package message

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

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

const (
	ScopeRewindSuppressionReason             = "rewind"
	ScopeRecallSuppressionReason             = "recall"
	ScopeCascadeReplyThreadSuppressionReason = "cascade_reply_thread_rewind"

	scopeSearchProjectionTableName        = "scope_message_search_docs"
	scopeSearchFullTextIndexName          = "scope_message_search_docs_content_fulltext_idx"
	scopeSearchFullTextParserName         = "MULTILINGUAL"
	scopeSearchFinalizeShadowTableName    = "scope_message_search_docs_finalize_shadow"
	scopeSearchFinalizeBackupTableName    = "scope_message_search_docs_finalize_backup"
	scopeSearchFinalizeStatusComplete     = "complete"
	scopeSearchBackfillStatusComplete     = "complete"
	scopeSearchMigrationStatusComplete    = "complete"
	scopeSearchBackfillStatusRunning      = "running"
	scopeSearchFinalizeStatusPending      = "pending"
	scopeSearchFinalizeStatusRunning      = "running"
	scopeSearchReadinessProjectionDefault = TiDBMessageSearchProjectionName
)

var (
	scopeReactionByActorQuery = scopeQueryDescriptor{
		Name:              "scope_reaction_by_actor",
		TargetIndex:       "scope_message_reactions_message_emoji_member_key(message_id, emoji, reactor_member_id)",
		AmplificationRisk: "bounded by one message, one emoji, and one reactor; used by reaction idempotency and removal",
		SQL: `
SELECT id, workspace_id, scope_id, message_id, emoji, reactor_member_id, created_at
FROM scope_message_reactions
WHERE message_id = $1
  AND emoji = $2
  AND reactor_member_id = $3
`,
	}
	scopeInsertReactionQuery = scopeQueryDescriptor{
		Name:              "scope_insert_reaction",
		TargetIndex:       "scope_message_reactions_message_emoji_member_key(message_id, emoji, reactor_member_id)",
		AmplificationRisk: "single immutable reaction fact insert; duplicate key returns existing reaction through the owner repository",
		SQL: `
INSERT INTO scope_message_reactions (id, workspace_id, scope_id, message_id, emoji, reactor_member_id, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`,
	}
	scopeDeleteReactionByActorQuery = scopeQueryDescriptor{
		Name:              "scope_delete_reaction_by_actor",
		TargetIndex:       "scope_message_reactions_message_emoji_member_key(message_id, emoji, reactor_member_id)",
		AmplificationRisk: "bounded by one message, one emoji, and one reactor; used by reaction remove",
		SQL: `
DELETE FROM scope_message_reactions
WHERE message_id = $1
  AND emoji = $2
  AND reactor_member_id = $3
`,
	}
	scopeSuppressReactionActivitiesQuery = scopeQueryDescriptor{
		Name:              "scope_suppress_reaction_activities",
		TargetIndex:       "scope_activity_events_ref_key(activity_ref_type, activity_ref_id, activity_type)",
		AmplificationRisk: "bounded by one reaction id and reaction activity type; used when a reaction is removed",
		SQL: `
UPDATE scope_activity_events
SET suppressed_at = $2
WHERE activity_ref_type = 'reaction'
  AND activity_ref_id = $1
  AND activity_type = 'reaction_added'
  AND suppressed_at IS NULL
`,
	}
	scopeSearchReadinessDetailQuery = scopeQueryDescriptor{
		Name:              "scope_search_readiness_detail",
		TargetIndex:       "scope_message_search_state PRIMARY(workspace_id, scope_id, projection_name)",
		AmplificationRisk: "point lookup for search readiness gate; search API must not infer readiness from docs rows",
		SQL: `
SELECT migration_status, backfill_status, finalize_status, read_ready, backfill_cursor_global_seq, created_at, updated_at
FROM scope_message_search_state
WHERE workspace_id = $1
  AND scope_id = $2
  AND projection_name = $3
`,
	}
	scopeSearchMessagesForBackfillQuery = scopeQueryDescriptor{
		Name:              "scope_search_messages_for_backfill",
		TargetIndex:       "scope_messages_scope_seq_key(scope_id, seq) + scope_message_search_docs PRIMARY(message_id)",
		AmplificationRisk: "bounded by one workspace, one scope, and explicit batch limit; backfill repairs missing docs in authoritative scope-local message order",
		SQL: `
SELECT m.id, m.workspace_id, m.scope_id, m.seq, m.workspace_global_seq, m.content, m.created_at
FROM scope_messages m
JOIN scopes searchable_scope
  ON searchable_scope.id = m.scope_id
 AND searchable_scope.workspace_id = m.workspace_id
 AND searchable_scope.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
LEFT JOIN scope_message_search_docs docs
  ON docs.message_id = m.id
WHERE m.workspace_id = $1
  AND m.scope_id = $2
  AND docs.message_id IS NULL
  AND ` + MessageNotRewoundSQL("m") + `
ORDER BY m.seq ASC
LIMIT $3
	`,
	}
	scopeSearchWorkspaceMessagesForBackfillQuery = scopeQueryDescriptor{
		Name:              "scope_search_workspace_messages_for_backfill",
		TargetIndex:       "scope_messages_workspace_global_seq_idx(workspace_id, workspace_global_seq) + scope_message_search_docs PRIMARY(message_id)",
		AmplificationRisk: "bounded by one workspace and explicit batch limit; workspace-global backfill repairs missing docs in authoritative workspace-global message order",
		SQL: `
	SELECT m.id, m.workspace_id, m.scope_id, m.workspace_global_seq, m.workspace_global_seq, m.content, m.created_at
	FROM scope_messages m
	JOIN scopes searchable_scope
	  ON searchable_scope.id = m.scope_id
	 AND searchable_scope.workspace_id = m.workspace_id
	 AND searchable_scope.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
	LEFT JOIN scope_message_search_docs docs
	  ON docs.message_id = m.id
	WHERE m.workspace_id = $1
	  AND docs.message_id IS NULL
	  AND ` + MessageNotRewoundSQL("m") + `
	ORDER BY m.workspace_global_seq ASC
	LIMIT $2
	`,
	}
	scopeSearchMissingDocsCountQuery = scopeQueryDescriptor{
		Name:              "scope_search_missing_docs_count",
		TargetIndex:       "scope_messages scope/global indexes + scope_message_search_docs PRIMARY(message_id)",
		AmplificationRisk: "bounded by one workspace and one scope; used before finalize/promote",
		SQL: `
SELECT COUNT(*)
FROM scope_messages m
JOIN scopes searchable_scope
  ON searchable_scope.id = m.scope_id
 AND searchable_scope.workspace_id = m.workspace_id
 AND searchable_scope.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
LEFT JOIN scope_message_search_docs docs
  ON docs.message_id = m.id
WHERE m.workspace_id = $1
  AND m.scope_id = $2
  AND docs.message_id IS NULL
  AND ` + MessageNotRewoundSQL("m") + `
	`,
	}
	scopeSearchWorkspaceMissingDocsCountQuery = scopeQueryDescriptor{
		Name:              "scope_search_workspace_missing_docs_count",
		TargetIndex:       "scope_messages_workspace_global_seq_idx(workspace_id, workspace_global_seq) + scope_message_search_docs PRIMARY(message_id)",
		AmplificationRisk: "bounded by one workspace; used before workspace-global finalize/promote",
		SQL: `
	SELECT COUNT(*)
	FROM scope_messages m
	JOIN scopes searchable_scope
	  ON searchable_scope.id = m.scope_id
	 AND searchable_scope.workspace_id = m.workspace_id
	 AND searchable_scope.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
	LEFT JOIN scope_message_search_docs docs
	  ON docs.message_id = m.id
	WHERE m.workspace_id = $1
	  AND docs.message_id IS NULL
	  AND ` + MessageNotRewoundSQL("m") + `
	`,
	}
	scopeSearchLatestScopeSeqQuery = scopeQueryDescriptor{
		Name:              "scope_search_latest_scope_seq",
		TargetIndex:       "scope_messages_scope_seq_key(scope_id, seq)",
		AmplificationRisk: "bounded aggregate over one workspace/scope to normalize the legacy backfill cursor to authoritative scope-local message order",
		SQL: `
SELECT COALESCE(MAX(seq), 0)
FROM scope_messages
WHERE workspace_id = $1
  AND scope_id = $2
	`,
	}
	scopeSearchLatestWorkspaceGlobalSeqQuery = scopeQueryDescriptor{
		Name:              "scope_search_latest_workspace_global_seq",
		TargetIndex:       "scope_messages_workspace_global_seq_idx(workspace_id, workspace_global_seq)",
		AmplificationRisk: "bounded aggregate over one workspace to normalize the workspace-global backfill cursor to authoritative workspace-global message order",
		SQL: `
	SELECT COALESCE(MAX(workspace_global_seq), 0)
	FROM scope_messages
	WHERE workspace_id = $1
	`,
	}
	scopeSearchUpdateStateQuery = scopeQueryDescriptor{
		Name:              "scope_search_update_state",
		TargetIndex:       "scope_message_search_state PRIMARY(workspace_id, scope_id, projection_name)",
		AmplificationRisk: "single projection state row update for search lifecycle",
		SQL: `
UPDATE scope_message_search_state
SET migration_status = $4,
    backfill_status = $5,
    backfill_cursor_global_seq = $6,
    finalize_status = $7,
    read_ready = $8,
    updated_at = $9
WHERE workspace_id = $1
  AND scope_id = $2
  AND projection_name = $3
`,
	}
	scopeSearchShowCreateDocsQuery = scopeQueryDescriptor{
		Name:              "scope_search_show_create_docs",
		TargetIndex:       "scope_message_search_docs physical table metadata",
		AmplificationRisk: "metadata-only lookup used by finalize/readiness verification; does not scan message rows",
		SQL:               `SHOW CREATE TABLE __TABLE__`,
	}
	scopeSearchFinalizeTableExistsQuery = scopeQueryDescriptor{
		Name:              "scope_search_finalize_table_exists",
		TargetIndex:       "information_schema.tables(table_schema, table_name)",
		AmplificationRisk: "metadata-only point lookup for finalize shadow/backup cleanup",
		SQL: `
SELECT COUNT(*)
FROM information_schema.tables
WHERE table_schema = DATABASE()
  AND table_name = $1
`,
	}
	scopeSearchFullTextEnabledQuery = scopeQueryDescriptor{
		Name:              "scope_search_fulltext_enabled",
		TargetIndex:       "TiDB variable tidb_enable_fulltext_index",
		AmplificationRisk: "metadata-only readiness check before creating the FULLTEXT artifact",
		SQL:               `SHOW VARIABLES LIKE 'tidb_enable_fulltext_index'`,
	}
	scopeSearchTiFlashRegisteredQuery = scopeQueryDescriptor{
		Name:              "scope_search_tiflash_registered",
		TargetIndex:       "information_schema.cluster_info(type)",
		AmplificationRisk: "metadata-only readiness check before creating the FULLTEXT artifact",
		SQL: `
SELECT COUNT(*)
FROM information_schema.cluster_info
WHERE type IN ('tiflash', 'tiflash_compute')
`,
	}
	scopeShareCardByMessageQuery = scopeQueryDescriptor{
		Name:              "scope_share_card_by_message",
		TargetIndex:       "scope_share_cards_scope_message_key(scope_id, message_id)",
		AmplificationRisk: "point lookup by anchor scope/message; used by share-card APIs and message-owned action state",
		SQL: `
SELECT id, workspace_id, scope_id, message_id, source_scope_id,
       COALESCE(source_message_id, ''), COALESCE(target_scope_id, ''), created_by_member_id,
       revoked_at, COALESCE(revoked_by_member_id, ''), payload, created_at, updated_at
FROM scope_share_cards
WHERE scope_id = $1
  AND message_id = $2
`,
	}
	scopeRevokeShareCardQuery = scopeQueryDescriptor{
		Name:              "scope_revoke_share_card",
		TargetIndex:       "scope_share_cards_scope_message_key(scope_id, message_id)",
		AmplificationRisk: "point update by anchor scope/message; preserves structured revoked_at/revoked_by lifecycle",
		SQL: `
UPDATE scope_share_cards
SET revoked_at = $3,
    revoked_by_member_id = $4,
    updated_at = $3
WHERE scope_id = $1
  AND message_id = $2
  AND revoked_at IS NULL
`,
	}
	scopeRewindSuffixMessagesQuery = scopeQueryDescriptor{
		Name:              "scope_rewind_suffix_messages",
		TargetIndex:       "scope_messages_scope_seq_key(scope_id, seq)",
		AmplificationRisk: "bounded by one scope suffix from a validated rewind plan",
		SQL: `
SELECT id, workspace_id, scope_id, seq, workspace_global_seq, author_member_id,
       author_type,
       COALESCE(NULLIF(notify_principal_member_id, ''), author_member_id),
       COALESCE(NULLIF(idempotency_subject_member_id, ''), author_member_id),
       COALESCE(client_message_id, ''), COALESCE(idempotency_payload_digest, ''),
       message_type, content, metadata, created_at, updated_at,
       edited_at, COALESCE(edited_by_member_id, ''), edit_revision
FROM scope_messages
WHERE scope_id = $1
  AND seq >= $2
  AND ` + MessageNotRewoundSQL("scope_messages") + `
  AND ` + MessageNotRewindSystemNoticeSQL("scope_messages") + `
ORDER BY seq ASC
`,
	}
	scopeRewindRecomputeStatsQuery = scopeQueryDescriptor{
		Name:              "scope_rewind_recompute_stats",
		TargetIndex:       "scope_messages_scope_seq_key(scope_id, seq)",
		AmplificationRisk: "bounded by one scope after suffix tombstone; recomputes physical high-water seq/global and visible summary state",
		SQL: `
UPDATE scopes
SET latest_seq = (SELECT COALESCE(MAX(seq), 0) FROM scope_messages WHERE scope_id = $1),
    latest_global_seq = (SELECT COALESCE(MAX(workspace_global_seq), 0) FROM scope_messages WHERE scope_id = $1),
    latest_message_at = (SELECT MAX(created_at) FROM scope_messages WHERE scope_id = $1 AND ` + HumanVisibleMessageTypeSQL("scope_messages") + `),
    visible_message_count = (SELECT COUNT(*) FROM scope_messages WHERE scope_id = $1 AND ` + HumanVisibleMessageTypeSQL("scope_messages") + `),
    visible_last_message_at = (SELECT MAX(created_at) FROM scope_messages WHERE scope_id = $1 AND ` + HumanVisibleMessageTypeSQL("scope_messages") + `),
    recent_author_member_ids = CASE
      WHEN type = 'thread' THEN (
        SELECT COALESCE(JSON_ARRAYAGG(recent.author_member_id), JSON_ARRAY())
        FROM (
          SELECT author_member_id, MAX(seq) AS latest_seq
          FROM scope_messages
          WHERE scope_id = $1
            AND ` + ReplyThreadPreviewAuthorSQL("scope_messages") + `
          GROUP BY author_member_id
          ORDER BY latest_seq DESC
          LIMIT 5
        ) recent
      )
      ELSE recent_author_member_ids
    END,
    unread_rewind_generation = unread_rewind_generation + 1,
    updated_at = $2
WHERE id = $1
`,
	}
	scopeRewindClampMembershipReadStateQuery = scopeQueryDescriptor{
		Name:              "scope_rewind_clamp_membership_read_state",
		TargetIndex:       "scope_memberships PRIMARY(scope_id, member_id)",
		AmplificationRisk: "bounded by one scope membership set after suffix tombstone",
		SQL: `
UPDATE scope_memberships
SET unread_generation = unread_generation + 1,
    last_read_seq = LEAST(last_read_seq, $2),
    last_read_global_seq = LEAST(last_read_global_seq, $3),
    owner_context_dismissed_through_global_seq = LEAST(owner_context_dismissed_through_global_seq, $3),
    updated_at = $4
WHERE scope_id = $1
  AND (
    last_read_seq > $2
    OR last_read_global_seq > $3
    OR owner_context_dismissed_through_global_seq > $3
  )
`,
	}
)

