package message

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

type scopeFanoutAdapterStub struct {
	err             error
	calls           []ScopeFanoutOutboxRecord
	failedOutboxes  []ScopeFanoutOutboxRecord
	failureMessages []string
}

func (adapter *scopeFanoutAdapterStub) Fanout(_ context.Context, outbox ScopeFanoutOutboxRecord) error {
	adapter.calls = append(adapter.calls, outbox)
	return adapter.err
}

func (adapter *scopeFanoutAdapterStub) FanoutFailed(_ context.Context, outbox ScopeFanoutOutboxRecord, cause error) {
	adapter.failedOutboxes = append(adapter.failedOutboxes, outbox)
	if cause != nil {
		adapter.failureMessages = append(adapter.failureMessages, cause.Error())
	}
}

type scopeFanoutDeferringAdapter struct {
	remainingDeferrals int
	calls              []ScopeFanoutOutboxRecord
}

func (adapter *scopeFanoutDeferringAdapter) Fanout(_ context.Context, outbox ScopeFanoutOutboxRecord) error {
	adapter.calls = append(adapter.calls, outbox)
	if adapter.remainingDeferrals <= 0 {
		return nil
	}
	adapter.remainingDeferrals--
	return errors.Join(ErrScopeFanoutDeferred, errors.New("unread roster is still warming"))
}

type scopeFanoutSequenceAdapter struct {
	errs  []error
	calls []ScopeFanoutOutboxRecord
}

