package events

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"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"
)

// ---------------------------------------------------------------------------
// Stubs for lineage tests
// ---------------------------------------------------------------------------

// payloadCapturingRepo extends outboxRepositoryStub to also capture the
// RecordEventPayload passed to Enqueue, so tests can inspect Metadata entries.
type payloadCapturingRepo struct {
	outboxRepositoryStub
	envelopes  []InternalRecordEventEnvelope
	enqueueErr error
}

func (r *payloadCapturingRepo) Enqueue(ctx context.Context, eventType string, payload interface{}) (string, error) {
	r.enqueueCalls++
	if p, ok := payload.(InternalRecordEventEnvelope); ok {
		r.envelopes = append(r.envelopes, p)
	}
	if r.enqueueErr != nil {
		return "", r.enqueueErr
	}
	return "evt-lineage-001", nil
}

func (r *payloadCapturingRepo) lastEnvelope(t *testing.T) InternalRecordEventEnvelope {
	t.Helper()
	if len(r.envelopes) == 0 {
		t.Fatal("expected at least one captured outbox envelope")
	}
	return r.envelopes[len(r.envelopes)-1]
}

// claimSuccessRepo extends outboxRepositoryStub to return a successful claim
// (non-empty id), letting processEventAtomic proceed through to Publish.
type claimSuccessRepo struct {
	outboxRepositoryStub
	updateStatuses []string
	retryCalls     int
	claimCalls     int
	claimInTx      bool
	refreshCalls   int
	refreshInTx    bool
	cleanupCalls   int
	cleanupInTx    bool
	cleanupDeleted int64
}

type orderedClaimRepo struct {
	claimSuccessRepo
	order []string
}

func (r *orderedClaimRepo) LockOutboxClaimCandidate(context.Context, string) error {
	r.order = append(r.order, "candidate")
	return nil
}

func (r *orderedClaimRepo) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	r.order = append(r.order, "claim")
	return r.claimSuccessRepo.ClaimEvent(ctx, id)
}

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

func (r *claimSuccessRepo) ClaimEvent(ctx context.Context, id string) (models.OutboxClaim, error) {
	r.claimCalls++
	r.claimInTx = inOutboxTestTx(ctx)
	return models.OutboxClaim{EventID: id}, nil
}

func (r *claimSuccessRepo) RefreshOutboxProcessingClaim(ctx context.Context, _ models.OutboxClaim) error {
	r.refreshCalls++
	r.refreshInTx = inOutboxTestTx(ctx)
	return nil
}

func (r *claimSuccessRepo) CleanupProcessedBatch(ctx context.Context, _ time.Duration, _ int) (int64, error) {
	r.cleanupCalls++
	r.cleanupInTx = inOutboxTestTx(ctx)
	return r.cleanupDeleted, nil
}

func (r *claimSuccessRepo) UpdateStatus(_ context.Context, _ models.OutboxClaim, status string, _ string) error {
	r.updateStatuses = append(r.updateStatuses, status)
	return nil
}

// contextCapturingEventBus is a minimal EventPublisher stub that records the
// context it received during Publish, so test 5 can assert typed-key restoration.
type contextCapturingEventBus struct {
	capturedCtx     context.Context
	capturedPayload interface{}
}

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

func (b *contextCapturingEventBus) Publish(ctx context.Context, _ sharedevents.EventType, payload interface{}) error {
	b.capturedCtx = ctx
	b.capturedPayload = payload
	return nil
}

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

// noopTxManager calls fn(ctx) directly without a real DB transaction.
type noopTxManager struct{}

func (m noopTxManager) WithTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	return fn(ctx)
}

func (m noopTxManager) WithPessimisticTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	return fn(ctx)
}

func (m noopTxManager) WithRetry(ctx context.Context, fn func(ctx context.Context) error, _ int) error {
	return fn(ctx)
}

func (m noopTxManager) WithPessimisticRetry(ctx context.Context, fn func(ctx context.Context) error, _ int) error {
	return fn(ctx)
}

func (m noopTxManager) IsTransactionActive(_ context.Context) bool { return false }

func (m noopTxManager) IsPessimisticTx(_ context.Context) bool { return false }

func (m noopTxManager) ExtractTx(_ context.Context) interface{} { return nil }

func (m noopTxManager) DetachTx(ctx context.Context) context.Context { return ctx }

type outboxTestTxContextKey struct{}

func inOutboxTestTx(ctx context.Context) bool {
	v, _ := ctx.Value(outboxTestTxContextKey{}).(bool)
	return v
}

type countingTxManager struct {
	noopTxManager
	withTxCalls               int
	withPessimisticCalls      int
	withPessimisticRetryCalls int
}

func (m *countingTxManager) WithTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	m.withTxCalls++
	return fn(ctx)
}

func (m *countingTxManager) WithPessimisticTransaction(ctx context.Context, fn func(ctx context.Context) error) error {
	m.withPessimisticCalls++
	return fn(context.WithValue(ctx, outboxTestTxContextKey{}, true))
}

func (m *countingTxManager) WithPessimisticRetry(ctx context.Context, fn func(ctx context.Context) error, _ int) error {
	m.withPessimisticRetryCalls++
	return fn(context.WithValue(ctx, outboxTestTxContextKey{}, true))
}

// ---------------------------------------------------------------------------
// Tests 3–6 (issue #633 test matrix)
// ---------------------------------------------------------------------------

