// Package automation — audit.go
//
// Async audit writer for _System_FlowAuditLog. Emits per-element execution
// rows from the Flow runtime (Pillar 5). L0 platform observability surface;
// never references tenant SObject names, ProfileIds, or picklist values.
// All object references are by BIGINT ID (ObjectMetadataId), NEVER by
// ObjectAPIName, per CLAUDE.md strong-reading.
//
// Production wiring: call SetAuditWriter(w) once at startup with a real Writer.
// Leave nil (default) to disable all audit emission with zero overhead.
package automation

import (
	"context"
	"log"
	"math/rand"
	"runtime/debug"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
)

// DefaultAuditSampleRate is the fraction of element executions to record when
// FlowDefinition.AuditSampleRate is zero/unset. The audit log's contract is
// "log every element completion (success or fail) so operators can reconstruct
// what happened" — a default below 1.0 silently drops most rows and makes the
// table useless for that purpose. Per-flow AuditSampleRate is still honored
// for tenants that explicitly opt down to a lower rate to manage volume; the
// default just stops being lossy out of the box.
const DefaultAuditSampleRate = 1.0

// auditChannelBufferSize is the bounded channel capacity for async row delivery.
// When the channel is full, Append drops the row (non-blocking backpressure)
// and increments DroppedCount for observability. 65536 entries gives ~2 minutes
// of buffer at 500 rows/sec sustained — bridging realistic TiDB / PD stalls
// without coupling flow execution latency to audit-log persistence. Operators
// monitor DroppedCount; if it grows during steady state, increase worker count
// or flip the table to a higher-durability transport. This is the documented
// at-most-once contract — see Append's docstring.
const auditChannelBufferSize = 65536

// auditBatchSize is the maximum number of rows to flush in one DB write.
const auditBatchSize = 100

// auditFlushInterval is the maximum time a worker will wait before flushing a
// partial batch to DB even if the batch is not full.
const auditFlushInterval = 1 * time.Second

const auditWriterFlushTimeout = 10 * time.Second

// maxAuditPayloadBytes caps the size of InputJSON / OutputJSON / ErrorPayload
// before INSERT to prevent oversized rows. MySQL/TiDB TEXT columns store at
// most 65,535 bytes (NOT 64*1024 = 65,536); a payload at exactly the byte
// boundary survived capPayload but failed INSERT with "data too long",
// failing the whole batch and losing up to auditBatchSize neighbouring rows.
// Keep this aligned with the physical TEXT column limit.
const maxAuditPayloadBytes = 65535

// Row is a single flow-element execution record destined for _System_FlowAuditLog.
// ObjectMetadataId stores object references by integer ID; never by name (strong
// reading per CLAUDE.md). ObjectAPIName / sObjectName are explicitly absent.
type Row struct {
	Id               string
	FlowExecutionId  string
	FlowDefinitionId string // ID of the FlowDefinition (from interview at emit time)
	FlowVersionId    string // ID of the active FlowVersion (from interview at emit time)
	ElementId        string
	ElementType      string
	Status           string // "success" | "error" | "skipped"
	DurationMs       int
	InputJSON        string
	OutputJSON       string
	ErrorPayload     string
	ObjectMetadataId int64 // nullable: 0 means absent
	RecordId         string
	CreatedDate      time.Time
}

// QueryFilter selects rows from _System_FlowAuditLog.
type QueryFilter struct {
	FlowExecutionId string
	ElementId       string
	Status          string
	Limit           int
}

// AuditStore is the narrow persistence interface used by Writer and Retention.
// Tests supply an in-memory implementation; production passes the real DB layer.
type AuditStore interface {
	// InsertRows inserts a batch of audit rows. The store generates DB-level IDs
	// when Row.Id is empty (but we populate it before calling).
	InsertRows(ctx context.Context, rows []Row) error
	// QueryRows returns rows matching filter. Used by Writer.Query.
	QueryRows(ctx context.Context, filter QueryFilter) ([]Row, error)
	// DeleteBefore removes rows whose CreatedDate is before the given time. Returns
	// the number of rows deleted.
	DeleteBefore(ctx context.Context, before time.Time) (int64, error)
	// CountRows returns the total number of rows in the table.
	CountRows(ctx context.Context) (int64, error)
	// DeleteOldest removes the N oldest rows (by CreatedDate). Used for row-count-cap
	// retention. Returns number deleted.
	DeleteOldest(ctx context.Context, n int64) (int64, error)
}

