runtime.go

// Package streaming runs a Murmur Pipeline in Live mode: it reads records from the
// pipeline's Source, applies the user's key/value extractors and monoid Combine, writes
// to the primary state store (and optionally a cache), and Acks the source record.
//
// Phase 1 is intentionally a single-goroutine sequential loop. Per-partition parallelism
// is a Phase 2 optimization gated on real benchmarks. franz-go batches fetches under the
// hood; for counter-class workloads at moderate rates, single-goroutine processing is
// not the bottleneck.
package streaming

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/gallowaysoftware/murmur/pkg/exec/processor"
	"github.com/gallowaysoftware/murmur/pkg/metrics"
	"github.com/gallowaysoftware/murmur/pkg/monoid/windowed"
	"github.com/gallowaysoftware/murmur/pkg/pipeline"
	"github.com/gallowaysoftware/murmur/pkg/source"
	"github.com/gallowaysoftware/murmur/pkg/state"
)

// RunOption configures Run.
type RunOption func(*runConfig)

type runConfig struct {
	processor.Config
	deadLetter       func(eventID string, err error)
	batchWindow      time.Duration
	maxBatchSize     int
	concurrency      int
	keyDebounce      time.Duration
	keyDebounceMax   int
	valueDebounce    time.Duration
	valueDebounceMax int

	// batchTick is the periodic flush interval for RecordBatch metrics.
	// Default 1 s. Pass <= 0 to disable. The mode label is ModeStreaming
	// so a single dashboard can compare streaming / bootstrap / replay
	// throughput.
	batchTick time.Duration

	// batchCount accumulates records-since-last-tick across all worker
	// paths (single, concurrent, aggregator). The tick goroutine swaps
	// it to zero on each tick and emits the count + elapsed time as a
	// RecordBatch.
	batchCount atomic.Int64

	// debouncer is materialized once in Run when keyDebounce > 0 and
	// reused across the per-record processing path. Concurrency-safe.
	debouncer *keyDebouncer

	// valueDebouncer is materialized once in Run when
	// valueDebounce > 0; same shape as debouncer but keyed by
	// (entityKey, fingerprintedValue).
	valueDebouncer *valueDebouncer
}

// debounceAllow returns false when a record's first-emitted key was
// already seen within the debounce window. Caller should Ack-and-skip
// when this returns false. Returns true (allow) when:
//   - the debouncer is unconfigured
//   - keys is empty (the underlying MergeMany no-ops on no keys anyway)
//   - the first key has not been seen within the window
func (c *runConfig) debounceAllow(keys []string) bool {
	if c.debouncer == nil {
		return true
	}
	if len(keys) == 0 {
		return true
	}
	return !c.debouncer.shouldDrop(keys[0])
}

// WithMetrics installs a metrics.Recorder on the runtime. Defaults to metrics.Noop{}.
func WithMetrics(r metrics.Recorder) RunOption {
	return func(c *runConfig) {
		if r != nil {
			c.Recorder = r
		}
	}
}

// WithMaxAttempts sets the per-record retry budget. Defaults to 3. A record that
// errors all the way through MaxAttempts is dead-lettered (see WithDeadLetter)
// and the runtime continues with the next record rather than crashing.
//
// Set to 1 to disable retries (any error → dead-letter immediately).
func WithMaxAttempts(n int) RunOption {
	return func(c *runConfig) {
		if n >= 1 {
			c.MaxAttempts = n
		}
	}
}

// WithRetryBackoff sets the per-attempt sleep schedule. Backoff doubles after
// each failure starting from base, capped at max, with full jitter. Defaults
// to 50 ms base / 5 s max.
func WithRetryBackoff(base, max time.Duration) RunOption {
	return func(c *runConfig) {
		if base > 0 {
			c.BackoffBase = base
		}
		if max > 0 {
			c.BackoffMax = max
		}
	}
}

// WithDeadLetter installs a callback invoked when a record fails every
// attempt. Use it to push the record's EventID (and the underlying error) to
// a DLQ table / topic / structured-log sink. Default is a no-op — the record
// is silently dropped and the runtime moves on. The runtime ALSO calls
// recorder.RecordError so DLQ counts surface in /api metrics regardless.
func WithDeadLetter(fn func(eventID string, err error)) RunOption {
	return func(c *runConfig) {
		if fn != nil {
			c.deadLetter = fn
		}
	}
}

