package automation

// audit_test.go — hermetic unit tests for T51 (audit table + async writer) and
// T52 (element executor emit points + sampling control).
//
// All tests are hermetic: no real TiDB connection required. The inMemoryStore
// implements AuditStore using in-memory slices protected by a mutex.

import (
	"context"
	"math"
	"math/rand"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
)

// systemTableDefinitionsForTest returns the canonical system-table catalog from
// the kernel schemacatalog package. It replaces the former in-package bootstrap
// GetSystemTableDefinitions() call (itself a thin wrapper over
// schemacatalog.SystemTableDefinitions) so these tests depend only on kernel
// packages and can move with the automation package to its sibling module.
func systemTableDefinitionsForTest(t *testing.T) []schema.TableDefinition {
	t.Helper()
	defs, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("schemacatalog.SystemTableDefinitions(): %v", err)
	}
	return defs
}

// ---------------------------------------------------------------------------
// Test double: in-memory AuditStore
// ---------------------------------------------------------------------------

type inMemoryStore struct {
	mu   sync.Mutex
	rows []Row
}

func newInMemoryStore() *inMemoryStore { return &inMemoryStore{} }

func (s *inMemoryStore) InsertRows(_ context.Context, rows []Row) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.rows = append(s.rows, rows...)
	return nil
}

func (s *inMemoryStore) QueryRows(_ context.Context, filter QueryFilter) ([]Row, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var out []Row
	for _, r := range s.rows {
		if filter.FlowExecutionId != "" && r.FlowExecutionId != filter.FlowExecutionId {
			continue
		}
		if filter.ElementId != "" && r.ElementId != filter.ElementId {
			continue
		}
		if filter.Status != "" && r.Status != filter.Status {
			continue
		}
		out = append(out, r)
		if filter.Limit > 0 && len(out) >= filter.Limit {
			break
		}
	}
	return out, nil
}

func (s *inMemoryStore) DeleteBefore(_ context.Context, before time.Time) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var kept []Row
	var deleted int64
	for _, r := range s.rows {
		if r.CreatedDate.Before(before) {
			deleted++
		} else {
			kept = append(kept, r)
		}
	}
	s.rows = kept
	return deleted, nil
}

func (s *inMemoryStore) CountRows(_ context.Context) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return int64(len(s.rows)), nil
}

// DeleteOldest removes the n oldest rows by CreatedDate.
func (s *inMemoryStore) DeleteOldest(_ context.Context, n int64) (int64, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if n <= 0 || len(s.rows) == 0 {
		return 0, nil
	}
	// Simple: sort rows by CreatedDate ascending, remove first n.
	// Use insertion sort on small slices.
	rows := make([]Row, len(s.rows))
	copy(rows, s.rows)
	for i := 1; i < len(rows); i++ {
		for j := i; j > 0 && rows[j].CreatedDate.Before(rows[j-1].CreatedDate); j-- {
			rows[j], rows[j-1] = rows[j-1], rows[j]
		}
	}
	if n > int64(len(rows)) {
		n = int64(len(rows))
	}
	s.rows = rows[n:]
	return n, nil
}

func (s *inMemoryStore) Rows() []Row {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]Row, len(s.rows))
	copy(out, s.rows)
	return out
}

func (s *inMemoryStore) Count() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return len(s.rows)
}

// ---------------------------------------------------------------------------
// T51 AC#1 — _System_FlowAuditLog registered in GetSystemTableDefinitions
// ---------------------------------------------------------------------------

func TestFlowAuditTableRegistered(t *testing.T) {
	defs := systemTableDefinitionsForTest(t)
	var found bool
	for _, d := range defs {
		if d.TableName == "_System_FlowAuditLog" {
			found = true
			break
		}
	}
	if !found {
		t.Fatal("GetSystemTableDefinitions() does not contain '_System_FlowAuditLog'")
	}
}

// T51 AC#1 — exactly one entry
func TestFlowAuditTableRegisteredExactlyOnce(t *testing.T) {
	defs := systemTableDefinitionsForTest(t)
	count := 0
	for _, d := range defs {
		if d.TableName == "_System_FlowAuditLog" {
			count++
		}
	}
	if count != 1 {
		t.Fatalf("expected exactly 1 '_System_FlowAuditLog' entry in GetSystemTableDefinitions(), got %d", count)
	}
}