// Writer is an asynchronous audit row emitter backed by a bounded channel and a
// configurable worker goroutine pool. Non-blocking: Append drops rows when the
// channel is full (backpressure) rather than blocking callers.
//
// Nil-safe: calling methods on a nil *Writer is a no-op. This preserves existing
// tests that construct FlowExecutor without wiring an audit writer.
type Writer struct {
	store   AuditStore
	ch      chan Row
	workers int

	dropped atomic.Int64

	rngMu sync.Mutex // guards rng for concurrent ShouldSample calls
	rng   *rand.Rand // injectable for test determinism; always accessed under rngMu

	done       chan struct{} // closed when all workers have exited after Shutdown
	shutdownMu sync.Mutex
	shutdown   bool // guarded by shutdownMu
}

// WriterOption configures a Writer at construction time.
type WriterOption func(*Writer)

// WithWorkerCount sets the number of background flush goroutines. Default is 1.
func WithWorkerCount(n int) WriterOption {
	return func(w *Writer) {
		if n > 0 {
			w.workers = n
		}
	}
}

// WithChannelBuffer sets the channel buffer size. Default is auditChannelBufferSize.
func WithChannelBuffer(n int) WriterOption {
	return func(w *Writer) {
		if n > 0 {
			w.ch = make(chan Row, n)
		}
	}
}

// WithRand injects a deterministic *rand.Rand for sampling in tests.
func WithRand(r *rand.Rand) WriterOption {
	return func(w *Writer) {
		w.rng = r
	}
}

// NewWriter constructs and starts a Writer backed by store.
// Call Shutdown when done to drain remaining rows and stop workers.
func NewWriter(store AuditStore, opts ...WriterOption) *Writer {
	w := &Writer{
		store:   store,
		ch:      make(chan Row, auditChannelBufferSize),
		workers: 1,
		done:    make(chan struct{}),
		rng:     rand.New(rand.NewSource(time.Now().UnixNano())), //nolint:gosec
	}
	for _, o := range opts {
		o(w)
	}
	w.start()
	return w
}

// start launches the worker goroutines. Called once by NewWriter.
func (w *Writer) start() {
	// Use a WaitGroup-like pattern via a counting goroutine.
	remaining := make(chan struct{}, w.workers)
	for i := 0; i < w.workers; i++ {
		go func() {
			w.runWorker()
			remaining <- struct{}{}
		}()
	}
	go func() {
		for i := 0; i < w.workers; i++ {
			<-remaining
		}
		close(w.done)
	}()
}

// runWorker is the long-running flush loop executed by each worker goroutine.
func (w *Writer) runWorker() {
	batch := make([]Row, 0, auditBatchSize)
	ticker := time.NewTicker(auditFlushInterval)
	defer ticker.Stop()

	flush := func() {
		if len(batch) == 0 {
			return
		}
		flushed := batch
		defer func() {
			if recovered := recover(); recovered != nil {
				log.Printf("audit writer: recovered panic while flushing %d rows: %v\n%s", len(flushed), recovered, debug.Stack())
			}
			batch = batch[:0]
		}()

		ctx, cancel := auditWriterFlushContext()
		defer cancel()
		if err := w.store.InsertRows(ctx, flushed); err != nil {
			log.Printf("audit writer: failed to flush %d rows: %v", len(flushed), err)
		}
	}

	for {
		select {
		case row, ok := <-w.ch:
			if !ok {
				// Channel closed — drain remaining and exit.
				// The channel has been closed by Shutdown; drain is done by
				// consuming from the closed channel until it is empty.
				for {
					select {
					case r2, ok2 := <-w.ch:
						if !ok2 {
							goto done
						}
						batch = append(batch, r2)
						if len(batch) >= auditBatchSize {
							flush()
						}
					default:
						goto done
					}
				}
			done:
				flush()
				return
			}
			batch = append(batch, row)
			if len(batch) >= auditBatchSize {
				flush()
			}
		case <-ticker.C:
			flush()
		}
	}
}

func auditWriterFlushContext() (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.Background(), auditWriterFlushTimeout)
}

