package events

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
)

// MockOutboxRepository implements ports.OutboxRepository for testing
type MockOutboxRepository struct {
	mock.Mock
}

func (m *MockOutboxRepository) Enqueue(ctx context.Context, eventType string, payload interface{}) (string, error) {
	args := m.Called(ctx, eventType, payload)
	return args.String(0), args.Error(1)
}

func (m *MockOutboxRepository) GetPendingEvents(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	args := m.Called(ctx, limit)
	if args.Get(0) == nil {
		return nil, args.Error(1)
	}
	return args.Get(0).([]models.OutboxEvent), args.Error(1)
}

func (m *MockOutboxRepository) ClaimEvent(ctx context.Context, id string) (string, error) {
	args := m.Called(ctx, id)
	return args.String(0), args.Error(1)
}

func (m *MockOutboxRepository) UpdateStatus(ctx context.Context, id, status, errorMsg string) error {
	args := m.Called(ctx, id, status, errorMsg)
	return args.Error(0)
}

func (m *MockOutboxRepository) IncrementRetry(ctx context.Context, id string, retryCount int, lastError string) error {
	args := m.Called(ctx, id, retryCount, lastError)
	return args.Error(0)
}

func (m *MockOutboxRepository) CleanupProcessed(ctx context.Context, olderThan time.Time) (int64, error) {
	args := m.Called(ctx, olderThan)
	return args.Get(0).(int64), args.Error(1)
}

// MockTransactionManager implements ports.TransactionManager for testing
type MockTransactionManager struct {
	mock.Mock
}

func (m *MockTransactionManager) WithTransaction(ctx context.Context, fn func(context.Context) error) error {
	// Just execute the function directly for unit testing
	return fn(ctx)
}

func (m *MockTransactionManager) WithRetry(ctx context.Context, fn func(context.Context) error, maxRetries int) error {
	return fn(ctx)
}

func (m *MockTransactionManager) InjectTx(ctx context.Context, tx interface{}) context.Context {
	return ctx
}

func (m *MockTransactionManager) ExtractTx(ctx context.Context) interface{} {
	return nil
}

func (m *MockTransactionManager) IsTransactionActive(ctx context.Context) bool {
	return true
}

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

// Tests

func TestOutboxService_EnqueueEvent(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	payload := RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "test-id"},
	}

	// Use mock.MatchedBy because EnqueueEvent initializes Metadata if nil
	mockRepo.On("Enqueue", mock.Anything, "record.created", mock.MatchedBy(func(p RecordEventPayload) bool {
		return p.ObjectAPIName == "Account" && p.Record["Id"] == "test-id"
	})).Return("outbox-123", nil)

	err := svc.EnqueueEvent(context.Background(), events.EventType("record.created"), payload)

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_EnqueueEvent_Error(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	payload := RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "test-id"},
	}

	expectedErr := errors.New("database error")
	mockRepo.On("Enqueue", mock.Anything, "record.created", mock.MatchedBy(func(p RecordEventPayload) bool {
		return p.ObjectAPIName == "Account"
	})).Return("", expectedErr)

	err := svc.EnqueueEvent(context.Background(), events.EventType("record.created"), payload)

	assert.Error(t, err)
	assert.Equal(t, expectedErr, err)
	mockRepo.AssertExpectations(t)
}

// stubMetadataLookup is the minimal fake used by the
// outboxNoSubscriberObjects tests. lastQueriedObject lets a test
// assert the lookup was actually consulted, which guards against a
// regression where the dynamic arm is bypassed.
type stubMetadataLookup struct {
	hasFlow            bool
	lastQueriedObject  string
	queryCount         int
}

func (s *stubMetadataLookup) HasActiveRecordTriggeredFlow(_ context.Context, objectAPIName string) bool {
	s.lastQueriedObject = objectAPIName
	s.queryCount++
	return s.hasFlow
}

// TestOutboxService_EnqueueEvent_SkipsNoSubscriberObject verifies that
// record events for objects on the noise candidate list are dropped at
// the EnqueueEvent boundary when the metadata lookup confirms no active
// record-triggered flow is wired against the object.
//
// Why this matters: every persistent flow run inserts a FlowInterview
// row through PersistenceService, which would otherwise enqueue a
// record.created event. The dispatcher then matches zero flows for
// FlowInterview and returns nil — the row was pure overhead. Under
// sustained flow load on sandbox new-prod-essential (2026-04-22) this
// self-amplification produced 4500+ pending events in <30 minutes,
// blocking real CRM events from being processed (issue #1658).
//
// AssertNotCalled on the repo + asserting the metadata stub was
// queried = a regression in either arm (skipping the filter, OR
// skipping the dynamic check) fails immediately.
func TestOutboxService_EnqueueEvent_SkipsNoSubscriberObject(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()
	stub := &stubMetadataLookup{hasFlow: false}
	svc := NewOutboxService(mockRepo, eventBus, mockTxManager).WithMetadataLookup(stub)

	payload := RecordEventPayload{
		ObjectAPIName: "FlowInterview",
		Record:        models.SObject{"Id": "fi-test-id"},
	}

	err := svc.EnqueueEvent(context.Background(), events.RecordCreated, payload)

	assert.NoError(t, err)
	mockRepo.AssertNotCalled(t, "Enqueue", mock.Anything, mock.Anything, mock.Anything)
	mockRepo.AssertExpectations(t)
	assert.Equal(t, 1, stub.queryCount, "metadata lookup must be consulted before drop")
	assert.Equal(t, "FlowInterview", stub.lastQueriedObject)
}

