package fanout

import (
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/api/dto/converter"
	"github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/rewind"
)

const (
	imEventTypeMessageCreated         = "message.created"
	imEventTypeMessageUpdated         = "message.updated"
	imEventTypeMessageRecalled        = "message.recalled"
	imEventTypeMessagePinsInvalidated = "message.pins.invalidated"
	imEventTypeReactionChanged        = "reaction.changed"
	imEventTypeThreadUpdated          = "thread.updated"
	imEventTypeMessageDelivery        = "message.delivery.updated"
	imEventTypeConversationRewound    = "conversation.rewound"
	imEventTypeTaskActivity           = "task.activity"
)

func publishMessageCreatedEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, scopeID string, appMessage message.AppMessage, delivery message.MessageDelivery) error {
	if writer == nil {
		return nil
	}
	// Ordinary message replay retains the historical no-op contract. A Task
	// message carries a frozen targeted audience, however, and the durable
	// outbox may be the first fanout attempt after a post-commit crash. Reapply
	// its stable message.created event so the Task Inbox row is recoverable.
	if delivery.IdempotentReplay && !delivery.HumanRecipientsTargeted && !delivery.RecoverableStreamProjection {
		return nil
	}
	message.RedactServiceAccountPublisherMetadataForSharedProjection(&appMessage)
	dtoMessage := converter.ToAppMessage(appMessage)
	payload := map[string]any{"message": dtoMessage}
	if delivery.SuppressBrowserNotification {
		payload["suppressBrowserNotification"] = true
	}
	if mention, recipients := messageMentionPayload(appMessage, delivery); mention != nil {
		payload["mention"] = mention
		payload["mentionRecipients"] = recipients
	}
	patches := map[string]any{}
	if delivery.ReplyThreadUpdated != nil {
		patches["thread"] = replyThreadUpdatedPayload(*delivery.ReplyThreadUpdated)
	}
	if len(patches) == 0 {
		patches = nil
	}
	entitySeq := appMessage.Seq
	conversation := conversationIndexFromMessage(dtoMessage)
	if delivery.Conversation != nil {
		conversation = conversationIndexFromDeliveryProjection(delivery.Conversation)
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID:        workspaceID,
		ScopeID:            scopeID,
		RecipientMemberIDs: append([]string(nil), delivery.HumanRecipientMemberIDs...),
		TargetedRecipients: delivery.HumanRecipientsTargeted,
		Event: im.StreamEvent{
			EventID:   messageCreatedStreamEventID(appMessage.ID),
			Type:      imEventTypeMessageCreated,
			EntityID:  appMessage.ID,
			EntitySeq: &entitySeq,
			Payload:   payload,
			Patches:   patches,
			CreatedAt: appMessage.CreatedAt,
		},
		Conversation: conversation,
	})
	if err != nil {
		return fmt.Errorf("append IM message stream event: %w", err)
	}
	return nil
}

func publishWorkflowMessageCreatedLiveEvent(
	writer StreamEventWriter,
	workspaceID string,
	scopeID string,
	appMessage message.AppMessage,
) {
	publisher, ok := writer.(ScopeLiveEventPublisher)
	if !ok {
		return
	}
	dtoMessage := converter.ToAppMessage(appMessage)
	entitySeq := appMessage.Seq
	publisher.PublishScopeLiveOnlyEvent(workspaceID, scopeID, im.StreamEvent{
		Type:      imEventTypeMessageCreated,
		EntityID:  appMessage.ID,
		EntitySeq: &entitySeq,
		Payload: map[string]any{
			"message":                     dtoMessage,
			"suppressBrowserNotification": true,
		},
		CreatedAt: appMessage.CreatedAt,
	})
}

func publishDeliveryIntentConversationEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, payload message.ScopeDeliveryIntentPayload) error {
	if writer == nil || payload.Conversation == nil || len(payload.HumanRecipientMemberIDs) == 0 {
		return nil
	}
	scopeID := strings.TrimSpace(payload.Message.TargetKey)
	eventID := strings.TrimSpace(payload.SourceRef.ID)
	if scopeID == "" || eventID == "" || payload.Message.TargetSeq <= 0 {
		return fmt.Errorf("task activity IM projection requires scope, source event, and sequence")
	}
	entitySeq := payload.Message.TargetSeq
	projection := conversationIndexFromDeliveryProjection(payload.Conversation)
	if projection == nil {
		return fmt.Errorf("task activity IM projection is incomplete")
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID:        strings.TrimSpace(workspaceID),
		ScopeID:            scopeID,
		RecipientMemberIDs: append([]string(nil), payload.HumanRecipientMemberIDs...),
		TargetedRecipients: payload.HumanRecipientsTargeted,
		Event: im.StreamEvent{
			EventID:        imEventTypeTaskActivity + ":" + eventID,
			Type:           imEventTypeTaskActivity,
			EntityID:       eventID,
			EntitySeq:      &entitySeq,
			ConversationID: strings.TrimSpace(payload.Conversation.ConversationID),
			Payload: map[string]any{
				"scopeId":      scopeID,
				"activityId":   eventID,
				"activityType": strings.TrimSpace(payload.Message.MessageType),
				"conversation": payload.Conversation.Projection,
			},
			CreatedAt: payload.Message.Timestamp,
		},
		Conversation: projection,
	})
	if err != nil {
		return fmt.Errorf("append IM task activity stream event: %w", err)
	}
	return nil
}

func conversationIndexFromDeliveryProjection(value *message.DeliveryConversationProjection) *im.ConversationIndexItem {
	if value == nil {
		return nil
	}
	projection := make(map[string]any, len(value.Projection))
	for key, item := range value.Projection {
		projection[key] = item
	}
	return &im.ConversationIndexItem{
		ConversationID: strings.TrimSpace(value.ConversationID),
		ScopeID:        strings.TrimSpace(value.ScopeID),
		Kind:           strings.TrimSpace(value.Kind),
		SortAt:         value.SortAt,
		Projection:     projection,
	}
}

func messageCreatedStreamEventID(messageID string) string {
	return imEventTypeMessageCreated + ":" + strings.TrimSpace(messageID)
}

func publishThreadUpdatedEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, updated message.ReplyThreadUpdated) error {
	if writer == nil {
		return nil
	}
	threadChannelID := strings.TrimSpace(updated.ThreadChannelID)
	if threadChannelID == "" {
		return nil
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     threadChannelID,
		Event: im.StreamEvent{
			Type:      imEventTypeThreadUpdated,
			EntityID:  strings.TrimSpace(updated.ParentMessageID),
			Payload:   replyThreadUpdatedPayload(updated),
			CreatedAt: replyThreadUpdatedCreatedAt(updated),
		},
	})
	if err != nil {
		return fmt.Errorf("append IM thread updated stream event: %w", err)
	}
	return nil
}

func publishMessageDeliveryUpdatedEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, scopeID string, messageID string) error {
	if writer == nil {
		return nil
	}
	trimmedMessageID := strings.TrimSpace(messageID)
	if strings.TrimSpace(scopeID) == "" || trimmedMessageID == "" {
		return nil
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     scopeID,
		Event: im.StreamEvent{
			Type:     imEventTypeMessageDelivery,
			EntityID: trimmedMessageID,
			Payload: map[string]any{
				"channelId":  strings.TrimSpace(scopeID),
				"messageIds": []string{trimmedMessageID},
			},
		},
	})
	if err != nil {
		return fmt.Errorf("append IM message delivery stream event: %w", err)
	}
	return nil
}

func publishMessageUpdatedEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, scopeID string, appMessage message.AppMessage, eventID string) error {
	if writer == nil {
		return nil
	}
	messageID := strings.TrimSpace(appMessage.ID)
	if messageID == "" {
		return nil
	}
	message.RedactServiceAccountPublisherMetadataForSharedProjection(&appMessage)
	entitySeq := appMessage.Seq
	payload := map[string]any{
		"scopeId":          strings.TrimSpace(scopeID),
		"messageId":        messageID,
		"seq":              appMessage.Seq,
		"content":          appMessage.Content,
		"metadata":         message.CopyMessageMetadata(appMessage.Metadata),
		"editedAt":         appMessage.EditedAt,
		"editedByMemberId": strings.TrimSpace(appMessage.EditedByMemberID),
		"editRevision":     appMessage.EditRevision,
	}
	createdAt := time.Now().UTC()
	if appMessage.EditedAt != nil && !appMessage.EditedAt.IsZero() {
		createdAt = appMessage.EditedAt.UTC()
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     strings.TrimSpace(scopeID),
		Event: im.StreamEvent{
			EventID:   strings.TrimSpace(eventID),
			Type:      imEventTypeMessageUpdated,
			EntityID:  messageID,
			EntitySeq: &entitySeq,
			Payload:   payload,
			CreatedAt: createdAt,
		},
	})
	if err != nil {
		return fmt.Errorf("append IM message update stream event: %w", err)
	}
	return nil
}

func publishMessageRecalledEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, scopeID string, appMessage message.AppMessage, updateConversation bool, eventID string) error {
	if writer == nil {
		return nil
	}
	messageID := strings.TrimSpace(appMessage.ID)
	if messageID == "" {
		return nil
	}
	message.RedactServiceAccountPublisherMetadataForSharedProjection(&appMessage)
	dtoMessage := converter.ToAppMessage(appMessage)
	entitySeq := appMessage.Seq
	payload := map[string]any{
		"scopeId":   strings.TrimSpace(scopeID),
		"messageId": messageID,
		"seq":       appMessage.Seq,
		"message":   dtoMessage,
	}
	if tombstone := message.CopyMessageMetadata(appMessage.Metadata)[message.MessageRecallMetadataKey]; tombstone != nil {
		payload["tombstone"] = tombstone
	}
	var conversation *im.ConversationIndexItem
	if updateConversation {
		conversation = conversationIndexFromMessage(dtoMessage)
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     strings.TrimSpace(scopeID),
		Event: im.StreamEvent{
			EventID:   strings.TrimSpace(eventID),
			Type:      imEventTypeMessageRecalled,
			EntityID:  messageID,
			EntitySeq: &entitySeq,
			Payload:   payload,
			CreatedAt: time.Now().UTC(),
		},
		Conversation: conversation,
	})
	if err != nil {
		return fmt.Errorf("append IM message recall stream event: %w", err)
	}
	return nil
}

func publishMessagePinsInvalidatedEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, scopeID string, messageIDs []string, eventID string) error {
	if writer == nil {
		return nil
	}
	scopeID = strings.TrimSpace(scopeID)
	if scopeID == "" {
		return nil
	}
	ids := make([]string, 0, len(messageIDs))
	for _, messageID := range messageIDs {
		if trimmed := strings.TrimSpace(messageID); trimmed != "" {
			ids = append(ids, trimmed)
		}
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     scopeID,
		Event: im.StreamEvent{
			EventID:  strings.TrimSpace(eventID),
			Type:     imEventTypeMessagePinsInvalidated,
			EntityID: scopeID,
			Payload: map[string]any{
				"scopeId":    scopeID,
				"messageIds": ids,
			},
			CreatedAt: time.Now().UTC(),
		},
	})
	if err != nil {
		return fmt.Errorf("append IM message pins invalidated stream event: %w", err)
	}
	return nil
}

func messageMutationStreamEventID(eventType, mutationID, scopeID string) string {
	mutationID = strings.TrimSpace(mutationID)
	if mutationID == "" {
		return ""
	}
	return strings.TrimSpace(eventType) + ":" + mutationID + ":" + strings.TrimSpace(scopeID)
}

func PublishConversationRewoundEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, plan rewind.Plan) error {
	if writer == nil {
		return nil
	}
	broadcastScopeID := strings.TrimSpace(plan.RealtimeBroadcastScopeID())
	rewoundScopeID := strings.TrimSpace(plan.RewindScopeID())
	if broadcastScopeID == "" || rewoundScopeID == "" {
		return nil
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     broadcastScopeID,
		Event: im.StreamEvent{
			Type:     imEventTypeConversationRewound,
			EntityID: strings.TrimSpace(plan.TargetMessageID),
			Payload: map[string]any{
				"channelId": rewoundScopeID,
			},
		},
	})
	if err != nil {
		return fmt.Errorf("append IM rewind stream event: %w", err)
	}
	return nil
}