// ---------------------------------------------------------------------------
// T51 AC#3 — Strong-reading literal grep: no forbidden column/field names
// ---------------------------------------------------------------------------

func TestFlowAuditStrongReading_NoForbiddenColumnNames(t *testing.T) {
	defs := systemTableDefinitionsForTest(t)
	var tbl *struct{ Columns []interface{} }
	_ = tbl // find the table by name

	var auditTableCols interface{}
	for _, d := range defs {
		if d.TableName == "_System_FlowAuditLog" {
			auditTableCols = d.Columns
			break
		}
	}
	if auditTableCols == nil {
		t.Fatal("_System_FlowAuditLog not found in definitions")
	}

	// (a) explicit literal grep for forbidden column names in schema
	for _, d := range defs {
		if d.TableName != "_System_FlowAuditLog" {
			continue
		}
		for _, col := range d.Columns {
			name := strings.ToLower(col.Name)
			if strings.Contains(name, "objectapiname") || strings.Contains(name, "sobjectname") {
				t.Fatalf("forbidden column name %q in _System_FlowAuditLog (CLAUDE.md strong-reading)", col.Name)
			}
		}
	}

	// (b) explicit forbidden field check on Row struct via reflect
	typ := reflect.TypeOf(Row{})
	for i := 0; i < typ.NumField(); i++ {
		name := strings.ToLower(typ.Field(i).Name)
		if strings.Contains(name, "objectapiname") || strings.Contains(name, "sobjectname") {
			t.Fatalf("forbidden field name %q in audit.Row (CLAUDE.md strong-reading)", typ.Field(i).Name)
		}
	}
}

// ---------------------------------------------------------------------------
// T51 AC#2 — Async writer + flush on shutdown (partial batch)
// ---------------------------------------------------------------------------

func TestFlowAuditWriter_ShutdownFlushesPartialBatch(t *testing.T) {
	store := newInMemoryStore()
	// Use small buffer to confirm shutdown drain works regardless of buffer size.
	w := NewWriter(store, WithChannelBuffer(200), WithWorkerCount(1))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	const N = 50 // less than auditBatchSize (100)
	for i := 0; i < N; i++ {
		w.Append(Row{
			FlowExecutionId: "exec-001",
			ElementId:       "elem-001",
			ElementType:     "Decision",
			Status:          "success",
			DurationMs:      1,
		})
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown() error: %v", err)
	}

	if got := store.Count(); got != N {
		t.Fatalf("expected %d rows after Shutdown, got %d", N, got)
	}
}

// T51 AC#2 — Batch flush triggers automatically when batch is full
func TestFlowAuditWriter_BatchFlushOnFullBatch(t *testing.T) {
	store := newInMemoryStore()
	// Use small buffer; auditBatchSize=100 so send 110 rows to trigger a flush.
	w := NewWriter(store, WithChannelBuffer(512), WithWorkerCount(1))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	const N = auditBatchSize + 10 // >= batch size → triggers at least one auto-flush
	for i := 0; i < N; i++ {
		w.Append(Row{
			FlowExecutionId: "exec-batch",
			ElementId:       "elem-batch",
			ElementType:     "Assignment",
			Status:          "success",
			DurationMs:      1,
		})
	}

	// Give worker a moment to flush the completed batch, then shut down.
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown() error: %v", err)
	}

	if got := store.Count(); got != N {
		t.Fatalf("expected %d rows after full-batch test, got %d", N, got)
	}
}

