package message

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/access"
)

const (
	HumanScopeSubscriptionStateActive = humanScopeSubscriptionStateActive
	HumanScopeSubscriptionStateQuiet  = humanScopeSubscriptionStateQuiet
	HumanScopeSubscriptionStateMute   = humanScopeSubscriptionStateMute
)

func NormalizeHumanScopeSubscriptionState(state string) (string, error) {
	return normalizeHumanScopeSubscriptionState(state)
}

func UpsertScopeNotificationState(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, state string, updatedAt time.Time) error {
	return upsertScopeNotificationStateWithExecer(ctx, execer, memberID, scopeID, state, updatedAt)
}

// UpsertTaskWatchNotificationState writes the Task-only v2 Watch authority and
// mirrors it to the legacy state column for old binaries during rolling
// deploys. Old binaries may subsequently overwrite state, but cannot change
// task_watch_state.
func UpsertTaskWatchNotificationState(ctx context.Context, execer CommandTagExecer, memberID string, scopeID string, state string, updatedAt time.Time) error {
	trimmedMemberID := strings.TrimSpace(memberID)
	trimmedScopeID := strings.TrimSpace(scopeID)
	normalizedState := strings.TrimSpace(state)
	if trimmedMemberID == "" || trimmedScopeID == "" || normalizedState == "" {
		return nil
	}
	actionAt := updatedAt.UTC()
	if actionAt.IsZero() {
		actionAt = time.Now().UTC()
	}
	if _, err := execer.Exec(ctx, `
INSERT INTO scope_notification_settings (
  scope_id, member_id, state, task_watch_state,
  created_at, updated_at, unread_generation
) VALUES ($1, $2, $3, $3, $4, $4, 1)
ON DUPLICATE KEY UPDATE
  unread_generation = CASE
    WHEN state <> VALUES(state)
      OR COALESCE(task_watch_state, '') <> VALUES(task_watch_state)
      THEN unread_generation + 1
    ELSE unread_generation
  END,
  updated_at = CASE
    WHEN state <> VALUES(state)
      OR COALESCE(task_watch_state, '') <> VALUES(task_watch_state)
      THEN VALUES(updated_at)
    ELSE updated_at
  END,
  state = VALUES(state),
  task_watch_state = VALUES(task_watch_state)
`, trimmedScopeID, trimmedMemberID, normalizedState, actionAt); err != nil {
		return fmt.Errorf("upsert task Watch notification state: %w", err)
	}
	return nil
}

func EnsureReplyThreadParticipantPresence(ctx context.Context, execer coredb.QueryExecer, threadChannelID string, participantID string, participatedAt time.Time) error {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	trimmedParticipantID := strings.TrimSpace(participantID)
	if trimmedThreadChannelID == "" || trimmedParticipantID == "" {
		return nil
	}
	memberIDs, err := resolveReplyThreadParticipantMemberIDs(ctx, execer, []string{trimmedParticipantID})
	if err != nil {
		return err
	}
	if len(memberIDs) == 0 {
		return nil
	}
	now := participatedAt.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	return ensureReplyThreadParticipantMembershipState(ctx, execer, trimmedThreadChannelID, memberIDs[0], "auto-wake", now, nil)
}

func EnsureReplyThreadParticipantProjection(ctx context.Context, execer coredb.QueryExecer, threadChannelID string, memberID string, agentState string, createdAt time.Time) error {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	trimmedMemberID := strings.TrimSpace(memberID)
	if trimmedThreadChannelID == "" || trimmedMemberID == "" {
		return nil
	}
	explicitlyUnsubscribed, err := isExplicitlyUnsubscribedFromThread(ctx, execer, trimmedMemberID, trimmedThreadChannelID)
	if err != nil {
		return err
	}
	if explicitlyUnsubscribed {
		return nil
	}
	return ensureReplyThreadParticipantProjectionState(ctx, execer, trimmedThreadChannelID, trimmedMemberID, agentState, createdAt)
}

func ensureReplyThreadParticipantProjectionState(ctx context.Context, execer coredb.QueryExecer, threadChannelID string, memberID string, agentState string, createdAt time.Time) error {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	trimmedMemberID := strings.TrimSpace(memberID)
	if trimmedThreadChannelID == "" || trimmedMemberID == "" {
		return nil
	}
	now := createdAt.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	notificationState := strings.TrimSpace(agentState)
	// The participant notification INSERT runs inside the same fence tx as the
	// membership write (via notify) so a concurrent remover cannot land between
	// them and leave a notification without its membership (or vice versa).
	notify := func(writeExecer coredb.QueryExecer) error {
		if _, err := writeExecer.Exec(ctx, `
INSERT IGNORE INTO scope_notification_settings (
  scope_id, member_id, state, created_at, updated_at
)
SELECT $1, $2, CASE WHEN type = 'agent' THEN $3 ELSE 'auto-wake' END, $4, $4
FROM members
WHERE id = $2
`, trimmedThreadChannelID, trimmedMemberID, notificationState, now); err != nil {
			return fmt.Errorf("insert reply thread participant subscription: %w", err)
		}
		if forceReplyThreadParticipantNotifyErrorForTest != nil {
			return forceReplyThreadParticipantNotifyErrorForTest
		}
		return nil
	}
	return ensureReplyThreadParticipantMembershipState(ctx, execer, trimmedThreadChannelID, trimmedMemberID, agentState, now, notify)
}

func ensureReplyThreadParticipantMembershipState(ctx context.Context, execer coredb.QueryExecer, threadChannelID string, memberID string, agentState string, participatedAt time.Time, notify func(coredb.QueryExecer) error) error {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	trimmedMemberID := strings.TrimSpace(memberID)
	if trimmedThreadChannelID == "" || trimmedMemberID == "" {
		return nil
	}
	now := participatedAt.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}

	return withChannelReplyThreadFence(ctx, execer, trimmedThreadChannelID, trimmedMemberID, func(writeExecer coredb.QueryExecer) error {
		// Resolve defaults under the parent lock (on the fence tx) so a concurrent
		// remover cannot leave the child write with a stale owner/visibility.
		visibility, ownerMemberID, _, err := scopeThreadMembershipDefaults(ctx, writeExecer, trimmedThreadChannelID, trimmedMemberID, agentState)
		if err != nil {
			return err
		}
		visibility = normalizedMembershipVisibility(visibility)
		ownerMemberID = strings.TrimSpace(ownerMemberID)

		openMembershipUpdate, err := writeExecer.Exec(ctx, `
UPDATE scope_memberships
SET unread_generation = CASE
        WHEN visibility <> $3 OR COALESCE(owner_member_id, '') <> $4 OR removed_at IS NOT NULL OR archived_at IS NOT NULL THEN unread_generation + 1
        ELSE unread_generation
    END,
    visibility = $3,
    owner_member_id = NULLIF($4, ''),
    authority_from_at = CASE
        WHEN removed_at IS NOT NULL OR archived_at IS NOT NULL OR effective_to_seq IS NOT NULL THEN LEAST($5, CURRENT_TIMESTAMP(6))
        ELSE COALESCE(authority_from_at, created_at)
    END,
    authority_to_at = CASE
        WHEN removed_at IS NOT NULL OR archived_at IS NOT NULL OR effective_to_seq IS NOT NULL THEN NULL
        ELSE authority_to_at
    END,
    updated_at = $5,
    removed_at = NULL,
    archived_at = NULL
WHERE scope_id = $1
  AND member_id = $2
  AND effective_to_seq IS NULL
`, trimmedThreadChannelID, trimmedMemberID, visibility, ownerMemberID, now)
		if err != nil {
			return fmt.Errorf("update reply thread participant membership: %w", err)
		}
		if openMembershipUpdate.RowsAffected() == 0 {
			if _, err := writeExecer.Exec(ctx, `
INSERT INTO scope_memberships (
  scope_id, member_id, visibility, owner_member_id, last_read_seq,
  last_read_global_seq, last_activity_global_seq,
  owner_context_dismissed_through_global_seq, effective_from_seq,
  effective_to_seq, authority_from_at, authority_to_at,
  created_at, updated_at, removed_at, archived_at
) VALUES ($1, $2, $3, NULLIF($4, ''), 0, 0, 0, 0, 0, NULL, LEAST($5, CURRENT_TIMESTAMP(6)), NULL, $5, $5, NULL, NULL)
ON DUPLICATE KEY UPDATE
  unread_generation = CASE
    WHEN scope_memberships.visibility <> VALUES(visibility)
      OR COALESCE(scope_memberships.owner_member_id, '') <> COALESCE(VALUES(owner_member_id), '')
      OR scope_memberships.effective_to_seq IS NOT NULL
      OR scope_memberships.removed_at IS NOT NULL
      OR scope_memberships.archived_at IS NOT NULL THEN scope_memberships.unread_generation + 1
    ELSE scope_memberships.unread_generation
  END,
  visibility = VALUES(visibility),
  owner_member_id = VALUES(owner_member_id),
  effective_to_seq = NULL,
  authority_from_at = CASE
    WHEN scope_memberships.removed_at IS NOT NULL OR scope_memberships.archived_at IS NOT NULL OR scope_memberships.effective_to_seq IS NOT NULL THEN VALUES(authority_from_at)
    ELSE COALESCE(scope_memberships.authority_from_at, scope_memberships.created_at)
  END,
  authority_to_at = CASE
    WHEN scope_memberships.removed_at IS NOT NULL OR scope_memberships.archived_at IS NOT NULL OR scope_memberships.effective_to_seq IS NOT NULL THEN NULL
    ELSE scope_memberships.authority_to_at
  END,
  removed_at = NULL,
  archived_at = NULL,
  updated_at = VALUES(updated_at)
`, trimmedThreadChannelID, trimmedMemberID, visibility, ownerMemberID, now); err != nil {
				return fmt.Errorf("ensure reply thread participant membership: %w", err)
			}
		}
		if notify != nil {
			return notify(writeExecer)
		}
		return nil
	})
}

func (txs TxStore) ListSubscribedThreadsForUser(ctx context.Context, userID string, workspaceID string) ([]SubscribedThread, error) {
	return txs.ListSubscribedThreadsForUserWithLimit(ctx, userID, workspaceID, 0)
}

func (txs TxStore) ListSubscribedThreadInboxPageForUser(ctx context.Context, userID string, workspaceID string, limit int, cursor ThreadInboxCursor) (ThreadInboxPage, error) {
	membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID))
	if err != nil {
		return ThreadInboxPage{}, mapAccessError(err)
	}

	normalizedCursor, err := normalizeThreadInboxCursor(cursor)
	if err != nil {
		return ThreadInboxPage{}, err
	}
	query, args, err := BuildListSubscribedThreadInboxPageForUserScopedQuery(workspaceID, membership.Member.ID, userID, limit, normalizedCursor)
	if err != nil {
		return ThreadInboxPage{}, err
	}

	rows, err := txs.tx.Query(ctx, query, args...)
	if err != nil {
		return ThreadInboxPage{}, fmt.Errorf("list subscribed thread inbox page: %w", err)
	}
	defer rows.Close()

	threads := make([]SubscribedThread, 0)
	threadChannelIDs := make([]string, 0)
	sortAts := make([]time.Time, 0)
	for rows.Next() {
		var thread SubscribedThread
		var rawParentMessageMetadata []byte
		var sortAt time.Time
		if err := rows.Scan(
			&thread.ThreadChannelID,
			&thread.ParentChannelID,
			&thread.ParentChannelType,
			&thread.ParentLabel,
			&thread.ParentMessageID,
			&thread.ParentMessageShortID,
			&thread.ParentMessagePreview,
			&rawParentMessageMetadata,
			&thread.ParentMessageAuthor,
			&thread.SubscriptionState,
			&thread.ReplyCount,
			&thread.LastReplyAt,
			&thread.LastActivityAt,
			&sortAt,
		); err != nil {
			return ThreadInboxPage{}, fmt.Errorf("scan subscribed thread inbox row: %w", err)
		}
		parentMessageMetadata, err := DecodeMessageMetadata(rawParentMessageMetadata)
		if err != nil {
			return ThreadInboxPage{}, err
		}
		thread.ParentMessageMetadata = parentMessageMetadata
		normalizeSubscribedThreadTitle(&thread)
		threads = append(threads, thread)
		threadChannelIDs = append(threadChannelIDs, thread.ThreadChannelID)
		sortAts = append(sortAts, sortAt)
	}
	if err := rows.Err(); err != nil {
		return ThreadInboxPage{}, fmt.Errorf("iterate subscribed thread inbox rows: %w", err)
	}

	hasMore := len(threads) > limit
	if hasMore {
		threads = threads[:limit]
		threadChannelIDs = threadChannelIDs[:limit]
		sortAts = sortAts[:limit]
	}

	participantDetailsMap, authorMap, err := loadReplyThreadParticipantAndAuthorDetailsMaps(ctx, txs.tx, threadChannelIDs)
	if err != nil {
		return ThreadInboxPage{}, err
	}
	for i := range threads {
		threads[i].ParticipantIDs = replyThreadParticipantIDsFromDetails(participantDetailsMap[threads[i].ThreadChannelID])
		threads[i].ReplyAuthors = authorMap[threads[i].ThreadChannelID]
		threads[i].ReplyAuthorIDs = ReplyThreadRecentAuthorIDsFromDetails(threads[i].ReplyAuthors)
	}
	if err := attachSubscribedThreadParentResolvedMentions(ctx, txs.tx, workspaceID, membership.Member.ID, threads); err != nil {
		return ThreadInboxPage{}, err
	}

	nextCursor := ThreadInboxCursor{}
	if hasMore && len(threads) > 0 {
		nextCursor, err = normalizeThreadInboxCursor(ThreadInboxCursor{
			SortAt:          sortAts[len(sortAts)-1],
			ThreadChannelID: threads[len(threads)-1].ThreadChannelID,
		})
		if err != nil {
			return ThreadInboxPage{}, err
		}
	}

	return ThreadInboxPage{
		Threads:    threads,
		HasMore:    hasMore,
		NextCursor: nextCursor,
	}, nil
}

func (txs TxStore) CountSubscribedThreadsForUser(ctx context.Context, userID string, workspaceID string) (ThreadInboxCount, error) {
	membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID))
	if err != nil {
		return ThreadInboxCount{}, mapAccessError(err)
	}

	query, args, err := BuildCountSubscribedThreadsForUserScopedQuery(workspaceID, membership.Member.ID, userID)
	if err != nil {
		return ThreadInboxCount{}, err
	}

	var count int64
	if err := txs.tx.QueryRow(ctx, query, args...).Scan(&count); err != nil {
		return ThreadInboxCount{}, fmt.Errorf("count subscribed threads: %w", err)
	}
	return ThreadInboxCount{Count: count}, nil
}

