Vellum/internal/media/discover.go
mithral c50b3cfb3b Wave 1: survive a real library
A 50k-file library used to wedge the database. This wave fixes the four
reasons why, plus a timestamp-ordering bug the work exposed.

Chunked scans (internal/library/scan.go)
- One transaction wrapped the whole reconciliation, holding SQLite's single
  write lock for the entire scan. Now 500 groups per transaction, and the two
  missing-file passes are batched too.
- Verified: a 20,000-file scan reports partial progress (scanned=1500 mid-scan)
  and a concurrent progress save returns 200 throughout. Under the old code
  that save blocked for the whole scan.
- Behaviour change: a scan is no longer all-or-nothing. A crash now leaves a
  consistent partial scan. The scan is idempotent, so re-running finishes it.
- res.Errors is capped at 100 (a broken mount used to append one string per
  file, then serialise the lot into the job row); the overflow is counted.

Concurrent job workers (internal/jobs/jobs.go)
- Run() was single-threaded, so a scan stalled every enrichment job behind it.
  Now 4 workers with jittered idle waits.
- Handlers run concurrently, so Register/lookup take an RWMutex, and a
  panicking handler is recovered and marked failed rather than killing a worker
  and stranding a row in 'running'.
- RecoverStale() requeues jobs left 'running' by a dead process; called at boot
  from main.go. Previously such a job stayed 'running' forever and the UI
  polled it forever.

Explore (internal/media/discover.go)
- Full-scanned and sorted the whole library on every home load, and was
  deterministic, so it showed the same 30 items every visit. Now a per-kind
  fair-share id range scan with a random start, wrap-around, dedup and
  interleaving. The doc comment claimed per-kind sampling before; now it does.

Indexes (migrations/0008_query_indexes.sql)
- Benchmarked with ANALYZE over 60k items: browse page 1 13.37ms -> 0.03ms
  (484x, full scan -> covering index), deep page 8x, recently added 3.2x.
- A candidate index for the Continue shelf was built, measured and REJECTED:
  the planner never chose it because `finished = 0` selects the majority of a
  user's rows, so it had no selectivity and would have cost a write on every
  progress save. Recorded in the migration rather than shipped as dead weight.

Canonical timestamps (internal/core/stamp.go)
- time.RFC3339Nano trims trailing zeros, so its fraction width varies and the
  strings are not lexicographically sortable: ".5Z" > ".55Z" as strings though
  .500s < .550s. That broke every ORDER BY <timestamp> and MAX(timestamp),
  including the Continue shelf and Most Viewed. It was wrong even in columns
  written 100% by Go, so it was not a cross-format issue.
- All 14 write sites now use core.Stamp, matching SQLite's own default format
  exactly. Read paths are untouched: RFC3339Nano parses both shapes, so
  existing rows keep working.

Also: busy_timeout 5s -> 15s. Chunked scans still delay a concurrent write, and
measured tail latency on a first scan of 15k new files reached 3.3s, leaving
almost no margin at 5s. Waiting beats failing with SQLITE_BUSY and losing a
playback position. A 10ms inter-batch yield was implemented, measured, found to
have no effect distinguishable from noise (overlapping ranges), and removed.

Verified end to end, not just compiled: orphan recovery logged at boot, scan
returns 202 in 0.7ms, Explore varies across loads, concurrent writes survive a
scan. gofmt/vet clean, 14 packages pass, -race clean on the queue, svelte-check
0/0, fresh install applies all 8 migrations.
2026-10-11 13:10:34 -04:00

384 lines
11 KiB
Go

