package events

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedevents "github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestEnqueueEventNoSubscriberDropUsesRegisteredObjects(t *testing.T) {
	restore := ResetDefaultNoSubscriberRecordEventObjectsForTest()
	t.Cleanup(restore)

	repo := &outboxRepositoryStub{}
	service := NewOutboxService(repo, nil, nil).
		WithMetadataLookup(staticOutboxMetadataLookup{hasActiveRecordTriggeredFlow: false})
	err := service.EnqueueEvent(context.Background(), sharedevents.RecordCreated, RecordEventPayload{
		ObjectAPIName: constants.TableFlowInterview,
		Record: models.SObject{
			"Id": "4FI000000000001AAA",
		},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}
	if repo.enqueueCalls != 1 {
		t.Fatalf("enqueue calls without registered no-subscriber object = %d, want 1", repo.enqueueCalls)
	}

	RegisterDefaultNoSubscriberRecordEventObjects(constants.TableFlowInterview)

	tests := []struct {
		name         string
		metadata     OutboxMetadataLookup
		wantEnqueues int
	}{
		{
			name:         "drops when trigger subscription lookup reports no active subscriber",
			metadata:     staticOutboxMetadataLookup{hasActiveRecordTriggeredFlow: false},
			wantEnqueues: 0,
		},
		{
			name:         "keeps when trigger subscription lookup reports an active subscriber",
			metadata:     staticOutboxMetadataLookup{hasActiveRecordTriggeredFlow: true},
			wantEnqueues: 1,
		},
		{
			name:         "keeps when metadata lookup is unwired",
			metadata:     nil,
			wantEnqueues: 1,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			repo := &outboxRepositoryStub{}
			service := NewOutboxService(repo, nil, nil)
			if tt.metadata != nil {
				service.WithMetadataLookup(tt.metadata)
			}

			err := service.EnqueueEvent(context.Background(), sharedevents.RecordCreated, RecordEventPayload{
				ObjectAPIName: constants.TableFlowInterview,
				Record: models.SObject{
					"Id": "4FI000000000001AAA",
				},
			})
			if err != nil {
				t.Fatalf("EnqueueEvent() error = %v", err)
			}
			if repo.enqueueCalls != tt.wantEnqueues {
				t.Fatalf("enqueue calls = %d, want %d", repo.enqueueCalls, tt.wantEnqueues)
			}
		})
	}
}