// TestOutboxService_EnqueueEvent_SkipsNoSubscriberObject_CaseInsensitive
// guards against a regression where the table-style lowercase variant of
// the object name (PersistenceService normalises some names to the table
// alias) bypasses the filter.
func TestOutboxService_EnqueueEvent_SkipsNoSubscriberObject_CaseInsensitive(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()
	stub := &stubMetadataLookup{hasFlow: false}
	svc := NewOutboxService(mockRepo, eventBus, mockTxManager).WithMetadataLookup(stub)

	payload := RecordEventPayload{
		ObjectAPIName: "flowinterview",
		Record:        models.SObject{"Id": "fi-test-id"},
	}

	err := svc.EnqueueEvent(context.Background(), events.RecordUpdated, payload)

	assert.NoError(t, err)
	mockRepo.AssertNotCalled(t, "Enqueue", mock.Anything, mock.Anything, mock.Anything)
	mockRepo.AssertExpectations(t)
	assert.Equal(t, 1, stub.queryCount)
}

// TestOutboxService_EnqueueEvent_NoSubscriberObject_PreservesIfActiveFlow
// covers the F2 reviewer concern: a user CAN wire a record-triggered
// flow against FlowInterview via the frontend's TriggerConfigPanel. If
// such a flow is active, the noise filter must NOT drop the event —
// otherwise the dispatcher silently never fires.
func TestOutboxService_EnqueueEvent_NoSubscriberObject_PreservesIfActiveFlow(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()
	stub := &stubMetadataLookup{hasFlow: true} // user wired a flow against FlowInterview
	svc := NewOutboxService(mockRepo, eventBus, mockTxManager).WithMetadataLookup(stub)

	payload := RecordEventPayload{
		ObjectAPIName: "FlowInterview",
		Record:        models.SObject{"Id": "fi-test-id"},
	}

	mockRepo.On("Enqueue", mock.Anything, "record.created", mock.MatchedBy(func(p RecordEventPayload) bool {
		return p.ObjectAPIName == "FlowInterview" && p.Record["Id"] == "fi-test-id"
	})).Return("outbox-id-123", nil)

	err := svc.EnqueueEvent(context.Background(), events.RecordCreated, payload)

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
	assert.Equal(t, 1, stub.queryCount, "metadata lookup must be consulted")
}

// TestOutboxService_EnqueueEvent_NoSubscriberObject_FailSafeWhenLookupUnwired
// pins the safety contract: when WithMetadataLookup has not been
// called (default constructor only), an event for a noise-listed
// object MUST still enqueue. Silently dropping events when we cannot
// prove there is no subscriber is the worse failure mode.
func TestOutboxService_EnqueueEvent_NoSubscriberObject_FailSafeWhenLookupUnwired(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()
	svc := NewOutboxService(mockRepo, eventBus, mockTxManager) // no WithMetadataLookup

	payload := RecordEventPayload{
		ObjectAPIName: "FlowInterview",
		Record:        models.SObject{"Id": "fi-test-id"},
	}

	mockRepo.On("Enqueue", mock.Anything, "record.created", mock.MatchedBy(func(p RecordEventPayload) bool {
		return p.ObjectAPIName == "FlowInterview"
	})).Return("outbox-id-456", nil)

	err := svc.EnqueueEvent(context.Background(), events.RecordCreated, payload)

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_NoPendingEvents(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return([]models.OutboxEvent{}, nil)

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_Success(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Register a handler to receive the event
	var receivedPayload interface{}
	var wg sync.WaitGroup
	wg.Add(1)
	eventBus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		receivedPayload = p
		wg.Done()
		return nil
	})

	payload := RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "test-id"},
	}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{
			ID:         "outbox-1",
			EventType:  "record.created",
			Payload:    string(payloadJSON),
			RetryCount: 0,
		},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusProcessed, "").Return(nil)

	err := svc.ProcessOutbox(context.Background())

	wg.Wait()
	assert.NoError(t, err)
	assert.NotNil(t, receivedPayload)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_EventAlreadyClaimed(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "test-id"}}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(payloadJSON), RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("", nil) // Already claimed

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
	// UpdateStatus should NOT be called since event wasn't claimed
	mockRepo.AssertNotCalled(t, "UpdateStatus")
}

