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
- Variables
- func PathFor(dbPath string) string
- func StringPtr(s string) *string
- type FailureText
- type Model
- func (m *Model) CatchUp(ctx context.Context) error
- func (m *Model) Close() error
- func (m *Model) EventsHandler() http.Handler
- func (m *Model) JournalCursor() int64
- func (m *Model) ProjectCounts(ctx context.Context) (map[string]map[string]int, error)
- func (m *Model) Run(ctx context.Context) error
- func (m *Model) Stats(ctx context.Context, filter TaskFilter) (map[string]int, map[string]map[string]int, error)
- func (m *Model) StatusCounts(ctx context.Context) (map[string]int, error)
- func (m *Model) Tasks(ctx context.Context, filter TaskFilter) ([]TaskRow, error)
- func (m *Model) Watch(ctx context.Context) <-chan TaskRow
- func (m *Model) WatchSeq(ctx context.Context) <-chan metaengine.SeqValue[TaskRow]
- type Option
- type RequeueClass
- type RowSnapshot
- type RowSource
- type StoreRows
- type TaskFilter
- type TaskList
- type TaskRow
Constants ¶
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 ¶
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.
Functions ¶
Types ¶
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 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 ¶
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) Close ¶
Close releases the watcher and the projection database. The journal source is owned by the caller.
func (*Model) EventsHandler ¶
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 ¶
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 ¶
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 ¶
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 ¶
StatusCounts counts the ledger rows per status — the GROUP BY pushdown behind the stats surfaces (dashboard chips, /api/stats, tq stats).
func (*Model) Tasks ¶
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 ¶
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.
type Option ¶
type Option func(*Model)
Option configures the Model.
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 WithRowSource ¶
WithRowSource overrides the thin-enqueue side channel (nil disables it). The default adapts the journal source's Get.
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 ¶
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 ¶
StoreRows adapts a queue.Store into a RowSource via Get.
func (StoreRows) EnqueueRow ¶
EnqueueRow implements RowSource.
type TaskFilter ¶
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 ¶
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.
Source Files