package message

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

	"github.com/stretchr/testify/require"

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

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	now := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)

	workspaceID := "wsp_scope_store"
	ownerUserID := "usr_scope_store_owner"
	ownerMemberID := "mem_scope_store_owner"
	privateMemberID := "mem_scope_store_private"
	channelScopeID := "scp_scope_store_channel"
	threadScopeID := "scp_scope_store_thread"
	rootMessageID := "msg_scope_store_root"
	replyMessageID := "msg_scope_store_reply"
	grantID := "grant_scope_store_private"
	activityEventID := "act_scope_store_reaction"
	outboxID := "out_scope_store_wake"
	reactionID := "rxn_scope_store"
	shareCardID := "shr_scope_store"

	insertMessagingWorkspace(t, ctx, db, workspaceID, now)
	insertMessagingUser(t, ctx, db, ownerUserID, "scope-store-owner@example.com", now)
	insertMessagingHumanMember(t, ctx, db, ownerMemberID, workspaceID, ownerUserID, now)
	insertMessagingAgentMember(t, ctx, db, privateMemberID, workspaceID, now)

	require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                channelScopeID,
		WorkspaceID:       workspaceID,
		Type:              "channel",
		IdentityKey:       "channel:scope-store",
		LatestSeq:         1,
		LatestGlobalSeq:   10,
		LatestMessageAt:   &now,
		Status:            "active",
		CreatedByMemberID: ownerMemberID,
		CreatedAt:         now,
		UpdatedAt:         now,
	}))
	require.NoError(t, repos.Scopes.UpsertChannelProfile(ctx, ScopeChannelProfileRecord{
		ScopeID:       channelScopeID,
		WorkspaceID:   workspaceID,
		Slug:          "scope-store",
		DisplayName:   "Scope Store",
		Description:   "scope store repository test",
		Visibility:    "private",
		OwnerMemberID: ownerMemberID,
		CreatedAt:     now,
		UpdatedAt:     now,
	}))
	require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                 threadScopeID,
		WorkspaceID:        workspaceID,
		Type:               "thread",
		ParentScopeID:      channelScopeID,
		RootMessageID:      rootMessageID,
		RootMessageShortID: "0001",
		IdentityKey:        "thread:scope-store",
		LatestSeq:          1,
		LatestGlobalSeq:    11,
		LatestMessageAt:    &now,
		Status:             "active",
		CreatedByMemberID:  ownerMemberID,
		CreatedAt:          now,
		UpdatedAt:          now,
	}))
	require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{
		ScopeID:                               channelScopeID,
		MemberID:                              ownerMemberID,
		Visibility:                            "public",
		LastReadSeq:                           1,
		LastReadGlobalSeq:                     10,
		OwnerContextDismissedThroughGlobalSeq: 10,
		CreatedAt:                             now,
		UpdatedAt:                             now,
	}))
	require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{
		ScopeID:           threadScopeID,
		MemberID:          privateMemberID,
		Visibility:        "private",
		OwnerMemberID:     ownerMemberID,
		LastReadSeq:       0,
		LastReadGlobalSeq: 0,
		CreatedAt:         now,
		UpdatedAt:         now,
	}))
	require.NoError(t, repos.Notifications.Upsert(ctx, ScopeNotificationSettingRecord{
		ScopeID:   channelScopeID,
		MemberID:  privateMemberID,
		State:     "auto-wake",
		CreatedAt: now,
		UpdatedAt: now,
	}))
	require.NoError(t, repos.PrivateVisibility.InsertGrant(ctx, ScopePrivateGrantRecord{
		ID:                grantID,
		WorkspaceID:       workspaceID,
		ScopeID:           threadScopeID,
		OwnerMemberID:     ownerMemberID,
		GranteeMemberID:   privateMemberID,
		GrantedByMemberID: ownerMemberID,
		CreatedAt:         now,
	}))
	require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{
		ID:                 rootMessageID,
		WorkspaceID:        workspaceID,
		ScopeID:            channelScopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 10,
		AuthorMemberID:     ownerMemberID,
		AuthorType:         "human",
		MessageType:        "text",
		Content:            "root",
		Metadata:           []byte(`{"kind":"root"}`),
		CreatedAt:          now,
		UpdatedAt:          now,
	}))
	require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{
		ID:                 replyMessageID,
		WorkspaceID:        workspaceID,
		ScopeID:            threadScopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 11,
		AuthorMemberID:     ownerMemberID,
		AuthorType:         "human",
		MessageType:        "text",
		Content:            "@private please read",
		Metadata:           []byte(`{"kind":"reply"}`),
		CreatedAt:          now,
		UpdatedAt:          now,
	}))
	require.NoError(t, repos.Sources.Insert(ctx, ScopeMessageSourceRecord{
		TargetScopeID:   threadScopeID,
		MessageID:       replyMessageID,
		Ordinal:         0,
		SourceScopeID:   channelScopeID,
		SourceSeq:       1,
		SourceMessageID: rootMessageID,
		SourceKind:      "root",
		SourceDedupeKey: "source:scope-store",
		CreatedAt:       now,
	}))
	activityGlobalSeq := int64(12)
	require.NoError(t, repos.ActivityEvents.Insert(ctx, ScopeActivityEventRecord{
		ID:                activityEventID,
		WorkspaceID:       workspaceID,
		ScopeID:           threadScopeID,
		Seq:               activityGlobalSeq,
		ActivityType:      "reaction_added",
		ActivityRefType:   "reaction",
		ActivityRefID:     reactionID,
		ActivityGlobalSeq: activityGlobalSeq,
		SourceMessageID:   replyMessageID,
		ActorMemberID:     ownerMemberID,
		Reason:            "reaction",
		CreatedAt:         now,
	}))
	require.NoError(t, repos.Outbox.Insert(ctx, ScopeFanoutOutboxRecord{
		ID:            outboxID,
		WorkspaceID:   workspaceID,
		ScopeID:       threadScopeID,
		MessageID:     replyMessageID,
		EventRefType:  "message",
		EventRefID:    replyMessageID,
		FanoutType:    "scope_fanout",
		DedupeKey:     "wake:scope-store",
		Payload:       []byte(`{"recipient":"private"}`),
		NextAttemptAt: now,
		CreatedAt:     now,
	}))
	require.NoError(t, repos.Reactions.Insert(ctx, ScopeMessageReactionRecord{
		ID:              reactionID,
		WorkspaceID:     workspaceID,
		ScopeID:         threadScopeID,
		MessageID:       replyMessageID,
		Emoji:           "👍",
		ReactorMemberID: ownerMemberID,
		CreatedAt:       now,
	}))
	require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{
		MessageID:   replyMessageID,
		WorkspaceID: workspaceID,
		ScopeID:     threadScopeID,
		GlobalSeq:   11,
		Content:     "scope store search",
		CreatedAt:   now,
	}))
	require.NoError(t, repos.Search.UpsertState(ctx, ScopeSearchStateRecord{
		WorkspaceID:             workspaceID,
		ScopeID:                 threadScopeID,
		ProjectionName:          "tidb",
		MigrationStatus:         "complete",
		BackfillStatus:          "complete",
		BackfillCursorGlobalSeq: 11,
		FinalizeStatus:          "complete",
		ReadReady:               true,
		CreatedAt:               now,
		UpdatedAt:               now,
	}))
	require.NoError(t, repos.ShareCards.Insert(ctx, ScopeShareCardRecord{
		ID:                shareCardID,
		WorkspaceID:       workspaceID,
		ScopeID:           threadScopeID,
		MessageID:         replyMessageID,
		SourceScopeID:     channelScopeID,
		SourceMessageID:   rootMessageID,
		TargetScopeID:     threadScopeID,
		CreatedByMemberID: ownerMemberID,
		Payload:           []byte(`{"share":true}`),
		CreatedAt:         now,
		UpdatedAt:         now,
	}))

	loaded, err := repos.Scopes.LoadScope(ctx, threadScopeID)
	require.NoError(t, err)
	require.Equal(t, "thread", loaded.Type)
	require.Equal(t, channelScopeID, loaded.ParentScopeID)

	messages, err := repos.Messages.ListBeforeSeq(ctx, threadScopeID, 2, 10)
	require.NoError(t, err)
	require.Len(t, messages, 1)
	require.Equal(t, replyMessageID, messages[0].ID)
	require.JSONEq(t, `{"kind":"reply"}`, string(messages[0].Metadata))

	activities, err := repos.ActivityEvents.ListAfterCursor(ctx, workspaceID, threadScopeID, 0, 10)
	require.NoError(t, err)
	require.Len(t, activities, 1)
	require.Equal(t, activityEventID, activities[0].ID)
	require.Equal(t, activityGlobalSeq, activities[0].ActivityGlobalSeq)

	pendingOutbox, err := repos.Outbox.ListPending(ctx, 10)
	require.NoError(t, err)
	require.Len(t, pendingOutbox, 1)
	require.Equal(t, outboxID, pendingOutbox[0].ID)
	require.JSONEq(t, `{"recipient":"private"}`, string(pendingOutbox[0].Payload))

	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_channel_profiles", "scope_id = $1", channelScopeID))
	require.Equal(t, int64(2), countScopeRows(t, ctx, db, "scope_memberships", "scope_id IN ($1, $2)", channelScopeID, threadScopeID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_notification_settings", "scope_id = $1", channelScopeID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_private_grants", "id = $1", grantID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_message_sources", "message_id = $1", replyMessageID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_message_reactions", "id = $1", reactionID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_message_search_docs", "message_id = $1", replyMessageID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_message_search_state", "workspace_id = $1 AND scope_id = $2", workspaceID, threadScopeID))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_share_cards", "id = $1", shareCardID))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	now := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
	editedAt := now.Add(5 * time.Minute)

	fixture := seedScopeMessageEditFixture(t, ctx, db, "records_edit_fields", "", now)

	tx, err := db.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	updated, err := UpdateScopeMessageContentAndSearchDoc(ctx, tx, ScopeMessageContentUpdate{
		WorkspaceID:      fixture.workspaceID,
		ScopeID:          fixture.scopeID,
		MessageID:        fixture.messageID,
		Content:          "edited content",
		Metadata:         []byte(`{"mentions":[{"id":"mem"}]}`),
		EditedByMemberID: fixture.authorMemberID,
		EditedAt:         editedAt,
	})
	require.NoError(t, err)
	require.NoError(t, tx.Commit(ctx))
	require.Equal(t, "edited content", updated.Content)
	require.JSONEq(t, `{"mentions":[{"id":"mem"}]}`, string(updated.Metadata))
	require.NotNil(t, updated.EditedAt)
	require.True(t, updated.EditedAt.Equal(editedAt))
	require.Equal(t, fixture.authorMemberID, updated.EditedByMemberID)
	require.EqualValues(t, 1, updated.EditRevision)

	loaded, err := repos.Messages.LoadByID(ctx, fixture.workspaceID, fixture.scopeID, fixture.messageID)
	require.NoError(t, err)
	require.Equal(t, updated.Content, loaded.Content)
	require.NotNil(t, loaded.EditedAt)
	require.True(t, loaded.EditedAt.Equal(editedAt))
	require.EqualValues(t, 1, loaded.EditRevision)

	var searchContent string
	require.NoError(t, db.QueryRow(ctx, `
SELECT content
FROM scope_message_search_docs
WHERE message_id = $1
`, fixture.messageID).Scan(&searchContent))
	require.Equal(t, "edited content", searchContent)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	now := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
	editedAt := now.Add(5 * time.Minute)

	nonAuthorFixture := seedScopeMessageEditFixture(t, ctx, db, "non_author_reject", "", now)
	tx, err := db.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	_, err = UpdateScopeMessageContentAndSearchDoc(ctx, tx, ScopeMessageContentUpdate{
		WorkspaceID:      nonAuthorFixture.workspaceID,
		ScopeID:          nonAuthorFixture.scopeID,
		MessageID:        nonAuthorFixture.messageID,
		Content:          "edited by someone else",
		Metadata:         []byte(`{}`),
		EditedByMemberID: "mem_message_edit_store_intruder",
		EditedAt:         editedAt,
	})
	require.Error(t, err)
	require.NoError(t, tx.Rollback(ctx))
	loaded, err := repos.Messages.LoadByID(ctx, nonAuthorFixture.workspaceID, nonAuthorFixture.scopeID, nonAuthorFixture.messageID)
	require.NoError(t, err)
	require.Equal(t, "original content", loaded.Content)
	require.EqualValues(t, 0, loaded.EditRevision)

	typedFixture := seedScopeMessageEditFixture(t, ctx, db, "typed_reject", MessageTypeOwnerQuote, now)
	tx, err = db.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	_, err = UpdateScopeMessageContentAndSearchDoc(ctx, tx, ScopeMessageContentUpdate{
		WorkspaceID:      typedFixture.workspaceID,
		ScopeID:          typedFixture.scopeID,
		MessageID:        typedFixture.messageID,
		Content:          "edited typed message",
		Metadata:         []byte(`{}`),
		EditedByMemberID: typedFixture.authorMemberID,
		EditedAt:         editedAt,
	})
	require.Error(t, err)
	require.NoError(t, tx.Rollback(ctx))
	loaded, err = repos.Messages.LoadByID(ctx, typedFixture.workspaceID, typedFixture.scopeID, typedFixture.messageID)
	require.NoError(t, err)
	require.Equal(t, "original content", loaded.Content)
	require.EqualValues(t, 0, loaded.EditRevision)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	now := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
	editedAt := now.Add(5 * time.Minute)

	fixture := seedScopeMessageEditFixture(t, ctx, db, "task_discussion_edit", MessageTypeTaskDiscussion, now)
	tx, err := db.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	updated, err := UpdateScopeMessageContentAndSearchDoc(ctx, tx, ScopeMessageContentUpdate{
		WorkspaceID:      fixture.workspaceID,
		ScopeID:          fixture.scopeID,
		MessageID:        fixture.messageID,
		Content:          "edited task discussion",
		Metadata:         []byte(`{}`),
		EditedByMemberID: fixture.authorMemberID,
		EditedAt:         editedAt,
	})
	require.NoError(t, err)
	require.NoError(t, tx.Commit(ctx))
	require.Equal(t, MessageTypeTaskDiscussion, updated.MessageType)
	require.Equal(t, "edited task discussion", updated.Content)
	require.NotNil(t, updated.EditedAt)
	require.EqualValues(t, 1, updated.EditRevision)

	loaded, err := repos.Messages.LoadByID(ctx, fixture.workspaceID, fixture.scopeID, fixture.messageID)
	require.NoError(t, err)
	require.Equal(t, "edited task discussion", loaded.Content)
	require.EqualValues(t, 1, loaded.EditRevision)

	var searchContent string
	require.NoError(t, db.QueryRow(ctx, `
SELECT content
FROM scope_message_search_docs
WHERE message_id = $1
`, fixture.messageID).Scan(&searchContent))
	require.Equal(t, "edited task discussion", searchContent)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	now := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
	editedAt := now.Add(5 * time.Minute)
	fixture := seedScopeMessageEditFixture(t, ctx, db, "search_rollback", "", now)

	tx, err := db.BeginTx(ctx, coredb.TxOptions{})
	require.NoError(t, err)
	failingTx := failSearchDocInsertTx{Tx: tx}
	_, err = UpdateScopeMessageContentAndSearchDoc(ctx, failingTx, ScopeMessageContentUpdate{
		WorkspaceID:      fixture.workspaceID,
		ScopeID:          fixture.scopeID,
		MessageID:        fixture.messageID,
		Content:          "edited but rolled back",
		Metadata:         []byte(`{}`),
		EditedByMemberID: fixture.authorMemberID,
		EditedAt:         editedAt,
	})
	require.ErrorContains(t, err, "boom search projection")
	require.NoError(t, tx.Rollback(ctx))

	loaded, err := repos.Messages.LoadByID(ctx, fixture.workspaceID, fixture.scopeID, fixture.messageID)
	require.NoError(t, err)
	require.Equal(t, "original content", loaded.Content)
	require.Nil(t, loaded.EditedAt)
	require.EqualValues(t, 0, loaded.EditRevision)

	var searchContent string
	require.NoError(t, db.QueryRow(ctx, `
SELECT content
FROM scope_message_search_docs
WHERE message_id = $1
`, fixture.messageID).Scan(&searchContent))
	require.Equal(t, "original content", searchContent)
}