type ScopeReactionMutationResult struct {
	Changed  bool
	Reaction ScopeMessageReactionRecord
}

type ScopeSearchReadiness struct {
	WorkspaceID             string
	ScopeID                 string
	ProjectionName          string
	MigrationStatus         string
	BackfillStatus          string
	FinalizeStatus          string
	ReadReady               bool
	BackfillCursorGlobalSeq int64
	CreatedAt               time.Time
	UpdatedAt               time.Time
}

type ScopeSearchBackfillResult struct {
	ProcessedRows int
	State         ScopeSearchReadiness
}

type scopeSearchBackfillCandidate struct {
	Doc        ScopeSearchDocRecord
	MessageSeq int64
}

type ScopeRewindSuffixRequest struct {
	WorkspaceID       string
	ScopeID           string
	FromSeq           int64
	RewoundAt         time.Time
	SuppressionReason string
}

type ScopeRewindSuffixResult struct {
	RewoundMessageIDs           []string
	SuppressedActivityCount     int64
	SuppressedOutboxCount       int64
	ClampedScopeSyncDirtyCount  int64
	ClampedMemberSyncDirtyCount int64
	LatestSeq                   int64
	LatestGlobalSeq             int64
	// DeletedAttachmentDerivatives are (workspaceID, attachmentID) pairs whose
	// thumbnail objects should be cleaned from OSS post-commit (best-effort).
	DeletedAttachmentDerivatives []DeletedAttachmentDerivative
}

type ScopeRecallMessageRequest struct {
	WorkspaceID         string
	ScopeID             string
	MessageID           string
	RecalledByMemberID  string
	ExpectedAuthorType  string
	ExpectedAuthorID    string
	ExpectedMessageType string
	PlaceholderContent  string
	Metadata            []byte
	RecalledAt          time.Time
}

type ScopeRecallMessageResult struct {
	Message                 ScopeMessageRecord
	SuppressedActivityCount int64
	SuppressedOutboxCount   int64
	IsLatestVisibleMessage  bool
	// DeletedAttachmentDerivatives are (workspaceID, attachmentID) pairs whose
	// thumbnail objects should be cleaned from OSS post-commit (best-effort).
	DeletedAttachmentDerivatives []DeletedAttachmentDerivative
}

type scopeMessageTombstoneProjectionStep struct {
	name                string
	impactedTable       string
	readPathLabel string
	statement           string
}

// scopeMessageTombstoneProjectionSteps is the shared recall/rewind
// read-path cleanup contract. Recall and rewind both keep the transcript row
// and replace it with a lifecycle tombstone; every message-bound sidecar or
// projection that could expose original content, snippets, reactions, source
// refs, share-card payloads, pins, or attachments must be scrubbed through this
// single contract.
var scopeMessageTombstoneProjectionSteps = []scopeMessageTombstoneProjectionStep{
	{
		name:                "share_cards",
		impactedTable:       "scope_share_cards",
		readPathLabel: "share-card action-state reads",
		statement: `
DELETE FROM scope_share_cards
WHERE (scope_id = $1 AND message_id IN (__MESSAGE_IDS__))
   OR (source_scope_id = $1 AND source_message_id IN (__MESSAGE_IDS__))
`,
	},
	{
		name:                "message_reactions",
		impactedTable:       "scope_message_reactions",
		readPathLabel: "reaction aggregate reads",
		statement: `
DELETE FROM scope_message_reactions
WHERE scope_id = $1
  AND message_id IN (__MESSAGE_IDS__)
`,
	},
	{
		name:                "message_pins",
		impactedTable:       "scope_message_pins",
		readPathLabel: "message pin list reads",
		statement: `
DELETE FROM scope_message_pins
WHERE scope_id = $1
  AND message_id IN (__MESSAGE_IDS__)
`,
	},
	{
		name:                "message_top_notices",
		impactedTable:       "scope_message_top_notices",
		readPathLabel: "message top banner reads",
		statement: `
DELETE FROM scope_message_top_notices
WHERE scope_id = $1
  AND message_id IN (__MESSAGE_IDS__)
`,
	},
	{
		name:                "message_search_docs",
		impactedTable:       "scope_message_search_docs",
		readPathLabel: "search reads",
		statement: `
DELETE FROM scope_message_search_docs
WHERE scope_id = $1
  AND message_id IN (__MESSAGE_IDS__)
`,
	},
	{
		name:                "message_sources",
		impactedTable:       "scope_message_sources",
		readPathLabel: "root/source context reads",
		statement: `
DELETE FROM scope_message_sources
WHERE (target_scope_id = $1 AND message_id IN (__MESSAGE_IDS__))
   OR (source_scope_id = $1 AND source_message_id IN (__MESSAGE_IDS__))
`,
	},
	{
		name:                "attachments",
		impactedTable:       "attachments",
		readPathLabel: "attachment list/download reads",
		// Derivative cleanup (thumbnail objects in OSS) happens post-commit via
		// collectDeletedAttachmentDerivatives (same tx, before DELETE) +
		// Store wrapper best-effort (after commit). See scrubScopeMessageTombstoneProjections.
		statement: `
DELETE FROM attachments
WHERE scope_id = $1
  AND message_id IN (__MESSAGE_IDS__)
`,
	},
	{
		name:                "channel_attachment_projection",
		impactedTable:       "channel_attachment_projection",
		readPathLabel: "channel attachment list reads",
		statement: `
DELETE FROM channel_attachment_projection
WHERE scope_id = $1
  AND message_id IN (__MESSAGE_IDS__)
`,
	},
}