// TestEnqueueEventStampsLineageOnFreshContext (test 3) — EnqueueEvent called
// with context.Background() (no chain lineage) mints a fresh UUID chain_id and
// stamps chain_depth="0" into the outbox payload Metadata.
// Parity test for Fact 4 (EnqueueEvent is the sole stamp site).
func TestEnqueueEventStampsLineageOnFreshContext(t *testing.T) {
	t.Parallel()

	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)

	err := svc.EnqueueEvent(context.Background(), sharedevents.RecordCreated, RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "001000000000001AAA"},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}

	internalMeta := repo.lastEnvelope(t).InternalMetadata
	chainID := internalMeta[string(constants.ContextKeyAfterSaveChainID)]
	chainDepth := internalMeta[string(constants.ContextKeyAfterSaveChainDepth)]

	if chainID == "" {
		t.Fatal("expected non-empty chain_id in Metadata on fresh context, got empty")
	}
	if chainDepth != "0" {
		t.Fatalf("expected chain_depth=\"0\" in Metadata on fresh context, got %q", chainDepth)
	}
}

func TestDurableExecutionBindingIsOpaqueInternalMetadata(t *testing.T) {
	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)
	ctx := WithDurableExecutionBinding(context.Background(), `{"schema":"binding/v1"}`)
	if err := svc.EnqueueEvent(ctx, sharedevents.RecordCreated, RecordEventPayload{
		ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"},
		Metadata: map[string]string{durableExecutionBindingMetadataKey: "public-spoof"},
	}); err != nil {
		t.Fatal(err)
	}
	envelope := repo.lastEnvelope(t)
	if got := envelope.InternalMetadata[durableExecutionBindingMetadataKey]; got != `{"schema":"binding/v1"}` {
		t.Fatalf("internal binding = %q", got)
	}
	if _, leaked := stripRegisteredInternalMetadata(envelope.PublicPayload.Metadata)[durableExecutionBindingMetadataKey]; leaked {
		t.Fatal("durable execution binding leaked into public event metadata")
	}
}

func TestDurableExecutionBinderInstallsContextBeforePublish(t *testing.T) {
	type boundKey struct{}
	repo := &claimSuccessRepo{}
	bus := &contextCapturingEventBus{}
	svc := NewOutboxService(repo, bus, noopTxManager{}).WithDurableExecutionBinder(
		func(_ context.Context, binding string) (DurableExecutionInstaller, error) {
			if binding != "exact-view" {
				t.Fatalf("binding = %q", binding)
			}
			return func(ctx context.Context) (context.Context, error) {
				return context.WithValue(ctx, boundKey{}, true), nil
			}, nil
		},
	)
	payload, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload:    RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"}},
		InternalMetadata: map[string]string{durableExecutionBindingMetadataKey: "exact-view"},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := svc.processEventAtomic(context.Background(), "event-1", string(sharedevents.RecordCreated), string(payload)); err != nil {
		t.Fatal(err)
	}
	if bound, _ := bus.capturedCtx.Value(boundKey{}).(bool); !bound {
		t.Fatal("event published without the verified durable execution binding")
	}

	missing := InternalRecordEventEnvelope{PublicPayload: RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-2"}}}
	missingPayload, _ := json.Marshal(missing)
	if _, err := svc.prepareClaimedEvent(context.Background(), "event-2", string(missingPayload)); err == nil {
		t.Fatal("metadata-driven worker accepted an unbound durable event")
	}
}

func TestDurableExecutionInstallerFailureFailsClosedBeforePublish(t *testing.T) {
	installErr := errors.New("install complete durable execution envelope")
	bus := &contextCapturingEventBus{}
	svc := NewOutboxService(nil, bus, nil).WithDurableExecutionBinder(
		func(context.Context, string) (DurableExecutionInstaller, error) {
			return func(context.Context) (context.Context, error) {
				return nil, installErr
			}, nil
		},
	)
	payload, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload:    RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"}},
		InternalMetadata: map[string]string{durableExecutionBindingMetadataKey: "exact-view"},
	})
	if err != nil {
		t.Fatal(err)
	}
	err = svc.publishClaimedEvent(context.Background(), "event-1", string(sharedevents.RecordCreated), string(payload))
	if !errors.Is(err, installErr) {
		t.Fatalf("publish error = %v, want installer failure", err)
	}
	if bus.capturedCtx != nil {
		t.Fatal("event published after durable execution envelope installation failed")
	}
}

