package message

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

	"github.com/stretchr/testify/require"

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

func TestScopeReactionStoreAddRollsBackWhenActivityInsertFails(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_reaction_add_rollback")
	message := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_reaction_add_rollback", 1, 100)
	require.NoError(t, repos.Messages.Insert(ctx, message))

	reaction := ScopeMessageReactionRecord{
		ID:                "rxn_scope_specialty_reaction_add_rollback",
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		MessageID:         message.ID,
		Emoji:             "👀",
		ReactorMemberID:   fixture.ownerMemberID,
		ActivitySeq:       101,
		ActivityGlobalSeq: 101,
		CreatedAt:         fixture.now,
	}
	failingRepos := NewScopeRepositories(scopeReactionFailActivityInsertDB{DB: db})
	_, err := failingRepos.Reactions.Add(ctx, reaction)
	require.ErrorContains(t, err, "add scope reaction activity")
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_reactions", "id = $1", reaction.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_activity_events", "activity_ref_id = $1", reaction.ID))
}

func TestScopeSearchReadinessAndShareCardLifecycle(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_share")
	message := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_share", 1, 100)
	source := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_source", 2, 101)
	require.NoError(t, repos.Messages.Insert(ctx, message))
	require.NoError(t, repos.Messages.Insert(ctx, source))

	_, err := repos.Search.LoadReadiness(ctx, fixture.workspaceID, fixture.scopeID, TiDBMessageSearchProjectionName)
	require.ErrorIs(t, err, ErrNotFound)
	readiness, err := repos.Search.BeginBackfill(ctx, fixture.workspaceID, fixture.scopeID, "", fixture.now)
	require.NoError(t, err)
	require.False(t, readiness.Ready())
	require.Equal(t, "running", readiness.BackfillStatus)

	backfill, err := repos.Search.Backfill(ctx, fixture.workspaceID, fixture.scopeID, "", 1, fixture.now.Add(time.Minute))
	require.NoError(t, err)
	require.Equal(t, 1, backfill.ProcessedRows)
	require.Equal(t, "running", backfill.State.BackfillStatus)
	require.Equal(t, message.Seq, backfill.State.BackfillCursorGlobalSeq)

	backfill, err = repos.Search.Backfill(ctx, fixture.workspaceID, fixture.scopeID, "", 10, fixture.now.Add(2*time.Minute))
	require.NoError(t, err)
	require.Equal(t, 1, backfill.ProcessedRows)
	require.Equal(t, "complete", backfill.State.BackfillStatus)
	require.Equal(t, source.Seq, backfill.State.BackfillCursorGlobalSeq)

	_, err = repos.Search.Finalize(ctx, fixture.workspaceID, fixture.scopeID, "", fixture.now.Add(3*time.Minute))
	require.Error(t, err)
	require.Contains(t, err.Error(), "scope message search")
	readiness, err = repos.Search.LoadReadiness(ctx, fixture.workspaceID, fixture.scopeID, "")
	require.NoError(t, err)
	require.False(t, readiness.Ready())
	require.Equal(t, "pending", readiness.FinalizeStatus)
	require.Equal(t, source.Seq, readiness.BackfillCursorGlobalSeq)

	require.NoError(t, repos.Search.UpsertState(ctx, ScopeSearchStateRecord{
		WorkspaceID:             fixture.workspaceID,
		ScopeID:                 fixture.scopeID,
		ProjectionName:          TiDBMessageSearchProjectionName,
		MigrationStatus:         "complete",
		BackfillStatus:          "complete",
		BackfillCursorGlobalSeq: source.WorkspaceGlobalSeq,
		FinalizeStatus:          "complete",
		ReadReady:               true,
		CreatedAt:               fixture.now,
		UpdatedAt:               fixture.now.Add(4 * time.Minute),
	}))
	readiness, err = repos.Search.LoadReadiness(ctx, fixture.workspaceID, fixture.scopeID, "")
	require.NoError(t, err)
	require.True(t, readiness.Ready())

	readiness, err = repos.Search.BeginBackfill(ctx, fixture.workspaceID, fixture.scopeID, "", fixture.now.Add(5*time.Minute))
	require.NoError(t, err)
	require.True(t, readiness.Ready())
	require.Equal(t, "complete", readiness.FinalizeStatus)

	backfill, err = repos.Search.Backfill(ctx, fixture.workspaceID, fixture.scopeID, "", 10, fixture.now.Add(6*time.Minute))
	require.NoError(t, err)
	require.Zero(t, backfill.ProcessedRows)
	require.True(t, backfill.State.Ready())
	require.Equal(t, "complete", backfill.State.FinalizeStatus)

	readiness, err = repos.Search.Finalize(ctx, fixture.workspaceID, fixture.scopeID, "", fixture.now.Add(7*time.Minute))
	require.Error(t, err)
	require.Contains(t, err.Error(), "scope message search")
	require.False(t, readiness.Ready())
	card := ScopeShareCardRecord{
		ID:                "shr_scope_specialty_share",
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		MessageID:         message.ID,
		SourceScopeID:     fixture.scopeID,
		SourceMessageID:   source.ID,
		TargetScopeID:     fixture.scopeID,
		CreatedByMemberID: fixture.ownerMemberID,
		Payload:           []byte(`{"kind":"share"}`),
		CreatedAt:         fixture.now,
		UpdatedAt:         fixture.now,
	}
	require.NoError(t, repos.ShareCards.Insert(ctx, card))
	loaded, err := repos.ShareCards.LoadByMessage(ctx, fixture.scopeID, message.ID)
	require.NoError(t, err)
	require.Equal(t, card.ID, loaded.ID)
	require.Equal(t, source.ID, loaded.SourceMessageID)
	require.Nil(t, loaded.RevokedAt)
	require.JSONEq(t, `{"kind":"share"}`, string(loaded.Payload))

	revokedAt := fixture.now.Add(time.Minute)
	changed, err := repos.ShareCards.Revoke(ctx, fixture.scopeID, message.ID, fixture.ownerMemberID, revokedAt)
	require.NoError(t, err)
	require.True(t, changed)
	loaded, err = repos.ShareCards.LoadByMessage(ctx, fixture.scopeID, message.ID)
	require.NoError(t, err)
	require.NotNil(t, loaded.RevokedAt)
	require.Equal(t, revokedAt, *loaded.RevokedAt)
	require.Equal(t, fixture.ownerMemberID, loaded.RevokedByMemberID)

	changed, err = repos.ShareCards.Revoke(ctx, fixture.scopeID, message.ID, fixture.ownerMemberID, revokedAt.Add(time.Minute))
	require.NoError(t, err)
	require.False(t, changed)
}