func TestFlowAuditWriter_RecoversFlushPanicAndKeepsWorkerAlive(t *testing.T) {
	store := &panicOnceAuditStore{
		inMemoryStore: newInMemoryStore(),
		panicked:      make(chan struct{}),
	}
	w := NewWriter(store, WithChannelBuffer(auditBatchSize+2), WithWorkerCount(1))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	for i := 0; i < auditBatchSize; i++ {
		w.Append(Row{
			FlowExecutionId: "panic-batch",
			ElementId:       "first",
			ElementType:     "Assignment",
			Status:          "success",
		})
	}

	select {
	case <-store.panicked:
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for audit flush panic path")
	}

	w.Append(Row{
		FlowExecutionId: "after-panic",
		ElementId:       "second",
		ElementType:     "Assignment",
		Status:          "success",
	})

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown() error after recovered flush panic: %v", err)
	}

	rows := store.Rows()
	if len(rows) != 1 {
		t.Fatalf("stored rows after recovered panic = %d, want 1", len(rows))
	}
	if rows[0].FlowExecutionId != "after-panic" {
		t.Fatalf("stored row FlowExecutionId = %q, want after-panic", rows[0].FlowExecutionId)
	}
}

type panicOnceAuditStore struct {
	*inMemoryStore
	panicOnce sync.Once
	panicked  chan struct{}
}

func (s *panicOnceAuditStore) InsertRows(ctx context.Context, rows []Row) error {
	didPanic := false
	s.panicOnce.Do(func() {
		didPanic = true
		close(s.panicked)
	})
	if didPanic {
		panic("audit flush test panic")
	}
	return s.inMemoryStore.InsertRows(ctx, rows)
}

// ---------------------------------------------------------------------------
// T51 AC#4 — Backpressure: DroppedCount > 0 AND Append is non-blocking
// ---------------------------------------------------------------------------

func TestFlowAuditWriter_Backpressure(t *testing.T) {
	// blockingStore holds inserts until unblocked so the worker stays busy.
	type blockingStore struct {
		inMemoryStore
		block chan struct{}
	}
	bs := &blockingStore{block: make(chan struct{})}
	bs.block = make(chan struct{}) // open = blocked
	bsInsert := func(ctx context.Context, rows []Row) error {
		select {
		case <-bs.block: // unblocked
		case <-ctx.Done():
		}
		return bs.inMemoryStore.InsertRows(ctx, rows)
	}
	var storeWrapper fakeBlockingStore
	storeWrapper.insertFn = bsInsert
	storeWrapper.base = &bs.inMemoryStore

	// Buffer of 4 + 1 worker blocked on first insert → channel fills fast.
	w := NewWriter(&storeWrapper, WithChannelBuffer(4), WithWorkerCount(1))

	// Give the worker a moment to start and pick up the first item (blocking it).
	time.Sleep(20 * time.Millisecond)

	const numAppends = 100
	start := time.Now()
	for i := 0; i < numAppends; i++ {
		w.Append(Row{ElementType: "Decision", Status: "success"})
	}
	elapsed := time.Since(start)

	// (b) assert no Append call took > 50ms in aggregate (non-blocking).
	if elapsed > 50*time.Millisecond {
		t.Fatalf("Append loop took %s for %d calls; expected < 50ms (non-blocking)", elapsed, numAppends)
	}

	// Unblock the store.
	close(bs.block)

	// Shutdown drains remaining.
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = w.Shutdown(ctx)

	// (a) assert DroppedCount > 0.
	if w.DroppedCount() == 0 {
		t.Fatal("expected DroppedCount() > 0 when channel is saturated, got 0")
	}
}

// fakeBlockingStore wraps inMemoryStore with an injectable InsertRows func.
type fakeBlockingStore struct {
	base     *inMemoryStore
	insertFn func(ctx context.Context, rows []Row) error
}

func (f *fakeBlockingStore) InsertRows(ctx context.Context, rows []Row) error {
	return f.insertFn(ctx, rows)
}
func (f *fakeBlockingStore) QueryRows(ctx context.Context, filter QueryFilter) ([]Row, error) {
	return f.base.QueryRows(ctx, filter)
}
func (f *fakeBlockingStore) DeleteBefore(ctx context.Context, before time.Time) (int64, error) {
	return f.base.DeleteBefore(ctx, before)
}
func (f *fakeBlockingStore) CountRows(ctx context.Context) (int64, error) {
	return f.base.CountRows(ctx)
}
func (f *fakeBlockingStore) DeleteOldest(ctx context.Context, n int64) (int64, error) {
	return f.base.DeleteOldest(ctx, n)
}

// ---------------------------------------------------------------------------
// T51 AC#5 — Retention: row-count-cap and max-age modes
// ---------------------------------------------------------------------------