func TestDurableExecutionArtifactIOFinishesBeforeEffectTransaction(t *testing.T) {
	type boundKey struct{}
	repo := &claimSuccessRepo{}
	bus := &contextCapturingEventBus{}
	txManager := &markingTxManager{}
	binderCalls := 0
	svc := NewOutboxService(repo, bus, txManager).WithDurableExecutionBinder(
		func(ctx context.Context, binding string) (DurableExecutionInstaller, error) {
			binderCalls++
			if ctx.Value(outboxPublishTxMarkerKey{}) == true {
				t.Fatal("immutable metadata was opened inside the effect transaction")
			}
			if binding != "exact-view" {
				t.Fatalf("binding = %q", binding)
			}
			return func(txCtx context.Context) (context.Context, error) {
				if txCtx.Value(outboxPublishTxMarkerKey{}) != true {
					t.Fatal("verified metadata was not installed in the effect transaction")
				}
				return context.WithValue(txCtx, boundKey{}, true), nil
			}, nil
		},
	)
	payload, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload:    RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"}},
		InternalMetadata: map[string]string{durableExecutionBindingMetadataKey: "exact-view"},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := svc.processClaimedEventAtomic(context.Background(), models.OutboxClaim{EventID: "event-1"}, string(sharedevents.RecordCreated), string(payload)); err != nil {
		t.Fatal(err)
	}
	if binderCalls != 1 || txManager.withPessimisticRetryCalls != 1 || txManager.withPessimisticCalls != 0 {
		t.Fatalf("binder calls = %d, retry transactions = %d, plain transactions = %d", binderCalls, txManager.withPessimisticRetryCalls, txManager.withPessimisticCalls)
	}
	if bus.capturedCtx == nil || bus.capturedCtx.Value(boundKey{}) != true {
		t.Fatal("publisher did not receive the transaction-bound immutable view")
	}
}

func TestLostRuntimeOwnershipCannotPublishOrMutateRetryState(t *testing.T) {
	repo := &claimSuccessRepo{}
	bus := &contextCapturingEventBus{}
	svc := NewOutboxService(repo, bus, noopTxManager{}).
		WithDurableExecutionBinder(func(context.Context, string) (DurableExecutionInstaller, error) {
			return func(ctx context.Context) (context.Context, error) { return ctx, nil }, nil
		}).
		WithDurableAttemptFence(func(context.Context) error { return persistence.ErrRuntimeWorkOwnershipLost })
	payload, _ := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload:    RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"}},
		InternalMetadata: map[string]string{durableExecutionBindingMetadataKey: "exact-view"},
	})
	err := svc.processClaimedEventAtomic(context.Background(), models.OutboxClaim{EventID: "event-stale"}, string(sharedevents.RecordCreated), string(payload))
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("error = %v", err)
	}
	if bus.capturedCtx != nil || len(repo.updateStatuses) != 0 || repo.retryCalls != 0 {
		t.Fatalf("stale worker produced side effects: publish=%v statuses=%v retry=%d", bus.capturedCtx != nil, repo.updateStatuses, repo.retryCalls)
	}
}

func TestLostRuntimeOwnershipCannotClaimAndHideWorkFromNewOwner(t *testing.T) {
	repo := &claimSuccessRepo{}
	txManager := &countingTxManager{}
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableExecutionBinder(func(context.Context, string) (DurableExecutionInstaller, error) {
			return func(ctx context.Context) (context.Context, error) { return ctx, nil }, nil
		}).
		WithDurableAttemptFence(func(context.Context) error { return persistence.ErrRuntimeWorkOwnershipLost })
	_, err := svc.claimEvent(context.Background(), "event-stale")
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("error = %v", err)
	}
	if repo.claimCalls != 0 || txManager.withPessimisticCalls != 1 {
		t.Fatalf("stale claim calls = %d, transactions = %d", repo.claimCalls, txManager.withPessimisticCalls)
	}
}

func TestDurableClaimFencesOwnershipAndClaimInOneTransaction(t *testing.T) {
	repo := &claimSuccessRepo{}
	txManager := &countingTxManager{}
	fenceInTx := false
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableExecutionBinder(func(context.Context, string) (DurableExecutionInstaller, error) {
			return func(ctx context.Context) (context.Context, error) { return ctx, nil }, nil
		}).
		WithDurableAttemptFence(func(ctx context.Context) error {
			fenceInTx = inOutboxTestTx(ctx)
			return nil
		})
	claimed, err := svc.claimEvent(context.Background(), "event-owned")
	if err != nil || claimed.EventID != "event-owned" || !fenceInTx || !repo.claimInTx || repo.claimCalls != 1 {
		t.Fatalf("claimed=%+v err=%v fenceInTx=%v claimInTx=%v calls=%d", claimed, err, fenceInTx, repo.claimInTx, repo.claimCalls)
	}
}

func TestDurableClaimUsesDeadlockFreeAdmissionOrder(t *testing.T) {
	repo := &orderedClaimRepo{}
	txManager := &countingTxManager{}
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableAdmissionFence(func(context.Context) error {
			repo.order = append(repo.order, "fence")
			return nil
		})
	if _, err := svc.claimEvent(context.Background(), "event-owned"); err != nil {
		t.Fatal(err)
	}
	if got := strings.Join(repo.order, ","); got != "fence,candidate,claim" {
		t.Fatalf("admission lock order = %q", got)
	}
}

func TestLostRuntimeOwnershipCannotRefreshProcessingClaim(t *testing.T) {
	repo := &claimSuccessRepo{}
	txManager := &countingTxManager{}
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableAttemptFence(func(context.Context) error { return persistence.ErrRuntimeWorkOwnershipLost })

	err := svc.refreshProcessingClaim(context.Background(), models.OutboxClaim{EventID: "event-stale"})
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("error = %v", err)
	}
	if repo.refreshCalls != 0 || txManager.withPessimisticCalls != 1 {
		t.Fatalf("stale heartbeat calls = %d, transactions = %d", repo.refreshCalls, txManager.withPessimisticCalls)
	}
}

