package automation

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

	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestApprovalActionReplayWorker_DrainsPendingRecallFieldUpdateAfterRestart(t *testing.T) {
	ctx := context.Background()
	store := newMemoryApprovalReplayStore(t)
	store.addPending(t, ApprovalActionReplayPayload{
		Version:             1,
		LifecycleEvent:      constants.ApprovalActionEventRecall,
		ProcessInstanceID:   "pi-restart",
		ProcessDefinitionID: "pd-restart",
		ObjectAPIName:       "GenericObject",
		TargetObjectID:      "target-restart",
		User:                &models.UserSession{ID: "005-submitter"},
		Actions: []ApprovalActionReplayFieldUpdate{{
			ActionID:           "pa-restart",
			ActionName:         "Set Recall State",
			FieldAPIName:       "LifecycleState",
			FieldValue:         "recalled",
			UpdateObjectName:   "GenericObject",
			UpdateRecordID:     "target-restart",
			UpdateFieldAPIName: "LifecycleState",
		}},
	}, time.Date(2026, 5, 15, 8, 0, 0, 0, time.UTC))

	persistence := &replayPersistenceStub{
		records: map[string]models.SObject{
			"GenericObject/target-restart": {
				"Id":             "target-restart",
				"LifecycleState": "submitted",
			},
		},
	}
	svc := NewApprovalService(persistence, nil, nil, nil, nil, nil, nil, logging.Default)
	worker := NewApprovalActionReplayWorker(store, replayTxStub{}, svc, ApprovalActionReplayConfig{
		HolderID:       "restart-worker",
		MaxAttempts:    3,
		InitialBackoff: time.Second,
		LeaseDuration:  time.Minute,
	})

	processed, err := worker.ProcessOnce(ctx)
	if err != nil {
		t.Fatalf("ProcessOnce() error = %v", err)
	}
	if processed != 1 {
		t.Fatalf("processed rows = %d, want 1", processed)
	}
	if got := persistence.records["GenericObject/target-restart"]["LifecycleState"]; got != "recalled" {
		t.Fatalf("LifecycleState = %v, want recalled", got)
	}
	row := store.rows[1]
	if row.status != ApprovalActionReplayStatusProcessed {
		t.Fatalf("row status = %q, want processed", row.status)
	}
	if row.processedAt == nil {
		t.Fatal("ProcessedAt was not set")
	}
}

func TestApprovalActionReplayWorker_StopCancelsProcessOnceContext(t *testing.T) {
	store := &blockingApprovalReplayStore{
		claimStarted: make(chan struct{}),
		claimErr:     make(chan error, 1),
	}
	svc := NewApprovalService(&replayPersistenceStub{records: map[string]models.SObject{}}, nil, nil, nil, nil, nil, nil, logging.Default)
	worker := NewApprovalActionReplayWorker(store, replayTxStub{}, svc, ApprovalActionReplayConfig{
		HolderID:      "stop-cancel-worker",
		PollInterval:  time.Hour,
		LeaseDuration: time.Minute,
	})

	worker.Start()
	select {
	case <-store.claimStarted:
	case <-time.After(time.Second):
		t.Fatal("worker did not enter ClaimDue")
	}

	stopped := make(chan struct{})
	go func() {
		worker.Stop()
		close(stopped)
	}()
	select {
	case <-stopped:
	case <-time.After(time.Second):
		t.Fatal("Stop() did not cancel the active ProcessOnce context")
	}
	select {
	case err := <-store.claimErr:
		if !errors.Is(err, context.Canceled) {
			t.Fatalf("ClaimDue context error = %v, want context.Canceled", err)
		}
	default:
		t.Fatal("ClaimDue did not observe context cancellation")
	}
}

