package audit

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

type fakeOutboxStore struct {
	readyErr     error
	acquireErr   error
	acquireQueue [][]OutboxLease
	acquireCalls int
	commitErr    error
	bumpErr      error
	poisonErr    error
	commitCalls  int
	bumpCalls    int
	poisonCalls  int
	lastRows     []FieldRow
	lastClass    PoisonClass
	queueState   OutboxQueueState
	observeCalls int
}

func (f *fakeOutboxStore) CheckReady(context.Context) error { return f.readyErr }
func (f *fakeOutboxStore) AcquireLease(context.Context, string, int, time.Duration, int) ([]OutboxLease, error) {
	f.acquireCalls++
	if f.acquireErr != nil {
		return nil, f.acquireErr
	}
	if len(f.acquireQueue) == 0 {
		return nil, nil
	}
	leases := f.acquireQueue[0]
	f.acquireQueue = f.acquireQueue[1:]
	return leases, nil
}

func TestOutboxWorkerStopsWhenDeploymentOwnershipIsLost(t *testing.T) {
	lease := testLease()
	lease.Row.Payload = payloadForTest(t)
	store := &fakeOutboxStore{acquireQueue: [][]OutboxLease{{lease}}, commitErr: ErrOutboxWorkOwnershipLost}
	worker := NewOutboxWorker(store, OutboxWorkerConfig{HolderID: lease.HolderID})
	if err := worker.Prepare(context.Background()); err != nil {
		t.Fatalf("Prepare: %v", err)
	}
	if err := worker.Run(context.Background()); !errors.Is(err, ErrOutboxWorkOwnershipLost) {
		t.Fatalf("Run error=%v, want work ownership lost", err)
	}
	if store.bumpCalls != 0 || store.poisonCalls != 0 || store.commitCalls != 1 {
		t.Fatalf("stale deployment performed disposition: bump=%d poison=%d commit=%d", store.bumpCalls, store.poisonCalls, store.commitCalls)
	}
}
func (f *fakeOutboxStore) CommitFanout(_ context.Context, _ OutboxLease, rows []FieldRow) error {
	f.commitCalls++
	f.lastRows = rows
	return f.commitErr
}
func (f *fakeOutboxStore) BumpAttemptsAfterRollback(context.Context, OutboxLease, int, string) (bool, error) {
	f.bumpCalls++
	return false, f.bumpErr
}
func (f *fakeOutboxStore) MarkPoisonAfterRollback(_ context.Context, _ OutboxLease, class PoisonClass, _ string) error {
	f.poisonCalls++
	f.lastClass = class
	return f.poisonErr
}
func (f *fakeOutboxStore) RepairPoison(context.Context, OutboxRepair) error { return nil }
func (f *fakeOutboxStore) ObserveQueue(context.Context) (OutboxQueueState, error) {
	f.observeCalls++
	return f.queueState, nil
}

func TestOutboxWorkerPreparePreservesInitialQueueForRun(t *testing.T) {
	lease := testLease()
	lease.Row.Payload = payloadForTest(t)
	store := &fakeOutboxStore{
		acquireQueue: [][]OutboxLease{{lease}, nil},
		queueState:   OutboxQueueState{Depth: 3, PoisonDepth: 2, OldestAge: 2 * time.Second},
	}
	worker := NewOutboxWorker(store, OutboxWorkerConfig{
		HolderID:     lease.HolderID,
		StopWhenIdle: true,
		PollInterval: time.Millisecond,
	})
	if err := worker.Prepare(context.Background()); err != nil {
		t.Fatalf("Prepare: %v", err)
	}
	worker.Run(context.Background())
	if store.acquireCalls != 2 || store.commitCalls != 1 {
		t.Fatalf("calls acquire=%d commit=%d", store.acquireCalls, store.commitCalls)
	}
	if len(store.lastRows) != 1 || store.lastRows[0].Outcome != OutcomeSuccess {
		t.Fatalf("initial rows = %#v", store.lastRows)
	}
	snapshot := worker.Snapshot()
	if snapshot.Attempts != 1 || snapshot.Processed != 1 || snapshot.Failures != 0 || snapshot.InFlight != 0 {
		t.Fatalf("snapshot = %+v", snapshot)
	}
	if snapshot.PollChecks != 2 || snapshot.ConfirmedIdle != 1 || snapshot.ClaimAttempts != 2 || snapshot.EmptyClaims != 1 || snapshot.Claimed != 1 {
		t.Fatalf("work-loop snapshot = %+v", snapshot)
	}
	if store.observeCalls != 0 {
		t.Fatalf("business worker observation calls = %d, want zero", store.observeCalls)
	}
}