func TestDurableHeartbeatFencesOwnershipAndRefreshInOneTransaction(t *testing.T) {
	repo := &claimSuccessRepo{}
	txManager := &countingTxManager{}
	fenceInTx := false
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableAttemptFence(func(ctx context.Context) error {
			fenceInTx = inOutboxTestTx(ctx)
			return nil
		})

	if err := svc.refreshProcessingClaim(context.Background(), models.OutboxClaim{EventID: "event-owned"}); err != nil {
		t.Fatal(err)
	}
	if !fenceInTx || !repo.refreshInTx || repo.refreshCalls != 1 || txManager.withPessimisticCalls != 1 {
		t.Fatalf("fenceInTx=%v refreshInTx=%v calls=%d transactions=%d",
			fenceInTx, repo.refreshInTx, repo.refreshCalls, txManager.withPessimisticCalls)
	}
}

func TestLostRuntimeOwnershipCannotDeleteCompletedQueueRows(t *testing.T) {
	repo := &claimSuccessRepo{cleanupDeleted: 2}
	txManager := &countingTxManager{}
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableAttemptFence(func(context.Context) error { return persistence.ErrRuntimeWorkOwnershipLost })

	deleted, err := svc.CleanupProcessedBatch(context.Background(), 24*time.Hour, 10)
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) || deleted != 0 {
		t.Fatalf("deleted=%d error=%v", deleted, err)
	}
	if repo.cleanupCalls != 0 || txManager.withPessimisticCalls != 1 {
		t.Fatalf("stale cleanup calls=%d transactions=%d", repo.cleanupCalls, txManager.withPessimisticCalls)
	}
}

func TestDurableCleanupFencesOwnershipAndDeleteInOneTransaction(t *testing.T) {
	repo := &claimSuccessRepo{cleanupDeleted: 2}
	txManager := &countingTxManager{}
	fenceInTx := false
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableAttemptFence(func(ctx context.Context) error {
			fenceInTx = inOutboxTestTx(ctx)
			return nil
		})

	deleted, err := svc.CleanupProcessedBatch(context.Background(), 24*time.Hour, 10)
	if err != nil || deleted != 2 || !fenceInTx || !repo.cleanupInTx || repo.cleanupCalls != 1 || txManager.withPessimisticCalls != 1 {
		t.Fatalf("deleted=%d err=%v fenceInTx=%v cleanupInTx=%v calls=%d transactions=%d",
			deleted, err, fenceInTx, repo.cleanupInTx, repo.cleanupCalls, txManager.withPessimisticCalls)
	}
}

func TestOwnershipFlipBeforeRetryBookkeepingCannotRewriteClaim(t *testing.T) {
	repo := &recordingOutboxRepository{updateStatusErr: errOutboxStatusWriteForTest}
	txManager := &markingTxManager{}
	fenceCalls := 0
	svc := NewOutboxService(repo, &contextCapturingEventBus{}, txManager).
		WithDurableExecutionBinder(func(context.Context, string) (DurableExecutionInstaller, error) {
			return func(ctx context.Context) (context.Context, error) { return ctx, nil }, nil
		}).
		WithDurableAttemptFence(func(context.Context) error {
			fenceCalls++
			if fenceCalls == 1 {
				return nil
			}
			return persistence.ErrRuntimeWorkOwnershipLost
		})
	payload, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload:    RecordEventPayload{ObjectAPIName: "Issue", Record: models.SObject{"Id": "issue-1"}},
		InternalMetadata: map[string]string{durableExecutionBindingMetadataKey: "exact-view"},
	})
	if err != nil {
		t.Fatal(err)
	}

	err = svc.processClaimedEventAtomic(context.Background(), models.OutboxClaim{EventID: "event-stale"}, string(sharedevents.RecordCreated), string(payload))
	if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("error = %v", err)
	}
	if repo.incrementRetryID != "" || fenceCalls != 2 || txManager.withPessimisticRetryCalls != 1 || txManager.withPessimisticCalls != 1 {
		t.Fatalf("stale bookkeeping mutated retry=%q fenceCalls=%d retryTransactions=%d plainTransactions=%d",
			repo.incrementRetryID, fenceCalls, txManager.withPessimisticRetryCalls, txManager.withPessimisticCalls)
	}
}

// TestEnqueueEventInheritsLineageOnNestedContext (test 4) — EnqueueEvent called
// with a context carrying existing chain lineage (as happens during AfterSave
// self-DML) inherits the parent chain_id and increments chain_depth.
func TestEnqueueEventInheritsLineageOnNestedContext(t *testing.T) {
	t.Parallel()

	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)

	parentCtx := WithLineage(context.Background(), "parent-chain-abc", 2)

	err := svc.EnqueueEvent(parentCtx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Opportunity",
		Record:        models.SObject{"Id": "006000000000001AAA"},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}

	internalMeta := repo.lastEnvelope(t).InternalMetadata
	chainID := internalMeta[string(constants.ContextKeyAfterSaveChainID)]
	chainDepth := internalMeta[string(constants.ContextKeyAfterSaveChainDepth)]

	if chainID != "parent-chain-abc" {
		t.Fatalf("expected inherited chain_id=%q, got %q", "parent-chain-abc", chainID)
	}
	if chainDepth != "3" {
		t.Fatalf("expected inherited chain_depth=\"3\" (2+1), got %q", chainDepth)
	}
}