func TestEnqueueEventRoutesRecordEventsToDedicatedOutboxWhenEnabled(t *testing.T) {
	t.Setenv(recordEventOutboxEnv, "true")

	mainRepo := &outboxRepositoryStub{}
	recordRepo := &outboxRepositoryStub{}
	service := NewOutboxService(mainRepo, nil, nil).
		WithRecordEventRepositories(recordRepo, nil)

	err := service.EnqueueEvent(context.Background(), sharedevents.RecordCreated, RecordEventPayload{
		ObjectAPIName: "Lead",
		Record: models.SObject{
			"Id": "00Q-record",
		},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}
	if mainRepo.enqueueCalls != 0 {
		t.Fatalf("main enqueue calls = %d, want 0", mainRepo.enqueueCalls)
	}
	if recordRepo.enqueueCalls != 1 {
		t.Fatalf("record-event enqueue calls = %d, want 1", recordRepo.enqueueCalls)
	}
}

func TestEnqueueEventKeepsRecordEventsOnMainOutboxWhenDedicatedRouteDisabled(t *testing.T) {
	t.Setenv(recordEventOutboxEnv, "false")

	mainRepo := &outboxRepositoryStub{}
	recordRepo := &outboxRepositoryStub{}
	service := NewOutboxService(mainRepo, nil, nil).
		WithRecordEventRepositories(recordRepo, nil)

	err := service.EnqueueEvent(context.Background(), sharedevents.RecordCreated, RecordEventPayload{
		ObjectAPIName: "Lead",
		Record: models.SObject{
			"Id": "00Q-record",
		},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}
	if mainRepo.enqueueCalls != 1 {
		t.Fatalf("main enqueue calls = %d, want 1", mainRepo.enqueueCalls)
	}
	if recordRepo.enqueueCalls != 0 {
		t.Fatalf("record-event enqueue calls = %d, want 0", recordRepo.enqueueCalls)
	}
}

func TestEnqueueEventExplicitDedicatedRouteDoesNotDependOnProcessEnvironment(t *testing.T) {
	t.Setenv(recordEventOutboxEnv, "false")

	mainRepo := &outboxRepositoryStub{}
	recordRepo := &outboxRepositoryStub{}
	service := NewOutboxService(mainRepo, nil, nil).
		WithRecordEventRepositories(recordRepo, nil).
		WithDedicatedRecordEventRouting()

	err := service.EnqueueEvent(context.Background(), sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}
	if mainRepo.enqueueCalls != 0 || recordRepo.enqueueCalls != 1 {
		t.Fatalf("enqueue calls main=%d record=%d, want 0/1", mainRepo.enqueueCalls, recordRepo.enqueueCalls)
	}
}

func TestEnqueueEventDoesNotRouteNonRecordEvents(t *testing.T) {
	t.Setenv(recordEventOutboxEnv, "true")

	mainRepo := &outboxRepositoryStub{}
	recordRepo := &outboxRepositoryStub{}
	service := NewOutboxService(mainRepo, nil, nil).
		WithRecordEventRepositories(recordRepo, nil)

	err := service.EnqueueEvent(context.Background(), sharedevents.ObjectCreated, RecordEventPayload{
		ObjectAPIName: "Object",
		Record: models.SObject{
			"Id": "obj-1",
		},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}
	if mainRepo.enqueueCalls != 1 {
		t.Fatalf("main enqueue calls = %d, want 1", mainRepo.enqueueCalls)
	}
	if recordRepo.enqueueCalls != 0 {
		t.Fatalf("record-event enqueue calls = %d, want 0", recordRepo.enqueueCalls)
	}
}

func TestOutboxWorkerUsesDetachedProcessContext(t *testing.T) {
	src := readEventsSourceForOutboxStructure(t, "outbox_service.go")

	if strings.Contains(src, "ProcessOutbox(context.Background())") {
		t.Fatal("outbox worker must not process events with an unbounded context.Background()")
	}
	for _, forbidden := range []string{
		"outboxWorkerProcessContext(os.stopCh)",
		"func outboxEventProcessContext(",
		"func outboxEventPublishContext(",
		"context.WithTimeout(",
		"os.processEventAtomic(eventCtx, env.ID, env.EventType, env.Payload)",
	} {
		if strings.Contains(src, forbidden) {
			t.Fatalf("outbox_service.go still contains unsafe worker context fragment %q", forbidden)
		}
	}
	for _, symbol := range []string{
		"func outboxWorkerProcessContext() (context.Context, context.CancelFunc)",
		"workerCtx, cancel := outboxWorkerProcessContext()",
		"go os.runOutboxPullWorker(workerID, interval)",
		"processed, err := os.processNextOutboxEvent(workerCtx, workerID)",
		"os.processEventAtomic(ctx, ev.ID, ev.EventType, ev.Payload)",
		"os.publishOutboxEvent(pubCtx, eventType, publishPayload)",
		"return os.eventBus.Publish(ctx, events.EventType(eventType), payload)",
	} {
		if !strings.Contains(src, symbol) {
			t.Fatalf("outbox_service.go is missing detached worker context fragment %q", symbol)
		}
	}
}

func TestOutboxWorkerProcessContextHasNoBatchDeadline(t *testing.T) {
	ctx, cancel := outboxWorkerProcessContext()
	defer cancel()

	if _, ok := ctx.Deadline(); ok {
		t.Fatal("outbox worker batch context must not impose a fixed deadline on the whole batch")
	}
}

func TestOutboxPullWorkerFlagDefaultsOff(t *testing.T) {
	t.Setenv(outboxPullWorkerEnv, "")
	if outboxPullWorkerEnabled() {
		t.Fatal("OUTBOX_PULL_WORKER blank enabled pull workers; want legacy fallback by default")
	}

	t.Setenv(outboxPullWorkerEnv, "true")
	if !outboxPullWorkerEnabled() {
		t.Fatal("OUTBOX_PULL_WORKER=true did not enable pull workers")
	}
}

func TestStartWorkerBlockedEventDoesNotBlockUnrelatedEvent(t *testing.T) {
	t.Setenv(outboxPullWorkerEnv, "true")
	t.Setenv(outboxParallelismEnv, "2")

	blockedPayload := marshalOutboxTestEnvelope(t, "Lead", "blocked-record")
	freePayload := marshalOutboxTestEnvelope(t, "Lead", "free-record")
	repo := newIsolationOutboxRepository([]models.OutboxEvent{
		{ID: "evt-blocked", EventType: string(sharedevents.RecordUpdated), Payload: blockedPayload},
		{ID: "evt-free", EventType: string(sharedevents.RecordUpdated), Payload: freePayload},
	})
	bus := newBlockingEventBus("blocked-record")
	svc := NewOutboxService(repo, bus, noopTxManager{})

	svc.StartWorker(5 * time.Millisecond)
	defer func() {
		bus.Unblock()
		svc.StopWorker()
	}()

	select {
	case <-bus.blockedSeen:
	case <-time.After(2 * time.Second):
		t.Fatal("blocked event was not claimed by a worker")
	}

	select {
	case <-repo.processed("evt-free"):
	case <-time.After(2 * time.Second):
		t.Fatal("unrelated event did not process while another worker was blocked")
	}

	if repo.statusOf("evt-blocked") != "processing" {
		t.Fatalf("blocked event status = %q, want processing", repo.statusOf("evt-blocked"))
	}
}

func TestStartWorkerBatchModeUsesLegacyPendingSelection(t *testing.T) {
	t.Setenv(outboxPullWorkerEnv, "")
	t.Setenv(outboxParallelismEnv, "2")

	firstPayload := marshalOutboxTestEnvelope(t, "Lead", "legacy-one")
	secondPayload := marshalOutboxTestEnvelope(t, "Lead", "legacy-two")
	repo := &batchModeOutboxRepository{
		isolationOutboxRepository: newIsolationOutboxRepository([]models.OutboxEvent{
			{ID: "evt-legacy-one", EventType: string(sharedevents.RecordUpdated), Payload: firstPayload},
			{ID: "evt-legacy-two", EventType: string(sharedevents.RecordUpdated), Payload: secondPayload},
		}),
	}
	svc := NewOutboxService(repo, newBlockingEventBus("never-block"), noopTxManager{})

	stopped := false
	svc.StartWorker(5 * time.Millisecond)
	defer func() {
		if !stopped {
			svc.StopWorker()
		}
	}()

	for _, id := range []string{"evt-legacy-one", "evt-legacy-two"} {
		select {
		case <-repo.processed(id):
		case <-time.After(2 * time.Second):
			t.Fatalf("batch worker did not process %s with legacy pending selection", id)
		}
	}

	svc.StopWorker()
	stopped = true

	if repo.fifoCallCount() != 0 {
		t.Fatal("batch worker used RecordKey FIFO selection; want legacy pending selection while OUTBOX_PULL_WORKER is off")
	}
	if repo.backfillCallCount() != 0 {
		t.Fatal("batch worker started RecordKey backfill while OUTBOX_PULL_WORKER is off")
	}
}

func TestStartWorkerPullModeFallsBackToBatchWhenRecordKeyColumnMissing(t *testing.T) {
	t.Setenv(outboxPullWorkerEnv, "true")
	t.Setenv(outboxParallelismEnv, "2")

	firstPayload := marshalOutboxTestEnvelope(t, "Lead", "legacy-one")
	secondPayload := marshalOutboxTestEnvelope(t, "Lead", "legacy-two")
	repo := &batchModeOutboxRepository{
		isolationOutboxRepository: newIsolationOutboxRepository([]models.OutboxEvent{
			{ID: "evt-legacy-one", EventType: string(sharedevents.RecordUpdated), Payload: firstPayload},
			{ID: "evt-legacy-two", EventType: string(sharedevents.RecordUpdated), Payload: secondPayload},
		}),
		recordKeyColumnPresent: false,
	}
	svc := NewOutboxService(repo, newBlockingEventBus("never-block"), noopTxManager{})

	stopped := false
	svc.StartWorker(5 * time.Millisecond)
	defer func() {
		if !stopped {
			svc.StopWorker()
		}
	}()

	for _, id := range []string{"evt-legacy-one", "evt-legacy-two"} {
		select {
		case <-repo.processed(id):
		case <-time.After(2 * time.Second):
			t.Fatalf("fallback batch worker did not process %s when RecordKey column is missing", id)
		}
	}

	svc.StopWorker()
	stopped = true

	if repo.fifoCallCount() != 0 {
		t.Fatal("pull fallback used RecordKey FIFO selection despite missing column")
	}
	if repo.backfillCallCount() != 0 {
		t.Fatal("pull fallback started RecordKey backfill despite missing column")
	}
}

func TestProcessOutboxUsesLegacyPendingSelectionWhenRecordKeyColumnMissing(t *testing.T) {
	firstPayload := marshalOutboxTestEnvelope(t, "Lead", "legacy-one")
	secondPayload := marshalOutboxTestEnvelope(t, "Lead", "legacy-two")
	repo := &batchModeOutboxRepository{
		isolationOutboxRepository: newIsolationOutboxRepository([]models.OutboxEvent{
			{ID: "evt-legacy-one", EventType: string(sharedevents.RecordUpdated), Payload: firstPayload},
			{ID: "evt-legacy-two", EventType: string(sharedevents.RecordUpdated), Payload: secondPayload},
		}),
		recordKeyColumnPresent: false,
	}
	svc := NewOutboxService(repo, newBlockingEventBus("never-block"), noopTxManager{})

	if err := svc.ProcessOutbox(context.Background()); err != nil {
		t.Fatalf("ProcessOutbox: %v", err)
	}

	for _, id := range []string{"evt-legacy-one", "evt-legacy-two"} {
		if got := repo.statusOf(id); got != OutboxStatusProcessed {
			t.Fatalf("status(%s) = %q, want %s", id, got, OutboxStatusProcessed)
		}
	}
	if repo.fifoCallCount() != 0 {
		t.Fatal("ProcessOutbox used RecordKey FIFO selection despite missing column")
	}
}

func TestStartWorkerPullModeStartsWhenRecordKeyBackfillFails(t *testing.T) {
	t.Setenv(outboxPullWorkerEnv, "true")
	t.Setenv(outboxParallelismEnv, "1")

	payload := marshalOutboxTestEnvelope(t, "Lead", "ready-record")
	repo := &failingBackfillIsolationOutboxRepository{
		isolationOutboxRepository: newIsolationOutboxRepository([]models.OutboxEvent{
			{ID: "evt-ready", EventType: string(sharedevents.RecordUpdated), Payload: payload},
		}),
	}
	svc := NewOutboxService(repo, &publishContextRecordingBus{}, noopTxManager{})

	stopped := false
	svc.StartWorker(5 * time.Millisecond)
	defer func() {
		if !stopped {
			svc.StopWorker()
		}
	}()

	select {
	case <-repo.processed("evt-ready"):
	case <-time.After(2 * time.Second):
		t.Fatal("pull worker did not process an event after RecordKey backfill failed once")
	}

	svc.StopWorker()
	stopped = true
}

func TestOutboxServiceUsesWorkerRepositoryForDrainBookkeeping(t *testing.T) {
	primary := &outboxRepositoryStub{}
	worker := newIsolationOutboxRepository([]models.OutboxEvent{{
		ID:         "evt-worker-1",
		EventType:  string(sharedevents.RecordCreated),
		Payload:    `{"objectApiName":"Account","record":{"Id":"001-worker"}}`,
		RetryCount: 0,
	}})
	bus := &publishContextRecordingBus{}

	svc := NewOutboxService(primary, bus, noopTxManager{}).WithWorkerRepository(worker)
	if err := svc.ProcessOutbox(context.Background()); err != nil {
		t.Fatalf("ProcessOutbox: %v", err)
	}

	if primary.enqueueCalls != 0 {
		t.Fatalf("primary enqueue calls = %d, want 0", primary.enqueueCalls)
	}
	worker.mu.Lock()
	status := worker.status["evt-worker-1"]
	worker.mu.Unlock()
	if status != OutboxStatusProcessed {
		t.Fatalf("worker status = %q, want %q", status, OutboxStatusProcessed)
	}
	if bus.publishCalls != 1 {
		t.Fatal("event bus was not called")
	}
}

func TestRecordKeyBackfillWorkerUsesWorkerRepositoryGuard(t *testing.T) {
	primary := &outboxRepositoryStub{}
	worker := &batchModeOutboxRepository{
		isolationOutboxRepository: newIsolationOutboxRepository(nil),
		recordKeyColumnPresent:    true,
	}
	svc := NewOutboxService(primary, newBlockingEventBus("never-block"), noopTxManager{}).
		WithWorkerRepository(worker)

	stopped := false
	svc.startOutboxRecordKeyBackfillWorker(time.Millisecond)
	defer func() {
		if !stopped {
			svc.StopWorker()
		}
	}()

	deadline := time.After(2 * time.Second)
	for worker.backfillCallCount() == 0 {
		select {
		case <-deadline:
			t.Fatal("RecordKey backfill worker did not use worker repository for startup guard")
		default:
			time.Sleep(time.Millisecond)
		}
	}

	svc.StopWorker()
	stopped = true
}

func TestPullWorkerRefreshesProcessingClaimWhileHandlerRuns(t *testing.T) {
	t.Setenv(outboxPullWorkerEnv, "true")
	t.Setenv(outboxParallelismEnv, "1")
	oldHeartbeatInterval := outboxProcessingHeartbeatInterval
	outboxProcessingHeartbeatInterval = 5 * time.Millisecond
	t.Cleanup(func() { outboxProcessingHeartbeatInterval = oldHeartbeatInterval })

	payload := marshalOutboxTestEnvelope(t, "Lead", "blocked-record")
	repo := &heartbeatRecordingOutboxRepository{
		isolationOutboxRepository: newIsolationOutboxRepository([]models.OutboxEvent{
			{ID: "evt-blocked", EventType: string(sharedevents.RecordUpdated), Payload: payload},
		}),
		heartbeatSeen: make(chan struct{}),
	}
	bus := newBlockingEventBus("blocked-record")
	svc := NewOutboxService(repo, bus, noopTxManager{})

	stopped := false
	svc.StartWorker(5 * time.Millisecond)
	defer func() {
		bus.Unblock()
		if !stopped {
			svc.StopWorker()
		}
	}()

	select {
	case <-bus.blockedSeen:
	case <-time.After(2 * time.Second):
		t.Fatal("blocked event was not claimed by a pull worker")
	}
	select {
	case <-repo.heartbeatSeen:
	case <-time.After(2 * time.Second):
		t.Fatal("pull worker did not refresh the visible processing claim while handler was blocked")
	}

	bus.Unblock()
	svc.StopWorker()
	stopped = true
}

type cancelBlockingHeartbeatRepository struct {
	outboxRepositoryStub
	started chan struct{}
	once    sync.Once
}

func (r *cancelBlockingHeartbeatRepository) RefreshOutboxProcessingClaim(ctx context.Context, _ models.OutboxClaim) error {
	r.once.Do(func() { close(r.started) })
	<-ctx.Done()
	return ctx.Err()
}

func TestHeartbeatStopCancelsBlockedRepositoryCall(t *testing.T) {
	oldHeartbeatInterval := outboxProcessingHeartbeatInterval
	outboxProcessingHeartbeatInterval = time.Millisecond
	t.Cleanup(func() { outboxProcessingHeartbeatInterval = oldHeartbeatInterval })
	repo := &cancelBlockingHeartbeatRepository{started: make(chan struct{})}
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, noopTxManager{}).
		WithDurableAttemptFence(func(context.Context) error { return nil })
	stop := svc.startOutboxProcessingHeartbeat(context.Background(), models.OutboxClaim{
		EventID: "event-1", RecordKey: "Issue/issue-1",
	})
	select {
	case <-repo.started:
	case <-time.After(time.Second):
		t.Fatal("heartbeat repository call did not start")
	}
	done := make(chan struct{})
	go func() {
		stop()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(time.Second):
		t.Fatal("heartbeat stop did not cancel blocked repository call")
	}
}

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

	payloadJSON := marshalOutboxTestEnvelope(t, "Lead", "00Q000000000001AAA")
	repo := &recordingOutboxRepository{}
	tx := &markingTxManager{}
	svc := NewOutboxService(repo, panicEventBus{}, tx)

	if err := svc.processEventAtomic(context.Background(), "evt-panic-001", string(sharedevents.RecordCreated), payloadJSON); err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}

	if repo.incrementRetryID != "evt-panic-001" {
		t.Fatalf("IncrementRetry id = %q, want evt-panic-001", repo.incrementRetryID)
	}
	if repo.incrementRetryCount != 1 {
		t.Fatalf("IncrementRetry count = %d, want 1", repo.incrementRetryCount)
	}
	if !strings.Contains(repo.incrementRetryMessage, "panic while processing event") {
		t.Fatalf("IncrementRetry message = %q, want panic detail", repo.incrementRetryMessage)
	}
	if !repo.incrementRetrySawPublishTx {
		t.Fatal("IncrementRetry did not run inside the claim/publish transaction")
	}
	if !repo.savepointCreated || !repo.savepointRolledBack || !repo.savepointReleased {
		t.Fatalf("savepoint create/rollback/release = %v/%v/%v, want true/true/true", repo.savepointCreated, repo.savepointRolledBack, repo.savepointReleased)
	}
	if tx.lastPessimisticErr != nil {
		t.Fatalf("pessimistic transaction error = %v, want nil retry bookkeeping commit", tx.lastPessimisticErr)
	}
	if repo.updateStatus != "" {
		t.Fatalf("UpdateStatus = %q, want no terminal status on first panic", repo.updateStatus)
	}
}