var scopeMessageTombstoneTranscriptStep = scopeMessageTombstoneProjectionStep{
	name:                "scope_messages",
	impactedTable:       "scope_messages",
	readPathLabel: "history/sync/attachment/diagnostics transcript authority",
	statement: `
UPDATE scope_messages
SET content = '',
    metadata = $2,
    updated_at = $3
WHERE scope_id = $1
  AND id IN (__MESSAGE_IDS__)
`,
}

func (store ReactionStore) Add(ctx context.Context, record ScopeMessageReactionRecord) (ScopeReactionMutationResult, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeReactionMutationResult{}, err
	}
	if strings.TrimSpace(record.ID) == "" || strings.TrimSpace(record.WorkspaceID) == "" || strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.MessageID) == "" || strings.TrimSpace(record.Emoji) == "" || strings.TrimSpace(record.ReactorMemberID) == "" {
		return ScopeReactionMutationResult{}, invalidArgument("reaction id, workspace, scope, message, emoji, and reactor are required")
	}
	if tx, ok := store.execer.(coredb.Tx); ok {
		return ReactionStore{scopeStoreBase: scopeStoreBase{execer: tx}}.addInTx(ctx, record)
	}
	txBeginner, ok := store.execer.(coredb.TxBeginner)
	if !ok {
		return ScopeReactionMutationResult{}, invalidArgument("scope reaction add requires a transaction-capable repository")
	}
	var added ScopeReactionMutationResult
	err := coredb.WithTx(ctx, txBeginner, func(tx coredb.Tx) error {
		var err error
		added, err = ReactionStore{scopeStoreBase: scopeStoreBase{execer: tx}}.addInTx(ctx, record)
		return err
	})
	if err != nil {
		return ScopeReactionMutationResult{}, err
	}
	return added, nil
}

func (store ReactionStore) addInTx(ctx context.Context, record ScopeMessageReactionRecord) (ScopeReactionMutationResult, error) {
	existing, found, err := store.LoadByActor(ctx, record.MessageID, record.Emoji, record.ReactorMemberID)
	if err != nil {
		return ScopeReactionMutationResult{}, err
	}
	if found {
		return ScopeReactionMutationResult{Changed: false, Reaction: existing}, nil
	}
	if _, err := store.execer.Exec(ctx, scopeInsertReactionQuery.SQL, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.MessageID), strings.TrimSpace(record.Emoji), strings.TrimSpace(record.ReactorMemberID), record.CreatedAt); err != nil {
		if uniqueViolationMatches(err, "scope_message_reactions_message_emoji_member_key") {
			existing, found, loadErr := store.LoadByActor(ctx, record.MessageID, record.Emoji, record.ReactorMemberID)
			if loadErr != nil {
				return ScopeReactionMutationResult{}, loadErr
			}
			if found {
				return ScopeReactionMutationResult{Changed: false, Reaction: existing}, nil
			}
		}
		return ScopeReactionMutationResult{}, fmt.Errorf("add scope reaction: %w", err)
	}
	if record.ActivityGlobalSeq <= 0 {
		record.ActivityGlobalSeq = record.ActivitySeq
	}
	if record.ActivitySeq > 0 {
		activityStore := ActivityEventStore{scopeStoreBase: store.scopeStoreBase}
		if err := activityStore.Insert(ctx, ScopeActivityEventRecord{
			ID:                scopePlanStableID("act", record.ID, "reaction_added"),
			WorkspaceID:       record.WorkspaceID,
			ScopeID:           record.ScopeID,
			Seq:               record.ActivitySeq,
			ActivityType:      "reaction_added",
			ActivityRefType:   "reaction",
			ActivityRefID:     record.ID,
			ActivityGlobalSeq: record.ActivityGlobalSeq,
			SourceMessageID:   record.MessageID,
			ActorMemberID:     record.ReactorMemberID,
			Reason:            "reaction",
			CreatedAt:         record.CreatedAt,
		}); err != nil {
			return ScopeReactionMutationResult{}, fmt.Errorf("add scope reaction activity: %w", err)
		}
	}
	return ScopeReactionMutationResult{Changed: true, Reaction: record}, nil
}

func (store ReactionStore) Remove(ctx context.Context, messageID string, emoji string, reactorMemberID string, removedAt time.Time) (ScopeReactionMutationResult, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeReactionMutationResult{}, err
	}
	if tx, ok := store.execer.(coredb.Tx); ok {
		return ReactionStore{scopeStoreBase: scopeStoreBase{execer: tx}}.removeInTx(ctx, messageID, emoji, reactorMemberID, removedAt)
	}
	txBeginner, ok := store.execer.(coredb.TxBeginner)
	if !ok {
		return ScopeReactionMutationResult{}, invalidArgument("scope reaction remove requires a transaction-capable repository")
	}
	var removed ScopeReactionMutationResult
	if err := coredb.WithTx(ctx, txBeginner, func(tx coredb.Tx) error {
		result, err := ReactionStore{scopeStoreBase: scopeStoreBase{execer: tx}}.removeInTx(ctx, messageID, emoji, reactorMemberID, removedAt)
		if err != nil {
			return err
		}
		removed = result
		return nil
	}); err != nil {
		return ScopeReactionMutationResult{}, err
	}
	return removed, nil
}

func (store ReactionStore) removeInTx(ctx context.Context, messageID string, emoji string, reactorMemberID string, removedAt time.Time) (ScopeReactionMutationResult, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeReactionMutationResult{}, err
	}
	existing, found, err := store.LoadByActor(ctx, messageID, emoji, reactorMemberID)
	if err != nil {
		return ScopeReactionMutationResult{}, err
	}
	if !found {
		return ScopeReactionMutationResult{Changed: false}, nil
	}
	if _, err := store.execer.Exec(ctx, scopeDeleteReactionByActorQuery.SQL, strings.TrimSpace(messageID), strings.TrimSpace(emoji), strings.TrimSpace(reactorMemberID)); err != nil {
		return ScopeReactionMutationResult{}, fmt.Errorf("delete scope reaction: %w", err)
	}
	if _, err := store.execer.Exec(ctx, scopeSuppressReactionActivitiesQuery.SQL, existing.ID, removedAt); err != nil {
		return ScopeReactionMutationResult{}, fmt.Errorf("suppress scope reaction activity: %w", err)
	}
	return ScopeReactionMutationResult{Changed: true, Reaction: existing}, nil
}

func (store ReactionStore) LoadByActor(ctx context.Context, messageID string, emoji string, reactorMemberID string) (ScopeMessageReactionRecord, bool, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeMessageReactionRecord{}, false, err
	}
	var record ScopeMessageReactionRecord
	err := store.execer.QueryRow(ctx, scopeReactionByActorQuery.SQL, strings.TrimSpace(messageID), strings.TrimSpace(emoji), strings.TrimSpace(reactorMemberID)).Scan(
		&record.ID,
		&record.WorkspaceID,
		&record.ScopeID,
		&record.MessageID,
		&record.Emoji,
		&record.ReactorMemberID,
		&record.CreatedAt,
	)
	if errorsIsNoRows(err) {
		return ScopeMessageReactionRecord{}, false, nil
	}
	if err != nil {
		return ScopeMessageReactionRecord{}, false, fmt.Errorf("load scope reaction by actor: %w", err)
	}
	return record, true, nil
}

func (store SearchStore) LoadReadiness(ctx context.Context, workspaceID string, scopeID string, projectionName string) (ScopeSearchReadiness, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeSearchReadiness{}, err
	}
	projectionName = strings.TrimSpace(projectionName)
	if projectionName == "" {
		projectionName = TiDBMessageSearchProjectionName
	}
	var record ScopeSearchReadiness
	record.WorkspaceID = strings.TrimSpace(workspaceID)
	record.ScopeID = strings.TrimSpace(scopeID)
	record.ProjectionName = projectionName
	err := store.execer.QueryRow(ctx, scopeSearchReadinessDetailQuery.SQL, record.WorkspaceID, record.ScopeID, record.ProjectionName).Scan(
		&record.MigrationStatus,
		&record.BackfillStatus,
		&record.FinalizeStatus,
		&record.ReadReady,
		&record.BackfillCursorGlobalSeq,
		&record.CreatedAt,
		&record.UpdatedAt,
	)
	if errorsIsNoRows(err) {
		return ScopeSearchReadiness{}, notFound("scope search readiness not found")
	}
	if err != nil {
		return ScopeSearchReadiness{}, fmt.Errorf("load scope search readiness: %w", err)
	}
	return record, nil
}