// TestProcessEventAtomicRestoresLineageOntoCtx (test 5) — processEventAtomic
// restores chain_id and chain_depth from an outbox row's Metadata onto pubCtx
// as typed constants.ContextKey values (not raw strings), so that downstream
// code can read them via ctx.Value(constants.ContextKeyAfterSaveChainID).
// Parity test for Fact 3 (context restore typed-key contract).
func TestProcessEventAtomicRestoresLineageOntoCtx(t *testing.T) {
	t.Parallel()

	payloadJSON, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload: RecordEventPayload{
			ObjectAPIName: "Contact",
			Record:        models.SObject{"Id": "003000000000001AAA"},
		},
		InternalMetadata: map[string]string{
			string(constants.ContextKeyAfterSaveChainID):    "test-chain-restore",
			string(constants.ContextKeyAfterSaveChainDepth): "3",
		},
	})
	if err != nil {
		t.Fatalf("json.Marshal payload: %v", err)
	}

	bus := &contextCapturingEventBus{}
	repo := &claimSuccessRepo{}
	svc := NewOutboxService(repo, bus, noopTxManager{})

	err = svc.processEventAtomic(context.Background(), "evt-restore-001", string(sharedevents.RecordUpdated), string(payloadJSON))
	if err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}
	if bus.capturedCtx == nil {
		t.Fatal("expected eventBus.Publish to be called; capturedCtx is nil")
	}

	// Assert typed-key lookup works after restore (the contract this test pins).
	gotChainID := bus.capturedCtx.Value(constants.ContextKeyAfterSaveChainID)
	if gotChainID == nil {
		t.Fatal("ctx.Value(ContextKeyAfterSaveChainID) is nil; restore loop may be using raw string keys instead of typed constants.ContextKey")
	}
	if gotChainID.(string) != "test-chain-restore" {
		t.Fatalf("chain_id: expected %q, got %q", "test-chain-restore", gotChainID)
	}

	gotDepth := bus.capturedCtx.Value(constants.ContextKeyAfterSaveChainDepth)
	if gotDepth == nil {
		t.Fatal("ctx.Value(ContextKeyAfterSaveChainDepth) is nil after restore")
	}
	if gotDepth.(string) != "3" {
		t.Fatalf("chain_depth: expected %q, got %q", "3", gotDepth)
	}

	// End-to-end: LineageFromContext on the restored context must inherit correctly.
	chainID, chainDepth, isFresh := LineageFromContext(bus.capturedCtx)
	if chainID != "test-chain-restore" {
		t.Fatalf("LineageFromContext: expected chainID=%q, got %q", "test-chain-restore", chainID)
	}
	if chainDepth != 4 { // 3 + 1 (inherit increments depth)
		t.Fatalf("LineageFromContext: expected chainDepth=4, got %d", chainDepth)
	}
	if isFresh {
		t.Fatal("LineageFromContext: expected isFresh=false for inherited chain")
	}
}

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

	ctx := restoreInternalMetadata(context.Background(), map[string]string{
		"unregistered_internal_key": "must-not-be-restored",
	})

	if got := ctx.Value(constants.ContextKey("unregistered_internal_key")); got != nil {
		t.Fatalf("unregistered internal metadata was restored onto context: %#v", got)
	}
}

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

	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)
	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "generic-connector")

	err := svc.EnqueueEvent(ctx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "001000000000001AAA"},
		Metadata: map[string]string{
			constants.InternalMetadataKeyConnectorSyncOrigin: "public-leak",
		},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}

	envelope := repo.lastEnvelope(t)
	if got := envelope.InternalMetadata[constants.InternalMetadataKeyConnectorSyncOrigin]; got != "generic-connector" {
		t.Fatalf("InternalMetadata connector_sync_origin = %q, want %q", got, "generic-connector")
	}

	payloadJSON, err := json.Marshal(envelope)
	if err != nil {
		t.Fatalf("json.Marshal envelope: %v", err)
	}
	bus := &contextCapturingEventBus{}
	processRepo := &claimSuccessRepo{}
	processSvc := NewOutboxService(processRepo, bus, noopTxManager{})

	if err := processSvc.processEventAtomic(context.Background(), "evt-origin-restore", string(sharedevents.RecordUpdated), string(payloadJSON)); err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}
	if bus.capturedCtx == nil {
		t.Fatal("expected eventBus.Publish to be called; capturedCtx is nil")
	}
	if got, _ := bus.capturedCtx.Value(constants.ContextKeyConnectorSyncOrigin).(string); got != "generic-connector" {
		t.Fatalf("published ctx connector_sync_origin = %q, want %q", got, "generic-connector")
	}
	published, ok := bus.capturedPayload.(RecordEventPayload)
	if !ok {
		t.Fatalf("published payload type = %T, want RecordEventPayload", bus.capturedPayload)
	}
	if _, ok := published.Metadata[constants.InternalMetadataKeyConnectorSyncOrigin]; ok {
		t.Fatalf("connector_sync_origin leaked into public payload metadata: %#v", published.Metadata)
	}
}

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

	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)
	ctx := context.WithValue(context.Background(), constants.ContextKeySessionID, "session-validated-1")

	err := svc.EnqueueEvent(ctx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "001000000000001AAA"},
		Metadata: map[string]string{
			constants.InternalMetadataKeyOriginatingSessionID: "public-leak",
		},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}

	envelope := repo.lastEnvelope(t)
	if got := envelope.InternalMetadata[constants.InternalMetadataKeyOriginatingSessionID]; got != "session-validated-1" {
		t.Fatalf("InternalMetadata originating session = %q, want %q", got, "session-validated-1")
	}

	payloadJSON, err := json.Marshal(envelope)
	if err != nil {
		t.Fatalf("json.Marshal envelope: %v", err)
	}
	bus := &contextCapturingEventBus{}
	processRepo := &claimSuccessRepo{}
	processSvc := NewOutboxService(processRepo, bus, noopTxManager{})

	if err := processSvc.processEventAtomic(context.Background(), "evt-session-restore", string(sharedevents.RecordUpdated), string(payloadJSON)); err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}
	if bus.capturedCtx == nil {
		t.Fatal("expected eventBus.Publish to be called; capturedCtx is nil")
	}
	if got, _ := bus.capturedCtx.Value(constants.ContextKeySessionID).(string); got != "session-validated-1" {
		t.Fatalf("published ctx session selector = %q, want %q", got, "session-validated-1")
	}
	published, ok := bus.capturedPayload.(RecordEventPayload)
	if !ok {
		t.Fatalf("published payload type = %T, want RecordEventPayload", bus.capturedPayload)
	}
	if _, ok := published.Metadata[constants.InternalMetadataKeyOriginatingSessionID]; ok {
		t.Fatalf("originating session selector leaked into public payload metadata: %#v", published.Metadata)
	}
}

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

	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)
	ctx := context.WithValue(context.Background(), constants.ContextKeySuppressRecordTriggeredAutomation, "Account/001000000000001AAA")

	err := svc.EnqueueEvent(ctx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "001000000000001AAA"},
		Metadata: map[string]string{
			constants.InternalMetadataKeySuppressRecordTriggeredAutomation: "public-leak",
		},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() error = %v", err)
	}

	envelope := repo.lastEnvelope(t)
	if got := envelope.InternalMetadata[constants.InternalMetadataKeySuppressRecordTriggeredAutomation]; got != "Account/001000000000001AAA" {
		t.Fatalf("InternalMetadata suppress_record_triggered_automation = %q, want %q", got, "Account/001000000000001AAA")
	}

	payloadJSON, err := json.Marshal(envelope)
	if err != nil {
		t.Fatalf("json.Marshal envelope: %v", err)
	}
	bus := &contextCapturingEventBus{}
	processRepo := &claimSuccessRepo{}
	processSvc := NewOutboxService(processRepo, bus, noopTxManager{})

	if err := processSvc.processEventAtomic(context.Background(), "evt-suppress-restore", string(sharedevents.RecordUpdated), string(payloadJSON)); err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}
	if bus.capturedCtx == nil {
		t.Fatal("expected eventBus.Publish to be called; capturedCtx is nil")
	}
	if got, _ := bus.capturedCtx.Value(constants.ContextKeySuppressRecordTriggeredAutomation).(string); got != "Account/001000000000001AAA" {
		t.Fatalf("published ctx suppress_record_triggered_automation = %q, want %q", got, "Account/001000000000001AAA")
	}
	published, ok := bus.capturedPayload.(RecordEventPayload)
	if !ok {
		t.Fatalf("published payload type = %T, want RecordEventPayload", bus.capturedPayload)
	}
	if _, ok := published.Metadata[constants.InternalMetadataKeySuppressRecordTriggeredAutomation]; ok {
		t.Fatalf("suppress_record_triggered_automation leaked into public payload metadata: %#v", published.Metadata)
	}
}

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

	payloadJSON, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload: RecordEventPayload{
			ObjectAPIName: "Contact",
			Record:        models.SObject{"Id": "003000000000001AAA"},
		},
	})
	if err != nil {
		t.Fatalf("json.Marshal payload: %v", err)
	}

	bus := &contextCapturingEventBus{}
	repo := &claimSuccessRepo{}
	txm := &countingTxManager{}
	svc := NewOutboxService(repo, bus, txm)

	err = svc.processEventAtomic(context.Background(), "evt-pessimistic-001", string(sharedevents.RecordUpdated), string(payloadJSON))
	if err != nil {
		t.Fatalf("processEventAtomic() error = %v", err)
	}
	if txm.withPessimisticRetryCalls != 1 || txm.withPessimisticCalls != 0 {
		t.Fatalf("WithPessimisticRetry/WithPessimisticTransaction calls = %d/%d, want 1/0", txm.withPessimisticRetryCalls, txm.withPessimisticCalls)
	}
	if txm.withTxCalls != 0 {
		t.Fatalf("WithTransaction calls = %d, want 0", txm.withTxCalls)
	}
	if bus.capturedCtx == nil {
		t.Fatal("expected publish to run")
	}
	if !inOutboxTestTx(bus.capturedCtx) {
		t.Fatal("publish did not run inside the outbox claim/finalize transaction")
	}
	if len(repo.updateStatuses) != 1 || repo.updateStatuses[0] != OutboxStatusProcessed {
		t.Fatalf("UpdateStatus calls = %#v, want [%q]", repo.updateStatuses, OutboxStatusProcessed)
	}
	if repo.refreshCalls != 0 {
		t.Fatalf("RefreshOutboxProcessingClaim calls = %d, want 0 for a claim created in the same transaction", repo.refreshCalls)
	}
}