func TestOutboxService_ProcessOutbox_InvalidPayload(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: "invalid json", RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusFailed, mock.MatchedBy(func(s string) bool {
		return len(s) > 0 // Error message should be non-empty
	})).Return(nil)

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_RetryOnPublishError(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Register handler that fails
	eventBus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		return errors.New("publish failed")
	})

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "test-id"}}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(payloadJSON), RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("IncrementRetry", mock.Anything, "outbox-1", 1, mock.MatchedBy(func(s string) bool {
		return len(s) > 0
	})).Return(nil)

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_MaxRetriesExceeded(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Register handler that fails
	eventBus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		return errors.New("publish failed")
	})

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "test-id"}}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(payloadJSON), RetryCount: MaxRetryAttempts - 1},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusFailed, mock.MatchedBy(func(s string) bool {
		return len(s) > 0
	})).Return(nil)

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_CleanupProcessed(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	olderThan := 24 * time.Hour
	mockRepo.On("CleanupProcessed", mock.Anything, mock.MatchedBy(func(t time.Time) bool {
		// Should be approximately 24 hours ago
		return time.Since(t) < 25*time.Hour && time.Since(t) > 23*time.Hour
	})).Return(int64(10), nil)

	count, err := svc.CleanupProcessed(context.Background(), olderThan)

	assert.NoError(t, err)
	assert.Equal(t, int64(10), count)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_StartStopWorker(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Allow GetPendingEvents to be called
	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return([]models.OutboxEvent{}, nil).Maybe()

	// Start worker with short interval
	svc.StartWorker(50 * time.Millisecond)

	// Let it run briefly
	time.Sleep(100 * time.Millisecond)

	// Stop should complete without hanging
	done := make(chan struct{})
	go func() {
		svc.StopWorker()
		close(done)
	}()

	select {
	case <-done:
		// Success
	case <-time.After(2 * time.Second):
		t.Fatal("StopWorker did not complete in time")
	}
}

func TestNewOutboxService(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	require.NotNil(t, svc)
	assert.NotNil(t, svc.repo)
	assert.NotNil(t, svc.eventBus)
	assert.NotNil(t, svc.txManager)
	assert.NotNil(t, svc.stopCh)
}

func TestOutboxService_EnqueueEvent_WithTraceContext(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	payload := RecordEventPayload{
		ObjectAPIName: "Contact",
		Record:        models.SObject{"Id": "contact-1"},
	}

	// The handler should receive the payload with Metadata populated (traceparent injected)
	mockRepo.On("Enqueue", mock.Anything, "record.updated", mock.MatchedBy(func(p RecordEventPayload) bool {
		return p.ObjectAPIName == "Contact" &&
			p.Metadata != nil &&
			p.Metadata[string(constants.ContextKeyTraceparent)] == "00-traceid-spanid-01"
	})).Return("outbox-456", nil)

	ctx := context.WithValue(context.Background(), constants.ContextKeyTraceparent, "00-traceid-spanid-01")
	err := svc.EnqueueEvent(ctx, events.EventType("record.updated"), payload)

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_EnqueueEvent_PreservesExistingMetadata(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	payload := RecordEventPayload{
		ObjectAPIName: "Lead",
		Record:        models.SObject{"Id": "lead-1"},
		Metadata:      map[string]string{"custom_key": "custom_value"},
	}

	mockRepo.On("Enqueue", mock.Anything, "record.created", mock.MatchedBy(func(p RecordEventPayload) bool {
		return p.Metadata != nil &&
			p.Metadata["custom_key"] == "custom_value"
	})).Return("outbox-789", nil)

	err := svc.EnqueueEvent(context.Background(), events.EventType("record.created"), payload)

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_GetPendingEventsError(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	expectedErr := errors.New("database connection failed")
	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(nil, expectedErr)

	err := svc.ProcessOutbox(context.Background())

	assert.Error(t, err)
	assert.Equal(t, expectedErr, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_ClaimEventError(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "test-id"}}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(payloadJSON), RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("", errors.New("claim failed"))

	// ProcessOutbox itself succeeds (logs the error) but the tx callback returns error
	err := svc.ProcessOutbox(context.Background())

	// processEventAtomic wraps the claim error; ProcessOutbox logs it but does not propagate
	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_InvalidPayload_UpdateStatusError(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: "not-json", RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusFailed, mock.Anything).
		Return(errors.New("update status failed"))

	err := svc.ProcessOutbox(context.Background())

	// The error from UpdateStatus propagates through the tx, logged by ProcessOutbox
	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_IncrementRetryError(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Register handler that fails
	eventBus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		return errors.New("handler failed")
	})

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "test-id"}}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(payloadJSON), RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("IncrementRetry", mock.Anything, "outbox-1", 1, mock.Anything).
		Return(errors.New("increment retry failed"))

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_UpdateStatusError_OnProcessed(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Register handler that succeeds
	eventBus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		return nil
	})

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "test-id"}}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(payloadJSON), RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusProcessed, "").
		Return(errors.New("status update failed"))

	err := svc.ProcessOutbox(context.Background())

	// Error from UpdateStatus is wrapped and logged
	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_MultipleEvents(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	var processedCount int
	var mu sync.Mutex
	eventBus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		mu.Lock()
		processedCount++
		mu.Unlock()
		return nil
	})

	payload1 := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "a1"}}
	payload2 := RecordEventPayload{ObjectAPIName: "Contact", Record: models.SObject{"Id": "c1"}}
	p1JSON, _ := json.Marshal(payload1)
	p2JSON, _ := json.Marshal(payload2)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(p1JSON), RetryCount: 0},
		{ID: "outbox-2", EventType: "record.created", Payload: string(p2JSON), RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-2").Return("outbox-2", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusProcessed, "").Return(nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-2", OutboxStatusProcessed, "").Return(nil)

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	assert.Equal(t, 2, processedCount)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_MaxRetries_MarksFailed(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Handler that always fails
	eventBus.Subscribe(events.EventType("record.deleted"), func(ctx context.Context, p interface{}) error {
		return errors.New("permanent failure")
	})

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "test-id"}}
	payloadJSON, _ := json.Marshal(payload)

	// Event at max retries - 1 (next failure will exceed max)
	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.deleted", Payload: string(payloadJSON), RetryCount: MaxRetryAttempts - 1},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusFailed, mock.MatchedBy(func(s string) bool {
		return len(s) > 0 && assert.Contains(t, s, "max retries exceeded")
	})).Return(nil)

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_ProcessOutbox_MetadataContextRestoration(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	// Register handler that checks context
	var receivedCtxValues map[string]string
	eventBus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		receivedCtxValues = make(map[string]string)
		if v, ok := ctx.Value("trace_key").(string); ok {
			receivedCtxValues["trace_key"] = v
		}
		return nil
	})

	payload := RecordEventPayload{
		ObjectAPIName: "Account",
		Record:        models.SObject{"Id": "test-id"},
		Metadata:      map[string]string{"trace_key": "trace_value"},
	}
	payloadJSON, _ := json.Marshal(payload)

	pendingEvents := []models.OutboxEvent{
		{ID: "outbox-1", EventType: "record.created", Payload: string(payloadJSON), RetryCount: 0},
	}

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return(pendingEvents, nil)
	mockRepo.On("ClaimEvent", mock.Anything, "outbox-1").Return("outbox-1", nil)
	mockRepo.On("UpdateStatus", mock.Anything, "outbox-1", OutboxStatusProcessed, "").Return(nil)

	err := svc.ProcessOutbox(context.Background())

	assert.NoError(t, err)
	assert.Equal(t, "trace_value", receivedCtxValues["trace_key"])
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_CleanupProcessed_Error(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	expectedErr := errors.New("cleanup failed")
	mockRepo.On("CleanupProcessed", mock.Anything, mock.Anything).Return(int64(0), expectedErr)

	count, err := svc.CleanupProcessed(context.Background(), 24*time.Hour)

	assert.Error(t, err)
	assert.Equal(t, int64(0), count)
	mockRepo.AssertExpectations(t)
}