func publishReactionChangedEvent(ctx context.Context, writer StreamEventWriter, workspaceID string, scopeID string, messageID string, reactions []message.MessageReactionAggregate, notification *message.HumanReactionNotification) error {
	if writer == nil {
		return nil
	}
	trimmedMessageID := strings.TrimSpace(messageID)
	if strings.TrimSpace(scopeID) == "" || trimmedMessageID == "" {
		return nil
	}
	payload := map[string]any{
		"messageId": trimmedMessageID,
		"reactions": converter.ToMessageReactionAggregates(reactions),
	}
	if notificationPayload := reactionNotificationPayload(notification); notificationPayload != nil {
		payload["reaction"] = notificationPayload
	}
	_, err := writer.AppendScopeEvent(ctx, im.ScopeEventRequest{
		WorkspaceID: workspaceID,
		ScopeID:     scopeID,
		Event: im.StreamEvent{
			Type:     imEventTypeReactionChanged,
			EntityID: trimmedMessageID,
			Payload:  payload,
		},
	})
	if err != nil {
		return fmt.Errorf("append IM reaction stream event: %w", err)
	}
	return nil
}

func conversationIndexFromMessage(message dto.AppMessage) *im.ConversationIndexItem {
	scopeID := strings.TrimSpace(message.ChannelID)
	if scopeID == "" {
		return nil
	}
	if strings.TrimSpace(message.ChannelType) == "task" {
		return nil
	}
	projection := jsonMap(message)
	projection["lastMessage"] = message
	if strings.TrimSpace(message.ChannelType) == "service_account" {
		profile := message.AuthorProfile
		title := "Service account"
		peerID := scopeID
		subtitle := ""
		avatarURL := ""
		if profile != nil {
			title = firstNonEmptyConversationValue(strings.TrimSpace(profile.DisplayName), strings.TrimSpace(profile.ServiceSlug), title)
			peerID = firstNonEmptyConversationValue(strings.TrimSpace(profile.PrincipalID), peerID)
			subtitle = strings.TrimSpace(profile.ServiceSlug)
			avatarURL = strings.TrimSpace(profile.AvatarURL)
		}
		projection["title"] = title
		projection["subtitle"] = subtitle
		projection["avatarUrl"] = avatarURL
		projection["peerType"] = "service"
		projection["peerId"] = peerID
		projection["preview"] = notificationPreview(message.Content)
		projection["lastAuthorName"] = title
		projection["lastMessageAt"] = message.CreatedAt
		projection["lastActivityAt"] = message.CreatedAt
		projection["subscriptionState"] = "active"
		projection["route"] = "/inbox/service-accounts/" + url.PathEscape(peerID)
		return &im.ConversationIndexItem{
			ConversationID: "serviceAccount:" + scopeID,
			ScopeID:        scopeID,
			Kind:           "service_account",
			SortAt:         message.CreatedAt,
			Projection:     projection,
		}
	}
	return &im.ConversationIndexItem{
		ScopeID:    scopeID,
		SortAt:     message.CreatedAt,
		Projection: projection,
	}
}

func firstNonEmptyConversationValue(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}

func replyThreadUpdatedPayload(updated message.ReplyThreadUpdated) map[string]any {
	return map[string]any{
		"parentMessageId": strings.TrimSpace(updated.ParentMessageID),
		"threadChannelId": strings.TrimSpace(updated.ThreadChannelID),
		"replyCount":      updated.ReplyCount,
		"lastReplyAt":     updated.LastReplyAt,
		"participantIds":  append([]string{}, updated.ParticipantIDs...),
		"participants":    converter.ToReplyThreadParticipants(updated.Participants),
		"replyAuthorIds":  append([]string{}, updated.ReplyAuthorIDs...),
		"replyAuthors":    converter.ToReplyThreadParticipants(updated.ReplyAuthors),
	}
}

func replyThreadUpdatedCreatedAt(updated message.ReplyThreadUpdated) time.Time {
	if updated.LastReplyAt != nil && !updated.LastReplyAt.IsZero() {
		return updated.LastReplyAt.UTC()
	}
	return time.Now().UTC()
}