func TestTerminalOutboxFailurePersistsExactAttemptCount(t *testing.T) {
	repo := &recordingOutboxRepository{}
	svc := NewOutboxService(repo, nil, nil).WithMaxRetryAttempts(1)

	if err := svc.handleClaimedEventFailure(context.Background(), models.OutboxClaim{EventID: "evt-terminal"}, errOutboxPublishForTest); err != nil {
		t.Fatalf("handleClaimedEventFailure: %v", err)
	}
	if repo.terminalFailureID != "evt-terminal" || repo.terminalFailureCount != 1 {
		t.Fatalf("terminal failure = id=%q attempts=%d, want evt-terminal/1", repo.terminalFailureID, repo.terminalFailureCount)
	}
	if !strings.Contains(repo.terminalFailureMessage, "max retries exceeded") {
		t.Fatalf("terminal failure message = %q, want retry exhaustion detail", repo.terminalFailureMessage)
	}
}

func TestTerminalOutboxFailureUsesDurableClaimCountNotDiscoveryCount(t *testing.T) {
	repo := &recordingOutboxRepository{claimRetryCount: 2}
	svc := NewOutboxService(repo, errorEventBus{err: errOutboxPublishForTest}, &markingTxManager{}).
		WithMaxRetryAttempts(3)
	payloadJSON := marshalOutboxTestEnvelope(t, "Lead", "00Q000000000003AAA")

	// Discovery says zero, but ClaimEvent returns the authoritative durable
	// count of two. This attempt must therefore terminalize as attempt three.
	if err := svc.processEventAtomic(context.Background(), "evt-durable-count", string(sharedevents.RecordUpdated), payloadJSON); err != nil {
		t.Fatalf("processEventAtomic: %v", err)
	}
	if repo.terminalFailureCount != 3 {
		t.Fatalf("terminal failure count = %d, want durable claim count + 1 = 3", repo.terminalFailureCount)
	}
	if repo.incrementRetryID != "" {
		t.Fatalf("stale discovery count incorrectly selected retry path for %q", repo.incrementRetryID)
	}
}