// TestSelfDmlInheritsChainIdAndIncrementsDepth (test 6) — simulates the
// self-DML path: a flow's recordUpdate element executes under an AfterSave
// context that already carries chain lineage. When persistence eventually calls
// EnqueueEvent with this context, the new outbox event must carry the same
// chain_id and chain_depth+1.
func TestSelfDmlInheritsChainIdAndIncrementsDepth(t *testing.T) {
	t.Parallel()

	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)

	// Simulate the AfterSave execution context: processEventAtomic restored
	// chain lineage onto pubCtx, and this ctx flows through FlowExecutor into
	// the persistence.Update call that triggers EnqueueEvent.
	aftersaveCtx := WithLineage(context.Background(), "self-dml-chain", 1)

	err := svc.EnqueueEvent(aftersaveCtx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Opportunity",
		Record:        models.SObject{"Id": "006000000000002AAA"},
	})
	if err != nil {
		t.Fatalf("EnqueueEvent() in self-DML simulation: error = %v", err)
	}

	internalMeta := repo.lastEnvelope(t).InternalMetadata
	chainID := internalMeta[string(constants.ContextKeyAfterSaveChainID)]
	chainDepth := internalMeta[string(constants.ContextKeyAfterSaveChainDepth)]

	if chainID != "self-dml-chain" {
		t.Fatalf("self-DML: expected inherited chain_id=%q, got %q", "self-dml-chain", chainID)
	}
	if chainDepth != "2" {
		t.Fatalf("self-DML: expected chain_depth=\"2\" (1+1), got %q", chainDepth)
	}
}