func messageMentionPayload(appMessage message.AppMessage, delivery message.MessageDelivery) (map[string]any, []map[string]string) {
	if strings.TrimSpace(appMessage.ID) == "" || len(delivery.MentionedHumans) == 0 {
		return nil, nil
	}
	recipients := make([]map[string]string, 0, len(delivery.MentionedHumans))
	seen := make(map[string]struct{}, len(delivery.MentionedHumans))
	for _, recipient := range delivery.MentionedHumans {
		userID := strings.TrimSpace(recipient.UserID)
		memberID := strings.TrimSpace(recipient.MemberID)
		if userID == "" || memberID == "" {
			continue
		}
		key := userID + "\x00" + memberID
		if _, ok := seen[key]; ok {
			continue
		}
		seen[key] = struct{}{}
		recipients = append(recipients, map[string]string{"userId": userID, "memberId": memberID})
	}
	if len(recipients) == 0 {
		return nil, nil
	}
	payload := map[string]any{
		"messageId":         strings.TrimSpace(appMessage.ID),
		"channelId":         strings.TrimSpace(appMessage.ChannelID),
		"channelType":       strings.TrimSpace(delivery.MentionContext.ChannelType),
		"parentChannelId":   strings.TrimSpace(delivery.MentionContext.ParentChannelID),
		"parentChannelType": strings.TrimSpace(delivery.MentionContext.ParentChannelType),
		"parentMessageId":   strings.TrimSpace(delivery.MentionContext.ParentMessageID),
		"authorProfile":     converter.ToAuthorProfile(appMessage.EffectiveAuthorProfile()),
		"contentPreview":    notificationPreview(appMessage.Content),
	}
	if taskContext := taskNotificationContextPayload(appMessage.TaskContext); taskContext != nil {
		payload["taskContext"] = taskContext
	}
	return payload, recipients
}

func taskNotificationContextPayload(context *message.AppMessageTaskContext) map[string]any {
	if context == nil {
		return nil
	}
	channelID := strings.TrimSpace(context.ChannelID)
	if channelID == "" || context.TaskNumber <= 0 {
		return nil
	}
	return map[string]any{
		"channelId":  channelID,
		"taskNumber": context.TaskNumber,
	}
}

func reactionNotificationPayload(notification *message.HumanReactionNotification) map[string]any {
	if notification == nil {
		return nil
	}
	if strings.TrimSpace(notification.ReactionID) == "" || strings.TrimSpace(notification.MessageID) == "" || strings.TrimSpace(notification.ChannelID) == "" || !notification.AuthorProfile.Normalize().Valid() {
		return nil
	}
	payload := map[string]any{
		"reactionId":        strings.TrimSpace(notification.ReactionID),
		"messageId":         strings.TrimSpace(notification.MessageID),
		"channelId":         strings.TrimSpace(notification.ChannelID),
		"channelType":       strings.TrimSpace(notification.ChannelType),
		"parentChannelId":   strings.TrimSpace(notification.ParentChannelID),
		"parentChannelType": strings.TrimSpace(notification.ParentChannelType),
		"parentMessageId":   strings.TrimSpace(notification.ParentMessageID),
		"authorProfile":     converter.ToAuthorProfile(notification.AuthorProfile),
		"emoji":             strings.TrimSpace(notification.Emoji),
		"contentPreview":    strings.TrimSpace(notification.ContentPreview),
	}
	if taskContext := taskNotificationContextPayload(notification.TaskContext); taskContext != nil {
		payload["taskContext"] = taskContext
	}
	return payload
}

func notificationPreview(content string) string {
	trimmed := strings.TrimSpace(content)
	if trimmed == "" {
		return ""
	}
	const maxRunes = 140
	runes := []rune(trimmed)
	if len(runes) <= maxRunes {
		return trimmed
	}
	return string(runes[:maxRunes]) + "..."
}

func jsonMap(value any) map[string]any {
	if value == nil {
		return map[string]any{}
	}
	encoded, err := json.Marshal(value)
	if err != nil {
		return map[string]any{}
	}
	var decoded map[string]any
	if err := json.Unmarshal(encoded, &decoded); err != nil || decoded == nil {
		return map[string]any{}
	}
	return decoded
}