func TestInvalidOutboxPayloadPersistsRepairableTerminalReceipt(t *testing.T) {
	repo := &recordingOutboxRepository{}
	svc := NewOutboxService(repo, nil, &markingTxManager{}).WithMaxRetryAttempts(3)

	if err := svc.processEventAtomic(context.Background(), "evt-invalid-payload", string(sharedevents.RecordUpdated), "{"); err != nil {
		t.Fatalf("processEventAtomic: %v", err)
	}
	if repo.terminalFailureID != "evt-invalid-payload" || repo.terminalFailureCount != 1 {
		t.Fatalf("terminal failure = id=%q attempts=%d, want evt-invalid-payload/1", repo.terminalFailureID, repo.terminalFailureCount)
	}
	if !strings.Contains(repo.terminalFailureMessage, "invalid payload") {
		t.Fatalf("terminal failure message = %q, want invalid payload", repo.terminalFailureMessage)
	}
}

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

	payloadJSON := marshalOutboxTestEnvelope(t, "Lead", "00Q000000000002AAA")
	repo := newRollbackRaceOutboxRepository()
	tx := &markingTxManager{}
	svc := NewOutboxService(repo, errorEventBus{err: errOutboxPublishForTest}, tx)

	if err := svc.processEventAtomic(context.Background(), "evt-race-001", string(sharedevents.RecordUpdated), payloadJSON); err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}

	if repo.contenderClaimed {
		t.Fatal("a second worker claimed the row between savepoint rollback and retry bookkeeping")
	}
	if got := repo.statusOf("evt-race-001"); got != OutboxStatusPending {
		t.Fatalf("status after retry bookkeeping = %q, want %s", got, OutboxStatusPending)
	}
	if got := repo.retryCountOf("evt-race-001"); got != 1 {
		t.Fatalf("retry count = %d, want 1", got)
	}
	if tx.lastPessimisticErr != nil {
		t.Fatalf("pessimistic transaction error = %v, want nil retry bookkeeping commit", tx.lastPessimisticErr)
	}
}

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

	payloadJSON := marshalOutboxTestEnvelope(t, "Contact", "003000000000001AAA")
	repo := &claimContextRecordingRepository{
		recordingOutboxRepository: recordingOutboxRepository{
			refreshErr: persistence.ErrOutboxEventAlreadyClaimed,
		},
	}
	bus := &publishContextRecordingBus{}
	tx := &markingTxManager{}
	svc := NewOutboxService(repo, bus, tx)

	if err := svc.processEventAtomic(context.Background(), "evt-process-001", string(sharedevents.RecordUpdated), payloadJSON); err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}

	if !repo.claimSawPublishTx {
		t.Fatal("ClaimEvent did not run inside the outbox transaction")
	}
	if repo.refreshCalls != 0 {
		t.Fatalf("RefreshOutboxProcessingClaim calls = %d, want 0 for a claim created in the same transaction", repo.refreshCalls)
	}
	if !bus.publishSawTx {
		t.Fatal("EventBus.Publish did not receive the outbox transaction context")
	}
	if !repo.updateStatusSawPublishTx {
		t.Fatal("UpdateStatus did not run inside the outbox transaction context")
	}
	if !repo.savepointCreated || !repo.savepointReleased || repo.savepointRolledBack {
		t.Fatalf("savepoint create/release/rollback = %v/%v/%v, want true/true/false", repo.savepointCreated, repo.savepointReleased, repo.savepointRolledBack)
	}
	if tx.withTxCalls != 0 || tx.withPessimisticRetryCalls != 1 || tx.withPessimisticCalls != 0 {
		t.Fatalf("tx manager calls: WithTransaction=%d WithPessimisticRetry=%d WithPessimisticTransaction=%d, want 0/1/0", tx.withTxCalls, tx.withPessimisticRetryCalls, tx.withPessimisticCalls)
	}
	if repo.updateStatusID != "evt-process-001" || repo.updateStatus != OutboxStatusProcessed {
		t.Fatalf("UpdateStatus id/status = %q/%q, want evt-process-001/%s", repo.updateStatusID, repo.updateStatus, OutboxStatusProcessed)
	}
}

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

	payloadJSON := marshalOutboxTestEnvelope(t, "Contact", "003000000000003AAA")
	repo := &recordingOutboxRepository{}
	bus := &publishContextRecordingBus{}
	tx := &markingTxManager{}
	svc := NewOutboxService(repo, bus, tx)

	if err := svc.processClaimedEventAtomic(context.Background(), models.OutboxClaim{EventID: "evt-claimed-001"}, string(sharedevents.RecordUpdated), payloadJSON); err != nil {
		t.Fatalf("processClaimedEventAtomic() error = %v", err)
	}

	if repo.claimedID != "" {
		t.Fatalf("ClaimEvent id = %q, want no claim for already-claimed path", repo.claimedID)
	}
	if repo.refreshCalls != 1 {
		t.Fatalf("RefreshOutboxProcessingClaim calls = %d, want 1 for a previously committed claim", repo.refreshCalls)
	}
	if !bus.publishSawTx {
		t.Fatal("EventBus.Publish did not receive the outbox transaction context")
	}
	if !repo.updateStatusSawPublishTx {
		t.Fatal("UpdateStatus did not run inside the outbox transaction context")
	}
	if !repo.savepointCreated || !repo.savepointReleased || repo.savepointRolledBack {
		t.Fatalf("savepoint create/release/rollback = %v/%v/%v, want true/true/false", repo.savepointCreated, repo.savepointReleased, repo.savepointRolledBack)
	}
	if tx.withTxCalls != 0 || tx.withPessimisticRetryCalls != 1 || tx.withPessimisticCalls != 0 {
		t.Fatalf("tx manager calls: WithTransaction=%d WithPessimisticRetry=%d WithPessimisticTransaction=%d, want 0/1/0", tx.withTxCalls, tx.withPessimisticRetryCalls, tx.withPessimisticCalls)
	}
	if repo.updateStatusID != "evt-claimed-001" || repo.updateStatus != OutboxStatusProcessed {
		t.Fatalf("UpdateStatus id/status = %q/%q, want evt-claimed-001/%s", repo.updateStatusID, repo.updateStatus, OutboxStatusProcessed)
	}
}

type historyPrelockReplaySignal struct{}

func (historyPrelockReplaySignal) Error() string { return "late history field discovered" }

func (historyPrelockReplaySignal) HistorySequencePrelockFields() []string {
	return []string{"field-a", "field-z"}
}

type historyPrelockReplayBus struct {
	publishContextRecordingBus
}

func (b *historyPrelockReplayBus) Publish(ctx context.Context, _ sharedevents.EventType, _ interface{}) error {
	b.publishCalls++
	b.publishSawTx = ctx.Value(outboxPublishTxMarkerKey{}) == true
	if b.publishCalls == 1 {
		return historyPrelockReplaySignal{}
	}
	return nil
}

type replayingHistoryTxManager struct {
	markingTxManager
	attempts   int
	maxRetries int
}

func (m *replayingHistoryTxManager) WithPessimisticRetry(ctx context.Context, fn func(context.Context) error, maxRetries int) error {
	m.withPessimisticRetryCalls++
	m.maxRetries = maxRetries
	for attempt := 0; attempt < 2; attempt++ {
		m.attempts++
		err := fn(context.WithValue(ctx, outboxPublishTxMarkerKey{}, true))
		if err == nil {
			return nil
		}
		var planner interface{ HistorySequencePrelockFields() []string }
		if !errors.As(err, &planner) {
			return err
		}
	}
	return errors.New("history prelock replay did not converge")
}

