package testdelivery

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/agent"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

type candidate struct {
	payload      []byte
	scopeID      string
	seq          int64
	global       int64
	eventRefType string
	sortAt       time.Time
}

// ReceiveAgentMessagesFromFanout is a fixture-only bridge for legacy integration
// assertions that used to inspect server-side pending deliveries. Production
// daemon delivery is driven by agent:deliver payloads; this helper only decodes
// persisted scope_fanout_outbox payloads and advances fixture read state so
// existing tests can assert the write-time routing plan without reviving server
// receive code.
func ReceiveAgentMessagesFromFanout(ctx context.Context, db coredb.DB, authenticated agent.AuthenticatedAgent, limit int) ([]message.ReceiveMessage, error) {
	messages, messageCursors, activityCursors, err := listAgentFanoutMessages(ctx, db, authenticated, limit)
	if err != nil {
		return nil, err
	}
	if err := advanceFixtureReadState(ctx, db, strings.TrimSpace(authenticated.MemberID), messageCursors, activityCursors); err != nil {
		return nil, err
	}
	return messages, nil
}

// AgentHasUnreadFanoutMessages reports whether the fixture fanout payload store
// still contains any unconsumed write-time delivery for the agent.
func AgentHasUnreadFanoutMessages(ctx context.Context, db coredb.DB, authenticated agent.AuthenticatedAgent) (bool, error) {
	messages, _, _, err := listAgentFanoutMessages(ctx, db, authenticated, 1)
	if err != nil {
		return false, err
	}
	return len(messages) > 0, nil
}

type cursorUpdate struct {
	seq    int64
	global int64
}

type cursorScopeUpdate struct {
	scopeID string
	cursor  cursorUpdate
}

func listAgentFanoutMessages(ctx context.Context, db coredb.DB, authenticated agent.AuthenticatedAgent, limit int) ([]message.ReceiveMessage, map[string]cursorUpdate, map[string]cursorUpdate, error) {
	if db == nil {
		return nil, nil, nil, fmt.Errorf("fanout fixture db is nil")
	}
	workspaceID := strings.TrimSpace(authenticated.WorkspaceID)
	memberID := strings.TrimSpace(authenticated.MemberID)
	if workspaceID == "" || memberID == "" {
		return nil, nil, nil, fmt.Errorf("fanout fixture workspace and member are required")
	}
	if limit <= 0 {
		limit = 50
	}
	candidateLimit := limit * 20
	if candidateLimit < 50 {
		candidateLimit = 50
	}
	if candidateLimit > 1000 {
		candidateLimit = 1000
	}
	candidates, err := loadFanoutCandidates(ctx, db, workspaceID, memberID, candidateLimit)
	if err != nil {
		return nil, nil, nil, err
	}
	service := message.NewScopeAPIService(message.New(db, nil, 0, message.Options{}))
	messages := make([]message.ReceiveMessage, 0, limit)
	messageCursors := make(map[string]cursorUpdate)
	activityCursors := make(map[string]cursorUpdate)
	for _, row := range candidates {
		var msg message.ReceiveMessage
		var scopeID string
		var seq int64
		var global int64
		isDeliveryIntent := false
		switch strings.TrimSpace(row.eventRefType) {
		case message.DeliverySourceRefMessage:
			payload, err := message.UnmarshalScopeFanoutPayload(row.payload)
			if err != nil {
				return nil, nil, nil, err
			}
			if strings.TrimSpace(payload.AppMessage.AuthorMemberID()) == memberID {
				continue
			}
			recipient, ok := fanoutPayloadRecipientForMember(payload, memberID)
			if !ok {
				continue
			}
			msg = payload.Delivery.Message
			if strings.TrimSpace(msg.MessageID) == "" {
				msg = receiveMessageFromPayload(payload)
			}
			if msg.Source.IsZero() {
				source := payload.AppMessage.Source.Normalize()
				if source.IsZero() {
					source = message.DefaultContributionSourceForAppMessage(payload.AppMessage)
				}
				msg.Source = source
			}
			msg = message.ReceiveMessageForAgentRecipient(msg, recipient)
			scopeID = firstNonEmpty(row.scopeID, msg.TargetKey, payload.AppMessage.ChannelID)
			seq = firstPositive(row.seq, msg.TargetSeq, payload.AppMessage.Seq, payload.Delivery.Seq)
			global = firstPositive(row.global, payload.Delivery.GlobalSeq)
		default:
			isDeliveryIntent = true
			payload, err := message.UnmarshalScopeDeliveryIntentPayload(row.payload)
			if err != nil {
				return nil, nil, nil, err
			}
			if strings.TrimSpace(payload.Message.AuthorProfile.MemberID) == memberID {
				continue
			}
			recipient, ok := deliveryIntentRecipientForMember(payload, memberID)
			if !ok {
				continue
			}
			msg = message.ReceiveMessageForAgentRecipient(payload.Message, recipient)
			scopeID = firstNonEmpty(row.scopeID, msg.TargetKey)
			seq = firstPositive(row.seq, msg.TargetSeq)
			global = firstPositive(row.global)
		}
		msg, err = service.PreloadReplyThreadRootMetadata(ctx, workspaceID, memberID, message.ScopeActorAgent, msg)
		if err != nil {
			return nil, nil, nil, err
		}
		messages = append(messages, msg)
		if scopeID != "" {
			cursors := messageCursors
			if isDeliveryIntent {
				cursors = activityCursors
			}
			current := cursors[scopeID]
			if seq > current.seq || (seq == current.seq && global > current.global) {
				cursors[scopeID] = cursorUpdate{seq: seq, global: global}
			}
		}
		if len(messages) >= limit {
			break
		}
	}
	return messages, messageCursors, activityCursors, nil
}

func loadFanoutCandidates(ctx context.Context, db coredb.Queryer, workspaceID string, memberID string, limit int) ([]candidate, error) {
	rows, err := db.Query(ctx, `
SELECT payload, scope_id, seq, global_seq, event_ref_type, sort_at
FROM (
  SELECT sfo.payload AS payload, m.scope_id AS scope_id, m.seq AS seq, m.workspace_global_seq AS global_seq,
         sfo.event_ref_type AS event_ref_type, m.created_at AS sort_at, sfo.id AS outbox_id
  FROM scope_fanout_outbox sfo
  JOIN scope_messages m
    ON m.id = sfo.message_id
   AND m.workspace_id = sfo.workspace_id
  JOIN scope_memberships sm
    ON sm.scope_id = m.scope_id
   AND sm.member_id = $2
   AND sm.removed_at IS NULL
   AND sm.effective_to_seq IS NULL
  WHERE sfo.workspace_id = $1
    AND sfo.event_ref_type = 'message'
    AND sfo.fanout_type = 'scope_fanout'
    AND sfo.payload IS NOT NULL
    AND LENGTH(sfo.payload) > 0
    AND m.seq > COALESCE(sm.last_read_seq, 0)
  UNION ALL
  SELECT sfo.payload AS payload, sfo.scope_id AS scope_id, COALESCE(ae.seq, 0) AS seq,
         COALESCE(ae.activity_global_seq, 0) AS global_seq, sfo.event_ref_type AS event_ref_type,
         sfo.created_at AS sort_at, sfo.id AS outbox_id
  FROM scope_fanout_outbox sfo
  LEFT JOIN scope_activity_events ae
    ON ae.id = sfo.event_ref_id
   AND ae.workspace_id = sfo.workspace_id
   AND ae.scope_id = sfo.scope_id
   AND ae.suppressed_at IS NULL
  LEFT JOIN scope_memberships sm
    ON sm.scope_id = sfo.scope_id
   AND sm.member_id = $2
   AND sm.removed_at IS NULL
   AND sm.effective_to_seq IS NULL
  WHERE sfo.workspace_id = $1
    AND sfo.event_ref_type <> 'message'
    AND sfo.fanout_type = 'scope_fanout'
    AND sfo.payload IS NOT NULL
    AND LENGTH(sfo.payload) > 0
    AND (sm.member_id IS NULL OR COALESCE(ae.activity_global_seq, 0) > COALESCE(sm.last_activity_global_seq, 0))
) candidates
ORDER BY sort_at ASC, scope_id ASC, seq ASC, outbox_id ASC
LIMIT $3
`, workspaceID, memberID, limit)
	if err != nil {
		return nil, fmt.Errorf("list fanout fixture candidates: %w", err)
	}
	defer rows.Close()

	result := make([]candidate, 0)
	for rows.Next() {
		var row candidate
		if err := rows.Scan(&row.payload, &row.scopeID, &row.seq, &row.global, &row.eventRefType, &row.sortAt); err != nil {
			return nil, fmt.Errorf("scan fanout fixture candidate: %w", err)
		}
		row.payload = append([]byte(nil), row.payload...)
		result = append(result, row)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate fanout fixture candidates: %w", err)
	}
	return result, nil
}

