ingest

Documentation

Index

• Constants
• func StartIngestWorker(ctx context.Context, nc *nats.Conn, cache cache.Cache, ...) (func(context.Context) error, error)
• type EventMessage
• type IngestWorker
• type Sweeper
  • func NewSweeper(js jetstream.JetStream, gapWindow time.Duration, logger *slog.Logger) *Sweeper
  • func (s *Sweeper) Start(ctx context.Context)

Constants

const BufferConsumerName = "buffer-consumer"

BufferConsumerName is the durable consumer name used by the ingest worker. The Active Sweeper references this to read the AckFloor.

Functions

func StartIngestWorker

func StartIngestWorker(
	ctx context.Context, nc *nats.Conn, cache cache.Cache,
	chHost, chHTTPPort, chHTTPScheme, chUser, chPassword, chDB string,
) (func(context.Context) error, error)

Types

type EventMessage

type EventMessage struct {
	TableName         string         `json:"table_name"`
	Scope             string         `json:"scope,omitempty"`
	ReceivedTimestamp string         `json:"received_timestamp"`
	Data              map[string]any `json:"data"`
}

EventMessage is the wire format published to the MQ.

type IngestWorker

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

type Sweeper

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

Sweeper implements the Active Sweeper pattern. It runs every minute and purges messages from the JetStream stream that satisfy BOTH conditions:

• ACKed by the buffer consumer (written to ClickHouse)
• Older than the gap window (no longer needed for SSE replay)

Purge target = MIN(buffer_ack_floor + 1, gap_window_seq). This guarantees: healthy state keeps exactly gap_window of rolling data; ClickHouse down freezes purging; catastrophic outage fills to MaxBytes and triggers backpressure via DiscardNew.

func NewSweeper

func NewSweeper(js jetstream.JetStream, gapWindow time.Duration, logger *slog.Logger) *Sweeper

NewSweeper creates an Active Sweeper. TODO: (future) need leader election or shared lock to only run one instance of the sweeper in clustered mode

func (*Sweeper) Start

func (s *Sweeper) Start(ctx context.Context)

Start runs the sweep loop every minute. Blocks until ctx is cancelled.