func (store SearchStore) BeginBackfill(ctx context.Context, workspaceID string, scopeID string, projectionName string, updatedAt time.Time) (ScopeSearchReadiness, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeSearchReadiness{}, err
	}
	if err := store.requireOrdinarySearchScope(ctx, workspaceID, scopeID); err != nil {
		return ScopeSearchReadiness{}, err
	}
	projectionName = normalizeScopeSearchProjectionName(projectionName)
	if existing, err := store.LoadReadiness(ctx, workspaceID, scopeID, projectionName); err == nil && existing.Ready() {
		return existing, nil
	} else if err != nil && !errors.Is(err, ErrNotFound) {
		return ScopeSearchReadiness{}, err
	}
	record := ScopeSearchStateRecord{
		WorkspaceID:             strings.TrimSpace(workspaceID),
		ScopeID:                 strings.TrimSpace(scopeID),
		ProjectionName:          projectionName,
		MigrationStatus:         "complete",
		BackfillStatus:          "running",
		BackfillCursorGlobalSeq: 0,
		FinalizeStatus:          "pending",
		ReadReady:               false,
		CreatedAt:               updatedAt,
		UpdatedAt:               updatedAt,
	}
	if record.WorkspaceID == "" {
		return ScopeSearchReadiness{}, invalidArgument("scope search backfill workspace is required")
	}
	if err := store.UpsertState(ctx, record); err != nil {
		return ScopeSearchReadiness{}, err
	}
	return store.LoadReadiness(ctx, record.WorkspaceID, record.ScopeID, projectionName)
}

func (store SearchStore) Backfill(ctx context.Context, workspaceID string, scopeID string, projectionName string, batchSize int, updatedAt time.Time) (ScopeSearchBackfillResult, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeSearchBackfillResult{}, err
	}
	if batchSize <= 0 {
		return ScopeSearchBackfillResult{}, invalidArgument("scope search backfill batch size must be positive")
	}
	projectionName = normalizeScopeSearchProjectionName(projectionName)
	var result ScopeSearchBackfillResult
	err := store.withSearchTx(ctx, func(txStore SearchStore) error {
		if err := txStore.requireOrdinarySearchScope(ctx, workspaceID, scopeID); err != nil {
			return err
		}
		state, err := txStore.LoadReadiness(ctx, workspaceID, scopeID, projectionName)
		if err != nil {
			if !errors.Is(err, ErrNotFound) {
				return err
			}
			state, err = txStore.BeginBackfill(ctx, workspaceID, scopeID, projectionName, updatedAt)
			if err != nil {
				return err
			}
		}
		if strings.TrimSpace(state.MigrationStatus) != scopeSearchMigrationStatusComplete {
			return invalidArgument("scope search migration must be complete before backfill")
		}
		latestSeq, err := txStore.loadBackfillLatestScopeSeq(ctx, state.WorkspaceID, state.ScopeID)
		if err != nil {
			return err
		}
		docs, err := txStore.loadBackfillDocs(ctx, state.WorkspaceID, state.ScopeID, batchSize)
		if err != nil {
			return err
		}
		result.ProcessedRows = len(docs)
		cursorAfter := normalizeScopeSearchBackfillCursor(state.BackfillCursorGlobalSeq, latestSeq)
		for _, doc := range docs {
			if err := txStore.InsertDoc(ctx, doc.Doc); err != nil {
				return err
			}
			if doc.MessageSeq > cursorAfter {
				cursorAfter = doc.MessageSeq
			}
		}
		hasMore := len(docs) == batchSize
		if hasMore {
			missing, err := txStore.countMissingSearchDocs(ctx, state.WorkspaceID, state.ScopeID)
			if err != nil {
				return err
			}
			hasMore = missing > 0
		}
		if !hasMore {
			cursorAfter = latestSeq
		}
		if state.Ready() {
			if cursorAfter != state.BackfillCursorGlobalSeq {
				if err := txStore.updateReadinessState(ctx, state.WorkspaceID, state.ScopeID, projectionName, scopeSearchMigrationStatusComplete, scopeSearchBackfillStatusComplete, cursorAfter, scopeSearchFinalizeStatusComplete, true, updatedAt); err != nil {
					return err
				}
				state, err = txStore.LoadReadiness(ctx, state.WorkspaceID, state.ScopeID, projectionName)
				if err != nil {
					return err
				}
			}
			result.State = state
			if hasMore {
				result.State.BackfillStatus = scopeSearchBackfillStatusRunning
			}
			return nil
		}
		backfillStatus := scopeSearchBackfillStatusRunning
		if !hasMore {
			backfillStatus = scopeSearchBackfillStatusComplete
		}
		finalizeStatus := strings.TrimSpace(state.FinalizeStatus)
		if finalizeStatus == "" {
			finalizeStatus = scopeSearchFinalizeStatusPending
		}
		readReady := state.ReadReady &&
			strings.TrimSpace(state.MigrationStatus) == scopeSearchMigrationStatusComplete &&
			backfillStatus == scopeSearchBackfillStatusComplete &&
			finalizeStatus == scopeSearchFinalizeStatusComplete
		if err := txStore.updateReadinessState(ctx, state.WorkspaceID, state.ScopeID, projectionName, scopeSearchMigrationStatusComplete, backfillStatus, cursorAfter, finalizeStatus, readReady, updatedAt); err != nil {
			return err
		}
		updated, err := txStore.LoadReadiness(ctx, state.WorkspaceID, state.ScopeID, projectionName)
		if err != nil {
			return err
		}
		result.State = updated
		return nil
	})
	if err != nil {
		return ScopeSearchBackfillResult{}, err
	}
	return result, nil
}

func (store SearchStore) Finalize(ctx context.Context, workspaceID string, scopeID string, projectionName string, updatedAt time.Time) (ScopeSearchReadiness, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeSearchReadiness{}, err
	}
	if err := store.requireOrdinarySearchScope(ctx, workspaceID, scopeID); err != nil {
		return ScopeSearchReadiness{}, err
	}
	projectionName = normalizeScopeSearchProjectionName(projectionName)
	state, err := store.LoadReadiness(ctx, workspaceID, scopeID, projectionName)
	if err != nil {
		return ScopeSearchReadiness{}, err
	}
	if strings.TrimSpace(state.MigrationStatus) != scopeSearchMigrationStatusComplete {
		return ScopeSearchReadiness{}, invalidArgument("scope search migration must be complete before finalize")
	}
	if strings.TrimSpace(state.BackfillStatus) != scopeSearchBackfillStatusComplete {
		return ScopeSearchReadiness{}, invalidArgument("scope search backfill must be complete before finalize")
	}
	missing, err := store.countMissingSearchDocs(ctx, state.WorkspaceID, state.ScopeID)
	if err != nil {
		return ScopeSearchReadiness{}, err
	}
	if missing > 0 {
		return ScopeSearchReadiness{}, invalidArgument("scope search projection has missing rows; run scope search backfill")
	}
	if err := store.ensureSearchFinalizeArtifacts(ctx); err != nil {
		return ScopeSearchReadiness{}, err
	}
	if err := store.updateReadinessState(ctx, state.WorkspaceID, state.ScopeID, projectionName, scopeSearchMigrationStatusComplete, scopeSearchBackfillStatusComplete, state.BackfillCursorGlobalSeq, scopeSearchFinalizeStatusComplete, state.ReadReady, updatedAt); err != nil {
		return ScopeSearchReadiness{}, err
	}
	return store.LoadReadiness(ctx, state.WorkspaceID, state.ScopeID, projectionName)
}

func (store SearchStore) PromoteReady(ctx context.Context, workspaceID string, scopeID string, projectionName string, updatedAt time.Time) (ScopeSearchReadiness, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeSearchReadiness{}, err
	}
	if err := store.requireOrdinarySearchScope(ctx, workspaceID, scopeID); err != nil {
		return ScopeSearchReadiness{}, err
	}
	projectionName = normalizeScopeSearchProjectionName(projectionName)
	var readiness ScopeSearchReadiness
	err := store.withSearchTx(ctx, func(txStore SearchStore) error {
		state, err := txStore.LoadReadiness(ctx, workspaceID, scopeID, projectionName)
		if err != nil {
			return err
		}
		if strings.TrimSpace(state.MigrationStatus) != scopeSearchMigrationStatusComplete {
			return invalidArgument("scope search migration must be complete before promote-ready")
		}
		if strings.TrimSpace(state.BackfillStatus) != scopeSearchBackfillStatusComplete {
			return invalidArgument("scope search backfill must be complete before promote-ready")
		}
		if strings.TrimSpace(state.FinalizeStatus) != scopeSearchFinalizeStatusComplete {
			return invalidArgument("scope search finalize must be complete before promote-ready")
		}
		indexReady, err := txStore.scopeSearchFullTextIndexReadyOnTable(ctx, scopeSearchProjectionTableName)
		if err != nil {
			return err
		}
		if !indexReady {
			return invalidArgument("scope search projection is not finalized with the multilingual FULLTEXT artifact")
		}
		missing, err := txStore.countMissingSearchDocs(ctx, state.WorkspaceID, state.ScopeID)
		if err != nil {
			return err
		}
		if missing > 0 {
			return invalidArgument("scope search projection has missing rows; run scope search backfill")
		}
		if err := txStore.updateReadinessState(ctx, state.WorkspaceID, state.ScopeID, projectionName, scopeSearchMigrationStatusComplete, scopeSearchBackfillStatusComplete, state.BackfillCursorGlobalSeq, scopeSearchFinalizeStatusComplete, true, updatedAt); err != nil {
			return err
		}
		readiness, err = txStore.LoadReadiness(ctx, state.WorkspaceID, state.ScopeID, projectionName)
		return err
	})
	if err != nil {
		return ScopeSearchReadiness{}, err
	}
	return readiness, nil
}

func (store SearchStore) requireOrdinarySearchScope(ctx context.Context, workspaceID string, scopeID string) error {
	workspaceID = strings.TrimSpace(workspaceID)
	scopeID = strings.TrimSpace(scopeID)
	if scopeID == "" {
		return nil
	}
	var marker int
	err := store.execer.QueryRow(ctx, `
SELECT 1
FROM scopes
WHERE workspace_id = $1
  AND id = $2
  AND type IN ('channel', 'dm', 'thread', 'task', 'service_account')
LIMIT 1
`, workspaceID, scopeID).Scan(&marker)
	if errorsIsNoRows(err) {
		return invalidArgument("scope does not support ordinary message search")
	}
	if err != nil {
		return fmt.Errorf("validate scope message search target: %w", err)
	}
	return nil
}

func (record ScopeSearchReadiness) Ready() bool {
	return record.ReadReady &&
		record.MigrationStatus == scopeSearchMigrationStatusComplete &&
		record.BackfillStatus == scopeSearchBackfillStatusComplete &&
		record.FinalizeStatus == scopeSearchFinalizeStatusComplete
}