func TestApprovalActionReplayWorker_RecoversProcessOncePanic(t *testing.T) {
	store := &panicOnceApprovalReplayStore{calls: make(chan int, 4)}
	svc := NewApprovalService(&replayPersistenceStub{records: map[string]models.SObject{}}, nil, nil, nil, nil, nil, nil, logging.Default)
	worker := NewApprovalActionReplayWorker(store, replayTxStub{}, svc, ApprovalActionReplayConfig{
		HolderID:      "panic-recover-worker",
		PollInterval:  time.Millisecond,
		LeaseDuration: time.Minute,
	})

	worker.Start()
	defer worker.Stop()

	for want := 1; want <= 2; want++ {
		select {
		case got := <-store.calls:
			if got != want {
				t.Fatalf("ClaimDue call = %d, want %d", got, want)
			}
		case <-time.After(time.Second):
			t.Fatalf("worker did not reach ClaimDue call %d after panic recovery", want)
		}
	}
}

func TestApprovalActionReplayWorker_RecoversProcessRowPanicAndContinuesBatch(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 5, 15, 8, 30, 0, 0, time.UTC)
	store := newMemoryApprovalReplayStore(t)
	store.addPending(t, ApprovalActionReplayPayload{
		Version:             1,
		LifecycleEvent:      constants.ApprovalActionEventRecall,
		ProcessInstanceID:   "pi-poison",
		ProcessDefinitionID: "pd-poison",
		ObjectAPIName:       "GenericObject",
		TargetObjectID:      "target-poison",
		User:                &models.UserSession{ID: "005-submitter"},
		Actions: []ApprovalActionReplayFieldUpdate{{
			ActionID:           "pa-poison",
			ActionName:         "Poison Row",
			FieldAPIName:       "LifecycleState",
			FieldValue:         "recalled-poison",
			UpdateObjectName:   "GenericObject",
			UpdateRecordID:     "target-poison",
			UpdateFieldAPIName: "LifecycleState",
		}},
	}, now)
	store.addPending(t, ApprovalActionReplayPayload{
		Version:             1,
		LifecycleEvent:      constants.ApprovalActionEventRecall,
		ProcessInstanceID:   "pi-ok",
		ProcessDefinitionID: "pd-ok",
		ObjectAPIName:       "GenericObject",
		TargetObjectID:      "target-ok",
		User:                &models.UserSession{ID: "005-submitter"},
		Actions: []ApprovalActionReplayFieldUpdate{{
			ActionID:           "pa-ok",
			ActionName:         "Set Recall State",
			FieldAPIName:       "LifecycleState",
			FieldValue:         "recalled",
			UpdateObjectName:   "GenericObject",
			UpdateRecordID:     "target-ok",
			UpdateFieldAPIName: "LifecycleState",
		}},
	}, now)

	persistence := &replayPersistenceStub{
		records: map[string]models.SObject{
			"GenericObject/target-poison": {
				"Id":             "target-poison",
				"LifecycleState": "submitted",
			},
			"GenericObject/target-ok": {
				"Id":             "target-ok",
				"LifecycleState": "submitted",
			},
		},
	}
	svc := NewApprovalService(persistence, nil, nil, nil, nil, nil, nil, logging.Default)
	txRunner := &panicFirstReplayTxStub{}
	worker := NewApprovalActionReplayWorker(store, txRunner, svc, ApprovalActionReplayConfig{
		HolderID:       "row-panic-worker",
		BatchSize:      2,
		MaxAttempts:    3,
		InitialBackoff: time.Second,
		LeaseDuration:  time.Minute,
	})
	worker.clock = func() time.Time { return now }

	processed, err := worker.ProcessOnce(ctx)
	if err != nil {
		t.Fatalf("ProcessOnce() error = %v", err)
	}
	if processed != 2 {
		t.Fatalf("processed rows = %d, want 2", processed)
	}
	if txRunner.pessimisticCalls != 2 {
		t.Fatalf("WithPessimisticTransaction calls = %d, want 2", txRunner.pessimisticCalls)
	}

	first := store.rows[1]
	if first.status != ApprovalActionReplayStatusPending {
		t.Fatalf("first row status = %q, want pending retry", first.status)
	}
	if first.attempts != 1 {
		t.Fatalf("first row attempts = %d, want 1", first.attempts)
	}
	if !first.nextAttemptAt.Equal(now.Add(time.Second)) {
		t.Fatalf("first row NextAttemptAt = %s, want %s", first.nextAttemptAt, now.Add(time.Second))
	}
	if first.lastErr == "" {
		t.Fatal("first row LastError was not recorded")
	}
	if got := persistence.records["GenericObject/target-poison"]["LifecycleState"]; got != "submitted" {
		t.Fatalf("target-poison LifecycleState = %v, want submitted", got)
	}

	second := store.rows[2]
	if second.status != ApprovalActionReplayStatusProcessed {
		t.Fatalf("second row status = %q, want processed", second.status)
	}
	if second.processedAt == nil {
		t.Fatal("second row ProcessedAt was not set")
	}
	if got := persistence.records["GenericObject/target-ok"]["LifecycleState"]; got != "recalled" {
		t.Fatalf("target-ok LifecycleState = %v, want recalled", got)
	}
}

