package im

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/api"
	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/api/dto/converter"
	"github.com/tidbcloud/AgentTeam/internal/api/im/fanout"
	storeim "github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

const (
	defaultIMSearchLimit = 20
	maxIMSearchLimit     = 50

	errorCodeIMMessageSearchUnsupported = "message_search_unsupported"
)

// Delivery handles GET /api/v1/workspaces/:workspaceId/messages/:messageId/delivery.
func (h *MessagesHandler) Delivery(r api.WorkspaceScopedRequest) (api.Response, error) {
	if h.queries == nil {
		return api.Response{}, api.NewHTTPError(http.StatusInternalServerError, "Internal server error")
	}
	messageID, err := routeStandaloneMessageID(r)
	if err != nil {
		return api.Response{}, err
	}
	diagnostics, err := h.queries.GetMessageDeliveryDiagnosticsForUser(r.Context(), r.UserID(), r.WorkspaceID(), messageID)
	if err != nil {
		return api.Response{}, err
	}
	return api.OK(converter.ToMessageDeliveryDiagnostics(diagnostics)), nil
}

// Search handles GET /api/v1/workspaces/:workspaceId/messages/search.
func (h *MessagesHandler) Search(r api.WorkspaceScopedRequest) (api.Response, error) {
	if h.queries == nil || !h.queries.SupportsMessageSearch(r.Context()) {
		return api.Response{}, api.NewHTTPErrorWithCode(http.StatusNotImplemented, "Message search is not supported without a TiDB core database", errorCodeIMMessageSearchUnsupported)
	}
	query := strings.TrimSpace(r.Gin.Query("q"))
	if query == "" {
		return api.Response{}, api.NewHTTPError(http.StatusBadRequest, "q is required")
	}
	limit, err := parseLimit(r.Gin.Query("limit"), defaultIMSearchLimit, maxIMSearchLimit)
	if err != nil {
		return api.Response{}, err
	}
	offset := 0
	rawOffset := strings.TrimSpace(r.Gin.Query("offset"))
	if rawOffset != "" {
		parsed, err := parseOptionalInt64(rawOffset, "offset")
		if err != nil || parsed == nil {
			return api.Response{}, api.NewHTTPError(http.StatusBadRequest, "invalid offset")
		}
		offset = int(*parsed)
	}
	cursor := message.MessageSearchCursor{}
	rawCursor := strings.TrimSpace(r.Gin.Query("cursor"))
	if rawCursor != "" {
		if rawOffset != "" {
			return api.Response{}, api.NewHTTPError(http.StatusBadRequest, "cursor and offset cannot both be set")
		}
		parsed, err := converter.FromMessageSearchCursorToken(rawCursor)
		if err != nil {
			return api.Response{}, api.NewHTTPError(http.StatusBadRequest, "invalid cursor")
		}
		cursor = parsed
	}
	scopeID := strings.TrimSpace(r.Gin.Query("scope_id"))
	if scopeID == "" {
		scopeID = strings.TrimSpace(r.Gin.Query("channel_id"))
	}
	authorFilter, err := parseMessageAuthorFilter(r)
	if err != nil {
		return api.Response{}, err
	}
	// An empty cursor is the first cursor page. Only an explicit offset opts
	// into legacy offset pagination.
	useCursor := rawOffset == ""
	result, err := h.queries.SearchMessagesForUser(r.Context(), r.UserID(), r.WorkspaceID(), query, scopeID, message.MessageSearchOptions{
		Limit:        limit,
		Offset:       offset,
		Cursor:       cursor,
		UseCursor:    useCursor,
		AuthorFilter: authorFilter,
	})
	if err != nil {
		return api.Response{}, err
	}
	return api.OK(converter.ToMessageSearchResult(result)), nil
}

func parseMessageAuthorFilter(r api.WorkspaceScopedRequest) (message.MessageAuthorFilter, error) {
	rawKind, hasKind := r.Gin.GetQuery("author_kind")
	rawID, hasID := r.Gin.GetQuery("author_id")
	if !hasKind && !hasID {
		return message.MessageAuthorFilter{}, nil
	}
	if hasKind != hasID {
		return message.MessageAuthorFilter{}, api.NewHTTPError(http.StatusBadRequest, "author_kind and author_id must be provided together")
	}
	filter, err := message.NormalizeMessageAuthorFilter(message.MessageAuthorFilter{
		Kind: message.MessageAuthorFilterKind(rawKind),
		ID:   rawID,
	})
	if err != nil {
		return message.MessageAuthorFilter{}, api.NewHTTPError(http.StatusBadRequest, err.Error())
	}
	if filter.IsZero() {
		return message.MessageAuthorFilter{}, api.NewHTTPError(http.StatusBadRequest, "author_kind and author_id must be provided together")
	}
	return filter, nil
}

