package audit

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

	"github.com/go-sql-driver/mysql"
)

// fakeStore is an in-memory OutboxStore for unit tests. It implements the
// minimal set of operations the worker uses: lease-acquire, field-insert
// (with per-payload injectable error), finalize-success, mark-poison, bump-
// attempts. Thread-safe via a single mutex — deliberately simple; the
// correctness properties we assert (no double-process, P99 latency) are
// the worker's behavior under a well-behaved store, not the store itself.
type fakeStore struct {
	mu   sync.Mutex
	rows map[int64]*fakeRow

	// insertErr lets a test make the field-row fan-out fail N times.
	// Keyed by outbox row id; decrements on each call until zero, then returns nil.
	insertErr map[int64]func() error

	// inserted counts successful field-row inserts per outbox id (used to
	// detect double-process in concurrency test).
	inserted map[int64]int

	// latencies captured: EnqueuedAt -> ProcessedAt per row, set at finalize.
	processedAt map[int64]time.Time

	nextID int64
	now    func() time.Time
}

type fakeRow struct {
	id             int64
	payload        []byte
	enqueuedAt     time.Time
	processedAt    *time.Time
	attempts       int
	leaseHolder    string
	leaseExpiresAt *time.Time
	poisonClass    PoisonClass
	lastError      string
}

func newFakeStore() *fakeStore {
	return &fakeStore{
		rows:        map[int64]*fakeRow{},
		insertErr:   map[int64]func() error{},
		inserted:    map[int64]int{},
		processedAt: map[int64]time.Time{},
		now:         time.Now,
	}
}

func (s *fakeStore) enqueue(payload []byte) int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.nextID++
	id := s.nextID
	s.rows[id] = &fakeRow{
		id:         id,
		payload:    payload,
		enqueuedAt: s.now(),
	}
	return id
}

func (s *fakeStore) AcquireLease(ctx context.Context, holderID string, batchSize int, lease time.Duration, maxAttempts int) ([]OutboxRow, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	now := s.now()
	var pending []*fakeRow
	for _, r := range s.rows {
		if r.processedAt != nil {
			continue
		}
		if r.poisonClass != PoisonClassNone {
			// §3.3: idx_unprocessed excludes any non-NULL PoisonClass.
			continue
		}
		if r.leaseExpiresAt != nil && r.leaseExpiresAt.After(now) {
			continue
		}
		pending = append(pending, r)
	}
	sort.Slice(pending, func(i, j int) bool { return pending[i].enqueuedAt.Before(pending[j].enqueuedAt) })
	if len(pending) > batchSize {
		pending = pending[:batchSize]
	}
	exp := now.Add(lease)
	out := make([]OutboxRow, 0, len(pending))
	for _, r := range pending {
		r.leaseHolder = holderID
		r.leaseExpiresAt = &exp
		out = append(out, OutboxRow{
			ID:         r.id,
			Payload:    append([]byte(nil), r.payload...),
			EnqueuedAt: r.enqueuedAt,
			Attempts:   r.attempts,
		})
	}
	return out, nil
}

func (s *fakeStore) InsertFieldRows(ctx context.Context, outboxID int64, rows []FieldRow) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if f, ok := s.insertErr[outboxID]; ok {
		if err := f(); err != nil {
			return err
		}
	}
	s.inserted[outboxID] += len(rows)
	return nil
}

func (s *fakeStore) FinalizeSuccess(ctx context.Context, outboxID int64) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	r, ok := s.rows[outboxID]
	if !ok {
		return fmt.Errorf("row %d missing", outboxID)
	}
	now := s.now()
	r.processedAt = &now
	r.leaseHolder = ""
	r.leaseExpiresAt = nil
	s.processedAt[outboxID] = now
	return nil
}