// WithBatchWindow enables write aggregation: per-(entity, bucket) deltas
// are accumulated in memory for `window` time, then flushed as a SINGLE
// store.MergeUpdate per key. This is the only way to keep up with hot
// keys at scale — a celebrity post taking 50k like-events/sec lands as
// ONE atomic-ADD per flush window per worker instead of 50k.
//
// Tradeoffs:
//
//   - Latency: a record's contribution is invisible to readers for up to
//     `window` time after acceptance. Default unset (no batching) gives
//     immediate-merge semantics; common production values are 500ms–5s
//     depending on read-staleness tolerance.
//   - Durability under crash: records are Ack'd to the source AFTER the
//     batch flushes, so a worker crash drops at most `window`-worth of
//     accumulated records and the source redelivers them on restart.
//     What the redelivery is worth depends on whether a Deduper is wired:
//     without WithDedup, those records never reached the store, so the
//     redelivery is their first apply and nothing is lost. WithDedup, the
//     claim is taken when the record ENTERS the accumulator, not at
//     flush — so the claim survives the crash, the redelivery is
//     dedup-skipped, and the in-flight batch's contribution is lost for
//     good. Pairing WithBatchWindow with WithDedup trades a possible
//     double-count for a possible silent loss of up to one flush window
//     (or maxBatch records per key) per crash; pick the failure you can
//     live with. There is no release-on-crash: flushOne dead-letters and
//     Acks a batch whose merge exhausted its retries but leaves the
//     claims standing, so replaying those EventIDs by hand is suppressed
//     until the dedup TTL expires.
//   - Memory: at most `maxBatch` records per (entity, bucket) before
//     forced flush. Default 1024 if unset. The number of concurrent keys
//     in flight is unbounded — for high-cardinality pipelines (per-user
//     keys), keep `window` short so each batch stays small.
//
// Pass window = 0 (or omit the option) to disable batching entirely;
// records flow through processor.MergeMany one at a time.
//
// Strongly recommended for any pipeline with a known hot-key distribution.
func WithBatchWindow(window time.Duration, maxBatch int) RunOption {
	return func(c *runConfig) {
		if window > 0 {
			c.batchWindow = window
		}
		if maxBatch > 0 {
			c.maxBatchSize = maxBatch
		}
	}
}

// WithConcurrency configures the runtime's intra-worker parallelism.
// Default 1 (the historical single-goroutine loop). The value is
// reused in two places:
//
//  1. **Source-dispatch routing.** When no aggregator is configured
//     (no WithBatchWindow), N worker goroutines drain the source
//     channel and records are routed by hash(first-emitted-key) — so
//     SAME-key records always land on the same worker. This preserves
//     per-key arrival order for non-additive monoids (First, Last,
//     Min, Max) where ordering still matters.
//
//  2. **Aggregator flush fan-out.** When WithBatchWindow IS configured,
//     each (entity, bucket) batch has already coalesced its
//     contributions into ONE delta at accept-time, so per-key flushes
//     touch distinct store keys. flushAll dispatches the per-key
//     MergeUpdates across N worker goroutines (bounded by the number
//     of distinct keys in the flush). This is the change that unlocks
//     real I/O parallelism inside a single worker instance: a flush
//     touching 1000 keys against a 5ms-per-call store completes in
//     ~5s at concurrency=1 vs ~315ms at concurrency=16.
//
// When to raise it:
//
//   - The single-goroutine ceiling (~5–10k events/s/worker against
//     DDB-local; documented in README) is your bottleneck.
//   - The aggregator flush is bottlenecked on store latency (large
//     coalesced-key set per window, DDB UpdateItem at ~5ms each).
//   - Fan-out via additional Kafka partitions isn't an option (e.g.,
//     downstream key cardinality demands a single partition).
//
// When NOT to raise it:
//
//   - Your store is CAS-heavy (BytesStore on a hot key). N workers
//     pre-routing by key removes cross-worker CAS contention but a
//     single worker on a hot CAS key is already the bottleneck.
//   - You have very few distinct keys per flush window — the flush
//     fan-out is bounded by len(keys), so concurrency above that
//     produces no extra speedup.
//
// The source-dispatch router routes by first-key hash. Hierarchical-
// rollup pipelines (KeyByMany) partition on the FIRST emitted key —
// so a record's "post:X" delta lands on the same worker as another
// "post:X" record, even if the records also fan out to "country:Y" /
// "global" keys that happen to hash differently. This keeps the
// per-key ordering guarantee meaningful where it matters most.
func WithConcurrency(n int) RunOption {
	return func(c *runConfig) {
		if n >= 1 {
			c.concurrency = n
		}
	}
}

