Backend Phase C: Sync-Kern + Caddy behind-proxy

Sync-Kern:
- internal/sync/hlc.go: Hybrid Logical Clock (wall_ms<<16|counter)
  Tick/Now/After, global mutex, strikt monoton + kausal korrekt
- internal/sync/hlc_test.go: Unit-Tests (monoton, kausal, keine Duplikate)

Store-Schicht:
- internal/store/opstore.go: AppendOps idempotent via UNIQUE(client_id,
  client_seq) ON CONFLICT DO NOTHING; LWW-Projektion (list_create/
  rename/delete, item_add/update/remove) in derselben Transaktion;
  PullOps mit Cursor (seq > since, 500er Pages)
- internal/store/liststore.go: CreateList / GetLists / GetList
- internal/store/itemstore.go: GetItems (nicht-gelöschte Items)
- internal/store/suggeststore.go: Search (pg_trgm + LIKE-fallback, 10)

HTTP-Handler:
- internal/httpapi/lists.go: GET/POST /api/lists, GET /api/lists/{id}
- internal/httpapi/ops.go: POST /api/lists/{id}/ops (Push),
  GET /api/lists/{id}/ops (Pull ?since=)
- internal/httpapi/suggest.go: GET /api/suggestions?q=
- internal/httpapi/api.go: alle Routen verdrahtet (RequireAuth)

Deployment:
- deploy/Caddyfile.behind-proxy: auto_https off, trusted_proxies
- deploy/Caddyfile: X-Forwarded-Proto hinzugefügt, Kommentar aktualisiert
- deploy/docker-compose.yml: CADDY_HTTP_PORT + CADDY_HTTPS_PORT
- deploy/.env.example: Caddy-Port-Variablen dokumentiert

go build ./... && go vet ./... && go test ./... 
HLC-Tests: monoton, kausal, keine Duplikate 
AGENTS.md: Phase C vollständig 
This commit is contained in:
Tronax 2026-08-05 19:56:05 +02:00
parent a5ef8cf3ba
commit 895725b5e5
Signed by: Tronax
SSH key fingerprint: SHA256:2pKKXDZucWvaF/GzXNz0FY53EAO1YDLN80bqS+TTz/o
15 changed files with 1257 additions and 51 deletions

View file