type scopeMessageEditFixture struct {
	workspaceID    string
	authorMemberID string
	scopeID        string
	messageID      string
}

func seedScopeMessageEditFixture(t *testing.T, ctx context.Context, db coredb.DB, suffix string, messageType string, now time.Time) scopeMessageEditFixture {
	t.Helper()
	repos := NewScopeRepositories(db)
	workspaceID := "wsp_message_edit_store_" + suffix
	userID := "usr_message_edit_store_author_" + suffix
	authorMemberID := "mem_message_edit_store_author_" + suffix
	scopeID := "scp_message_edit_store_channel_" + suffix
	messageID := "msg_message_edit_store_" + suffix

	insertMessagingWorkspace(t, ctx, db, workspaceID, now)
	insertMessagingUser(t, ctx, db, userID, "message-edit-store-"+suffix+"@example.com", now)
	insertMessagingHumanMember(t, ctx, db, authorMemberID, workspaceID, userID, now)
	require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                scopeID,
		WorkspaceID:       workspaceID,
		Type:              "channel",
		IdentityKey:       "channel:message-edit-store-" + suffix,
		LatestSeq:         1,
		LatestGlobalSeq:   10,
		LatestMessageAt:   &now,
		Status:            "active",
		CreatedByMemberID: authorMemberID,
		CreatedAt:         now,
		UpdatedAt:         now,
	}))
	require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{
		ID:                 messageID,
		WorkspaceID:        workspaceID,
		ScopeID:            scopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 10,
		AuthorMemberID:     authorMemberID,
		AuthorType:         "human",
		MessageType:        messageType,
		Content:            "original content",
		Metadata:           []byte(`{"mentions":[]}`),
		CreatedAt:          now,
		UpdatedAt:          now,
	}))
	require.NoError(t, repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{
		MessageID:   messageID,
		WorkspaceID: workspaceID,
		ScopeID:     scopeID,
		GlobalSeq:   10,
		Content:     "original content",
		CreatedAt:   now,
	}))
	return scopeMessageEditFixture{
		workspaceID:    workspaceID,
		authorMemberID: authorMemberID,
		scopeID:        scopeID,
		messageID:      messageID,
	}
}