func TestFlowAuditRetention_RowCountCap(t *testing.T) {
	store := newInMemoryStore()

	// Insert 100 rows with distinct timestamps.
	base := time.Now().Add(-1 * time.Hour)
	for i := 0; i < 100; i++ {
		store.InsertRows(context.Background(), []Row{{
			Id:          "r" + string(rune('A'+i%26)) + string(rune('0'+i/26)),
			CreatedDate: base.Add(time.Duration(i) * time.Second),
			Status:      "success",
		}})
	}

	cfg := RetentionConfig{MaxRows: 20}
	ret := NewRetention(store, cfg)
	n, err := ret.Run(context.Background())
	if err != nil {
		t.Fatalf("Run() error: %v", err)
	}
	if n == 0 {
		t.Fatal("expected deleted count > 0")
	}

	remaining := store.Count()
	if remaining > 20 {
		t.Fatalf("expected <= 20 rows after retention, got %d", remaining)
	}
}

func TestFlowAuditRetention_MaxAge(t *testing.T) {
	store := newInMemoryStore()

	// Insert 10 rows in the past (older than 1ms).
	old := time.Now().Add(-10 * time.Millisecond)
	for i := 0; i < 10; i++ {
		store.InsertRows(context.Background(), []Row{{
			Id:          "old" + string(rune('0'+i)),
			CreatedDate: old,
			Status:      "success",
		}})
	}
	// Insert 5 fresh rows.
	now := time.Now().Add(10 * time.Second) // future-dated so they survive
	for i := 0; i < 5; i++ {
		store.InsertRows(context.Background(), []Row{{
			Id:          "new" + string(rune('0'+i)),
			CreatedDate: now,
			Status:      "success",
		}})
	}

	// Sleep a tiny bit so "now" for MaxAge calculation is definitely past `old`.
	time.Sleep(2 * time.Millisecond)

	cfg := RetentionConfig{MaxAge: 1 * time.Millisecond}
	ret := NewRetention(store, cfg)
	_, err := ret.Run(context.Background())
	if err != nil {
		t.Fatalf("Run() error: %v", err)
	}

	remaining := store.Count()
	if remaining != 5 {
		t.Fatalf("expected 5 rows after age-based retention, got %d", remaining)
	}
}

// ---------------------------------------------------------------------------
// T51 AC#6 — Write API: Append is non-blocking; Query returns matching rows
// ---------------------------------------------------------------------------

func TestFlowAuditWriter_AppendNonBlocking(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store)
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	// Append must return immediately (non-blocking even on full channel).
	start := time.Now()
	w.Append(Row{FlowExecutionId: "e1", Status: "success"})
	if elapsed := time.Since(start); elapsed > 5*time.Millisecond {
		t.Fatalf("Append() took %s; expected < 5ms (must be non-blocking)", elapsed)
	}
}

func TestFlowAuditWriter_QueryReturnsMatchingRows(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store)
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	rows := []Row{
		{FlowExecutionId: "exec-q1", ElementId: "e1", Status: "success"},
		{FlowExecutionId: "exec-q1", ElementId: "e2", Status: "error"},
		{FlowExecutionId: "exec-q2", ElementId: "e1", Status: "success"},
	}
	for _, r := range rows {
		w.Append(r)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown() error: %v", err)
	}

	got, err := w.Query(context.Background(), QueryFilter{FlowExecutionId: "exec-q1"})
	if err != nil {
		t.Fatalf("Query() error: %v", err)
	}
	if len(got) != 2 {
		t.Fatalf("Query(exec-q1) returned %d rows, want 2", len(got))
	}
}

// ---------------------------------------------------------------------------
// T52 AC#2 — Rate = 1.0 → all 50 rows emitted
// ---------------------------------------------------------------------------

func TestFlowAuditSampling_Rate1_0_AllEmitted(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(42))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	const N = 50
	for i := 0; i < N; i++ {
		w.MaybeEmit(Row{FlowExecutionId: "rate1", Status: "success"}, 1.0)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	if got := store.Count(); got != N {
		t.Fatalf("rate=1.0: expected %d rows, got %d", N, got)
	}
}

