readmodel

package module

Go to main page

Versions in this module

v0

v0.4.1

Oct 30, 2026

Changes in this version

+ const DefaultTaskPageSize

+ const FoldLockSuffix

+ var ErrFoldLocked = errors.New("readmodel: projection fold lock held by another folder")

+ func FoldLockPath(home string) string

+ func WorkerFailedMarkerPath(dbPath string) string

+ func WriteWorkerFailedMarker(dbPath string, marker WorkerFailedMarker) error

+ type EngineHealthRow struct

+ ConsecutiveFailures int

+ LastError string

+ Name string

+ State string

+ type FoldLock struct

+ func AcquireFoldLock(home string, wait time.Duration) (*FoldLock, error)

+ func (l *FoldLock) Release() error

+ type FromProject string

+ type FromStatus string

type Model

+ func (m *Model) ProjectionHealth(ctx context.Context) ProjectionHealth

+ func (m *Model) ReadOnly() bool

+ func (m *Model) TasksPage(ctx context.Context, filter TaskFilter, page TaskPage) ([]TaskRow, string, error)

type Option

+ func WithFoldLock(lock *FoldLock) Option

+ func WithReadOnly() Option

+ type ProjectionHealth struct

+ Checkpoint int64

+ Checkpointed bool

+ Cursor int64

+ Engines []EngineHealthRow

+ Head int64

+ Lag int64

+ type TaskPage struct

+ Cursor string

+ Limit int

+ type WorkerFailedMarker struct

+ FailedAt time.Time

+ Projection string

+ func ReadWorkerFailedMarker(dbPath string) (marker WorkerFailedMarker, exists bool, err error)

v0.4.0

Oct 11, 2026

v0.3.4

Oct 11, 2026

Changes in this version

+ var ProjectionHomeCallerPragmas = projectionHomeCallerPragmas()

+ func ConvergeProjectionHome(path string) error

type Option

+ func WithEngine(eng metaengine.Engine) Option

v0.3.3

Oct 7, 2026

Changes in this version

+ func DLQPathFor(modelPath string) string

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

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

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

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

+ type DeadLetters struct

+ func OpenDeadLetters(ctx context.Context, modelPath string) (*DeadLetters, error)

+ func (d *DeadLetters) Close() error

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

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

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

+ type FoldProjection struct

+ func NewFoldProjection(m *Model) FoldProjection

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

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

+ func (p FoldProjection) Name() string

+ type ProjectionHost struct

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

+ func (h *ProjectionHost) Close() error

+ type ProjectionHostOptions struct

+ BatchSize int

+ CheckpointEvery int

+ DeadLetterStore projectionhost.DeadLetterStore

+ DeadLetterThreshold int

+ MaxRestarts int

+ type WatermarkCheckpoints struct

+ Src queue.Store

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

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

v0.3.2

Oct 6, 2026

v0.3.1

Oct 5, 2026

v0.3.0

Oct 5, 2026

Changes in this version

+ const CursorConsumer

+ const DefaultBatch

+ const DefaultPoll

+ const DefaultReplayCapacity

+ const SSEHeartbeat

+ const SSETimeout

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

+ func PathFor(dbPath string) string

+ func StringPtr(s string) *string

+ type FailureText string

+ type Model struct

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

+ 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 func(*Model)

+ func WithBatch(n int) Option

+ func WithDurableCursor() Option

+ func WithPoll(d time.Duration) Option

+ func WithRowSource(rs RowSource) Option

+ type RequeueClass string

+ type RowSnapshot struct

+ CreatedAt int64

+ DedupKey string

+ Priority int

+ Project string

+ Type string

+ type RowSource interface

+ EnqueueRow func(ctx context.Context, id string) (row RowSnapshot, ok bool, err error)

+ type StoreRows struct

+ Store queue.Store

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

+ type TaskFilter struct

+ Project *string

+ Status *string

+ type TaskList struct

+ Project *string

+ Status *string

+ type TaskRow struct

+ Attempts int

+ CreatedAt int64

+ DedupKey string

+ ID string

+ LastError string

+ NotBefore int64

+ ParkedBy string

+ Priority int

+ Project string

+ Status string

+ Type string

+ UpdatedAt int64

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL