package im

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
)

const (
	conversationIndexUpsertArgCount  = 10
	conversationIndexUpsertBatchSize = 200
)

// ScopeEventRequest describes one recoverable IM event produced by a scope
// write. When Conversation is set, the same write also refreshes the
// member-local conversation projection for each recipient.
type ScopeEventRequest struct {
	WorkspaceID string
	ScopeID     string
	// RecipientMemberIDs optionally narrows fanout to an immutable write-time
	// audience. Empty preserves ordinary scope-wide IM fanout. Task attention
	// events always pass the Human recipient snapshot so participation alone
	// never creates Inbox activity.
	RecipientMemberIDs []string
	// TargetedRecipients preserves the distinction between an absent audience
	// (ordinary scope-wide fanout) and an explicit empty audience (nobody).
	// Task attention always sets this flag.
	TargetedRecipients bool
	Event              StreamEvent
	Conversation       *ConversationIndexItem
}

// AppendScopeEvent fans out one event to every active member in the scope and
// updates the corresponding conversation projection rows in the same TiDB
// transaction. It is intentionally scoped to current scope_memberships rather
// than user IDs so it matches the rest of the IM/read-state schema.
func (store *Store) AppendScopeEvent(ctx context.Context, request ScopeEventRequest) ([]StreamEvent, error) {
	if err := store.Validate(); err != nil {
		return nil, err
	}
	workspaceID := normalizedString(request.WorkspaceID)
	scopeID := normalizedString(request.ScopeID)
	if workspaceID == "" {
		return nil, fmt.Errorf("workspace id is required")
	}
	if scopeID == "" {
		return nil, fmt.Errorf("scope id is required")
	}
	var appended []StreamEvent
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		recipients, err := loadScopeEventRecipientsForMembers(ctx, tx, workspaceID, scopeID, request.RecipientMemberIDs, request.TargetedRecipients)
		if err != nil {
			return err
		}
		appended, err = appendEventsForRecipients(ctx, tx, workspaceID, scopeID, request.Event, request.Conversation, recipients)
		return err
	})
	if err != nil {
		return nil, err
	}
	return appended, nil
}

// AppendMemberEvent appends one member-local event and optional conversation
// projection update. It is used for self-only mutations such as read state,
// pin/archive, notification, and local conversation actions.
func (store *Store) AppendMemberEvent(ctx context.Context, workspaceID string, memberID string, scopeID string, event StreamEvent, conversation *ConversationIndexItem) (StreamEvent, error) {
	if err := store.Validate(); err != nil {
		return StreamEvent{}, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return StreamEvent{}, err
	}
	scopeID = normalizedString(scopeID)
	if scopeID == "" {
		return StreamEvent{}, fmt.Errorf("scope id is required")
	}
	var appended []StreamEvent
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		recipient, err := loadMemberScopeEventRecipient(ctx, tx, workspaceID, scopeID, memberID)
		if err != nil && conversation != nil && coredb.IsNotFound(err) {
			// Reply-thread conversations can be visible through parent-scope
			// access and notification state before a child membership row exists.
			recipient, err = loadMemberConversationEventRecipient(ctx, tx, workspaceID, scopeID, memberID)
		}
		if err != nil {
			return err
		}
		appended, err = appendEventsForRecipients(ctx, tx, workspaceID, scopeID, event, conversation, []scopeEventRecipient{recipient})
		return err
	})
	if err != nil {
		return StreamEvent{}, err
	}
	if len(appended) == 0 {
		return StreamEvent{}, fmt.Errorf("member event was not appended")
	}
	return appended[0], nil
}

// AppendMemberWorkspaceEvent appends one member-local workspace event that is
// not tied to a single visible scope, for example a bulk inbox mutation.
func (store *Store) AppendMemberWorkspaceEvent(ctx context.Context, workspaceID string, memberID string, event StreamEvent) (StreamEvent, error) {
	if err := store.Validate(); err != nil {
		return StreamEvent{}, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return StreamEvent{}, err
	}
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		event, err = appendMemberWorkspaceEventTx(ctx, tx, workspaceID, memberID, event)
		return err
	})
	if err != nil {
		return StreamEvent{}, err
	}
	return event, nil
}

// AppendConversationBulkReadEvent appends a member-local bulk read event.
// Conversation unread badges are served by UnreadRuntime, so the IM projection
// is not an unread mutation target.
func (store *Store) AppendConversationBulkReadEvent(ctx context.Context, workspaceID string, memberID string, event StreamEvent, kinds []string) (StreamEvent, error) {
	if err := store.Validate(); err != nil {
		return StreamEvent{}, err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return StreamEvent{}, err
	}
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		event, err = appendMemberWorkspaceEventTx(ctx, tx, workspaceID, memberID, event)
		return err
	})
	if err != nil {
		return StreamEvent{}, err
	}
	return event, nil
}

func appendMemberWorkspaceEventTx(ctx context.Context, tx coredb.Tx, workspaceID string, memberID string, event StreamEvent) (StreamEvent, error) {
	event.WorkspaceID = workspaceID
	event.MemberID = memberID
	event.Type = normalizedString(event.Type)
	if event.Type == "" {
		return StreamEvent{}, fmt.Errorf("event type is required")
	}
	if event.CreatedAt.IsZero() {
		event.CreatedAt = nowUTC()
	} else {
		event.CreatedAt = event.CreatedAt.UTC()
	}
	if event.EventID == "" {
		id, err := ensureEventID("")
		if err != nil {
			return StreamEvent{}, err
		}
		event.EventID = id
	}
	nextSeq, err := nextEventSeq(ctx, tx, workspaceID, memberID)
	if err != nil {
		return StreamEvent{}, err
	}
	if event.EventSeq <= 0 {
		event.EventSeq = nextSeq
	}
	if _, err := insertStreamEvent(ctx, tx, event); err != nil {
		return StreamEvent{}, err
	}
	return event, nil
}

func normalizeConversationIndexKinds(kinds []string) []string {
	if len(kinds) == 0 {
		return nil
	}
	seen := make(map[string]struct{}, len(kinds))
	normalized := make([]string, 0, len(kinds))
	for _, kind := range kinds {
		var value string
		switch strings.ToLower(normalizedString(kind)) {
		case "channel":
			value = "channel"
		case "replythread", "thread":
			value = "thread"
		case "humandm", "agentdm", "dm":
			value = "dm"
		case "serviceaccount", "service_account":
			value = "service_account"
		case "task":
			value = "task"
		}
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		normalized = append(normalized, value)
	}
	return normalized
}

type scopeEventRecipient struct {
	MemberID        string
	ScopeKind       string
	ParentScopeType string
	SortAt          time.Time
	PinnedAt        *time.Time
	ArchivedAt      *time.Time
	TaskWatchState  string
}

type pendingScopeEvent struct {
	event                      StreamEvent
	recipient                  scopeEventRecipient
	skipConversationProjection bool
}

type memberEventSeq struct {
	memberID string
	eventSeq int64
}

func loadScopeEventRecipients(ctx context.Context, queryer coredb.Queryer, workspaceID string, scopeID string) ([]scopeEventRecipient, error) {
	rows, err := queryer.Query(ctx, scopeEventRecipientSQL(``), workspaceID, scopeID)
	if err != nil {
		return nil, fmt.Errorf("load im scope event recipients: %w", err)
	}
	defer rows.Close()

	recipients := make([]scopeEventRecipient, 0)
	for rows.Next() {
		recipient, err := scanScopeEventRecipient(rows)
		if err != nil {
			return nil, err
		}
		recipients = append(recipients, recipient)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate im scope event recipients: %w", err)
	}
	return recipients, nil
}

func loadScopeEventRecipientsForMembers(ctx context.Context, queryer coredb.Queryer, workspaceID string, scopeID string, memberIDs []string, targeted bool) ([]scopeEventRecipient, error) {
	normalizedMemberIDs := normalizeMemberIDs(memberIDs)
	if len(normalizedMemberIDs) == 0 {
		if targeted {
			return []scopeEventRecipient{}, nil
		}
		return loadScopeEventRecipients(ctx, queryer, workspaceID, scopeID)
	}
	requestedMembersCTE, memberArgs, _, err := coredb.BuildStringValuesCTE("requested_members", "member_id", normalizedMemberIDs, 3)
	if err != nil {
		return nil, err
	}
	rows, err := queryer.Query(ctx, fmt.Sprintf(`
WITH %s
%s
`, requestedMembersCTE, scopeEventRecipientSQL(`  AND EXISTS (
    SELECT 1
    FROM requested_members
    WHERE requested_members.member_id = sm.member_id
  )
`)), append([]any{workspaceID, scopeID}, memberArgs...)...)
	if err != nil {
		return nil, fmt.Errorf("load targeted im scope event recipients: %w", err)
	}
	defer rows.Close()

	recipients := make([]scopeEventRecipient, 0, len(normalizedMemberIDs))
	for rows.Next() {
		recipient, err := scanScopeEventRecipient(rows)
		if err != nil {
			return nil, err
		}
		recipients = append(recipients, recipient)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate targeted im scope event recipients: %w", err)
	}
	return recipients, nil
}

func normalizeMemberIDs(values []string) []string {
	if len(values) == 0 {
		return nil
	}
	normalized := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		value = normalizedString(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		normalized = append(normalized, value)
	}
	return normalized
}

func loadMemberScopeEventRecipient(ctx context.Context, queryer coredb.Queryer, workspaceID string, scopeID string, memberID string) (scopeEventRecipient, error) {
	row := queryer.QueryRow(ctx, scopeEventRecipientSQL(`  AND sm.member_id = $3
`), workspaceID, scopeID, memberID)
	recipient, err := scanScopeEventRecipient(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return scopeEventRecipient{}, fmt.Errorf("member is not active in scope: %w", err)
		}
		return scopeEventRecipient{}, err
	}
	return recipient, nil
}

func loadMemberConversationEventRecipient(ctx context.Context, queryer coredb.Queryer, workspaceID string, scopeID string, memberID string) (scopeEventRecipient, error) {
	row := queryer.QueryRow(ctx, `
SELECT m.id,
       s.type,
       CASE
           WHEN parent_cp.subsystem = 'kb' THEN 'kb'
           ELSE CASE WHEN parent_scope.type IS NULL THEN '' ELSE parent_scope.type END
       END AS parent_scope_type,
       COALESCE(s.visible_last_message_at, s.latest_message_at, s.updated_at, s.created_at) AS sort_at,
       cms.pinned_at,
       CASE
           WHEN cms.hidden_at IS NOT NULL
            AND COALESCE(s.latest_seq, 0) <= COALESCE(cms.hidden_through_seq, 0)
           THEN cms.hidden_at
           ELSE NULL
       END AS archived_at,
       COALESCE(sns.task_watch_state, '') AS task_watch_state
FROM scopes s
JOIN members m
  ON m.id = $3
 AND m.workspace_id = s.workspace_id
 AND m.status = 'active'
LEFT JOIN conversation_member_states cms
  ON cms.scope_id = s.id
 AND cms.member_id = m.id
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = s.id
 AND sns.member_id = m.id
LEFT JOIN scopes parent_scope
  ON parent_scope.id = s.parent_scope_id
 AND parent_scope.workspace_id = s.workspace_id
LEFT JOIN scope_channel_profiles parent_cp
  ON parent_cp.scope_id = parent_scope.id
 AND parent_cp.workspace_id = s.workspace_id
WHERE s.workspace_id = $1
  AND s.id = $2
  AND s.status = 'active'
  AND s.archived_at IS NULL
  AND s.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
`, workspaceID, scopeID, memberID)
	recipient, err := scanScopeEventRecipient(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return scopeEventRecipient{}, fmt.Errorf("member conversation scope is not available: %w", err)
		}
		return scopeEventRecipient{}, err
	}
	return recipient, nil
}

func scopeEventRecipientSQL(extraPredicate string) string {
	return `
SELECT sm.member_id,
       s.type,
       CASE
           WHEN parent_cp.subsystem = 'kb' THEN 'kb'
           ELSE CASE WHEN parent_scope.type IS NULL THEN '' ELSE parent_scope.type END
       END AS parent_scope_type,
       COALESCE(s.visible_last_message_at, s.latest_message_at, s.updated_at, s.created_at) AS sort_at,
       cms.pinned_at,
       CASE
           WHEN cms.hidden_at IS NOT NULL
            AND COALESCE(s.latest_seq, 0) <= COALESCE(cms.hidden_through_seq, 0)
           THEN cms.hidden_at
           ELSE NULL
       END AS archived_at,
       COALESCE(sns.task_watch_state, '') AS task_watch_state
FROM scopes s
JOIN scope_memberships sm
  ON sm.scope_id = s.id
 AND sm.removed_at IS NULL
 AND sm.archived_at IS NULL
 AND sm.effective_to_seq IS NULL
JOIN members m
  ON m.id = sm.member_id
 AND m.workspace_id = s.workspace_id
 AND m.status = 'active'
LEFT JOIN conversation_member_states cms
  ON cms.scope_id = sm.scope_id
 AND cms.member_id = sm.member_id
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = sm.scope_id
 AND sns.member_id = sm.member_id
LEFT JOIN scopes parent_scope
  ON parent_scope.id = s.parent_scope_id
 AND parent_scope.workspace_id = s.workspace_id
LEFT JOIN scope_channel_profiles parent_cp
  ON parent_cp.scope_id = parent_scope.id
 AND parent_cp.workspace_id = s.workspace_id
WHERE s.workspace_id = $1
  AND s.id = $2
  AND s.status = 'active'
  AND s.archived_at IS NULL
  AND s.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
` + extraPredicate + `ORDER BY sm.member_id ASC
`
}

func scanScopeEventRecipient(row interface{ Scan(...any) error }) (scopeEventRecipient, error) {
	var recipient scopeEventRecipient
	var sortAt sql.NullTime
	var pinnedAt sql.NullTime
	var archivedAt sql.NullTime
	if err := row.Scan(
		&recipient.MemberID,
		&recipient.ScopeKind,
		&recipient.ParentScopeType,
		&sortAt,
		&pinnedAt,
		&archivedAt,
		&recipient.TaskWatchState,
	); err != nil {
		return scopeEventRecipient{}, fmt.Errorf("scan im scope event recipient: %w", err)
	}
	if sortAt.Valid {
		recipient.SortAt = sortAt.Time.UTC()
	}
	if pinnedAt.Valid {
		value := pinnedAt.Time.UTC()
		recipient.PinnedAt = &value
	}
	if archivedAt.Valid {
		value := archivedAt.Time.UTC()
		recipient.ArchivedAt = &value
	}
	return recipient, nil
}

func appendEventsForRecipients(ctx context.Context, tx coredb.Tx, workspaceID string, scopeID string, baseEvent StreamEvent, conversation *ConversationIndexItem, recipients []scopeEventRecipient) ([]StreamEvent, error) {
	if len(recipients) == 0 {
		return nil, nil
	}
	workspaceID = normalizedString(workspaceID)
	scopeID = normalizedString(scopeID)
	baseEvent.WorkspaceID = workspaceID
	baseEvent.ScopeID = firstNonEmptyString(baseEvent.ScopeID, scopeID)
	baseEvent.Type = normalizedString(baseEvent.Type)
	if baseEvent.Type == "" {
		return nil, fmt.Errorf("event type is required")
	}
	if baseEvent.CreatedAt.IsZero() {
		baseEvent.CreatedAt = nowUTC()
	} else {
		baseEvent.CreatedAt = baseEvent.CreatedAt.UTC()
	}
	if baseEvent.EventID == "" {
		id, err := ensureEventID("")
		if err != nil {
			return nil, err
		}
		baseEvent.EventID = id
	}

	nextSeqByMember, err := nextEventSeqs(ctx, tx, workspaceID, memberIDsForScopeEventRecipients(recipients))
	if err != nil {
		return nil, err
	}
	if conversation != nil || taskConversationActionNeedsRecipientRefresh(baseEvent, recipients) {
		if taskLifecycleProtectedRecipientRefresh(ctx, workspaceID, scopeID, recipients) {
			recipients, err = refreshProtectedTaskScopeEventRecipientsForProjection(ctx, tx, workspaceID, scopeID, recipients)
		} else {
			recipients, err = refreshScopeEventRecipientsForProjection(ctx, tx, workspaceID, scopeID, recipients)
		}
		if err != nil {
			return nil, err
		}
	}

	appended := make([]StreamEvent, 0, len(recipients))
	events := make([]StreamEvent, 0, len(recipients))
	pending := make([]pendingScopeEvent, 0, len(recipients))
	for _, recipient := range recipients {
		event := baseEvent
		event.WorkspaceID = workspaceID
		event.MemberID = recipient.MemberID
		var skipConversationProjection bool
		event, skipConversationProjection, err = reconcileTaskConversationActionEvent(event, recipient)
		if err != nil {
			return nil, err
		}
		if event.ConversationID == "" && conversation != nil {
			event.ConversationID = normalizedString(conversation.ConversationID)
		}
		if event.ConversationID == "" {
			event.ConversationID = conversationIDForScopeKind(recipient.ScopeKind, scopeID)
		}
		if event.EventSeq <= 0 {
			nextSeq := nextSeqByMember[recipient.MemberID]
			if nextSeq <= 0 {
				nextSeq = 1
			}
			event.EventSeq = nextSeq
			nextSeqByMember[recipient.MemberID] = nextSeq + 1
		}
		events = append(events, event)
		pending = append(pending, pendingScopeEvent{
			event:                      event,
			recipient:                  recipient,
			skipConversationProjection: skipConversationProjection,
		})
	}
	insertedEvents, err := insertStreamEvents(ctx, tx, events)
	if err != nil {
		return nil, err
	}
	pendingByEvent := make(map[memberEventSeq]pendingScopeEvent, len(pending))
	for _, item := range pending {
		pendingByEvent[memberEventSeq{memberID: item.event.MemberID, eventSeq: item.event.EventSeq}] = item
	}
	conversations := make([]ConversationIndexItem, 0, len(insertedEvents))
	for _, event := range insertedEvents {
		item, ok := pendingByEvent[memberEventSeq{memberID: event.MemberID, eventSeq: event.EventSeq}]
		if !ok {
			return nil, fmt.Errorf("inserted im stream event recipient not found")
		}
		if conversation != nil && !item.skipConversationProjection {
			conversations = append(conversations, conversationIndexItemForRecipient(*conversation, event, item.recipient, scopeID))
		}
		appended = append(appended, event)
	}
	if err := upsertConversationsTx(ctx, tx, workspaceID, conversations); err != nil {
		return nil, err
	}
	return appended, nil
}

func memberIDsForScopeEventRecipients(recipients []scopeEventRecipient) []string {
	memberIDs := make([]string, 0, len(recipients))
	for _, recipient := range recipients {
		memberIDs = append(memberIDs, recipient.MemberID)
	}
	return memberIDs
}

func taskLifecycleProtectedRecipientRefresh(
	ctx context.Context,
	workspaceID string,
	scopeID string,
	recipients []scopeEventRecipient,
) bool {
	if !taskLifecycleScopeIsProtected(ctx, workspaceID, scopeID) || len(recipients) == 0 {
		return false
	}
	// The context marker alone is intentionally insufficient. It is an
	// optimization/correctness capability granted only to Task projections;
	// malformed or mixed recipient data must retain the ordinary joined
	// locking read rather than accidentally weakening another scope kind.
	for _, recipient := range recipients {
		if normalizedString(recipient.ScopeKind) != "task" {
			return false
		}
	}
	return true
}

type taskScopeProjectionMemberState struct {
	pinnedAt                 *time.Time
	hiddenAt                 *time.Time
	hiddenThroughSeq         int64
	hiddenThroughActivitySeq int64
	taskWatchState           string
}

type taskScopeProjectionScopeState struct {
	scopeKind         string
	parentScopeType   string
	sortAt            time.Time
	latestSeq         int64
	latestActivitySeq int64
}

// refreshProtectedTaskScopeEventRecipientsForProjection is the inner half of
// the Task lifecycle hybrid lock protocol. The outer guard already holds the
// owning Task rows, so taking a locking read on scopes here would reintroduce a
// task -> scope edge and deadlock with rolling, legacy scope -> task discussion
// writers.
//
// Member streams have already been locked by nextEventSeqs. We current-lock
// exact CMS and SNS primary keys in deterministic order and retain the values
// returned by those locking reads. TiDB RR ordinary reads may use the
// transaction start snapshot even after a blocking current read, so it is not
// correct to lock these rows and then re-read them through a plain join. Only
// immutable/lifecycle-protected scope projection fields are read ordinarily.
//
// A missing CMS/SNS row is represented by defaults. We deliberately do not
// assume a missing-key SELECT FOR UPDATE prevents a concurrent insert: an
// action that inserts after this sample must append its self event after our
// already-held member stream row and thereby repair the final projection.
func refreshProtectedTaskScopeEventRecipientsForProjection(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID string,
	scopeID string,
	recipients []scopeEventRecipient,
) ([]scopeEventRecipient, error) {
	memberIDs := normalizeMemberIDs(memberIDsForScopeEventRecipients(recipients))
	if len(memberIDs) == 0 {
		return nil, nil
	}
	sort.Strings(memberIDs)

	memberStateByID := make(map[string]taskScopeProjectionMemberState, len(memberIDs))
	for _, memberID := range memberIDs {
		state, err := currentLockTaskScopeConversationMemberState(ctx, tx, scopeID, memberID)
		if err != nil {
			return nil, err
		}
		memberStateByID[memberID] = state
	}
	for _, memberID := range memberIDs {
		state := memberStateByID[memberID]
		taskWatchState, err := currentLockTaskScopeNotificationState(ctx, tx, scopeID, memberID)
		if err != nil {
			return nil, err
		}
		state.taskWatchState = taskWatchState
		memberStateByID[memberID] = state
	}

	scopeState, err := loadProtectedTaskScopeProjectionState(ctx, tx, workspaceID, scopeID)
	if err != nil {
		return nil, err
	}
	if normalizedString(scopeState.scopeKind) != "task" {
		return nil, fmt.Errorf("protected task conversation scope is not a Task")
	}

	refreshed := make([]scopeEventRecipient, 0, len(recipients))
	for _, recipient := range recipients {
		memberState, ok := memberStateByID[recipient.MemberID]
		if !ok {
			return nil, fmt.Errorf("protected task projection member state is not available for %q", recipient.MemberID)
		}
		current := scopeEventRecipient{
			MemberID:        recipient.MemberID,
			ScopeKind:       scopeState.scopeKind,
			ParentScopeType: scopeState.parentScopeType,
			SortAt:          scopeState.sortAt,
			PinnedAt:        memberState.pinnedAt,
			TaskWatchState:  memberState.taskWatchState,
		}
		if memberState.hiddenAt != nil &&
			scopeState.latestSeq <= memberState.hiddenThroughSeq {
			current.ArchivedAt = memberState.hiddenAt
		}
		refreshed = append(refreshed, current)
	}
	return refreshed, nil
}

func currentLockTaskScopeConversationMemberState(
	ctx context.Context,
	queryer coredb.QueryRower,
	scopeID string,
	memberID string,
) (taskScopeProjectionMemberState, error) {
	var pinnedAt sql.NullTime
	var hiddenAt sql.NullTime
	var hiddenThroughSeq sql.NullInt64
	var hiddenThroughActivitySeq int64
	err := queryer.QueryRow(ctx, `
SELECT pinned_at,
       hidden_at,
       hidden_through_seq,
       hidden_through_activity_seq
FROM conversation_member_states
WHERE scope_id = $1
  AND member_id = $2
FOR UPDATE
`, scopeID, memberID).Scan(
		&pinnedAt,
		&hiddenAt,
		&hiddenThroughSeq,
		&hiddenThroughActivitySeq,
	)
	if errors.Is(err, sql.ErrNoRows) || coredb.IsNotFound(err) {
		return taskScopeProjectionMemberState{}, nil
	}
	if err != nil {
		return taskScopeProjectionMemberState{}, fmt.Errorf(
			"lock task conversation member state for member %q: %w",
			memberID,
			err,
		)
	}

	state := taskScopeProjectionMemberState{
		hiddenThroughActivitySeq: hiddenThroughActivitySeq,
	}
	if pinnedAt.Valid {
		value := pinnedAt.Time.UTC()
		state.pinnedAt = &value
	}
	if hiddenAt.Valid {
		value := hiddenAt.Time.UTC()
		state.hiddenAt = &value
	}
	if hiddenThroughSeq.Valid {
		state.hiddenThroughSeq = hiddenThroughSeq.Int64
	}
	return state, nil
}

func currentLockTaskScopeNotificationState(
	ctx context.Context,
	queryer coredb.QueryRower,
	scopeID string,
	memberID string,
) (string, error) {
	var taskWatchState string
	err := queryer.QueryRow(ctx, `
SELECT COALESCE(task_watch_state, '')
FROM scope_notification_settings
WHERE scope_id = $1
  AND member_id = $2
FOR UPDATE
`, scopeID, memberID).Scan(&taskWatchState)
	if errors.Is(err, sql.ErrNoRows) || coredb.IsNotFound(err) {
		return "", nil
	}
	if err != nil {
		return "", fmt.Errorf(
			"lock task notification state for member %q: %w",
			memberID,
			err,
		)
	}
	return normalizedString(taskWatchState), nil
}

func loadProtectedTaskScopeProjectionState(
	ctx context.Context,
	queryer coredb.QueryRower,
	workspaceID string,
	scopeID string,
) (taskScopeProjectionScopeState, error) {
	var state taskScopeProjectionScopeState
	var sortAt sql.NullTime
	err := queryer.QueryRow(ctx, `
SELECT s.type,
       CASE
           WHEN parent_cp.subsystem = 'kb' THEN 'kb'
           ELSE CASE WHEN parent_scope.type IS NULL THEN '' ELSE parent_scope.type END
       END AS parent_scope_type,
       COALESCE(s.visible_last_message_at, s.latest_message_at, s.updated_at, s.created_at) AS sort_at,
       COALESCE(s.latest_seq, 0),
       COALESCE(s.latest_activity_seq, 0)
FROM scopes s
LEFT JOIN scopes parent_scope
  ON parent_scope.id = s.parent_scope_id
 AND parent_scope.workspace_id = s.workspace_id
LEFT JOIN scope_channel_profiles parent_cp
  ON parent_cp.scope_id = parent_scope.id
 AND parent_cp.workspace_id = s.workspace_id
WHERE s.workspace_id = $1
  AND s.id = $2
  AND s.status = 'active'
  AND s.archived_at IS NULL
`, workspaceID, scopeID).Scan(
		&state.scopeKind,
		&state.parentScopeType,
		&sortAt,
		&state.latestSeq,
		&state.latestActivitySeq,
	)
	if errors.Is(err, sql.ErrNoRows) || coredb.IsNotFound(err) {
		return taskScopeProjectionScopeState{}, fmt.Errorf(
			"protected task conversation scope is not available: %w",
			sql.ErrNoRows,
		)
	}
	if err != nil {
		return taskScopeProjectionScopeState{}, fmt.Errorf("read protected task conversation scope: %w", err)
	}
	state.scopeKind = normalizedString(state.scopeKind)
	state.parentScopeType = normalizedString(state.parentScopeType)
	if sortAt.Valid {
		state.sortAt = sortAt.Time.UTC()
	}
	return state, nil
}