func TestApprovalActionReplayWorker_ExhaustionMarksDurableFailureAndAlert(t *testing.T) {
	ctx := context.Background()
	now := time.Date(2026, 5, 15, 9, 0, 0, 0, time.UTC)
	store := newMemoryApprovalReplayStore(t)
	store.addPending(t, ApprovalActionReplayPayload{
		Version:             1,
		LifecycleEvent:      constants.ApprovalActionEventRecall,
		ProcessInstanceID:   "pi-fail",
		ProcessDefinitionID: "pd-fail",
		ObjectAPIName:       "GenericObject",
		TargetObjectID:      "target-fail",
		User:                &models.UserSession{ID: "005-submitter"},
		Actions: []ApprovalActionReplayFieldUpdate{{
			ActionID:           "pa-fail",
			ActionName:         "Set Recall State",
			FieldAPIName:       "LifecycleState",
			FieldValue:         "recalled",
			UpdateObjectName:   "GenericObject",
			UpdateRecordID:     "target-fail",
			UpdateFieldAPIName: "LifecycleState",
		}},
	}, now)

	persistence := &replayPersistenceStub{
		records: map[string]models.SObject{
			"GenericObject/target-fail": {
				"Id":             "target-fail",
				"LifecycleState": "submitted",
			},
		},
		updateErr: errors.New("permanent target update failure"),
	}
	svc := NewApprovalService(persistence, nil, nil, nil, nil, nil, nil, logging.Default)
	worker := NewApprovalActionReplayWorker(store, replayTxStub{}, svc, ApprovalActionReplayConfig{
		HolderID:       "failure-worker",
		MaxAttempts:    2,
		InitialBackoff: 10 * time.Second,
		MaxBackoff:     10 * time.Second,
		LeaseDuration:  time.Minute,
	})
	worker.clock = func() time.Time { return now }

	if _, err := worker.ProcessOnce(ctx); err != nil {
		t.Fatalf("first ProcessOnce() error = %v", err)
	}
	row := store.rows[1]
	if row.status != ApprovalActionReplayStatusPending {
		t.Fatalf("row status after first failure = %q, want pending retry", row.status)
	}
	if row.attempts != 1 {
		t.Fatalf("attempts after first failure = %d, want 1", row.attempts)
	}
	if !row.nextAttemptAt.Equal(now.Add(10 * time.Second)) {
		t.Fatalf("NextAttemptAt = %s, want %s", row.nextAttemptAt, now.Add(10*time.Second))
	}
	if row.alertedAt != nil {
		t.Fatalf("AlertedAt after first failure = %v, want nil", row.alertedAt)
	}

	later := now.Add(10 * time.Second)
	worker.clock = func() time.Time { return later }
	if _, err := worker.ProcessOnce(ctx); err != nil {
		t.Fatalf("second ProcessOnce() error = %v", err)
	}
	row = store.rows[1]
	if row.status != ApprovalActionReplayStatusFailed {
		t.Fatalf("row status after exhaustion = %q, want failed", row.status)
	}
	if row.attempts != 2 {
		t.Fatalf("attempts after exhaustion = %d, want 2", row.attempts)
	}
	if row.failedAt == nil || row.alertedAt == nil {
		t.Fatalf("failedAt=%v alertedAt=%v, want both set", row.failedAt, row.alertedAt)
	}
	if row.lastErr == "" {
		t.Fatal("LastError was not recorded")
	}
}

