package pl.allegro.tech.hermes.management.api;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import jakarta.annotation.security.RolesAllowed;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.DELETE;
import jakarta.ws.rs.DefaultValue;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.HttpMethod;
import jakarta.ws.rs.NotFoundException;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.PUT;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.container.ContainerRequestContext;
import jakarta.ws.rs.core.Context;
import jakarta.ws.rs.core.Response;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import pl.allegro.tech.hermes.api.MessageTextPreview;
import pl.allegro.tech.hermes.api.OwnerId;
import pl.allegro.tech.hermes.api.PatchData;
import pl.allegro.tech.hermes.api.Query;
import pl.allegro.tech.hermes.api.Topic;
import pl.allegro.tech.hermes.api.TopicMetrics;
import pl.allegro.tech.hermes.api.TopicName;
import pl.allegro.tech.hermes.api.TopicWithSchema;
import pl.allegro.tech.hermes.common.exception.BrokerNotFoundForPartitionException;
import pl.allegro.tech.hermes.management.api.auth.HermesSecurityAwareRequestUser;
import pl.allegro.tech.hermes.management.api.auth.ManagementRights;
import pl.allegro.tech.hermes.management.api.auth.Roles;
import pl.allegro.tech.hermes.management.domain.auth.RequestUser;
import pl.allegro.tech.hermes.management.domain.owner.OwnerSource;
import pl.allegro.tech.hermes.management.domain.owner.OwnerSourceNotFound;
import pl.allegro.tech.hermes.management.domain.owner.OwnerSources;
import pl.allegro.tech.hermes.management.domain.topic.CreatorRights;
import pl.allegro.tech.hermes.management.domain.topic.SingleMessageReaderException;
import pl.allegro.tech.hermes.management.domain.topic.TopicService;

import java.util.List;
import java.util.Optional;

import static com.google.common.base.Strings.isNullOrEmpty;
import static jakarta.ws.rs.core.MediaType.APPLICATION_JSON;
import static jakarta.ws.rs.core.MediaType.APPLICATION_OCTET_STREAM;
import static jakarta.ws.rs.core.Response.status;
import static java.lang.String.format;

@Component
@Path("/topics")
@Api(value = "/topics", description = "Operations on topics")
public class TopicsEndpoint {

    private final TopicService topicService;
    private final ManagementRights managementRights;
    private final OwnerSources ownerSources;

    @Autowired
    public TopicsEndpoint(TopicService topicService,
                          ManagementRights managementRights,
                          OwnerSources ownerSources) {
        this.topicService = topicService;
        this.managementRights = managementRights;
        this.ownerSources = ownerSources;
    }

    @GET
    @Produces(APPLICATION_JSON)
    @ApiOperation(value = "List topics from group", response = List.class, httpMethod = HttpMethod.GET)
    public List<String> list(
            @DefaultValue("") @QueryParam("groupName") String groupName,
            @DefaultValue("false") @QueryParam("tracked") boolean tracked) {

        return tracked ? listTracked(groupName) : listNames(groupName);
    }

    @POST
    @Consumes(APPLICATION_JSON)
    @Produces(APPLICATION_JSON)
    @Path("/query")
    @ApiOperation(value = "Queries topics from group", response = List.class, httpMethod = HttpMethod.POST)
    public List<String> queryList(
            @DefaultValue("") @QueryParam("groupName") String groupName,
            Query<Topic> query) {

        return isNullOrEmpty(groupName)
                ? topicService.listFilteredTopicNames(query)
                : topicService.listFilteredTopicNames(groupName, query);
    }

    @POST
    @Consumes(APPLICATION_JSON)
    @Produces(APPLICATION_JSON)
    @RolesAllowed(Roles.ANY)
    @ApiOperation(value = "Create topic", httpMethod = HttpMethod.POST)
    public Response create(TopicWithSchema topicWithSchema, @Context ContainerRequestContext requestContext) {
        RequestUser requestUser = new HermesSecurityAwareRequestUser(requestContext);
        CreatorRights isAllowedToManage = checkedTopic -> managementRights.isUserAllowedToManageTopic(checkedTopic, requestContext);
        topicService.createTopicWithSchema(topicWithSchema, requestUser, isAllowedToManage);
        return status(Response.Status.CREATED).build();
    }

    @DELETE
    @Produces(APPLICATION_JSON)
    @Path("/{topicName}")
    @RolesAllowed({Roles.ADMIN, Roles.TOPIC_OWNER})
    @ApiOperation(value = "Remove topic", httpMethod = HttpMethod.DELETE)
    public Response remove(@PathParam("topicName") String qualifiedTopicName, @Context ContainerRequestContext requestContext) {
        RequestUser requestUser = new HermesSecurityAwareRequestUser(requestContext);
        topicService.removeTopicWithSchema(topicService.getTopicDetails(TopicName.fromQualifiedName(qualifiedTopicName)),
                requestUser);
        return status(Response.Status.OK).build();
    }