func (txs TxStore) ListSubscribedThreadsForUserWithLimit(ctx context.Context, userID string, workspaceID string, limit int) ([]SubscribedThread, error) {
	membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID))
	if err != nil {
		return nil, mapAccessError(err)
	}

	query, args, err := BuildListSubscribedThreadsForUserScopedQuery(workspaceID, membership.Member.ID, userID, limit)
	if err != nil {
		return nil, err
	}

	rows, err := txs.tx.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list subscribed threads: %w", err)
	}
	defer rows.Close()

	threads := make([]SubscribedThread, 0)
	threadChannelIDs := make([]string, 0)
	for rows.Next() {
		var thread SubscribedThread
		var rawParentMessageMetadata []byte
		if err := rows.Scan(
			&thread.ThreadChannelID,
			&thread.ParentChannelID,
			&thread.ParentChannelType,
			&thread.ParentLabel,
			&thread.ParentMessageID,
			&thread.ParentMessageShortID,
			&thread.ParentMessagePreview,
			&rawParentMessageMetadata,
			&thread.ParentMessageAuthor,
			&thread.SubscriptionState,
			&thread.ReplyCount,
			&thread.LastReplyAt,
			&thread.LastActivityAt,
		); err != nil {
			return nil, fmt.Errorf("scan subscribed thread: %w", err)
		}
		parentMessageMetadata, err := DecodeMessageMetadata(rawParentMessageMetadata)
		if err != nil {
			return nil, err
		}
		thread.ParentMessageMetadata = parentMessageMetadata
		normalizeSubscribedThreadTitle(&thread)
		threads = append(threads, thread)
		threadChannelIDs = append(threadChannelIDs, thread.ThreadChannelID)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate subscribed threads: %w", err)
	}

	participantDetailsMap, authorMap, err := loadReplyThreadParticipantAndAuthorDetailsMaps(ctx, txs.tx, threadChannelIDs)
	if err != nil {
		return nil, err
	}
	for i := range threads {
		threads[i].ParticipantIDs = replyThreadParticipantIDsFromDetails(participantDetailsMap[threads[i].ThreadChannelID])
		threads[i].ReplyAuthors = authorMap[threads[i].ThreadChannelID]
		threads[i].ReplyAuthorIDs = ReplyThreadRecentAuthorIDsFromDetails(threads[i].ReplyAuthors)
	}
	if err := attachSubscribedThreadParentResolvedMentions(ctx, txs.tx, workspaceID, membership.Member.ID, threads); err != nil {
		return nil, err
	}
	return threads, nil
}

func attachSubscribedThreadParentResolvedMentions(ctx context.Context, queryer coredb.Queryer, workspaceID string, viewerMemberID string, threads []SubscribedThread) error {
	if len(threads) == 0 {
		return nil
	}
	parentMessages := make([]AppMessage, 0, len(threads))
	threadIndexes := make([]int, 0, len(threads))
	for index, thread := range threads {
		if strings.TrimSpace(thread.ParentChannelID) == "" || strings.TrimSpace(thread.ParentMessagePreview) == "" {
			continue
		}
		parentMessages = append(parentMessages, AppMessage{
			ID:        strings.TrimSpace(thread.ParentMessageID),
			ChannelID: strings.TrimSpace(thread.ParentChannelID),
			Content:   thread.ParentMessagePreview,
			Metadata:  thread.ParentMessageMetadata,
		})
		threadIndexes = append(threadIndexes, index)
	}
	if len(parentMessages) == 0 {
		return nil
	}
	if err := AttachResolvedMentionsForAppMessages(ctx, queryer, workspaceID, viewerMemberID, parentMessages); err != nil {
		return err
	}
	for index, appMessage := range parentMessages {
		threads[threadIndexes[index]].ParentMessageResolvedMentions = appMessage.ResolvedMentions
	}
	return nil
}

func (txs TxStore) GetReplyThreadForUser(ctx context.Context, userID string, workspaceID string, threadChannelID string) (SubscribedThread, error) {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	if trimmedThreadChannelID == "" {
		return SubscribedThread{}, invalidArgument("threadChannelId is required")
	}

	membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID))
	if err != nil {
		return SubscribedThread{}, mapAccessError(err)
	}
	if _, err := RequireMemberReplyThreadRead(ctx, txs.tx, membership.Member.ID, strings.TrimSpace(workspaceID), trimmedThreadChannelID); err != nil {
		return SubscribedThread{}, err
	}

	thread, err := txs.loadReplyThreadForMember(ctx, strings.TrimSpace(workspaceID), membership.Member.ID, trimmedThreadChannelID)
	if err != nil {
		return SubscribedThread{}, err
	}
	participantMap, authorMap, err := loadReplyThreadParticipantAndAuthorDetailsMaps(ctx, txs.tx, []string{trimmedThreadChannelID})
	if err != nil {
		return SubscribedThread{}, err
	}
	thread.Participants = participantMap[trimmedThreadChannelID]
	thread.Participants, err = txs.mergeVisibleReplyThreadAuthorParticipants(
		ctx,
		strings.TrimSpace(workspaceID),
		membership.Member.ID,
		trimmedThreadChannelID,
		thread.Participants,
	)
	if err != nil {
		return SubscribedThread{}, err
	}
	thread.ParticipantIDs = replyThreadParticipantIDsFromDetails(thread.Participants)
	thread.ParticipantCount = len(thread.Participants)
	thread.ReplyAuthors = authorMap[trimmedThreadChannelID]
	thread.ReplyAuthorIDs = ReplyThreadRecentAuthorIDsFromDetails(thread.ReplyAuthors)
	return thread, nil
}