// refreshScopeEventRecipientsForProjection re-reads mutable member projection
// state after the member stream rows have been locked. This gives every
// projection write, plus Task self actions that have no materialized Inbox row
// yet, the same stream -> member state -> conversation index lock order. A
// source event sampled before a concurrent hide/watch/pin mutation therefore
// cannot publish or persist the superseded member-local value.
func refreshScopeEventRecipientsForProjection(ctx context.Context, queryer coredb.Queryer, workspaceID string, scopeID string, recipients []scopeEventRecipient) ([]scopeEventRecipient, error) {
	memberIDs := memberIDsForScopeEventRecipients(recipients)
	if len(memberIDs) == 0 {
		return nil, nil
	}
	args := make([]any, 0, len(memberIDs)+2)
	args = append(args, workspaceID, scopeID)
	// The initial recipient load already established active membership/member
	// access. Re-materialize only those IDs here so the current read locks the
	// single scope row and matching CMS rows, not every member row in a large
	// channel fanout.
	requestedMemberSelects := make([]string, 0, len(memberIDs))
	for index, memberID := range memberIDs {
		args = append(args, memberID)
		selectMember := fmt.Sprintf("SELECT $%d", len(args))
		if index == 0 {
			selectMember += " AS member_id"
		}
		requestedMemberSelects = append(requestedMemberSelects, selectMember)
	}
	rows, err := queryer.Query(ctx, fmt.Sprintf(`
SELECT requested.member_id,
       s.type,
       CASE
           WHEN parent_cp.subsystem = 'kb' THEN 'kb'
           ELSE CASE WHEN parent_scope.type IS NULL THEN '' ELSE parent_scope.type END
       END AS parent_scope_type,
       COALESCE(s.visible_last_message_at, s.latest_message_at, s.updated_at, s.created_at) AS sort_at,
       cms.pinned_at,
       CASE
           WHEN cms.hidden_at IS NOT NULL
            AND COALESCE(s.latest_seq, 0) <= COALESCE(cms.hidden_through_seq, 0)
           THEN cms.hidden_at
           ELSE NULL
       END AS archived_at,
       COALESCE(sns.task_watch_state, '') AS task_watch_state
FROM scopes s
JOIN (
  %s
) requested ON TRUE
LEFT JOIN conversation_member_states cms
  ON cms.scope_id = s.id
 AND cms.member_id = requested.member_id
LEFT JOIN scope_notification_settings sns
  ON sns.scope_id = s.id
 AND sns.member_id = requested.member_id
LEFT JOIN scopes parent_scope
  ON parent_scope.id = s.parent_scope_id
 AND parent_scope.workspace_id = s.workspace_id
LEFT JOIN scope_channel_profiles parent_cp
  ON parent_cp.scope_id = parent_scope.id
 AND parent_cp.workspace_id = s.workspace_id
WHERE s.workspace_id = $1
  AND s.id = $2
  AND s.status = 'active'
  AND s.archived_at IS NULL
ORDER BY requested.member_id ASC
FOR UPDATE
`, strings.Join(requestedMemberSelects, "\n  UNION ALL\n  ")), args...)
	if err != nil {
		return nil, fmt.Errorf("refresh im scope event recipients: %w", err)
	}
	defer rows.Close()

	refreshedByMember := make(map[string]scopeEventRecipient, len(recipients))
	for rows.Next() {
		recipient, err := scanScopeEventRecipient(rows)
		if err != nil {
			return nil, err
		}
		refreshedByMember[recipient.MemberID] = recipient
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate refreshed im scope event recipients: %w", err)
	}

	refreshed := make([]scopeEventRecipient, 0, len(recipients))
	for _, recipient := range recipients {
		current, ok := refreshedByMember[recipient.MemberID]
		if !ok {
			return nil, fmt.Errorf("member conversation scope is not available: %w", sql.ErrNoRows)
		}
		refreshed = append(refreshed, current)
	}
	return refreshed, nil
}

func conversationIndexItemForRecipient(base ConversationIndexItem, event StreamEvent, recipient scopeEventRecipient, scopeID string) ConversationIndexItem {
	item := base
	item.WorkspaceID = event.WorkspaceID
	item.MemberID = recipient.MemberID
	item.ScopeID = firstNonEmptyString(item.ScopeID, scopeID)
	item.Kind = firstNonEmptyString(item.Kind, recipient.ScopeKind)
	item.ConversationID = firstNonEmptyString(item.ConversationID, event.ConversationID, conversationIDForScopeKind(item.Kind, item.ScopeID))
	if item.SortAt.IsZero() {
		item.SortAt = firstNonZeroTime(event.CreatedAt, recipient.SortAt, nowUTC())
	} else {
		item.SortAt = item.SortAt.UTC()
	}
	item.LastEventSeq = event.EventSeq
	// Mutable member-local state is re-read under the member stream lock in the
	// append transaction. It is newer than any action projection loaded before
	// this append, so it must win even when the stale projection explicitly
	// carried nil (unpin) or non-nil (hide) values.
	item.PinnedAt = recipient.PinnedAt
	item.PinnedAtSet = true
	item.ArchivedAt = recipient.ArchivedAt
	item.ParentScopeType = normalizedString(recipient.ParentScopeType)
	item.ParentScopeTypeSet = true
	if item.Kind == "task" {
		item.Projection = reconcileTaskConversationProjection(item.Projection, recipient)
	}
	item.Projection = mergeConversationProjection(item.Projection, item)
	return item
}

func taskConversationActionNeedsRecipientRefresh(event StreamEvent, recipients []scopeEventRecipient) bool {
	action, _ := event.Payload["action"].(string)
	switch normalizedString(action) {
	case "mute", "unmute", "pin", "unpin", "hide", "mark_read", "mark_unread":
	default:
		return false
	}
	for _, recipient := range recipients {
		if normalizedString(recipient.ScopeKind) == "task" {
			return true
		}
	}
	return false
}