func (adapter *scopeFanoutSequenceAdapter) Fanout(_ context.Context, outbox ScopeFanoutOutboxRecord) error {
	adapter.calls = append(adapter.calls, outbox)
	if len(adapter.errs) == 0 {
		return nil
	}
	err := adapter.errs[0]
	adapter.errs = adapter.errs[1:]
	return err
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos, fixture, _, _, outbox := seedScopeFanoutWorkerFixture(t, ctx, db, "deliver")
	now := fixture.now.Add(time.Hour)
	adapter := &scopeFanoutAdapterStub{}
	worker := NewScopeFanoutWorkerWithClock(db, adapter, "worker-deliver", func() time.Time { return now })

	result, err := worker.RunOnce(ctx, 10)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Delivered: 1}, result)
	require.Len(t, adapter.calls, 1)
	require.Equal(t, outbox.ID, adapter.calls[0].ID)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'delivered' AND delivered_at IS NOT NULL", outbox.ID))
	require.Equal(t, int64(0), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND lease_owner IS NOT NULL", outbox.ID))
	loaded, err := repos.Outbox.ListPending(ctx, 10)
	require.NoError(t, err)
	require.Empty(t, loaded)
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "non_message_intent")
	message := ScopeMessageRecord{
		ID:                 "msg_scope_fanout_non_message_intent",
		WorkspaceID:        fixture.workspaceID,
		ScopeID:            fixture.scopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 100,
		AuthorMemberID:     fixture.ownerMemberID,
		AuthorType:         ScopeActorHuman,
		MessageType:        "text",
		Content:            "activity source",
		CreatedAt:          fixture.now,
		UpdatedAt:          fixture.now,
	}
	require.NoError(t, repos.Messages.Insert(ctx, message))
	activity := scopeActivityEventForTest(fixture, message.ID, "act_scope_fanout_non_message_intent", 101, nil)
	require.NoError(t, repos.ActivityEvents.Insert(ctx, activity))
	outbox, err := BuildScopeDeliveryIntentOutboxRecord(DeliveryIntent{
		WorkspaceID: fixture.workspaceID,
		ScopeID:     fixture.scopeID,
		SourceRef: DeliverySourceRef{
			Type: DeliverySourceRefActivityEvent,
			ID:   activity.ID,
		},
		ReceiveMessage: ReceiveMessage{
			MessageID: activity.ID,
			TargetKey: fixture.scopeID,
			TargetSeq: activity.Seq,
			Content:   "activity wake",
		},
		WakeRecipients: []AgentRecipient{{AgentID: "agt_non_message_intent"}},
		CreatedAt:      fixture.now,
		NextAttemptAt:  fixture.now.Add(-time.Minute),
	})
	require.NoError(t, err)
	require.Empty(t, outbox.MessageID)
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND message_id IS NULL", outbox.ID))

	pending, err := repos.Outbox.ListPending(ctx, 10)
	require.NoError(t, err)
	require.Len(t, pending, 1)
	require.Equal(t, outbox.ID, pending[0].ID)
	require.Empty(t, pending[0].MessageID)

	now := fixture.now.Add(time.Hour)
	adapter := &scopeFanoutAdapterStub{}
	worker := NewScopeFanoutWorkerWithClock(db, adapter, "worker-non-message-intent", func() time.Time { return now })

	result, err := worker.RunOnce(ctx, 10)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Delivered: 1}, result)
	require.Len(t, adapter.calls, 1)
	require.Equal(t, outbox.ID, adapter.calls[0].ID)
	require.Empty(t, adapter.calls[0].MessageID)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'delivered' AND delivered_at IS NOT NULL", outbox.ID))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	_, fixture, _, activity, outbox := seedScopeFanoutWorkerFixture(t, ctx, db, "suppressed")
	now := fixture.now.Add(time.Hour)
	_, err := db.Exec(ctx, `UPDATE scope_activity_events SET suppressed_at = $1 WHERE id = $2`, now, activity.ID)
	require.NoError(t, err)

	adapter := &scopeFanoutAdapterStub{}
	worker := NewScopeFanoutWorkerWithClock(db, adapter, "worker-suppressed", func() time.Time { return now })
	result, err := worker.RunOnce(ctx, 10)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Suppressed: 1}, result)
	require.Empty(t, adapter.calls)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'suppressed' AND suppression_reason = 'activity_suppressed'", outbox.ID))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	_, fixture, message, _, outbox := seedScopeFanoutWorkerFixture(t, ctx, db, "missing_payload")
	now := fixture.now.Add(time.Hour)
	_, err := db.Exec(ctx, `
UPDATE scope_fanout_outbox
SET event_ref_type = 'message', event_ref_id = $1, payload = NULL
WHERE id = $2
`, message.ID, outbox.ID)
	require.NoError(t, err)

	adapter := &scopeFanoutAdapterStub{}
	worker := NewScopeFanoutWorkerWithClock(db, adapter, "worker-missing-payload", func() time.Time { return now })
	result, err := worker.RunOnce(ctx, 10)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Suppressed: 1}, result)
	require.Empty(t, adapter.calls)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'suppressed' AND suppression_reason = 'payload_missing'", outbox.ID))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	_, fixture, _, _, outbox := seedScopeFanoutWorkerFixture(t, ctx, db, "retry")
	now := fixture.now.Add(time.Hour)
	adapter := &scopeFanoutAdapterStub{err: errors.New("runtime unavailable")}
	worker := NewScopeFanoutWorkerWithClock(db, adapter, "worker-retry", func() time.Time { return now })
	worker.backoff = func(int) time.Duration { return 5 * time.Second }

	result, err := worker.RunOnce(ctx, 10)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Retried: 1}, result)
	require.Empty(t, adapter.failedOutboxes)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'pending' AND attempts = 1 AND last_error = 'runtime unavailable'", outbox.ID))
	_, err = db.Exec(ctx, `UPDATE scope_fanout_outbox SET attempts = max_attempts - 1, next_attempt_at = $1 WHERE id = $2`, now.Add(-time.Second), outbox.ID)
	require.NoError(t, err)
	result, err = worker.RunOnce(ctx, 10)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Failed: 1}, result)
	require.Len(t, adapter.failedOutboxes, 1)
	require.Equal(t, outbox.ID, adapter.failedOutboxes[0].ID)
	require.Equal(t, []string{"runtime unavailable"}, adapter.failureMessages)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'failed' AND attempts = max_attempts", outbox.ID))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	_, groupAFixture, _, _, groupAOutbox := seedScopeFanoutWorkerFixture(t, ctx, db, "workspace_group_a")
	_, groupBFixture, _, _, groupBOutbox := seedScopeFanoutWorkerFixture(t, ctx, db, "workspace_group_b")
	now := time.Now().UTC().Add(time.Hour).Truncate(time.Second)

	_, err := db.Exec(ctx, `
INSERT INTO workspace_placements (workspace_id, workspace_group_id, state, created_at, updated_at)
VALUES ($1, 'scope-fanout-group-a', 'active', $3, $3),
       ($2, 'scope-fanout-group-b', 'moving', $3, $3)
`, groupAFixture.workspaceID, groupBFixture.workspaceID, now.Add(-time.Hour))
	require.NoError(t, err)
	_, err = db.Exec(ctx, `
UPDATE scope_fanout_outbox
SET next_attempt_at = CASE id WHEN $1 THEN $3 ELSE $4 END
WHERE id IN ($1, $2)
`, groupAOutbox.ID, groupBOutbox.ID, now.Add(-2*time.Minute), now.Add(-time.Minute))
	require.NoError(t, err)

	adapter := &scopeFanoutAdapterStub{}
	worker := NewScopeFanoutWorkerWithOptions(db, adapter, "worker-workspace-group-b", ScopeFanoutWorkerOptions{
		WorkspaceGroupID: " scope-fanout-group-b ",
	})
	worker.now = func() time.Time { return now }

	result, err := worker.RunOnce(ctx, 1)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Delivered: 1}, result)
	require.Len(t, adapter.calls, 1)
	require.Equal(t, groupBOutbox.ID, adapter.calls[0].ID)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'pending' AND attempts = 0", groupAOutbox.ID))

	// Reuse the same rows as expired leases. The older row belongs to group A,
	// so a global LIMIT 1 would starve group B unless placement ownership is
	// applied in the expired-candidate SQL before ORDER BY/LIMIT.
	now = now.Add(time.Hour)
	_, err = db.Exec(ctx, `
UPDATE scope_fanout_outbox
SET status = 'leased', attempts = 1, lease_owner = 'expired-worker',
    lease_expires_at = CASE id WHEN $1 THEN $3 ELSE $4 END,
    next_attempt_at = CASE id WHEN $1 THEN $5 ELSE $6 END,
    delivered_at = NULL
WHERE id IN ($1, $2)
`, groupAOutbox.ID, groupBOutbox.ID,
		now.Add(-2*time.Minute), now.Add(-time.Minute),
		now.Add(-4*time.Minute), now.Add(-3*time.Minute))
	require.NoError(t, err)
	_, err = db.Exec(ctx, `UPDATE workspace_placements SET state = 'draining', updated_at = $2 WHERE workspace_id = $1`, groupBFixture.workspaceID, now)
	require.NoError(t, err)
	adapter.calls = nil

	result, err = worker.RunOnce(ctx, 1)
	require.NoError(t, err)
	require.Equal(t, ScopeFanoutRunResult{Leased: 1, Delivered: 1}, result)
	require.Len(t, adapter.calls, 1)
	require.Equal(t, groupBOutbox.ID, adapter.calls[0].ID)
	require.Equal(t, int64(1), countScopeRows(t, ctx, db, "scope_fanout_outbox", "id = $1 AND status = 'leased' AND lease_owner = 'expired-worker' AND attempts = 1", groupAOutbox.ID))
}

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

	ctx := context.Background()
	deliveredAt := time.Date(2026, 5, 21, 12, 0, 0, 0, time.FixedZone("CST", 8*60*60))
	execer := &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 1}

	err := NewScopeRepositories(execer).Outbox.MarkDeliveredForMessage(ctx, " msg-1 ", deliveredAt)

	require.NoError(t, err)
	require.Contains(t, execer.query, "status = 'pending'")
	require.NotContains(t, execer.query, "'leased'")
	require.Contains(t, execer.query, "fanout_type = 'scope_fanout'")
	require.Equal(t, []any{"msg-1", deliveredAt.UTC()}, execer.args)
}

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

	ctx := context.Background()
	execer := &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 0}

	err := NewScopeRepositories(execer).Outbox.MarkDeliveredForMessage(ctx, "msg-missing", time.Date(2026, 5, 21, 12, 0, 0, 0, time.UTC))

	require.ErrorIs(t, err, ErrNotFound)
	require.Contains(t, execer.query, "status = 'pending'")
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos, fixture, sourceMessage, _, _ := seedScopeFanoutWorkerFixture(t, ctx, db, "inline_receipt")
	outbox := BuildScopeMessageFanoutOutboxRecord(ScopeMessageFanoutOutboxRecordParams{
		WorkspaceID:   fixture.workspaceID,
		ScopeID:       fixture.scopeID,
		MessageID:     sourceMessage.ID,
		Payload:       []byte(`{"version":1}`),
		CreatedAt:     fixture.now,
		NextAttemptAt: fixture.now.Add(time.Hour),
	})
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))
	now := fixture.now.Add(time.Hour)

	claimed, ok, err := repos.Outbox.ClaimForMessageInline(
		ctx,
		sourceMessage.ID,
		"inline-owner-a",
		now,
		now.Add(time.Minute),
	)
	require.NoError(t, err)
	require.True(t, ok)
	require.Equal(t, outbox.ID, claimed.ID)
	require.Equal(t, "inline-owner-a", claimed.LeaseOwner)
	require.NotNil(t, claimed.LeaseExpiresAt)
	require.NoError(t, repos.Outbox.RenewLease(
		ctx,
		outbox.ID,
		"inline-owner-a",
		*claimed.LeaseExpiresAt,
	), "a second-precision no-op renewal must preserve the current owner's lease")
	require.ErrorIs(t, repos.Outbox.RenewLease(
		ctx,
		outbox.ID,
		"stale-inline-owner",
		*claimed.LeaseExpiresAt,
	), ErrNotFound, "a zero-row renewal must still reject a stale owner")

	_, ok, err = repos.Outbox.ClaimForMessageInline(
		ctx,
		sourceMessage.ID,
		"inline-owner-b",
		now.Add(time.Second),
		now.Add(2*time.Minute),
	)
	require.NoError(t, err)
	require.False(t, ok, "a live inline owner must exclude the worker and other request paths")

	recorded, err := repos.Outbox.HasAgentDeliveryReceipt(ctx, DeliverySourceRefMessage, sourceMessage.ID, "agt_receipt")
	require.NoError(t, err)
	require.False(t, recorded)
	require.NoError(t, repos.Outbox.RecordAgentDeliveryReceipt(
		ctx,
		outbox.ID,
		"inline-owner-a",
		DeliverySourceRefMessage,
		sourceMessage.ID,
		"agt_receipt",
		now.Add(2*time.Second),
	))
	recorded, err = repos.Outbox.HasAgentDeliveryReceipt(ctx, DeliverySourceRefMessage, sourceMessage.ID, "agt_receipt")
	require.NoError(t, err)
	require.True(t, recorded)

	require.ErrorIs(t, repos.Outbox.RecordAgentDeliveryReceipt(
		ctx,
		outbox.ID,
		"stale-owner",
		DeliverySourceRefMessage,
		sourceMessage.ID,
		"agt_stale",
		now.Add(3*time.Second),
	), ErrNotFound)
	require.NoError(t, repos.Outbox.MarkRetryOrFailed(ctx, outbox.ID, "inline-owner-a", now.Add(4*time.Second), "partial delivery"))

	claimed, ok, err = repos.Outbox.ClaimForMessageInline(
		ctx,
		sourceMessage.ID,
		"inline-owner-b",
		now.Add(5*time.Second),
		now.Add(2*time.Minute),
	)
	require.NoError(t, err)
	require.True(t, ok)
	require.Equal(t, "inline-owner-b", claimed.LeaseOwner)
	recorded, err = repos.Outbox.HasAgentDeliveryReceipt(ctx, DeliverySourceRefMessage, sourceMessage.ID, "agt_receipt")
	require.NoError(t, err)
	require.True(t, recorded, "a successful recipient must be skipped by a partial retry")
	require.NoError(t, repos.Outbox.RenewLease(ctx, outbox.ID, "inline-owner-b", now.Add(3*time.Minute)))
	require.ErrorIs(t, repos.Outbox.RenewLease(ctx, outbox.ID, "inline-owner-a", now.Add(3*time.Minute)), ErrNotFound)
	require.NoError(t, repos.Outbox.MarkDelivered(ctx, outbox.ID, "inline-owner-b", now.Add(6*time.Second)))
}

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

	ctx := context.Background()
	db := startMessagingTestPool(t, ctx)
	repos, fixture, sourceMessage, _, _ := seedScopeFanoutWorkerFixture(t, ctx, db, "renew_same_expiry")
	outbox := BuildScopeMessageFanoutOutboxRecord(ScopeMessageFanoutOutboxRecordParams{
		WorkspaceID:   fixture.workspaceID,
		ScopeID:       fixture.scopeID,
		MessageID:     sourceMessage.ID,
		Payload:       []byte(`{"version":1}`),
		CreatedAt:     fixture.now,
		NextAttemptAt: fixture.now,
	})
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))

	leaseUntil := fixture.now.Add(time.Minute).Truncate(time.Millisecond)
	claimed, ok, err := repos.Outbox.ClaimForMessageInline(
		ctx,
		sourceMessage.ID,
		"inline-same-expiry",
		fixture.now,
		leaseUntil,
	)
	require.NoError(t, err)
	require.True(t, ok)
	require.NotNil(t, claimed.LeaseExpiresAt)

	require.NoError(t, repos.Outbox.RenewLease(
		ctx,
		outbox.ID,
		"inline-same-expiry",
		claimed.LeaseExpiresAt.UTC(),
	))
	require.ErrorIs(t, repos.Outbox.RenewLease(
		ctx,
		outbox.ID,
		"stale-owner",
		claimed.LeaseExpiresAt.UTC(),
	), ErrNotFound)
}

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

	ctx := context.Background()
	deliveredAt := time.Date(2026, 5, 21, 13, 0, 0, 0, time.FixedZone("CST", 8*60*60))
	execer := &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 1}

	err := NewScopeRepositories(execer).Outbox.MarkDeliveredForSource(ctx, " activity_event ", " act-1 ", "", deliveredAt)

	require.NoError(t, err)
	require.Contains(t, execer.query, "event_ref_type = $1")
	require.Contains(t, execer.query, "status IN ('pending', 'leased', 'pending_v2', 'leased_v2')")
	require.Equal(t, []any{"activity_event", "act-1", ScopeOutboxFanoutScope, deliveredAt.UTC()}, execer.args)

	execer = &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 0}
	err = NewScopeRepositories(execer).Outbox.MarkDeliveredForSource(ctx, "activity_event", "act-missing", ScopeOutboxFanoutScope, deliveredAt)

	require.ErrorIs(t, err, ErrNotFound)
	require.Contains(t, execer.query, "status IN ('pending', 'leased', 'pending_v2', 'leased_v2')")
}

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

	ctx := context.Background()
	nextAttemptAt := time.Date(2026, 7, 29, 2, 30, 0, 0, time.UTC)
	execer := &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 1}

	err := NewScopeRepositories(execer).Outbox.MarkDeferred(
		ctx,
		" out-v2 ",
		" worker-v2 ",
		nextAttemptAt,
		" roster warming ",
	)

	require.NoError(t, err)
	require.Contains(t, execer.query, "status = 'pending_v2'")
	require.Contains(t, execer.query, "attempts = GREATEST(attempts - 1, 0)")
	require.Contains(t, execer.query, "AND status = 'leased_v2'")
	require.NotContains(t, execer.query, "status = 'pending'")
	require.Equal(t, []any{"out-v2", nextAttemptAt, "roster warming", "worker-v2"}, execer.args)

	execer = &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 0}
	err = NewScopeRepositories(execer).Outbox.MarkDeferred(
		ctx,
		"out-v1-or-stale",
		"worker-v2",
		nextAttemptAt,
		"roster warming",
	)
	require.ErrorIs(t, err, ErrNotFound)
}

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

	ctx := context.Background()
	suppressedAt := time.Date(2026, 5, 21, 13, 30, 0, 0, time.FixedZone("CST", 8*60*60))
	execer := &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 1}

	err := NewScopeRepositories(execer).Outbox.MarkSuppressedForSource(ctx, " activity_event ", " act-1 ", "", suppressedAt, " daemon_notify_failed ")

	require.NoError(t, err)
	require.Contains(t, execer.query, "event_ref_type = $1")
	require.Contains(t, execer.query, "status IN ('pending', 'leased', 'pending_v2', 'leased_v2')")
	require.Equal(t, []any{"activity_event", "act-1", ScopeOutboxFanoutScope, suppressedAt.UTC(), "daemon_notify_failed"}, execer.args)

	execer = &scopeOutboxMarkDeliveredExec{t: t, rowsAffected: 0}
	err = NewScopeRepositories(execer).Outbox.MarkSuppressedForSource(ctx, "activity_event", "act-missing", ScopeOutboxFanoutScope, suppressedAt, "daemon_notify_failed")

	require.ErrorIs(t, err, ErrNotFound)
	require.Contains(t, execer.query, "status IN ('pending', 'leased', 'pending_v2', 'leased_v2')")
}