// ---------------------------------------------------------------------------
// T52 AC#3 — Rate = 0.0 → no rows emitted
// ---------------------------------------------------------------------------

func TestFlowAuditSampling_Rate0_0_NoneEmitted(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(42))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	const N = 50
	for i := 0; i < N; i++ {
		w.MaybeEmit(Row{FlowExecutionId: "rate0", Status: "success"}, 0.0)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	if got := store.Count(); got != 0 {
		t.Fatalf("rate=0.0: expected 0 rows, got %d", got)
	}
}

// ---------------------------------------------------------------------------
// T52 AC#4 — Rate = 0.5 statistical (N=1000, seed=42)
// ---------------------------------------------------------------------------

func TestFlowAuditSampling_Rate0_5_Statistical(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(42))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	const N = 1000
	for i := 0; i < N; i++ {
		w.MaybeEmit(Row{FlowExecutionId: "stat", Status: "success"}, 0.5)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	emittedCount := store.Count()
	skippedCount := N - emittedCount
	actualRate := float64(emittedCount) / float64(N)

	// (4a) statistical assertion — actual rate within tolerance
	if math.Abs(actualRate-0.5) >= 0.05 {
		t.Fatalf("rate=0.5 statistical: actualRate=%.4f is not within ±0.05 of 0.5", actualRate)
	}

	// (4b) non-degenerate — both emitted and skipped > 0
	if emittedCount <= 0 {
		t.Fatalf("rate=0.5: no rows emitted (emittedCount=%d)", emittedCount)
	}
	if skippedCount <= 0 {
		t.Fatalf("rate=0.5: no rows skipped (skippedCount=%d)", skippedCount)
	}
}

// ---------------------------------------------------------------------------
// T52 AC#5 — DefaultAuditSampleRate == 1.0 (every element completion logged)
// ---------------------------------------------------------------------------

func TestFlowAuditDefaultSampleRate(t *testing.T) {
	// Assert the exported constant value. The audit log contract is "log every
	// element completion"; a sub-1.0 default would silently drop most rows
	// (Lane C round-3 finding: rollup pillar5-audit emit hook contract).
	if DefaultAuditSampleRate != 1.0 {
		t.Fatalf("DefaultAuditSampleRate = %v, want 1.0", DefaultAuditSampleRate)
	}
}

// TestCapPayload_TextColumnBoundary guards Lane C round-3 r2 Pass B Finding 2:
// MySQL/TiDB TEXT columns max out at 65,535 bytes, NOT 64*1024 = 65,536. The
// previous cap value let a 65,536-byte payload through, which then failed the
// INSERT batch with "data too long", losing up to auditBatchSize neighbouring
// rows. Verify the boundary explicitly.
func TestCapPayload_TextColumnBoundary(t *testing.T) {
	// At the boundary: a 65,535-byte payload is the largest TEXT can hold and
	// must pass capPayload unchanged.
	atLimit := strings.Repeat("a", 65535)
	if got := capPayload(atLimit); len(got) != 65535 {
		t.Fatalf("capPayload(65535-byte input) = %d bytes, want 65535", len(got))
	}

	// One byte over the boundary: must be truncated to exactly 65,535 so the
	// INSERT does not blow up.
	overLimit := strings.Repeat("a", 65536)
	if got := capPayload(overLimit); len(got) != 65535 {
		t.Fatalf("capPayload(65536-byte input) = %d bytes, want 65535 (TEXT max)", len(got))
	}

	// A small payload is unchanged.
	if got := capPayload("ok"); got != "ok" {
		t.Fatalf("capPayload(\"ok\") = %q, want \"ok\"", got)
	}
}

func TestFlowAuditSampleRateFromCtx_FallsBackToDefault(t *testing.T) {
	ctx := context.Background() // no rate stored → should return default
	rate := auditSampleRateFromCtx(ctx)
	if rate != DefaultAuditSampleRate {
		t.Fatalf("auditSampleRateFromCtx(empty ctx) = %v, want %v (DefaultAuditSampleRate)", rate, DefaultAuditSampleRate)
	}
}