// reconcileTaskConversationActionEvent projects current member-local Task
// policy state into an action event while the member stream and the mutable
// CMS/notification rows are locked in one append transaction. Task actions
// commit their source mutation before appending the durable self event, so a
// later mutation may already have superseded the action's preloaded payload.
// Publishing that stale payload would make stream order disagree with database
// truth even though the conversation index itself can be repaired.
func reconcileTaskConversationActionEvent(event StreamEvent, recipient scopeEventRecipient) (StreamEvent, bool, error) {
	if normalizedString(recipient.ScopeKind) != "task" || len(event.Payload) == 0 {
		return event, false, nil
	}
	action, _ := event.Payload["action"].(string)
	action = normalizedString(action)
	switch action {
	case "mute", "unmute", "pin", "unpin", "hide", "mark_read", "mark_unread":
	default:
		return event, false, nil
	}

	payload := cloneAnyMap(event.Payload)
	conversation, err := model.ConvertToJSONMap(payload["conversation"])
	if err != nil {
		return StreamEvent{}, false, fmt.Errorf("decode Task conversation action payload: %w", err)
	}
	skipConversationProjection := false

	// Every Task action payload that carries a conversation should expose the
	// current member-local overlays, not the row sampled by the API handler.
	if conversation != nil {
		subscriptionState := frontendTaskWatchState(recipient.TaskWatchState)
		conversation["subscriptionState"] = subscriptionState
		if recipient.PinnedAt == nil {
			conversation["pinnedAt"] = nil
		} else {
			conversation["pinnedAt"] = recipient.PinnedAt.UTC().Format(time.RFC3339Nano)
		}
		conversation["archived"] = recipient.ArchivedAt != nil
		conversation["sidebarHidden"] = recipient.ArchivedAt != nil
		payload["conversation"] = conversation
	}

	switch action {
	case "mute", "unmute":
		state := frontendTaskWatchState(recipient.TaskWatchState)
		payload["subscriptionState"] = state
		if state == "active" {
			payload["action"] = "unmute"
		} else {
			payload["action"] = "mute"
		}
	case "pin", "unpin":
		if recipient.PinnedAt == nil {
			payload["action"] = "unpin"
			payload["pinnedAt"] = nil
		} else {
			payload["action"] = "pin"
			payload["pinnedAt"] = recipient.PinnedAt.UTC().Format(time.RFC3339Nano)
		}
	case "hide":
		if recipient.ArchivedAt == nil {
			// A later Task message has already crossed the hidden message
			// frontier. Keep a durable stream checkpoint for this completed
			// request, but make it an identity-only no-op so a stale archive
			// payload cannot remove the now-current Inbox row. Activity-only
			// reopening stays disabled during the expand-only release.
			event.Type = "conversation.updated"
			payload = taskConversationActionIdentityPayload(payload)
			skipConversationProjection = true
		}
	}
	event.Payload = payload
	return event, skipConversationProjection, nil
}

func frontendTaskWatchState(value string) string {
	if normalizedString(value) == "active" {
		return "active"
	}
	return "mute"
}

func reconcileTaskConversationProjection(projection map[string]any, recipient scopeEventRecipient) map[string]any {
	current := cloneAnyMap(projection)
	current["subscriptionState"] = frontendTaskWatchState(recipient.TaskWatchState)
	current["archived"] = recipient.ArchivedAt != nil
	current["sidebarHidden"] = recipient.ArchivedAt != nil
	if recipient.PinnedAt == nil {
		delete(current, "pinnedAt")
	} else {
		current["pinnedAt"] = recipient.PinnedAt.UTC()
	}
	return current
}

func cloneAnyMap(value map[string]any) map[string]any {
	cloned := make(map[string]any, len(value))
	for key, item := range value {
		cloned[key] = item
	}
	return cloned
}

func taskConversationActionIdentityPayload(payload map[string]any) map[string]any {
	identity := make(map[string]any, 3)
	for _, key := range []string{"kind", "scopeId", "taskId"} {
		if value, ok := payload[key]; ok {
			identity[key] = value
		}
	}
	return identity
}

// UpsertConversationTx is the transaction-capable form of
// Store.UpsertConversation: it upserts one owner-local im_conversation_index row
// against an arbitrary QueryExecer (a coredb.Tx or the shared DB). It is
// idempotent and out-of-order safe: the underlying upsert keeps the greatest
// sort_at and never lets an older projection overwrite a newer one.
func UpsertConversationTx(ctx context.Context, execer coredb.QueryExecer, workspaceID string, memberID string, item ConversationIndexItem) error {
	return upsertConversationTx(ctx, execer, workspaceID, memberID, item)
}

// InsertConversationIfMissingTx inserts one owner-local im_conversation_index row
// ONLY if no row exists for the member's (workspace_id, member_id,
// conversation_id) primary key. An existing row — including its projection_json,
// sort_at, last_event_seq, pinned_at, archived_at, created_at, and any member
// overlay — is preserved untouched. This is the canonical Task Watch Inbox
// materialization semantic: Watch only creates the missing task:<scope> row and
// must never regress an existing conversation's history/pin/hide facts.
func InsertConversationIfMissingTx(ctx context.Context, execer coredb.QueryExecer, workspaceID string, memberID string, item ConversationIndexItem) error {
	if err := validateConversationIndexItem(item); err != nil {
		return err
	}
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return err
	}
	args, err := conversationIndexUpsertArgs(workspaceID, memberID, item)
	if err != nil {
		return err
	}
	_, err = execer.Exec(ctx, `
INSERT INTO im_conversation_index (
  workspace_id, member_id, conversation_id, scope_id, kind, sort_at,
  last_event_seq, pinned_at, archived_at,
  projection_json, created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON DUPLICATE KEY UPDATE
  scope_id = im_conversation_index.scope_id,
  kind = im_conversation_index.kind,
  sort_at = im_conversation_index.sort_at,
  last_event_seq = im_conversation_index.last_event_seq,
  pinned_at = im_conversation_index.pinned_at,
  archived_at = im_conversation_index.archived_at,
  projection_json = im_conversation_index.projection_json,
  updated_at = im_conversation_index.updated_at
`, args...)
	if err != nil {
		return fmt.Errorf("insert im conversation index if missing: %w", err)
	}
	return nil
}

func validateConversationIndexItem(item ConversationIndexItem) error {
	if strings.TrimSpace(item.ConversationID) == "" {
		return fmt.Errorf("conversation id is required")
	}
	if strings.TrimSpace(item.Kind) == "" {
		return fmt.Errorf("conversation kind is required")
	}
	return nil
}

func upsertConversationTx(ctx context.Context, execer coredb.QueryExecer, workspaceID string, memberID string, item ConversationIndexItem) error {
	item.MemberID = memberID
	return upsertConversationsTx(ctx, execer, workspaceID, []ConversationIndexItem{item})
}