type scopeOutboxMarkDeliveredExec struct {
	t            *testing.T
	rowsAffected int64
	query        string
	args         []any
}

func (execer *scopeOutboxMarkDeliveredExec) Exec(ctx context.Context, query string, args ...any) (coredb.Result, error) {
	execer.t.Helper()
	execer.query = query
	execer.args = append([]any(nil), args...)
	return scopeOutboxMarkDeliveredResult{rowsAffected: execer.rowsAffected}, nil
}

func (execer *scopeOutboxMarkDeliveredExec) Query(context.Context, string, ...any) (coredb.Rows, error) {
	return nil, errors.New("unexpected Query")
}

func (execer *scopeOutboxMarkDeliveredExec) QueryRow(context.Context, string, ...any) coredb.Row {
	return scopeOutboxMarkDeliveredErrorRow{err: errors.New("unexpected QueryRow")}
}

type scopeOutboxMarkDeliveredResult struct {
	rowsAffected int64
}

func (result scopeOutboxMarkDeliveredResult) RowsAffected() int64 {
	return result.rowsAffected
}

type scopeOutboxMarkDeliveredErrorRow struct {
	err error
}

func (row scopeOutboxMarkDeliveredErrorRow) Scan(...any) error {
	return row.err
}

func seedScopeFanoutWorkerFixture(t *testing.T, ctx context.Context, db coredb.QueryExecer, suffix string) (ScopeRepositories, scopeStoreFixture, ScopeMessageRecord, ScopeActivityEventRecord, ScopeFanoutOutboxRecord) {
	t.Helper()

	repos := NewScopeRepositories(db)
	fixture := seedScopeStoreFixture(t, ctx, db, "fanout_"+suffix)
	message := ScopeMessageRecord{
		ID:                 "msg_scope_fanout_" + suffix,
		WorkspaceID:        fixture.workspaceID,
		ScopeID:            fixture.scopeID,
		Seq:                1,
		WorkspaceGlobalSeq: 100,
		AuthorMemberID:     fixture.ownerMemberID,
		AuthorType:         ScopeActorHuman,
		MessageType:        "text",
		Content:            "fanout",
		CreatedAt:          fixture.now,
		UpdatedAt:          fixture.now,
	}
	require.NoError(t, repos.Messages.Insert(ctx, message))
	activity := scopeActivityEventForTest(fixture, message.ID, "act_scope_fanout_"+suffix, 101, nil)
	require.NoError(t, repos.ActivityEvents.Insert(ctx, activity))
	outbox := scopeOutboxForTest(fixture, message.ID, "out_scope_fanout_"+suffix, "dedupe:fanout:"+suffix, fixture.now.Add(-time.Minute), "pending")
	outbox.EventRefType = "activity_event"
	outbox.EventRefID = activity.ID
	outbox.FanoutType = ScopeOutboxFanoutScope
	require.NoError(t, repos.Outbox.Insert(ctx, outbox))
	return repos, fixture, message, activity, outbox
}