func TestScopeSearchReadinessRequiresCompleteLifecycle(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_search_gate")

	require.NoError(t, repos.Search.UpsertState(ctx, ScopeSearchStateRecord{
		WorkspaceID:             fixture.workspaceID,
		ScopeID:                 fixture.scopeID,
		ProjectionName:          TiDBMessageSearchProjectionName,
		MigrationStatus:         "complete",
		BackfillStatus:          "running",
		BackfillCursorGlobalSeq: 7,
		FinalizeStatus:          "complete",
		ReadReady:               true,
		CreatedAt:               fixture.now,
		UpdatedAt:               fixture.now,
	}))

	err := repos.Search.RequireReadReady(ctx, fixture.workspaceID, fixture.scopeID)
	require.ErrorIs(t, err, ErrMessageSearchUnavailable)
}

func TestScopeSearchReadinessFallsBackToWorkspaceGlobalRow(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_search_workspace_fallback")

	require.NoError(t, repos.Search.UpsertState(ctx, ScopeSearchStateRecord{
		WorkspaceID:             fixture.workspaceID,
		ScopeID:                 "",
		ProjectionName:          TiDBMessageSearchProjectionName,
		MigrationStatus:         "complete",
		BackfillStatus:          "complete",
		BackfillCursorGlobalSeq: 99,
		FinalizeStatus:          "complete",
		ReadReady:               true,
		CreatedAt:               fixture.now,
		UpdatedAt:               fixture.now,
	}))

	require.NoError(t, repos.Search.RequireReadReady(ctx, fixture.workspaceID, fixture.scopeID))
	require.NoError(t, repos.Search.RequireReadReady(ctx, fixture.workspaceID, ""))
}

func TestScopeSearchBackfillSupportsWorkspaceGlobalTarget(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_search_workspace_global")

	secondScopeID := fixture.scopeID + "_other"
	require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                secondScopeID,
		WorkspaceID:       fixture.workspaceID,
		Type:              "channel",
		IdentityKey:       "channel:scope-store:specialty-search-workspace-global:other",
		Status:            "active",
		CreatedByMemberID: fixture.ownerMemberID,
		CreatedAt:         fixture.now,
		UpdatedAt:         fixture.now,
	}))
	require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{
		ScopeID:    secondScopeID,
		MemberID:   fixture.ownerMemberID,
		Visibility: "public",
		CreatedAt:  fixture.now,
		UpdatedAt:  fixture.now,
	}))

	first := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_workspace_global_first", 1, 100)
	second := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_workspace_global_second", 2, 101)
	second.ScopeID = secondScopeID
	require.NoError(t, repos.Messages.Insert(ctx, first))
	require.NoError(t, repos.Messages.Insert(ctx, second))

	_, err := repos.Search.LoadReadiness(ctx, fixture.workspaceID, "", TiDBMessageSearchProjectionName)
	require.ErrorIs(t, err, ErrNotFound)

	readiness, err := repos.Search.BeginBackfill(ctx, fixture.workspaceID, "", "", fixture.now)
	require.NoError(t, err)
	require.Equal(t, "", readiness.ScopeID)
	require.Equal(t, "running", readiness.BackfillStatus)

	backfill, err := repos.Search.Backfill(ctx, fixture.workspaceID, "", "", 1, fixture.now.Add(time.Minute))
	require.NoError(t, err)
	require.Equal(t, 1, backfill.ProcessedRows)
	require.Equal(t, "running", backfill.State.BackfillStatus)
	require.Equal(t, first.WorkspaceGlobalSeq, backfill.State.BackfillCursorGlobalSeq)

	backfill, err = repos.Search.Backfill(ctx, fixture.workspaceID, "", "", 10, fixture.now.Add(2*time.Minute))
	require.NoError(t, err)
	require.Equal(t, 1, backfill.ProcessedRows)
	require.Equal(t, "complete", backfill.State.BackfillStatus)
	require.Equal(t, second.WorkspaceGlobalSeq, backfill.State.BackfillCursorGlobalSeq)
	require.Equal(t, "", backfill.State.ScopeID)
	require.Equal(t, int64(2), countScopeRows(t, ctx, db, "scope_message_search_docs", "workspace_id = $1", fixture.workspaceID))
}

func TestScopeSearchBackfillRepairsMixedHistoricalAndCompatibilitySeqRows(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_search_mixed_cursor")

	historical := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_search_old", 1, 100)
	compat := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_search_new", 2, 2)
	require.NoError(t, repos.Messages.Insert(ctx, historical))
	require.NoError(t, repos.Messages.Insert(ctx, compat))
	require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{
		MessageID:   historical.ID,
		WorkspaceID: historical.WorkspaceID,
		ScopeID:     historical.ScopeID,
		GlobalSeq:   historical.WorkspaceGlobalSeq,
		Content:     historical.Content,
		CreatedAt:   historical.CreatedAt,
	}))
	require.NoError(t, repos.Search.UpsertState(ctx, ScopeSearchStateRecord{
		WorkspaceID:             fixture.workspaceID,
		ScopeID:                 fixture.scopeID,
		ProjectionName:          TiDBMessageSearchProjectionName,
		MigrationStatus:         "complete",
		BackfillStatus:          "running",
		BackfillCursorGlobalSeq: historical.WorkspaceGlobalSeq,
		FinalizeStatus:          "pending",
		ReadReady:               false,
		CreatedAt:               fixture.now,
		UpdatedAt:               fixture.now,
	}))

	backfill, err := repos.Search.Backfill(ctx, fixture.workspaceID, fixture.scopeID, "", 10, fixture.now.Add(time.Minute))
	require.NoError(t, err)
	require.Equal(t, 1, backfill.ProcessedRows)
	require.Equal(t, "complete", backfill.State.BackfillStatus)
	require.Equal(t, compat.Seq, backfill.State.BackfillCursorGlobalSeq)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_message_search_docs", "message_id = $1 AND global_seq = $2", compat.ID, compat.WorkspaceGlobalSeq))
}

