durableemitter

type AuthConfig struct {
	AuthHeaders      map[string]string
	AuthHeadersTTL   time.Duration
	AuthPublicKeyHex string
	// AuthKeySigner may be nil at init time for LOOP plugins; call SetGlobalSigner
	// after the CSA keystore is available.
	AuthKeySigner Signer
}

type BatchEmitter interface {
	// QueueMessage enqueues a single CloudEvent for batched delivery.
	// Returns an error only if the internal buffer is full or the client
	// has been stopped. Callers must treat a non-nil return as a
	// drop (the event is still persisted; retransmit will retry).
	QueueMessage(event *chipingress.CloudEventPb, callback func(error)) error
	// Start begins background processing. Must be called before QueueMessage.
	Start(ctx context.Context)
	// Stop flushes any queued events, waits for all in-flight network calls
	// and callbacks to complete, then closes the underlying transport.
	Stop()
}

type BatchInserter interface {
	InsertBatch(ctx context.Context, payloads [][]byte) ([]int64, error)
}

type Config struct {
	// RetransmitInterval controls how often the retransmit loop ticks.
	RetransmitInterval time.Duration
	// RetransmitAfter is the minimum age of an event before the retransmit
	// loop considers it. This gives the batch publish path time to succeed.
	RetransmitAfter time.Duration
	// RetransmitBatchSize caps how many pending rows are listed per retransmit tick.
	RetransmitBatchSize int
	// ExpiryInterval controls how often the expiry loop ticks.
	ExpiryInterval time.Duration
	// EventTTL is the maximum age of an event before it is expired.
	EventTTL time.Duration
	// PublishTimeout is the deadline for DB operations in delivery callbacks
	// (BatchDelete). The actual gRPC publish timeout is configured on
	// the BatchEmitter (batch.Client) directly.
	PublishTimeout time.Duration
	// DisablePruning disables the background expiry (DeleteExpired) loop.
	// Events then remain in the DB even after they age past EventTTL. Useful for
	// post-test analysis of created_at timestamps.
	DisablePruning bool
	// Hooks is optional instrumentation (load tests, profiling). Nil fields are skipped.
	// Callbacks may run from many goroutines; implementations must be thread-safe.
	Hooks *Hooks
	// Metrics enables OpenTelemetry instruments (queue, publish, store, optional process stats).
	// When non-nil, a meter must be supplied to NewDurableEmitter; nil disables instrumentation.
	Metrics *DurableEmitterMetricsConfig
	// InsertBatchSize enables write coalescing when > 0 and the store implements
	// BatchInserter. Multiple concurrent Emit() calls are grouped into a single
	// multi-row INSERT, dramatically reducing per-event transaction overhead.
	// Each coalescer worker collects up to InsertBatchSize payloads before flushing.
	InsertBatchSize int
	// InsertBatchFlushInterval is the linger time after the first payload arrives
	// in a coalescing batch. Zero defaults to 2ms.
	InsertBatchFlushInterval time.Duration
	// InsertBatchWorkers is the number of concurrent batch-insert goroutines.
	// Zero defaults to 4.
	InsertBatchWorkers int
	// DeleteBatchSize enables delete coalescing when > 0. Instead of issuing one
	// DELETE per delivered event from the delivery callback, ids are funneled to
	// background workers that collapse many ids into a single BatchDelete,
	// drastically reducing per-event DELETE/connection churn.
	// Each worker collects up to DeleteBatchSize ids before flushing.
	DeleteBatchSize int
	// DeleteBatchFlushInterval is the linger time after the first id arrives in a
	// coalescing batch before it is flushed. Zero defaults to 100ms.
	DeleteBatchFlushInterval time.Duration
	// DeleteBatchWorkers is the number of concurrent batch-delete goroutines.
	// Zero defaults to 2.
	DeleteBatchWorkers int
}

type DurableEmitter struct {
	services.Service
	// contains filtered or unexported fields
}

type DurableEmitterMetricsConfig struct {
	// PollInterval is how often queue and optional process gauges refresh. Zero = 10s.
	PollInterval time.Duration
	// MaxQueuePayloadBytes, if > 0, records capacity_usage_ratio = queue_payload_bytes / max.
	MaxQueuePayloadBytes int64
}

type DurableEvent struct {
	ID        int64
	Payload   []byte // serialized CloudEventPb proto
	CreatedAt time.Time
}