// WithBatchTick overrides the RecordBatch flush interval. The default 1 s
// fits most production deployments; raise it to reduce metrics chatter
// for very slow pipelines, or pass <= 0 to disable periodic batch flushes
// entirely.
func WithBatchTick(d time.Duration) RunOption {
	return func(c *runConfig) {
		c.batchTick = d
	}
}

// WithDedup installs a state.Deduper. Each record's EventID is claimed via
// MarkSeen before the merge runs; if MarkSeen reports the EventID was
// already claimed (a duplicate from a worker crash mid-write), the runtime
// skips the merge and Acks the record so the source advances.
//
// This makes at-least-once delivery idempotent at the monoid layer for any
// pipeline whose Combine is non-commutative or non-idempotent (e.g. Sum,
// HLL.Add). Pipelines whose Combine is already idempotent (Set, Min, Max)
// don't strictly need a Deduper but adding one is harmless.
//
// The Deduper itself is typically backed by a small DDB table with TTL —
// see pkg/state/dynamodb.NewDeduper.
//
// Note the interaction with WithBatchWindow. On the unbatched path the
// claim is released when the merge fails, so a redelivery is re-applied.
// The aggregator claims on accept and never releases, so a crash before
// the flush loses the accumulated records rather than double-counting
// them — see WithBatchWindow's "Durability under crash" note.
func WithDedup(d state.Deduper) RunOption {
	return func(c *runConfig) {
		if d != nil {
			c.Dedup = d
		}
	}
}

// Run executes the pipeline in Live mode until ctx is canceled or the source returns.
// Returns nil on graceful shutdown.
//
// Per-record processing errors are retried up to MaxAttempts with exponential
// backoff, then dead-lettered (callback + RecordError) and skipped. The runtime
// itself only returns early if the source returns or ctx is canceled — a
// stuck downstream does not take the worker down with it.
//
// The pipeline must have been configured with From / Key / Value / Aggregate / StoreIn
// before Run is called; Build is invoked internally to validate.
func Run[T any, V any](ctx context.Context, p *pipeline.Pipeline[T, V], opts ...RunOption) error {
	if err := p.Build(); err != nil {
		return fmt.Errorf("streaming.Run: %w", err)
	}
	if p.Source() == nil {
		return fmt.Errorf("streaming.Run: %w", pipeline.ErrMissingSource)
	}

	cfg := runConfig{Config: processor.Defaults(), batchTick: time.Second}
	for _, o := range opts {
		o(&cfg)
	}
	if cfg.keyDebounce > 0 {
		cfg.debouncer = newKeyDebouncer(cfg.keyDebounce, cfg.keyDebounceMax)
	}
	if cfg.valueDebounce > 0 {
		cfg.valueDebouncer = newValueDebouncer(cfg.valueDebounce, cfg.valueDebounceMax)
	}

	// Periodic RecordBatch goroutine. Reads batchCount atomically every
	// batchTick and emits a single RecordBatch call with the pipeline name,
	// mode=streaming, the records seen since the last tick, and the elapsed
	// wall time. Stops when batchDone is closed (Run is returning).
	//
	// The default metrics.Noop recorder no-ops on RecordBatch, so a pipeline
	// running without explicit metrics pays a single atomic-load per tick
	// and nothing more.
	batchDone := make(chan struct{})
	if cfg.batchTick > 0 {
		go runBatchTicker(&cfg, p.Name(), metrics.ModeStreaming, batchDone)
	}
	defer close(batchDone)

	name := p.Name()
	src := p.Source()
	keysFn := p.KeysFn() // multi-key aware; falls back to wrapping KeyFn in a 1-element slice
	valueFn := p.ValueFn()
	store := p.Store()
	cache := p.CacheStore()
	window := p.Window()

	// Buffered channel decouples the source's network polling from the processing loop.
	// 1024 records of slack absorbs short DDB latency spikes without backpressuring Kafka.
	records := make(chan source.Record[T], 1024)

	// Source pump goroutine.
	srcDone := make(chan error, 1)
	go func() {
		srcDone <- src.Read(ctx, records)
		close(records)
	}()

	// If write aggregation is enabled, spin up the aggregator + flush loop.
	// Records still pass through processWithRetry's retry-and-deadletter
	// shape, but the actual store write is batched per (entity, bucket).
	var agg *aggregator[T, V]
	var flushDone chan struct{}
	if cfg.batchWindow > 0 {
		agg = newAggregator(&cfg, p.Monoid(), keysFn, valueFn, store, cache, window, name, cfg.maxBatchSize)
		flushDone = make(chan struct{})
		go func() {
			defer close(flushDone)
			agg.runFlushLoop(ctx, cfg.batchWindow)
		}()
	}

	// Concurrency > 1: route records to N worker goroutines by
	// hash(first-key). Same-key records always go to the same worker,
	// preserving per-key order. concurrency = 1 retains the historical
	// single-goroutine loop.
	if cfg.concurrency > 1 {
		err := runConcurrent(ctx, name, records, srcDone, keysFn, valueFn, store, cache, window, agg, &cfg)
		if agg != nil {
			// Source closed naturally — force a final flush so the
			// caller sees the post-drain state without having to wait
			// for ctx cancellation. Matches the sequential path's
			// "force a final flush" behavior below. The runFlushLoop
			// goroutine continues to fire on its ticker until ctx is
			// canceled; flushAll is safe to call concurrently with
			// the ticker firing.
			agg.flushAll(ctx)
		}
		return err
	}

	for {
		select {
		case <-ctx.Done():
			if agg != nil {
				<-flushDone // wait for final drain
			}
			return nil
		case err := <-srcDone:
			drainAndProcess(ctx, name, records, keysFn, valueFn, store, cache, window, agg, &cfg)
			if agg != nil {
				// Source closed naturally; force a final flush in addition
				// to whatever the flush loop was doing.
				agg.flushAll(ctx)
			}
			return err
		case rec, ok := <-records:
			if !ok {
				return <-srcDone
			}
			cfg.batchCount.Add(1)
			if agg != nil {
				if dup := agg.accept(ctx, rec); dup && rec.Ack != nil {
					_ = rec.Ack()
				}
			} else {
				processWithRetry(ctx, name, rec, keysFn, valueFn, store, cache, window, &cfg)
			}
		}
	}
}

