// Package store provides database access helpers for the mitbringsl backend. // Op-log operations: idempotent append + pull, with inline LWW projection. package store import ( "context" "encoding/json" "fmt" "log/slog" "strings" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" hlcsync "github.com/mitbringsl/backend/internal/sync" ) // OpType constants – must stay in sync with op_log.op_type check. const ( OpListCreate = "list_create" OpListRename = "list_rename" OpListDelete = "list_delete" OpItemAdd = "item_add" OpItemUpdate = "item_update" OpItemRemove = "item_remove" ) // maxPushBatch is the maximum number of ops accepted in a single push request. const maxPushBatch = 100 // pullPageSize is the maximum number of ops returned by a single pull. const pullPageSize = 500 // IncomingOp is a single operation sent by the client. type IncomingOp struct { ClientSeq int64 `json:"client_seq"` OpType string `json:"op_type"` TargetID uuid.UUID `json:"target_id"` HLCTS int64 `json:"hlc_ts"` Payload json.RawMessage `json:"payload"` } // OpResult is returned for each op after a push – maps client_seq → server seq + hlc. type OpResult struct { ClientSeq int64 `json:"client_seq"` Seq int64 `json:"seq"` HLCTS int64 `json:"hlc_ts"` } // Op is a fully hydrated op_log row, returned on pull. type Op struct { Seq int64 `json:"seq"` ClientID uuid.UUID `json:"client_id"` OpType string `json:"op_type"` TargetID uuid.UUID `json:"target_id"` Payload json.RawMessage `json:"payload"` ClientSeq int64 `json:"client_seq"` HLCTS int64 `json:"hlc_ts"` CreatedAt time.Time `json:"created_at"` } // OpStore handles op_log reads and writes. type OpStore struct { pool *pgxpool.Pool } // NewOpStore creates an OpStore backed by the given pool. func NewOpStore(pool *pgxpool.Pool) *OpStore { return &OpStore{pool: pool} } // MaxPushBatch exposes the configured limit so handlers can validate early. func MaxPushBatch() int { return maxPushBatch } // AppendOps idempotently inserts the given ops for (listID, userID, clientID) // into op_log, applies the LWW projection to items/lists within the same // transaction, and returns one OpResult per input op. // // Ops that already exist (UNIQUE client_id+client_seq) are silently skipped; // the previously stored seq+hlc is returned for them. func (s *OpStore) AppendOps( ctx context.Context, listID, userID, clientID uuid.UUID, ops []IncomingOp, ) ([]OpResult, error) { results := make([]OpResult, len(ops)) tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{}) if err != nil { return nil, fmt.Errorf("opstore: begin tx: %w", err) } defer func() { _ = tx.Rollback(ctx) }() for i, op := range ops { // Tick the server HLC, incorporating the client's value. serverHLC := hlcsync.Tick(op.HLCTS) // Validate op_type. if !isKnownOpType(op.OpType) { return nil, fmt.Errorf("opstore: unknown op_type %q at index %d", op.OpType, i) } payloadBytes := []byte(op.Payload) if len(payloadBytes) == 0 { payloadBytes = []byte("{}") } var seq int64 var storedHLC int64 err := tx.QueryRow(ctx, ` INSERT INTO op_log (list_id, user_id, client_id, op_type, target_id, payload, client_seq, hlc_ts) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (client_id, client_seq) DO NOTHING RETURNING seq, hlc_ts`, listID, userID, clientID, op.OpType, op.TargetID, payloadBytes, op.ClientSeq, serverHLC, ).Scan(&seq, &storedHLC) if err == pgx.ErrNoRows { // Already stored – look up the existing row. if err2 := tx.QueryRow(ctx, ` SELECT seq, hlc_ts FROM op_log WHERE client_id = $1 AND client_seq = $2`, clientID, op.ClientSeq, ).Scan(&seq, &storedHLC); err2 != nil { return nil, fmt.Errorf("opstore: lookup existing op: %w", err2) } // Skip projection for already-applied ops. results[i] = OpResult{ClientSeq: op.ClientSeq, Seq: seq, HLCTS: storedHLC} continue } if err != nil { return nil, fmt.Errorf("opstore: insert op: %w", err) } // Apply the projection within the same transaction. if err := applyProjection(ctx, tx, listID, op, storedHLC); err != nil { return nil, fmt.Errorf("opstore: apply projection for op %d: %w", i, err) } results[i] = OpResult{ClientSeq: op.ClientSeq, Seq: seq, HLCTS: storedHLC} } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("opstore: commit: %w", err) } return results, nil } // PullOps returns up to pullPageSize ops for listID with seq > since. // The caller can detect more pages by comparing len(result) == pullPageSize. func (s *OpStore) PullOps(ctx context.Context, listID uuid.UUID, since int64) ([]Op, error) { rows, err := s.pool.Query(ctx, ` SELECT seq, client_id, op_type, target_id, payload, client_seq, hlc_ts, created_at FROM op_log WHERE list_id = $1 AND seq > $2 ORDER BY seq ASC LIMIT $3`, listID, since, pullPageSize, ) if err != nil { return nil, fmt.Errorf("opstore: pull ops: %w", err) } defer rows.Close() var ops []Op for rows.Next() { var o Op if err := rows.Scan( &o.Seq, &o.ClientID, &o.OpType, &o.TargetID, &o.Payload, &o.ClientSeq, &o.HLCTS, &o.CreatedAt, ); err != nil { return nil, fmt.Errorf("opstore: scan op row: %w", err) } ops = append(ops, o) } return ops, rows.Err() } // --------------------------------------------------------------------------- // Projection (LWW register per entity, Tombstones) // --------------------------------------------------------------------------- // applyProjection updates the items/lists projection tables based on op. // All writes use LWW semantics: UPDATE ... WHERE hlc_ts < incomingHLC. // Tombstoned rows (deleted_at IS NOT NULL) are never revived. func applyProjection(ctx context.Context, tx pgx.Tx, listID uuid.UUID, op IncomingOp, serverHLC int64) error { switch op.OpType { case OpListCreate: return projListCreate(ctx, tx, listID, op, serverHLC) case OpListRename: return projListRename(ctx, tx, op, serverHLC) case OpListDelete: return projListDelete(ctx, tx, op, serverHLC) case OpItemAdd: return projItemAdd(ctx, tx, listID, op, serverHLC) case OpItemUpdate: return projItemUpdate(ctx, tx, op, serverHLC) case OpItemRemove: return projItemRemove(ctx, tx, op, serverHLC) default: // Unknown op types are logged but do not fail the transaction. slog.Warn("applyProjection: unhandled op_type", "op_type", op.OpType) return nil } } // payloadField extracts a string field from a JSON payload. func payloadField(payload json.RawMessage, field string) string { var m map[string]json.RawMessage if err := json.Unmarshal(payload, &m); err != nil { return "" } v, ok := m[field] if !ok { return "" } var s string if err := json.Unmarshal(v, &s); err != nil { return "" } return s } func payloadBool(payload json.RawMessage, field string) (bool, bool) { var m map[string]json.RawMessage if err := json.Unmarshal(payload, &m); err != nil { return false, false } v, ok := m[field] if !ok { return false, false } var b bool if err := json.Unmarshal(v, &b); err != nil { return false, false } return b, true } func projListCreate(ctx context.Context, tx pgx.Tx, listID uuid.UUID, op IncomingOp, hlc int64) error { name := payloadField(op.Payload, "name") ownerID := op.TargetID // for list_create, target_id IS the list id; owner comes from payload ownerStr := payloadField(op.Payload, "owner_id") if ownerStr != "" { if id, err := uuid.Parse(ownerStr); err == nil { ownerID = id } } // Use target_id as the list id so the client can refer to it. _, err := tx.Exec(ctx, ` INSERT INTO lists (id, name, owner_id, hlc_ts) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO NOTHING`, op.TargetID, name, ownerID, hlc, ) return err } func projListRename(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error { name := payloadField(op.Payload, "name") if name == "" { return nil } _, err := tx.Exec(ctx, ` UPDATE lists SET name = $1, hlc_ts = $2, updated_at = now() WHERE id = $3 AND hlc_ts < $2 AND deleted_at IS NULL`, name, hlc, op.TargetID, ) return err } func projListDelete(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error { _, err := tx.Exec(ctx, ` UPDATE lists SET deleted_at = now(), hlc_ts = $1, updated_at = now() WHERE id = $2 AND hlc_ts < $1 AND deleted_at IS NULL`, hlc, op.TargetID, ) return err } func projItemAdd(ctx context.Context, tx pgx.Tx, listID uuid.UUID, op IncomingOp, hlc int64) error { name := payloadField(op.Payload, "name") quantity := payloadField(op.Payload, "quantity") _, err := tx.Exec(ctx, ` INSERT INTO items (id, list_id, name, quantity, hlc_ts) VALUES ($1, $2, $3, NULLIF($4,''), $5) ON CONFLICT (id) DO NOTHING`, op.TargetID, listID, name, quantity, hlc, ) if err != nil { return err } // Update suggestions table. if name != "" { return upsertItemName(ctx, tx, strings.ToLower(name)) } return nil } func projItemUpdate(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error { // Build a dynamic UPDATE only touching provided payload fields. setClauses := []string{"hlc_ts = $1", "updated_at = now()"} args := []any{hlc, op.TargetID} if name := payloadField(op.Payload, "name"); name != "" { args = append(args, name) setClauses = append(setClauses, fmt.Sprintf("name = $%d", len(args))) } if qty := payloadField(op.Payload, "quantity"); qty != "" { args = append(args, qty) setClauses = append(setClauses, fmt.Sprintf("quantity = $%d", len(args))) } if checked, ok := payloadBool(op.Payload, "checked"); ok { args = append(args, checked) setClauses = append(setClauses, fmt.Sprintf("checked = $%d", len(args))) if checked { setClauses = append(setClauses, "checked_at = now()") } else { setClauses = append(setClauses, "checked_at = NULL") } } query := fmt.Sprintf( `UPDATE items SET %s WHERE id = $2 AND hlc_ts < $1 AND deleted_at IS NULL`, strings.Join(setClauses, ", "), ) _, err := tx.Exec(ctx, query, args...) return err } func projItemRemove(ctx context.Context, tx pgx.Tx, op IncomingOp, hlc int64) error { _, err := tx.Exec(ctx, ` UPDATE items SET deleted_at = now(), hlc_ts = $1, updated_at = now() WHERE id = $2 AND hlc_ts < $1 AND deleted_at IS NULL`, hlc, op.TargetID, ) return err } // upsertItemName increments the usage count for a lowercased item name in the // suggestions table. Called within the projection transaction. func upsertItemName(ctx context.Context, tx pgx.Tx, name string) error { _, err := tx.Exec(ctx, ` INSERT INTO item_names (name, usage_count, last_used_at) VALUES ($1, 1, now()) ON CONFLICT (name) DO UPDATE SET usage_count = item_names.usage_count + 1, last_used_at = now()`, name, ) return err } // isKnownOpType validates the op_type string. func isKnownOpType(t string) bool { switch t { case OpListCreate, OpListRename, OpListDelete, OpItemAdd, OpItemUpdate, OpItemRemove: return true } return false }