func TestScopeSearchBackfillRepairsReadyProjectionWithoutDowngradingReadiness(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_search_ready_repair")

	historical := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_ready_old", 1, 100)
	compat := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_ready_new", 2, 2)
	require.NoError(t, repos.Messages.Insert(ctx, historical))
	require.NoError(t, repos.Messages.Insert(ctx, compat))
	require.NoError(t, repos.Search.UpsertState(ctx, ScopeSearchStateRecord{
		WorkspaceID:             fixture.workspaceID,
		ScopeID:                 fixture.scopeID,
		ProjectionName:          TiDBMessageSearchProjectionName,
		MigrationStatus:         "complete",
		BackfillStatus:          "complete",
		BackfillCursorGlobalSeq: historical.WorkspaceGlobalSeq,
		FinalizeStatus:          "complete",
		ReadReady:               true,
		CreatedAt:               fixture.now,
		UpdatedAt:               fixture.now,
	}))

	first, err := repos.Search.Backfill(ctx, fixture.workspaceID, fixture.scopeID, "", 1, fixture.now.Add(time.Minute))
	require.NoError(t, err)
	require.Equal(t, 1, first.ProcessedRows)
	require.Equal(t, "running", first.State.BackfillStatus)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_message_search_docs", "scope_id = $1", fixture.scopeID))

	persisted, err := repos.Search.LoadReadiness(ctx, fixture.workspaceID, fixture.scopeID, "")
	require.NoError(t, err)
	require.True(t, persisted.Ready())
	require.Equal(t, "complete", persisted.BackfillStatus)
	require.Equal(t, "complete", persisted.FinalizeStatus)

	second, err := repos.Search.Backfill(ctx, fixture.workspaceID, fixture.scopeID, "", 1, fixture.now.Add(2*time.Minute))
	require.NoError(t, err)
	require.Equal(t, 1, second.ProcessedRows)
	require.Equal(t, "complete", second.State.BackfillStatus)
	require.Equal(t, compat.Seq, second.State.BackfillCursorGlobalSeq)
	require.Equal(t, int64(2), countScopeRows(t, ctx, db, "scope_message_search_docs", "scope_id = $1", fixture.scopeID))

	persisted, err = repos.Search.LoadReadiness(ctx, fixture.workspaceID, fixture.scopeID, "")
	require.NoError(t, err)
	require.True(t, persisted.Ready())
	require.Equal(t, "complete", persisted.BackfillStatus)
	require.Equal(t, compat.Seq, persisted.BackfillCursorGlobalSeq)
}

func TestScopeSearchFullTextIndexReadyInCreateTableRequiresMultilingualParser(t *testing.T) {
	t.Parallel()

	require.False(t, scopeSearchFullTextIndexReadyInCreateTable("CREATE TABLE `scope_message_search_docs` ()"))
	require.False(t, scopeSearchFullTextIndexReadyInCreateTable(
		"CREATE TABLE `scope_message_search_docs` (`content` longtext, FULLTEXT KEY `scope_message_search_docs_content_fulltext_idx` (`content`) WITH PARSER STANDARD)",
	))
	require.True(t, scopeSearchFullTextIndexReadyInCreateTable(
		"CREATE TABLE `scope_message_search_docs` (`content` longtext, FULLTEXT KEY `scope_message_search_docs_content_fulltext_idx` (`content`) WITH PARSER MULTILINGUAL)",
	))
	require.True(t, scopeSearchFullTextIndexReadyInCreateTable(
		"CREATE TABLE `scope_message_search_docs` (`content` longtext, FULLTEXT INDEX `scope_message_search_docs_content_fulltext_idx` (`content`) WITH PARSER MULTILINGUAL)",
	))
}

func TestScopeMessageTombstoneProjectionLifecycleIsExplicit(t *testing.T) {
	t.Parallel()

	readVisibilitySteps := append(
		append([]scopeMessageTombstoneProjectionStep{}, scopeMessageTombstoneProjectionSteps...),
		scopeMessageTombstoneTranscriptStep,
	)
	stepNames := make([]string, 0, len(readVisibilitySteps))
	stepStatements := make(map[string]string, len(readVisibilitySteps))
	stepImpactedTables := make(map[string]string, len(readVisibilitySteps))
	stepReadPaths := make(map[string]string, len(readVisibilitySteps))
	for _, step := range readVisibilitySteps {
		stepNames = append(stepNames, step.name)
		stepStatements[step.name] = step.statement
		stepImpactedTables[step.name] = step.impactedTable
		stepReadPaths[step.name] = step.readPathLabel
	}

	require.Equal(t, []string{
		"share_cards",
		"message_reactions",
		"message_pins",
		"message_top_notices",
		"message_search_docs",
		"message_sources",
		"attachments",
		"channel_attachment_projection",
		"scope_messages",
	}, stepNames)
	require.Contains(t, stepStatements["share_cards"], "DELETE FROM scope_share_cards")
	require.Contains(t, stepStatements["message_reactions"], "DELETE FROM scope_message_reactions")
	require.Contains(t, stepStatements["message_pins"], "DELETE FROM scope_message_pins")
	require.Contains(t, stepStatements["message_top_notices"], "DELETE FROM scope_message_top_notices")
	require.Contains(t, stepStatements["message_search_docs"], "DELETE FROM scope_message_search_docs")
	require.Contains(t, stepStatements["message_sources"], "target_scope_id = $1")
	require.Contains(t, stepStatements["message_sources"], "source_scope_id = $1")
	require.Contains(t, stepStatements["message_sources"], "source_message_id IN")
	require.Contains(t, stepStatements["attachments"], "DELETE FROM attachments")
	// Verify the DELETE scope matches what collectDeletedAttachmentDerivatives
	// queries, so the pre-delete collection sees exactly the rows about to be
	// removed (no missed derivatives, no phantom entries).
	require.Contains(t, stepStatements["attachments"], "scope_id = $1")
	require.Contains(t, stepStatements["attachments"], "message_id IN")
	require.Contains(t, stepStatements["channel_attachment_projection"], "DELETE FROM channel_attachment_projection")
	require.Contains(t, stepStatements["scope_messages"], "UPDATE scope_messages")
	require.NotContains(t, stepStatements["scope_messages"], "DELETE FROM scope_messages")
	require.Equal(t, "scope_messages", stepNames[len(stepNames)-1], "transcript rows are tombstoned after read/search/source projections")
	require.Equal(t, map[string]string{
		"share_cards":                   "scope_share_cards",
		"message_reactions":             "scope_message_reactions",
		"message_pins":                  "scope_message_pins",
		"message_top_notices":           "scope_message_top_notices",
		"message_search_docs":           "scope_message_search_docs",
		"message_sources":               "scope_message_sources",
		"attachments":                   "attachments",
		"channel_attachment_projection": "channel_attachment_projection",
		"scope_messages":                "scope_messages",
	}, stepImpactedTables)
	require.Equal(t, map[string]string{
		"share_cards":                   "share-card action-state reads",
		"message_reactions":             "reaction aggregate reads",
		"message_pins":                  "message pin list reads",
		"message_top_notices":           "message top banner reads",
		"message_search_docs":           "search reads",
		"message_sources":               "root/source context reads",
		"attachments":                   "attachment list/download reads",
		"channel_attachment_projection": "channel attachment list reads",
		"scope_messages":                "history/sync/attachment/diagnostics transcript authority",
	}, stepReadPaths)
	for _, step := range scopeMessageTombstoneProjectionSteps {
		require.NotEmpty(t, step.impactedTable)
		require.NotEmpty(t, step.readPathLabel)
		require.NotContains(t, step.statement, "DELETE FROM scope_messages", "projection impact steps must not delete transcript rows")
	}
}