func (s *fakeStore) MarkPoison(ctx context.Context, outboxID int64, class PoisonClass, lastErr string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	r, ok := s.rows[outboxID]
	if !ok {
		return fmt.Errorf("row %d missing", outboxID)
	}
	r.poisonClass = class
	r.lastError = lastErr
	r.leaseHolder = ""
	r.leaseExpiresAt = nil
	return nil
}

func (s *fakeStore) BumpAttempts(ctx context.Context, outboxID int64, lastErr string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	r, ok := s.rows[outboxID]
	if !ok {
		return fmt.Errorf("row %d missing", outboxID)
	}
	r.attempts++
	r.lastError = lastErr
	r.leaseHolder = ""
	r.leaseExpiresAt = nil
	return nil
}

func (s *fakeStore) snapshot(id int64) fakeRow {
	s.mu.Lock()
	defer s.mu.Unlock()
	return *s.rows[id]
}

// ---- test helpers -------------------------------------------------------

func enqueueEvent(s *fakeStore, fieldCount int) int64 {
	ev := AuditEvent{
		ObjectApiName: "Account",
		RecordId:      "001xx",
		Action:        ActionUpdate,
		ChangedById:   "005xx",
		Entrypoint:    EntrypointRestAPI,
		Outcome:       OutcomeSuccess,
	}
	for i := 0; i < fieldCount; i++ {
		ev.FieldDiffs = append(ev.FieldDiffs, FieldDiff{
			FieldName: fmt.Sprintf("Field%d", i),
			OldValue:  "old",
			NewValue:  "new",
		})
	}
	payload, _ := json.Marshal(ev)
	return s.enqueue(payload)
}

// ---- P99 <5s SLO test ---------------------------------------------------

func TestOutboxWorker_P99SLO_500Rows(t *testing.T) {
	store := newFakeStore()
	const N = 500
	ids := make([]int64, 0, N)
	for i := 0; i < N; i++ {
		ids = append(ids, enqueueEvent(store, 3))
	}
	cfg := OutboxWorkerConfig{
		BatchSize:     100,
		PollInterval:  10 * time.Millisecond, // tight for test
		LeaseDuration: 30 * time.Second,
		MaxAttempts:   5,
		HolderID:      "test-w1",
		StopWhenIdle:  true,
	}
	w := NewOutboxWorker(store, cfg)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	start := time.Now()
	w.Run(ctx)
	total := time.Since(start)

	// Collect per-row latencies.
	lats := make([]time.Duration, 0, N)
	for _, id := range ids {
		r := store.snapshot(id)
		if r.processedAt == nil {
			t.Fatalf("row %d not processed (total elapsed %v)", id, total)
		}
		lats = append(lats, r.processedAt.Sub(r.enqueuedAt))
	}
	sort.Slice(lats, func(i, j int) bool { return lats[i] < lats[j] })
	p99 := lats[int(float64(N)*0.99)-1]
	if p99 >= 5*time.Second {
		t.Fatalf("P99 latency %v exceeds 5s SLO (total elapsed %v)", p99, total)
	}
	t.Logf("P99=%v over %d rows in %v", p99, N, total)
}

// ---- concurrency test ---------------------------------------------------