func TestApprovalActionReplayWorker_UsesFrozenRelatedRecordTarget(t *testing.T) {
	ctx := context.Background()
	store := newMemoryApprovalReplayStore(t)
	store.addPending(t, ApprovalActionReplayPayload{
		Version:             1,
		LifecycleEvent:      constants.ApprovalActionEventRecall,
		ProcessInstanceID:   "pi-related",
		ProcessDefinitionID: "pd-related",
		ObjectAPIName:       "GenericObject",
		TargetObjectID:      "target-related",
		User:                &models.UserSession{ID: "005-submitter"},
		Actions: []ApprovalActionReplayFieldUpdate{{
			ActionID:           "pa-related",
			ActionName:         "Freeze Related State",
			FieldAPIName:       "ParentId.LifecycleState",
			FieldValue:         "recalled",
			UpdateObjectName:   "RelatedObject",
			UpdateRecordID:     "parent-at-recall",
			UpdateFieldAPIName: "LifecycleState",
		}},
	}, time.Date(2026, 5, 15, 8, 0, 0, 0, time.UTC))

	persistence := &replayPersistenceStub{
		records: map[string]models.SObject{
			"GenericObject/target-related": {
				"Id":       "target-related",
				"ParentId": "parent-after-recall",
			},
			"RelatedObject/parent-at-recall": {
				"Id":             "parent-at-recall",
				"LifecycleState": "submitted",
			},
			"RelatedObject/parent-after-recall": {
				"Id":             "parent-after-recall",
				"LifecycleState": "submitted",
			},
		},
	}
	svc := NewApprovalService(persistence, nil, nil, nil, nil, nil, nil, logging.Default)
	worker := NewApprovalActionReplayWorker(store, replayTxStub{}, svc, ApprovalActionReplayConfig{
		HolderID:       "related-worker",
		MaxAttempts:    3,
		InitialBackoff: time.Second,
		LeaseDuration:  time.Minute,
	})

	if _, err := worker.ProcessOnce(ctx); err != nil {
		t.Fatalf("ProcessOnce() error = %v", err)
	}
	if got := persistence.records["RelatedObject/parent-at-recall"]["LifecycleState"]; got != "recalled" {
		t.Fatalf("parent-at-recall LifecycleState = %v, want recalled", got)
	}
	if got := persistence.records["RelatedObject/parent-after-recall"]["LifecycleState"]; got != "submitted" {
		t.Fatalf("parent-after-recall LifecycleState = %v, want submitted", got)
	}
	if len(persistence.updates) != 1 {
		t.Fatalf("updates = %d, want 1", len(persistence.updates))
	}
	update := persistence.updates[0]
	if update.objectName != "RelatedObject" || update.id != "parent-at-recall" {
		t.Fatalf("update target = %s/%s, want RelatedObject/parent-at-recall", update.objectName, update.id)
	}
}

type replayTxStub struct{}

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

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

type panicFirstReplayTxStub struct {
	pessimisticCalls int
}

func (s *panicFirstReplayTxStub) WithTransaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}

func (s *panicFirstReplayTxStub) WithPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	s.pessimisticCalls++
	if s.pessimisticCalls == 1 {
		panic("process row panic")
	}
	return fn(ctx)
}

type blockingApprovalReplayStore struct {
	claimStarted chan struct{}
	claimErr     chan error
}

func (s *blockingApprovalReplayStore) EnqueueRecallFieldUpdates(ctx context.Context, payload ApprovalActionReplayPayload) (int64, error) {
	panic("blockingApprovalReplayStore.EnqueueRecallFieldUpdates: not implemented")
}

func (s *blockingApprovalReplayStore) ClaimDue(ctx context.Context, holderID string, batchSize int, lease time.Duration, now time.Time) ([]ApprovalActionReplayRow, error) {
	select {
	case <-s.claimStarted:
	default:
		close(s.claimStarted)
	}
	<-ctx.Done()
	err := ctx.Err()
	s.claimErr <- err
	return nil, err
}

func (s *blockingApprovalReplayStore) MarkProcessed(ctx context.Context, id int64, holderID string, now time.Time) error {
	panic("blockingApprovalReplayStore.MarkProcessed: not implemented")
}

func (s *blockingApprovalReplayStore) ScheduleRetry(ctx context.Context, id int64, holderID string, attempts int, nextAttemptAt time.Time, lastErr string, now time.Time) error {
	panic("blockingApprovalReplayStore.ScheduleRetry: not implemented")
}

func (s *blockingApprovalReplayStore) MarkFailed(ctx context.Context, id int64, holderID string, attempts int, lastErr string, now time.Time) error {
	panic("blockingApprovalReplayStore.MarkFailed: not implemented")
}

func (s *blockingApprovalReplayStore) Counts(ctx context.Context) (ApprovalActionReplayCounts, error) {
	panic("blockingApprovalReplayStore.Counts: not implemented")
}

type panicOnceApprovalReplayStore struct {
	calls chan int
	mu    sync.Mutex
	count int
}

func (s *panicOnceApprovalReplayStore) EnqueueRecallFieldUpdates(ctx context.Context, payload ApprovalActionReplayPayload) (int64, error) {
	panic("panicOnceApprovalReplayStore.EnqueueRecallFieldUpdates: not implemented")
}

func (s *panicOnceApprovalReplayStore) ClaimDue(ctx context.Context, holderID string, batchSize int, lease time.Duration, now time.Time) ([]ApprovalActionReplayRow, error) {
	s.mu.Lock()
	s.count++
	n := s.count
	s.mu.Unlock()
	select {
	case s.calls <- n:
	default:
	}
	if n == 1 {
		panic("claim due panic")
	}
	return nil, nil
}

func (s *panicOnceApprovalReplayStore) MarkProcessed(ctx context.Context, id int64, holderID string, now time.Time) error {
	panic("panicOnceApprovalReplayStore.MarkProcessed: not implemented")
}

func (s *panicOnceApprovalReplayStore) ScheduleRetry(ctx context.Context, id int64, holderID string, attempts int, nextAttemptAt time.Time, lastErr string, now time.Time) error {
	panic("panicOnceApprovalReplayStore.ScheduleRetry: not implemented")
}

func (s *panicOnceApprovalReplayStore) MarkFailed(ctx context.Context, id int64, holderID string, attempts int, lastErr string, now time.Time) error {
	panic("panicOnceApprovalReplayStore.MarkFailed: not implemented")
}

func (s *panicOnceApprovalReplayStore) Counts(ctx context.Context) (ApprovalActionReplayCounts, error) {
	panic("panicOnceApprovalReplayStore.Counts: not implemented")
}

type replayPersistenceStub struct {
	records   map[string]models.SObject
	updateErr error
	updates   []replayUpdateCall
}

type replayUpdateCall struct {
	objectName string
	id         string
	updates    models.SObject
}

func (s *replayPersistenceStub) Insert(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
	panic("replayPersistenceStub.Insert: not implemented")
}

func (s *replayPersistenceStub) PreviewBeforeInsert(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
	panic("replayPersistenceStub.PreviewBeforeInsert: not implemented")
}

func (s *replayPersistenceStub) PrelockUpdateParents(ctx context.Context, tableName string, id string, updates models.SObject, user *models.UserSession) error {
	return nil
}