// TestCollectDeletedAttachmentDerivativesQueriesCorrectRows verifies that the
// pre-DELETE collection query matches the scope of the tombstone cascade —
// it should see exactly the attachments that are about to be deleted.
func TestCollectDeletedAttachmentDerivativesQueriesCorrectRows(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	fixture := seedScopeStoreFixture(t, ctx, db, "collect_derivatives")
	messageOne := scopeSpecialtyMessageForTest(fixture, "msg_collect_keep", 1, 100)
	messageTwo := scopeSpecialtyMessageForTest(fixture, "msg_collect_delete", 2, 200)
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, messageOne))
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, messageTwo))
	// Attach a row to messageTwo (the one that will be rewound).
	mustExecMessaging(t, ctx, db, `
INSERT INTO attachments (id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, created_at)
VALUES ($1, $2, $3, $4, $5, 'photo.png', 'image/png', 1024, NOW())
`, "att_collect_test", fixture.workspaceID, fixture.scopeID, fixture.ownerMemberID, messageTwo.ID)
	// Attach a row to messageOne (should NOT be collected — it's not being rewound).
	mustExecMessaging(t, ctx, db, `
INSERT INTO attachments (id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, created_at)
VALUES ($1, $2, $3, $4, $5, 'keep.png', 'image/png', 512, NOW())
`, "att_keep_test", fixture.workspaceID, fixture.scopeID, fixture.ownerMemberID, messageOne.ID)

	err := coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		derivatives, err := collectDeletedAttachmentDerivatives(ctx, tx, fixture.scopeID, []string{messageTwo.ID})
		require.NoError(t, err)
		require.Len(t, derivatives, 1, "should collect exactly the attachment being deleted")
		require.Equal(t, "att_collect_test", derivatives[0].AttachmentID)
		require.Equal(t, fixture.workspaceID, derivatives[0].WorkspaceID)
		// Verify the keep attachment is NOT collected.
		for _, d := range derivatives {
			require.NotEqual(t, "att_keep_test", d.AttachmentID, "keep attachment must not be collected")
		}
		return nil
	})
	require.NoError(t, err)
}

func TestScrubScopeMessageTombstoneProjectionsReturnsDerivatives(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	fixture := seedScopeStoreFixture(t, ctx, db, "scrub_derivatives")
	messageOne := scopeSpecialtyMessageForTest(fixture, "msg_scrub_keep", 1, 100)
	messageTwo := scopeSpecialtyMessageForTest(fixture, "msg_scrub_delete", 2, 200)
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, messageOne))
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, messageTwo))
	mustExecMessaging(t, ctx, db, `
INSERT INTO attachments (id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, created_at)
VALUES ($1, $2, $3, $4, $5, 'photo.png', 'image/png', 1024, NOW())
`, "att_scrub_test", fixture.workspaceID, fixture.scopeID, fixture.ownerMemberID, messageTwo.ID)

	err := coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		derivatives, err := scrubScopeMessageTombstoneProjections(ctx, tx, fixture.scopeID, []string{messageTwo.ID}, "test scrub")
		require.NoError(t, err)
		require.Len(t, derivatives, 1, "should return the deleted attachment's derivative target")
		require.Equal(t, "att_scrub_test", derivatives[0].AttachmentID)
		// Verify the attachment was actually deleted by the SQL cascade.
		var count int
		require.NoError(t, tx.QueryRow(ctx, `SELECT COUNT(*) FROM attachments WHERE id = $1`, "att_scrub_test").Scan(&count))
		require.Zero(t, count, "attachment should be deleted after scrub")
		return nil
	})
	require.NoError(t, err)
}

// TestRewindScopeSuffixCallsDerivativeCleanerPostCommit verifies that the
// DeleteAttachmentDerivatives callback fires AFTER the tx commits (not during),
// with the correct (workspaceID, attachmentID) pairs for the rewound message's
// attachments.
func TestRewindScopeSuffixCallsDerivativeCleanerPostCommit(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "rewind_cleanup")
	messageOne := scopeSpecialtyMessageForTest(fixture, "msg_rewind_cleanup_keep", 1, 100)
	messageTwo := scopeSpecialtyMessageForTest(fixture, "msg_rewind_cleanup_delete", 2, 200)
	require.NoError(t, repos.Messages.Insert(ctx, messageOne))
	require.NoError(t, repos.Messages.Insert(ctx, messageTwo))
	mustExecMessaging(t, ctx, db, `
INSERT INTO attachments (id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, created_at)
VALUES ($1, $2, $3, $4, $5, 'photo.png', 'image/png', 1024, NOW())
`, "att_rewind_cleanup", fixture.workspaceID, fixture.scopeID, fixture.ownerMemberID, messageTwo.ID)

	var cleanerCalls []struct {
		WorkspaceID  string
		AttachmentID string
	}
	store := New(db, nil, 0, Options{
		DeleteAttachmentDerivatives: func(_ context.Context, workspaceID string, attachmentID string) {
			cleanerCalls = append(cleanerCalls, struct {
				WorkspaceID  string
				AttachmentID string
			}{WorkspaceID: workspaceID, AttachmentID: attachmentID})
		},
	})

	// Verify callback hasn't fired yet (pre-commit).
	require.Empty(t, cleanerCalls)

	_, err := store.RewindScopeSuffix(ctx, ScopeRewindSuffixRequest{
		WorkspaceID: fixture.workspaceID,
		ScopeID:     fixture.scopeID,
		FromSeq:     2,
		RewoundAt:   fixture.now.Add(time.Hour),
	})
	require.NoError(t, err)

	// Post-commit: callback fired exactly once with the deleted attachment's IDs.
	require.Len(t, cleanerCalls, 1, "derivative cleaner should fire post-commit")
	require.Equal(t, fixture.workspaceID, cleanerCalls[0].WorkspaceID)
	require.Equal(t, "att_rewind_cleanup", cleanerCalls[0].AttachmentID)
}