func TestProcessClaimedEventAtomicReplaysLateHistoryPrelockAtTransactionRoot(t *testing.T) {
	payloadJSON := marshalOutboxTestEnvelope(t, "CampaignMember", "00v000000000001AAA")
	repo := &recordingOutboxRepository{}
	bus := &historyPrelockReplayBus{}
	tx := &replayingHistoryTxManager{}
	svc := NewOutboxService(repo, bus, tx)

	err := svc.processClaimedEventAtomic(
		context.Background(),
		models.OutboxClaim{EventID: "evt-history-prelock"},
		string(sharedevents.RecordUpdated),
		payloadJSON,
	)
	if err != nil {
		t.Fatalf("processClaimedEventAtomic() error = %v", err)
	}
	if tx.withPessimisticRetryCalls != 1 || tx.attempts != 2 || tx.maxRetries != outboxEffectTransactionRetries {
		t.Fatalf("retry calls/attempts/max = %d/%d/%d, want 1/2/%d", tx.withPessimisticRetryCalls, tx.attempts, tx.maxRetries, outboxEffectTransactionRetries)
	}
	if bus.publishCalls != 2 || repo.refreshCalls != 2 {
		t.Fatalf("publish/claim-refresh calls = %d/%d, want 2/2", bus.publishCalls, repo.refreshCalls)
	}
	if repo.incrementRetryID != "" || repo.terminalFailureID != "" {
		t.Fatalf("history replay was converted into durable retry/poison state: retry=%q failed=%q", repo.incrementRetryID, repo.terminalFailureID)
	}
	if repo.updateStatusID != "evt-history-prelock" || repo.updateStatus != OutboxStatusProcessed {
		t.Fatalf("terminal status = %q/%q, want processed", repo.updateStatusID, repo.updateStatus)
	}
}

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

	payloadJSON := marshalOutboxTestEnvelope(t, "Contact", "003000000000002AAA")
	repo := &recordingOutboxRepository{updateStatusErr: errOutboxStatusWriteForTest}
	bus := &publishContextRecordingBus{}
	tx := &markingTxManager{}
	svc := NewOutboxService(repo, bus, tx)

	err := svc.processEventAtomic(context.Background(), "evt-status-fail-001", string(sharedevents.RecordUpdated), payloadJSON)
	if err == nil {
		t.Fatal("processEventAtomic() error = nil, want status write failure")
	}
	if !strings.Contains(err.Error(), "failed to mark as processed") {
		t.Fatalf("processEventAtomic() error = %v, want processed status failure", err)
	}
	if !bus.publishSawTx {
		t.Fatal("EventBus.Publish did not run inside the outbox transaction")
	}
	if !repo.updateStatusSawPublishTx {
		t.Fatal("UpdateStatus did not run inside the outbox transaction")
	}
	if tx.lastPessimisticErr == nil {
		t.Fatal("pessimistic transaction error = nil, want status failure to roll back published side effects")
	}
}

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

	payloadJSON := marshalOutboxTestEnvelope(t, "Campaign", "701000000000001AAA")
	repo := newLiveAttemptOutboxRepository("evt-live-001")
	bus := newBlockingEventBus("701000000000001AAA")
	svc := NewOutboxService(repo, bus, &markingTxManager{})

	done := make(chan error, 1)
	go func() {
		done <- svc.processEventAtomic(context.Background(), "evt-live-001", string(sharedevents.RecordUpdated), payloadJSON)
	}()

	select {
	case <-bus.blockedSeen:
	case <-time.After(2 * time.Second):
		t.Fatal("first live attempt did not reach Publish")
	}

	if err := svc.processEventAtomic(context.Background(), "evt-live-001", string(sharedevents.RecordUpdated), payloadJSON); err != nil {
		t.Fatalf("second processEventAtomic() error = %v", err)
	}
	if got := bus.PublishCount(); got != 1 {
		t.Fatalf("publish count while first attempt is live = %d, want 1", got)
	}

	bus.Unblock()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("first processEventAtomic() error = %v", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("first live attempt did not finish after unblock")
	}
	if got := repo.statusOf("evt-live-001"); got != OutboxStatusProcessed {
		t.Fatalf("status after first attempt = %q, want %s", got, OutboxStatusProcessed)
	}
}

func readEventsSourceForOutboxStructure(t *testing.T, filename string) string {
	t.Helper()

	_, here, _, ok := runtime.Caller(0)
	if !ok {
		t.Fatal("runtime.Caller failed")
	}
	raw, err := os.ReadFile(filepath.Join(filepath.Dir(here), filename))
	if err != nil {
		t.Fatalf("read %s: %v", filename, err)
	}
	return string(raw)
}

type staticOutboxMetadataLookup struct {
	hasActiveRecordTriggeredFlow bool
}

func (s staticOutboxMetadataLookup) HasActiveRecordTriggeredFlow(ctx context.Context, objectAPIName string) bool {
	return s.hasActiveRecordTriggeredFlow
}

type outboxRepositoryStub struct {
	enqueueCalls int
}

func (s *outboxRepositoryStub) Enqueue(ctx context.Context, eventType string, payload interface{}) (string, error) {
	s.enqueueCalls++
	return "evt-001", nil
}

func (s *outboxRepositoryStub) GetPendingEvents(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	return nil, nil
}

func (s *outboxRepositoryStub) GetPendingEventsWithRecordFIFO(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	return s.GetPendingEvents(ctx, limit)
}

func (s *outboxRepositoryStub) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	return models.OutboxClaim{}, nil
}

func (s *outboxRepositoryStub) CreateOutboxAttemptSavepoint(ctx context.Context) error {
	return nil
}

func (s *outboxRepositoryStub) RollbackToOutboxAttemptSavepoint(ctx context.Context) error {
	return nil
}

func (s *outboxRepositoryStub) ReleaseOutboxAttemptSavepoint(ctx context.Context) error {
	return nil
}

func (s *outboxRepositoryStub) RefreshOutboxProcessingClaim(ctx context.Context, claim models.OutboxClaim) error {
	return nil
}

func (s *outboxRepositoryStub) UpdateStatus(ctx context.Context, claim models.OutboxClaim, status string, errMessage string) error {
	return nil
}

func (s *outboxRepositoryStub) MarkFailed(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	return s.UpdateStatus(ctx, claim, OutboxStatusFailed, errMessage)
}

func (s *outboxRepositoryStub) IncrementRetry(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	return nil
}

func (s *outboxRepositoryStub) CleanupProcessed(ctx context.Context, cutoff time.Time) (int64, error) {
	return 0, nil
}

type atomicNoWorkOutboxRepository struct {
	outboxRepositoryStub
	atomicClaimCalls int
	pendingCalls     int
}

type atomicBatchOutboxRepository struct {
	outboxRepositoryStub
	events      []models.OutboxEvent
	claimCalls  int
	claimLimits []int
	claimInTx   bool
}

func (r *atomicBatchOutboxRepository) ClaimEventBatchWithRecordFIFO(ctx context.Context, limit int) ([]models.OutboxEvent, []models.OutboxClaim, error) {
	r.claimCalls++
	r.claimLimits = append(r.claimLimits, limit)
	r.claimInTx = r.claimInTx || ctx.Value(outboxPublishTxMarkerKey{}) == true
	if limit > len(r.events) {
		limit = len(r.events)
	}
	events := append([]models.OutboxEvent(nil), r.events[:limit]...)
	r.events = r.events[limit:]
	claims := make([]models.OutboxClaim, 0, len(events))
	for _, event := range events {
		claims = append(claims, models.OutboxClaim{EventID: event.ID, RecordKey: event.RecordKey, Token: "claim-" + event.ID})
	}
	return events, claims, nil
}

func (r *atomicNoWorkOutboxRepository) GetPendingEventsWithRecordFIFO(context.Context, int) ([]models.OutboxEvent, error) {
	r.pendingCalls++
	return nil, errOutboxRecordFIFOSelectionForTest
}

func (r *atomicNoWorkOutboxRepository) ClaimEventBatchWithRecordFIFO(context.Context, int) ([]models.OutboxEvent, []models.OutboxClaim, error) {
	r.atomicClaimCalls++
	return nil, nil, nil
}