func upsertConversationsTx(ctx context.Context, execer coredb.QueryExecer, workspaceID string, items []ConversationIndexItem) error {
	for start := 0; start < len(items); start += conversationIndexUpsertBatchSize {
		end := start + conversationIndexUpsertBatchSize
		if end > len(items) {
			end = len(items)
		}
		if err := upsertConversationBatchTx(ctx, execer, workspaceID, items[start:end]); err != nil {
			return err
		}
	}
	return nil
}

func upsertConversationBatchTx(ctx context.Context, execer coredb.QueryExecer, workspaceID string, items []ConversationIndexItem) error {
	if len(items) == 0 {
		return nil
	}
	args := make([]any, 0, len(items)*conversationIndexUpsertArgCount)
	values := make([]string, 0, len(items))
	for _, item := range items {
		itemArgs, err := conversationIndexUpsertArgs(workspaceID, item.MemberID, item)
		if err != nil {
			return err
		}
		args = append(args, itemArgs...)
		values = append(values, conversationIndexUpsertPlaceholders(len(args)-conversationIndexUpsertArgCount+1))
	}
	_, err := execer.Exec(ctx, fmt.Sprintf(`
INSERT INTO im_conversation_index (
  workspace_id, member_id, conversation_id, scope_id, kind, sort_at,
  last_event_seq, pinned_at, archived_at,
  projection_json, created_at, updated_at
) VALUES %s
ON DUPLICATE KEY UPDATE
  scope_id = VALUES(scope_id),
  kind = VALUES(kind),
  sort_at = GREATEST(sort_at, VALUES(sort_at)),
  last_event_seq = GREATEST(last_event_seq, VALUES(last_event_seq)),
  pinned_at = VALUES(pinned_at),
  archived_at = VALUES(archived_at),
  projection_json = CASE
    WHEN VALUES(sort_at) < sort_at THEN projection_json
    ELSE VALUES(projection_json)
  END,
  updated_at = CURRENT_TIMESTAMP
`, strings.Join(values, ", ")), args...)
	if err != nil {
		return fmt.Errorf("upsert im conversation index: %w", err)
	}
	return nil
}

func conversationIndexUpsertPlaceholders(start int) string {
	placeholders := make([]string, 0, conversationIndexUpsertArgCount+2)
	for idx := 0; idx < conversationIndexUpsertArgCount; idx++ {
		placeholders = append(placeholders, fmt.Sprintf("$%d", start+idx))
	}
	placeholders = append(placeholders, "CURRENT_TIMESTAMP", "CURRENT_TIMESTAMP")
	return "(" + strings.Join(placeholders, ", ") + ")"
}

func conversationIndexUpsertArgs(workspaceID string, memberID string, item ConversationIndexItem) ([]any, error) {
	workspaceID, memberID, err := requireWorkspaceMember(workspaceID, memberID)
	if err != nil {
		return nil, err
	}
	conversationID := normalizedString(item.ConversationID)
	if conversationID == "" {
		return nil, fmt.Errorf("conversation id is required")
	}
	kind := normalizedString(item.Kind)
	switch kind {
	case "channel", "dm", "thread", "task", "service_account":
		// Ordinary Inbox projection kinds.
	default:
		return nil, fmt.Errorf("unsupported conversation kind %q", kind)
	}
	sortAt := item.SortAt.UTC()
	if sortAt.IsZero() {
		sortAt = nowUTC()
	}
	item.ConversationID = conversationID
	item.ScopeID = normalizedString(item.ScopeID)
	item.Kind = kind
	item.SortAt = sortAt
	if item.PinnedAt != nil {
		normalizedPinnedAt := item.PinnedAt.UTC()
		item.PinnedAt = &normalizedPinnedAt
	}
	item.Projection = mergeConversationProjection(item.Projection, item)
	projection, err := encodeJSONMap(item.Projection, true)
	if err != nil {
		return nil, err
	}
	return []any{
		workspaceID,
		memberID,
		conversationID,
		normalizedString(item.ScopeID),
		kind,
		sortAt,
		item.LastEventSeq,
		item.PinnedAt,
		item.ArchivedAt,
		projection,
	}, nil
}

func conversationIDForScopeKind(kind string, scopeID string) string {
	kind = normalizedString(kind)
	scopeID = normalizedString(scopeID)
	if scopeID == "" {
		return ""
	}
	switch kind {
	case "channel":
		return "channel:" + scopeID
	case "thread":
		return "thread:" + scopeID
	case "dm":
		return "dm:" + scopeID
	case "task":
		return "task:" + scopeID
	case "service_account":
		return "serviceAccount:" + scopeID
	default:
		return scopeID
	}
}

func mergeConversationProjection(projection map[string]any, item ConversationIndexItem) map[string]any {
	merged := make(map[string]any, len(projection)+8)
	for key, value := range projection {
		merged[key] = value
	}
	delete(merged, "unreadCount")
	delete(merged, "hasUnread")
	delete(merged, "manualUnread")
	delete(merged, "pinnedAt")
	merged["id"] = item.ConversationID
	merged["scopeId"] = item.ScopeID
	merged["kind"] = item.Kind
	merged["sortAt"] = item.SortAt
	merged["lastEventSeq"] = item.LastEventSeq
	merged["archived"] = item.ArchivedAt != nil
	if item.PinnedAt != nil {
		merged["pinnedAt"] = item.PinnedAt.UTC()
	}
	if item.ParentScopeTypeSet {
		merged["parentScopeType"] = normalizedString(item.ParentScopeType)
	}
	return merged
}

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

func firstNonZeroTime(values ...time.Time) time.Time {
	for _, value := range values {
		if !value.IsZero() {
			return value.UTC()
		}
	}
	return time.Time{}
}