// SubmitAgentDelivery handles POST /api/v1/workspaces/:workspaceId/messages/:messageId/agent-delivery.
// It keeps the browser mutation on the Workspace IM HTTP surface and avoids
// emitting legacy Socket.IO IM events; browser-side updates are returned in the
// response and subsequent realtime fanout belongs to im_stream_events.
func (h *MessagesHandler) SubmitAgentDelivery(r api.WorkspaceScopedJSON[dto.AgentDeliveryCreateOptions]) (api.Response, error) {
	if h.agentDeliveries == nil {
		return api.Response{}, api.NewHTTPError(http.StatusNotImplemented, "agentDelivery is not supported")
	}
	messageID, err := routeStandaloneMessageID(r.WorkspaceScopedRequest)
	if err != nil {
		return api.Response{}, err
	}
	interruptPolicy, err := normalizeIMAgentDeliveryInterruptPolicy(&r.Body)
	if err != nil {
		return api.Response{}, err
	}
	if interruptPolicy == "" {
		return api.Response{}, api.NewHTTPError(http.StatusBadRequest, "interruptPolicy is required")
	}

	_, delivery, err := h.agentDeliveries.PrepareMessageAgentDeliveryForUser(r.Context(), r.UserID(), r.WorkspaceID(), messageID, interruptPolicy)
	if err != nil {
		return api.Response{}, err
	}
	pendingMessage, err := h.agentDeliveries.MarkMessageAgentDeliveryStatusForUser(r.Context(), r.UserID(), r.WorkspaceID(), messageID, interruptPolicy, message.AgentDeliveryStatusPending)
	if err != nil {
		return api.Response{}, err
	}
	if err := h.publishMessageDeliveryStatusUpdate(r.Context(), r.WorkspaceID(), pendingMessage); err != nil {
		return api.Response{}, err
	}
	if err := h.notifyAgentsStrict(r.Context(), r.WorkspaceID(), delivery); err != nil {
		failedMessage, markErr := h.agentDeliveries.MarkMessageAgentDeliveryStatusForUser(r.Context(), r.UserID(), r.WorkspaceID(), messageID, interruptPolicy, message.AgentDeliveryStatusFailed)
		var publishErr error
		if markErr == nil {
			publishErr = h.publishMessageDeliveryStatusUpdate(r.Context(), r.WorkspaceID(), failedMessage)
		}
		return api.Response{}, errors.Join(err, markErr, publishErr)
	}
	h.markScopeFanoutDelivered(r.Context(), delivery)
	appMessage, err := h.agentDeliveries.CommitMessageAgentDeliveryForUser(r.Context(), r.UserID(), r.WorkspaceID(), messageID, interruptPolicy)
	if err != nil {
		return api.Response{}, err
	}
	if decorator, ok := h.writer.(SentMessageDecorator); ok {
		if decorated, decorateErr := decorator.DecorateAppMessageForUser(r.Context(), r.UserID(), r.WorkspaceID(), appMessage); decorateErr == nil {
			appMessage = decorated
		}
	}
	if err := h.publishMessageDeliveryStatusUpdate(r.Context(), r.WorkspaceID(), appMessage); err != nil {
		return api.Response{}, err
	}
	return api.OK(converter.ToAppMessage(appMessage)), nil
}

func (h *MessagesHandler) publishMessageDeliveryStatusUpdate(ctx context.Context, workspaceID string, appMessage message.AppMessage) error {
	if !message.IsHumanVisibleAppMessage(appMessage) {
		return nil
	}
	return publishMessageDeliveryUpdatedEvent(ctx, h.streamWriter, workspaceID, appMessage.ChannelID, appMessage.ID)
}

func (h *MessagesHandler) deliverCommittedMessage(ctx context.Context, workspaceID string, appMessage message.AppMessage, delivery message.MessageDelivery) error {
	if h.streamWriter == nil {
		return fmt.Errorf("IM stream writer is not configured")
	}
	return fanout.NewDeliveryFanout(h.agentNotifier, h.streamWriter, h.logger).
		WithReplyThreadRootPreloader(h.replyRootPreloader).
		WithScopeFanoutMarker(h.scopeFanoutMarker).
		WithTaskScopeLifecycleGuard(h.taskScopeLifecycle).
		WithUnreadRuntime(h.unreadRuntime).
		WithUnreadDeltaSink(h.unreadSink).
		DeliverCommittedMessage(ctx, workspaceID, appMessage, delivery)
}

func (h *MessagesHandler) withActiveTaskScope(
	ctx context.Context,
	workspaceID string,
	scopeID string,
	effect func(context.Context) error,
) (bool, error) {
	if effect == nil {
		return false, fmt.Errorf("Task scope effect callback is required")
	}
	if !isTaskScopeMessageTarget(scopeID) || h == nil || h.taskScopeLifecycle == nil {
		return true, effect(ctx)
	}
	workspaceID = strings.TrimSpace(workspaceID)
	scopeID = strings.TrimSpace(scopeID)
	protectedCtx := storeim.WithTaskLifecycleProtectedScopes(ctx, workspaceID, []string{scopeID})
	return h.taskScopeLifecycle.WithActiveTaskScopes(
		ctx,
		workspaceID,
		[]string{scopeID},
		func() error {
			return effect(protectedCtx)
		},
	)
}