// TestRewindNoTargetsDoesNotCallDerivativeCleaner verifies that a rewind with
// no attachments (empty scope, zero messages rewound) does not call the cleaner.
func TestRewindNoTargetsDoesNotCallDerivativeCleaner(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	fixture := seedScopeStoreFixture(t, ctx, db, "rewind_no_targets")

	cleanerCalled := false
	store := New(db, nil, 0, Options{
		DeleteAttachmentDerivatives: func(context.Context, string, string) {
			cleanerCalled = true
		},
	})

	// Rewind a scope with no messages — succeeds but there are no attachments.
	_, err := store.RewindScopeSuffix(ctx, ScopeRewindSuffixRequest{
		WorkspaceID: fixture.workspaceID,
		ScopeID:     fixture.scopeID,
		FromSeq:     1,
		RewoundAt:   fixture.now.Add(time.Hour),
	})
	require.NoError(t, err)
	require.False(t, cleanerCalled, "cleaner must not fire when no attachments were deleted")
}

// TestScrubDoesNotDeleteNonTargetAttachments verifies that scrubbing a set of
// message IDs does not delete attachments belonging to other messages.
func TestScrubDoesNotDeleteNonTargetAttachments(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	fixture := seedScopeStoreFixture(t, ctx, db, "scrub_non_target")
	message := scopeSpecialtyMessageForTest(fixture, "msg_scrub_non_target", 1, 100)
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, message))
	mustExecMessaging(t, ctx, db, `
INSERT INTO attachments (id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, created_at)
VALUES ($1, $2, $3, $4, $5, 'photo.png', 'image/png', 1024, NOW())
`, "att_scrub_non_target", fixture.workspaceID, fixture.scopeID, fixture.ownerMemberID, message.ID)

	err := coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		// Scrub a non-existent message ID — succeeds with zero rows affected.
		_, err := scrubScopeMessageTombstoneProjections(ctx, tx, fixture.scopeID, []string{"nonexistent_message_id"}, "non-target test")
		return err
	})
	require.NoError(t, err)

	var count int
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM attachments WHERE id = $1`, "att_scrub_non_target").Scan(&count))
	require.Equal(t, 1, count, "attachment must survive when its message was not scrubbed")
}

// TestScrubRollbackPreservesAttachmentsOnForcedRollback verifies that when the
// tx is forced to rollback AFTER scrub has deleted attachment rows, the rows
// are restored by the DB transaction and the Store-level derivative cleaner
// is NOT called (because WithTx returned error → post-commit cleanup skipped).
func TestScrubRollbackPreservesAttachmentsOnForcedRollback(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	fixture := seedScopeStoreFixture(t, ctx, db, "scrub_forced_rollback")
	message := scopeSpecialtyMessageForTest(fixture, "msg_scrub_forced_rollback", 1, 100)
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, message))
	attID := "att_scrub_forced_rollback"
	mustExecMessaging(t, ctx, db, `
INSERT INTO attachments (id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, created_at)
VALUES ($1, $2, $3, $4, $5, 'photo.png', 'image/png', 1024, NOW())
`, attID, fixture.workspaceID, fixture.scopeID, fixture.ownerMemberID, message.ID)

	// Use coredb.WithTx directly to prove: scrub deletes attachment rows within
	// the tx, then a forced error rolls back the tx → attachment rows restored.
	err := coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		// Scrub the real message — attachment row is deleted within this tx.
		_, err := scrubScopeMessageTombstoneProjections(ctx, tx, fixture.scopeID, []string{message.ID}, "forced rollback test")
		if err != nil {
			return err
		}
		// Verify the attachment IS deleted within the tx (before rollback).
		var inTxCount int
		require.NoError(t, tx.QueryRow(ctx, `SELECT COUNT(*) FROM attachments WHERE id = $1`, attID).Scan(&inTxCount))
		require.Zero(t, inTxCount, "attachment should be deleted within tx before forced rollback")
		// Force rollback.
		return errors.New("force rollback")
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "force rollback")

	// Post-rollback: attachment row must be restored by the DB transaction.
	var postTxCount int
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM attachments WHERE id = $1`, attID).Scan(&postTxCount))
	require.Equal(t, 1, postTxCount, "attachment must be restored after tx rollback")

	// Post-rollback: attachment row must be restored by the DB.
	var count int
	require.NoError(t, db.QueryRow(ctx, `SELECT COUNT(*) FROM attachments WHERE id = $1`, attID).Scan(&count))
	require.Equal(t, 1, count, "attachment must be restored after tx rollback")
}