// Append enqueues row for async write. Non-blocking: when the channel is full,
// the row is dropped and DroppedCount is incremented. Safe to call on a nil *Writer.
// If Shutdown has already been called, the row is silently dropped.
//
// Concurrency safety: shutdownMu is held across the channel send so that
// Shutdown's close(w.ch) cannot race with the send. The select is non-blocking
// (default case) so the mutex is held only for the duration of the channel probe.
func (w *Writer) Append(row Row) {
	if w == nil {
		return
	}
	if row.Id == "" {
		if id, err := uuid.NewRandom(); err == nil {
			row.Id = id.String()
		}
	}
	if row.CreatedDate.IsZero() {
		row.CreatedDate = time.Now()
	}
	row.InputJSON = capPayload(row.InputJSON)
	row.OutputJSON = capPayload(row.OutputJSON)
	row.ErrorPayload = capPayload(row.ErrorPayload)

	w.shutdownMu.Lock()
	defer w.shutdownMu.Unlock()
	if w.shutdown {
		return
	}
	select {
	case w.ch <- row:
	default:
		w.dropped.Add(1)
		log.Printf("audit writer: channel full, row dropped (total dropped: %d)", w.dropped.Load())
	}
}

// capPayload truncates s to maxAuditPayloadBytes to prevent oversized rows.
func capPayload(s string) string {
	if len(s) <= maxAuditPayloadBytes {
		return s
	}
	return s[:maxAuditPayloadBytes]
}

// MaybeEmit emits row to the writer when ShouldSample(rate) returns true.
// Safe to call on a nil *Writer (no-op). Call sites in element executors
// MUST go through emitFlowAudit (below) instead of MaybeEmit directly so the
// preview-mode suppression is enforced uniformly.
func (w *Writer) MaybeEmit(row Row, rate float64) {
	if w == nil {
		return
	}
	if !w.ShouldSample(rate) {
		return
	}
	w.Append(row)
}

// emitFlowAudit is the single chokehold every flow element executor uses to
// emit an audit row. It wraps pkgAuditWriter.MaybeEmit with two extra checks:
//
//  1. preview-mode suppression: when ctx was marked via WithPreviewMode (i.e.
//     this execution was started by FlowExecutor.PreviewRecordCreateFields),
//     no row is emitted. _System_FlowAuditLog is real DB state and writing
//     to it from a preview path would violate the no-DML preview contract.
//  2. nil-safety on a nil writer: identical to the underlying MaybeEmit.
//
// Element executors call this from their defer-emit blocks; replacing it
// with a direct pkgAuditWriter.MaybeEmit re-introduces the Pillar-3 audit
// leak that PR #273 round 7 closes (IANTHEREAL P1 Blocker 1).
func emitFlowAudit(ctx context.Context, row Row, rate float64) {
	if isPreviewMode(ctx) {
		return
	}
	pkgAuditWriter.MaybeEmit(row, rate)
}

// ShouldSample returns true with probability rate. Uses the injectable rng for
// test determinism. Safe to call on a nil *Writer (returns false).
// rngMu guards w.rng so concurrent calls from multiple goroutines are safe.
func (w *Writer) ShouldSample(rate float64) bool {
	if w == nil {
		return false
	}
	if rate <= 0 {
		return false
	}
	if rate >= 1.0 {
		return true
	}
	w.rngMu.Lock()
	v := w.rng.Float64()
	w.rngMu.Unlock()
	return v < rate
}

// DroppedCount returns the total number of rows dropped due to backpressure.
// Safe to call on a nil *Writer (returns 0).
func (w *Writer) DroppedCount() int64 {
	if w == nil {
		return 0
	}
	return w.dropped.Load()
}

// Query synchronously returns rows matching filter. Safe to call on a nil *Writer.
func (w *Writer) Query(ctx context.Context, filter QueryFilter) ([]Row, error) {
	if w == nil {
		return nil, nil
	}
	return w.store.QueryRows(ctx, filter)
}