func TestOutboxWorker_ThreeWorkers_NoDoubleProcess(t *testing.T) {
	store := newFakeStore()
	const N = 200
	ids := make([]int64, 0, N)
	for i := 0; i < N; i++ {
		ids = append(ids, enqueueEvent(store, 2))
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var wg sync.WaitGroup
	for i := 0; i < 3; i++ {
		wg.Add(1)
		holder := fmt.Sprintf("w%d", i)
		go func() {
			defer wg.Done()
			w := NewOutboxWorker(store, OutboxWorkerConfig{
				BatchSize:     50,
				PollInterval:  5 * time.Millisecond,
				LeaseDuration: 30 * time.Second,
				MaxAttempts:   5,
				HolderID:      holder,
				StopWhenIdle:  true,
			})
			w.Run(ctx)
		}()
	}
	wg.Wait()

	// All processed, and field-insert count per row == len(FieldDiffs) (= 2)
	// confirming each row processed exactly once (no double-fan-out).
	for _, id := range ids {
		r := store.snapshot(id)
		if r.processedAt == nil {
			t.Fatalf("row %d not processed", id)
		}
		store.mu.Lock()
		cnt := store.inserted[id]
		store.mu.Unlock()
		if cnt != 2 {
			t.Fatalf("row %d inserted %d times, want 2 (double-process?)", id, cnt)
		}
	}
	t.Logf("3 workers × %d rows, zero duplicates", N)
}

// ---- poison path test ---------------------------------------------------

func TestOutboxWorker_FKViolation_MarksPoison(t *testing.T) {
	store := newFakeStore()
	id := enqueueEvent(store, 1)

	// Fail InsertFieldRows once with an FK violation.
	calls := atomic.Int32{}
	store.insertErr[id] = func() error {
		calls.Add(1)
		return &mysql.MySQLError{Number: 1452, Message: "FK constraint fails"}
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	cfg := OutboxWorkerConfig{
		BatchSize: 10, PollInterval: 10 * time.Millisecond,
		LeaseDuration: 30 * time.Second, MaxAttempts: 5, HolderID: "w",
		StopWhenIdle: true,
	}
	NewOutboxWorker(store, cfg).Run(ctx)
	_ = calls.Load() // silence unused

	r := store.snapshot(id)
	if r.poisonClass != PoisonClassFKViolation {
		t.Fatalf("PoisonClass=%q, want %q", r.poisonClass, PoisonClassFKViolation)
	}
	if r.processedAt != nil {
		t.Fatalf("ProcessedAt must remain NULL on poison, got %v", r.processedAt)
	}
	if r.lastError == "" {
		t.Fatalf("LastError must be populated")
	}
}

// ---- retry exhaustion test ----------------------------------------------

func TestOutboxWorker_RetryExhaustion_MarksRetryExhausted(t *testing.T) {
	store := newFakeStore()
	id := enqueueEvent(store, 1)

	// Transient deadlock every attempt; worker should bump Attempts 5×
	// then assign RETRY_EXHAUSTED.
	store.insertErr[id] = func() error {
		return &mysql.MySQLError{Number: 1213, Message: "Deadlock"}
	}

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	cfg := OutboxWorkerConfig{
		BatchSize: 10, PollInterval: 5 * time.Millisecond,
		LeaseDuration: 30 * time.Second, MaxAttempts: 5, HolderID: "w",
		StopWhenIdle: true,
	}
	NewOutboxWorker(store, cfg).Run(ctx)

	r := store.snapshot(id)
	if r.attempts < 5 {
		t.Fatalf("Attempts=%d, want >=5", r.attempts)
	}
	if r.poisonClass != PoisonClassRetryExhausted {
		t.Fatalf("PoisonClass=%q, want %q", r.poisonClass, PoisonClassRetryExhausted)
	}
	if r.processedAt != nil {
		t.Fatalf("ProcessedAt must remain NULL on retry-exhausted")
	}
}

// ---- parse failure path -------------------------------------------------

func TestOutboxWorker_ParseFail_MarksPoison(t *testing.T) {
	store := newFakeStore()
	// Enqueue malformed JSON payload directly.
	id := store.enqueue([]byte(`{"not valid json`))

	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancel()
	cfg := OutboxWorkerConfig{
		BatchSize: 10, PollInterval: 5 * time.Millisecond,
		LeaseDuration: 30 * time.Second, MaxAttempts: 5, HolderID: "w",
		StopWhenIdle: true,
	}
	NewOutboxWorker(store, cfg).Run(ctx)

	r := store.snapshot(id)
	if r.poisonClass != PoisonClassParseFail {
		t.Fatalf("PoisonClass=%q, want %q", r.poisonClass, PoisonClassParseFail)
	}
}