func TestFlowAuditSampleRateFromCtx_UsesStoredRate(t *testing.T) {
	ctx := withAuditSampleRate(context.Background(), 0.75)
	rate := auditSampleRateFromCtx(ctx)
	if rate != 0.75 {
		t.Fatalf("auditSampleRateFromCtx() = %v, want 0.75", rate)
	}
}

func TestFlowAuditSampleRateFromCtx_ZeroFallsBackToDefault(t *testing.T) {
	// A zero-valued rate in ctx should still return DefaultAuditSampleRate.
	ctx := withAuditSampleRate(context.Background(), 0.0)
	rate := auditSampleRateFromCtx(ctx)
	if rate != DefaultAuditSampleRate {
		t.Fatalf("auditSampleRateFromCtx(zero) = %v, want %v (default)", rate, DefaultAuditSampleRate)
	}
}

// ---------------------------------------------------------------------------
// T52 AC#1 — Each executor emits exactly once (success + error)
// Tested via MaybeEmit with deterministic rate=1.0 and a fake store.
// ---------------------------------------------------------------------------

// TestFlowAuditEmit_SuccessPath confirms a single success row is produced when
// a MaybeEmit call with rate=1.0 is used in the defer pattern.
func TestFlowAuditEmit_SuccessPath(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(1))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	// Simulate a defer-pattern emit on success path.
	var execErr error
	start := time.Now()
	status := "success"
	if execErr != nil {
		status = "error"
	}
	w.MaybeEmit(Row{
		FlowExecutionId: "exec-s1",
		ElementId:       "dec-001",
		ElementType:     "Decision",
		Status:          status,
		DurationMs:      int(time.Since(start).Milliseconds()),
	}, 1.0)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	rows := store.Rows()
	if len(rows) != 1 {
		t.Fatalf("success path: expected 1 row, got %d", len(rows))
	}
	if rows[0].Status != "success" {
		t.Fatalf("success path: expected Status=success, got %q", rows[0].Status)
	}
}

// TestFlowAuditEmit_ErrorPath confirms a single error row is produced.
func TestFlowAuditEmit_ErrorPath(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(1))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	// Simulate a defer-pattern emit on error path.
	execErr := context.DeadlineExceeded // non-nil = error
	start := time.Now()
	status := "success"
	if execErr != nil {
		status = "error"
	}
	w.MaybeEmit(Row{
		FlowExecutionId: "exec-e1",
		ElementId:       "dec-001",
		ElementType:     "Decision",
		Status:          status,
		DurationMs:      int(time.Since(start).Milliseconds()),
		ErrorPayload:    execErr.Error(),
	}, 1.0)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	rows := store.Rows()
	if len(rows) != 1 {
		t.Fatalf("error path: expected 1 row, got %d", len(rows))
	}
	if rows[0].Status != "error" {
		t.Fatalf("error path: expected Status=error, got %q", rows[0].Status)
	}
}

// ---------------------------------------------------------------------------
// Additional: NilWriter safety (all methods are nil-safe)
// ---------------------------------------------------------------------------

func TestFlowAuditWriter_NilSafety(t *testing.T) {
	var w *Writer
	w.Append(Row{})
	w.MaybeEmit(Row{}, 1.0)
	if w.ShouldSample(1.0) {
		t.Fatal("nil Writer.ShouldSample should return false")
	}
	if w.DroppedCount() != 0 {
		t.Fatal("nil Writer.DroppedCount should return 0")
	}
	rows, err := w.Query(context.Background(), QueryFilter{})
	if err != nil || rows != nil {
		t.Fatal("nil Writer.Query should return nil, nil")
	}
	if err := w.Shutdown(context.Background()); err != nil {
		t.Fatalf("nil Writer.Shutdown should return nil, got %v", err)
	}
}

// ---------------------------------------------------------------------------
// Finding 2 — FlowDefinitionId and FlowVersionId populated on emit
// ---------------------------------------------------------------------------