// TestEnqueueEventIncrementsTupleReservationsPerEmittedRow verifies that the
// writer advances the current event's reservation counts once per emitted
// outbox row, even when the caller reuses the same dispatch context.
func TestEnqueueEventIncrementsTupleReservationsPerEmittedRow(t *testing.T) {
	t.Parallel()

	repo := &payloadCapturingRepo{}
	svc := NewOutboxService(repo, nil, nil)
	tupleA := "001:001000000000001AAA:300000000000001AAA"
	tupleB := "001:001000000000001AAA:300000000000002AAA"

	ctx := WithCurrentEventRunnableTuples(
		WithAfterSaveTupleReservationCounts(
			WithLineage(context.Background(), "chain-a", 0),
			map[string]int{tupleA: 1},
		),
		map[string]struct{}{
			tupleA: {},
			tupleB: {},
		},
	)

	err := svc.EnqueueEvent(ctx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "001000000000001AAA"},
	})
	if err != nil {
		t.Fatalf("first EnqueueEvent() error = %v", err)
	}
	err = svc.EnqueueEvent(ctx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "001000000000001AAA"},
	})
	if err != nil {
		t.Fatalf("second EnqueueEvent() error = %v", err)
	}
	if len(repo.envelopes) != 2 {
		t.Fatalf("captured envelopes = %d, want 2", len(repo.envelopes))
	}

	first := parseTupleReservationCounts(repo.envelopes[0].InternalMetadata[string(constants.ContextKeyAfterSaveTupleCounts)])
	if first[tupleA] != 2 || first[tupleB] != 1 {
		t.Fatalf("first emitted row counts = %#v, want tupleA=2 tupleB=1", first)
	}

	second := parseTupleReservationCounts(repo.envelopes[1].InternalMetadata[string(constants.ContextKeyAfterSaveTupleCounts)])
	if second[tupleA] != 3 || second[tupleB] != 2 {
		t.Fatalf("second emitted row counts = %#v, want tupleA=3 tupleB=2", second)
	}
}

// TestEnqueueEventDoesNotAdvanceLiveTupleReservationsWhenEnqueueFails verifies
// the emitted envelope uses the incremented snapshot while the caller's shared
// reservation state stays unchanged if the outbox write fails.
func TestEnqueueEventDoesNotAdvanceLiveTupleReservationsWhenEnqueueFails(t *testing.T) {
	t.Parallel()

	repo := &payloadCapturingRepo{enqueueErr: errors.New("enqueue failed")}
	svc := NewOutboxService(repo, nil, nil)
	tupleA := "001:001000000000001AAA:300000000000001AAA"
	tupleB := "001:001000000000001AAA:300000000000002AAA"

	ctx := WithCurrentEventRunnableTuples(
		WithAfterSaveTupleReservationCounts(
			WithLineage(context.Background(), "chain-a", 0),
			map[string]int{tupleA: 1},
		),
		map[string]struct{}{
			tupleA: {},
			tupleB: {},
		},
	)

	err := svc.EnqueueEvent(ctx, sharedevents.RecordUpdated, RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "001000000000001AAA"},
	})
	if err == nil {
		t.Fatal("EnqueueEvent() error = nil, want enqueue failure")
	}

	if len(repo.envelopes) != 1 {
		t.Fatalf("captured envelopes = %d, want 1", len(repo.envelopes))
	}

	attempted := parseTupleReservationCounts(repo.envelopes[0].InternalMetadata[string(constants.ContextKeyAfterSaveTupleCounts)])
	if attempted[tupleA] != 2 || attempted[tupleB] != 1 {
		t.Fatalf("attempted emitted row counts = %#v, want tupleA=2 tupleB=1", attempted)
	}

	live := AfterSaveTupleReservationCounts(ctx)
	if live[tupleA] != 1 {
		t.Fatalf("live reservation counts after failed enqueue = %#v, want tupleA=1", live)
	}
	if _, ok := live[tupleB]; ok {
		t.Fatalf("live reservation counts after failed enqueue = %#v, want tupleB absent", live)
	}
}