func (s *replayPersistenceStub) Update(ctx context.Context, tableName string, id string, updates models.SObject, user *models.UserSession) error {
	if s.updateErr != nil {
		return s.updateErr
	}
	key := tableName + "/" + id
	record := s.records[key]
	if record == nil {
		record = models.SObject{"Id": id}
		s.records[key] = record
	}
	for k, v := range updates {
		record[k] = v
	}
	s.updates = append(s.updates, replayUpdateCall{objectName: tableName, id: id, updates: cloneReplaySObject(updates)})
	return nil
}

func (s *replayPersistenceStub) PreviewBeforeUpdate(ctx context.Context, tableName string, id string, updates models.SObject, internalStamps models.SObject, user *models.UserSession) (*models.PreparedRecordUpdate, error) {
	panic("replayPersistenceStub.PreviewBeforeUpdate: not implemented")
}

func (s *replayPersistenceStub) CommitPreparedUpdate(ctx context.Context, tableName string, id string, prepared *models.PreparedRecordUpdate, extraUpdates models.SObject, user *models.UserSession) error {
	panic("replayPersistenceStub.CommitPreparedUpdate: not implemented")
}

func (s *replayPersistenceStub) Delete(ctx context.Context, tableName string, id string, user *models.UserSession) error {
	panic("replayPersistenceStub.Delete: not implemented")
}

func (s *replayPersistenceStub) GetRaw(ctx context.Context, objectName string, id string) (models.SObject, error) {
	key := objectName + "/" + id
	record := s.records[key]
	if record == nil {
		return nil, fmt.Errorf("record %s not found", key)
	}
	return cloneReplaySObject(record), nil
}

func (s *replayPersistenceStub) ReadSelectedSystemRecord(ctx context.Context, objectName, id string, fields []string) (models.SObject, error) {
	record, err := s.GetRaw(ctx, objectName, id)
	if err != nil || record == nil {
		return record, err
	}
	selected := make(models.SObject, len(fields))
	for _, field := range fields {
		if value, ok := record[field]; ok {
			selected[field] = value
		}
	}
	return selected, nil
}

func (s *replayPersistenceStub) GetLock(ctx context.Context, tableName string, id string) (models.SObject, error) {
	panic("replayPersistenceStub.GetLock: not implemented")
}

func (s *replayPersistenceStub) GetRecordType(ctx context.Context, id string) (string, error) {
	panic("replayPersistenceStub.GetRecordType: not implemented")
}

func (s *replayPersistenceStub) RunInTransaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}

func (s *replayPersistenceStub) IsTransactionActive(context.Context) bool { return false }

func (s *replayPersistenceStub) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	return fn(ctx)
}

func (s *replayPersistenceStub) DetachTx(ctx context.Context) context.Context {
	return ctx
}

func (s *replayPersistenceStub) RecomputeActiveFormulaProjection(_ context.Context, _ *models.UserSession, _, _ string, _ []string) (models.SObject, error) {
	return nil, nil
}

type memoryApprovalReplayStore struct {
	t    *testing.T
	rows map[int64]*memoryApprovalReplayRow
	next int64
}

type memoryApprovalReplayRow struct {
	id             int64
	lifecycleEvent string
	payload        []byte
	status         string
	attempts       int
	nextAttemptAt  time.Time
	leaseHolderID  string
	leaseExpires   time.Time
	lastErr        string
	processedAt    *time.Time
	failedAt       *time.Time
	alertedAt      *time.Time
}

func newMemoryApprovalReplayStore(t *testing.T) *memoryApprovalReplayStore {
	t.Helper()
	return &memoryApprovalReplayStore{t: t, rows: make(map[int64]*memoryApprovalReplayRow)}
}

func (s *memoryApprovalReplayStore) addPending(t *testing.T, payload ApprovalActionReplayPayload, due time.Time) int64 {
	t.Helper()
	body, err := json.Marshal(payload)
	if err != nil {
		t.Fatalf("marshal payload: %v", err)
	}
	s.next++
	s.rows[s.next] = &memoryApprovalReplayRow{
		id:             s.next,
		lifecycleEvent: payload.LifecycleEvent,
		payload:        body,
		status:         ApprovalActionReplayStatusPending,
		nextAttemptAt:  due,
	}
	return s.next
}