func TestProcessNextUsesAtomicFIFOClaimInsideDurableAdmission(t *testing.T) {
	repo := &atomicNoWorkOutboxRepository{}
	tx := &markingTxManager{}
	service := NewOutboxService(repo, nil, tx).
		WithDurableAdmissionFence(func(ctx context.Context) error {
			if ctx.Value(outboxPublishTxMarkerKey{}) != true {
				t.Fatal("durable admission fence ran outside transaction")
			}
			return nil
		})

	processed, err := service.ProcessNext(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	if processed || repo.atomicClaimCalls != 1 || repo.pendingCalls != 0 || tx.withPessimisticCalls != 1 {
		t.Fatalf("processed=%t atomic=%d pending=%d pessimisticTx=%d, want false/1/0/1", processed, repo.atomicClaimCalls, repo.pendingCalls, tx.withPessimisticCalls)
	}
}

func TestDurableRecordEventBatchUsesOneFenceAndRespectsCapacity(t *testing.T) {
	repo := &atomicBatchOutboxRepository{events: []models.OutboxEvent{
		{ID: "event-1", RecordKey: "Issue/one"},
		{ID: "event-2", RecordKey: "Issue/two"},
		{ID: "event-3", RecordKey: "Issue/three"},
	}}
	tx := &markingTxManager{}
	fenceCalls := 0
	service := NewOutboxService(repo, nil, tx).
		WithDurableAdmissionFence(func(ctx context.Context) error {
			fenceCalls++
			if ctx.Value(outboxPublishTxMarkerKey{}) != true {
				t.Fatal("durable batch admission fence ran outside transaction")
			}
			return nil
		})

	first, err := service.ClaimDurableRecordEventBatch(t.Context(), 2)
	if err != nil {
		t.Fatal(err)
	}
	if len(first) != 2 || first[0].event.ID != "event-1" || first[1].event.ID != "event-2" {
		t.Fatalf("first claimed batch = %+v, want first two FIFO heads", first)
	}
	if fenceCalls != 1 || tx.withPessimisticCalls != 1 || repo.claimCalls != 1 || !repo.claimInTx {
		t.Fatalf("first batch fence/tx/claims/in-tx = %d/%d/%d/%t, want 1/1/1/true", fenceCalls, tx.withPessimisticCalls, repo.claimCalls, repo.claimInTx)
	}

	second, err := service.ClaimDurableRecordEventBatch(t.Context(), 2)
	if err != nil {
		t.Fatal(err)
	}
	if len(second) != 1 || second[0].event.ID != "event-3" {
		t.Fatalf("second claimed batch = %+v, want final FIFO head", second)
	}
	if fenceCalls != 2 || tx.withPessimisticCalls != 2 || repo.claimCalls != 2 || !reflect.DeepEqual(repo.claimLimits, []int{2, 2}) {
		t.Fatalf("all batch fence/tx/claims/limits = %d/%d/%d/%v, want 2/2/2/[2 2]", fenceCalls, tx.withPessimisticCalls, repo.claimCalls, repo.claimLimits)
	}
}

func TestDurableRecordEventBatchDoesNotStartEffectBeforeExplicitDispatch(t *testing.T) {
	payload, err := json.Marshal(RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "one"}})
	if err != nil {
		t.Fatal(err)
	}
	repo := &atomicBatchOutboxRepository{events: []models.OutboxEvent{{
		ID: "event-1", EventType: string(sharedevents.RecordUpdated), Payload: string(payload), RecordKey: "Issue/one",
	}}}
	bus := &contextCapturingEventBus{}
	service := NewOutboxService(repo, bus, &markingTxManager{}).
		WithDurableAdmissionFence(func(context.Context) error { return nil }).
		WithDurableAttemptFence(func(context.Context) error { return nil })

	claimed, err := service.ClaimDurableRecordEventBatch(t.Context(), 1)
	if err != nil || len(claimed) != 1 {
		t.Fatalf("ClaimDurableRecordEventBatch() = %+v, %v", claimed, err)
	}
	if bus.capturedCtx != nil {
		t.Fatal("record-event effect started before the durable batch claim returned")
	}
	if err := service.ProcessDurableRecordEventClaim(t.Context(), claimed[0]); err != nil {
		t.Fatal(err)
	}
	if bus.capturedCtx == nil {
		t.Fatal("explicit durable record-event dispatch did not publish the effect")
	}
}

type staleClaimOutboxRepository struct {
	outboxRepositoryStub
	refreshCalls int
}

func (r *staleClaimOutboxRepository) RefreshOutboxProcessingClaim(context.Context, models.OutboxClaim) error {
	r.refreshCalls++
	return persistence.ErrOutboxEventAlreadyClaimed
}

func TestNonLiveOutboxClaimIsRejectedBeforeSubscriberEffects(t *testing.T) {
	repo := &staleClaimOutboxRepository{}
	bus := &publishContextRecordingBus{}
	service := NewOutboxService(repo, bus, nil)
	payload, err := json.Marshal(RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"}})
	if err != nil {
		t.Fatal(err)
	}
	err = service.processClaimedEventAtomic(t.Context(), models.OutboxClaim{
		EventID: "event-1", Token: "00000000-0000-4000-8000-000000000001",
	}, string(sharedevents.RecordUpdated), string(payload))
	if err != nil {
		t.Fatalf("stale claim should be a benign lost race: %v", err)
	}
	if repo.refreshCalls != 1 || bus.publishCalls != 0 {
		t.Fatalf("refresh calls=%d publish calls=%d, want 1/0", repo.refreshCalls, bus.publishCalls)
	}
}

type bookkeepingClaimRevalidationRepository struct {
	outboxRepositoryStub
	refreshCalls int
	updateCalls  int
	retryCalls   int
}

func (r *bookkeepingClaimRevalidationRepository) RefreshOutboxProcessingClaim(context.Context, models.OutboxClaim) error {
	r.refreshCalls++
	if r.refreshCalls == 1 {
		return nil
	}
	return persistence.ErrOutboxEventAlreadyClaimed
}

func (r *bookkeepingClaimRevalidationRepository) UpdateStatus(context.Context, models.OutboxClaim, string, string) error {
	r.updateCalls++
	return errors.New("acknowledgement failed")
}

func (r *bookkeepingClaimRevalidationRepository) IncrementRetry(context.Context, models.OutboxClaim, string) error {
	r.retryCalls++
	return nil
}

func TestFailureBookkeepingCannotUseClaimThatExpiredAfterEffectTransaction(t *testing.T) {
	repo := &bookkeepingClaimRevalidationRepository{}
	bus := &publishContextRecordingBus{}
	txManager := &countingTxManager{}
	service := NewOutboxService(repo, bus, txManager).
		WithDurableAttemptFence(func(context.Context) error { return nil })
	payload, err := json.Marshal(RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"}})
	if err != nil {
		t.Fatal(err)
	}
	err = service.processClaimedEventAtomic(t.Context(), models.OutboxClaim{
		EventID: "event-1", Token: "00000000-0000-4000-8000-000000000001",
	}, string(sharedevents.RecordUpdated), string(payload))
	if err != nil {
		t.Fatalf("lost bookkeeping claim should be a benign race: %v", err)
	}
	if repo.refreshCalls != 2 || repo.updateCalls != 1 || repo.retryCalls != 0 {
		t.Fatalf("refresh=%d update=%d retry=%d, want 2/1/0", repo.refreshCalls, repo.updateCalls, repo.retryCalls)
	}
	if bus.publishCalls != 1 || txManager.withPessimisticRetryCalls != 1 || txManager.withPessimisticCalls != 1 {
		t.Fatalf("publish=%d retryTransactions=%d plainTransactions=%d, want 1/1/1", bus.publishCalls, txManager.withPessimisticRetryCalls, txManager.withPessimisticCalls)
	}
}

type isolationOutboxRepository struct {
	outboxRepositoryStub
	mu          sync.Mutex
	events      []models.OutboxEvent
	status      map[string]string
	processedCh map[string]chan struct{}
}

func newIsolationOutboxRepository(events []models.OutboxEvent) *isolationOutboxRepository {
	r := &isolationOutboxRepository{
		events:      events,
		status:      make(map[string]string, len(events)),
		processedCh: make(map[string]chan struct{}, len(events)),
	}
	for _, ev := range events {
		r.status[ev.ID] = OutboxStatusPending
		r.processedCh[ev.ID] = make(chan struct{})
	}
	return r
}

func (r *isolationOutboxRepository) GetPendingEvents(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	for _, ev := range r.events {
		if r.status[ev.ID] == OutboxStatusPending {
			return []models.OutboxEvent{ev}, nil
		}
	}
	return nil, nil
}

func (r *isolationOutboxRepository) GetPendingEventsWithRecordFIFO(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	return r.GetPendingEvents(ctx, limit)
}

func (r *isolationOutboxRepository) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.status[id] != OutboxStatusPending {
		return models.OutboxClaim{}, persistence.ErrOutboxEventAlreadyClaimed
	}
	r.status[id] = "processing"
	return models.OutboxClaim{EventID: id}, nil
}

func (r *isolationOutboxRepository) UpdateStatus(ctx context.Context, claim models.OutboxClaim, status string, errMessage string) error {
	id := claim.EventID
	r.mu.Lock()
	defer r.mu.Unlock()
	r.status[id] = status
	if status == OutboxStatusProcessed {
		close(r.processedCh[id])
	}
	return nil
}

func (r *isolationOutboxRepository) MarkFailed(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	return r.UpdateStatus(ctx, claim, OutboxStatusFailed, errMessage)
}

func (r *isolationOutboxRepository) IncrementRetry(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	id := claim.EventID
	r.mu.Lock()
	defer r.mu.Unlock()
	r.status[id] = OutboxStatusPending
	return nil
}

func (r *isolationOutboxRepository) processed(id string) <-chan struct{} {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.processedCh[id]
}

func (r *isolationOutboxRepository) statusOf(id string) string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.status[id]
}