// TestWithdrawServiceNotificationMessageTxReturnsDerivatives verifies that the
// App withdraw path collects DeletedAttachmentDerivatives (same pipeline as
// recall). The post-commit consumption is wired identically to recall in
// MutateAppNotification (hoisted messageMutation + CleanupAttachmentDerivatives).
func TestWithdrawServiceNotificationMessageTxReturnsDerivatives(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	fixture := seedScopeStoreFixture(t, ctx, db, "withdraw_derivatives")
	messageOne := scopeSpecialtyMessageForTest(fixture, "msg_withdraw_keep", 1, 100)
	messageTwo := scopeSpecialtyMessageForTest(fixture, "msg_withdraw_delete", 2, 200)
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, messageOne))
	require.NoError(t, NewScopeRepositories(db).Messages.Insert(ctx, messageTwo))
	mustExecMessaging(t, ctx, db, `
INSERT INTO attachments (id, workspace_id, scope_id, uploaded_by_member_id, message_id, filename, content_type, size_bytes, created_at)
VALUES ($1, $2, $3, $4, $5, 'photo.png', 'image/png', 1024, NOW())
`, "att_withdraw_test", fixture.workspaceID, fixture.scopeID, fixture.ownerMemberID, messageTwo.ID)
	// Make messageTwo look like a service-authored notification message.
	// lockServiceNotificationMessage requires author_type='service',
	// matching author_member_id, and message_type IN ('', app_interactive_card).
	mustExecMessaging(t, ctx, db, `
UPDATE scope_messages
SET author_type = 'service', author_member_id = $3, message_type = ''
WHERE id = $1 AND scope_id = $2
`, messageTwo.ID, fixture.scopeID, fixture.ownerMemberID)

	_, _, err := EncodeMessageMetadata(BuildMessageRecallMetadata(fixture.ownerMemberID, "2026-07-18T00:00:00Z", "通知已撤回"))
	require.NoError(t, err)

	// WithdrawServiceNotificationMessageTx internally builds its own tombstone
	// metadata from PlaceholderContent; we just need to pass the minimal fields.
	var result ServiceNotificationMessageMutationResult
	err = coredb.WithTx(ctx, db, func(tx coredb.Tx) error {
		result, err = WithdrawServiceNotificationMessageTx(ctx, tx, ServiceNotificationMessageWithdrawal{
			WorkspaceID:        fixture.workspaceID,
			ScopeID:            fixture.scopeID,
			MessageID:          messageTwo.ID,
			AuthorMemberID:     fixture.ownerMemberID,
			PlaceholderContent: "通知已撤回",
			Reason:             "test",
			WithdrawnAt:        fixture.now.Add(time.Hour),
		})
		return err
	})
	require.NoError(t, err)
	// The withdraw path must return the deleted attachment's derivative target.
	require.Len(t, result.DeletedAttachmentDerivatives, 1)
	require.Equal(t, fixture.workspaceID, result.DeletedAttachmentDerivatives[0].WorkspaceID)
	require.Equal(t, "att_withdraw_test", result.DeletedAttachmentDerivatives[0].AttachmentID)
}