func TestFlowAuditRow_ScopeFieldsPopulated(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(1))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	w.MaybeEmit(Row{
		FlowExecutionId:  "exec-scope",
		FlowDefinitionId: "def-001",
		FlowVersionId:    "ver-001",
		ElementId:        "elem-001",
		ElementType:      "Decision",
		Status:           "success",
		DurationMs:       5,
	}, 1.0)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	rows := store.Rows()
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
	if rows[0].FlowDefinitionId != "def-001" {
		t.Errorf("FlowDefinitionId = %q, want %q", rows[0].FlowDefinitionId, "def-001")
	}
	if rows[0].FlowVersionId != "ver-001" {
		t.Errorf("FlowVersionId = %q, want %q", rows[0].FlowVersionId, "ver-001")
	}
}

func TestFlowAuditRow_ErrorPayloadPopulatedOnError(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(1))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	w.MaybeEmit(Row{
		FlowExecutionId: "exec-err",
		FlowVersionId:   "ver-002",
		ElementId:       "elem-002",
		ElementType:     "RecordCreate",
		Status:          "error",
		DurationMs:      2,
		ErrorPayload:    "context deadline exceeded",
	}, 1.0)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(ctx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	rows := store.Rows()
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
	if rows[0].ErrorPayload == "" {
		t.Error("ErrorPayload must not be empty on error path")
	}
	if rows[0].FlowVersionId != "ver-002" {
		t.Errorf("FlowVersionId = %q, want %q", rows[0].FlowVersionId, "ver-002")
	}
}

// ---------------------------------------------------------------------------
// Finding 3 — NewRetentionFromTableDefinition reads table-level config
// ---------------------------------------------------------------------------

func TestRetentionFromTableDefinition_ReadsRetentionConfig(t *testing.T) {
	store := newInMemoryStore()
	ret, err := NewRetentionFromTableDefinition("_System_FlowAuditLog", store)
	if err != nil {
		t.Fatalf("NewRetentionFromTableDefinition: %v", err)
	}
	if ret == nil {
		t.Fatal("expected non-nil Retention")
	}
	// Assert the config was read from the JSON (retentionMaxRows: 1000000)
	if ret.config.MaxRows != 1000000 {
		t.Errorf("MaxRows = %d, want 1000000", ret.config.MaxRows)
	}
	// retentionMaxAgeSeconds: 7776000 → 90 days
	if ret.config.MaxAge == 0 {
		t.Error("MaxAge must be non-zero (retentionMaxAgeSeconds declared in system_tables.json)")
	}
}

func TestRetentionFromTableDefinition_UnknownTable(t *testing.T) {
	store := newInMemoryStore()
	_, err := NewRetentionFromTableDefinition("_System_NonExistent", store)
	if err == nil {
		t.Fatal("expected error for unknown table, got nil")
	}
}

// ---------------------------------------------------------------------------
// Finding 4 — Concurrency: Append + ShouldSample + Shutdown race-safe
// ---------------------------------------------------------------------------

func TestFlowAuditConcurrency(t *testing.T) {
	store := newInMemoryStore()
	// Use default (production-like) Writer — no injected rng so ShouldSample uses shared rng.
	w := NewWriter(store, WithChannelBuffer(512), WithWorkerCount(2))

	const goroutines = 50
	var wg sync.WaitGroup
	wg.Add(goroutines)

	panicCh := make(chan interface{}, goroutines)

	for i := 0; i < goroutines; i++ {
		go func(id int) {
			defer wg.Done()
			defer func() {
				if r := recover(); r != nil {
					panicCh <- r
				}
			}()
			for j := 0; j < 20; j++ {
				w.Append(Row{
					FlowExecutionId: "exec-concurrent",
					ElementId:       "e1",
					ElementType:     "Decision",
					Status:          "success",
				})
				// ShouldSample must not race-detect on w.rng
				_ = w.ShouldSample(0.5)
			}
		}(i)
	}

	// Shutdown from a separate goroutine while appends are still running.
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	}()

	wg.Wait()
	close(panicCh)

	for p := range panicCh {
		t.Fatalf("goroutine panicked: %v", p)
	}

	// Drain any remaining items (Shutdown may have already closed the channel).
	shutCtx, shutCancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer shutCancel()
	_ = w.Shutdown(shutCtx) // idempotent — safe to call again
}

// ---------------------------------------------------------------------------
// Round-2 Finding 1 — FlowDefinitionId threads from ctx into emitted Row
// ---------------------------------------------------------------------------