func (store SearchStore) withSearchTx(ctx context.Context, fn func(SearchStore) error) error {
	if tx, ok := store.execer.(coredb.Tx); ok {
		return fn(SearchStore{scopeStoreBase: scopeStoreBase{execer: tx}})
	}
	if beginner, ok := store.execer.(coredb.TxBeginner); ok {
		return coredb.WithTx(ctx, beginner, func(tx coredb.Tx) error {
			return fn(SearchStore{scopeStoreBase: scopeStoreBase{execer: tx}})
		})
	}
	return fn(store)
}

func (store SearchStore) loadBackfillDocs(ctx context.Context, workspaceID string, scopeID string, limit int) ([]scopeSearchBackfillCandidate, error) {
	workspaceID = strings.TrimSpace(workspaceID)
	scopeID = strings.TrimSpace(scopeID)
	query := scopeSearchMessagesForBackfillQuery.SQL
	args := []any{workspaceID, scopeID, limit}
	if scopeID == "" {
		query = scopeSearchWorkspaceMessagesForBackfillQuery.SQL
		args = []any{workspaceID, limit}
	}
	rows, err := store.execer.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("load scope search backfill messages: %w", err)
	}
	defer rows.Close()
	docs := make([]scopeSearchBackfillCandidate, 0, limit)
	for rows.Next() {
		var candidate scopeSearchBackfillCandidate
		if err := rows.Scan(
			&candidate.Doc.MessageID,
			&candidate.Doc.WorkspaceID,
			&candidate.Doc.ScopeID,
			&candidate.MessageSeq,
			&candidate.Doc.GlobalSeq,
			&candidate.Doc.Content,
			&candidate.Doc.CreatedAt,
		); err != nil {
			return nil, fmt.Errorf("scan scope search backfill message: %w", err)
		}
		docs = append(docs, candidate)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate scope search backfill messages: %w", err)
	}
	return docs, nil
}

func (store SearchStore) loadBackfillLatestScopeSeq(ctx context.Context, workspaceID string, scopeID string) (int64, error) {
	var latestSeq int64
	workspaceID = strings.TrimSpace(workspaceID)
	scopeID = strings.TrimSpace(scopeID)
	query := scopeSearchLatestScopeSeqQuery.SQL
	args := []any{workspaceID, scopeID}
	if scopeID == "" {
		query = scopeSearchLatestWorkspaceGlobalSeqQuery.SQL
		args = []any{workspaceID}
	}
	if err := store.execer.QueryRow(ctx, query, args...).Scan(&latestSeq); err != nil {
		return 0, fmt.Errorf("load scope search backfill latest seq: %w", err)
	}
	return latestSeq, nil
}

func normalizeScopeSearchBackfillCursor(cursor int64, latestSeq int64) int64 {
	if cursor < 0 {
		return 0
	}
	if latestSeq < 0 {
		return cursor
	}
	if cursor > latestSeq {
		return latestSeq
	}
	return cursor
}

func (store SearchStore) countMissingSearchDocs(ctx context.Context, workspaceID string, scopeID string) (int64, error) {
	var missing int64
	workspaceID = strings.TrimSpace(workspaceID)
	scopeID = strings.TrimSpace(scopeID)
	query := scopeSearchMissingDocsCountQuery.SQL
	args := []any{workspaceID, scopeID}
	if scopeID == "" {
		query = scopeSearchWorkspaceMissingDocsCountQuery.SQL
		args = []any{workspaceID}
	}
	if err := store.execer.QueryRow(ctx, query, args...).Scan(&missing); err != nil {
		return 0, fmt.Errorf("count missing scope search docs: %w", err)
	}
	return missing, nil
}

func (store SearchStore) ensureSearchFinalizeArtifacts(ctx context.Context) error {
	indexReady, err := store.scopeSearchFullTextIndexReadyOnTable(ctx, scopeSearchProjectionTableName)
	if err != nil {
		return err
	}
	if indexReady {
		if err := store.catchUpScopeSearchFinalizeBackup(ctx); err != nil {
			return err
		}
		if err := store.dropScopeSearchFinalizeTableIfExists(ctx, scopeSearchFinalizeShadowTableName); err != nil {
			return err
		}
		if err := store.dropScopeSearchFinalizeTableIfExists(ctx, scopeSearchFinalizeBackupTableName); err != nil {
			return err
		}
		return nil
	}

	if err := store.requireScopeSearchFullTextEnvironment(ctx); err != nil {
		return err
	}

	backupExists, err := store.scopeSearchFinalizeTableExists(ctx, scopeSearchFinalizeBackupTableName)
	if err != nil {
		return err
	}
	if backupExists {
		return fmt.Errorf("scope message search finalize backup table exists but the active projection still has no FULLTEXT index")
	}

	if err := store.dropScopeSearchFinalizeTableIfExists(ctx, scopeSearchFinalizeShadowTableName); err != nil {
		return err
	}
	if err := store.createScopeSearchFinalizeShadowTable(ctx); err != nil {
		return err
	}

	shadowReady, err := store.scopeSearchFullTextIndexReadyOnTable(ctx, scopeSearchFinalizeShadowTableName)
	if err != nil {
		return err
	}
	if !shadowReady {
		_ = store.dropScopeSearchFinalizeTableIfExists(ctx, scopeSearchFinalizeShadowTableName)
		return fmt.Errorf("scope message search finalize shadow table did not materialize a multilingual FULLTEXT index")
	}

	if err := store.copyScopeSearchProjectionRows(ctx, scopeSearchProjectionTableName, scopeSearchFinalizeShadowTableName); err != nil {
		return err
	}
	if err := store.swapScopeSearchProjectionTables(ctx); err != nil {
		return err
	}
	if err := store.catchUpScopeSearchFinalizeBackup(ctx); err != nil {
		return err
	}

	indexReady, err = store.scopeSearchFullTextIndexReadyOnTable(ctx, scopeSearchProjectionTableName)
	if err != nil {
		return err
	}
	if !indexReady {
		return fmt.Errorf("scope message search multilingual fulltext index is not visible after swapping in the finalized projection table")
	}

	if err := store.dropScopeSearchFinalizeTableIfExists(ctx, scopeSearchFinalizeBackupTableName); err != nil {
		return err
	}
	return nil
}

func (store SearchStore) requireScopeSearchFullTextEnvironment(ctx context.Context) error {
	fullTextEnabled, err := store.scopeSearchFullTextEnabled(ctx)
	if err != nil {
		return err
	}
	if !fullTextEnabled {
		return fmt.Errorf("tidb_enable_fulltext_index is not enabled; scope message search fulltext finalize requires it")
	}

	tiFlashReady, err := store.scopeSearchTiFlashNodeRegistered(ctx)
	if err != nil {
		return err
	}
	if !tiFlashReady {
		return fmt.Errorf("no TiFlash node is registered; scope message search fulltext finalize requires TiFlash")
	}
	return nil
}

func (store SearchStore) scopeSearchFullTextEnabled(ctx context.Context) (bool, error) {
	rows, err := store.execer.Query(ctx, scopeSearchFullTextEnabledQuery.SQL)
	if err != nil {
		return false, fmt.Errorf("query tidb_enable_fulltext_index: %w", err)
	}
	defer rows.Close()

	if !rows.Next() {
		if err := rows.Err(); err != nil {
			return false, fmt.Errorf("iterate tidb_enable_fulltext_index: %w", err)
		}
		return false, nil
	}

	var name, value string
	if err := rows.Scan(&name, &value); err != nil {
		return false, fmt.Errorf("scan tidb_enable_fulltext_index: %w", err)
	}
	if err := rows.Err(); err != nil {
		return false, fmt.Errorf("iterate tidb_enable_fulltext_index: %w", err)
	}
	return strings.EqualFold(strings.TrimSpace(value), "ON"), nil
}

func (store SearchStore) scopeSearchTiFlashNodeRegistered(ctx context.Context) (bool, error) {
	var count int
	if err := store.execer.QueryRow(ctx, scopeSearchTiFlashRegisteredQuery.SQL).Scan(&count); err != nil {
		return false, fmt.Errorf("query tiflash node registration: %w", err)
	}
	return count > 0, nil
}

func (store SearchStore) scopeSearchFinalizeTableExists(ctx context.Context, tableName string) (bool, error) {
	var count int
	if err := store.execer.QueryRow(ctx, scopeSearchFinalizeTableExistsQuery.SQL, strings.TrimSpace(tableName)).Scan(&count); err != nil {
		return false, fmt.Errorf("check table %s: %w", tableName, err)
	}
	return count == 1, nil
}

func (store SearchStore) dropScopeSearchFinalizeTableIfExists(ctx context.Context, tableName string) error {
	if _, err := store.execer.Exec(ctx, fmt.Sprintf("DROP TABLE IF EXISTS %s", quoteScopeSearchIdentifier(tableName))); err != nil {
		return fmt.Errorf("drop table %s: %w", tableName, err)
	}
	return nil
}

func (store SearchStore) createScopeSearchFinalizeShadowTable(ctx context.Context) error {
	if _, err := store.execer.Exec(ctx, fmt.Sprintf(`
CREATE TABLE %s (
    message_id VARCHAR(255) PRIMARY KEY,
    workspace_id VARCHAR(255) NOT NULL,
    scope_id VARCHAR(255) NOT NULL,
    global_seq BIGINT NOT NULL,
    content LONGTEXT NOT NULL,
    created_at TIMESTAMP NOT NULL,
    KEY scope_message_search_docs_workspace_global_seq_idx (workspace_id, global_seq),
    KEY scope_message_search_docs_scope_global_seq_idx (scope_id, global_seq),
    FULLTEXT KEY %s (content) WITH PARSER %s,
    CONSTRAINT fk_scope_message_search_docs_finalize_shadow_message
        FOREIGN KEY (message_id) REFERENCES scope_messages(id) ON DELETE CASCADE,
    CONSTRAINT fk_scope_message_search_docs_finalize_shadow_workspace
        FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE,
    CONSTRAINT fk_scope_message_search_docs_finalize_shadow_scope
        FOREIGN KEY (scope_id) REFERENCES scopes(id) ON DELETE CASCADE
)`,
		quoteScopeSearchIdentifier(scopeSearchFinalizeShadowTableName),
		quoteScopeSearchIdentifier(scopeSearchFullTextIndexName),
		scopeSearchFullTextParserName,
	)); err != nil {
		return fmt.Errorf("create scope message search finalize shadow table: %w", err)
	}
	return nil
}