package media
import (
"context"
"database/sql"
"fmt"
"math/rand"
"strings"
)
// RecordView increments a user's view count for an item.
//
// Called when an item is opened in a reader, so the count reflects consumption
// rather than a detail-page fetch. Opening the same thing repeatedly still
// counts: "most viewed" should reflect habit, not just distinct items.
func RecordView(ctx context.Context, db *sql.DB, userID, itemID int64) (int, error) {
var views int
err := db.QueryRowContext(ctx, `
INSERT INTO item_views (user_id, item_id, views, last_viewed_at)
VALUES (?, ?, 1, ?)
ON CONFLICT (user_id, item_id)
DO UPDATE SET views = views + 1, last_viewed_at = excluded.last_viewed_at
RETURNING views`,
userID, itemID, nowRFC3339()).Scan(&views)
if err != nil {
return 0, fmt.Errorf("media: record view: %w", err)
}
return views, nil
}
// MostViewed returns the user's most-opened items within the given libraries.
func MostViewed(ctx context.Context, db *sql.DB, userID int64, libraryIDs []int64, limit int) ([]shelfItem, error) {
if len(libraryIDs) == 0 {
return []shelfItem{}, nil
}
if limit <= 0 || limit > 100 {
limit = 20
}
in, args := inClause(libraryIDs)
args = append([]any{userID}, args...)
args = append(args, limit)
query := `
SELECT i.id, i.library_id, i.kind, i.title, v.views, v.last_viewed_at
FROM item_views v
JOIN media_items i ON i.id = v.item_id
WHERE v.user_id = ? AND i.missing = 0
AND i.library_id IN (` + in + `)
ORDER BY v.views DESC, v.last_viewed_at DESC
LIMIT ?`
rows, err := db.QueryContext(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("media: most viewed: %w", err)
}
defer rows.Close()
out := []shelfItem{}
for rows.Next() {
var s shelfItem
if err := rows.Scan(&s.ID, &s.LibraryID, &s.Kind, &s.Title, &s.Views, &s.UpdatedAt); err != nil {
return nil, err
}
out = append(out, s)
}
return out, rows.Err()
}
// itemRef is a lightweight item used by the suggestion and explore queries.
type itemRef struct {
ID int64
LibraryID int64
Kind string
Title string
Year int
}
// engagement describes what a user has been consuming, which is the basis for
// suggestions.
type engagement struct {
Kind string
Title string
Author string
}
// topEngagement finds the kind and author the user most recently engaged with,
// so suggestions can be explained in plain language.
func topEngagement(ctx context.Context, db *sql.DB, userID int64) (engagement, bool, error) {
var e engagement
err := db.QueryRowContext(ctx, `
SELECT i.kind, i.title,
COALESCE((SELECT p.name FROM item_people ip
JOIN people p ON p.id = ip.person_id
WHERE ip.item_id = i.id AND ip.role = 'author'
ORDER BY ip.ord LIMIT 1), '')
FROM progress p
JOIN media_items i ON i.id = p.item_id
WHERE p.user_id = ? AND i.missing = 0
ORDER BY p.updated_at DESC
LIMIT 1`, userID).Scan(&e.Kind, &e.Title, &e.Author)
if err == sql.ErrNoRows {
return engagement{}, false, nil
}
if err != nil {
return engagement{}, false, fmt.Errorf("media: top engagement: %w", err)
}
return e, true, nil
}
// Suggestions returns unstarted items related to what the user has been
// consuming: same author first, then the same kind, then anything recent.
//
// The rule is deliberately legible — a suggestion a person cannot understand is
// just noise. Each result carries the reason it was chosen.
func Suggestions(ctx context.Context, db *sql.DB, userID int64, libraryIDs []int64, limit int) ([]suggestion, error) {
if len(libraryIDs) == 0 {
return []suggestion{}, nil
}
if limit <= 0 || limit > 50 {
limit = 20
}
e, hasEngagement, err := topEngagement(ctx, db, userID)
if err != nil {
return nil, err
}
in, inArgs := inClause(libraryIDs)
out := []suggestion{}
seen := map[int64]bool{}
add := func(rows *sql.Rows, reason string) error {
defer rows.Close()
for rows.Next() {
var r itemRef
if err := rows.Scan(&r.ID, &r.LibraryID, &r.Kind, &r.Title, &r.Year); err != nil {
return err
}
if seen[r.ID] || len(out) >= limit {
continue
}
seen[r.ID] = true
out = append(out, suggestion{
ID: r.ID, LibraryID: r.LibraryID, Kind: r.Kind,
Title: r.Title, Year: r.Year, Reason: reason,
})
}
return rows.Err()
}
// Unstarted = no progress row for this user. Note the parameter order: the
// IN list comes first in the SQL, then the user id inside NOT EXISTS.
base := `
SELECT i.id, i.library_id, i.kind, i.title, COALESCE(i.year, 0)
FROM media_items i
WHERE i.missing = 0 AND i.library_id IN (` + in + `)
AND NOT EXISTS (SELECT 1 FROM progress p WHERE p.item_id = i.id AND p.user_id = ?)`
if hasEngagement {
// 1. Same author.
if e.Author != "" {
rows, err := db.QueryContext(ctx, base+`
AND EXISTS (SELECT 1 FROM item_people ip JOIN people p ON p.id = ip.person_id
WHERE ip.item_id = i.id AND p.name = ?)
ORDER BY i.added_at DESC LIMIT ?`,
withLibs(inArgs, userID, e.Author, limit)...)
if err != nil {
return nil, fmt.Errorf("media: suggestions by author: %w", err)
}
if err := add(rows, "More by "+e.Author); err != nil {
return nil, err
}
}
// 2. Same kind.
rows, err := db.QueryContext(ctx, base+`
AND i.kind = ? ORDER BY i.added_at DESC LIMIT ?`,
withLibs(inArgs, userID, e.Kind, limit)...)
if err != nil {
return nil, fmt.Errorf("media: suggestions by kind: %w", err)
}
if err := add(rows, "More "+kindNoun(e.Kind)); err != nil {
return nil, err
}
}
// 3. Anything recent, as a floor so the shelf is never empty.
rows, err := db.QueryContext(ctx, base+`
ORDER BY i.added_at DESC LIMIT ?`,
withLibs(inArgs, userID, limit)...)
if err != nil {
return nil, fmt.Errorf("media: suggestions fallback: %w", err)
}
if err := add(rows, "Recently added"); err != nil {
return nil, err
}
return out, nil
}
// Explore returns a mixed sample across every kind the user can see.
//
// Excludes anything already on the Continue shelf so the zone feels like
// discovery rather than a repeat. The shelf is sampled per kind: each kind
// present in the visible libraries gets a fair share of the limit, so one
// large library or format cannot drown out the rest. Within a kind the sample
// starts at a random id and walks an indexed, id-ordered range scan that stops
// once it has enough rows, so repeat visits surface different items without
// scanning or sorting the whole library.
func Explore(ctx context.Context, db *sql.DB, userID int64, libraryIDs []int64, limit int) ([]shelfItem, error) {
if len(libraryIDs) == 0 {
return []shelfItem{}, nil
}
if limit <= 0 || limit > 100 {
limit = 30
}
in, inArgs := inClause(libraryIDs)
// One small grouped query tells us which kinds are actually present, so we
// can give each of them a fair share of the shelf. Kinds with no visible
// items are ignored.
kindRows, err := db.QueryContext(ctx, `
SELECT kind, count(*)
FROM media_items
WHERE missing = 0 AND library_id IN (`+in+`)
GROUP BY kind`, inArgs...)
if err != nil {
return nil, fmt.Errorf("media: explore: %w", err)
}
kinds := []string{}
for kindRows.Next() {
var kind string
var n int64
if err := kindRows.Scan(&kind, &n); err != nil {
kindRows.Close()
return nil, err
}
if n > 0 {
kinds = append(kinds, kind)
}
}
if err := kindRows.Err(); err != nil {
kindRows.Close()
return nil, err
}
kindRows.Close()
if len(kinds) == 0 {
return []shelfItem{}, nil
}
// Divide the limit across the kinds that exist: one each, then spread the
// remainder over the first kinds. If there are more kinds than slots, the
// extras get nothing on this visit.
base := limit / len(kinds)
rem := limit % len(kinds)
seen := map[int64]bool{}
out := []shelfItem{}
// appendRows runs one bounded, id-ordered range scan for a kind and appends
// any rows not already seen. before selects the wrap-around half of the id
// range (id < start) instead of the forward half (id >= start). It returns
// how many new rows were added.
appendRows := func(kind string, start int64, before bool, n int) (int, error) {
if n <= 0 {
return 0, nil
}
cmp := ">="
if before {
cmp = "<"
}
q := `
SELECT i.id, i.library_id, i.kind, i.title, 0, i.added_at
FROM media_items i
WHERE i.missing = 0 AND i.library_id IN (` + in + `)
AND i.kind = ?
AND i.id ` + cmp + ` ?
AND NOT EXISTS (
SELECT 1 FROM progress p
WHERE p.item_id = i.id AND p.user_id = ? AND p.finished = 0
)
ORDER BY i.id
LIMIT ?`
rows, err := db.QueryContext(ctx, q, withLibs(inArgs, kind, start, userID, n)...)
if err != nil {
return 0, fmt.Errorf("media: explore: %w", err)
}
defer rows.Close()
added := 0
for rows.Next() {
var s shelfItem
if err := rows.Scan(&s.ID, &s.LibraryID, &s.Kind, &s.Title, &s.Percent, &s.UpdatedAt); err != nil {
return added, err
}
if seen[s.ID] {
continue
}
seen[s.ID] = true
out = append(out, s)
added++
}
return added, rows.Err()
}
for i, kind := range kinds {
share := base
if i < rem {
share++
}
if share <= 0 {
continue
}
var maxID int64
if err := db.QueryRowContext(ctx, `
SELECT COALESCE(MAX(id), 0)
FROM media_items
WHERE missing = 0 AND library_id IN (`+in+`) AND kind = ?`,
withLibs(inArgs, kind)...).Scan(&maxID); err != nil {
return nil, fmt.Errorf("media: explore: %w", err)
}
if maxID <= 0 {
continue
}
// A random start keeps repeat visits from showing the same slice.
start := rand.Int63n(maxID) + 1
got, err := appendRows(kind, start, false, share)
if err != nil {
return nil, err
}
if got < share {
// The start landed near the end of the id range; wrap to the
// beginning to fill the kind's share.
if _, err := appendRows(kind, start, true, share-got); err != nil {
return nil, err
}
}
}
// Mix the kinds instead of leaving them in contiguous blocks, then trim.
rand.Shuffle(len(out), func(a, b int) { out[a], out[b] = out[b], out[a] })
if len(out) > limit {
out = out[:limit]
}
return out, nil
}
// withLibs builds a query argument list: the IN-list args first (they appear
// first in these queries), then whatever follows.
//
// Written explicitly because the parameter order in the SQL is not the order
// the caller naturally thinks in, and getting it wrong silently swaps a user id
// for a library id.
func withLibs(inArgs []any, rest ...any) []any {
out := make([]any, 0, len(inArgs)+len(rest))
out = append(out, inArgs...)
out = append(out, rest...)
return out
}
// kindNoun turns a media kind into something readable in a sentence.
func kindNoun(kind string) string {
switch kind {
case "ebooks":
return "books"
case "comics":
return "comics"
case "audiobooks":
return "audiobooks"
case "music":
return "music"
case "photos":
return "photos"
case "games":
return "games"
default:
return strings.ReplaceAll(kind, "-", " ")
}
}