func TestOutboxWorkerClaimFailureRemainsObservableAndFailsPreparation(t *testing.T) {
	store := &fakeOutboxStore{acquireErr: errors.New("claim unavailable")}
	worker := NewOutboxWorker(store, OutboxWorkerConfig{HolderID: "worker-1"})
	if err := worker.Prepare(t.Context()); err == nil {
		t.Fatal("Prepare succeeded after claim failure")
	}
	snapshot := worker.Snapshot()
	if snapshot.PollChecks != 1 || snapshot.ClaimAttempts != 1 || snapshot.ConfirmedIdle != 0 || snapshot.EmptyClaims != 0 || snapshot.Claimed != 0 || snapshot.Failures != 1 {
		t.Fatalf("snapshot = %+v", snapshot)
	}
}

func TestOutboxWorkerKnownRollbackUsesSingleDisposition(t *testing.T) {
	lease := testLease()
	lease.Row.Payload = payloadForTest(t)
	store := &fakeOutboxStore{commitErr: &OutboxOperationError{
		CommitState: OutboxCommitRolledBack,
		Err:         errors.New("lock wait timeout"),
	}}
	worker := NewOutboxWorker(store, OutboxWorkerConfig{HolderID: lease.HolderID, MaxAttempts: 5})
	worker.processLease(context.Background(), lease)
	if store.bumpCalls != 1 || store.poisonCalls != 0 || store.commitCalls != 1 {
		t.Fatalf("calls bump=%d poison=%d commit=%d", store.bumpCalls, store.poisonCalls, store.commitCalls)
	}
}

func TestOutboxWorkerDeterministicRollbackPoisons(t *testing.T) {
	lease := testLease()
	lease.Row.Payload = payloadForTest(t)
	store := &fakeOutboxStore{commitErr: &OutboxOperationError{
		CommitState:   OutboxCommitRolledBack,
		Deterministic: true,
		PoisonClass:   PoisonClassFKViolation,
		Err:           errors.New("foreign key"),
	}}
	worker := NewOutboxWorker(store, OutboxWorkerConfig{HolderID: lease.HolderID})
	worker.processLease(context.Background(), lease)
	if store.poisonCalls != 1 || store.lastClass != PoisonClassFKViolation || store.bumpCalls != 0 {
		t.Fatalf("calls poison=%d class=%q bump=%d", store.poisonCalls, store.lastClass, store.bumpCalls)
	}
}

func TestOutboxWorkerNeverDisposesCommitUnknown(t *testing.T) {
	lease := testLease()
	lease.Row.Payload = payloadForTest(t)
	for _, tt := range []struct {
		name      string
		commitErr error
	}{
		{name: "commit unknown", commitErr: &OutboxOperationError{CommitState: OutboxCommitUnknown, Err: errors.New("bad connection")}},
	} {
		t.Run(tt.name, func(t *testing.T) {
			store := &fakeOutboxStore{commitErr: tt.commitErr}
			worker := NewOutboxWorker(store, OutboxWorkerConfig{HolderID: lease.HolderID})
			worker.processLease(context.Background(), lease)
			if store.bumpCalls != 0 || store.poisonCalls != 0 {
				t.Fatalf("commit-unknown path disposed: bump=%d poison=%d", store.bumpCalls, store.poisonCalls)
			}
		})
	}
}

func TestOutboxWorkerParseFailurePoisonsOnlyOutsideShutdown(t *testing.T) {
	lease := testLease()
	lease.Row.Payload = []byte("{")
	store := &fakeOutboxStore{}
	worker := NewOutboxWorker(store, OutboxWorkerConfig{HolderID: lease.HolderID})
	worker.processLease(context.Background(), lease)
	if store.poisonCalls != 1 || store.lastClass != PoisonClassParseFail {
		t.Fatalf("parse poison calls=%d class=%q", store.poisonCalls, store.lastClass)
	}

	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	store.poisonCalls = 0
	worker.processLease(ctx, lease)
	if store.poisonCalls != 0 {
		t.Fatal("shutdown context entered AfterRollback")
	}
}

func TestClassifyErrorNeverDirectlyPoisonsUnknownOrDriverErrors(t *testing.T) {
	if got := ClassifyError(errors.New("opaque driver failure")); got != PoisonClassNone {
		t.Fatalf("unknown classification = %q", got)
	}
}