func TestOutboxService_StopWorker_Idempotent(t *testing.T) {
	mockRepo := new(MockOutboxRepository)
	mockTxManager := new(MockTransactionManager)
	eventBus := NewEventBus()

	svc := NewOutboxService(mockRepo, eventBus, mockTxManager)

	mockRepo.On("GetPendingEvents", mock.Anything, 100).Return([]models.OutboxEvent{}, nil).Maybe()

	svc.StartWorker(50 * time.Millisecond)
	time.Sleep(100 * time.Millisecond)

	// Stopping twice should not panic
	svc.StopWorker()
	assert.NotPanics(t, func() {
		svc.StopWorker()
	})
}

func TestOutboxConstants(t *testing.T) {
	assert.Equal(t, "pending", OutboxStatusPending)
	assert.Equal(t, "processed", OutboxStatusProcessed)
	assert.Equal(t, "failed", OutboxStatusFailed)
	assert.Equal(t, 5, MaxRetryAttempts)
}

// --- Parallelism regression tests (Phase 1 fix for outbox throughput) ---
//
// These tests guard the fix in /Users/amber/Documents/nexus/flow/DIAGNOSIS.md:
// ProcessOutbox must dispatch a batch of N distinct events with bounded
// concurrency rather than one-at-a-time. Under pure serial execution the
// overlap assertion (handlers in flight at the same moment) fails.