// runConcurrent drains `records` across cfg.concurrency worker
// goroutines. A small router goroutine reads records, hashes the
// first-emitted-key, and dispatches to the matching worker channel.
// Same-key records always land on the same worker.
//
// Returns when the source channel closes or ctx cancels. Workers are
// drained on the way out so no record is dropped mid-flight.
func runConcurrent[T any, V any](
	ctx context.Context,
	name string,
	records <-chan source.Record[T],
	srcDone <-chan error,
	keysFn func(T) []string,
	valueFn func(T) V,
	store state.Store[V],
	cache state.Cache[V],
	window *windowed.Config,
	agg *aggregator[T, V],
	cfg *runConfig,
) error {
	n := cfg.concurrency
	workers := make([]chan source.Record[T], n)
	for i := range workers {
		// Per-worker buffer of 256 (1024/4 ~ chunked across N workers
		// in expectation; absorb short latency spikes without forcing
		// the router to block).
		workers[i] = make(chan source.Record[T], 256)
	}

	var wg sync.WaitGroup
	for i := range workers {
		wg.Add(1)
		go func(ch <-chan source.Record[T]) {
			defer wg.Done()
			for rec := range ch {
				cfg.batchCount.Add(1)
				if agg != nil {
					if dup := agg.accept(ctx, rec); dup && rec.Ack != nil {
						_ = rec.Ack()
					}
				} else {
					processWithRetry(ctx, name, rec, keysFn, valueFn, store, cache, window, cfg)
				}
			}
		}(workers[i])
	}

	// Router: route every record to one worker by first-key hash.
	// Records with no emitted keys (KeysFn returns nil/empty — rare)
	// route to worker 0; processor.MergeMany handles the empty-keys
	// case as a no-op.
	routerDone := make(chan struct{})
	go func() {
		defer close(routerDone)
		defer func() {
			for _, ch := range workers {
				close(ch)
			}
		}()
		for {
			select {
			case <-ctx.Done():
				return
			case rec, ok := <-records:
				if !ok {
					return
				}
				idx := routeKey(keysFn(rec.Value), n)
				select {
				case workers[idx] <- rec:
				case <-ctx.Done():
					return
				}
			}
		}
	}()

	<-routerDone
	wg.Wait()
	return <-srcDone
}

// runBatchTicker emits one RecordBatch every cfg.batchTick with the records
// observed since the last tick and the elapsed wall time. Returns when done
// is closed; on the way out it flushes any pending records so the final
// batch isn't lost on graceful shutdown.
func runBatchTicker(cfg *runConfig, pipelineName, mode string, done <-chan struct{}) {
	t := time.NewTicker(cfg.batchTick)
	defer t.Stop()
	last := time.Now()
	flush := func() {
		n := cfg.batchCount.Swap(0)
		if n == 0 {
			return
		}
		cfg.Recorder.RecordBatch(pipelineName, mode, int(n), time.Since(last))
		last = time.Now()
	}
	for {
		select {
		case <-done:
			flush()
			return
		case <-t.C:
			flush()
		}
	}
}