// mergeVisibleReplyThreadAuthorParticipants keeps the detailed Thread GET
// roster aligned with the messages this viewer can actually see. Durable
// notification participants remain the primary projection; visible message
// authors fill the "Other participants" fallback without exposing authors from
// private messages that the current viewer cannot read.
func (txs TxStore) mergeVisibleReplyThreadAuthorParticipants(
	ctx context.Context,
	workspaceID string,
	viewerMemberID string,
	threadChannelID string,
	participants []ReplyThreadParticipant,
) ([]ReplyThreadParticipant, error) {
	result := append([]ReplyThreadParticipant(nil), participants...)
	seenMemberIDs := make(map[string]struct{}, len(result))
	for _, participant := range result {
		if memberID := strings.TrimSpace(participant.MemberID); memberID != "" {
			seenMemberIDs[memberID] = struct{}{}
		}
	}

	service := NewScopeAPIServiceForQueryer(txs.tx)
	visibleAuthorMemberIDs, err := service.loadVisibleAuthorMemberIDs(
		ctx,
		workspaceID,
		threadChannelID,
		viewerMemberID,
	)
	if err != nil {
		return nil, fmt.Errorf("load visible reply thread author participants: %w", err)
	}
	missingMemberIDs := make([]string, 0, len(visibleAuthorMemberIDs))
	for _, memberID := range visibleAuthorMemberIDs {
		if _, exists := seenMemberIDs[memberID]; exists {
			continue
		}
		seenMemberIDs[memberID] = struct{}{}
		missingMemberIDs = append(missingMemberIDs, memberID)
	}
	presentations, err := loadScopeMemberPresentations(
		ctx,
		txs.tx,
		workspaceID,
		missingMemberIDs,
	)
	if err != nil {
		return nil, err
	}
	for _, memberID := range missingMemberIDs {
		presentation, ok := presentations[memberID]
		if !ok {
			continue
		}
		result = append(result, ReplyThreadParticipant{
			ID:          firstNonEmpty(presentation.PublicID, presentation.MemberID),
			Type:        strings.TrimSpace(presentation.MemberType),
			DisplayName: firstNonEmpty(presentation.DisplayName, presentation.PublicID, presentation.MemberID),
			AvatarURL:   strings.TrimSpace(presentation.AvatarURL),
			MemberID:    strings.TrimSpace(presentation.MemberID),
		})
	}
	return result, nil
}

func (txs TxStore) GetThreadMembershipForUser(ctx context.Context, userID string, workspaceID string, threadChannelID string) (ThreadMembership, error) {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	if trimmedThreadChannelID == "" {
		return ThreadMembership{}, invalidArgument("threadChannelId is required")
	}

	membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID))
	if err != nil {
		return ThreadMembership{}, mapAccessError(err)
	}
	if _, err := RequireMemberReplyThreadRead(ctx, txs.tx, membership.Member.ID, strings.TrimSpace(workspaceID), trimmedThreadChannelID); err != nil {
		return ThreadMembership{}, err
	}

	var state string
	err = txs.tx.QueryRow(ctx, `
SELECT state
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
  AND state IN ('active', 'quiet', 'mute')
LIMIT 1
`, trimmedThreadChannelID, membership.Member.ID).Scan(&state)
	if err != nil {
		if coredb.IsNotFound(err) {
			return ThreadMembership{
				ThreadChannelID: trimmedThreadChannelID,
				Subscribed:      false,
			}, nil
		}
		return ThreadMembership{}, fmt.Errorf("load thread membership: %w", err)
	}

	return ThreadMembership{
		ThreadChannelID:   trimmedThreadChannelID,
		Subscribed:        true,
		SubscriptionState: strings.TrimSpace(state),
	}, nil
}