@ -7,36 +7,48 @@ import (
"github.com/mitbringsl/backend/internal/auth"
"github.com/mitbringsl/backend/internal/config"
"github.com/mitbringsl/backend/internal/store"
)
// API bundles all handler groups and wires the router.
// Handler groups are added in subsequent phases (lists, items, suggestions).
type API struct {
cfg *config.Config
pool *pgxpool.Pool
health *HealthHandler
auth *AuthHandler
health *HealthHandler
auth *AuthHandler
lists *ListHandler
ops *OpsHandler
suggest *SuggestHandler
}
// NewAPI constructs the API with all handler groups.
func NewAPI(cfg *config.Config, pool *pgxpool.Pool) *API {
// Auth stores
users := auth.NewUserStore(pool)
sessions := auth.NewSessionStore(pool, auth.SessionConfig{
TokenBytes: cfg.SessionTokenBytes,
TTL: cfg.SessionTokenTTL,
})
oidcSvc := auth.NewOIDCService(buildProviders(cfg))
// When no provider is enabled, pass nil so the endpoint returns a clear
// "OIDC disabled" message instead of "unknown provider" for every request.
if len(oidcSvc.Providers()) == 0 {
oidcSvc = nil
}
// Sync stores
listStore := store.NewListStore(pool)
itemStore := store.NewItemStore(pool)
opStore := store.NewOpStore(pool)
suggestStore := store.NewSuggestStore(pool)
return &API{
cfg: cfg,
pool: pool,
health: &HealthHandler{Pool: pool},
auth: NewAuthHandler(users, sessions, oidcSvc, cfg),
cfg: cfg,
pool: pool,
health: &HealthHandler{Pool: pool},
auth: NewAuthHandler(users, sessions, oidcSvc, cfg),
lists: NewListHandler(listStore, itemStore),
ops: NewOpsHandler(opStore, listStore),
suggest: NewSuggestHandler(suggestStore),
}
}
@ -67,11 +79,13 @@ func (a *API) Handler() http.Handler {
mux.HandleFunc("POST /auth/oidc", a.auth.OIDC)
mux.HandleFunc("POST /auth/logout", a.auth.Logout)
// --- authenticated API endpoints (added in Phase C) ---
// mux.HandleFunc("GET /api/lists", a.RequireAuth(http.HandlerFunc(a.lists.List)))
// mux.HandleFunc("GET /api/lists/{id}/ops", a.RequireAuth(http.HandlerFunc(a.items.PullOps)))
// mux.HandleFunc("POST /api/lists/{id}/ops", a.RequireAuth(http.HandlerFunc(a.items.PushOps)))
// mux.HandleFunc("GET /api/suggestions", a.RequireAuth(http.HandlerFunc(a.suggest.Suggest)))
// --- authenticated API endpoints ---
mux.Handle("GET /api/lists", a.RequireAuth(http.HandlerFunc(a.lists.List)))
mux.Handle("POST /api/lists", a.RequireAuth(http.HandlerFunc(a.lists.Create)))
mux.Handle("GET /api/lists/{id}", a.RequireAuth(http.HandlerFunc(a.lists.Get)))
mux.Handle("POST /api/lists/{id}/ops", a.RequireAuth(http.HandlerFunc(a.ops.Push)))
mux.Handle("GET /api/lists/{id}/ops", a.RequireAuth(http.HandlerFunc(a.ops.Pull)))
mux.Handle("GET /api/suggestions", a.RequireAuth(http.HandlerFunc(a.suggest.Search)))
return Chain(
mux,

View file

@ -0,0 +1,162 @@
package httpapi
import (
"log/slog"
"net/http"
"strings"
"github.com/google/uuid"
"github.com/mitbringsl/backend/internal/store"
)
// ListHandler handles the /api/lists endpoints.
type ListHandler struct {
lists *store.ListStore
items *store.ItemStore
}
// NewListHandler creates a ListHandler with the given stores.
func NewListHandler(lists *store.ListStore, items *store.ItemStore) *ListHandler {
return &ListHandler{lists: lists, items: items}
}
// --- request / response types -----------------------------------------------
type createListRequest struct {
Name string `json:"name"`
}
type listDTO struct {
ID string `json:"id"`
Name string `json:"name"`
UpdatedAt string `json:"updated_at"`
HLCTS int64 `json:"hlc_ts"`
}
type listDetailDTO struct {
listDTO
Items []store.Item `json:"items"`
}
type listListResponse struct {
Lists []listDTO `json:"lists"`
}
// --- handlers ---------------------------------------------------------------
// List returns all non-deleted lists for the authenticated user.
// GET /api/lists
func (h *ListHandler) List(w http.ResponseWriter, r *http.Request) {
userID, ok := userIDFromCtx(r)
if !ok {
renderError(w, http.StatusUnauthorized, "Unauthorized", "Missing user context.")
return
}
ls, err := h.lists.GetLists(r.Context(), userID)
if err != nil {
slog.Error("list lists failed", "error", err, "user_id", userID)
renderError(w, http.StatusInternalServerError, "Internal error", "Could not load lists.")
return
}
dtos := make([]listDTO, len(ls))
for i, l := range ls {
dtos[i] = listDTO{
ID: l.ID.String(),
Name: l.Name,
UpdatedAt: l.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"),
HLCTS: l.HLCTS,
}
}
renderJSON(w, http.StatusOK, listListResponse{Lists: dtos})
}
// Create creates a new list for the authenticated user.
// POST /api/lists
func (h *ListHandler) Create(w http.ResponseWriter, r *http.Request) {
userID, ok := userIDFromCtx(r)
if !ok {
renderError(w, http.StatusUnauthorized, "Unauthorized", "Missing user context.")
return
}
var req createListRequest
if !decodeJSON(w, r, &req) {
return
}
name := strings.TrimSpace(req.Name)
if name == "" {
renderError(w, http.StatusBadRequest, "Bad request", "List name must not be empty.")
return
}
l, err := h.lists.CreateList(r.Context(), userID, name)
if err != nil {
slog.Error("create list failed", "error", err, "user_id", userID)
renderError(w, http.StatusInternalServerError, "Internal error", "Could not create list.")
return
}
renderJSON(w, http.StatusCreated, listDTO{
ID: l.ID.String(),
Name: l.Name,
UpdatedAt: l.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"),
HLCTS: l.HLCTS,
})
}
// Get returns a single list with its items.
// GET /api/lists/{id}
func (h *ListHandler) Get(w http.ResponseWriter, r *http.Request) {
userID, ok := userIDFromCtx(r)
if !ok {
renderError(w, http.StatusUnauthorized, "Unauthorized", "Missing user context.")
return
}
listID, err := uuid.Parse(r.PathValue("id"))
if err != nil {
renderError(w, http.StatusBadRequest, "Bad request", "Invalid list ID.")
return
}
l, err := h.lists.GetList(r.Context(), listID, userID)
if err != nil {
if !apiError(w, err) {
slog.Error("get list failed", "error", err, "list_id", listID)
renderError(w, http.StatusInternalServerError, "Internal error", "Could not load list.")
}
return
}
its, err := h.items.GetItems(r.Context(), listID)
if err != nil {
slog.Error("get items failed", "error", err, "list_id", listID)
renderError(w, http.StatusInternalServerError, "Internal error", "Could not load items.")
return
}
if its == nil {
its = []store.Item{}
}
renderJSON(w, http.StatusOK, listDetailDTO{
listDTO: listDTO{
ID: l.ID.String(),
Name: l.Name,
UpdatedAt: l.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"),
HLCTS: l.HLCTS,
},
Items: its,
})
}
// userIDFromCtx extracts the user UUID from the request context (set by RequireAuth).
func userIDFromCtx(r *http.Request) (uuid.UUID, bool) {
v := r.Context().Value(ctxKeyUserID)
if v == nil {
return uuid.Nil, false
}
id, ok := v.(uuid.UUID)
return id, ok
}

View file

@ -0,0 +1,147 @@
package httpapi
import (
"fmt"
"log/slog"
"net/http"
"strconv"
"github.com/google/uuid"
"github.com/mitbringsl/backend/internal/store"
)
// OpsHandler handles push and pull of op_log entries.
type OpsHandler struct {
ops *store.OpStore
lists *store.ListStore
}
// NewOpsHandler creates an OpsHandler with the given stores.
func NewOpsHandler(ops *store.OpStore, lists *store.ListStore) *OpsHandler {
return &OpsHandler{ops: ops, lists: lists}
}
// --- request / response types -----------------------------------------------
type pushRequest struct {
ClientID uuid.UUID `json:"client_id"`
Ops []store.IncomingOp `json:"ops"`
}
type pushResponse struct {
Results []store.OpResult `json:"results"`
}
type pullResponse struct {
Ops []store.Op `json:"ops"`
HasMore bool `json:"has_more"`
}
// --- handlers ---------------------------------------------------------------
// Push accepts a batch of ops from the client, inserts them idempotently into
// op_log, applies the LWW projection, and returns the server-assigned seq+hlc
// for each op.
//
// POST /api/lists/{id}/ops
func (h *OpsHandler) Push(w http.ResponseWriter, r *http.Request) {
userID, ok := userIDFromCtx(r)
if !ok {
renderError(w, http.StatusUnauthorized, "Unauthorized", "Missing user context.")
return
}
listID, err := uuid.Parse(r.PathValue("id"))
if err != nil {
renderError(w, http.StatusBadRequest, "Bad request", "Invalid list ID.")
return
}
// Verify the user owns (or is a member of) the list.
if _, err := h.lists.GetList(r.Context(), listID, userID); err != nil {
if !apiError(w, err) {
renderError(w, http.StatusNotFound, "Not found", "List not found.")
}
return
}
var req pushRequest
if !decodeJSON(w, r, &req) {
return
}
if req.ClientID == uuid.Nil {
renderError(w, http.StatusBadRequest, "Bad request", "client_id must be a non-nil UUID.")
return
}
if len(req.Ops) == 0 {
renderJSON(w, http.StatusOK, pushResponse{Results: []store.OpResult{}})
return
}
maxBatch := store.MaxPushBatch()
if len(req.Ops) > maxBatch {
renderError(w, http.StatusBadRequest, "Bad request",
fmt.Sprintf("Too many ops in a single request (max %d).", maxBatch))
return
}
results, err := h.ops.AppendOps(r.Context(), listID, userID, req.ClientID, req.Ops)
if err != nil {
slog.Error("push ops failed", "error", err, "list_id", listID, "user_id", userID)
renderError(w, http.StatusInternalServerError, "Internal error", "Could not store ops.")
return
}
renderJSON(w, http.StatusOK, pushResponse{Results: results})
}
// Pull returns ops for a list since a given server seq cursor.
//
// GET /api/lists/{id}/ops?since=0
func (h *OpsHandler) Pull(w http.ResponseWriter, r *http.Request) {
userID, ok := userIDFromCtx(r)
if !ok {
renderError(w, http.StatusUnauthorized, "Unauthorized", "Missing user context.")
return
}
listID, err := uuid.Parse(r.PathValue("id"))
if err != nil {
renderError(w, http.StatusBadRequest, "Bad request", "Invalid list ID.")
return
}
// Verify list access.
if _, err := h.lists.GetList(r.Context(), listID, userID); err != nil {
if !apiError(w, err) {
renderError(w, http.StatusNotFound, "Not found", "List not found.")
}
return
}
since := int64(0)
if s := r.URL.Query().Get("since"); s != "" {
v, err := strconv.ParseInt(s, 10, 64)
if err != nil || v < 0 {
renderError(w, http.StatusBadRequest, "Bad request", "since must be a non-negative integer.")
return
}
since = v
}
ops, err := h.ops.PullOps(r.Context(), listID, since)
if err != nil {
slog.Error("pull ops failed", "error", err, "list_id", listID, "since", since)
renderError(w, http.StatusInternalServerError, "Internal error", "Could not load ops.")
return
}
if ops == nil {
ops = []store.Op{}
}
// has_more is true when we returned exactly the page-size limit.
// The client should pull again with the last returned seq.
hasMore := len(ops) == 500 // pullPageSize is 500 in opstore.go
renderJSON(w, http.StatusOK, pullResponse{Ops: ops, HasMore: hasMore})
}

View file

@ -0,0 +1,47 @@
package httpapi
import (
"log/slog"
"net/http"
"strings"
"github.com/mitbringsl/backend/internal/store"
)
// SuggestHandler handles the /api/suggestions endpoint.
type SuggestHandler struct {
suggest *store.SuggestStore
}
// NewSuggestHandler creates a SuggestHandler backed by the given store.
func NewSuggestHandler(suggest *store.SuggestStore) *SuggestHandler {
return &SuggestHandler{suggest: suggest}
}
type suggestResponse struct {
Suggestions []string `json:"suggestions"`
}
// Search returns autocomplete suggestions for item names matching the query.
//
// GET /api/suggestions?q=<query>
func (h *SuggestHandler) Search(w http.ResponseWriter, r *http.Request) {
q := strings.TrimSpace(r.URL.Query().Get("q"))
if q == "" {
renderError(w, http.StatusBadRequest, "Bad request", "Query parameter 'q' is required.")
return
}
// Lowercase so the trgm index can match correctly.
q = strings.ToLower(q)
names, err := h.suggest.Search(r.Context(), q)
if err != nil {
slog.Error("suggest search failed", "error", err, "q", q)
renderError(w, http.StatusInternalServerError, "Internal error", "Could not load suggestions.")
return
}
if names == nil {
names = []string{}
}
renderJSON(w, http.StatusOK, suggestResponse{Suggestions: names})
}

View file

@ -0,0 +1,64 @@
package store
import (
"context"
"fmt"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
)
// Item is the projection of an items row.
type Item struct {
ID uuid.UUID `json:"id"`
ListID uuid.UUID `json:"list_id"`
Name string `json:"name"`
Quantity *string `json:"quantity,omitempty"`
Checked bool `json:"checked"`
SortOrder *int `json:"sort_order,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
DeletedAt *time.Time `json:"deleted_at,omitempty"`
HLCTS int64 `json:"hlc_ts"`
ClientID *uuid.UUID `json:"client_id,omitempty"`
CheckedAt *time.Time `json:"checked_at,omitempty"`
}
// ItemStore provides read access to the items projection.
type ItemStore struct {
pool *pgxpool.Pool
}
// NewItemStore creates an ItemStore backed by the given pool.
func NewItemStore(pool *pgxpool.Pool) *ItemStore { return &ItemStore{pool: pool} }
// GetItems returns all non-deleted items for listID, ordered by sort_order then creation time.
func (s *ItemStore) GetItems(ctx context.Context, listID uuid.UUID) ([]Item, error) {
rows, err := s.pool.Query(ctx, `
SELECT id, list_id, name, quantity, checked, sort_order,
created_at, updated_at, hlc_ts, client_id, checked_at
FROM items
WHERE list_id = $1 AND deleted_at IS NULL
ORDER BY sort_order ASC NULLS LAST, created_at ASC`,
listID,
)
if err != nil {
return nil, fmt.Errorf("itemstore: get items: %w", err)
}
defer rows.Close()
var items []Item
for rows.Next() {
var it Item
if err := rows.Scan(
&it.ID, &it.ListID, &it.Name, &it.Quantity, &it.Checked,
&it.SortOrder, &it.CreatedAt, &it.UpdatedAt,
&it.HLCTS, &it.ClientID, &it.CheckedAt,
); err != nil {
return nil, fmt.Errorf("itemstore: scan item row: %w", err)
}
items = append(items, it)
}
return items, rows.Err()
}

View file

@ -0,0 +1,88 @@
package store
import (
"context"
"fmt"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
)
// List is the projection of a list row.
type List struct {
ID uuid.UUID `json:"id"`
Name string `json:"name"`
OwnerID uuid.UUID `json:"owner_id"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
DeletedAt *time.Time `json:"deleted_at,omitempty"`
HLCTS int64 `json:"hlc_ts"`
}
// ListStore provides read/write access to the lists projection.
type ListStore struct {
pool *pgxpool.Pool
}
// NewListStore creates a ListStore backed by the given pool.
func NewListStore(pool *pgxpool.Pool) *ListStore { return &ListStore{pool: pool} }
// CreateList inserts a new list owned by ownerID.
// In the MVP, list creation is a direct INSERT (no op required).
// The caller is responsible for emitting a list_create op if desired for full
// op-log coverage; for Phase C the REST endpoint does a direct insert.
func (s *ListStore) CreateList(ctx context.Context, ownerID uuid.UUID, name string) (*List, error) {
var l List
err := s.pool.QueryRow(ctx, `
INSERT INTO lists (name, owner_id)
VALUES ($1, $2)
RETURNING id, name, owner_id, created_at, updated_at, hlc_ts`,
name, ownerID,
).Scan(&l.ID, &l.Name, &l.OwnerID, &l.CreatedAt, &l.UpdatedAt, &l.HLCTS)
if err != nil {
return nil, fmt.Errorf("liststore: create list: %w", err)
}
return &l, nil
}
// GetLists returns all non-deleted lists owned by ownerID, newest first.
func (s *ListStore) GetLists(ctx context.Context, ownerID uuid.UUID) ([]List, error) {
rows, err := s.pool.Query(ctx, `
SELECT id, name, owner_id, created_at, updated_at, hlc_ts
FROM lists
WHERE owner_id = $1 AND deleted_at IS NULL
ORDER BY created_at DESC`,
ownerID,
)
if err != nil {
return nil, fmt.Errorf("liststore: get lists: %w", err)
}
defer rows.Close()
var lists []List
for rows.Next() {
var l List
if err := rows.Scan(&l.ID, &l.Name, &l.OwnerID, &l.CreatedAt, &l.UpdatedAt, &l.HLCTS); err != nil {
return nil, fmt.Errorf("liststore: scan list row: %w", err)
}
lists = append(lists, l)
}
return lists, rows.Err()
}
// GetList returns a single non-deleted list, checking that userID is the owner.
// Returns pgx.ErrNoRows if not found or not owned by userID.
func (s *ListStore) GetList(ctx context.Context, listID, userID uuid.UUID) (*List, error) {
var l List
err := s.pool.QueryRow(ctx, `
SELECT id, name, owner_id, created_at, updated_at, hlc_ts
FROM lists
WHERE id = $1 AND owner_id = $2 AND deleted_at IS NULL`,
listID, userID,
).Scan(&l.ID, &l.Name, &l.OwnerID, &l.CreatedAt, &l.UpdatedAt, &l.HLCTS)
if err != nil {
return nil, fmt.Errorf("liststore: get list: %w", err)
}
return &l, nil
}

View file

@ -0,0 +1,368 @@
// 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
}

View file

@ -0,0 +1,46 @@
package store
import (
"context"
"fmt"
"github.com/jackc/pgx/v5/pgxpool"
)
const suggestLimit = 10
// SuggestStore provides autocomplete search over aggregated item names.
type SuggestStore struct {
pool *pgxpool.Pool
}
// NewSuggestStore creates a SuggestStore backed by the given pool.
func NewSuggestStore(pool *pgxpool.Pool) *SuggestStore { return &SuggestStore{pool: pool} }
// Search returns up to suggestLimit item name suggestions matching q using
// pg_trgm similarity search. Results are ordered by usage_count DESC so the
// most-used names appear first.
func (s *SuggestStore) Search(ctx context.Context, q string) ([]string, error) {
rows, err := s.pool.Query(ctx, `
SELECT name
FROM item_names
WHERE name % $1 OR name LIKE $2
ORDER BY usage_count DESC, last_used_at DESC
LIMIT $3`,
q, q+"%", suggestLimit,
)
if err != nil {
return nil, fmt.Errorf("suggeststore: search: %w", err)
}
defer rows.Close()
var names []string
for rows.Next() {
var name string
if err := rows.Scan(&name); err != nil {
return nil, fmt.Errorf("suggeststore: scan name: %w", err)
}
names = append(names, name)
}
return names, rows.Err()
}

View file

@ -0,0 +1,120 @@
// Package sync provides synchronisation primitives for the mitbringsl backend.
// The Hybrid Logical Clock (HLC) combines a physical wall-clock with a logical
// counter so that every event gets a strictly monotonically increasing int64
// timestamp that is compatible with client-generated HLC values.
//
// Encoding: the top 48 bits hold wall-clock milliseconds since the Unix epoch;
// the bottom 16 bits hold a counter that disambiguates events within the same
// millisecond. This lets us store the full HLC value in a single BIGINT column
// while still being orderable and human-readable.
package sync
import (
"sync"
"time"
)
const (
// counterBits is the number of bits reserved for the intra-ms counter.
counterBits = 16
// maxCounter is the maximum value the intra-ms counter can hold.
maxCounter = (1 << counterBits) - 1
)
// clock is the process-global HLC state. All exported functions are safe for
// concurrent use via the embedded mutex.
var clock struct {
mu sync.Mutex
last int64
}
// Now returns the current HLC value without advancing the state.
// Useful for initialisation checks; prefer Tick for actual event timestamps.
func Now() int64 {
clock.mu.Lock()
defer clock.mu.Unlock()
return tick(clock.last)
}
// Tick advances the global HLC past the given external value (e.g. a
// client-supplied hlc_ts) and returns the new value. It is safe to call with
// 0 when there is no external value to incorporate.
//
// Guarantees:
// - result > last (strictly monotonic)
// - result >= externalHLC (causal ordering preserved)
func Tick(externalHLC int64) int64 {
clock.mu.Lock()
defer clock.mu.Unlock()
wall := wallMS()
prev := clock.last
// Determine the maximum physical time we're aware of.
maxWall := max3(wall, wallOf(prev), wallOf(externalHLC))
var counter int64
switch {
case maxWall == wallOf(prev) && maxWall == wallOf(externalHLC):
// Both are in the same ms bucket: take the larger counter + 1.
c := counterOf(prev)
if ec := counterOf(externalHLC); ec > c {
c = ec
}
counter = c + 1
case maxWall == wallOf(prev):
counter = counterOf(prev) + 1
case maxWall == wallOf(externalHLC):
counter = counterOf(externalHLC) + 1
default:
// Wall time advanced; reset counter.
counter = 0
}
if counter > maxCounter {
// Counter overflow: steal a millisecond from the future.
maxWall++
counter = 0
}
next := (maxWall << counterBits) | counter
clock.last = next
return next
}
// After reports whether a causally follows b (i.e. a > b as integers).
func After(a, b int64) bool { return a > b }
// wallMS returns the current Unix epoch time in milliseconds.
func wallMS() int64 { return time.Now().UnixMilli() }
// wallOf extracts the wall-clock ms from an HLC value.
func wallOf(hlc int64) int64 { return hlc >> counterBits }
// counterOf extracts the logical counter from an HLC value.
func counterOf(hlc int64) int64 { return hlc & maxCounter }
// tick is the internal (lock-held) Tick implementation against wall time only.
func tick(last int64) int64 {
wall := wallMS()
maxWall := wallOf(last)
if wall > maxWall {
return wall << counterBits
}
counter := counterOf(last) + 1
if counter > maxCounter {
maxWall++
counter = 0
}
return (maxWall << counterBits) | counter
}
func max3(a, b, c int64) int64 {
if b > a {
a = b
}
if c > a {
a = c
}
return a
}

View file

@ -0,0 +1,61 @@
package sync
import (
"testing"
)
func TestTick_StrictlyMonotonic(t *testing.T) {
// Reset global state.
clock.mu.Lock()
clock.last = 0
clock.mu.Unlock()
prev := int64(0)
for i := 0; i < 1000; i++ {
v := Tick(0)
if v <= prev {
t.Fatalf("iteration %d: Tick() = %d, want > %d", i, v, prev)
}
prev = v
}
}
func TestTick_CausalOrdering(t *testing.T) {
clock.mu.Lock()
clock.last = 0
clock.mu.Unlock()
// Simulate a client HLC far in the future.
future := int64(9_999_999_999) << counterBits
v := Tick(future)
if !After(v, future) {
t.Fatalf("Tick(future) = %d, want > %d", v, future)
}
}
func TestTick_NoDuplicates(t *testing.T) {
clock.mu.Lock()
clock.last = 0
clock.mu.Unlock()
seen := make(map[int64]bool)
for i := 0; i < 10_000; i++ {
v := Tick(0)
if seen[v] {
t.Fatalf("duplicate HLC value %d at iteration %d", v, i)
}
seen[v] = true
}
}
func TestAfter(t *testing.T) {
if !After(2, 1) {
t.Fatal("After(2,1) must be true")
}
if After(1, 2) {
t.Fatal("After(1,2) must be false")
}
if After(1, 1) {
t.Fatal("After(1,1) must be false")
}
}