func (s *memoryApprovalReplayStore) EnqueueRecallFieldUpdates(ctx context.Context, payload ApprovalActionReplayPayload) (int64, error) {
	return s.addPending(s.t, payload, time.Now().UTC()), nil
}

func (s *memoryApprovalReplayStore) ClaimDue(ctx context.Context, holderID string, batchSize int, lease time.Duration, now time.Time) ([]ApprovalActionReplayRow, error) {
	var out []ApprovalActionReplayRow
	ids := make([]int64, 0, len(s.rows))
	for id := range s.rows {
		ids = append(ids, id)
	}
	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
	for _, id := range ids {
		if len(out) >= batchSize {
			break
		}
		row := s.rows[id]
		duePending := row.status == ApprovalActionReplayStatusPending && !row.nextAttemptAt.After(now)
		staleProcessing := row.status == ApprovalActionReplayStatusProcessing && row.leaseExpires.Before(now)
		if !duePending && !staleProcessing {
			continue
		}
		row.status = ApprovalActionReplayStatusProcessing
		row.leaseHolderID = holderID
		row.leaseExpires = now.Add(lease)
		out = append(out, ApprovalActionReplayRow{
			ID:             row.id,
			LifecycleEvent: row.lifecycleEvent,
			Payload:        append([]byte(nil), row.payload...),
			Attempts:       row.attempts,
		})
	}
	return out, nil
}

func (s *memoryApprovalReplayStore) MarkProcessed(ctx context.Context, id int64, holderID string, now time.Time) error {
	row := s.mustLeased(id, holderID)
	row.status = ApprovalActionReplayStatusProcessed
	row.leaseHolderID = ""
	row.leaseExpires = time.Time{}
	t := now.UTC()
	row.processedAt = &t
	return nil
}

func (s *memoryApprovalReplayStore) ScheduleRetry(ctx context.Context, id int64, holderID string, attempts int, nextAttemptAt time.Time, lastErr string, now time.Time) error {
	row := s.mustLeased(id, holderID)
	row.status = ApprovalActionReplayStatusPending
	row.attempts = attempts
	row.nextAttemptAt = nextAttemptAt.UTC()
	row.lastErr = lastErr
	row.leaseHolderID = ""
	row.leaseExpires = time.Time{}
	return nil
}

func (s *memoryApprovalReplayStore) MarkFailed(ctx context.Context, id int64, holderID string, attempts int, lastErr string, now time.Time) error {
	row := s.mustLeased(id, holderID)
	row.status = ApprovalActionReplayStatusFailed
	row.attempts = attempts
	row.lastErr = lastErr
	t := now.UTC()
	row.failedAt = &t
	row.alertedAt = &t
	row.leaseHolderID = ""
	row.leaseExpires = time.Time{}
	return nil
}

func (s *memoryApprovalReplayStore) Counts(ctx context.Context) (ApprovalActionReplayCounts, error) {
	var counts ApprovalActionReplayCounts
	for _, row := range s.rows {
		switch row.status {
		case ApprovalActionReplayStatusPending:
			counts.Pending++
		case ApprovalActionReplayStatusProcessing:
			counts.Processing++
		case ApprovalActionReplayStatusProcessed:
			counts.Processed++
		case ApprovalActionReplayStatusFailed:
			counts.Failed++
		}
	}
	return counts, nil
}

func (s *memoryApprovalReplayStore) mustLeased(id int64, holderID string) *memoryApprovalReplayRow {
	row := s.rows[id]
	if row == nil {
		s.t.Fatalf("row %d not found", id)
	}
	if row.status != ApprovalActionReplayStatusProcessing || row.leaseHolderID != holderID {
		s.t.Fatalf("row %d lease = status %q holder %q, want processing/%s", id, row.status, row.leaseHolderID, holderID)
	}
	return row
}

func cloneReplaySObject(in models.SObject) models.SObject {
	out := make(models.SObject, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}