func (txs TxStore) loadReplyThreadForMember(ctx context.Context, workspaceID string, memberID string, threadChannelID string) (SubscribedThread, error) {
	row := txs.tx.QueryRow(ctx, `
SELECT
    reply_scope.id AS thread_channel_id,
    parent_scope.id AS parent_channel_id,
    CASE
        WHEN parent_scope.type = 'channel' AND parent_cp.subsystem = 'kb' THEN 'kb'
        ELSE parent_scope.type
    END AS parent_channel_type,
    CASE
        WHEN parent_scope.type = 'channel' THEN COALESCE(NULLIF(parent_cp.display_name, ''), NULLIF(parent_cp.slug, ''), parent_scope.id)
        WHEN parent_peer.type = 'agent' THEN COALESCE(NULLIF(parent_peer.display_name, ''), NULLIF(parent_peer_agent.name, ''), '')
        WHEN parent_peer.type = 'human' THEN COALESCE(NULLIF(parent_peer_user.display_name, ''), NULLIF(parent_peer.display_name, ''), parent_peer_user.name, '')
        ELSE ''
    END AS parent_label,
    COALESCE(reply_scope.root_message_id, '') AS parent_message_id,
    COALESCE(reply_scope.root_message_short_id, '') AS parent_message_short_id,
    COALESCE(pm.content, '') AS parent_message_preview,
    COALESCE(pm.metadata, '{}') AS parent_message_metadata,
    CASE
        WHEN pm.author_type = 'agent' THEN COALESCE(NULLIF(author.display_name, ''), NULLIF(author_agent.name, ''), '')
        WHEN pm.author_type = 'human' THEN COALESCE(NULLIF(author_user.display_name, ''), NULLIF(author.display_name, ''), author_user.name, '')
        WHEN pm.author_type = 'service' THEN COALESCE(NULLIF(author.display_name, ''), author.id)
        ELSE ''
    END AS parent_message_author,
    COALESCE(ss.state, '') AS subscription_state,
    COALESCE(reply_scope.latest_seq, 0) AS reply_count,
    reply_scope.latest_message_at AS last_reply_at,
    COALESCE(reply_scope.latest_message_at, ss.updated_at, reply_scope.updated_at) AS last_activity_at
FROM scopes reply_scope
JOIN scopes parent_scope
  ON parent_scope.id = reply_scope.parent_scope_id
 AND parent_scope.workspace_id = reply_scope.workspace_id
LEFT JOIN scope_channel_profiles parent_cp ON parent_cp.scope_id = parent_scope.id
LEFT JOIN scope_memberships parent_peer_membership
  ON parent_peer_membership.scope_id = parent_scope.id
 AND parent_peer_membership.member_id <> $2
 AND parent_peer_membership.removed_at IS NULL
 AND parent_peer_membership.archived_at IS NULL
 AND parent_peer_membership.effective_to_seq IS NULL
 AND parent_scope.type = 'dm'
LEFT JOIN members parent_peer ON parent_peer.id = parent_peer_membership.member_id
LEFT JOIN users parent_peer_user ON parent_peer.type = 'human' AND parent_peer_user.id = parent_peer.user_id
LEFT JOIN agents parent_peer_agent ON parent_peer.type = 'agent' AND parent_peer_agent.member_id = parent_peer.id
LEFT JOIN scope_messages pm ON pm.id = reply_scope.root_message_id AND pm.scope_id = reply_scope.parent_scope_id
LEFT JOIN members author ON author.id = pm.author_member_id
LEFT JOIN users author_user ON pm.author_type = 'human' AND author_user.id = author.user_id
LEFT JOIN agents author_agent ON pm.author_type = 'agent' AND author_agent.member_id = author.id
LEFT JOIN scope_notification_settings ss
  ON ss.scope_id = reply_scope.id
 AND ss.member_id = $2
 AND ss.state IN ('active', 'quiet', 'mute')
WHERE reply_scope.workspace_id = $1
  AND reply_scope.id = $3
  AND reply_scope.type = 'thread'
  AND reply_scope.status = 'active'
LIMIT 1
`, workspaceID, memberID, threadChannelID)

	var thread SubscribedThread
	var rawParentMessageMetadata []byte
	if err := row.Scan(
		&thread.ThreadChannelID,
		&thread.ParentChannelID,
		&thread.ParentChannelType,
		&thread.ParentLabel,
		&thread.ParentMessageID,
		&thread.ParentMessageShortID,
		&thread.ParentMessagePreview,
		&rawParentMessageMetadata,
		&thread.ParentMessageAuthor,
		&thread.SubscriptionState,
		&thread.ReplyCount,
		&thread.LastReplyAt,
		&thread.LastActivityAt,
	); err != nil {
		if coredb.IsNotFound(err) {
			return SubscribedThread{}, ErrNotFound
		}
		return SubscribedThread{}, fmt.Errorf("load reply thread: %w", err)
	}
	parentMessageMetadata, err := DecodeMessageMetadata(rawParentMessageMetadata)
	if err != nil {
		return SubscribedThread{}, err
	}
	thread.ParentMessageMetadata = parentMessageMetadata
	normalizeSubscribedThreadTitle(&thread)
	threads := []SubscribedThread{thread}
	if err := attachSubscribedThreadParentResolvedMentions(ctx, txs.tx, workspaceID, memberID, threads); err != nil {
		return SubscribedThread{}, err
	}
	thread = threads[0]
	return thread, nil
}

func normalizeSubscribedThreadTitle(thread *SubscribedThread) {
	if thread == nil {
		return
	}
	thread.Title = DeriveSubscribedThreadTitle(*thread)
}

func DeriveSubscribedThreadTitle(thread SubscribedThread) string {
	for _, candidate := range []string{
		thread.Title,
		thread.ParentMessagePreview,
		thread.ParentLabel,
		thread.ParentMessageShortID,
		thread.ThreadChannelID,
	} {
		if title := strings.TrimSpace(candidate); title != "" {
			return title
		}
	}
	return "thread"
}

func (txs TxStore) UnsubscribeThreadForUser(ctx context.Context, userID string, workspaceID string, threadChannelID string) error {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	if trimmedThreadChannelID == "" {
		return badRequest("threadChannelId is required")
	}

	membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID))
	if err != nil {
		return mapAccessError(err)
	}
	if _, err := RequireMemberReplyThreadRead(ctx, txs.tx, membership.Member.ID, workspaceID, trimmedThreadChannelID); err != nil {
		return err
	}

	now := time.Now().UTC()
	if err := UpsertThreadSubscription(ctx, txs.tx, membership.Member.ID, trimmedThreadChannelID, now, &now); err != nil {
		return fmt.Errorf("unsubscribe thread: %w", err)
	}
	if err := archiveReplyThreadConversationIndexRow(ctx, txs.tx, workspaceID, membership.Member.ID, trimmedThreadChannelID, now); err != nil {
		return err
	}
	return nil
}