func (store SearchStore) copyScopeSearchProjectionRows(ctx context.Context, sourceTable string, targetTable string) error {
	if _, err := store.execer.Exec(ctx, fmt.Sprintf(`
INSERT INTO %s (message_id, workspace_id, scope_id, global_seq, content, created_at)
SELECT source.message_id, source.workspace_id, source.scope_id, source.global_seq, source.content, source.created_at
FROM %s source
JOIN scopes searchable_scope
  ON searchable_scope.id = source.scope_id
 AND searchable_scope.workspace_id = source.workspace_id
 AND searchable_scope.type IN ('channel', 'dm', 'thread', 'task', 'service_account')
ON DUPLICATE KEY UPDATE
    workspace_id = VALUES(workspace_id),
    scope_id = VALUES(scope_id),
    global_seq = VALUES(global_seq),
    content = VALUES(content),
    created_at = VALUES(created_at)
`,
		quoteScopeSearchIdentifier(targetTable),
		quoteScopeSearchIdentifier(sourceTable),
	)); err != nil {
		return fmt.Errorf("copy scope search projection rows from %s to %s: %w", sourceTable, targetTable, err)
	}
	return nil
}

func (store SearchStore) swapScopeSearchProjectionTables(ctx context.Context) error {
	if _, err := store.execer.Exec(ctx, fmt.Sprintf(
		"RENAME TABLE %s TO %s, %s TO %s",
		quoteScopeSearchIdentifier(scopeSearchProjectionTableName),
		quoteScopeSearchIdentifier(scopeSearchFinalizeBackupTableName),
		quoteScopeSearchIdentifier(scopeSearchFinalizeShadowTableName),
		quoteScopeSearchIdentifier(scopeSearchProjectionTableName),
	)); err != nil {
		return fmt.Errorf("swap finalized scope message search projection table: %w", err)
	}
	return nil
}

func (store SearchStore) catchUpScopeSearchFinalizeBackup(ctx context.Context) error {
	backupExists, err := store.scopeSearchFinalizeTableExists(ctx, scopeSearchFinalizeBackupTableName)
	if err != nil {
		return err
	}
	if !backupExists {
		return nil
	}
	if err := store.copyScopeSearchProjectionRows(ctx, scopeSearchFinalizeBackupTableName, scopeSearchProjectionTableName); err != nil {
		return fmt.Errorf("catch up scope message search finalize backup: %w", err)
	}
	return nil
}

func (store SearchStore) scopeSearchFullTextIndexReadyOnTable(ctx context.Context, tableName string) (bool, error) {
	createTableSQL, err := store.showScopeSearchCreateTable(ctx, tableName)
	if err != nil {
		return false, err
	}
	return scopeSearchFullTextIndexReadyInCreateTable(createTableSQL), nil
}

func (store SearchStore) showScopeSearchCreateTable(ctx context.Context, tableName string) (string, error) {
	var ignoredName, createStmt string
	query := strings.Replace(scopeSearchShowCreateDocsQuery.SQL, "__TABLE__", quoteScopeSearchIdentifier(tableName), 1)
	if err := store.execer.QueryRow(ctx, query).Scan(&ignoredName, &createStmt); err != nil {
		return "", fmt.Errorf("show create table %s: %w", tableName, err)
	}
	return createStmt, nil
}

func scopeSearchFullTextIndexReadyInCreateTable(createTableSQL string) bool {
	normalized := strings.ToLower(strings.Join(strings.Fields(createTableSQL), " "))
	indexName := strings.ToLower(quoteScopeSearchIdentifier(scopeSearchFullTextIndexName))
	multilingualParser := strings.ToLower(scopeSearchFullTextParserName)
	return (strings.Contains(normalized, "fulltext index "+indexName) ||
		strings.Contains(normalized, "fulltext key "+indexName)) &&
		strings.Contains(normalized, "`content`") &&
		strings.Contains(normalized, "with parser "+multilingualParser)
}

func quoteScopeSearchIdentifier(identifier string) string {
	return "`" + strings.ReplaceAll(identifier, "`", "``") + "`"
}

func (store SearchStore) updateReadinessState(ctx context.Context, workspaceID string, scopeID string, projectionName string, migrationStatus string, backfillStatus string, backfillCursorGlobalSeq int64, finalizeStatus string, readReady bool, updatedAt time.Time) error {
	_, err := store.execer.Exec(ctx, scopeSearchUpdateStateQuery.SQL, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID), normalizeScopeSearchProjectionName(projectionName), strings.TrimSpace(migrationStatus), strings.TrimSpace(backfillStatus), backfillCursorGlobalSeq, strings.TrimSpace(finalizeStatus), readReady, updatedAt)
	if err != nil {
		return fmt.Errorf("update scope search readiness: %w", err)
	}
	return nil
}

func normalizeScopeSearchProjectionName(projectionName string) string {
	projectionName = strings.TrimSpace(projectionName)
	if projectionName == "" {
		return TiDBMessageSearchProjectionName
	}
	return projectionName
}

func (store ShareCardStore) LoadByMessage(ctx context.Context, scopeID string, messageID string) (ScopeShareCardRecord, error) {
	if err := store.requireExecer(); err != nil {
		return ScopeShareCardRecord{}, err
	}
	var record ScopeShareCardRecord
	var revokedAt sql.NullTime
	var payload []byte
	err := store.execer.QueryRow(ctx, scopeShareCardByMessageQuery.SQL, strings.TrimSpace(scopeID), strings.TrimSpace(messageID)).Scan(
		&record.ID,
		&record.WorkspaceID,
		&record.ScopeID,
		&record.MessageID,
		&record.SourceScopeID,
		&record.SourceMessageID,
		&record.TargetScopeID,
		&record.CreatedByMemberID,
		&revokedAt,
		&record.RevokedByMemberID,
		&payload,
		&record.CreatedAt,
		&record.UpdatedAt,
	)
	if errorsIsNoRows(err) {
		return ScopeShareCardRecord{}, notFound("scope share card not found")
	}
	if err != nil {
		return ScopeShareCardRecord{}, fmt.Errorf("load scope share card by message: %w", err)
	}
	record.RevokedAt = nullableTimePtr(revokedAt)
	record.Payload = append([]byte(nil), payload...)
	return record, nil
}

func (store ShareCardStore) Revoke(ctx context.Context, scopeID string, messageID string, revokedByMemberID string, revokedAt time.Time) (bool, error) {
	if err := store.requireExecer(); err != nil {
		return false, err
	}
	result, err := store.execer.Exec(ctx, scopeRevokeShareCardQuery.SQL, strings.TrimSpace(scopeID), strings.TrimSpace(messageID), revokedAt, strings.TrimSpace(revokedByMemberID))
	if err != nil {
		return false, fmt.Errorf("revoke scope share card: %w", err)
	}
	return result.RowsAffected() > 0, nil
}

func (store *Store) RewindScopeSuffix(ctx context.Context, request ScopeRewindSuffixRequest) (ScopeRewindSuffixResult, error) {
	if store == nil || store.db == nil {
		return ScopeRewindSuffixResult{}, fmt.Errorf("scope rewind store is nil")
	}
	var result ScopeRewindSuffixResult
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		rewound, err := RewindScopeSuffixTx(ctx, tx, request)
		if err != nil {
			return err
		}
		result = rewound
		return nil
	})
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	// Post-commit best-effort derivative cleanup. Failures are logged but do
	// not affect the rewind result — derivatives are regenerable cache.
	store.CleanupAttachmentDerivatives(ctx, result.DeletedAttachmentDerivatives)
	return result, nil
}

func RewindScopeSuffixTx(ctx context.Context, tx coredb.Tx, request ScopeRewindSuffixRequest) (ScopeRewindSuffixResult, error) {
	repos := NewScopeRepositories(tx)
	rewoundAt := request.RewoundAt.UTC()
	if rewoundAt.IsZero() {
		rewoundAt = time.Now().UTC()
	}
	messages, err := repos.Messages.ListSuffix(ctx, request.ScopeID, request.FromSeq)
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	var result ScopeRewindSuffixResult
	if len(messages) == 0 {
		latestSeq, latestGlobalSeq, err := recomputeScopeStats(ctx, tx, request.ScopeID, rewoundAt)
		if err != nil {
			return ScopeRewindSuffixResult{}, err
		}
		result.LatestSeq = latestSeq
		result.LatestGlobalSeq = latestGlobalSeq
		return result, nil
	}
	messageIDs := make([]string, 0, len(messages))
	workspaceID := strings.TrimSpace(request.WorkspaceID)
	for _, message := range messages {
		if workspaceID == "" {
			workspaceID = strings.TrimSpace(message.WorkspaceID)
		}
		if strings.TrimSpace(message.WorkspaceID) != workspaceID {
			return ScopeRewindSuffixResult{}, invalidArgument("rewind scope workspace mismatch")
		}
		messageIDs = append(messageIDs, message.ID)
	}
	result.RewoundMessageIDs = append(result.RewoundMessageIDs, messageIDs...)
	suppressionReason := firstNonEmpty(request.SuppressionReason, ScopeRewindSuppressionReason)
	suppressedActivity, err := suppressScopeActivityEventsForMessages(ctx, tx, request.ScopeID, messageIDs, rewoundAt)
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	result.SuppressedActivityCount = suppressedActivity
	suppressedOutbox, err := suppressScopeOutboxForMessages(ctx, tx, request.ScopeID, messageIDs, rewoundAt, suppressionReason)
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	result.SuppressedOutboxCount = suppressedOutbox
	rewoundMetadata, _, err := EncodeMessageMetadata(BuildMessageRewoundMetadata(rewoundAt.Format(time.RFC3339Nano)))
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	if _, err := tombstoneScopeRewindSuffixMessages(ctx, tx, request.ScopeID, messageIDs, rewoundMetadata, rewoundAt); err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	rewindDerivatives, err := scrubScopeRewindSuffixProjections(ctx, tx, request.ScopeID, messageIDs)
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	result.DeletedAttachmentDerivatives = rewindDerivatives
	if err := scrubScopeRewindDurableIMProjections(ctx, tx, workspaceID, request.ScopeID, messageIDs, rewoundAt); err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	latestSeq, latestGlobalSeq, err := recomputeScopeStats(ctx, tx, request.ScopeID, rewoundAt)
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	result.LatestSeq = latestSeq
	result.LatestGlobalSeq = latestGlobalSeq
	if err := clampScopeMembershipReadState(ctx, tx, request.ScopeID, latestSeq, latestGlobalSeq, rewoundAt); err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	clampedScopeDirty, clampedMemberDirty, err := clampScopeMessageSyncDirtyState(ctx, tx, workspaceID, request.ScopeID, latestSeq, rewoundAt)
	if err != nil {
		return ScopeRewindSuffixResult{}, err
	}
	result.ClampedScopeSyncDirtyCount = clampedScopeDirty
	result.ClampedMemberSyncDirtyCount = clampedMemberDirty
	return result, nil
}