func fanoutPayloadRecipientForMember(payload message.ScopeFanoutPayload, memberID string) (message.AgentRecipient, bool) {
	memberID = strings.TrimSpace(memberID)
	var fallback message.AgentRecipient
	var found bool
	for _, recipient := range payload.Delivery.Recipients {
		if strings.TrimSpace(recipient.AgentMemberID) == memberID {
			if len(recipient.TeamContexts) > 0 {
				return recipient, true
			}
			if !found {
				fallback = recipient
				found = true
			}
		}
	}
	for _, recipient := range payload.Delivery.WakeRecipients {
		if strings.TrimSpace(recipient.AgentMemberID) == memberID {
			if len(recipient.TeamContexts) > 0 {
				return recipient, true
			}
			if !found {
				fallback = recipient
				found = true
			}
		}
	}
	return fallback, found
}

func deliveryIntentRecipientForMember(payload message.ScopeDeliveryIntentPayload, memberID string) (message.AgentRecipient, bool) {
	memberID = strings.TrimSpace(memberID)
	var fallback message.AgentRecipient
	var found bool
	for _, recipient := range payload.Recipients {
		if strings.TrimSpace(recipient.AgentMemberID) == memberID {
			if len(recipient.TeamContexts) > 0 {
				return recipient, true
			}
			if !found {
				fallback = recipient
				found = true
			}
		}
	}
	for _, recipient := range payload.WakeRecipients {
		if strings.TrimSpace(recipient.AgentMemberID) == memberID {
			if len(recipient.TeamContexts) > 0 {
				return recipient, true
			}
			if !found {
				fallback = recipient
				found = true
			}
		}
	}
	return fallback, found
}

func receiveMessageFromPayload(payload message.ScopeFanoutPayload) message.ReceiveMessage {
	appMessage := payload.AppMessage
	return message.ReceiveMessage{
		MessageID:         appMessage.ID,
		TargetKey:         appMessage.ChannelID,
		TargetSeq:         appMessage.Seq,
		ChannelType:       appMessage.ChannelType,
		AuthorProfile:     appMessage.EffectiveAuthorProfile(),
		MessageType:       appMessage.MessageType,
		Content:           appMessage.Content,
		Metadata:          appMessage.Metadata,
		Timestamp:         appMessage.CreatedAt,
		Attachments:       appMessage.Attachments,
		ActiveTaskTarget:  payload.Delivery.Message.ActiveTaskTarget,
		ParentChannelType: appMessage.ParentChannelType,
		ParentChannelName: payload.Delivery.Message.ParentChannelName,
	}
}

func advanceFixtureReadState(ctx context.Context, db coredb.QueryExecer, memberID string, messageCursors map[string]cursorUpdate, activityCursors map[string]cursorUpdate) error {
	now := time.Now().UTC()
	if err := advanceFixtureMessageReadState(ctx, db, memberID, messageCursors, now); err != nil {
		return err
	}
	if err := advanceFixtureActivityReadState(ctx, db, memberID, activityCursors, now); err != nil {
		return err
	}
	return nil
}