type failSearchDocInsertTx struct {
	coredb.Tx
}

func (tx failSearchDocInsertTx) Exec(ctx context.Context, query string, args ...any) (coredb.Result, error) {
	if strings.Contains(query, "scope_message_search_docs") {
		return nil, errors.New("boom search projection")
	}
	return tx.Tx.Exec(ctx, query, args...)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)

	require.ErrorIs(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{}), domain.ErrInvalidArgument)
	require.ErrorIs(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{}), domain.ErrInvalidArgument)
	require.ErrorIs(t, repos.PrivateVisibility.InsertGrant(ctx, ScopePrivateGrantRecord{}), domain.ErrInvalidArgument)
	require.ErrorIs(t, repos.Messages.Insert(ctx, ScopeMessageRecord{}), domain.ErrInvalidArgument)
}

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

	tests := []struct {
		name        string
		descriptor  scopeQueryDescriptor
		index       string
		queryNeedle string
	}{
		{
			name:        "history page",
			descriptor:  scopeMessagesBeforeSeqQuery,
			index:       "scope_messages_scope_seq_key",
			queryNeedle: "ORDER BY seq DESC",
		},
		{
			name:        "activity after cursor",
			descriptor:  scopeActivityEventsAfterCursorQuery,
			index:       "scope_activity_events_scope_seq_key",
			queryNeedle: "suppressed_at IS NULL",
		},
		{
			name:        "fanout outbox",
			descriptor:  scopePendingFanoutOutboxQuery,
			index:       "scope_fanout_outbox_pending_idx",
			queryNeedle: "next_attempt_at <= CURRENT_TIMESTAMP",
		},
		{
			name:        "fanout outbox lease pending",
			descriptor:  scopeLeasePendingFanoutOutboxQuery,
			index:       "scope_fanout_outbox_pending_idx",
			queryNeedle: "next_attempt_at <= $1",
		},
		{
			name:        "fanout outbox lease pending workspace group",
			descriptor:  scopeLeasePendingFanoutOutboxForWorkspaceGroupQuery,
			index:       "workspace_placements_group_state_idx",
			queryNeedle: "placement.state IN ('active', 'draining', 'moving')",
		},
		{
			name:        "fanout outbox lease expired",
			descriptor:  scopeLeaseExpiredFanoutOutboxQuery,
			index:       "scope_fanout_outbox_lease_idx",
			queryNeedle: "lease_expires_at <= $1",
		},
		{
			name:        "fanout outbox lease expired workspace group",
			descriptor:  scopeLeaseExpiredFanoutOutboxForWorkspaceGroupQuery,
			index:       "workspace_placements_group_state_idx",
			queryNeedle: "placement.state IN ('active', 'draining', 'moving')",
		},
		{
			name:        "fanout outbox mark leased",
			descriptor:  scopeMarkFanoutOutboxLeasedQuery,
			index:       "scope_fanout_outbox PRIMARY",
			queryNeedle: "attempts = attempts + 1",
		},
		{
			name:        "fanout outbox payload target",
			descriptor:  scopeMessageFanoutOutboxPayloadTargetCountQuery,
			index:       "scope_fanout_outbox_event_ref_idx",
			queryNeedle: "fanout_type = 'scope_fanout'",
		},
		{
			name:        "fanout outbox delivered",
			descriptor:  scopeMarkFanoutOutboxDeliveredQuery,
			index:       "scope_fanout_outbox PRIMARY",
			queryNeedle: "lease_owner = $3",
		},
		{
			name:        "fanout outbox retry",
			descriptor:  scopeMarkFanoutOutboxRetryQuery,
			index:       "scope_fanout_outbox PRIMARY",
			queryNeedle: "lease_owner = $4",
		},
		{
			name:        "fanout outbox suppressed",
			descriptor:  scopeMarkFanoutOutboxSuppressedQuery,
			index:       "scope_fanout_outbox PRIMARY",
			queryNeedle: "lease_owner = $4",
		},
		{
			name:        "fanout outbox leased payload recheck",
			descriptor:  scopeLeasedFanoutOutboxByOwnerQuery,
			index:       "scope_fanout_outbox PRIMARY",
			queryNeedle: "lease_owner = $2",
		},
		{
			name:        "fanout outbox message mutation",
			descriptor:  scopeMessageFanoutOutboxForMutationQuery,
			index:       "scope_fanout_outbox_event_ref_idx",
			queryNeedle: "event_ref_id = $1",
		},
		{
			name:        "visibility history",
			descriptor:  scopeMessagesVisibleToMemberQuery,
			index:       "scope_messages_scope_seq_key",
			queryNeedle: "JOIN scope_memberships",
		},
		{
			name:        "search visibility",
			descriptor:  scopeSearchVisibilityCandidatesQuery,
			index:       "scope_message_search_docs_workspace_global_seq_idx",
			queryNeedle: "__MATCH_CONDITION__",
		},
		{
			name:        "search readiness",
			descriptor:  scopeSearchReadinessQuery,
			index:       "scope_message_search_state PRIMARY",
			queryNeedle: "read_ready",
		},
		{
			name:        "message visibility by id",
			descriptor:  scopeMessageVisibilityByIDQuery,
			index:       "scope_messages PRIMARY",
			queryNeedle: "m.id = $3",
		},
		{
			name:        "member display",
			descriptor:  scopeMemberDisplayRowsQuery,
			index:       "scope_memberships_scope_visibility_idx",
			queryNeedle: "sm.visibility = 'public'",
		},
		{
			name:        "scope presentations",
			descriptor:  scopePresentationsByIDQuery,
			index:       "scopes PRIMARY",
			queryNeedle: "LEFT JOIN scope_channel_profiles",
		},
		{
			name:        "member presentations",
			descriptor:  scopeMemberPresentationsByIDQuery,
			index:       "members PRIMARY",
			queryNeedle: "LEFT JOIN agents",
		},
		{
			name:        "advance read state",
			descriptor:  scopeAdvanceMembershipReadStateQuery,
			index:       "scope_memberships PRIMARY",
			queryNeedle: "effective_to_seq IS NULL",
		},
		{
			name:        "activity event by id",
			descriptor:  scopeActivityEventByIDQuery,
			index:       "scope_activity_events PRIMARY",
			queryNeedle: "WHERE id = $1",
		},
		{
			name:        "reaction by actor",
			descriptor:  scopeReactionByActorQuery,
			index:       "scope_message_reactions_message_emoji_member_key",
			queryNeedle: "reactor_member_id = $3",
		},
		{
			name:        "insert reaction",
			descriptor:  scopeInsertReactionQuery,
			index:       "scope_message_reactions_message_emoji_member_key",
			queryNeedle: "INSERT INTO scope_message_reactions",
		},
		{
			name:        "delete reaction",
			descriptor:  scopeDeleteReactionByActorQuery,
			index:       "scope_message_reactions_message_emoji_member_key",
			queryNeedle: "DELETE FROM scope_message_reactions",
		},
		{
			name:        "suppress reaction activities",
			descriptor:  scopeSuppressReactionActivitiesQuery,
			index:       "scope_activity_events_ref_key",
			queryNeedle: "activity_ref_type = 'reaction'",
		},
		{
			name:        "search readiness detail",
			descriptor:  scopeSearchReadinessDetailQuery,
			index:       "scope_message_search_state PRIMARY",
			queryNeedle: "migration_status",
		},
		{
			name:        "search backfill candidates",
			descriptor:  scopeSearchMessagesForBackfillQuery,
			index:       "scope_messages_scope_seq_key",
			queryNeedle: "docs.message_id IS NULL",
		},
		{
			name:        "search missing docs count",
			descriptor:  scopeSearchMissingDocsCountQuery,
			index:       "scope_messages",
			queryNeedle: "docs.message_id IS NULL",
		},
		{
			name:        "search latest scope seq",
			descriptor:  scopeSearchLatestScopeSeqQuery,
			index:       "scope_messages_scope_seq_key",
			queryNeedle: "COALESCE(MAX(seq), 0)",
		},
		{
			name:        "search update state",
			descriptor:  scopeSearchUpdateStateQuery,
			index:       "scope_message_search_state PRIMARY",
			queryNeedle: "read_ready = $8",
		},
		{
			name:        "search show create docs",
			descriptor:  scopeSearchShowCreateDocsQuery,
			index:       "scope_message_search_docs physical table metadata",
			queryNeedle: "SHOW CREATE TABLE",
		},
		{
			name:        "search finalize table exists",
			descriptor:  scopeSearchFinalizeTableExistsQuery,
			index:       "information_schema.tables",
			queryNeedle: "table_name = $1",
		},
		{
			name:        "search fulltext enabled",
			descriptor:  scopeSearchFullTextEnabledQuery,
			index:       "tidb_enable_fulltext_index",
			queryNeedle: "SHOW VARIABLES",
		},
		{
			name:        "search tiflash registered",
			descriptor:  scopeSearchTiFlashRegisteredQuery,
			index:       "information_schema.cluster_info",
			queryNeedle: "tiflash_compute",
		},
		{
			name:        "share card by message",
			descriptor:  scopeShareCardByMessageQuery,
			index:       "scope_share_cards_scope_message_key",
			queryNeedle: "revoked_by_member_id",
		},
		{
			name:        "revoke share card",
			descriptor:  scopeRevokeShareCardQuery,
			index:       "scope_share_cards_scope_message_key",
			queryNeedle: "revoked_at IS NULL",
		},
		{
			name:        "rewind suffix messages",
			descriptor:  scopeRewindSuffixMessagesQuery,
			index:       "scope_messages_scope_seq_key",
			queryNeedle: "seq >= $2",
		},
		{
			name:        "rewind recompute stats",
			descriptor:  scopeRewindRecomputeStatsQuery,
			index:       "scope_messages_scope_seq_key",
			queryNeedle: "MAX(workspace_global_seq)",
		},
		{
			name:        "rewind clamp read state",
			descriptor:  scopeRewindClampMembershipReadStateQuery,
			index:       "scope_memberships PRIMARY",
			queryNeedle: "LEAST(last_read_seq",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.NotEmpty(t, tt.descriptor.Name)
			require.Contains(t, tt.descriptor.TargetIndex, tt.index)
			require.NotEmpty(t, tt.descriptor.AmplificationRisk)
			require.Contains(t, tt.descriptor.SQL, tt.queryNeedle)
		})
	}
}

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

	now := time.Date(2026, 6, 5, 10, 0, 0, 0, time.UTC)
	pending := []ScopeFanoutOutboxRecord{
		{ID: "out_pending_1", NextAttemptAt: now.Add(time.Second)},
		{ID: "out_pending_2", NextAttemptAt: now.Add(4 * time.Second)},
	}
	expired := []ScopeFanoutOutboxRecord{
		{ID: "out_expired_1", NextAttemptAt: now.Add(2 * time.Second)},
		{ID: "out_expired_2", NextAttemptAt: now.Add(4 * time.Second)},
	}

	merged := mergeScopeFanoutLeaseCandidates(pending, expired, 3)

	require.Equal(t, []ScopeFanoutOutboxRecord{
		{ID: "out_pending_1", NextAttemptAt: now.Add(time.Second)},
		{ID: "out_expired_1", NextAttemptAt: now.Add(2 * time.Second)},
		{ID: "out_expired_2", NextAttemptAt: now.Add(4 * time.Second)},
	}, merged)
}

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

	for name, query := range map[string]string{
		"history page":      scopeMessagesBeforeSeqQuery.SQL,
		"visibility page":   scopeMessagesVisibleToMemberQuery.SQL,
		"search visibility": scopeSearchVisibilityCandidatesQuery.SQL,
		"visibility by id":  scopeMessageVisibilityByIDQuery.SQL,
		"rewind suffix":     scopeRewindSuffixMessagesQuery.SQL,
	} {
		t.Run(name, func(t *testing.T) {
			require.Contains(t, query, "notify_principal_member_id")
			require.Contains(t, query, "idempotency_subject_member_id")
			require.Contains(t, query, "idempotency_payload_digest")
		})
	}
}

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

	require.NotContains(t, scopeLeasePendingFanoutOutboxQuery.SQL, " OR ")
	require.NotContains(t, scopeLeaseExpiredFanoutOutboxQuery.SQL, " OR ")
	require.NotContains(t, scopeLeasePendingFanoutOutboxForWorkspaceGroupQuery.SQL, " OR ")
	require.NotContains(t, scopeLeaseExpiredFanoutOutboxForWorkspaceGroupQuery.SQL, " OR ")
}

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

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

	first := ScopeMessageRecord{
		ID:                         "msg_scope_store_idem_first",
		WorkspaceID:                fixture.workspaceID,
		ScopeID:                    fixture.scopeID,
		Seq:                        1,
		WorkspaceGlobalSeq:         100,
		AuthorMemberID:             fixture.ownerMemberID,
		AuthorType:                 "human",
		NotifyPrincipalMemberID:    fixture.ownerMemberID,
		IdempotencySubjectMemberID: fixture.ownerMemberID,
		ClientMessageID:            "client-scope-store-idem",
		IdempotencyPayloadDigest:   strings.Repeat("a", 64),
		MessageType:                "text",
		Content:                    "first",
		CreatedAt:                  fixture.now,
		UpdatedAt:                  fixture.now,
	}
	result, err := repos.Messages.InsertIdempotent(ctx, first)
	require.NoError(t, err)
	require.True(t, result.Inserted)
	require.Equal(t, first.ID, result.Message.ID)

	duplicate := first
	duplicate.ID = "msg_scope_store_idem_second"
	duplicate.Seq = 2
	duplicate.WorkspaceGlobalSeq = 101
	duplicate.Content = first.Content
	result, err = repos.Messages.InsertIdempotent(ctx, duplicate)
	require.NoError(t, err)
	require.False(t, result.Inserted)
	require.Equal(t, first.ID, result.Message.ID)
	require.Equal(t, first.Content, result.Message.Content)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_messages", "scope_id = $1", fixture.scopeID))

	digestConflict := duplicate
	digestConflict.ID = "msg_scope_store_idem_digest_conflict"
	digestConflict.Seq = 3
	digestConflict.WorkspaceGlobalSeq = 102
	digestConflict.Content = "different"
	digestConflict.IdempotencyPayloadDigest = strings.Repeat("b", 64)
	_, err = repos.Messages.InsertIdempotent(ctx, digestConflict)
	require.ErrorIs(t, err, ErrConflict)

	conflict := first
	conflict.ID = "msg_scope_store_idem_conflict"
	conflict.ClientMessageID = "client-scope-store-conflict"
	require.Error(t, repos.Messages.Insert(ctx, conflict))
}

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

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

	for seq := int64(1); seq <= 3; seq++ {
		require.NoError(t, repos.Messages.Insert(ctx, ScopeMessageRecord{
			ID:                 fmt.Sprintf("msg_scope_store_cursor_%d", seq),
			WorkspaceID:        fixture.workspaceID,
			ScopeID:            fixture.scopeID,
			Seq:                seq,
			WorkspaceGlobalSeq: 200 + seq,
			AuthorMemberID:     fixture.ownerMemberID,
			AuthorType:         "human",
			MessageType:        "text",
			Content:            "cursor",
			CreatedAt:          fixture.now.Add(time.Duration(seq) * time.Second),
			UpdatedAt:          fixture.now.Add(time.Duration(seq) * time.Second),
		}))
	}

	latest, err := repos.Messages.ListBeforeSeq(ctx, fixture.scopeID, 0, 2)
	require.NoError(t, err)
	require.Len(t, latest, 2)
	require.Equal(t, int64(3), latest[0].Seq)
	require.Equal(t, int64(2), latest[1].Seq)

	beforeThree, err := repos.Messages.ListBeforeSeq(ctx, fixture.scopeID, 3, 2)
	require.NoError(t, err)
	require.Len(t, beforeThree, 2)
	require.Equal(t, int64(2), beforeThree[0].Seq)
	require.Equal(t, int64(1), beforeThree[1].Seq)

	beforeTwo, err := repos.Messages.ListBeforeSeq(ctx, fixture.scopeID, 2, 1)
	require.NoError(t, err)
	require.Len(t, beforeTwo, 1)
	require.Equal(t, int64(1), beforeTwo[0].Seq)
}

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

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

	message := ScopeMessageRecord{
		ID:                 "msg_scope_store_pending",
		WorkspaceID:        fixture.workspaceID,
		ScopeID:            fixture.scopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 300,
		AuthorMemberID:     fixture.ownerMemberID,
		AuthorType:         "human",
		MessageType:        "text",
		Content:            "pending",
		CreatedAt:          fixture.now,
		UpdatedAt:          fixture.now,
	}
	require.NoError(t, repos.Messages.Insert(ctx, message))

	suppressedAt := fixture.now
	require.NoError(t, repos.ActivityEvents.Insert(ctx, scopeActivityEventForTest(fixture, message.ID, "act_scope_store_pending_30", 30, nil)))
	require.NoError(t, repos.ActivityEvents.Insert(ctx, scopeActivityEventForTest(fixture, message.ID, "act_scope_store_pending_20", 20, nil)))
	require.NoError(t, repos.ActivityEvents.Insert(ctx, scopeActivityEventForTest(fixture, message.ID, "act_scope_store_pending_suppressed", 15, &suppressedAt)))
	activities, err := repos.ActivityEvents.ListAfterCursor(ctx, fixture.workspaceID, fixture.scopeID, 0, 10)
	require.NoError(t, err)
	require.Len(t, activities, 2)
	require.Equal(t, int64(20), activities[0].ActivityGlobalSeq)
	require.Equal(t, int64(30), activities[1].ActivityGlobalSeq)

	actualNow := time.Now().UTC().Truncate(time.Second)
	require.NoError(t, repos.Outbox.Insert(ctx, scopeOutboxForTest(fixture, message.ID, "out_scope_store_pending_b", "dedupe:pending:b", actualNow.Add(-time.Minute), "pending")))
	require.NoError(t, repos.Outbox.Insert(ctx, scopeOutboxForTest(fixture, message.ID, "out_scope_store_pending_a", "dedupe:pending:a", actualNow.Add(-2*time.Minute), "pending")))
	require.NoError(t, repos.Outbox.Insert(ctx, scopeOutboxForTest(fixture, message.ID, "out_scope_store_pending_future", "dedupe:pending:future", actualNow.Add(time.Hour), "pending")))
	require.NoError(t, repos.Outbox.Insert(ctx, scopeOutboxForTest(fixture, message.ID, "out_scope_store_pending_delivered", "dedupe:pending:delivered", actualNow.Add(-3*time.Minute), "delivered")))
	pendingOutbox, err := repos.Outbox.ListPending(ctx, 10)
	require.NoError(t, err)
	require.Len(t, pendingOutbox, 2)
	require.Equal(t, "out_scope_store_pending_a", pendingOutbox[0].ID)
	require.Equal(t, "out_scope_store_pending_b", pendingOutbox[1].ID)
}

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

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

	messageBacked := ScopeActivityEventRecord{
		ID:                "act_scope_store_activity_validation_message",
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		Seq:               10,
		ActivityType:      "reaction_added",
		ActivityRefType:   "message",
		ActivityRefID:     "msg_scope_store_activity_validation",
		ActivityGlobalSeq: 10,
		CreatedAt:         fixture.now,
	}
	err := repos.ActivityEvents.Insert(ctx, messageBacked)
	require.ErrorContains(t, err, "message-backed activity events are not allowed")

	messageCreated := messageBacked
	messageCreated.ActivityRefType = "reaction"
	messageCreated.ActivityType = "message_created"
	err = repos.ActivityEvents.Insert(ctx, messageCreated)
	require.ErrorContains(t, err, "message-backed activity events are not allowed")

	missingGlobalSeq := messageBacked
	missingGlobalSeq.ActivityRefType = "reaction"
	missingGlobalSeq.ActivityType = "reaction_added"
	missingGlobalSeq.Seq = 0
	err = repos.ActivityEvents.Insert(ctx, missingGlobalSeq)
	require.ErrorContains(t, err, "activity seq is required")

	missingRequired := missingGlobalSeq
	missingRequired.Seq = 11
	missingRequired.ActivityRefID = ""
	err = repos.ActivityEvents.Insert(ctx, missingRequired)
	require.ErrorContains(t, err, "activity id, workspace, scope, type, and ref are required")

	valid := scopeActivityEventForTest(fixture, "", "act_scope_store_activity_validation_valid", 12, nil)
	require.NoError(t, repos.ActivityEvents.Insert(ctx, valid))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_activity_events", "id = $1", valid.ID))
}

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

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

	message := ScopeMessageRecord{
		ID:                 "msg_scope_store_dedupe",
		WorkspaceID:        fixture.workspaceID,
		ScopeID:            fixture.scopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 400,
		AuthorMemberID:     fixture.ownerMemberID,
		AuthorType:         "human",
		MessageType:        "text",
		Content:            "dedupe",
		CreatedAt:          fixture.now,
		UpdatedAt:          fixture.now,
	}
	require.NoError(t, repos.Messages.Insert(ctx, message))

	activity := scopeActivityEventForTest(fixture, message.ID, "act_scope_store_dedupe_first", 400, nil)
	require.NoError(t, repos.ActivityEvents.Insert(ctx, activity))
	activity.ID = "act_scope_store_dedupe_second"
	require.NoError(t, repos.ActivityEvents.Insert(ctx, activity))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_activity_events", "activity_ref_id = $1", activity.ActivityRefID))

	outbox := scopeOutboxForTest(fixture, message.ID, "out_scope_store_dedupe_first", "dedupe:scope-store", fixture.now, "pending")
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))
	outbox.ID = "out_scope_store_dedupe_second"
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "dedupe_key = $1", "dedupe:scope-store"))

	reaction := ScopeMessageReactionRecord{
		ID:              "rxn_scope_store_dedupe_first",
		WorkspaceID:     fixture.workspaceID,
		ScopeID:         fixture.scopeID,
		MessageID:       message.ID,
		Emoji:           "👍",
		ReactorMemberID: fixture.ownerMemberID,
		CreatedAt:       fixture.now,
	}
	require.NoError(t, repos.Reactions.Insert(ctx, reaction))
	reaction.ID = "rxn_scope_store_dedupe_second"
	require.NoError(t, repos.Reactions.Insert(ctx, reaction))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_message_reactions", "message_id = $1 AND emoji = $2", message.ID, "👍"))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "set_payload_missing")
	message := scopeMessageForPayloadTest(fixture, "msg_scope_store_set_payload_missing")
	require.NoError(t, repos.Messages.Insert(ctx, message))

	// A different fanout type tied to the same message must not satisfy the
	// authoritative message delivery payload target.
	otherFanout := scopeOutboxForTest(
		fixture,
		message.ID,
		"out_scope_store_set_payload_divergent",
		"dedupe:scope-store:set-payload:divergent",
		fixture.now,
		"pending",
	)
	otherFanout.FanoutType = ScopeOutboxFanoutMessageMutation
	require.NoError(t, repos.Outbox.Insert(ctx, otherFanout))

	payload := scopeFanoutPayloadForMessageTest(t, message.ID)
	err := repos.Outbox.SetPayloadForMessage(ctx, message.ID, payload)
	require.ErrorIs(t, err, ErrNotFound)

	var storedPayload []byte
	require.NoError(t, db.QueryRow(ctx, `SELECT payload FROM scope_fanout_outbox WHERE id = $1`, otherFanout.ID).Scan(&storedPayload))
	require.Empty(t, storedPayload)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "set_payload_update")
	message := scopeMessageForPayloadTest(fixture, "msg_scope_store_set_payload_update")
	require.NoError(t, repos.Messages.Insert(ctx, message))
	outbox := scopeOutboxForTest(
		fixture,
		message.ID,
		"out_scope_store_set_payload_update",
		"dedupe:scope-store:set-payload:update",
		fixture.now,
		"pending",
	)
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))

	payload := scopeFanoutPayloadForMessageTest(t, message.ID)
	require.NoError(t, repos.Outbox.SetPayloadForMessage(ctx, message.ID, payload))

	var storedPayload []byte
	var status string
	require.NoError(t, db.QueryRow(ctx, `SELECT payload, status FROM scope_fanout_outbox WHERE id = $1`, outbox.ID).Scan(&storedPayload, &status))
	require.JSONEq(t, string(payload), string(storedPayload))
	require.Equal(t, "pending", status)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "set_payload_identity")
	targetMessage := scopeMessageForPayloadTest(fixture, "msg_scope_store_set_payload_target")
	require.NoError(t, repos.Messages.Insert(ctx, targetMessage))
	outbox := scopeOutboxForTest(
		fixture,
		targetMessage.ID,
		"out_scope_store_set_payload_identity",
		"dedupe:scope-store:set-payload:identity",
		fixture.now,
		"pending",
	)
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))

	payload := scopeFanoutPayloadForMessageTest(t, "msg_scope_store_set_payload_other")
	err := repos.Outbox.SetPayloadForMessage(ctx, targetMessage.ID, payload)
	require.ErrorIs(t, err, domain.ErrInvalidArgument)

	var storedPayload []byte
	require.NoError(t, db.QueryRow(ctx, `SELECT payload FROM scope_fanout_outbox WHERE id = $1`, outbox.ID).Scan(&storedPayload))
	require.Empty(t, storedPayload)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "set_payload_same")
	message := scopeMessageForPayloadTest(fixture, "msg_scope_store_set_payload_same")
	require.NoError(t, repos.Messages.Insert(ctx, message))
	payload := scopeFanoutPayloadForMessageTest(t, message.ID)
	outbox := scopeOutboxForTest(
		fixture,
		message.ID,
		"out_scope_store_set_payload_same",
		"dedupe:scope-store:set-payload:same",
		fixture.now,
		"pending",
	)
	outbox.Payload = payload
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))

	// PostgreSQL counts this as one matched row, while TiDB/MySQL can count it
	// as zero changed rows. Both database result semantics must be accepted.
	require.NoError(t, repos.Outbox.SetPayloadForMessage(ctx, message.ID, payload))

	var storedPayload []byte
	require.NoError(t, db.QueryRow(ctx, `SELECT payload FROM scope_fanout_outbox WHERE id = $1`, outbox.ID).Scan(&storedPayload))
	require.JSONEq(t, string(payload), string(storedPayload))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "set_payload_multiple")
	message := scopeMessageForPayloadTest(fixture, "msg_scope_store_set_payload_multiple")
	require.NoError(t, repos.Messages.Insert(ctx, message))

	first := scopeOutboxForTest(
		fixture,
		message.ID,
		"out_scope_store_set_payload_multiple_first",
		"dedupe:scope-store:set-payload:multiple:first",
		fixture.now,
		"pending",
	)
	second := scopeOutboxForTest(
		fixture,
		message.ID,
		"out_scope_store_set_payload_multiple_second",
		"dedupe:scope-store:set-payload:multiple:second",
		fixture.now,
		"pending",
	)
	payload := scopeFanoutPayloadForMessageTest(t, message.ID)
	first.Payload = payload
	require.NoError(t, repos.Outbox.Insert(ctx, first))
	require.NoError(t, repos.Outbox.Insert(ctx, second))

	err := repos.Outbox.SetPayloadForMessage(ctx, message.ID, payload)
	require.ErrorIs(t, err, ErrConflict)
	var firstPayload, secondPayload []byte
	require.NoError(t, db.QueryRow(ctx, `SELECT payload FROM scope_fanout_outbox WHERE id = $1`, first.ID).Scan(&firstPayload))
	require.NoError(t, db.QueryRow(ctx, `SELECT payload FROM scope_fanout_outbox WHERE id = $1`, second.ID).Scan(&secondPayload))
	require.JSONEq(t, string(payload), string(firstPayload))
	require.Empty(t, secondPayload, "duplicate-target validation must fail before modifying either authoritative row")
}