// concurrentMockRepo is a minimal thread-safe repo stub. testify/mock.Mock
// has internal locking that serialises method calls, which masks actual
// handler concurrency — so we use a hand-rolled stub for the parallelism
// tests and keep testify/mock for the other cases.
type concurrentMockRepo struct {
	mu           sync.Mutex
	pending      []models.OutboxEvent
	claimed      map[string]bool
	processed    map[string]bool
	claimCalls   int
	updateCalls  int
	retryCalls   int
	claimAttempt map[string]int // count of ClaimEvent attempts per id
}

func newConcurrentMockRepo(pending []models.OutboxEvent) *concurrentMockRepo {
	return &concurrentMockRepo{
		pending:      pending,
		claimed:      make(map[string]bool),
		processed:    make(map[string]bool),
		claimAttempt: make(map[string]int),
	}
}

func (m *concurrentMockRepo) Enqueue(_ context.Context, _ string, _ interface{}) (string, error) {
	return "", nil
}

func (m *concurrentMockRepo) GetPendingEvents(_ context.Context, _ int) ([]models.OutboxEvent, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	out := make([]models.OutboxEvent, len(m.pending))
	copy(out, m.pending)
	return out, nil
}

// ClaimEvent simulates the atomic UPDATE ... WHERE Status='pending' in the
// real repo: exactly one caller per id wins; the rest see RowsAffected()==0
// which the repo returns as ErrOutboxEventAlreadyClaimed. We return
// ("", nil) here (the legacy benign-skip signal) because processEventAtomic
// treats both ("", nil) AND the sentinel as "skip this event" — either
// pattern exercises the same code path and the authoritative
// double-dispatch check is in repo.processedCount().
func (m *concurrentMockRepo) ClaimEvent(_ context.Context, id string) (string, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.claimAttempt[id]++
	m.claimCalls++
	if m.claimed[id] {
		return "", nil
	}
	m.claimed[id] = true
	return id, nil
}

func (m *concurrentMockRepo) UpdateStatus(_ context.Context, id, status, _ string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.updateCalls++
	if status == OutboxStatusProcessed {
		m.processed[id] = true
	}
	return nil
}

func (m *concurrentMockRepo) IncrementRetry(_ context.Context, _ string, _ int, _ string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.retryCalls++
	return nil
}

func (m *concurrentMockRepo) CleanupProcessed(_ context.Context, _ time.Time) (int64, error) {
	return 0, nil
}

func (m *concurrentMockRepo) processedCount() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.processed)
}