// Shutdown gracefully drains the channel, flushes all buffered rows to DB,
// and waits for all workers to exit. Returns when all rows have been written
// or the context deadline fires. Safe to call more than once (idempotent).
func (w *Writer) Shutdown(ctx context.Context) error {
	if w == nil {
		return nil
	}
	w.shutdownMu.Lock()
	if !w.shutdown {
		w.shutdown = true
		close(w.ch)
	}
	w.shutdownMu.Unlock()

	select {
	case <-w.done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// ----------------------------------------------------------------------------
// Package-level writer (set at startup; nil = disabled)
// ----------------------------------------------------------------------------

// pkgAuditWriter is the package-level Writer used by element executors.
// nil by default; production startup calls SetAuditWriter.
var pkgAuditWriter *Writer

// SetAuditWriter wires a Writer into the automation package so all element
// executors can emit audit rows. Call once at startup. Pass nil to disable.
// Subsequent calls replace the writer (last-write-wins; safe for tests).
func SetAuditWriter(w *Writer) {
	pkgAuditWriter = w
}

// GetAuditWriter returns the currently registered package-level writer (may
// be nil if not yet wired).
func GetAuditWriter() *Writer {
	return pkgAuditWriter
}

// resolveAuditSampleRate normalizes an admin-authored FlowDefinition.AuditSampleRate
// into the runtime rate used by emitters. Zero/negative/NaN falls back to
// DefaultAuditSampleRate; values >= 1.0 are clamped to 1.0 (always emit).
func resolveAuditSampleRate(rate float64) float64 {
	// NaN comparisons always return false; check explicitly to avoid leaking
	// a NaN back into the sampling decision (which would silently disable audit).
	if rate != rate || rate <= 0 {
		return DefaultAuditSampleRate
	}
	if rate >= 1.0 {
		return 1.0
	}
	return rate
}

// auditSampleRateContextKey is the context key used to carry the resolved
// FlowDefinition.AuditSampleRate into element executors (set once per
// executeFlowMetadata call, read by each element executor's defer).
type auditSampleRateContextKey struct{}

// withAuditSampleRate stores rate on ctx for downstream element executors.
func withAuditSampleRate(ctx context.Context, rate float64) context.Context {
	return context.WithValue(ctx, auditSampleRateContextKey{}, rate)
}

// auditSampleRateFromCtx retrieves the rate stored by withAuditSampleRate.
// Falls back to DefaultAuditSampleRate when absent.
func auditSampleRateFromCtx(ctx context.Context) float64 {
	if v, ok := ctx.Value(auditSampleRateContextKey{}).(float64); ok && v > 0 {
		return v
	}
	return DefaultAuditSampleRate
}

// auditFlowDefinitionIdContextKey carries the resolved FlowDefinition.Id into
// element executors so each defer-emitted Row can be tagged with its owning
// FlowDefinition (for cross-version aggregation in the audit log).
type auditFlowDefinitionIdContextKey struct{}

// withAuditFlowDefinitionId stores the FlowDefinition.Id on ctx so downstream
// element executors can populate Row.FlowDefinitionId without a per-element
// metadata lookup.
func withAuditFlowDefinitionId(ctx context.Context, id string) context.Context {
	return context.WithValue(ctx, auditFlowDefinitionIdContextKey{}, id)
}

// auditFlowDefinitionIdFromCtx retrieves the FlowDefinition.Id stored by
// withAuditFlowDefinitionId. Returns "" when absent (e.g. unit tests that
// invoke an element executor directly without going through executeFlowVersion).
func auditFlowDefinitionIdFromCtx(ctx context.Context) string {
	if v, ok := ctx.Value(auditFlowDefinitionIdContextKey{}).(string); ok {
		return v
	}
	return ""
}

// ----------------------------------------------------------------------------
// Strong-reading helpers used in tests (T51 AC#3)
// ----------------------------------------------------------------------------

// forbiddenAuditColumnSubstrings lists substrings whose presence in a column
// or struct-field name indicates a CLAUDE.md strong-reading violation in the
// new audit schema.
var forbiddenAuditColumnSubstrings = []string{
	"objectapiname",
	"sobjectname",
}

// containsForbiddenAuditName reports whether name (case-insensitive) matches
// any strong-reading-forbidden substring.
func containsForbiddenAuditName(name string) bool {
	lower := strings.ToLower(name)
	for _, sub := range forbiddenAuditColumnSubstrings {
		if strings.Contains(lower, sub) {
			return true
		}
	}
	return false
}