type failingBackfillIsolationOutboxRepository struct {
	*isolationOutboxRepository

	backfillMu    sync.Mutex
	backfillCalls int
}

func (r *failingBackfillIsolationOutboxRepository) BackfillRecordKeys(ctx context.Context, limit int) (int, error) {
	r.backfillMu.Lock()
	defer r.backfillMu.Unlock()
	r.backfillCalls++
	if r.backfillCalls == 1 {
		return 0, errOutboxRecordKeyBackfillForTest
	}
	return 0, nil
}

type heartbeatRecordingOutboxRepository struct {
	*isolationOutboxRepository

	heartbeatOnce sync.Once
	heartbeatSeen chan struct{}
}

func (r *heartbeatRecordingOutboxRepository) RefreshOutboxProcessingClaim(ctx context.Context, claim models.OutboxClaim) error {
	id := claim.EventID
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.status[id] != "processing" {
		return persistence.ErrOutboxEventAlreadyClaimed
	}
	r.heartbeatOnce.Do(func() { close(r.heartbeatSeen) })
	return nil
}

type batchModeOutboxRepository struct {
	*isolationOutboxRepository

	pathMu                 sync.Mutex
	fifoCalls              int
	backfillCalls          int
	recordKeyColumnPresent bool
}

func (r *batchModeOutboxRepository) GetPendingEvents(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	r.mu.Lock()
	defer r.mu.Unlock()

	var pending []models.OutboxEvent
	for _, ev := range r.events {
		if r.status[ev.ID] == OutboxStatusPending {
			pending = append(pending, ev)
			if len(pending) >= limit {
				break
			}
		}
	}
	return pending, nil
}

func (r *batchModeOutboxRepository) GetPendingEventsWithRecordFIFO(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	r.pathMu.Lock()
	r.fifoCalls++
	r.pathMu.Unlock()
	return nil, errOutboxRecordFIFOSelectionForTest
}

func (r *batchModeOutboxRepository) BackfillRecordKeys(ctx context.Context, limit int) (int, error) {
	r.pathMu.Lock()
	r.backfillCalls++
	r.pathMu.Unlock()
	return 0, errOutboxRecordKeyBackfillForTest
}

func (r *batchModeOutboxRepository) HasRecordKeyColumn(ctx context.Context) (bool, error) {
	return r.recordKeyColumnPresent, nil
}

func (r *batchModeOutboxRepository) fifoCallCount() int {
	r.pathMu.Lock()
	defer r.pathMu.Unlock()
	return r.fifoCalls
}

func (r *batchModeOutboxRepository) backfillCallCount() int {
	r.pathMu.Lock()
	defer r.pathMu.Unlock()
	return r.backfillCalls
}

type blockingEventBus struct {
	blockRecordID string
	blockedSeen   chan struct{}
	unblock       chan struct{}
	once          sync.Once
	unblockOnce   sync.Once
	mu            sync.Mutex
	publishCount  int
}

func newBlockingEventBus(blockRecordID string) *blockingEventBus {
	return &blockingEventBus{
		blockRecordID: blockRecordID,
		blockedSeen:   make(chan struct{}),
		unblock:       make(chan struct{}),
	}
}

func (b *blockingEventBus) Subscribe(_ sharedevents.EventType, _ EventHandler) func() {
	return func() {}
}

func (b *blockingEventBus) Publish(ctx context.Context, _ sharedevents.EventType, payload interface{}) error {
	b.mu.Lock()
	b.publishCount++
	b.mu.Unlock()
	if p, ok := payload.(RecordEventPayload); ok && p.Record.GetString("Id") == b.blockRecordID {
		b.once.Do(func() { close(b.blockedSeen) })
		<-b.unblock
	}
	return nil
}

func (b *blockingEventBus) PublishAsyncWithContext(context.Context, sharedevents.EventType, interface{}) {
}

func (b *blockingEventBus) Unblock() {
	b.unblockOnce.Do(func() { close(b.unblock) })
}

func (b *blockingEventBus) PublishCount() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.publishCount
}

type rollbackRaceOutboxRepository struct {
	outboxRepositoryStub

	mu               sync.Mutex
	status           map[string]string
	retryCount       map[string]int
	lockHeld         bool
	contenderClaimed bool
}

func newRollbackRaceOutboxRepository() *rollbackRaceOutboxRepository {
	return &rollbackRaceOutboxRepository{
		status:     map[string]string{"evt-race-001": OutboxStatusPending},
		retryCount: map[string]int{"evt-race-001": 0},
	}
}

func (r *rollbackRaceOutboxRepository) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.lockHeld {
		return models.OutboxClaim{}, persistence.ErrOutboxEventAlreadyClaimed
	}
	if r.status[id] != OutboxStatusPending {
		return models.OutboxClaim{}, persistence.ErrOutboxEventAlreadyClaimed
	}
	if ctx.Value(outboxPublishTxMarkerKey{}) == true {
		r.status[id] = "processing"
		r.lockHeld = true
		return models.OutboxClaim{EventID: id}, nil
	}
	r.status[id] = "processing"
	return models.OutboxClaim{EventID: id}, nil
}

func (r *rollbackRaceOutboxRepository) CreateOutboxAttemptSavepoint(ctx context.Context) error {
	return nil
}

func (r *rollbackRaceOutboxRepository) RollbackToOutboxAttemptSavepoint(ctx context.Context) error {
	claimed, claimErr := r.ClaimEvent(context.Background(), "evt-race-001")
	r.mu.Lock()
	r.contenderClaimed = claimErr == nil && claimed.EventID != ""
	r.mu.Unlock()
	return nil
}

func (r *rollbackRaceOutboxRepository) ReleaseOutboxAttemptSavepoint(ctx context.Context) error {
	return nil
}

func (r *rollbackRaceOutboxRepository) UpdateStatus(ctx context.Context, claim models.OutboxClaim, status string, errMessage string) error {
	id := claim.EventID
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.status[id] != "processing" {
		return persistence.ErrOutboxEventAlreadyClaimed
	}
	r.status[id] = status
	r.lockHeld = false
	return nil
}

func (r *rollbackRaceOutboxRepository) MarkFailed(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	return r.UpdateStatus(ctx, claim, OutboxStatusFailed, errMessage)
}

func (r *rollbackRaceOutboxRepository) IncrementRetry(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	id := claim.EventID
	r.mu.Lock()
	defer r.mu.Unlock()

	if r.status[id] != "processing" {
		return persistence.ErrOutboxEventAlreadyClaimed
	}
	r.retryCount[id] = claim.RetryCount + 1
	r.status[id] = OutboxStatusPending
	r.lockHeld = false
	return nil
}

func (r *rollbackRaceOutboxRepository) statusOf(id string) string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.status[id]
}