// TestFlowAudit_FlowDefinitionIdPopulated mirrors the executor defer pattern:
// each element executor reads auditFlowDefinitionIdFromCtx(ctx) and stamps it
// onto the emitted Row. This guards against the regression where the column
// shipped with no writer (always empty in production).
func TestFlowAudit_FlowDefinitionIdPopulated(t *testing.T) {
	store := newInMemoryStore()
	w := NewWriter(store, WithRand(rand.New(rand.NewSource(1))))
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = w.Shutdown(ctx)
	})

	// Simulate the executor entry point threading the FlowDefinitionId on ctx.
	ctx := withAuditFlowDefinitionId(context.Background(), "def-flow-001")

	// Defer-emit pattern equivalent to executor_*.go.
	w.MaybeEmit(Row{
		FlowExecutionId:  "exec-defid",
		FlowDefinitionId: auditFlowDefinitionIdFromCtx(ctx),
		FlowVersionId:    "ver-defid",
		ElementId:        "elem-defid",
		ElementType:      "Decision",
		Status:           "success",
		DurationMs:       1,
	}, 1.0)

	shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := w.Shutdown(shutCtx); err != nil {
		t.Fatalf("Shutdown(): %v", err)
	}

	rows := store.Rows()
	if len(rows) != 1 {
		t.Fatalf("expected 1 row, got %d", len(rows))
	}
	if rows[0].FlowDefinitionId != "def-flow-001" {
		t.Errorf("FlowDefinitionId = %q, want %q", rows[0].FlowDefinitionId, "def-flow-001")
	}
}

// TestFlowAudit_FlowDefinitionIdAbsentWhenCtxMissing confirms the helper
// returns an empty string when the ctx key is unset, matching the
// pre-wiring fallback behaviour.
func TestFlowAudit_FlowDefinitionIdAbsentWhenCtxMissing(t *testing.T) {
	if got := auditFlowDefinitionIdFromCtx(context.Background()); got != "" {
		t.Errorf("auditFlowDefinitionIdFromCtx(empty ctx) = %q, want empty", got)
	}
}

// ---------------------------------------------------------------------------
// Round-2 Finding 2 — resolveAuditSampleRate clamps + falls back as documented
// ---------------------------------------------------------------------------

func TestResolveAuditSampleRate(t *testing.T) {
	cases := []struct {
		name string
		in   float64
		want float64
	}{
		{"zero falls back to default", 0, DefaultAuditSampleRate},
		{"negative falls back to default", -0.1, DefaultAuditSampleRate},
		{"NaN falls back to default", math.NaN(), DefaultAuditSampleRate},
		{"in-range passes through", 0.25, 0.25},
		{"one passes through", 1.0, 1.0},
		{"above-one clamps to 1", 2.5, 1.0},
	}
	for _, tc := range cases {
		got := resolveAuditSampleRate(tc.in)
		// NaN comparison: any non-default would fail the != check below.
		if math.IsNaN(tc.in) && got != DefaultAuditSampleRate {
			t.Errorf("%s: got %v, want %v", tc.name, got, DefaultAuditSampleRate)
			continue
		}
		if !math.IsNaN(tc.in) && got != tc.want {
			t.Errorf("%s: resolveAuditSampleRate(%v) = %v, want %v", tc.name, tc.in, got, tc.want)
		}
	}
}

// ---------------------------------------------------------------------------
// Additional: containsForbiddenAuditName helper covers both (a) and (b) logic
// ---------------------------------------------------------------------------

func TestContainsForbiddenAuditName(t *testing.T) {
	cases := []struct {
		name string
		want bool
	}{
		{"ObjectAPIName", true},
		{"objectapiname", true},
		{"SObjectName", true},
		{"sobjectname", true},
		{"ObjectMetadataId", false},
		{"RecordId", false},
		{"FlowExecutionId", false},
	}
	for _, tc := range cases {
		got := containsForbiddenAuditName(tc.name)
		if got != tc.want {
			t.Errorf("containsForbiddenAuditName(%q) = %v, want %v", tc.name, got, tc.want)
		}
	}
}