// routeKey maps the first-emitted key to a worker index. FNV-1a hash
// over the byte content; cheap and good-enough distribution. No keys
// → worker 0 (caller's processor.MergeMany no-ops on empty key list).
func routeKey(keys []string, n int) int {
	if len(keys) == 0 || n <= 1 {
		return 0
	}
	const (
		offset uint32 = 2166136261
		prime  uint32 = 16777619
	)
	h := offset
	for _, c := range []byte(keys[0]) {
		h ^= uint32(c)
		h *= prime
	}
	return int(h % uint32(n))
}

func drainAndProcess[T any, V any](
	ctx context.Context,
	name string,
	records <-chan source.Record[T],
	keysFn func(T) []string,
	valueFn func(T) V,
	store state.Store[V],
	cache state.Cache[V],
	window *windowed.Config,
	agg *aggregator[T, V],
	cfg *runConfig,
) {
	deadline := time.NewTimer(2 * time.Second)
	defer deadline.Stop()
	for {
		select {
		case <-deadline.C:
			return
		case r, ok := <-records:
			if !ok {
				return
			}
			cfg.batchCount.Add(1)
			if agg != nil {
				if dup := agg.accept(ctx, r); dup && r.Ack != nil {
					_ = r.Ack()
				}
			} else {
				processWithRetry(ctx, name, r, keysFn, valueFn, store, cache, window, cfg)
			}
		}
	}
}

// processWithRetry delegates per-record processing to processor.MergeMany
// (retries, dedup, metrics, multi-key fanout) and adds the streaming-specific
// Ack-and-skip poison-record handling on top. Returns no error so the caller's
// loop is not coupled to this record's outcome.
func processWithRetry[T any, V any](
	ctx context.Context,
	name string,
	rec source.Record[T],
	keysFn func(T) []string,
	valueFn func(T) V,
	store state.Store[V],
	cache state.Cache[V],
	window *windowed.Config,
	cfg *runConfig,
) {
	eventTime := rec.EventTime
	if eventTime.IsZero() {
		eventTime = time.Now()
	}
	keys := keysFn(rec.Value)
	if !cfg.debounceAllow(keys) {
		// Duplicate-key record within the debounce window — drop the
		// processor work but still Ack so the source advances.
		cfg.Recorder.RecordEvent(name + ":debounce_skip")
		if rec.Ack != nil {
			_ = rec.Ack()
		}
		return
	}
	v := valueFn(rec.Value)
	if cfg.valueDebouncer != nil && len(keys) > 0 {
		// Same (entity, value) seen within the value-debounce window —
		// drop without processor work, but Ack so the source advances.
		if cfg.valueDebouncer.shouldDrop(keys[0], fingerprintValue(v)) {
			cfg.Recorder.RecordEvent(name + ":value_debounce_skip")
			if rec.Ack != nil {
				_ = rec.Ack()
			}
			return
		}
	}
	err := processor.MergeMany(ctx, &cfg.Config, name, rec.EventID, eventTime,
		keys, v, store, cache, window)
	if err != nil {
		// Shutdown is NOT a poison record. The comment here used to claim
		// cancellation would "just return", but the code fell through and
		// both dead-lettered and Acked — telling the source a record had been
		// handled when it had not been merged at all. Every ECS deploy,
		// scale-in and task replacement silently lost its in-flight records
		// that way, and Run still returned nil so the worker logged a clean
		// exit.
		//
		// Return WITHOUT Acking: the source has not advanced, so the record
		// is redelivered to whoever picks the partition up next. That is the
		// at-least-once contract doing its job. (processor.MergeMany releases
		// the dedup claim on a detached context for the same reason, so the
		// redelivery is re-applied rather than dedup-skipped.)
		//
		// errors.Is, not ==: processor wraps the cause.
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			cfg.Recorder.RecordEvent(name + ":shutdown_unacked")
			return
		}

		// Genuine retry exhaustion — a poison record. processor already
		// recorded the dead_letter metric; invoke the user's callback and Ack
		// past it so the pipeline makes progress.
		if cfg.deadLetter != nil {
			cfg.deadLetter(rec.EventID, err)
		}
		if rec.Ack != nil {
			_ = rec.Ack()
		}
		return
	}

	// Successful merge (or dedup-skip). Ack so the source advances; surface
	// Ack errors to the recorder rather than as a runtime error.
	if rec.Ack != nil {
		if err := rec.Ack(); err != nil {
			cfg.Recorder.RecordError(name, fmt.Errorf("source Ack: %w", err))
		}
	}
}