func (h *MessagesHandler) emitUnreadDeltas(workspaceID string, deltas []unread.UnreadDelta) {
	if h == nil || h.unreadSink == nil {
		return
	}
	for _, delta := range deltas {
		h.unreadSink.EmitUnreadUpdated(workspaceID, delta)
	}
}
func (h *MessagesHandler) notifyAgentsStrict(ctx context.Context, workspaceID string, delivery message.MessageDelivery) error {
	if h.agentNotifier == nil {
		return nil
	}
	var errs []error
	for _, recipient := range delivery.WakeRecipients {
		wakeMessage := receiveMessageForIMDelivery(delivery, recipient)
		if h.replyRootPreloader != nil {
			if decorated, preloadErr := h.replyRootPreloader.PreloadReplyThreadRootForRecipient(ctx, workspaceID, recipient, wakeMessage); preloadErr == nil {
				wakeMessage = decorated
			}
		}
		if err := h.agentNotifier.NotifyAgentMessage(ctx, recipient, delivery.Seq, wakeMessage); err != nil {
			errs = append(errs, fmt.Errorf("notify agent %s: %w", strings.TrimSpace(recipient.AgentID), err))
		}
	}
	return errors.Join(errs...)
}
func (h *MessagesHandler) markScopeFanoutDelivered(ctx context.Context, delivery message.MessageDelivery) {
	if h.scopeFanoutMarker == nil || delivery.IdempotentReplay {
		return
	}
	messageID := strings.TrimSpace(delivery.MessageID)
	if messageID == "" {
		return
	}
	_ = h.scopeFanoutMarker.MarkScopeFanoutDeliveredForMessage(ctx, messageID)
}

func receiveMessageForIMDelivery(delivery message.MessageDelivery, recipient message.AgentRecipient) message.ReceiveMessage {
	msg := delivery.Message
	if strings.TrimSpace(msg.MessageID) == "" {
		msg.MessageID = strings.TrimSpace(delivery.MessageID)
	}
	if strings.TrimSpace(msg.TargetKey) == "" {
		if delivery.AppMessage != nil && strings.TrimSpace(delivery.AppMessage.ChannelID) != "" {
			msg.TargetKey = strings.TrimSpace(delivery.AppMessage.ChannelID)
		} else if strings.TrimSpace(delivery.ThreadID) != "" {
			msg.TargetKey = strings.TrimSpace(delivery.ThreadID)
		}
	}
	if !msg.CarriesScopeTargetSeq() {
		msg.TargetSeq = 0
	} else if msg.TargetSeq <= 0 && !deliveryIsStartedReplyThreadRoot(delivery, msg) {
		if delivery.Seq > 0 {
			msg.TargetSeq = delivery.Seq
		} else if delivery.AppMessage != nil && delivery.AppMessage.Seq > 0 {
			msg.TargetSeq = delivery.AppMessage.Seq
		}
	}
	if strings.TrimSpace(msg.InterruptPolicy) == "" {
		msg.InterruptPolicy = strings.TrimSpace(delivery.InterruptPolicy)
	}
	if msg.Source.IsZero() && delivery.AppMessage != nil {
		source := delivery.AppMessage.Source.Normalize()
		if source.IsZero() {
			source = message.DefaultContributionSourceForAppMessage(*delivery.AppMessage)
		}
		msg.Source = source
	}
	return message.ReceiveMessageForAgentRecipient(msg, recipient)
}

func deliveryIsStartedReplyThreadRoot(delivery message.MessageDelivery, msg message.ReceiveMessage) bool {
	if delivery.StartedReplyThread == nil {
		return false
	}
	threadChannelID := strings.TrimSpace(delivery.StartedReplyThread.ThreadChannelID)
	if threadChannelID == "" {
		return false
	}
	if strings.TrimSpace(msg.TargetKey) != threadChannelID {
		return false
	}
	if messageID := strings.TrimSpace(delivery.MessageID); messageID != "" {
		return strings.TrimSpace(msg.MessageID) == messageID
	}
	return true
}

func normalizeIMAgentDeliveryInterruptPolicy(options *dto.AgentDeliveryCreateOptions) (string, error) {
	if options == nil {
		return "", nil
	}
	normalized, ok := message.NormalizeAgentInterruptPolicy(options.InterruptPolicy)
	if !ok {
		return "", api.NewHTTPError(http.StatusBadRequest, "unsupported interruptPolicy")
	}
	return normalized, nil
}

func routeStandaloneMessageID(r api.WorkspaceScopedRequest) (string, error) {
	messageID := strings.TrimSpace(r.Gin.Param("messageId"))
	if messageID == "" {
		return "", api.NewHTTPError(http.StatusBadRequest, "messageId is required")
	}
	return messageID, nil
}