func archiveReplyThreadConversationIndexRow(ctx context.Context, execer coredb.QueryExecer, workspaceID string, memberID string, threadChannelID string, archivedAt time.Time) error {
	workspaceID = strings.TrimSpace(workspaceID)
	memberID = strings.TrimSpace(memberID)
	threadChannelID = strings.TrimSpace(threadChannelID)
	if workspaceID == "" || memberID == "" || threadChannelID == "" {
		return nil
	}
	at := archivedAt.UTC()
	if at.IsZero() {
		at = time.Now().UTC()
	}
	if _, err := execer.Exec(ctx, `
UPDATE im_conversation_index
SET archived_at = COALESCE(archived_at, $4),
    updated_at = $4
WHERE workspace_id = $1
  AND member_id = $2
  AND scope_id = $3
  AND kind = 'thread'
`, workspaceID, memberID, threadChannelID, at); err != nil {
		return fmt.Errorf("archive thread conversation index row: %w", err)
	}
	return nil
}

func (txs TxStore) SubscribeThreadForUser(ctx context.Context, userID string, workspaceID string, threadChannelID string) error {
	return txs.SetThreadSubscriptionStateForUser(ctx, userID, workspaceID, threadChannelID, humanScopeSubscriptionStateActive)
}

func (txs TxStore) SetThreadSubscriptionStateForUser(ctx context.Context, userID string, workspaceID string, threadChannelID string, state string) error {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	if trimmedThreadChannelID == "" {
		return badRequest("threadChannelId is required")
	}
	normalizedState, err := normalizeHumanScopeSubscriptionState(state)
	if err != nil {
		return err
	}

	membership, err := access.LoadHumanWorkspaceMember(ctx, txs.tx, strings.TrimSpace(userID), strings.TrimSpace(workspaceID))
	if err != nil {
		return mapAccessError(err)
	}
	if _, err := RequireMemberReplyThreadRead(ctx, txs.tx, membership.Member.ID, workspaceID, trimmedThreadChannelID); err != nil {
		return err
	}

	now := time.Now().UTC()
	if _, err := txs.ensureReplyThreadScopeProjection(ctx, trimmedThreadChannelID); err != nil {
		return fmt.Errorf("subscribe thread scope projection: %w", err)
	}
	if _, err := upsertScopeThreadSubscriptionStateWithExecer(ctx, txs.tx, membership.Member.ID, trimmedThreadChannelID, normalizedState, 0, 0, now); err != nil {
		return fmt.Errorf("subscribe thread scope subscription state: %w", err)
	}
	return nil
}

// SilenceDerivedReplyThreadParticipantFence swallows ErrScopeParentMembershipInactive
// for DERIVED reply-thread participants (root author, mentions): a participant who
// has left the parent channel is simply not revived on the reply thread, and must
// NOT block the current legitimate writer. Direct/explicit writers keep the fence
// fail-closed and must not route through this.
func SilenceDerivedReplyThreadParticipantFence(err error) error {
	if errors.Is(err, ErrScopeParentMembershipInactive) {
		return nil
	}
	return err
}

// forceReplyThreadParticipantNotifyErrorForTest, when non-nil, is returned by the
// reply-thread participant notify step right after the notification INSERT, so a
// test can prove the membership + notification writes are atomic: the whole fence
// tx must roll back (both rows gone) rather than leave the membership behind. It
// is nil in production.
var forceReplyThreadParticipantNotifyErrorForTest error

func (txs TxStore) SyncReplyThreadSubscriptions(ctx context.Context, authorMemberID string, authorType domain.MemberType, threadChannelID string, parentMessageID string, mentionedMemberIDs []string, agentMentionMemberIDs []string, currentMessageSeq int64, currentMessageGlobalSeq int64, subscribedAt time.Time) error {
	trimmedAuthorMemberID := strings.TrimSpace(authorMemberID)
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	trimmedParentMessageID := strings.TrimSpace(parentMessageID)
	if trimmedAuthorMemberID == "" || trimmedThreadChannelID == "" || trimmedParentMessageID == "" {
		return fmt.Errorf("reply thread subscription context is incomplete")
	}
	if err := EnsureReplyThreadParentDMAgentParticipant(ctx, txs.tx, trimmedThreadChannelID, subscribedAt); err != nil {
		return err
	}

	parentAuthorMemberID, err := txs.loadReplyThreadParentAuthorMemberID(ctx, trimmedThreadChannelID, trimmedParentMessageID)
	if err != nil {
		return err
	}
	if parentAuthorMemberID != "" && parentAuthorMemberID != trimmedAuthorMemberID {
		if err := SilenceDerivedReplyThreadParticipantFence(EnsureReplyThreadParticipantProjection(ctx, txs.tx, trimmedThreadChannelID, parentAuthorMemberID, "auto-wake", subscribedAt)); err != nil {
			return err
		}
	}

	if authorType == domain.MemberTypeHuman {
		if err := ActivateThreadSubscriptionAtSeq(ctx, txs.tx, trimmedAuthorMemberID, trimmedThreadChannelID, currentMessageSeq, subscribedAt); err != nil {
			return err
		}
	}
	rootMentionMemberIDs, err := txs.loadReplyThreadRootMentionRecipientMemberIDs(ctx, trimmedParentMessageID)
	if err != nil {
		return err
	}

	for _, memberID := range mergeReplyThreadSubscriptionMemberIDs(mentionedMemberIDs, agentMentionMemberIDs) {
		if err := SilenceDerivedReplyThreadParticipantFence(ActivateThreadSubscriptionWithInitialReadAtSeq(ctx, txs.tx, memberID, trimmedThreadChannelID, currentMessageSeq-1, currentMessageGlobalSeq-1, currentMessageSeq, subscribedAt)); err != nil {
			return err
		}
	}
	for _, memberID := range mergeReplyThreadSubscriptionMemberIDs(rootMentionMemberIDs) {
		if err := SilenceDerivedReplyThreadParticipantFence(ActivateThreadSubscriptionWithInitialReadIfNotExplicitlyUnsubscribedAtSeq(ctx, txs.tx, memberID, trimmedThreadChannelID, currentMessageSeq-1, currentMessageGlobalSeq-1, currentMessageSeq, subscribedAt)); err != nil {
			return err
		}
	}
	return nil
}