func RecallScopeMessageTx(ctx context.Context, tx coredb.Tx, request ScopeRecallMessageRequest) (ScopeRecallMessageResult, error) {
	if tx == nil {
		return ScopeRecallMessageResult{}, invalidArgument("message recall transaction is required")
	}
	workspaceID := strings.TrimSpace(request.WorkspaceID)
	scopeID := strings.TrimSpace(request.ScopeID)
	messageID := strings.TrimSpace(request.MessageID)
	recalledByMemberID := strings.TrimSpace(request.RecalledByMemberID)
	if workspaceID == "" || scopeID == "" || messageID == "" || recalledByMemberID == "" {
		return ScopeRecallMessageResult{}, invalidArgument("message recall workspace, scope, message, and actor are required")
	}
	recalledAt := request.RecalledAt.UTC()
	if recalledAt.IsZero() {
		recalledAt = time.Now().UTC()
	}
	repos := NewScopeRepositories(tx)
	record, err := repos.Messages.Recall(ctx, ScopeMessageRecallUpdate{
		WorkspaceID:         workspaceID,
		ScopeID:             scopeID,
		MessageID:           messageID,
		Content:             strings.TrimSpace(request.PlaceholderContent),
		Metadata:            append([]byte(nil), request.Metadata...),
		RecalledByMemberID:  recalledByMemberID,
		ExpectedAuthorType:  strings.TrimSpace(request.ExpectedAuthorType),
		ExpectedAuthorID:    strings.TrimSpace(request.ExpectedAuthorID),
		ExpectedMessageType: strings.TrimSpace(request.ExpectedMessageType),
		RecalledAt:          recalledAt,
	})
	if err != nil {
		return ScopeRecallMessageResult{}, err
	}
	messageIDs := []string{record.ID}
	result := ScopeRecallMessageResult{Message: record}
	suppressedActivity, err := suppressScopeActivityEventsForMessages(ctx, tx, scopeID, messageIDs, recalledAt)
	if err != nil {
		return ScopeRecallMessageResult{}, err
	}
	result.SuppressedActivityCount = suppressedActivity
	suppressedOutbox, err := suppressScopeOutboxForMessages(ctx, tx, scopeID, messageIDs, recalledAt, ScopeRecallSuppressionReason)
	if err != nil {
		return ScopeRecallMessageResult{}, err
	}
	result.SuppressedOutboxCount = suppressedOutbox
	recallDerivatives, err := scrubScopeRecallMessageProjections(ctx, tx, scopeID, messageIDs)
	if err != nil {
		return ScopeRecallMessageResult{}, err
	}
	result.DeletedAttachmentDerivatives = recallDerivatives
	latest, err := isLatestVisibleScopeMessage(ctx, tx, workspaceID, scopeID, record.Seq)
	if err != nil {
		return ScopeRecallMessageResult{}, err
	}
	result.IsLatestVisibleMessage = latest
	if err := repos.SyncDirty.MarkMessageDirtyForScope(ctx, workspaceID, scopeID, record.Seq, ScopeRecallSuppressionReason, recalledAt); err != nil {
		return ScopeRecallMessageResult{}, err
	}
	return result, nil
}

func scrubScopeRewindSuffixProjections(ctx context.Context, tx coredb.Tx, scopeID string, messageIDs []string) ([]DeletedAttachmentDerivative, error) {
	derivatives, err := scrubScopeMessageTombstoneProjections(ctx, tx, scopeID, messageIDs, "rewind suffix")
	if err != nil {
		return nil, err
	}
	if _, err := execMessageIDStatementWithArgs(ctx, tx, `
DELETE FROM message_recovery_points
WHERE message_id IN (__MESSAGE_IDS__)
`, nil, 1, messageIDs); err != nil {
		return nil, fmt.Errorf("scrub rewind suffix message_recovery_points: %w", err)
	}
	return derivatives, nil
}

// DeletedAttachmentDerivative is a workspace+attachment pair whose thumbnail
// derivative object should be cleaned from OSS after the DB transaction commits.
type DeletedAttachmentDerivative struct {
	WorkspaceID  string
	AttachmentID string
}

// collectDeletedAttachmentDerivatives queries the attachments that will be
// deleted by the tombstone cascade (BEFORE the DELETE runs), so the caller can
// clean derivative objects post-commit. Called within the same tx.
func collectDeletedAttachmentDerivatives(ctx context.Context, tx coredb.Tx, scopeID string, messageIDs []string) ([]DeletedAttachmentDerivative, error) {
	ids := normalizeNonEmptyStrings(messageIDs)
	if len(ids) == 0 {
		return nil, nil
	}
	placeholders, idArgs := scopeSpecialtyPlaceholders(2, ids)
	query := `
SELECT workspace_id, id
FROM attachments
WHERE scope_id = $1
  AND message_id IN (` + placeholders + `)`
	args := append([]any{strings.TrimSpace(scopeID)}, idArgs...)
	rows, err := tx.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("query attachment derivatives: %w", err)
	}
	defer rows.Close()
	var result []DeletedAttachmentDerivative
	for rows.Next() {
		var d DeletedAttachmentDerivative
		if err := rows.Scan(&d.WorkspaceID, &d.AttachmentID); err != nil {
			return nil, fmt.Errorf("scan attachment derivative: %w", err)
		}
		result = append(result, d)
	}
	return result, rows.Err()
}

func scrubScopeMessageTombstoneProjections(ctx context.Context, tx coredb.Tx, scopeID string, messageIDs []string, operation string) ([]DeletedAttachmentDerivative, error) {
	operation = strings.TrimSpace(operation)
	if operation == "" {
		operation = "message tombstone"
	}
	// Collect derivative targets BEFORE the attachments DELETE step runs.
	// If collection fails, return error so the tx rolls back — we must not
	// delete DB rows without knowing which derivatives to clean.
	derivatives, err := collectDeletedAttachmentDerivatives(ctx, tx, scopeID, messageIDs)
	if err != nil {
		return nil, fmt.Errorf("collect %s attachment derivatives: %w", operation, err)
	}
	for _, step := range scopeMessageTombstoneProjectionSteps {
		if _, err := execMessageIDStatement(ctx, tx, step.statement, scopeID, messageIDs); err != nil {
			return nil, fmt.Errorf("scrub %s %s projection: %w", operation, step.name, err)
		}
	}
	return derivatives, nil
}

func tombstoneScopeRewindSuffixMessages(ctx context.Context, tx coredb.Tx, scopeID string, messageIDs []string, metadata []byte, rewoundAt time.Time) (int64, error) {
	result, err := execMessageIDStatementWithArgs(ctx, tx, `
`+strings.TrimSpace(scopeMessageTombstoneTranscriptStep.statement)+`
`, []any{strings.TrimSpace(scopeID), nullableJSON(metadata), rewoundAt.UTC()}, 4, messageIDs)
	if err != nil {
		return 0, fmt.Errorf("tombstone rewind suffix messages: %w", err)
	}
	return result.RowsAffected(), nil
}

func scrubScopeRewindDurableIMProjections(ctx context.Context, tx coredb.Tx, workspaceID string, scopeID string, messageIDs []string, rewoundAt time.Time) error {
	rewoundAtString := rewoundAt.UTC().Format(time.RFC3339Nano)
	rewoundMetadataSQL := `
JSON_OBJECT(
  'messageRecall',
  JSON_OBJECT(
    'status', $3,
    'rewoundAt', $4
  )
)`
	if _, err := execMessageIDStatementWithArgs(ctx, tx, `
UPDATE im_stream_events
SET payload_json = JSON_SET(
  JSON_REMOVE(
    payload_json,
    '$.mention',
    '$.mentionRecipients',
    '$.message.resolvedMentions',
    '$.message.attachments',
    '$.message.reactions',
    '$.message.source',
    '$.message.deliverySummaryPreview'
  ),
  '$.content', '',
  '$.metadata', `+rewoundMetadataSQL+`,
  '$.attachments', JSON_ARRAY(),
  '$.reactions', JSON_ARRAY(),
  '$.message.content', '',
  '$.message.metadata', `+rewoundMetadataSQL+`,
  '$.message.attachments', JSON_ARRAY(),
  '$.message.reactions', JSON_ARRAY()
)
WHERE workspace_id = $1
  AND scope_id = $2
  AND entity_id IN (__MESSAGE_IDS__)
  AND type IN ('message.created', 'message.updated')
`, []any{strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID), MessageLifecycleStatusRewound, rewoundAtString}, 5, messageIDs); err != nil {
		return fmt.Errorf("scrub rewind im stream projections: %w", err)
	}
	if _, err := execMessageIDStatementWithArgs(ctx, tx, `
UPDATE im_conversation_index
SET projection_json = JSON_SET(
  JSON_REMOVE(
    projection_json,
    '$.resolvedMentions',
    '$.attachments',
    '$.reactions',
    '$.source',
    '$.deliverySummaryPreview',
    '$.lastMessage.resolvedMentions',
    '$.lastMessage.attachments',
    '$.lastMessage.reactions',
    '$.lastMessage.source',
    '$.lastMessage.deliverySummaryPreview'
  ),
  '$.content', '',
  '$.metadata', `+rewoundMetadataSQL+`,
  '$.attachments', JSON_ARRAY(),
  '$.reactions', JSON_ARRAY(),
  '$.lastMessage.content', '',
  '$.lastMessage.metadata', `+rewoundMetadataSQL+`,
  '$.lastMessage.attachments', JSON_ARRAY(),
  '$.lastMessage.reactions', JSON_ARRAY()
)
WHERE workspace_id = $1
  AND scope_id = $2
  AND JSON_UNQUOTE(JSON_EXTRACT(projection_json, '$.lastMessage.id')) IN (__MESSAGE_IDS__)
`, []any{strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID), MessageLifecycleStatusRewound, rewoundAtString}, 5, messageIDs); err != nil {
		return fmt.Errorf("scrub rewind im conversation projection: %w", err)
	}
	return nil
}