func (r *rollbackRaceOutboxRepository) retryCountOf(id string) int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.retryCount[id]
}

type liveAttemptOutboxRepository struct {
	outboxRepositoryStub

	mu       sync.Mutex
	id       string
	status   string
	lockHeld bool
}

func newLiveAttemptOutboxRepository(id string) *liveAttemptOutboxRepository {
	return &liveAttemptOutboxRepository{
		id:     id,
		status: OutboxStatusPending,
	}
}

func (r *liveAttemptOutboxRepository) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	r.mu.Lock()
	defer r.mu.Unlock()

	if id != r.id || r.status != OutboxStatusPending || r.lockHeld {
		return models.OutboxClaim{}, persistence.ErrOutboxEventAlreadyClaimed
	}
	r.status = "processing"
	r.lockHeld = true
	return models.OutboxClaim{EventID: id}, nil
}

func (r *liveAttemptOutboxRepository) UpdateStatus(ctx context.Context, claim models.OutboxClaim, status string, errMessage string) error {
	id := claim.EventID
	r.mu.Lock()
	defer r.mu.Unlock()

	if id != r.id || r.status != "processing" || !r.lockHeld {
		return persistence.ErrOutboxEventAlreadyClaimed
	}
	r.status = status
	r.lockHeld = false
	return nil
}

func (r *liveAttemptOutboxRepository) IncrementRetry(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	id := claim.EventID
	r.mu.Lock()
	defer r.mu.Unlock()

	if id != r.id || r.status != "processing" || !r.lockHeld {
		return persistence.ErrOutboxEventAlreadyClaimed
	}
	r.status = OutboxStatusPending
	r.lockHeld = false
	return nil
}

func (r *liveAttemptOutboxRepository) statusOf(id string) string {
	r.mu.Lock()
	defer r.mu.Unlock()
	if id != r.id {
		return ""
	}
	return r.status
}

type recordingOutboxRepository struct {
	outboxRepositoryStub

	claimedID                  string
	claimRetryCount            int
	refreshCalls               int
	refreshErr                 error
	updateStatusID             string
	updateStatus               string
	updateStatusMessage        string
	updateStatusSawPublishTx   bool
	updateStatusErr            error
	terminalFailureID          string
	terminalFailureCount       int
	terminalFailureMessage     string
	incrementRetryID           string
	incrementRetryCount        int
	incrementRetryMessage      string
	incrementRetrySawPublishTx bool
	savepointCreated           bool
	savepointRolledBack        bool
	savepointReleased          bool
}

type outboxPublishTxMarkerKey struct{}

var errOutboxStatusWriteForTest = errString("status write failed")
var errOutboxRecordKeyBackfillForTest = errString("record key backfill failed")
var errOutboxRecordFIFOSelectionForTest = errString("record fifo selection should not be called")
var errOutboxPublishForTest = errString("publish failed")

type errString string

func (e errString) Error() string { return string(e) }

type markingTxManager struct {
	noopTxManager
	withTxCalls               int
	withPessimisticCalls      int
	withPessimisticRetryCalls int
	lastPessimisticErr        error
}

func (m *markingTxManager) WithTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	m.withTxCalls++
	return fn(context.WithValue(ctx, outboxPublishTxMarkerKey{}, true))
}

func (m *markingTxManager) WithPessimisticTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	m.withPessimisticCalls++
	err := fn(context.WithValue(ctx, outboxPublishTxMarkerKey{}, true))
	m.lastPessimisticErr = err
	return err
}

func (m *markingTxManager) WithPessimisticRetry(ctx context.Context, fn func(ctx context.Context) error, _ int) error {
	m.withPessimisticRetryCalls++
	err := fn(context.WithValue(ctx, outboxPublishTxMarkerKey{}, true))
	m.lastPessimisticErr = err
	return err
}

type claimContextRecordingRepository struct {
	recordingOutboxRepository
	claimSawPublishTx bool
}

func (r *claimContextRecordingRepository) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	r.claimSawPublishTx = ctx.Value(outboxPublishTxMarkerKey{}) == true
	return r.recordingOutboxRepository.ClaimEvent(ctx, id)
}

type publishContextRecordingBus struct {
	publishSawTx bool
	publishCalls int
}

func (b *publishContextRecordingBus) Subscribe(_ sharedevents.EventType, _ EventHandler) func() {
	return func() {}
}

func (b *publishContextRecordingBus) Publish(ctx context.Context, _ sharedevents.EventType, _ interface{}) error {
	b.publishCalls++
	b.publishSawTx = ctx.Value(outboxPublishTxMarkerKey{}) == true
	return nil
}

func (b *publishContextRecordingBus) PublishAsyncWithContext(context.Context, sharedevents.EventType, interface{}) {
}

func (r *recordingOutboxRepository) ClaimEvent(_ context.Context, id string) (models.OutboxClaim, error) {
	r.claimedID = id
	return models.OutboxClaim{EventID: id, RetryCount: r.claimRetryCount}, nil
}

func (r *recordingOutboxRepository) RefreshOutboxProcessingClaim(context.Context, models.OutboxClaim) error {
	r.refreshCalls++
	return r.refreshErr
}

func (r *recordingOutboxRepository) CreateOutboxAttemptSavepoint(ctx context.Context) error {
	r.savepointCreated = true
	return nil
}

func (r *recordingOutboxRepository) RollbackToOutboxAttemptSavepoint(ctx context.Context) error {
	r.savepointRolledBack = true
	return nil
}

func (r *recordingOutboxRepository) ReleaseOutboxAttemptSavepoint(ctx context.Context) error {
	r.savepointReleased = true
	return nil
}

func (r *recordingOutboxRepository) UpdateStatus(ctx context.Context, claim models.OutboxClaim, status string, errMessage string) error {
	r.updateStatusID = claim.EventID
	r.updateStatus = status
	r.updateStatusMessage = errMessage
	r.updateStatusSawPublishTx = ctx.Value(outboxPublishTxMarkerKey{}) == true
	if r.updateStatusErr != nil {
		return r.updateStatusErr
	}
	return nil
}

func (r *recordingOutboxRepository) MarkFailed(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	r.terminalFailureID = claim.EventID
	r.terminalFailureCount = claim.RetryCount + 1
	r.terminalFailureMessage = errMessage
	r.updateStatusID = claim.EventID
	r.updateStatus = OutboxStatusFailed
	r.updateStatusMessage = errMessage
	r.updateStatusSawPublishTx = ctx.Value(outboxPublishTxMarkerKey{}) == true
	return r.updateStatusErr
}

func (r *recordingOutboxRepository) IncrementRetry(ctx context.Context, claim models.OutboxClaim, errMessage string) error {
	r.incrementRetryID = claim.EventID
	r.incrementRetryCount = claim.RetryCount + 1
	r.incrementRetryMessage = errMessage
	r.incrementRetrySawPublishTx = ctx.Value(outboxPublishTxMarkerKey{}) == true
	return nil
}

type panicEventBus struct{}

func (panicEventBus) Subscribe(_ sharedevents.EventType, _ EventHandler) func() {
	return func() {}
}

func (panicEventBus) Publish(context.Context, sharedevents.EventType, interface{}) error {
	panic("publish exploded")
}

func (panicEventBus) PublishAsyncWithContext(context.Context, sharedevents.EventType, interface{}) {}

type errorEventBus struct {
	err error
}

func (errorEventBus) Subscribe(_ sharedevents.EventType, _ EventHandler) func() {
	return func() {}
}

func (b errorEventBus) Publish(context.Context, sharedevents.EventType, interface{}) error {
	return b.err
}

func (errorEventBus) PublishAsyncWithContext(context.Context, sharedevents.EventType, interface{}) {}

func marshalOutboxTestEnvelope(t *testing.T, objectAPIName string, recordID string) string {
	t.Helper()
	payloadJSON, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload: RecordEventPayload{
			ObjectAPIName: objectAPIName,
			Record:        models.SObject{"Id": recordID},
		},
	})
	if err != nil {
		t.Fatalf("json.Marshal payload: %v", err)
	}
	return string(payloadJSON)
}