159 lines
6.7 KiB
Go
159 lines
6.7 KiB
Go
package pgstore
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
// events.go: the frame history a book's event stream is resumed from.
|
|
//
|
|
// Frames are minted by the WRITER, inside the transaction that caused them, and not by the
|
|
// connection that sends them. That is forced rather than chosen: `Last-Event-ID` has to mean the
|
|
// same thing however a frame was carried (canon §EventEnvelope), so two people watching one book
|
|
// must see one frame under one id — which a per-connection counter cannot give and a stored one
|
|
// gives for free.
|
|
//
|
|
// It does NOT make the reporting database an event store. What is kept is a short live buffer,
|
|
// pruned as it grows: enough to continue a client that blinked, never enough to replay a history.
|
|
// A client that presents an id older than the buffer is told `resync_required` and re-reads, which
|
|
// is the same answer the contract gives after a wholesale replacement.
|
|
|
|
// Frame names, matching the contract's event table. Only HISTORY frames appear here: `hello`, `end`,
|
|
// `resync_required` and `session_ended` belong to a CONNECTION, carry the id of the last history
|
|
// frame and consume none of their own. (The fourth arrived with canon 0.8.0 — a revocation or an
|
|
// expired session ends the stream; see httpapi.eventSessionEnded.)
|
|
const (
|
|
FrameStatus = "status"
|
|
FrameProgress = "progress"
|
|
FrameChapter = "chapter"
|
|
FrameNote = "note"
|
|
FrameBank = "bank"
|
|
)
|
|
|
|
// bufferFrames is how many frames of one book are kept. It bounds the table rather than promising a
|
|
// window: the contract says explicitly that the buffer's size is not declared and that a client MUST
|
|
// NOT depend on any frame being resent.
|
|
const bufferFrames = 512
|
|
|
|
// Frame is one stored frame.
|
|
type Frame struct {
|
|
Position int64
|
|
Event string
|
|
// Data is the frame's payload exactly as it goes on the wire, built in the transaction that
|
|
// caused it. Re-deriving it at send time would answer a later state under an earlier id.
|
|
Data json.RawMessage
|
|
}
|
|
|
|
// emitFrame appends one frame inside the caller's transaction.
|
|
//
|
|
// The caller passes the payload WITHOUT the two book-scope numbers: every frame carries them
|
|
// (canon §EventBase), so they are stamped here from the row this transaction has already moved —
|
|
// which is also why this must be called AFTER the write it announces and never before.
|
|
func emitFrame(ctx context.Context, tx pgx.Tx, bookID, event string, payload map[string]any) error {
|
|
var position, revision int64
|
|
var structure int
|
|
err := tx.QueryRow(ctx, `
|
|
update books set event_position = event_position + 1
|
|
where id = $1
|
|
returning event_position, revision, structure_version`, bookID).
|
|
Scan(&position, &revision, &structure)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil // the book is gone; nothing is watching it
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: mint frame position: %w", err)
|
|
}
|
|
if payload == nil {
|
|
payload = map[string]any{}
|
|
}
|
|
payload["revision"] = revision
|
|
payload["structure_version"] = structure
|
|
data, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: encode frame: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
insert into book_events (book_id, position, event, data) values ($1, $2, $3, $4)`,
|
|
bookID, position, event, data); err != nil {
|
|
return fmt.Errorf("pgstore: append frame: %w", err)
|
|
}
|
|
// Pruned here rather than by a sweep: the buffer is bounded per book by construction, and a
|
|
// retention pass that has to be scheduled is a second thing to get wrong.
|
|
if _, err := tx.Exec(ctx,
|
|
`delete from book_events where book_id = $1 and position <= $2`,
|
|
bookID, position-bufferFrames); err != nil {
|
|
return fmt.Errorf("pgstore: prune frames: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// StreamState is what a stream needs to know about a book before and between frames.
|
|
type StreamState struct {
|
|
// Position is the book's last history frame, or 0 when it has produced none. History numbering
|
|
// starts at 1, so 0 can never collide with a real frame and is what a connection frame of a book
|
|
// with no history carries.
|
|
Position int64
|
|
Revision int64
|
|
StructureVersion int
|
|
// AtRest is "no run is live and no intake is in flight" — the condition the contract ends a
|
|
// stream on, and the one under which a presented id at or past Position is answered 204.
|
|
//
|
|
// ⚠ A materialization still owed is intake in flight. The tree lands a moment AFTER the parse
|
|
// commits, and a run's text a moment after the run closes; calling those moments rest sent `end`
|
|
// and then answered the browser's own reconnect 204, so the client stopped watching exactly when
|
|
// the chapters were about to appear.
|
|
AtRest bool
|
|
// Oldest is the oldest frame still buffered, or 0 when nothing is. A client presenting an id
|
|
// below it cannot be continued and is told to re-sync.
|
|
Oldest int64
|
|
}
|
|
|
|
// nothingIsRunning is the canon's "no run live and no intake in flight" (§streamBookEvents), over a
|
|
// `books b`. One string because both readers want the SAME condition and must not drift: the stream
|
|
// ends on it, and the materializer takes a book only while it holds (BooksOwedReadModel). What the
|
|
// two differ on is the debt beside it, which is not part of this.
|
|
const nothingIsRunning = `b.status not in ('uploading', 'parsing')
|
|
and not exists (select 1 from runs r where r.book_id = b.id and r.finished_at is null)`
|
|
|
|
// ReadStream reports a book's stream state, or ErrNoBook when the caller may not see it.
|
|
func (s *Store) ReadStream(ctx context.Context, userID, bookID string) (StreamState, error) {
|
|
const q = `
|
|
select b.event_position, b.revision, b.structure_version,
|
|
` + nothingIsRunning + ` and b.read_model_owed_at is null,
|
|
coalesce((select min(position) from book_events e where e.book_id = b.id), 0)
|
|
from books b where b.id = $1 and b.owner_id = $2`
|
|
var out StreamState
|
|
err := s.pool.QueryRow(ctx, q, bookID, userID).
|
|
Scan(&out.Position, &out.Revision, &out.StructureVersion, &out.AtRest, &out.Oldest)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return StreamState{}, ErrNoBook
|
|
}
|
|
if err != nil {
|
|
return StreamState{}, fmt.Errorf("pgstore: read stream state: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ReadFrames returns the buffered frames after a position, oldest first.
|
|
func (s *Store) ReadFrames(ctx context.Context, bookID string, after int64, limit int) ([]Frame, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
select position, event, data from book_events
|
|
where book_id = $1 and position > $2 order by position limit $3`, bookID, after, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgstore: read frames: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []Frame
|
|
for rows.Next() {
|
|
var f Frame
|
|
if err := rows.Scan(&f.Position, &f.Event, &f.Data); err != nil {
|
|
return nil, fmt.Errorf("pgstore: scan frame: %w", err)
|
|
}
|
|
out = append(out, f)
|
|
}
|
|
return out, rows.Err()
|
|
}
|