// TestProcessEventAtomicPublishesPublicPayloadOnly verifies that internal
// recursion transport metadata is restored onto ctx but never leaked to normal
// record-event subscribers.
func TestProcessEventAtomicPublishesPublicPayloadOnly(t *testing.T) {
	t.Parallel()

	payloadJSON, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload: RecordEventPayload{
			ObjectAPIName: "Contact",
			Record:        models.SObject{"Id": "003000000000001AAA"},
			Metadata: map[string]string{
				string(constants.ContextKeyTraceparent): "00-1234567890abcdef1234567890abcdef-1234567890abcdef-01",
			},
		},
		InternalMetadata: map[string]string{
			string(constants.ContextKeyAfterSaveChainID):     "test-chain-restore",
			string(constants.ContextKeyAfterSaveChainDepth):  "3",
			string(constants.ContextKeyAfterSaveTupleCounts): `{"001:003000000000001AAA:300000000000001AAA":2}`,
			internalMetadataKeyCurrentEventRunnableTuples:    `["001:003000000000001AAA:300000000000001AAA"]`,
		},
	})
	if err != nil {
		t.Fatalf("json.Marshal payload: %v", err)
	}

	bus := &contextCapturingEventBus{}
	repo := &claimSuccessRepo{}
	svc := NewOutboxService(repo, bus, noopTxManager{})

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

	published, ok := bus.capturedPayload.(RecordEventPayload)
	if !ok {
		t.Fatalf("published payload type = %T, want RecordEventPayload", bus.capturedPayload)
	}
	if published.Metadata == nil {
		t.Fatal("published payload metadata unexpectedly nil; trace metadata should remain intact")
	}
	if _, ok := published.Metadata[string(constants.ContextKeyAfterSaveChainID)]; ok {
		t.Fatal("published payload leaked internal after_save_chain_id metadata")
	}
	if _, ok := published.Metadata[string(constants.ContextKeyAfterSaveChainDepth)]; ok {
		t.Fatal("published payload leaked internal after_save_chain_depth metadata")
	}
	if _, ok := published.Metadata[string(constants.ContextKeyAfterSaveTupleCounts)]; ok {
		t.Fatal("published payload leaked internal after_save_tuple_counts metadata")
	}
	if published.Metadata[string(constants.ContextKeyTraceparent)] == "" {
		t.Fatal("published payload should retain traceparent metadata")
	}
	if _, ok := published.Metadata[internalMetadataKeyCurrentEventRunnableTuples]; ok {
		t.Fatal("published payload leaked internal current-event runnable tuple metadata")
	}
}

// TestProcessEventAtomicRestoresInternalRecursionMetadataToCtx verifies that
// the new internal envelope round-trips chain lineage, reservation counts, and
// current-event runnable tuples back onto the worker context.
func TestProcessEventAtomicRestoresInternalRecursionMetadataToCtx(t *testing.T) {
	t.Parallel()

	tupleA := "001:003000000000001AAA:300000000000001AAA"
	tupleB := "001:003000000000001AAA:300000000000002AAA"

	payloadJSON, err := json.Marshal(InternalRecordEventEnvelope{
		PublicPayload: RecordEventPayload{
			ObjectAPIName: "Contact",
			Record:        models.SObject{"Id": "003000000000001AAA"},
		},
		InternalMetadata: map[string]string{
			string(constants.ContextKeyAfterSaveChainID):     "test-chain-restore",
			string(constants.ContextKeyAfterSaveChainDepth):  "3",
			string(constants.ContextKeyAfterSaveTupleCounts): `{"001:003000000000001AAA:300000000000001AAA":2}`,
			internalMetadataKeyCurrentEventRunnableTuples:    `["001:003000000000001AAA:300000000000001AAA","001:003000000000001AAA:300000000000002AAA"]`,
		},
	})
	if err != nil {
		t.Fatalf("json.Marshal payload: %v", err)
	}

	bus := &contextCapturingEventBus{}
	repo := &claimSuccessRepo{}
	svc := NewOutboxService(repo, bus, noopTxManager{})

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

	if gotChainID := bus.capturedCtx.Value(constants.ContextKeyAfterSaveChainID); gotChainID == nil || gotChainID.(string) != "test-chain-restore" {
		t.Fatalf("ctx after restore chain_id = %#v, want %q", gotChainID, "test-chain-restore")
	}
	if gotCounts := AfterSaveTupleReservationCounts(bus.capturedCtx); gotCounts[tupleA] != 2 {
		t.Fatalf("restored reservation counts = %#v, want tupleA=2", gotCounts)
	}
	if gotTuples := CurrentEventRunnableTuples(bus.capturedCtx); len(gotTuples) != 2 {
		t.Fatalf("restored runnable tuples len = %d, want 2", len(gotTuples))
	} else {
		if _, ok := gotTuples[tupleA]; !ok {
			t.Fatalf("restored runnable tuples missing %q", tupleA)
		}
		if _, ok := gotTuples[tupleB]; !ok {
			t.Fatalf("restored runnable tuples missing %q", tupleB)
		}
	}
}