type scopeStoreFixture struct {
	workspaceID     string
	ownerUserID     string
	ownerMemberID   string
	privateMemberID string
	scopeID         string
	now             time.Time
}

func seedScopeStoreFixture(t *testing.T, ctx context.Context, db coredb.QueryExecer, suffix string) scopeStoreFixture {
	t.Helper()

	now := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
	fixture := scopeStoreFixture{
		workspaceID:     "wsp_scope_store_" + suffix,
		ownerUserID:     "usr_scope_store_" + suffix,
		ownerMemberID:   "mem_scope_store_" + suffix + "_owner",
		privateMemberID: "mem_scope_store_" + suffix + "_private",
		scopeID:         "scp_scope_store_" + suffix,
		now:             now,
	}
	insertMessagingWorkspace(t, ctx, db, fixture.workspaceID, now)
	insertMessagingUser(t, ctx, db, fixture.ownerUserID, "scope-store-"+suffix+"@example.com", now)
	insertMessagingHumanMember(t, ctx, db, fixture.ownerMemberID, fixture.workspaceID, fixture.ownerUserID, now)
	insertMessagingAgentMember(t, ctx, db, fixture.privateMemberID, fixture.workspaceID, now)

	repos := NewScopeRepositories(db)
	require.NoError(t, repos.Scopes.UpsertScope(ctx, ScopeRecord{
		ID:                fixture.scopeID,
		WorkspaceID:       fixture.workspaceID,
		Type:              "channel",
		IdentityKey:       "channel:scope-store:" + suffix,
		Status:            "active",
		CreatedByMemberID: fixture.ownerMemberID,
		CreatedAt:         now,
		UpdatedAt:         now,
	}))
	require.NoError(t, repos.Memberships.Upsert(ctx, ScopeMembershipRecord{
		ScopeID:    fixture.scopeID,
		MemberID:   fixture.ownerMemberID,
		Visibility: "public",
		CreatedAt:  now,
		UpdatedAt:  now,
	}))
	return fixture
}