// TestOutboxService_ProcessOutbox_DispatchesInParallel asserts that handlers
// for distinct events run concurrently, not strictly sequentially. The
// handler blocks on a barrier channel until all N events are in flight; if
// ProcessOutbox dispatched serially, the barrier would deadlock.
//
// With the fix (bounded goroutine pool) and parallelism >= N this PASSES.
// Without the fix (serial loop) this FAILS via timeout — proving the RED.
func TestOutboxService_ProcessOutbox_DispatchesInParallel(t *testing.T) {
	const n = 4

	// Use DISTINCT record ids — the partitioning fix (Round 2 H4)
	// intentionally serialises same-record events onto one slot. The
	// parallelism regression we care about is that DIFFERENT records
	// run concurrently, which is what the pre-fix serial loop regressed
	// and what the bounded-goroutine fix restores.
	pending := make([]models.OutboxEvent, n)
	for i := 0; i < n; i++ {
		payload := RecordEventPayload{
			ObjectAPIName: "Account",
			Record:        models.SObject{"Id": fmt.Sprintf("rec-%d", i)},
		}
		payloadJSON, _ := json.Marshal(payload)
		pending[i] = models.OutboxEvent{
			ID:         fmt.Sprintf("outbox-%d", i),
			EventType:  "record.created",
			Payload:    string(payloadJSON),
			RetryCount: 0,
		}
	}

	repo := newConcurrentMockRepo(pending)
	txm := new(MockTransactionManager)
	bus := NewEventBus()

	// Barrier: handlers block here until all N are in flight. If dispatch
	// is serial, the second handler never starts because the first is
	// blocked — deadlock → context timeout → test fails.
	inFlight := make(chan struct{}, n)
	release := make(chan struct{})
	bus.Subscribe(events.EventType("record.created"), func(ctx context.Context, _ interface{}) error {
		inFlight <- struct{}{}
		<-release
		return nil
	})

	svc := NewOutboxService(repo, bus, txm)

	// Fan the release once we've observed n concurrent in-flight handlers.
	go func() {
		// Wait for all N to arrive at the barrier, then release them.
		for i := 0; i < n; i++ {
			select {
			case <-inFlight:
			case <-time.After(3 * time.Second):
				return // timeout — handled by ProcessOutbox context below
			}
		}
		close(release)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	done := make(chan error, 1)
	go func() {
		done <- svc.ProcessOutbox(ctx)
	}()

	select {
	case err := <-done:
		assert.NoError(t, err)
	case <-time.After(5 * time.Second):
		t.Fatal("ProcessOutbox deadlocked — handlers were dispatched serially, not in parallel")
	}

	assert.Equal(t, n, repo.processedCount(), "all events should be marked processed")
}

// TestOutboxService_ProcessOutbox_ConcurrentClaimsNoDoubleDispatch asserts
// that even if ProcessOutbox is invoked from two goroutines simultaneously
// (e.g. two workers, or a stale tick overlapping a new one) AND both see
// the same pending list, each event is claimed exactly once and its
// handler invoked exactly once.
//
// The ClaimEvent mock simulates SELECT ... FOR UPDATE SKIP LOCKED — only
// the first caller per id wins. This test would regress if a future
// change moved claim-awareness outside the per-event txn.
func TestOutboxService_ProcessOutbox_ConcurrentClaimsNoDoubleDispatch(t *testing.T) {
	const n = 8

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "x"}}
	payloadJSON, _ := json.Marshal(payload)

	pending := make([]models.OutboxEvent, n)
	for i := 0; i < n; i++ {
		pending[i] = models.OutboxEvent{
			ID:         fmt.Sprintf("outbox-%d", i),
			EventType:  "record.created",
			Payload:    string(payloadJSON),
			RetryCount: 0,
		}
	}

	repo := newConcurrentMockRepo(pending)
	txm := new(MockTransactionManager)
	bus := NewEventBus()

	// No per-handler counter — the authoritative count is repo.processedCount(),
	// which is mutex-guarded and reflects exactly one UpdateStatus(processed)
	// per distinct event. The dead *int pointer the prior revision stored into
	// a sync.Map was never incremented and never read (Round 1 M3).
	bus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		_ = p.(RecordEventPayload)
		return nil
	})

	svc := NewOutboxService(repo, bus, txm)

	// Run ProcessOutbox from 3 goroutines concurrently.
	const workers = 3
	var wg sync.WaitGroup
	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = svc.ProcessOutbox(context.Background())
		}()
	}

	doneCh := make(chan struct{})
	go func() {
		wg.Wait()
		close(doneCh)
	}()

	select {
	case <-doneCh:
	case <-time.After(10 * time.Second):
		t.Fatal("concurrent ProcessOutbox did not complete in time")
	}

	// Exactly n events processed (no double-dispatch even with 3 concurrent workers).
	assert.Equal(t, n, repo.processedCount(), "each event should be processed exactly once across concurrent workers")

	// Each event's ClaimEvent was attempted at least once, and at most once succeeded.
	repo.mu.Lock()
	defer repo.mu.Unlock()
	for i := 0; i < n; i++ {
		id := fmt.Sprintf("outbox-%d", i)
		assert.True(t, repo.claimed[id], "event %s should be claimed", id)
		assert.True(t, repo.processed[id], "event %s should be processed exactly once", id)
	}
}