func mergeReplyThreadSubscriptionMemberIDs(groups ...[]string) []string {
	seen := make(map[string]struct{})
	merged := make([]string, 0)
	for _, group := range groups {
		for _, raw := range group {
			memberID := strings.TrimSpace(raw)
			if memberID == "" {
				continue
			}
			if _, ok := seen[memberID]; ok {
				continue
			}
			seen[memberID] = struct{}{}
			merged = append(merged, memberID)
		}
	}
	return merged
}

func (txs TxStore) EnsureReplyThreadParentAuthorParticipant(ctx context.Context, threadChannelID string, parentMessageID string, participatedAt time.Time) error {
	trimmedThreadChannelID := strings.TrimSpace(threadChannelID)
	trimmedParentMessageID := strings.TrimSpace(parentMessageID)
	if trimmedThreadChannelID == "" || trimmedParentMessageID == "" {
		return fmt.Errorf("reply thread parent participant context is incomplete")
	}

	parentAuthorMemberID, err := txs.loadReplyThreadParentAuthorMemberID(ctx, trimmedThreadChannelID, trimmedParentMessageID)
	if err != nil {
		return err
	}
	if strings.TrimSpace(parentAuthorMemberID) == "" {
		return nil
	}

	memberType, err := loadMemberTypeByID(ctx, txs.tx, parentAuthorMemberID)
	if err != nil {
		if errors.Is(err, ErrNotFound) {
			return nil
		}
		return err
	}

	switch memberType {
	case domain.MemberTypeHuman:
		return SilenceDerivedReplyThreadParticipantFence(EnsureReplyThreadParticipantProjection(ctx, txs.tx, trimmedThreadChannelID, parentAuthorMemberID, "auto-wake", participatedAt))
	default:
		return nil
	}
}

func loadMemberTypeByID(ctx context.Context, queryer coredb.Queryer, memberID string) (domain.MemberType, error) {
	row := queryer.QueryRow(ctx, `
SELECT type
FROM members
WHERE id = $1
`, strings.TrimSpace(memberID))

	var memberType string
	if err := row.Scan(&memberType); err != nil {
		if coredb.IsNotFound(err) {
			return "", ErrNotFound
		}
		return "", fmt.Errorf("load member type: %w", err)
	}
	return domain.MemberType(strings.TrimSpace(memberType)), nil
}

func (txs TxStore) loadReplyThreadRootMentionRecipientMemberIDs(ctx context.Context, parentMessageID string) ([]string, error) {
	trimmedParentMessageID := strings.TrimSpace(parentMessageID)
	if trimmedParentMessageID == "" {
		return nil, nil
	}

	var scopeID string
	var authorMemberID string
	var content string
	var rawMetadata []byte
	if err := txs.tx.QueryRow(ctx, `
SELECT scope_id, author_member_id, content, metadata
FROM scope_messages
WHERE id = $1
`, trimmedParentMessageID).Scan(&scopeID, &authorMemberID, &content, &rawMetadata); err != nil {
		if coredb.IsNotFound(err) {
			return nil, nil
		}
		return nil, fmt.Errorf("load reply thread root message mention context: %w", err)
	}

	metadata, err := DecodeMessageMetadata(rawMetadata)
	if err != nil {
		return nil, err
	}
	recipients, err := LoadScopeMentionedAgentRecipientsWithMetadata(ctx, txs.tx, scopeID, authorMemberID, content, metadata)
	if err != nil {
		return nil, err
	}
	memberIDs := make([]string, 0, len(recipients))
	for _, recipient := range recipients {
		memberIDs = append(memberIDs, recipient.AgentMemberID)
	}
	return mergeReplyThreadSubscriptionMemberIDs(memberIDs), nil
}

func (txs TxStore) loadReplyThreadParentAuthorMemberID(ctx context.Context, threadChannelID string, parentMessageID string) (string, error) {
	row := txs.tx.QueryRow(ctx, `
SELECT COALESCE(m.author_member_id, '')
FROM scopes reply_scope
LEFT JOIN scope_messages m
  ON m.id = reply_scope.root_message_id
 AND m.scope_id = reply_scope.parent_scope_id
WHERE reply_scope.id = $1
  AND reply_scope.root_message_id = $2
  AND reply_scope.type = 'thread'
  AND reply_scope.status = 'active'
`, strings.TrimSpace(threadChannelID), strings.TrimSpace(parentMessageID))

	var authorMemberID string
	if err := row.Scan(&authorMemberID); err != nil {
		if coredb.IsNotFound(err) {
			return "", nil
		}
		return "", fmt.Errorf("load parent message author for thread subscription: %w", err)
	}
	return strings.TrimSpace(authorMemberID), nil
}