func scrubScopeRecallMessageProjections(ctx context.Context, tx coredb.Tx, scopeID string, messageIDs []string) ([]DeletedAttachmentDerivative, error) {
	return scrubScopeMessageTombstoneProjections(ctx, tx, scopeID, messageIDs, "recall")
}

func isLatestVisibleScopeMessage(ctx context.Context, queryer coredb.Queryer, workspaceID string, scopeID string, seq int64) (bool, error) {
	if strings.TrimSpace(workspaceID) == "" || strings.TrimSpace(scopeID) == "" || seq <= 0 {
		return false, invalidArgument("latest-visible message check requires workspace, scope, and seq")
	}
	var latest bool
	if err := queryer.QueryRow(ctx, `
SELECT NOT EXISTS (
  SELECT 1
  FROM scope_messages
  WHERE workspace_id = $1
    AND scope_id = $2
    AND seq > $3
    AND `+HumanVisibleMessageTypeSQL("scope_messages")+`
)
`, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID), seq).Scan(&latest); err != nil {
		return false, fmt.Errorf("check latest visible scope message: %w", err)
	}
	return latest, nil
}

func (store MessageStore) ListSuffix(ctx context.Context, scopeID string, fromSeq int64) ([]ScopeMessageRecord, error) {
	if err := store.requireExecer(); err != nil {
		return nil, err
	}
	if strings.TrimSpace(scopeID) == "" || fromSeq <= 0 {
		return nil, invalidArgument("rewind suffix scope and positive from seq are required")
	}
	rows, err := store.execer.Query(ctx, scopeRewindSuffixMessagesQuery.SQL, strings.TrimSpace(scopeID), fromSeq)
	if err != nil {
		return nil, fmt.Errorf("list scope rewind suffix messages: %w", err)
	}
	defer rows.Close()
	messages := make([]ScopeMessageRecord, 0)
	for rows.Next() {
		var record ScopeMessageRecord
		var metadata []byte
		var editedAt sql.NullTime
		if err := rows.Scan(
			&record.ID,
			&record.WorkspaceID,
			&record.ScopeID,
			&record.Seq,
			&record.WorkspaceGlobalSeq,
			&record.AuthorMemberID,
			&record.AuthorType,
			&record.NotifyPrincipalMemberID,
			&record.IdempotencySubjectMemberID,
			&record.ClientMessageID,
			&record.IdempotencyPayloadDigest,
			&record.MessageType,
			&record.Content,
			&metadata,
			&record.CreatedAt,
			&record.UpdatedAt,
			&editedAt,
			&record.EditedByMemberID,
			&record.EditRevision,
		); err != nil {
			return nil, fmt.Errorf("scan scope rewind suffix message: %w", err)
		}
		record.EditedAt = nullableTimePtr(editedAt)
		record.Metadata = append([]byte(nil), metadata...)
		messages = append(messages, record)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate scope rewind suffix messages: %w", err)
	}
	return messages, nil
}

func suppressScopeActivityEventsForMessages(ctx context.Context, tx coredb.Tx, scopeID string, messageIDs []string, suppressedAt time.Time) (int64, error) {
	statement := `
UPDATE scope_activity_events
SET suppressed_at = $2
WHERE scope_id = $1
  AND suppressed_at IS NULL
  AND (
    source_message_id IN (__MESSAGE_IDS__)
    OR activity_ref_id IN (
      SELECT id
      FROM scope_message_reactions
      WHERE scope_id = $1
        AND message_id IN (__MESSAGE_IDS__)
    )
  )
`
	result, err := execMessageIDStatementWithArgs(ctx, tx, statement, []any{strings.TrimSpace(scopeID), suppressedAt}, 3, messageIDs)
	if err != nil {
		return 0, fmt.Errorf("suppress scope activity events for rewind: %w", err)
	}
	return result.RowsAffected(), nil
}

func suppressScopeOutboxForMessages(ctx context.Context, tx coredb.Tx, scopeID string, messageIDs []string, suppressedAt time.Time, suppressionReason string) (int64, error) {
	statement := `
UPDATE scope_fanout_outbox
SET status = 'suppressed',
    suppressed_at = $2,
    suppression_reason = $3
WHERE scope_id = $1
  AND status IN ('pending', 'leased', 'pending_v2', 'leased_v2')
  AND (
    message_id IN (__MESSAGE_IDS__)
    OR (event_ref_type = 'message' AND event_ref_id IN (__MESSAGE_IDS__))
    OR (
      event_ref_type = 'reaction'
      AND event_ref_id IN (
        SELECT id
        FROM scope_message_reactions
        WHERE scope_id = $1
          AND message_id IN (__MESSAGE_IDS__)
      )
    )
    OR (
      event_ref_type = 'activity_event'
      AND event_ref_id IN (
        SELECT id
        FROM scope_activity_events
        WHERE scope_id = $1
          AND (
            source_message_id IN (__MESSAGE_IDS__)
            OR activity_ref_id IN (
              SELECT id
              FROM scope_message_reactions
              WHERE scope_id = $1
                AND message_id IN (__MESSAGE_IDS__)
            )
          )
      )
    )
  )
`
	result, err := execMessageIDStatementWithArgs(ctx, tx, statement, []any{strings.TrimSpace(scopeID), suppressedAt, strings.TrimSpace(suppressionReason)}, 4, messageIDs)
	if err != nil {
		return 0, fmt.Errorf("suppress scope outbox for rewind: %w", err)
	}
	return result.RowsAffected(), nil
}

func clampScopeMessageSyncDirtyState(ctx context.Context, tx coredb.Tx, workspaceID string, scopeID string, latestSeq int64, updatedAt time.Time) (int64, int64, error) {
	if latestSeq < 0 {
		latestSeq = 0
	}
	trimmedWorkspaceID := strings.TrimSpace(workspaceID)
	trimmedScopeID := strings.TrimSpace(scopeID)
	scopeResult, err := tx.Exec(ctx, `
UPDATE scope_sync_dirty
SET latest_message_seq = $3,
    updated_at = $4
WHERE workspace_id = $1
  AND scope_id = $2
  AND latest_message_seq > $3
`, trimmedWorkspaceID, trimmedScopeID, latestSeq, updatedAt)
	if err != nil {
		return 0, 0, fmt.Errorf("clamp scope sync dirty for rewind: %w", err)
	}
	memberResult, err := tx.Exec(ctx, `
UPDATE member_scope_sync_dirty
SET latest_message_seq = $3,
    updated_at = $4
WHERE workspace_id = $1
  AND scope_id = $2
  AND latest_message_seq > $3
`, trimmedWorkspaceID, trimmedScopeID, latestSeq, updatedAt)
	if err != nil {
		return 0, 0, fmt.Errorf("clamp member sync dirty for rewind: %w", err)
	}
	return scopeResult.RowsAffected(), memberResult.RowsAffected(), nil
}

func recomputeScopeStats(ctx context.Context, tx coredb.Tx, scopeID string, updatedAt time.Time) (int64, int64, error) {
	if _, err := tx.Exec(ctx, scopeRewindRecomputeStatsQuery.SQL, strings.TrimSpace(scopeID), updatedAt); err != nil {
		return 0, 0, fmt.Errorf("recompute scope rewind stats: %w", err)
	}
	var latestSeq, latestGlobalSeq int64
	if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(seq), 0), COALESCE(MAX(workspace_global_seq), 0)
FROM scope_messages
WHERE scope_id = $1
  AND `+HumanVisibleMessageTypeSQL("scope_messages")+`
`, strings.TrimSpace(scopeID)).Scan(&latestSeq, &latestGlobalSeq); err != nil {
		return 0, 0, fmt.Errorf("load recomputed scope stats: %w", err)
	}
	return latestSeq, latestGlobalSeq, nil
}

func clampScopeMembershipReadState(ctx context.Context, tx coredb.Tx, scopeID string, latestSeq int64, latestGlobalSeq int64, updatedAt time.Time) error {
	if _, err := tx.Exec(ctx, scopeRewindClampMembershipReadStateQuery.SQL, strings.TrimSpace(scopeID), latestSeq, latestGlobalSeq, updatedAt); err != nil {
		return fmt.Errorf("clamp scope membership read state after rewind: %w", err)
	}
	return nil
}

func execMessageIDStatement(ctx context.Context, execer coredb.QueryExecer, statement string, scopeID string, messageIDs []string) (coredb.Result, error) {
	return execMessageIDStatementWithArgs(ctx, execer, statement, []any{strings.TrimSpace(scopeID)}, 2, messageIDs)
}

func execMessageIDStatementWithArgs(ctx context.Context, execer coredb.QueryExecer, statement string, prefixArgs []any, start int, messageIDs []string) (coredb.Result, error) {
	ids := normalizeNonEmptyStrings(messageIDs)
	if len(ids) == 0 {
		return scopeSpecialtyRowsAffectedResult(0), nil
	}
	placeholders, idArgs := scopeSpecialtyPlaceholders(start, ids)
	query := strings.ReplaceAll(statement, "__MESSAGE_IDS__", placeholders)
	args := append(prefixArgs, idArgs...)
	return execer.Exec(ctx, query, args...)
}

type scopeSpecialtyRowsAffectedResult int64

func (result scopeSpecialtyRowsAffectedResult) RowsAffected() int64 {
	return int64(result)
}

func scopeSpecialtyPlaceholders(start int, values []string) (string, []any) {
	placeholders := make([]string, len(values))
	args := make([]any, len(values))
	for index, value := range values {
		placeholders[index] = fmt.Sprintf("$%d", start+index)
		args[index] = value
	}
	return strings.Join(placeholders, ", "), args
}

func errorsIsNoRows(err error) bool {
	return errors.Is(err, sql.ErrNoRows)
}