// TestOutboxService_ProcessOutbox_SameRecordOrdering asserts per-record FIFO
// is preserved across parallel slots. Five events for the same (Object, Id)
// pair, interleaved with two events for other records, must arrive at the
// handler in their original CreatedDate order — they hash to the same slot
// and the slot's worker drains its channel sequentially.
func TestOutboxService_ProcessOutbox_SameRecordOrdering(t *testing.T) {
	t.Setenv(outboxParallelismEnv, "4")

	// Build 7 events: positions 0,2,4,5,6 are for Account/a1 (the record
	// whose ordering we assert); positions 1 and 3 are noise on other
	// records. The "seq" field inside Metadata lets the handler record
	// arrival order per (Object, Id) without dragging in more plumbing.
	makeEvent := func(idx int, object, recID string) models.OutboxEvent {
		p := RecordEventPayload{
			ObjectAPIName: object,
			Record:        models.SObject{"Id": recID},
			Metadata:      map[string]string{"seq": fmt.Sprintf("%d", idx)},
		}
		b, _ := json.Marshal(p)
		return models.OutboxEvent{
			ID:         fmt.Sprintf("outbox-%d", idx),
			EventType:  "record.created",
			Payload:    string(b),
			RetryCount: 0,
		}
	}
	pending := []models.OutboxEvent{
		makeEvent(0, "Account", "a1"),
		makeEvent(1, "Contact", "c1"),
		makeEvent(2, "Account", "a1"),
		makeEvent(3, "Contact", "c2"),
		makeEvent(4, "Account", "a1"),
		makeEvent(5, "Account", "a1"),
		makeEvent(6, "Account", "a1"),
	}

	repo := newConcurrentMockRepo(pending)
	txm := new(MockTransactionManager)
	bus := NewEventBus()

	var arrivalMu sync.Mutex
	// arrivals maps "<object>/<id>" → slice of seq values in handler order
	arrivals := make(map[string][]int)
	bus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		pl := p.(RecordEventPayload)
		seq, _ := strconv.Atoi(pl.Metadata["seq"])
		key := pl.ObjectAPIName + "/" + pl.Record["Id"].(string)
		arrivalMu.Lock()
		arrivals[key] = append(arrivals[key], seq)
		arrivalMu.Unlock()
		return nil
	})

	svc := NewOutboxService(repo, bus, txm)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	require.NoError(t, svc.ProcessOutbox(ctx))

	arrivalMu.Lock()
	defer arrivalMu.Unlock()
	// Account/a1 must appear in original (0,2,4,5,6) order — same slot,
	// sequential channel drain.
	assert.Equal(t, []int{0, 2, 4, 5, 6}, arrivals["Account/a1"],
		"same-record events must preserve FIFO order across parallel slots")
	// Noise records each fire once in the order they were queued; we only
	// assert they landed, not their cross-record interleaving.
	assert.Equal(t, []int{1}, arrivals["Contact/c1"])
	assert.Equal(t, []int{3}, arrivals["Contact/c2"])
}

// TestOutboxService_ProcessOutbox_CtxCancelDuringBatch asserts that a
// cancelled context during a large batch causes ProcessOutbox to return
// ctx.Err() without hanging on the undispatched tail — the core H1 fix.
// The handler blocks the first two slots indefinitely; once ctx is
// cancelled, the dispatch loop must bail out of the channel send.
func TestOutboxService_ProcessOutbox_CtxCancelDuringBatch(t *testing.T) {
	t.Setenv(outboxParallelismEnv, "2")

	// 100 events on the SAME record so they all hash to the same slot;
	// that slot's channel buffer (cap 4) fills, the slot worker blocks on
	// the release channel, and the dispatch loop can't send event #5+.
	// Cancelling ctx must unblock the send select arm.
	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "same"}}
	payloadJSON, _ := json.Marshal(payload)
	pending := make([]models.OutboxEvent, 100)
	for i := range pending {
		pending[i] = models.OutboxEvent{
			ID:         fmt.Sprintf("outbox-%d", i),
			EventType:  "record.created",
			Payload:    string(payloadJSON),
			RetryCount: 0,
		}
	}

	repo := newConcurrentMockRepo(pending)
	txm := new(MockTransactionManager)
	bus := NewEventBus()

	release := make(chan struct{})
	bus.Subscribe(events.EventType("record.created"), func(ctx context.Context, _ interface{}) error {
		<-release
		return nil
	})

	svc := NewOutboxService(repo, bus, txm)

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan error, 1)
	go func() {
		done <- svc.ProcessOutbox(ctx)
	}()

	// Give the dispatch loop time to fill the slot buffer + block on send.
	time.Sleep(100 * time.Millisecond)
	cancel()

	// After cancel, the select in the dispatch loop should fire the
	// ctx.Done() arm; close(slotChans[*]) then unblocks workers via
	// close-range-exit, release is still unclosed so any in-flight
	// handler stays blocked — we free them here to let the test exit.
	close(release)

	select {
	case err := <-done:
		assert.ErrorIs(t, err, context.Canceled, "ProcessOutbox should return ctx.Err() after cancel")
	case <-time.After(3 * time.Second):
		t.Fatal("ProcessOutbox did not return after ctx cancel — dispatch loop is blocking the semaphore without ctx.Done()")
	}
}

// TestResolveOutboxParallelism covers the env-var override clamp logic
// so a future change can't silently regress the [1, 64] bound.
func TestResolveOutboxParallelism(t *testing.T) {
	cases := []struct {
		name string
		env  string
		want int
	}{
		{"empty uses default", "", defaultOutboxParallelism},
		{"invalid uses default", "not-a-number", defaultOutboxParallelism},
		{"valid in range", "16", 16},
		{"below min clamps up", "0", minOutboxParallelism},
		{"negative clamps up", "-5", minOutboxParallelism},
		{"above max clamps down", "1000", maxOutboxParallelism},
		{"at min", "1", 1},
		{"at max", "64", 64},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			if tc.env == "" {
				t.Setenv(outboxParallelismEnv, "")
			} else {
				t.Setenv(outboxParallelismEnv, tc.env)
			}
			got := resolveOutboxParallelism()
			assert.Equal(t, tc.want, got)
		})
	}
}

