Memby v0.1.53: Android TV client plus gateway

Android TV client for Emby (Kotlin, Compose for TV) and the Memby gateway (Go, Postgres, Redis) that fronts it. Client: - Setup, profiles, home rows, Media3 playback, system screensaver (Dream) - Backend chosen at build time: gateway when memby.gatewayUrl is set, otherwise direct to Emby. Both paths stay working. - Server-composed home rows, rendered verbatim so new row types ship without an app release - Full-screen animated maintenance state, row engagement telemetry Gateway: - One request per TV screen; auth, caching, search and row shaping - Library import from Emby into Postgres (manual, then hourly incremental) - Recommendations from viewing history (recency-weighted genre affinity) - Admin page for imports, an offline switch, and per-row analytics - Video always direct-plays from Emby; only metadata passes through Identity is com.ponzischeme89.memby throughout, replacing com.mattcohen.embyclientsname. A changed applicationId installs as a new app: TVs need a fresh sign-in and the old package uninstalled. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-27 08:16:20 +12:00
commit 2ce405c540
99 changed files with 14433 additions and 0 deletions
@@ -0,0 +1,113 @@
+package store
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// RowEvent is one reported interaction with a home-screen row.
+type RowEvent struct {
+	OccurredAt time.Time
+	UserID     string
+	RowID      string
+	RowKind    string
+	Event      string
+	ItemID     string
+	DwellMs    int
+}
+
+// Event kinds. Impressions say a row was drawn; focus says the remote actually landed
+// on it and for how long; select says something was opened from it.
+const (
+	RowEventImpression = "impression"
+	RowEventFocus      = "focus"
+	RowEventSelect     = "select"
+)
+
+// RowStat is the aggregate the admin page renders.
+type RowStat struct {
+	RowID       string  `json:"rowId"`
+	RowKind     string  `json:"rowKind"`
+	Impressions int64   `json:"impressions"`
+	Focuses     int64   `json:"focuses"`
+	Selects     int64   `json:"selects"`
+	DwellMs     int64   `json:"dwellMs"`
+	Viewers     int64   `json:"viewers"`
+	SelectRate  float64 `json:"selectRate"`
+}
+
+func (s *Store) InsertRowEvents(ctx context.Context, events []RowEvent) error {
+	if len(events) == 0 {
+		return nil
+	}
+	batch := &pgx.Batch{}
+	for _, event := range events {
+		batch.Queue(`
+			INSERT INTO row_events (occurred_at, emby_user_id, row_id, row_kind, event, item_id, dwell_ms)
+			VALUES ($1,$2,$3,$4,$5,$6,$7)`,
+			event.OccurredAt, event.UserID, event.RowID, event.RowKind,
+			event.Event, event.ItemID, event.DwellMs)
+	}
+
+	results := s.pool.SendBatch(ctx, batch)
+	defer results.Close()
+	for range events {
+		if _, err := results.Exec(); err != nil {
+			return fmt.Errorf("store: insert row events: %w", err)
+		}
+	}
+	return nil
+}
+
+// RowStats aggregates engagement since a point in time, busiest row first.
+//
+// Dwell is the interesting number: impressions only say a row was on screen, whereas
+// dwell says someone actually stopped there.
+func (s *Store) RowStats(ctx context.Context, since time.Time) ([]RowStat, error) {
+	rows, err := s.pool.Query(ctx, `
+		SELECT row_id,
+		       (array_agg(row_kind ORDER BY occurred_at DESC))[1] AS row_kind,
+		       count(*) FILTER (WHERE event = 'impression') AS impressions,
+		       count(*) FILTER (WHERE event = 'focus')      AS focuses,
+		       count(*) FILTER (WHERE event = 'select')     AS selects,
+		       coalesce(sum(dwell_ms), 0)                   AS dwell_ms,
+		       count(DISTINCT emby_user_id)                 AS viewers
+		FROM row_events
+		WHERE occurred_at >= $1
+		GROUP BY row_id
+		ORDER BY dwell_ms DESC, impressions DESC`, since)
+	if err != nil {
+		return nil, fmt.Errorf("store: row stats: %w", err)
+	}
+	defer rows.Close()
+
+	stats := []RowStat{}
+	for rows.Next() {
+		var stat RowStat
+		if err := rows.Scan(&stat.RowID, &stat.RowKind, &stat.Impressions, &stat.Focuses,
+			&stat.Selects, &stat.DwellMs, &stat.Viewers); err != nil {
+			return nil, err
+		}
+		if stat.Impressions > 0 {
+			stat.SelectRate = float64(stat.Selects) / float64(stat.Impressions)
+		}
+		stats = append(stats, stat)
+	}
+	return stats, rows.Err()
+}
+
+// PruneRowEvents drops raw events past their retention window. Aggregates are computed
+// at read time, so nothing is preserved once the events go — which is the point: this is
+// engagement telemetry for tuning rows, not a permanent record of what people watched.
+func (s *Store) PruneRowEvents(ctx context.Context, olderThan time.Duration) (int64, error) {
+	tag, err := s.pool.Exec(ctx,
+		`DELETE FROM row_events WHERE occurred_at < now() - $1::interval`,
+		fmt.Sprintf("%d seconds", int64(olderThan.Seconds())))
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
@@ -0,0 +1,181 @@
+package store
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// LibraryItem is one imported Emby item. Payload is Emby's JSON verbatim; the flat
+// columns exist only so Postgres can filter and rank without opening the JSON.
+type LibraryItem struct {
+	ID              string
+	Type            string
+	Name            string
+	SeriesID        string
+	SeriesName      string
+	ProductionYear  *int
+	CommunityRating *float64
+	Genres          []string
+	Studios         []string
+	DateCreated     *time.Time
+	SearchText      string
+	Payload         json.RawMessage
+}
+
+// LibraryStats is what the admin page shows about the imported library.
+type LibraryStats struct {
+	Total      int64            `json:"total"`
+	ByType     map[string]int64 `json:"byType"`
+	LastSynced *time.Time       `json:"lastSynced"`
+}
+
+// UpsertLibraryItems writes a batch, refreshing synced_at on every row it touches.
+//
+// synced_at doubles as the mark-and-sweep marker: a full import stamps everything it
+// sees, then deletes whatever kept an older stamp.
+func (s *Store) UpsertLibraryItems(ctx context.Context, items []LibraryItem, syncedAt time.Time) (int64, error) {
+	if len(items) == 0 {
+		return 0, nil
+	}
+
+	batch := &pgx.Batch{}
+	for _, item := range items {
+		batch.Queue(`
+			INSERT INTO library_items (
+				id, type, name, series_id, series_name, production_year, community_rating,
+				genres, studios, date_created, search_text, payload, synced_at
+			) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12::jsonb,$13)
+			ON CONFLICT (id) DO UPDATE SET
+				type = EXCLUDED.type,
+				name = EXCLUDED.name,
+				series_id = EXCLUDED.series_id,
+				series_name = EXCLUDED.series_name,
+				production_year = EXCLUDED.production_year,
+				community_rating = EXCLUDED.community_rating,
+				genres = EXCLUDED.genres,
+				studios = EXCLUDED.studios,
+				date_created = EXCLUDED.date_created,
+				search_text = EXCLUDED.search_text,
+				payload = EXCLUDED.payload,
+				synced_at = EXCLUDED.synced_at`,
+			item.ID, item.Type, item.Name, item.SeriesID, item.SeriesName,
+			item.ProductionYear, item.CommunityRating, item.Genres, item.Studios,
+			item.DateCreated, item.SearchText, string(item.Payload), syncedAt)
+	}
+
+	results := s.pool.SendBatch(ctx, batch)
+	defer results.Close()
+
+	var written int64
+	for range items {
+		tag, err := results.Exec()
+		if err != nil {
+			return written, fmt.Errorf("store: upsert library items: %w", err)
+		}
+		written += tag.RowsAffected()
+	}
+	return written, nil
+}
+
+// DeleteLibraryItemsBefore removes anything a full import did not touch — items deleted
+// from Emby since the last run.
+func (s *Store) DeleteLibraryItemsBefore(ctx context.Context, cutoff time.Time) (int64, error) {
+	tag, err := s.pool.Exec(ctx, `DELETE FROM library_items WHERE synced_at < $1`, cutoff)
+	if err != nil {
+		return 0, fmt.Errorf("store: prune library: %w", err)
+	}
+	return tag.RowsAffected(), nil
+}
+
+// SearchLibrary answers from the imported library rather than Emby.
+//
+// Full-text match first, with a trailing ILIKE so partial words ("sever") still hit
+// before someone finishes typing on a remote.
+func (s *Store) SearchLibrary(ctx context.Context, term string, limit int) ([]json.RawMessage, error) {
+	trimmed := strings.TrimSpace(term)
+	if trimmed == "" {
+		return nil, nil
+	}
+	rows, err := s.pool.Query(ctx, `
+		SELECT payload
+		FROM library_items
+		WHERE search_tsv @@ plainto_tsquery('simple', $1)
+		   OR search_text ILIKE '%' || $1 || '%'
+		ORDER BY
+			ts_rank(search_tsv, plainto_tsquery('simple', $1)) DESC,
+			(lower(name) = lower($1)) DESC,
+			community_rating DESC NULLS LAST,
+			name ASC
+		LIMIT $2`, trimmed, limit)
+	if err != nil {
+		return nil, fmt.Errorf("store: search library: %w", err)
+	}
+	return collectPayloads(rows)
+}
+
+// LibraryCandidates returns unwatched-agnostic candidates in the given genres, for the
+// recommendation engine. User state is applied by the caller, which is the only place
+// that knows it.
+func (s *Store) LibraryCandidates(ctx context.Context, genres []string, limit int) ([]json.RawMessage, error) {
+	if len(genres) == 0 {
+		return nil, nil
+	}
+	rows, err := s.pool.Query(ctx, `
+		SELECT payload
+		FROM library_items
+		WHERE type IN ('Movie', 'Series')
+		  AND genres && $1
+		ORDER BY community_rating DESC NULLS LAST, date_created DESC NULLS LAST
+		LIMIT $2`, genres, limit)
+	if err != nil {
+		return nil, fmt.Errorf("store: library candidates: %w", err)
+	}
+	return collectPayloads(rows)
+}
+
+func (s *Store) LibraryStats(ctx context.Context) (LibraryStats, error) {
+	stats := LibraryStats{ByType: map[string]int64{}}
+
+	rows, err := s.pool.Query(ctx, `SELECT type, count(*) FROM library_items GROUP BY type`)
+	if err != nil {
+		return stats, fmt.Errorf("store: library stats: %w", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var itemType string
+		var count int64
+		if err := rows.Scan(&itemType, &count); err != nil {
+			return stats, err
+		}
+		stats.ByType[itemType] = count
+		stats.Total += count
+	}
+	if err := rows.Err(); err != nil {
+		return stats, err
+	}
+
+	var lastSynced *time.Time
+	if err := s.pool.QueryRow(ctx, `SELECT max(synced_at) FROM library_items`).Scan(&lastSynced); err != nil {
+		return stats, err
+	}
+	stats.LastSynced = lastSynced
+	return stats, nil
+}
+
+func collectPayloads(rows pgx.Rows) ([]json.RawMessage, error) {
+	defer rows.Close()
+	out := []json.RawMessage{}
+	for rows.Next() {
+		var payload []byte
+		if err := rows.Scan(&payload); err != nil {
+			return nil, err
+		}
+		out = append(out, json.RawMessage(payload))
+	}
+	return out, rows.Err()
+}
@@ -0,0 +1,88 @@
+-- Gateway sessions: one row per signed-in TV.
+--
+-- token_hash is SHA-256 of the bearer token handed to the device, so a database dump
+-- does not hand over working gateway tokens. emby_token IS the live upstream token and
+-- is stored as-is: treat this volume as a secret store.
+CREATE TABLE IF NOT EXISTS sessions (
+    token_hash   BYTEA PRIMARY KEY,
+    emby_user_id TEXT        NOT NULL,
+    emby_token   TEXT        NOT NULL,
+    username     TEXT        NOT NULL,
+    server_id    TEXT        NOT NULL DEFAULT '',
+    device_id    TEXT        NOT NULL DEFAULT '',
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS sessions_emby_user_idx ON sessions (emby_user_id);
+CREATE INDEX IF NOT EXISTS sessions_last_seen_idx ON sessions (last_seen_at);
+
+-- The imported library.
+--
+-- payload is Emby's item JSON verbatim, so rows served from here are byte-identical to
+-- rows served live. Deliberately holds NO per-user state: everything is imported with
+-- EnableUserData=false, because one household shares this table and watched/favourite
+-- flags are not shareable. Anything user-specific still comes from Emby live.
+CREATE TABLE IF NOT EXISTS library_items (
+    id               TEXT PRIMARY KEY,
+    type             TEXT        NOT NULL DEFAULT '',
+    name             TEXT        NOT NULL DEFAULT '',
+    series_id        TEXT        NOT NULL DEFAULT '',
+    series_name      TEXT        NOT NULL DEFAULT '',
+    production_year  INT,
+    community_rating REAL,
+    genres           TEXT[]      NOT NULL DEFAULT '{}',
+    studios          TEXT[]      NOT NULL DEFAULT '{}',
+    date_created     TIMESTAMPTZ,
+    search_text      TEXT        NOT NULL DEFAULT '',
+    payload          JSONB       NOT NULL,
+    synced_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    -- 'simple' rather than 'english': film titles are proper nouns, and stemming
+    -- "Arrival" into "arriv" helps nobody.
+    search_tsv       tsvector GENERATED ALWAYS AS (to_tsvector('simple', search_text)) STORED
+);
+
+CREATE INDEX IF NOT EXISTS library_items_search_idx ON library_items USING GIN (search_tsv);
+CREATE INDEX IF NOT EXISTS library_items_genres_idx ON library_items USING GIN (genres);
+CREATE INDEX IF NOT EXISTS library_items_type_created_idx ON library_items (type, date_created DESC);
+CREATE INDEX IF NOT EXISTS library_items_synced_idx ON library_items (synced_at);
+
+-- One row per import, so the admin page can show what happened and when.
+CREATE TABLE IF NOT EXISTS sync_runs (
+    id             BIGSERIAL PRIMARY KEY,
+    kind           TEXT        NOT NULL,                    -- full | incremental
+    trigger        TEXT        NOT NULL DEFAULT 'schedule', -- schedule | manual | startup
+    status         TEXT        NOT NULL,                    -- running | success | failed
+    started_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+    finished_at    TIMESTAMPTZ,
+    items_seen     INT         NOT NULL DEFAULT 0,
+    items_upserted INT         NOT NULL DEFAULT 0,
+    items_removed  INT         NOT NULL DEFAULT 0,
+    error          TEXT        NOT NULL DEFAULT ''
+);
+
+CREATE INDEX IF NOT EXISTS sync_runs_started_idx ON sync_runs (started_at DESC);
+
+-- Small key/value store for operator switches (currently just maintenance mode). Kept in
+-- Postgres rather than memory so a restart cannot silently bring the app back up.
+CREATE TABLE IF NOT EXISTS app_settings (
+    key        TEXT PRIMARY KEY,
+    value      JSONB       NOT NULL,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Row-level engagement. One row per reported event; aggregation happens at read time,
+-- which is fine at household scale and keeps the write path trivial.
+CREATE TABLE IF NOT EXISTS row_events (
+    id           BIGSERIAL PRIMARY KEY,
+    occurred_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
+    emby_user_id TEXT        NOT NULL,
+    row_id       TEXT        NOT NULL,
+    row_kind     TEXT        NOT NULL DEFAULT '',
+    event        TEXT        NOT NULL,                -- impression | focus | select
+    item_id      TEXT        NOT NULL DEFAULT '',
+    dwell_ms     INT         NOT NULL DEFAULT 0
+);
+
+CREATE INDEX IF NOT EXISTS row_events_time_idx ON row_events (occurred_at DESC);
+CREATE INDEX IF NOT EXISTS row_events_row_idx ON row_events (row_id, occurred_at DESC);
@@ -0,0 +1,80 @@
+package store
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// MaintenanceKey is the app_settings row backing maintenance mode.
+const MaintenanceKey = "maintenance"
+
+// Maintenance is the operator switch that takes Memby down independently of Emby.
+//
+// Deliberately durable: a restart must not quietly bring the app back up while someone
+// is still working on it.
+type Maintenance struct {
+	Enabled   bool      `json:"enabled"`
+	Message   string    `json:"message"`
+	UpdatedAt time.Time `json:"updatedAt"`
+}
+
+// DefaultMaintenanceMessage is shown on the TV when the operator did not write one.
+const DefaultMaintenanceMessage = "Memby is down for maintenance. Try again shortly."
+
+func (s *Store) Maintenance(ctx context.Context) (Maintenance, error) {
+	var raw []byte
+	err := s.pool.QueryRow(ctx, `SELECT value FROM app_settings WHERE key = $1`, MaintenanceKey).Scan(&raw)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return Maintenance{}, nil
+	}
+	if err != nil {
+		return Maintenance{}, fmt.Errorf("store: read maintenance: %w", err)
+	}
+
+	var state Maintenance
+	if err := json.Unmarshal(raw, &state); err != nil {
+		return Maintenance{}, fmt.Errorf("store: decode maintenance: %w", err)
+	}
+	return state, nil
+}
+
+func (s *Store) SetMaintenance(ctx context.Context, state Maintenance) error {
+	state.UpdatedAt = time.Now().UTC()
+	raw, err := json.Marshal(state)
+	if err != nil {
+		return err
+	}
+	_, err = s.pool.Exec(ctx, `
+		INSERT INTO app_settings (key, value, updated_at)
+		VALUES ($1, $2::jsonb, now())
+		ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`,
+		MaintenanceKey, string(raw))
+	if err != nil {
+		return fmt.Errorf("store: write maintenance: %w", err)
+	}
+	return nil
+}
+
+// NewestSession is the fallback credential for the library import: whichever TV signed
+// in most recently. It means a fresh deployment can import without configuring a
+// service account, at the cost of the import stopping if that user is ever removed.
+func (s *Store) NewestSession(ctx context.Context) (Session, error) {
+	var sess Session
+	err := s.pool.QueryRow(ctx, `
+		SELECT token_hash, emby_user_id, emby_token, username, server_id, device_id, last_seen_at
+		FROM sessions ORDER BY last_seen_at DESC LIMIT 1`).
+		Scan(&sess.TokenHash, &sess.EmbyUserID, &sess.EmbyToken, &sess.Username,
+			&sess.ServerID, &sess.DeviceID, &sess.LastSeenAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return Session{}, ErrNotFound
+	}
+	if err != nil {
+		return Session{}, fmt.Errorf("store: newest session: %w", err)
+	}
+	return sess, nil
+}
@@ -0,0 +1,113 @@
+// Package store persists gateway sessions in Postgres.
+package store
+
+import (
+	"context"
+	_ "embed"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+//go:embed schema.sql
+var schema string
+
+// ErrNotFound is returned when a token does not match a live session.
+var ErrNotFound = errors.New("store: session not found")
+
+type Session struct {
+	TokenHash  []byte
+	EmbyUserID string
+	EmbyToken  string
+	Username   string
+	ServerID   string
+	DeviceID   string
+	LastSeenAt time.Time
+}
+
+type Store struct {
+	pool *pgxpool.Pool
+}
+
+func Open(ctx context.Context, databaseURL string) (*Store, error) {
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		return nil, fmt.Errorf("store: connect: %w", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("store: ping: %w", err)
+	}
+	return &Store{pool: pool}, nil
+}
+
+func (s *Store) Close() { s.pool.Close() }
+
+func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }
+
+// Migrate applies the schema. It is idempotent, so it runs on every boot.
+func (s *Store) Migrate(ctx context.Context) error {
+	if _, err := s.pool.Exec(ctx, schema); err != nil {
+		return fmt.Errorf("store: migrate: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) CreateSession(ctx context.Context, sess Session) error {
+	_, err := s.pool.Exec(ctx, `
+		INSERT INTO sessions (token_hash, emby_user_id, emby_token, username, server_id, device_id)
+		VALUES ($1, $2, $3, $4, $5, $6)
+		ON CONFLICT (token_hash) DO UPDATE SET
+			emby_token   = EXCLUDED.emby_token,
+			username     = EXCLUDED.username,
+			server_id    = EXCLUDED.server_id,
+			device_id    = EXCLUDED.device_id,
+			last_seen_at = now()`,
+		sess.TokenHash, sess.EmbyUserID, sess.EmbyToken, sess.Username, sess.ServerID, sess.DeviceID)
+	if err != nil {
+		return fmt.Errorf("store: create session: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) SessionByTokenHash(ctx context.Context, hash []byte) (Session, error) {
+	var sess Session
+	err := s.pool.QueryRow(ctx, `
+		SELECT token_hash, emby_user_id, emby_token, username, server_id, device_id, last_seen_at
+		FROM sessions WHERE token_hash = $1`, hash).
+		Scan(&sess.TokenHash, &sess.EmbyUserID, &sess.EmbyToken, &sess.Username,
+			&sess.ServerID, &sess.DeviceID, &sess.LastSeenAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return Session{}, ErrNotFound
+	}
+	if err != nil {
+		return Session{}, fmt.Errorf("store: load session: %w", err)
+	}
+	return sess, nil
+}
+
+// Touch records activity. Cheap enough to call on the auth path, and it is what the
+// idle-expiry sweep reads.
+func (s *Store) Touch(ctx context.Context, hash []byte) error {
+	_, err := s.pool.Exec(ctx, `UPDATE sessions SET last_seen_at = now() WHERE token_hash = $1`, hash)
+	return err
+}
+
+func (s *Store) DeleteSession(ctx context.Context, hash []byte) error {
+	_, err := s.pool.Exec(ctx, `DELETE FROM sessions WHERE token_hash = $1`, hash)
+	return err
+}
+
+// DeleteIdleSessions retires tokens unused for longer than idle, returning how many went.
+func (s *Store) DeleteIdleSessions(ctx context.Context, idle time.Duration) (int64, error) {
+	tag, err := s.pool.Exec(ctx,
+		`DELETE FROM sessions WHERE last_seen_at < now() - $1::interval`,
+		fmt.Sprintf("%d seconds", int64(idle.Seconds())))
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
@@ -0,0 +1,101 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// SyncRun records one library import.
+type SyncRun struct {
+	ID            int64      `json:"id"`
+	Kind          string     `json:"kind"`
+	Trigger       string     `json:"trigger"`
+	Status        string     `json:"status"`
+	StartedAt     time.Time  `json:"startedAt"`
+	FinishedAt    *time.Time `json:"finishedAt"`
+	ItemsSeen     int        `json:"itemsSeen"`
+	ItemsUpserted int        `json:"itemsUpserted"`
+	ItemsRemoved  int        `json:"itemsRemoved"`
+	Error         string     `json:"error"`
+}
+
+const (
+	SyncStatusRunning = "running"
+	SyncStatusSuccess = "success"
+	SyncStatusFailed  = "failed"
+)
+
+func (s *Store) StartSyncRun(ctx context.Context, kind, trigger string) (int64, error) {
+	var id int64
+	err := s.pool.QueryRow(ctx,
+		`INSERT INTO sync_runs (kind, trigger, status) VALUES ($1, $2, $3) RETURNING id`,
+		kind, trigger, SyncStatusRunning).Scan(&id)
+	if err != nil {
+		return 0, fmt.Errorf("store: start sync run: %w", err)
+	}
+	return id, nil
+}
+
+func (s *Store) FinishSyncRun(ctx context.Context, id int64, run SyncRun) error {
+	_, err := s.pool.Exec(ctx, `
+		UPDATE sync_runs
+		SET status = $2, finished_at = now(), items_seen = $3,
+		    items_upserted = $4, items_removed = $5, error = $6
+		WHERE id = $1`,
+		id, run.Status, run.ItemsSeen, run.ItemsUpserted, run.ItemsRemoved, run.Error)
+	if err != nil {
+		return fmt.Errorf("store: finish sync run: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) RecentSyncRuns(ctx context.Context, limit int) ([]SyncRun, error) {
+	rows, err := s.pool.Query(ctx, `
+		SELECT id, kind, trigger, status, started_at, finished_at,
+		       items_seen, items_upserted, items_removed, error
+		FROM sync_runs ORDER BY started_at DESC LIMIT $1`, limit)
+	if err != nil {
+		return nil, fmt.Errorf("store: recent sync runs: %w", err)
+	}
+	defer rows.Close()
+
+	runs := []SyncRun{}
+	for rows.Next() {
+		var run SyncRun
+		if err := rows.Scan(&run.ID, &run.Kind, &run.Trigger, &run.Status, &run.StartedAt,
+			&run.FinishedAt, &run.ItemsSeen, &run.ItemsUpserted, &run.ItemsRemoved, &run.Error); err != nil {
+			return nil, err
+		}
+		runs = append(runs, run)
+	}
+	return runs, rows.Err()
+}
+
+// LastSuccessfulSyncAt is the watermark an incremental import asks Emby about: "what has
+// changed since?" Nil means nothing has ever completed, so a full import is required.
+func (s *Store) LastSuccessfulSyncAt(ctx context.Context) (*time.Time, error) {
+	var at *time.Time
+	err := s.pool.QueryRow(ctx,
+		`SELECT max(started_at) FROM sync_runs WHERE status = $1`, SyncStatusSuccess).Scan(&at)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("store: last successful sync: %w", err)
+	}
+	return at, nil
+}
+
+// MarkStaleRunsFailed cleans up runs left "running" by a crash or a restart mid-import.
+func (s *Store) MarkStaleRunsFailed(ctx context.Context) error {
+	_, err := s.pool.Exec(ctx, `
+		UPDATE sync_runs
+		SET status = $1, finished_at = now(),
+		    error = 'interrupted — the gateway restarted while this import was running'
+		WHERE status = $2`, SyncStatusFailed, SyncStatusRunning)
+	return err
+}