func scopeActivityEventForTest(fixture scopeStoreFixture, messageID string, eventID string, globalSeq int64, suppressedAt *time.Time) ScopeActivityEventRecord {
	return ScopeActivityEventRecord{
		ID:                eventID,
		WorkspaceID:       fixture.workspaceID,
		ScopeID:           fixture.scopeID,
		Seq:               globalSeq,
		ActivityType:      "reaction_added",
		ActivityRefType:   "reaction",
		ActivityRefID:     eventID + "_reaction",
		ActivityGlobalSeq: globalSeq,
		SourceMessageID:   messageID,
		ActorMemberID:     fixture.ownerMemberID,
		Reason:            "reaction",
		CreatedAt:         fixture.now,
		SuppressedAt:      suppressedAt,
	}
}

func scopeOutboxForTest(fixture scopeStoreFixture, messageID string, outboxID string, dedupeKey string, nextAttemptAt time.Time, status string) ScopeFanoutOutboxRecord {
	return ScopeFanoutOutboxRecord{
		ID:            outboxID,
		WorkspaceID:   fixture.workspaceID,
		ScopeID:       fixture.scopeID,
		MessageID:     messageID,
		EventRefType:  "message",
		EventRefID:    messageID,
		FanoutType:    "scope_fanout",
		DedupeKey:     dedupeKey,
		Status:        status,
		NextAttemptAt: nextAttemptAt,
		CreatedAt:     fixture.now,
	}
}

func scopeMessageForPayloadTest(fixture scopeStoreFixture, messageID string) ScopeMessageRecord {
	return ScopeMessageRecord{
		ID:                 messageID,
		WorkspaceID:        fixture.workspaceID,
		ScopeID:            fixture.scopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 1,
		AuthorMemberID:     fixture.ownerMemberID,
		AuthorType:         "human",
		MessageType:        "text",
		Content:            "scope fanout payload",
		CreatedAt:          fixture.now,
		UpdatedAt:          fixture.now,
	}
}

func scopeFanoutPayloadForMessageTest(t *testing.T, messageID string) []byte {
	t.Helper()
	payload, err := MarshalScopeFanoutPayload(
		AppMessage{ID: messageID},
		MessageDelivery{MessageID: messageID},
	)
	require.NoError(t, err)
	return payload
}

func countScopeRows(t *testing.T, ctx context.Context, db coredb.Queryer, table string, predicate string, args ...any) int64 {
	t.Helper()

	query := "SELECT COUNT(*) FROM " + table
	if predicate != "" {
		query += " WHERE " + predicate
	}
	var count int64
	require.NoError(t, db.QueryRow(ctx, query, args...).Scan(&count))
	return count
}