Help us shape the future of Go! Take the 2026 Go Developer Survey

readmodel

package module
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 11, 2026 License: MIT Imports: 30 Imported by: 0

Documentation

Overview

Package readmodel is the ADR-0019 S3 read model: tq's task ledger and status counts as go-cqrs-lite metaengine Store collections (planned tables), folded from the fact journal, with metaengine Watcher/ServeSSE as the live-fragment mechanism that replaces the hand journal-tailer → hub fan-out (ADR-0003 Phase D read path) once the S3 flip lands.

The Model opens its OWN sqlite database beside the queue DB (path decision: separate file — the projection is disposable and rebuildable from the journal, and a second writer connection on the queue file would contend with the store's single serialized writer). Feeding is journal-first: every lifecycle fact maps to an event applied through metaengine Store.ApplyRecord, so the collections are a fold projection of the journal exactly like every other tq consumer (ADR-0001).

THIN-ENQUEUE SIDE CHANNEL (interim): the upstream engine's task.enqueued fact carries only {project, type} today — priority, dedup key and the creation timestamp are not in the journal (S1 divergence, M4 ratification memo pending). While that holds, the Projector fills those fields from the store row on first sight of a task (RowSource; nil disables). The side channel dies with the upstream detail growth: the fold for task.enqueued then reads everything from the fact.

The package is read-only over the queue: it never mutates task state.

Index

Constants

View Source 
const (
	// DefaultPoll is the journal tail interval (webui parity).
	DefaultPoll = 500 * time.Millisecond
	// DefaultBatch is the per-poll fact batch cap (webui tailBatchLimit
	// parity): the pump only needs to converge, so a longer burst
	// finishes over the next ticks.
	DefaultBatch = 1000
	// DefaultReplayCapacity bounds the SSE replay journal (Last-Event-ID
	// reconnection window).
	DefaultReplayCapacity = 1024
	// SSEHeartbeat is the keepalive cadence for the events stream.
	SSEHeartbeat = 15 * time.Second
	// SSETimeout caps one stream's lifetime so server shutdowns and
	// stuck clients reclaim their goroutines.
	SSETimeout = 30 * time.Minute
	// CursorConsumer is the watermarks-table key the durable cursor
	// checkpoints under (queue.Store.SaveWatermark): the projection
	// resumes from it instead of replaying the whole journal on every
	// open. One key per queue db — the projection file is PathFor-derived,
	// so one store has exactly one durable projection.
	CursorConsumer = "readmodel"
)

Defaults for the projection pump and the SSE surface.

Variables

View Source 
var ErrNoSource = errors.New("readmodel: nil journal source")

ErrNoSource reports an Open call without a journal source: the model is a projection, and without a journal to fold there is nothing to serve.

View Source 
var ProjectionHomeCallerPragmas = projectionHomeCallerPragmas()

Functions

func ConvergeProjectionHome added in v0.3.4 

func ConvergeProjectionHome(path string) error

ConvergeProjectionHome reconciles an on-disk projection home with the current TaskRow model BEFORE any engine opens the file: a projection written by an older model (a planned-table column the current fold inserts does not exist) makes every task.enqueued fold fail, the metaengine's CREATE TABLE IF NOT EXISTS never alters an existing table, and the projection worker crash-loops on the first fact until it exhausts its restart budget — the Oct 2026 production outage where meta_planned_tasks predated not_before.

The projection is DISPOSABLE by contract (delete the file to force a full journal replay), so a divergent home is snapshotted aside — renamed, never deleted — and the next open recreates it from the current model; WithDurableCursor already replays from zero exactly once when the surviving watermark points past an empty projection. A healthy or absent home is left untouched.

func DLQPathFor added in v0.3.3 

func DLQPathFor(modelPath string) string

DLQPathFor derives the poison-fact sidecar path beside a projection database: one DLQ per projection home, discovered by both the serve wiring (OpenDeadLetters) and the doctor's projection section.

func EventIDToSeq added in v0.3.3 

func EventIDToSeq(eventID id.EventID) (int64, bool)

EventIDToSeq decodes a sequence-derived event ID back to its journal sequence; ok is false for any ID not minted by this layout (a random or foreign cursor), which callers must treat as "no position".

func Lag added in v0.3.3 

func Lag(h *projectionhost.Host) time.Duration

Lag reports the host's aggregate fold lag — the doctor/health surface (M09) reads it through the host the caller owns.

func PathFor 

func PathFor(dbPath string) string

PathFor derives the projection database path beside a queue database: one shared projection file per queue db (the sqliteengine opens it WAL + busy_timeout, so serve/api/stats processes can share it).

func SeqToEventID added in v0.3.3 

func SeqToEventID(seq int64) (id.EventID, error)

SeqToEventID encodes a journal sequence as the synthetic event ID the cqrs journal stamps on every fact event. Exported for the checkpoint adapter's consumers (doctor/health surfaces, composition root).

func StatusCounts added in v0.3.3 

func StatusCounts(ctx context.Context, model *Model, store queue.Store) (map[task.Status]int, error)

StatusCounts is the one seam both servers read status counts through: from the model when one is supplied, from the queue store otherwise. The dashboard server (internal/webui) and the machine API (internal/httpapi) shared this exact branch pair before it landed here.

func StringPtr 

func StringPtr(s string) *string

StringPtr is a small helper for building a TaskFilter inline.

Types

type DeadLetters added in v0.3.3 

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

DeadLetters is the projection host's poison-fact sidecar over one sqlite file beside the projection database (DLQPathFor). Serve opens it for the fold host's DeadLetterStore; doctor surfaces read counts and recent entries from the same file. Open creates the file and the dead-letter table; Close releases the handle.

func OpenDeadLetters added in v0.3.3 

func OpenDeadLetters(ctx context.Context, d config.Deployment) (*DeadLetters, error)

OpenDeadLetters opens (creating if needed) the DLQ sidecar for the projection home described by d (ADR-0022: the deployment is the ONE pragma source — the sidecar follows the same sync tier as both homes). The handle follows the queue's sqlite posture: WAL, busy timeout, MaxOpenConns(1).

func (*DeadLetters) Close added in v0.3.3 

func (d *DeadLetters) Close() error

Close releases the sidecar's handle.

func (*DeadLetters) Count added in v0.3.3 

func (d *DeadLetters) Count(ctx context.Context) (int64, error)

Count reports the stored poison facts across every projection.

func (*DeadLetters) Recent added in v0.3.3 

func (d *DeadLetters) Recent(ctx context.Context, limit int) ([]projectionhost.DeadLetterEntry, error)

Recent returns up to limit dead-letter entries, newest failure first.

func (*DeadLetters) Store added in v0.3.3 

func (d *DeadLetters) Store() *projectionhost.SQLiteDeadLetterStore

Store returns the platform DLQ for the projection host's DeadLetterStore option; the concrete store satisfies the interface.

type FailureText 

type FailureText string

FailureText is a failed attempt's error text — a named string so event structs carry at most one bare string (the fold key).

type FoldProjection added in v0.3.3 

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

FoldProjection feeds platform-delivered fact events through the same fold the hand pump uses (Model.apply) — one fold, two drivers.

func NewFoldProjection added in v0.3.3 

func NewFoldProjection(m *Model) FoldProjection

NewFoldProjection wraps the model's fold as a platform projection.

func (FoldProjection) EventTypes added in v0.3.3 

func (p FoldProjection) EventTypes() []event.Type

EventTypes delivers every fact family, matching the hand pump's deliver-everything behavior exactly: the fold's default arm skips the no-ledger families, and a future family must reach Handle without an edit here.

func (FoldProjection) Handle added in v0.3.3 

func (p FoldProjection) Handle(ctx context.Context, evt event.Event) error

Handle reconstructs the fact from the platform event and folds it. The payload was stamped by the cqrs journal's factEvent (event.New with the JSON codec), so DecodePayloadAuto is the matching decode path.

func (FoldProjection) Name added in v0.3.3 

func (p FoldProjection) Name() string

Name is the projection identity and the watermarks consumer.

type Model 

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

Model is the metaengine-backed read model over one queue's fact journal: the tasks collection (planned table) plus the Watcher/ServeSSE live surface. Create with Open, pump with Run (or CatchUp for one pass), read with Tasks/StatusCounts, stream with EventsHandler.

func Open 

func Open(ctx context.Context, path string, src queue.Store, opts ...Option) (*Model, error)

Open creates the Model over its own sqlite database file (the projection is disposable; delete the file to force a full journal replay on next open). src is the journal source — the same queue.Store the dashboards read; it is used read-only (Facts, HeadSeq, Get). ctx covers the durable-cursor load (watermark read + replay guard), not the pump (Run carries its own).

func (*Model) CatchUp 

func (m *Model) CatchUp(ctx context.Context) error

CatchUp applies every journal fact up to the current head once.

func (*Model) Close 

func (m *Model) Close() error

Close releases the watcher and the projection database. The journal source is owned by the caller.

func (*Model) EventsHandler 

func (m *Model) EventsHandler() http.Handler

EventsHandler serves the live ledger as Server-Sent Events: every fold update streams as one JSON TaskRow, with `id: <seq>` for Last-Event-ID reconnection (metaengine's replay journal). This is the Watcher/ServeSSE mechanism the S3 flip installs behind the dashboard fragments.

func (*Model) JournalCursor 

func (m *Model) JournalCursor() int64

JournalCursor reports the applied journal watermark: the seq of the last fact folded into the collections (0 before the first apply). Live consumers use it as the change-notification sequence, so a notification carries the same journal watermark the hand tailer used to report.

func (*Model) ProjectCounts 

func (m *Model) ProjectCounts(ctx context.Context) (map[string]map[string]int, error)

ProjectCounts counts the ledger rows per project per status — the same matrix the store-side ProjectCounts GROUP BY serves (queue.Store vocabulary parity for the projection).

func (*Model) Run 

func (m *Model) Run(ctx context.Context) error

Run pumps the journal until ctx is cancelled: catch up to the head, then tail for new facts. Without WithDurableCursor the cursor is in-process — a restarted model replays the journal from the beginning, which the folds converge on (every fold is an upsert keyed by task id); with it, the model resumes from the persisted checkpoint instead.

func (*Model) Stats 

func (m *Model) Stats(
	ctx context.Context,
	filter TaskFilter,
) (map[string]int, map[string]map[string]int, error)

Stats reads the status counts and the per-project-per-status counts as SQL GROUP BY pushdowns over the planned tasks table — O(statuses + statuses×groups) engine work, zero rows loaded. The counts derive from the same folded rows every Tasks read serves, so they are exact by construction (no separate counter state to drift on reclaims, rescues or dismissals, which have no statically knowable from-status).

filter narrows both matrices exactly like a Tasks read would: a Status filter leaves only that status's counts, a Project filter only that project's.

func (*Model) StatusCounts 

func (m *Model) StatusCounts(ctx context.Context) (map[string]int, error)

StatusCounts counts the ledger rows per status — the GROUP BY pushdown behind the stats surfaces (dashboard chips, /api/stats, tq stats).

func (*Model) Tasks 

func (m *Model) Tasks(ctx context.Context, filter TaskFilter) ([]TaskRow, error)

Tasks reads the ledger under the filter, newest creation first — the planned-table pushdown scan. Filters ride the documented TypedReader surface (conditional WithFilter options): the query-input dispatch (Store.ExecuteCtx) cannot express optional filters — a nil *string input field binds as a typed-nil interface and lands in SQL as `= NULL` (metaengine v4.14.0, extractValueByName), so every read here goes through the reader.

func (*Model) Watch 

func (m *Model) Watch(ctx context.Context) <-chan TaskRow

Watch subscribes to live ledger changes: every fold update arrives as the folded TaskRow (buffered, drop-oldest on a slow consumer). The subscription ends with ctx. This is the Watcher half of the S3 live fragments; EventsHandler is its SSE transport.

func (*Model) WatchSeq 

func (m *Model) WatchSeq(ctx context.Context) <-chan metaengine.SeqValue[TaskRow]

WatchSeq is Watch with the projection write sequence attached. The dashboard's notification pump consumes this: each delivery replaces one journal-tailer poll cycle — the change signal, not the payload.

type Option 

type Option func(*Model)

Option configures the Model.

func WithBatch 

func WithBatch(n int) Option

WithBatch overrides the per-poll fact batch cap.

func WithDurableCursor 

func WithDurableCursor() Option

WithDurableCursor checkpoints the projection cursor into the queue db's watermarks table after every applied batch and resumes from it on open: a restarted model folds only the facts appended since its last checkpoint instead of replaying the whole journal (the 9.5k-fact restart replay the adoption review measured). An empty projection under a nonzero checkpoint replays from zero once, so the documented "delete the file to force a replay" escape hatch keeps working. Checkpoint writes are monotonic upserts; a failed write is logged and retried on the next batch — a missed checkpoint only costs replay on the next restart, never a skipped fact.

func WithEngine added in v0.3.4 

func WithEngine(eng metaengine.Engine) Option

WithEngine adopts a caller-built engine instead of constructing one: the composition root opens the projection-home engine ONCE and hands it over (the single tq-owned constructor call per serve run). From adoption on the model owns the engine exactly as for a self-opened one — the FromDSN constructors mark their engine as the DB owner, so Close tears the connection down with the model.

func WithPoll 

func WithPoll(d time.Duration) Option

WithPoll overrides the journal tail interval.

func WithRowSource 

func WithRowSource(rs RowSource) Option

WithRowSource overrides the thin-enqueue side channel (nil disables it). The default adapts the journal source's Get.

type ProjectionHost added in v0.3.3 

type ProjectionHost struct {
	*projectionhost.Host
	// contains filtered or unexported fields
}

ProjectionHost pairs the platform host with the live tails' lifecycle: the platform never closes its subscriber (the Bus doc leaves cleanup to the owner via io.Closer), so Close here is the tail-ownership half and Stop stays the platform half. Embedding keeps Start, Stop, and Register on the wrapper.

func NewProjectionHost added in v0.3.3 

func NewProjectionHost(src queue.Store, m *Model, opts ProjectionHostOptions) (*ProjectionHost, error)

NewProjectionHost builds the managed host over the fact journal with the model's fold registered as its projection. Start/Stop stay with the caller (the composition root owns the lifetime); the journal is the queue store itself, read-only (Facts). The live tail after the initial drain polls the journal at DefaultPoll — the queue journal has no bus, and the webui contract is a tail, not a one-shot drain.

func (*ProjectionHost) Close added in v0.3.3 

func (h *ProjectionHost) Close() error

Close stops every tail goroutine the subscriber spawned (one per worker generation: a worker restart re-subscribes) and waits for their exit, so a closed journal never sees a late poll.

type ProjectionHostOptions added in v0.3.3 

type ProjectionHostOptions struct {
	// CheckpointEvery batches live-phase checkpoint saves; 0 keeps the
	// platform default.
	CheckpointEvery int
	// BatchSize caps the journal read per drain loop; 0 keeps the
	// platform default.
	BatchSize int
	// MaxRestarts bounds the crash-restart budget per worker; 0 keeps
	// the platform default (5).
	MaxRestarts int
	// DeadLetterStore captures poison facts after Threshold failures;
	// nil disables the DLQ.
	DeadLetterStore projectionhost.DeadLetterStore
	// DeadLetterThreshold is the failure count that routes a fact to the
	// DLQ; 0 keeps the platform default when a store is configured.
	DeadLetterThreshold int
}

ProjectionHostOptions are the platform knobs the composition root may tune; zero values take the projectionhost defaults except where the queue's operating posture demands one (checkpoint batching on).

type RequeueClass 

type RequeueClass string

RequeueClass is a requeue evidence class — a named string so the engine's type-based key inference keeps ID unambiguous (same rule as FailureText).

type RowSnapshot 

type RowSnapshot struct {
	Project   string
	Type      string
	Priority  int
	DedupKey  string
	CreatedAt int64
}

RowSnapshot is the enqueue-time slice of a task row.

type RowSource 

type RowSource interface {
	// EnqueueRow reports the enqueue-time projection of one task; a
	// miss (task already compacted away) returns ok=false and the fold
	// keeps the fact-detail fallbacks.
	EnqueueRow(ctx context.Context, id string) (row RowSnapshot, ok bool, err error)
}

RowSource supplies the enqueue-time fields the current engine's thin task.enqueued fact omits (project/type fall back to the fact detail; priority, dedup key and creation time exist only on the row). It dies with the upstream detail growth (M4 memo) — the fold then reads everything from the fact.

type StoreRows 

type StoreRows struct {
	Store queue.Store
}

StoreRows adapts a queue.Store into a RowSource via Get.

func (StoreRows) EnqueueRow 

func (s StoreRows) EnqueueRow(ctx context.Context, id string) (RowSnapshot, bool, error)

EnqueueRow implements RowSource.

type TaskFilter 

type TaskFilter struct {
	Status  *string
	Project *string
}

TaskFilter selects the Tasks read. Nil fields mean "no filter"; the values are the raw queue vocabulary (task.Status string, project name). Deliberately a separate type from TaskList (queries.go): TaskList is the metaengine query-input declaration carrier and MUST NOT dispatch reads through it (typed-nil binding hazard), while TaskFilter is the caller-facing filter the Model reads satisfy.

type TaskList 

type TaskList struct {
	Status  *string
	Project *string
}

TaskList is the query's declared input type — a declaration carrier only. Reads MUST NOT dispatch through it (Store.ExecuteCtx binds nil pointer fields as typed-nil interfaces → `= NULL`, and empty strings as real filters): Model.Tasks reads via the TypedReader instead.

type TaskRow 

type TaskRow struct {
	ID string `json:"id"`
	// Project is the task's repo/project scope (empty = unscoped).
	Project string `json:"project"`
	// Type is the executor task type (sh, agent, review, ...).
	Type string `json:"type"`
	// Status is the task.Status value (pending, running, completed,
	// dead, cancelled).
	Status string `json:"status"`
	// Priority is the STORED priority (ADR-0015); aging is scheduling
	// and never mutates it.
	Priority int `json:"priority"`
	// Attempts counts claimed leases — every claim is one attempt start.
	Attempts int `json:"attempts"`
	// CreatedAt/UpdatedAt are unix millis — the task-row storage format.
	CreatedAt int64 `json:"created_at"`
	UpdatedAt int64 `json:"updated_at"`
	// DedupKey is the harvest/enqueue identity (empty for ad-hoc tasks).
	DedupKey string `json:"dedup_key"`
	// LastError is the most recent failure text (empty when none).
	LastError string `json:"last_error"`
	// NotBefore is the earliest claim time (unix millis) of the latest
	// CLASS park (from the requeue evidence), zero when none is in
	// force. The store's backoff-after-failure parks are deliberately
	// NOT mirrored: task.failed facts carry no class or window, so the
	// projection records only what the journal proves.
	NotBefore int64 `json:"not_before"`
	// ParkedBy is the latest requeue class (queue.RequeueClass*;
	// "unknown" for legacy facts, empty until the first requeue). A
	// budget-parked task is Status==pending with ParkedBy=="budget" and
	// a future NotBefore. Cleared on claim and on failure (a burned
	// attempt is a retry, not a park).
	ParkedBy string `json:"parked_by"`
}

TaskRow is one ledger row in the tasks collection: the fields the queue views (board, table, stats) render, folded from the journal.

type WatermarkCheckpoints added in v0.3.3 

type WatermarkCheckpoints struct {
	Src queue.Store
}

WatermarkCheckpoints adapts the queue's watermarks table (one monotonic integer cursor per consumer) to the platform's event.CheckpointStore. The projection's Name IS the watermarks consumer, so the host-managed pump and the in-model durable cursor (WithDurableCursor) share one checkpoint slot by construction.

func (WatermarkCheckpoints) Load added in v0.3.3 

func (w WatermarkCheckpoints) Load(ctx context.Context, name string) (event.Checkpoint, error)

Load returns the stored checkpoint; no row (or a zero seq) is the zero Checkpoint — the platform's "no prior progress" shape, replaying from the journal start.

func (WatermarkCheckpoints) Save added in v0.3.3 

func (w WatermarkCheckpoints) Save(ctx context.Context, name string, cp event.Checkpoint) error

Save stores the checkpoint's event ID as the consumer's seq. A non-sequence-derived ID cannot come from this journal's events and is refused instead of silently wedging the cursor.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.