// TestOutboxService_SlotWorkerPanicIsolated covers the Round 3 C3 fix:
// when a handler panics on one event, the slot worker must continue
// draining its channel so the remaining same-slot events still dispatch.
// The prior Round 2 placement put `defer recover()` OUTSIDE the
// `for env := range in` loop, so a single panic unwound the whole
// goroutine — stranding every subsequent event on that slot for up to
// OutboxStaleProcessingSeconds (300s) before the reclaim predicate
// would pick them up on a later tick.
//
// We force same-slot routing by using identical (ObjectAPIName, Id) on
// every event; they all hash to the same partition. The handler panics
// on the 3rd arrival and succeeds otherwise.
func TestOutboxService_SlotWorkerPanicIsolated(t *testing.T) {
	// Force a single slot so every event lands on the same goroutine —
	// that's the only way the panic-isolation property is observable.
	t.Setenv(outboxParallelismEnv, "1")

	payload := RecordEventPayload{ObjectAPIName: "Account", Record: models.SObject{"Id": "same"}}
	payloadJSON, _ := json.Marshal(payload)

	const n = 5
	pending := make([]models.OutboxEvent, n)
	for i := 0; i < n; i++ {
		pending[i] = models.OutboxEvent{
			ID:         fmt.Sprintf("outbox-%d", i),
			EventType:  "record.created",
			Payload:    string(payloadJSON),
			RetryCount: 0,
		}
	}

	repo := newConcurrentMockRepo(pending)
	txm := new(MockTransactionManager)
	bus := NewEventBus()

	var mu sync.Mutex
	dispatched := make([]string, 0, n)
	bus.Subscribe(events.EventType("record.created"), func(ctx context.Context, p interface{}) error {
		mu.Lock()
		// The event id isn't carried in the payload; but the arrival
		// order on a single slot must match the pending-list order, so
		// we record the call count — that's enough to assert isolation.
		dispatched = append(dispatched, fmt.Sprintf("call-%d", len(dispatched)))
		idx := len(dispatched) - 1
		mu.Unlock()
		if idx == 2 {
			panic("simulated handler panic on 3rd event")
		}
		return nil
	})

	svc := NewOutboxService(repo, bus, txm)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	err := svc.ProcessOutbox(ctx)
	assert.NoError(t, err, "ProcessOutbox itself must not propagate a slot-worker panic")

	mu.Lock()
	defer mu.Unlock()
	// All 5 handlers must have been invoked, including the ones AFTER
	// the panicking call-2. If recover lived outside the for-range loop
	// this would be len==3 and the assertion would fail.
	assert.Equal(t, n, len(dispatched),
		"panic on one event must not strand sibling events on the same slot")
}

// TestSlotForEvent_SameRecordSameSlot asserts the partition function
// groups same-record events together and falls back to event ID when
// the payload lacks a record identity.
func TestSlotForEvent_SameRecordSameSlot(t *testing.T) {
	mkPayload := func(object, id string) string {
		p := RecordEventPayload{ObjectAPIName: object, Record: models.SObject{"Id": id}}
		b, _ := json.Marshal(p)
		return string(b)
	}

	t.Run("same record → same slot", func(t *testing.T) {
		p := mkPayload("Account", "a1")
		s1 := slotForEvent("id-1", p, 8)
		s2 := slotForEvent("id-2", p, 8)
		assert.Equal(t, s1, s2, "identical record keys must hash to the same slot")
	})

	t.Run("different records → usually different slots", func(t *testing.T) {
		// Probabilistic but stable: fnv32 of these two distinct keys
		// mod 8 is 5 vs 3 — checked at authoring time. If this ever
		// flakes, switch to asserting non-constant output across many
		// inputs instead.
		s1 := slotForEvent("id-1", mkPayload("Account", "a1"), 8)
		s2 := slotForEvent("id-2", mkPayload("Account", "a2"), 8)
		// We can't assert != strictly (hash collision possible), but we
		// can assert both are in range.
		assert.GreaterOrEqual(t, s1, 0)
		assert.Less(t, s1, 8)
		assert.GreaterOrEqual(t, s2, 0)
		assert.Less(t, s2, 8)
	})

	t.Run("empty payload → falls back to event ID", func(t *testing.T) {
		s := slotForEvent("id-1", "", 8)
		assert.GreaterOrEqual(t, s, 0)
		assert.Less(t, s, 8)
	})

	t.Run("parallelism=1 → always slot 0", func(t *testing.T) {
		s := slotForEvent("anything", mkPayload("X", "y"), 1)
		assert.Equal(t, 0, s)
	})
}