func TestScopeRewindSuffixTombstonesDerivedFactsAndRecomputesState(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_rewind")
	messageOne := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_rewind_keep", 1, 100)
	messageTwo := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_rewind_rewound", 2, 200)
	machineID := "mach_scope_specialty_rewind"
	agentID := "agt_scope_specialty_rewind"
	insertMessagingMachine(t, ctx, db, machineID, fixture.workspaceID, fixture.ownerUserID, fixture.now)
	insertMessagingAgent(t, ctx, db, agentID, fixture.workspaceID, machineID, fixture.privateMemberID, "scope-specialty-rewind", fixture.now)
	require.NoError(t, repos.Messages.Insert(ctx, messageOne))
	require.NoError(t, repos.Messages.Insert(ctx, messageTwo))
	mustExecMessaging(t, ctx, db, `
INSERT INTO message_recovery_points (
  workspace_id, message_id, agent_id, thread_id, recovery_strategy, recovery_ref, prepared_at
) VALUES ($1, $2, $3, $4, 'codex_app_server_turns', '{"threadId":"scp_scope_store_specialty_rewind","turnCount":1}', $5)
`, fixture.workspaceID, messageTwo.ID, agentID, fixture.scopeID, fixture.now)
	require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                fixture.scopeID,
		WorkspaceID:       fixture.workspaceID,
		Type:              "channel",
		IdentityKey:       "channel:scope-store:specialty_rewind",
		LatestSeq:         messageTwo.Seq,
		LatestGlobalSeq:   messageTwo.WorkspaceGlobalSeq,
		LatestMessageAt:   &messageTwo.CreatedAt,
		Status:            "active",
		CreatedByMemberID: fixture.ownerMemberID,
		CreatedAt:         fixture.now,
		UpdatedAt:         fixture.now,
	}))
	require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{
		ScopeID:                               fixture.scopeID,
		MemberID:                              fixture.ownerMemberID,
		Visibility:                            "public",
		LastReadSeq:                           2,
		LastReadGlobalSeq:                     200,
		OwnerContextDismissedThroughGlobalSeq: 200,
		CreatedAt:                             fixture.now,
		UpdatedAt:                             fixture.now,
	}))
	require.NoError(t, repos.Sources.Insert(ctx, ScopeMessageSourceRecord{
		TargetScopeID:   fixture.scopeID,
		MessageID:       messageOne.ID,
		Ordinal:         0,
		SourceScopeID:   fixture.scopeID,
		SourceSeq:       messageTwo.Seq,
		SourceMessageID: messageTwo.ID,
		SourceKind:      "quote",
		SourceDedupeKey: "source:scope-specialty-rewind",
		CreatedAt:       fixture.now,
	}))
	require.NoError(t, repos.Sources.Insert(ctx, ScopeMessageSourceRecord{
		TargetScopeID:   fixture.scopeID,
		MessageID:       messageTwo.ID,
		Ordinal:         0,
		SourceScopeID:   fixture.scopeID,
		SourceSeq:       messageOne.Seq,
		SourceMessageID: messageOne.ID,
		SourceKind:      "quote",
		SourceDedupeKey: "source:scope-specialty-rewind-target",
		CreatedAt:       fixture.now,
	}))
	require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{
		MessageID:   messageTwo.ID,
		WorkspaceID: fixture.workspaceID,
		ScopeID:     fixture.scopeID,
		GlobalSeq:   messageTwo.WorkspaceGlobalSeq,
		Content:     messageTwo.Content,
		CreatedAt:   messageTwo.CreatedAt,
	}))
	reaction := ScopeMessageReactionRecord{
		ID:              "rxn_scope_specialty_rewind",
		WorkspaceID:     fixture.workspaceID,
		ScopeID:         fixture.scopeID,
		MessageID:       messageTwo.ID,
		Emoji:           "👀",
		ReactorMemberID: fixture.ownerMemberID,
		CreatedAt:       fixture.now,
	}
	require.NoError(t, repos.Reactions.Insert(ctx, reaction))
	activity := scopeActivityEventForTest(fixture, messageTwo.ID, "act_scope_specialty_rewind_reaction", messageTwo.WorkspaceGlobalSeq+1, nil)
	activity.ActivityRefID = reaction.ID
	require.NoError(t, repos.ActivityEvents.Insert(ctx, activity))
	outbox := scopeOutboxForTest(fixture, messageTwo.ID, "out_scope_specialty_rewind", "dedupe:scope-specialty-rewind", fixture.now, "pending")
	outbox.EventRefType = "activity_event"
	outbox.EventRefID = activity.ID
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))
	require.NoError(t, repos.ShareCards.Insert(ctx, ScopeShareCardRecord{
		ID:                "shr_scope_specialty_rewind",
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		MessageID:         messageTwo.ID,
		SourceScopeID:     fixture.scopeID,
		SourceMessageID:   messageOne.ID,
		TargetScopeID:     fixture.scopeID,
		CreatedByMemberID: fixture.ownerMemberID,
		CreatedAt:         fixture.now,
		UpdatedAt:         fixture.now,
	}))
	pinned, err := repos.Pins.InsertActivePin(ctx, ScopeMessagePinRecord{
		ID:               "pin_scope_specialty_rewind",
		WorkspaceID:      fixture.workspaceID,
		ScopeID:          fixture.scopeID,
		MessageID:        messageTwo.ID,
		PinnedByMemberID: fixture.ownerMemberID,
		PinnedAt:         fixture.now,
		CreatedAt:        fixture.now,
		UpdatedAt:        fixture.now,
	})
	require.NoError(t, err)
	require.True(t, pinned.Changed)
	topped, err := repos.Pins.UpsertActiveTopNotice(ctx, ScopeMessageTopNoticeRecord{
		ID:            "top_scope_specialty_rewind",
		WorkspaceID:   fixture.workspaceID,
		ScopeID:       fixture.scopeID,
		MessageID:     messageTwo.ID,
		SetByMemberID: fixture.ownerMemberID,
		SetAt:         fixture.now,
		CreatedAt:     fixture.now,
		UpdatedAt:     fixture.now,
	})
	require.NoError(t, err)
	require.True(t, topped.Changed)
	hidden, err := repos.Pins.InsertTopNoticeHide(ctx, ScopeMessageTopNoticeHideRecord{
		ID:               "noticehide_scope_specialty_rewind",
		WorkspaceID:      fixture.workspaceID,
		ScopeID:          fixture.scopeID,
		TopNoticeID:      topped.TopNotice.ID,
		MessageID:        messageTwo.ID,
		HiddenByMemberID: fixture.ownerMemberID,
		HiddenAt:         fixture.now,
		CreatedAt:        fixture.now,
		UpdatedAt:        fixture.now,
	})
	require.NoError(t, err)
	require.True(t, hidden.Changed)
	mustExecMessaging(t, ctx, db, `
INSERT INTO scope_sync_dirty (
  workspace_id, scope_id, latest_message_seq, latest_activity_seq, reason, updated_at
) VALUES ($1, $2, 2, 99, 'message', $3)
`, fixture.workspaceID, fixture.scopeID, fixture.now)
	mustExecMessaging(t, ctx, db, `
INSERT INTO member_scope_sync_dirty (
  workspace_id, member_id, scope_id, latest_message_seq, latest_activity_seq, reason, updated_at
) VALUES ($1, $2, $3, 2, 88, 'message', $4)
`, fixture.workspaceID, fixture.ownerMemberID, fixture.scopeID, fixture.now)

	result, err := New(db, nil, 0, Options{}).RewindScopeSuffix(ctx, ScopeRewindSuffixRequest{
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		FromSeq:           2,
		RewoundAt:         fixture.now.Add(time.Hour),
		SuppressionReason: "test_rewind",
	})
	require.NoError(t, err)
	require.Equal(t, []string{messageTwo.ID}, result.RewoundMessageIDs)
	require.Equal(t, int64(1), result.SuppressedActivityCount)
	require.Equal(t, int64(1), result.SuppressedOutboxCount)
	require.Equal(t, int64(1), result.ClampedScopeSyncDirtyCount)
	require.Equal(t, int64(1), result.ClampedMemberSyncDirtyCount)
	require.Equal(t, int64(1), result.LatestSeq)
	require.Equal(t, int64(100), result.LatestGlobalSeq)

	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_messages", "id = $1", messageOne.ID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_messages", "id = $1", messageTwo.ID))
	var rewoundContent string
	var rewoundMetadata []byte
	require.NoError(t, db.QueryRow(ctx, `SELECT content, metadata FROM scope_messages WHERE id = $1`, messageTwo.ID).Scan(&rewoundContent, &rewoundMetadata))
	require.Empty(t, rewoundContent)
	decodedRewoundMetadata, err := DecodeMessageMetadata(rewoundMetadata)
	require.NoError(t, err)
	require.True(t, MessageMetadataIsRewound(decodedRewoundMetadata))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_sources", "source_message_id = $1", messageTwo.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_sources", "message_id = $1", messageTwo.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_reactions", "id = $1", reaction.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_pins", "id = $1", pinned.Pin.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_top_notices", "id = $1", topped.TopNotice.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_top_notice_hides", "id = $1", hidden.Hide.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_message_search_docs", "message_id = $1", messageTwo.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_share_cards", "message_id = $1", messageTwo.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "message_recovery_points", "message_id = $1", messageTwo.ID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_activity_events", "id = $1 AND suppressed_at IS NOT NULL", activity.ID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'suppressed' AND suppression_reason = 'test_rewind'", outbox.ID))

	var scopeDirtySeq, scopeDirtyActivitySeq, memberDirtySeq, memberDirtyActivitySeq int64
	require.NoError(t, db.QueryRow(ctx, `SELECT latest_message_seq, latest_activity_seq FROM scope_sync_dirty WHERE workspace_id = $1 AND scope_id = $2`, fixture.workspaceID, fixture.scopeID).Scan(&scopeDirtySeq, &scopeDirtyActivitySeq))
	require.Equal(t, int64(1), scopeDirtySeq)
	require.Equal(t, int64(99), scopeDirtyActivitySeq)
	require.NoError(t, db.QueryRow(ctx, `SELECT latest_message_seq, latest_activity_seq FROM member_scope_sync_dirty WHERE workspace_id = $1 AND member_id = $2 AND scope_id = $3`, fixture.workspaceID, fixture.ownerMemberID, fixture.scopeID).Scan(&memberDirtySeq, &memberDirtyActivitySeq))
	require.Equal(t, int64(1), memberDirtySeq)
	require.Equal(t, int64(88), memberDirtyActivitySeq)

	var latestSeq, latestGlobalSeq, visibleMessageCount, lastReadSeq, lastReadGlobalSeq, dismissedGlobalSeq int64
	var visibleLastMessageAt sql.NullTime
	require.NoError(t, db.QueryRow(ctx, `SELECT latest_seq, latest_global_seq, visible_message_count, visible_last_message_at FROM scopes WHERE id = $1`, fixture.scopeID).Scan(&latestSeq, &latestGlobalSeq, &visibleMessageCount, &visibleLastMessageAt))
	require.Equal(t, int64(2), latestSeq)
	require.Equal(t, int64(200), latestGlobalSeq)
	require.Equal(t, int64(1), visibleMessageCount)
	require.True(t, visibleLastMessageAt.Valid)
	require.Equal(t, messageOne.CreatedAt, visibleLastMessageAt.Time)
	require.NoError(t, db.QueryRow(ctx, `SELECT last_read_seq, last_read_global_seq, owner_context_dismissed_through_global_seq FROM scope_memberships WHERE scope_id = $1 AND member_id = $2`, fixture.scopeID, fixture.ownerMemberID).Scan(&lastReadSeq, &lastReadGlobalSeq, &dismissedGlobalSeq))
	require.Equal(t, int64(1), lastReadSeq)
	require.Equal(t, int64(100), lastReadGlobalSeq)
	require.Equal(t, int64(100), dismissedGlobalSeq)

	repeated, err := New(db, nil, 0, Options{}).RewindScopeSuffix(ctx, ScopeRewindSuffixRequest{
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		FromSeq:           2,
		RewoundAt:         fixture.now.Add(2 * time.Hour),
		SuppressionReason: "test_rewind",
	})
	require.NoError(t, err)
	require.Empty(t, repeated.RewoundMessageIDs)
	require.Equal(t, int64(0), repeated.SuppressedActivityCount)
	require.Equal(t, int64(0), repeated.SuppressedOutboxCount)
}