func advanceFixtureMessageReadState(ctx context.Context, db coredb.QueryExecer, memberID string, messageCursors map[string]cursorUpdate, now time.Time) error {
	updates := cursorScopeUpdates(messageCursors, func(cursor cursorUpdate) bool {
		return cursor.seq > 0
	})
	if len(updates) == 0 {
		return nil
	}

	args := []any{strings.TrimSpace(memberID), now}
	seqCases := make([]string, 0, len(updates))
	globalCases := make([]string, 0, len(updates))
	scopePlaceholders := make([]string, 0, len(updates))
	for _, update := range updates {
		scopeArg := len(args) + 1
		args = append(args, update.scopeID)
		seqArg := len(args) + 1
		args = append(args, update.cursor.seq)
		globalArg := len(args) + 1
		args = append(args, update.cursor.global)
		seqCases = append(seqCases, fmt.Sprintf("WHEN $%d THEN GREATEST(last_read_seq, $%d)", scopeArg, seqArg))
		globalCases = append(globalCases, fmt.Sprintf("WHEN $%d THEN GREATEST(last_read_global_seq, $%d)", scopeArg, globalArg))
		scopePlaceholders = append(scopePlaceholders, fmt.Sprintf("$%d", scopeArg))
	}

	_, err := db.Exec(ctx, fmt.Sprintf(`
UPDATE scope_memberships
SET unread_generation = CASE
  WHEN CASE scope_id
%s
    ELSE last_read_seq
  END <> last_read_seq OR CASE scope_id
%s
    ELSE last_read_global_seq
  END <> last_read_global_seq THEN unread_generation + 1
  ELSE unread_generation
END,
last_read_seq = CASE scope_id
%s
    ELSE last_read_seq
END,
    last_read_global_seq = CASE scope_id
    %s
    ELSE last_read_global_seq
  END,
    updated_at = $2
WHERE member_id = $1
  AND scope_id IN (%s)
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, strings.Join(seqCases, "\n    "), strings.Join(globalCases, "\n    "), strings.Join(seqCases, "\n    "), strings.Join(globalCases, "\n    "), strings.Join(scopePlaceholders, ", ")), args...)
	if err != nil {
		return fmt.Errorf("advance fanout fixture message read state: %w", err)
	}
	return nil
}

func advanceFixtureActivityReadState(ctx context.Context, db coredb.QueryExecer, memberID string, activityCursors map[string]cursorUpdate, now time.Time) error {
	updates := cursorScopeUpdates(activityCursors, func(cursor cursorUpdate) bool {
		return cursor.global > 0
	})
	if len(updates) == 0 {
		return nil
	}

	args := []any{strings.TrimSpace(memberID), now}
	globalCases := make([]string, 0, len(updates))
	scopePlaceholders := make([]string, 0, len(updates))
	for _, update := range updates {
		scopeArg := len(args) + 1
		args = append(args, update.scopeID)
		globalArg := len(args) + 1
		args = append(args, update.cursor.global)
		globalCases = append(globalCases, fmt.Sprintf("WHEN $%d THEN GREATEST(last_activity_global_seq, $%d)", scopeArg, globalArg))
		scopePlaceholders = append(scopePlaceholders, fmt.Sprintf("$%d", scopeArg))
	}

	_, err := db.Exec(ctx, fmt.Sprintf(`
UPDATE scope_memberships
SET unread_generation = CASE
  WHEN CASE scope_id
%s
    ELSE last_activity_global_seq
  END <> last_activity_global_seq THEN unread_generation + 1
  ELSE unread_generation
END,
last_activity_global_seq = CASE scope_id
%s
    ELSE last_activity_global_seq
END,
    updated_at = $2
WHERE member_id = $1
  AND scope_id IN (%s)
  AND removed_at IS NULL
  AND effective_to_seq IS NULL
`, strings.Join(globalCases, "\n    "), strings.Join(globalCases, "\n    "), strings.Join(scopePlaceholders, ", ")), args...)
	if err != nil {
		return fmt.Errorf("advance fanout fixture activity read state: %w", err)
	}
	return nil
}

func cursorScopeUpdates(cursors map[string]cursorUpdate, include func(cursorUpdate) bool) []cursorScopeUpdate {
	byScopeID := make(map[string]cursorUpdate, len(cursors))
	for scopeID, cursor := range cursors {
		trimmedScopeID := strings.TrimSpace(scopeID)
		if trimmedScopeID == "" || !include(cursor) {
			continue
		}
		if current, ok := byScopeID[trimmedScopeID]; !ok || cursorAfter(cursor, current) {
			byScopeID[trimmedScopeID] = cursor
		}
	}

	updates := make([]cursorScopeUpdate, 0, len(byScopeID))
	for scopeID, cursor := range byScopeID {
		updates = append(updates, cursorScopeUpdate{scopeID: scopeID, cursor: cursor})
	}
	sort.Slice(updates, func(i, j int) bool {
		return updates[i].scopeID < updates[j].scopeID
	})
	return updates
}

func cursorAfter(cursor cursorUpdate, current cursorUpdate) bool {
	return cursor.seq > current.seq || (cursor.seq == current.seq && cursor.global > current.global)
}

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

func firstPositive(values ...int64) int64 {
	for _, value := range values {
		if value > 0 {
			return value
		}
	}
	return 0
}