type DurableEventStore interface {
	// Insert persists a serialized event and returns its assigned ID.
	Insert(ctx context.Context, payload []byte) (int64, error)
	// Delete physically removes a row (corrupt payloads, policy drops, tests).
	Delete(ctx context.Context, id int64) error
	// BatchDelete records successful delivery of multiple events to Chip by
	// deleting them in a single operation (delete-on-delivery)
	BatchDelete(ctx context.Context, ids []int64) (int64, error)
	// ListPending returns undelivered events created before createdBefore,
	// ordered by (created_at, id) ascending and strictly after the
	// (afterCreatedAt, afterID) cursor, up to limit rows. Pass a zero cursor
	// (time.Time{}, 0) to start from the oldest. The retransmit loop pages
	// through the backlog with this cursor — advancing it each tick and wrapping
	// to a zero cursor at the end — so a persistently-failing event can't
	// monopolise the head of the list. Under delete-on-delivery every row still
	// present is undelivered, so this is the pending backlog.
	ListPending(ctx context.Context, createdBefore, afterCreatedAt time.Time, afterID int64, limit int) ([]DurableEvent, error)
	// DeleteExpired removes any events older than ttl and returns the count
	// deleted — a time-based garbage collector that also reclaims rows which
	// failed to delete on delivery (e.g. a DB error in the delivery callback), so
	// nothing lingers past EventTTL.
	DeleteExpired(ctx context.Context, ttl time.Duration) (int64, error)
}

type DurableQueueObserver interface {
	// ObserveDurableQueue returns live queue statistics. eventTTL matches Config and
	// is used to derive the remaining TTL budget of the oldest pending event.
	ObserveDurableQueue(ctx context.Context, eventTTL time.Duration) (DurableQueueStats, error)
}

type DurableQueueStats struct {
	// Depth is the number of undelivered (delivered_at IS NULL) rows — the
	// delivery backlog.
	Depth int64
	// TotalRows is the number of rows physically present in the table, including
	// delivered-but-not-yet-purged rows. This is the authoritative "queue depth"
	// (actual table count): it is read directly from the DB so it stays correct
	// regardless of how many writers share the table or which in-memory delta
	// updates were lost to failed/partial DB operations.
	TotalRows        int64
	PayloadBytes     int64
	OldestPendingAge time.Duration // 0 if the queue is empty
	// TTLBudget is the remaining time before the oldest still-pending event hits
	// EventTTL (i.e. EventTTL - OldestPendingAge). It is the headroom of the event
	// closest to expiry and serves as a DLQ-pressure proxy; there is no separate
	// dead-letter table in the default design. It goes negative when the expiry
	// loop is behind. Equals EventTTL when the queue is empty.
	TTLBudget time.Duration
}

type Hooks struct {
	// OnEmitInsert is called after each store.Insert in Emit (the DB write that
	// blocks the caller). elapsed covers only the INSERT; err is nil on success.
	OnEmitInsert func(elapsed time.Duration, err error)
	// OnBatchPublish is called from the delivery callback after each event's
	// batch is sent. elapsed is measured from QueueMessage call to callback
	// invocation; batchSize is always 1 (one callback per event); err is nil
	// on success.
	OnBatchPublish func(elapsed time.Duration, batchSize int, err error)
	// OnBatchDelete is called after BatchDelete following a successful delivery.
	OnBatchDelete func(elapsed time.Duration, count int)
}

type PgDurableEventStore struct {
	// contains filtered or unexported fields
}

type SetupConfig struct {
	// Endpoint is the gRPC address for the Chip Ingress service.
	Endpoint string
	// InsecureConnection disables TLS when true.
	InsecureConnection bool
	// Auth configures chip ingress credentials. AuthKeySigner may be nil at init
	// for LOOP plugins; call SetGlobalSigner after the CSA keystore is available.
	Auth AuthConfig
	// RetransmitEnabled controls whether the retransmit and cleanup loops run.
	// Set to true for the host (chainlink node) process.
	// Set to false for LOOP plugin processes — the host's retransmit loop picks
	// up any rows inserted by plugin-side DurableEmitters from the shared DB.
	RetransmitEnabled bool

	// Batch client tuning — zero values use package defaults.
	BatchSize          int           // default: 50
	BatchInterval      time.Duration // default: 50ms
	MaxConcurrentSends int           // default: 4
	MaxPublishTimeout  time.Duration // default: 5s
	ShutdownTimeout    time.Duration // default: 30s
	// MessageBufferSize is the capacity of the batch client's producer→batcher
	// channel. QueueMessage drops events (non-blocking send) when this is full,
	// which happens when emit throughput outpaces the batcher.
	MessageBufferSize int // default: 10000

	// EmitterConfig overrides DefaultConfig when non-nil.
	EmitterConfig *Config
	// Meter is the OpenTelemetry meter for instrumentation. Nil disables metrics.
	Meter metric.Meter
}

type Signer = chipingress.Signer