func TestScopeRewindSuffixOnlySuppressesUnfinishedOutbox(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "specialty_rewind_outbox")
	messageOne := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_rewind_outbox_keep", 1, 100)
	messageTwo := scopeSpecialtyMessageForTest(fixture, "msg_scope_specialty_rewind_outbox_rewound", 2, 200)
	require.NoError(t, repos.Messages.Insert(ctx, messageOne))
	require.NoError(t, repos.Messages.Insert(ctx, messageTwo))
	require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{
		ID:              fixture.scopeID,
		WorkspaceID:     fixture.workspaceID,
		Type:            "channel",
		IdentityKey:     "channel:scope-store:specialty_rewind_outbox",
		LatestSeq:       messageTwo.Seq,
		LatestGlobalSeq: messageTwo.WorkspaceGlobalSeq,
		LatestMessageAt: &messageTwo.CreatedAt,
		Status:          "active",
		CreatedAt:       fixture.now,
		UpdatedAt:       fixture.now,
	}))

	pendingOutbox := scopeOutboxForTest(fixture, messageTwo.ID, "out_scope_specialty_rewind_pending", "dedupe:scope-specialty-rewind-pending", fixture.now, "pending")
	leasedOutbox := scopeOutboxForTest(fixture, messageTwo.ID, "out_scope_specialty_rewind_leased", "dedupe:scope-specialty-rewind-leased", fixture.now, "leased")
	deliveredOutbox := scopeOutboxForTest(fixture, messageTwo.ID, "out_scope_specialty_rewind_delivered", "dedupe:scope-specialty-rewind-delivered", fixture.now, "delivered")
	require.NoError(t, repos.Outbox.Insert(ctx, pendingOutbox))
	require.NoError(t, repos.Outbox.Insert(ctx, leasedOutbox))
	require.NoError(t, repos.Outbox.Insert(ctx, deliveredOutbox))

	result, err := New(db, nil, 0, Options{}).RewindScopeSuffix(ctx, ScopeRewindSuffixRequest{
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		FromSeq:           2,
		RewoundAt:         fixture.now.Add(time.Hour),
		SuppressionReason: "test_rewind",
	})
	require.NoError(t, err)
	require.Equal(t, []string{messageTwo.ID}, result.RewoundMessageIDs)
	require.Equal(t, int64(2), result.SuppressedOutboxCount)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'suppressed' AND suppression_reason = 'test_rewind'", pendingOutbox.ID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'suppressed' AND suppression_reason = 'test_rewind'", leasedOutbox.ID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'delivered' AND suppressed_at IS NULL", deliveredOutbox.ID))
}

func scopeSpecialtyMessageForTest(fixture scopeStoreFixture, id string, seq int64, globalSeq int64) ScopeMessageRecord {
	createdAt := fixture.now.Add(time.Duration(seq) * time.Second)
	return ScopeMessageRecord{
		ID:                 id,
		WorkspaceID:        fixture.workspaceID,
		ScopeID:            fixture.scopeID,
		Seq:                seq,
		WorkspaceGlobalSeq: globalSeq,
		AuthorMemberID:     fixture.ownerMemberID,
		AuthorType:         ScopeActorHuman,
		MessageType:        "text",
		Content:            id,
		CreatedAt:          createdAt,
		UpdatedAt:          createdAt,
	}
}

type scopeReactionFailSuppressDB struct {
	coredb.DB
}

func (db scopeReactionFailSuppressDB) BeginTx(ctx context.Context, options coredb.TxOptions) (coredb.Tx, error) {
	tx, err := db.DB.BeginTx(ctx, options)
	if err != nil {
		return nil, err
	}
	return scopeReactionFailSuppressTx{Tx: tx}, nil
}

type scopeReactionFailSuppressTx struct {
	coredb.Tx
}

func (tx scopeReactionFailSuppressTx) Exec(ctx context.Context, query string, args ...any) (coredb.Result, error) {
	if strings.Contains(query, "UPDATE scope_activity_events") && strings.Contains(query, "activity_ref_type = 'reaction'") {
		return nil, errors.New("injected reaction delivery suppress failure")
	}
	return tx.Tx.Exec(ctx, query, args...)
}

type scopeReactionFailActivityInsertDB struct {
	coredb.DB
}

func (db scopeReactionFailActivityInsertDB) BeginTx(ctx context.Context, options coredb.TxOptions) (coredb.Tx, error) {
	tx, err := db.DB.BeginTx(ctx, options)
	if err != nil {
		return nil, err
	}
	return scopeReactionFailActivityInsertTx{Tx: tx}, nil
}

type scopeReactionFailActivityInsertTx struct {
	coredb.Tx
}

func (tx scopeReactionFailActivityInsertTx) Exec(ctx context.Context, query string, args ...any) (coredb.Result, error) {
	if strings.Contains(query, "INSERT INTO scope_activity_events") {
		return nil, errors.New("injected reaction activity insert failure")
	}
	return tx.Tx.Exec(ctx, query, args...)
}