    @PUT
    @Consumes(APPLICATION_JSON)
    @Produces(APPLICATION_JSON)
    @Path("/{topicName}")
    @RolesAllowed({Roles.ADMIN, Roles.TOPIC_OWNER})
    @ApiOperation(value = "Update topic", httpMethod = HttpMethod.PUT)
    public Response update(@PathParam("topicName") String qualifiedTopicName, PatchData patch,
                           @Context ContainerRequestContext requestContext) {
        RequestUser requestUser = new HermesSecurityAwareRequestUser(requestContext);
        topicService.updateTopicWithSchema(TopicName.fromQualifiedName(qualifiedTopicName), patch, requestUser);
        return status(Response.Status.OK).build();
    }

    @GET
    @Produces(APPLICATION_JSON)
    @Path("/{topicName}")
    @ApiOperation(value = "Topic details", httpMethod = HttpMethod.GET)
    public TopicWithSchema get(@PathParam("topicName") String qualifiedTopicName) {
        return topicService.getTopicWithSchema(TopicName.fromQualifiedName(qualifiedTopicName));
    }

    @GET
    @Produces(APPLICATION_JSON)
    @Path("/{topicName}/metrics")
    @ApiOperation(value = "Topic metrics", httpMethod = HttpMethod.GET)
    public TopicMetrics getMetrics(@PathParam("topicName") String qualifiedTopicName) {
        return topicService.getTopicMetrics(TopicName.fromQualifiedName(qualifiedTopicName));
    }

    @GET
    @Produces(APPLICATION_JSON)
    @Path("/{topicName}/preview")
    @ApiOperation(value = "Topic publisher preview", httpMethod = HttpMethod.GET)
    public List<MessageTextPreview> getPreview(@PathParam("topicName") String qualifiedTopicName) {
        return topicService.previewText(TopicName.fromQualifiedName(qualifiedTopicName));
    }

    @GET
    @Produces(APPLICATION_OCTET_STREAM)
    @Path("/{topicName}/preview/{idx}")
    @ApiOperation(value = "Topic publisher preview", httpMethod = HttpMethod.GET)
    public byte[] getPreviewRaw(@PathParam("topicName") String qualifiedTopicName, @PathParam("idx") Integer idx) {
        TopicName topicName = TopicName.fromQualifiedName(qualifiedTopicName);
        Optional<byte[]> preview = topicService.preview(topicName, idx);
        if (preview.isPresent()) {
            return preview.get();
        } else {
            throw new NotFoundException(format(
                    "Message preview not found for topic %s and offset %d",
                    topicName, idx
            ));
        }
    }

    @GET
    @Produces(APPLICATION_JSON)
    @Path("/{topicName}/preview/cluster/{brokersClusterName}/partition/{partition}/offset/{offset}")
    @RolesAllowed({Roles.ADMIN})
    @ApiOperation(value = "Fetch single message from specified brokers cluster", httpMethod = HttpMethod.GET)
    public String preview(@PathParam("topicName") String qualifiedTopicName,
                          @PathParam("brokersClusterName") String brokersClusterName,
                          @PathParam("partition") Integer partition,
                          @PathParam("offset") Long offset) {
        try {
            return topicService.fetchSingleMessageFromPrimary(brokersClusterName, TopicName.fromQualifiedName(qualifiedTopicName),
                    partition, offset);
        } catch (BrokerNotFoundForPartitionException | SingleMessageReaderException exception) {
            throw new NotFoundException(
                    format("Message not found for brokers cluster %s, topic %s, partition %d and offset %d", brokersClusterName,
                            qualifiedTopicName, partition, offset));
        }
    }

    @GET
    @Produces(APPLICATION_JSON)
    @Path("/owner/{ownerSourceName}/{ownerId}")
    public List<Topic> listForOwner(@PathParam("ownerSourceName") String ownerSourceName, @PathParam("ownerId") String id) {
        OwnerSource ownerSource = ownerSources.getByName(ownerSourceName)
                .orElseThrow(() -> new OwnerSourceNotFound(ownerSourceName));
        if (!ownerSource.exists(id)) {
            throw new OwnerSource.OwnerNotFound(ownerSourceName, id);
        }
        OwnerId ownerId = new OwnerId(ownerSource.name(), id);
        return topicService.listForOwnerId(ownerId);
    }

    private List<String> listTracked(String groupName) {
        return isNullOrEmpty(groupName) ? topicService.listTrackedTopicNames() : topicService.listTrackedTopicNames(groupName);
    }

    private List<String> listNames(String groupName) {
        return isNullOrEmpty(groupName) ? topicService.listQualifiedTopicNames() : topicService.listQualifiedTopicNames(groupName);
    }
}
