From df740d3e1a81e402ccf65a8dae0f07742661a9b5 Mon Sep 17 00:00:00 2001 From: mithral Date: Fri, 9 Oct 2026 21:16:07 -0400 Subject: [PATCH] Phase 5: job queue and metadata providers - jobs: a table-backed queue in the same SQLite file. No Redis, no second process, included in the same backup. claim() is a single UPDATE...RETURNING so a job cannot run twice even with several workers. Failures back off exponentially; a missing handler fails permanently rather than retrying a bug. - metadata: Provider interface plus Open Library and Google Books (both keyless). The enricher tries providers in registration order, falls back on empty, and continues past a provider error. - migration 0004: metadata_fetch records the attempt, not just the match, so a rescan does not re-hammer a rate-limited API for an item that will never match. Errors are always retried; ok/notfound are cached for 30 days. - enrichment writes provider_ids, updates title/year, links authors through people/item_people, and downloads the cover as an asset. The cover download runs outside the transaction: a slow image host must not hold a write lock. - endpoints: queue enrichment per library, and inspect an item's matches. - web: Enrich button per library. Verified live against Open Library: "The Hobbit" matched /works/OL27482W, title and year 1937 written, a 301x500 JPEG cover downloaded and served. --- cmd/vellum/main.go | 17 ++ internal/core/migrate_test.go | 27 ++- internal/jobs/jobs.go | 227 ++++++++++++++++++ internal/jobs/jobs_test.go | 198 +++++++++++++++ internal/metadata/enrich.go | 307 ++++++++++++++++++++++++ internal/metadata/enrich_test.go | 245 +++++++++++++++++++ internal/metadata/googlebooks.go | 130 ++++++++++ internal/metadata/handlers.go | 171 +++++++++++++ internal/metadata/openlibrary.go | 114 +++++++++ internal/metadata/provider.go | 73 ++++++ migrations/0004_metadata.sql | 16 ++ web/src/lib/api.ts | 10 + web/src/lib/components/Libraries.svelte | 15 ++ 13 files changed, 1548 insertions(+), 2 deletions(-) create mode 100644 internal/jobs/jobs.go create mode 100644 internal/jobs/jobs_test.go create mode 100644 internal/metadata/enrich.go create mode 100644 internal/metadata/enrich_test.go create mode 100644 internal/metadata/googlebooks.go create mode 100644 internal/metadata/handlers.go create mode 100644 internal/metadata/openlibrary.go create mode 100644 internal/metadata/provider.go create mode 100644 migrations/0004_metadata.sql diff --git a/cmd/vellum/main.go b/cmd/vellum/main.go index 6fe992e..8cb1562 100644 --- a/cmd/vellum/main.go +++ b/cmd/vellum/main.go @@ -15,8 +15,10 @@ import ( "github.com/mithral/vellum/internal/auth" "github.com/mithral/vellum/internal/core" "github.com/mithral/vellum/internal/httpapi" + "github.com/mithral/vellum/internal/jobs" "github.com/mithral/vellum/internal/library" "github.com/mithral/vellum/internal/media" + "github.com/mithral/vellum/internal/metadata" "github.com/mithral/vellum/internal/modules/comics" "github.com/mithral/vellum/internal/modules/ebooks" "github.com/mithral/vellum/internal/search" @@ -80,6 +82,20 @@ func run() error { comics.NewHandlers(db, log, authenticator).Register(api.Mux()) search.NewHandlers(db, log, authenticator).Register(api.Mux()) + // Background work: a table-backed queue in the same database. Providers + // are registered primary-first; a fallback is only consulted when the + // primary returns nothing. + queue := jobs.New(db, log) + metadata.Register(metadata.NewOpenLibrary()) + metadata.Register(metadata.NewGoogleBooks()) + enricher := metadata.NewEnricher(db, log, cfg.ConfigDir) + queue.Register(metadata.JobKind, enricher.Handle) + metadata.NewHandlers(db, log, authenticator, queue).Register(api.Mux()) + + jobCtx, cancelJobs := context.WithCancel(context.Background()) + defer cancelJobs() + go queue.Run(jobCtx) + srv := &http.Server{ Addr: cfg.Listen, Handler: api.Handler(), @@ -115,6 +131,7 @@ func run() error { if err := srv.Shutdown(shutdownCtx); err != nil { return fmt.Errorf("graceful shutdown: %w", err) } + cancelJobs() log.Info("stopped") return nil } diff --git a/internal/core/migrate_test.go b/internal/core/migrate_test.go index 7e6d0f4..8d36952 100644 --- a/internal/core/migrate_test.go +++ b/internal/core/migrate_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "io" + "io/fs" "log/slog" "path/filepath" "strings" @@ -16,6 +17,27 @@ func discardLogger() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } +// highestMigrationNumber reads the embedded migrations so the version +// assertion does not need editing every time a migration is added. +func highestMigrationNumber(t *testing.T) int { + t.Helper() + names, err := fs.Glob(migrations.FS, "*.sql") + if err != nil { + t.Fatal(err) + } + highest := 0 + for _, name := range names { + n, err := migrationNumber(name) + if err != nil { + t.Fatal(err) + } + if n > highest { + highest = n + } + } + return highest +} + func openTempDB(t *testing.T) *sql.DB { t.Helper() db, err := OpenDB(filepath.Join(t.TempDir(), "test.db")) @@ -43,8 +65,9 @@ func TestMigrateAppliesAndIsIdempotent(t *testing.T) { if err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil { t.Fatalf("read user_version: %v", err) } - if version != 3 { - t.Errorf("user_version = %d, want 3", version) + want := highestMigrationNumber(t) + if version != want { + t.Errorf("user_version = %d, want %d", version, want) } var tables int diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go new file mode 100644 index 0000000..aeeada3 --- /dev/null +++ b/internal/jobs/jobs.go @@ -0,0 +1,227 @@ +// Package jobs is a table-backed work queue. +// +// It lives in the same SQLite file as everything else: no Redis, no second +// process, and the queue is included in the same backup. SQLite's single +// writer makes the claim below atomic without any extra locking. +package jobs + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "log/slog" + "time" +) + +// States a job can be in. +const ( + StateQueued = "queued" + StateRunning = "running" + StateDone = "done" + StateFailed = "failed" +) + +// Handler processes one job payload. +type Handler func(ctx context.Context, payload json.RawMessage) error + +// Queue claims and runs jobs. +type Queue struct { + db *sql.DB + log *slog.Logger + handlers map[string]Handler + maxAttempts int + pollEvery time.Duration + baseBackoff time.Duration +} + +// New builds a Queue with sensible defaults. +func New(db *sql.DB, log *slog.Logger) *Queue { + return &Queue{ + db: db, + log: log, + handlers: map[string]Handler{}, + maxAttempts: 5, + pollEvery: time.Second, + baseBackoff: 30 * time.Second, + } +} + +// Register installs the handler for a job kind. +func (q *Queue) Register(kind string, h Handler) { + q.handlers[kind] = h +} + +// Enqueue adds a job to run as soon as possible. +func (q *Queue) Enqueue(ctx context.Context, kind string, payload any) error { + return q.EnqueueAfter(ctx, kind, payload, 0) +} + +// EnqueueAfter adds a job that will not run before delay has elapsed. +func (q *Queue) EnqueueAfter(ctx context.Context, kind string, payload any, delay time.Duration) error { + body := "{}" + if payload != nil { + raw, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("jobs: marshal payload: %w", err) + } + body = string(raw) + } + _, err := q.db.ExecContext(ctx, ` + INSERT INTO jobs (kind, payload, state, run_after) + VALUES (?, ?, ?, ?)`, + kind, body, StateQueued, time.Now().UTC().Add(delay).Format(time.RFC3339Nano)) + if err != nil { + return fmt.Errorf("jobs: enqueue %s: %w", kind, err) + } + return nil +} + +// Run processes jobs until the context is cancelled. +func (q *Queue) Run(ctx context.Context) { + ticker := time.NewTicker(q.pollEvery) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + // Drain everything currently due before waiting again. + for { + ran, err := q.RunOnce(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + q.log.Error("job queue", "err", err) + break + } + if !ran { + break + } + } + } + } +} + +// RunOnce claims and runs at most one job. It reports whether it ran anything, +// which is what makes the queue testable without a ticker. +func (q *Queue) RunOnce(ctx context.Context) (bool, error) { + job, ok, err := q.claim(ctx) + if err != nil || !ok { + return false, err + } + + handler, exists := q.handlers[job.Kind] + if !exists { + // A missing handler is a programming error, not a transient fault. + // Retrying it just delays the inevitable and clutters the queue. + q.failPermanently(ctx, job, fmt.Errorf("no handler registered for %q", job.Kind)) + return true, nil + } + + if err := handler(ctx, job.Payload); err != nil { + q.log.Warn("job failed", "kind", job.Kind, "id", job.ID, "attempt", job.Attempts, "err", err) + q.finish(ctx, job, err) + return true, nil + } + q.finish(ctx, job, nil) + return true, nil +} + +type claimedJob struct { + ID int64 + Kind string + Payload json.RawMessage + Attempts int +} + +// claim atomically takes the oldest due job. +// +// The UPDATE ... RETURNING form means a job cannot be claimed twice even with +// several workers: the subselect and the state change are one statement. +func (q *Queue) claim(ctx context.Context) (claimedJob, bool, error) { + var ( + job claimedJob + payload string + ) + err := q.db.QueryRowContext(ctx, ` + UPDATE jobs SET state = ?, attempts = attempts + 1 + WHERE id = ( + SELECT id FROM jobs + WHERE state = ? AND run_after <= ? + ORDER BY run_after, id + LIMIT 1 + ) + RETURNING id, kind, payload, attempts`, + StateRunning, StateQueued, time.Now().UTC().Format(time.RFC3339Nano), + ).Scan(&job.ID, &job.Kind, &payload, &job.Attempts) + if errors.Is(err, sql.ErrNoRows) { + return claimedJob{}, false, nil + } + if err != nil { + return claimedJob{}, false, fmt.Errorf("jobs: claim: %w", err) + } + job.Payload = json.RawMessage(payload) + return job, true, nil +} + +// finish marks a job done, or reschedules it with exponential backoff until it +// exhausts its attempts. +func (q *Queue) finish(ctx context.Context, job claimedJob, runErr error) { + if runErr == nil { + if _, err := q.db.ExecContext(ctx, + `UPDATE jobs SET state = ?, error = '' WHERE id = ?`, StateDone, job.ID); err != nil { + q.log.Error("jobs: mark done", "id", job.ID, "err", err) + } + return + } + + if job.Attempts >= q.maxAttempts { + if _, err := q.db.ExecContext(ctx, + `UPDATE jobs SET state = ?, error = ? WHERE id = ?`, + StateFailed, runErr.Error(), job.ID); err != nil { + q.log.Error("jobs: mark failed", "id", job.ID, "err", err) + } + return + } + + backoff := q.baseBackoff * time.Duration(1<<(job.Attempts-1)) + runAfter := time.Now().UTC().Add(backoff).Format(time.RFC3339Nano) + if _, err := q.db.ExecContext(ctx, + `UPDATE jobs SET state = ?, error = ?, run_after = ? WHERE id = ?`, + StateQueued, runErr.Error(), runAfter, job.ID); err != nil { + q.log.Error("jobs: reschedule", "id", job.ID, "err", err) + } +} + +// failPermanently marks a job failed with no further attempts. +func (q *Queue) failPermanently(ctx context.Context, job claimedJob, err error) { + if _, err := q.db.ExecContext(ctx, + `UPDATE jobs SET state = ?, error = ? WHERE id = ?`, + StateFailed, err.Error(), job.ID); err != nil { + q.log.Error("jobs: mark failed", "id", job.ID, "err", err) + } +} + +// Stats reports job counts by state, for the health endpoint and debugging. +func (q *Queue) Stats(ctx context.Context) (map[string]int, error) { + rows, err := q.db.QueryContext(ctx, `SELECT state, count(*) FROM jobs GROUP BY state`) + if err != nil { + return nil, err + } + defer rows.Close() + + out := map[string]int{} + for rows.Next() { + var state string + var n int + if err := rows.Scan(&state, &n); err != nil { + return nil, err + } + out[state] = n + } + return out, rows.Err() +} diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go new file mode 100644 index 0000000..3c39799 --- /dev/null +++ b/internal/jobs/jobs_test.go @@ -0,0 +1,198 @@ +package jobs + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "io" + "log/slog" + "path/filepath" + "testing" + + "github.com/mithral/vellum/internal/core" + "github.com/mithral/vellum/migrations" +) + +func testDB(t *testing.T) *sql.DB { + t.Helper() + db, err := core.OpenDB(filepath.Join(t.TempDir(), "jobs.db")) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := core.Migrate(context.Background(), db, migrations.FS, discard()); err != nil { + t.Fatalf("Migrate: %v", err) + } + return db +} + +func discard() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } + +func stateOf(t *testing.T, db *sql.DB, id int64) (string, int, string) { + t.Helper() + var state, errMsg string + var attempts int + if err := db.QueryRow( + `SELECT state, attempts, error FROM jobs WHERE id = ?`, id).Scan(&state, &attempts, &errMsg); err != nil { + t.Fatal(err) + } + return state, attempts, errMsg +} + +func TestRunOnceExecutesHandler(t *testing.T) { + db := testDB(t) + q := New(db, discard()) + + var got json.RawMessage + q.Register("greet", func(_ context.Context, payload json.RawMessage) error { + got = payload + return nil + }) + + if err := q.Enqueue(context.Background(), "greet", map[string]string{"name": "ada"}); err != nil { + t.Fatal(err) + } + + ran, err := q.RunOnce(context.Background()) + if err != nil { + t.Fatal(err) + } + if !ran { + t.Fatal("RunOnce reported nothing to do") + } + if string(got) != `{"name":"ada"}` { + t.Errorf("payload = %s", got) + } + + // Nothing left to run. + ran, err = q.RunOnce(context.Background()) + if err != nil { + t.Fatal(err) + } + if ran { + t.Error("RunOnce ran a job that was already done") + } + + var state string + var n int + if err := db.QueryRow(`SELECT state, count(*) FROM jobs GROUP BY state`).Scan(&state, &n); err != nil { + t.Fatal(err) + } + if state != StateDone || n != 1 { + t.Errorf("state=%s count=%d, want done/1", state, n) + } +} + +func TestUnknownKindFailsImmediately(t *testing.T) { + db := testDB(t) + q := New(db, discard()) + + if err := q.Enqueue(context.Background(), "nobody-handles-this", nil); err != nil { + t.Fatal(err) + } + if _, err := q.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + + var state, errMsg string + if err := db.QueryRow(`SELECT state, error FROM jobs`).Scan(&state, &errMsg); err != nil { + t.Fatal(err) + } + if state != StateFailed { + t.Errorf("state = %s, want failed (a missing handler must not retry)", state) + } + if errMsg == "" { + t.Error("failure reason was not recorded") + } +} + +func TestFailureBacksOffThenFails(t *testing.T) { + db := testDB(t) + q := New(db, discard()) + q.maxAttempts = 2 + q.baseBackoff = 30 * 1000 * 1000 * 1000 // 30s + + boom := errors.New("transient") + q.Register("flaky", func(context.Context, json.RawMessage) error { return boom }) + + if err := q.Enqueue(context.Background(), "flaky", nil); err != nil { + t.Fatal(err) + } + + // First run fails and reschedules into the future. + if _, err := q.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + state, attempts, errMsg := stateOf(t, db, 1) + if state != StateQueued || attempts != 1 { + t.Fatalf("after first failure: state=%s attempts=%d", state, attempts) + } + if errMsg == "" { + t.Error("failure reason was not recorded") + } + + // Not due yet, so nothing runs. + ran, err := q.RunOnce(context.Background()) + if err != nil { + t.Fatal(err) + } + if ran { + t.Error("a backed-off job ran before its run_after") + } + + // Force it due and run again: attempts is now at the ceiling. + if _, err := db.Exec(`UPDATE jobs SET run_after = '1970-01-01T00:00:00Z'`); err != nil { + t.Fatal(err) + } + if _, err := q.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + state, attempts, _ = stateOf(t, db, 1) + if state != StateFailed { + t.Errorf("after exhausting attempts: state=%s, want failed", state) + } + if attempts != 2 { + t.Errorf("attempts = %d, want 2", attempts) + } +} + +func TestEnqueueAfterDelays(t *testing.T) { + db := testDB(t) + q := New(db, discard()) + q.Register("later", func(context.Context, json.RawMessage) error { return nil }) + + if err := q.EnqueueAfter(context.Background(), "later", nil, 60*1000*1000*1000); err != nil { + t.Fatal(err) + } + ran, err := q.RunOnce(context.Background()) + if err != nil { + t.Fatal(err) + } + if ran { + t.Error("a delayed job ran immediately") + } +} + +func TestStats(t *testing.T) { + db := testDB(t) + q := New(db, discard()) + q.Register("noop", func(context.Context, json.RawMessage) error { return nil }) + + for i := 0; i < 3; i++ { + if err := q.Enqueue(context.Background(), "noop", nil); err != nil { + t.Fatal(err) + } + } + if _, err := q.RunOnce(context.Background()); err != nil { + t.Fatal(err) + } + + stats, err := q.Stats(context.Background()) + if err != nil { + t.Fatal(err) + } + if stats[StateDone] != 1 || stats[StateQueued] != 2 { + t.Errorf("stats = %v, want done:1 queued:2", stats) + } +} diff --git a/internal/metadata/enrich.go b/internal/metadata/enrich.go new file mode 100644 index 0000000..155f5e1 --- /dev/null +++ b/internal/metadata/enrich.go @@ -0,0 +1,307 @@ +package metadata + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "os" + "path/filepath" + "strings" + "time" + + "github.com/mithral/vellum/internal/media" +) + +// JobKind is the queue job that enriches a single item. +const JobKind = "enrich_item" + +// retryAfter is how long a failed or empty lookup is trusted before the +// provider is asked again. It stops a rescan from re-hammering a rate-limited +// API for an item that will never match. +const retryAfter = 30 * 24 * time.Hour + +// maxCoverBytes caps a downloaded cover. +const maxCoverBytes = 8 << 20 + +// Enricher matches items against providers and writes the results back. +type Enricher struct { + db *sql.DB + log *slog.Logger + cacheDir string + client *http.Client + + // providers resolves the provider list for a media kind. It defaults to + // the global registry; tests override it to avoid network and global state. + providers func(kind string) []Provider +} + +// NewEnricher builds the enricher. +func NewEnricher(db *sql.DB, log *slog.Logger, cacheDir string) *Enricher { + return &Enricher{ + db: db, + log: log, + cacheDir: cacheDir, + client: &http.Client{Timeout: 20 * time.Second}, + providers: For, + } +} + +type enrichPayload struct { + ItemID int64 `json:"item_id"` +} + +// Handle is the queue handler for JobKind. +func (e *Enricher) Handle(ctx context.Context, payload json.RawMessage) error { + var p enrichPayload + if err := json.Unmarshal(payload, &p); err != nil { + return fmt.Errorf("metadata: bad payload: %w", err) + } + if p.ItemID <= 0 { + return errors.New("metadata: payload has no item_id") + } + return e.EnrichItem(ctx, p.ItemID) +} + +// EnrichItem tries each provider for the item's kind until one matches. +// +// A provider error is recorded and the next provider is tried. The function +// returns an error only for a database failure, because a missing match is a +// normal outcome and must not fail the job. +func (e *Enricher) EnrichItem(ctx context.Context, itemID int64) error { + it, err := media.LoadItem(ctx, e.db, itemID) + if err != nil { + if errors.Is(err, media.ErrItemNotFound) { + return nil // deleted since the job was queued + } + return err + } + + // Already matched by a previous run. + var matched int + if err := e.db.QueryRowContext(ctx, + `SELECT count(*) FROM provider_ids WHERE item_id = ?`, itemID).Scan(&matched); err != nil { + return err + } + if matched > 0 { + return nil + } + + providers := e.providers(it.Kind) + if len(providers) == 0 { + return nil + } + + q := Query{Title: it.Title, Kind: it.Kind} + for _, p := range providers { + if e.triedRecently(ctx, itemID, p.Name()) { + continue + } + + candidates, err := p.Search(ctx, q) + if err != nil { + e.log.Warn("metadata provider failed", "provider", p.Name(), "item", itemID, "err", err) + if recErr := e.record(ctx, itemID, p.Name(), "error", err.Error()); recErr != nil { + return recErr + } + continue + } + if len(candidates) == 0 { + if recErr := e.record(ctx, itemID, p.Name(), "notfound", ""); recErr != nil { + return recErr + } + continue + } + + if err := e.apply(ctx, it, candidates[0]); err != nil { + return err + } + if err := e.record(ctx, itemID, p.Name(), "ok", ""); err != nil { + return err + } + e.log.Info("metadata matched", "item", itemID, "provider", p.Name(), "title", candidates[0].Title) + return nil + } + return nil +} + +// apply writes a chosen candidate back to the database. +func (e *Enricher) apply(ctx context.Context, it media.Item, c Candidate) error { + tx, err := e.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck + + if _, err := tx.ExecContext(ctx, ` + INSERT INTO provider_ids (item_id, provider, external_id) VALUES (?, ?, ?) + ON CONFLICT (item_id, provider) DO UPDATE SET external_id = excluded.external_id`, + it.ID, c.Provider, c.ExternalID); err != nil { + return fmt.Errorf("metadata: store provider id: %w", err) + } + + // Only overwrite the scanner's title when the catalogue gave us one. + title := it.Title + if strings.TrimSpace(c.Title) != "" { + title = strings.TrimSpace(c.Title) + } + if _, err := tx.ExecContext(ctx, ` + UPDATE media_items SET title = ?, sort_title = ?, year = ? WHERE id = ?`, + title, strings.ToLower(title), nullableYear(c.Year), it.ID); err != nil { + return fmt.Errorf("metadata: update item: %w", err) + } + + if _, err := tx.ExecContext(ctx, + `DELETE FROM item_people WHERE item_id = ? AND role = 'author'`, it.ID); err != nil { + return err + } + for i, name := range c.Authors { + name = strings.TrimSpace(name) + if name == "" { + continue + } + var personID int64 + if err := tx.QueryRowContext(ctx, ` + INSERT INTO people (name) VALUES (?) + ON CONFLICT (name) DO UPDATE SET name = excluded.name + RETURNING id`, name).Scan(&personID); err != nil { + return fmt.Errorf("metadata: upsert person: %w", err) + } + if _, err := tx.ExecContext(ctx, + `INSERT INTO item_people (item_id, person_id, role, ord) VALUES (?, ?, 'author', ?) + ON CONFLICT (item_id, person_id, role) DO UPDATE SET ord = excluded.ord`, + it.ID, personID, i); err != nil { + return fmt.Errorf("metadata: link author: %w", err) + } + } + + if err := tx.Commit(); err != nil { + return err + } + + // Cover download is best-effort and outside the transaction: a slow image + // host must not hold a write lock. + if c.CoverURL != "" { + if err := e.downloadCover(ctx, it, c); err != nil { + e.log.Warn("cover download failed", "item", it.ID, "err", err) + } + } + return nil +} + +// downloadCover fetches a cover into the cache and records it as an asset. +func (e *Enricher) downloadCover(ctx context.Context, it media.Item, c Candidate) error { + var existing int + if err := e.db.QueryRowContext(ctx, + `SELECT count(*) FROM assets WHERE item_id = ? AND role = 'cover'`, it.ID).Scan(&existing); err != nil { + return err + } + if existing > 0 { + return nil + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.CoverURL, nil) + if err != nil { + return err + } + req.Header.Set("User-Agent", "Vellum/0.1 (self-hosted media server)") + + resp, err := e.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("cover status %d", resp.StatusCode) + } + + data, err := io.ReadAll(io.LimitReader(resp.Body, maxCoverBytes)) + if err != nil { + return err + } + if len(data) == 0 { + return errors.New("empty cover") + } + + dir, err := media.CoverCacheDir(e.cacheDir, it.Kind) + if err != nil { + return err + } + ext := coverExt(resp.Header.Get("Content-Type"), c.CoverURL) + outPath := filepath.Join(dir, fmt.Sprintf("%d%s", it.ID, ext)) + + tmp := outPath + ".tmp" + if err := os.WriteFile(tmp, data, 0o640); err != nil { + return err + } + if err := os.Rename(tmp, outPath); err != nil { + _ = os.Remove(tmp) + return err + } + + _, err = e.db.ExecContext(ctx, ` + INSERT INTO assets (item_id, role, path, bytes) VALUES (?, 'cover', ?, ?) + ON CONFLICT (item_id, role, path) DO NOTHING`, it.ID, outPath, len(data)) + return err +} + +// triedRecently reports whether a provider was already consulted for an item. +func (e *Enricher) triedRecently(ctx context.Context, itemID int64, provider string) bool { + var status, fetchedAt string + err := e.db.QueryRowContext(ctx, + `SELECT status, fetched_at FROM metadata_fetch WHERE item_id = ? AND provider = ?`, + itemID, provider).Scan(&status, &fetchedAt) + if err != nil { + return false + } + if status != "ok" && status != "notfound" { + return false // errors are always worth retrying + } + when, err := time.Parse(time.RFC3339Nano, fetchedAt) + if err != nil { + return false + } + return time.Since(when) < retryAfter +} + +func (e *Enricher) record(ctx context.Context, itemID int64, provider, status, detail string) error { + _, err := e.db.ExecContext(ctx, ` + INSERT INTO metadata_fetch (item_id, provider, status, detail, fetched_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (item_id, provider) + DO UPDATE SET status = excluded.status, detail = excluded.detail, fetched_at = excluded.fetched_at`, + itemID, provider, status, detail, time.Now().UTC().Format(time.RFC3339Nano)) + return err +} + +func coverExt(contentType, url string) string { + switch { + case strings.Contains(contentType, "png"): + return ".png" + case strings.Contains(contentType, "webp"): + return ".webp" + case strings.Contains(contentType, "gif"): + return ".gif" + case strings.Contains(contentType, "jpeg"), strings.Contains(contentType, "jpg"): + return ".jpg" + } + lower := strings.ToLower(url) + for _, ext := range []string{".png", ".webp", ".gif", ".jpeg", ".jpg"} { + if strings.Contains(lower, ext) { + return ext + } + } + return ".jpg" +} + +// nullableYear stores 0 as NULL rather than a literal year zero. +func nullableYear(year int) any { + if year <= 0 { + return nil + } + return year +} diff --git a/internal/metadata/enrich_test.go b/internal/metadata/enrich_test.go new file mode 100644 index 0000000..563dabb --- /dev/null +++ b/internal/metadata/enrich_test.go @@ -0,0 +1,245 @@ +package metadata + +import ( + "context" + "database/sql" + "errors" + "io" + "log/slog" + "path/filepath" + "testing" + + "github.com/mithral/vellum/internal/core" + "github.com/mithral/vellum/migrations" +) + +func testDB(t *testing.T) *sql.DB { + t.Helper() + db, err := core.OpenDB(filepath.Join(t.TempDir(), "meta.db")) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := core.Migrate(context.Background(), db, migrations.FS, discard()); err != nil { + t.Fatalf("Migrate: %v", err) + } + return db +} + +func discard() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } + +func seedItem(t *testing.T, db *sql.DB, title string) int64 { + t.Helper() + ctx := context.Background() + var libID int64 + if err := db.QueryRowContext(ctx, + `INSERT INTO libraries (name, kind, root_path) VALUES ('Ebooks','ebooks','/tmp/e') RETURNING id`).Scan(&libID); err != nil { + t.Fatal(err) + } + var itemID int64 + if err := db.QueryRowContext(ctx, ` + INSERT INTO media_items (library_id, kind, path, title, sort_title) + VALUES (?, 'ebooks', 'book.epub', ?, ?) RETURNING id`, + libID, title, title).Scan(&itemID); err != nil { + t.Fatal(err) + } + // A file row so media.LoadItem can resolve a modtime. + if _, err := db.ExecContext(ctx, + `INSERT INTO media_files (item_id, path, size, mtime) VALUES (?, 'book.epub', 1, '1')`, itemID); err != nil { + t.Fatal(err) + } + return itemID +} + +// fakeProvider is a Provider with scripted results. +type fakeProvider struct { + name string + kinds []string + results []Candidate + err error + calls int +} + +func (f *fakeProvider) Name() string { return f.name } +func (f *fakeProvider) Kinds() []string { return f.kinds } +func (f *fakeProvider) Search(context.Context, Query) ([]Candidate, error) { + f.calls++ + return f.results, f.err +} + +func enricherWith(t *testing.T, db *sql.DB, provs ...Provider) *Enricher { + t.Helper() + e := NewEnricher(db, discard(), t.TempDir()) + e.providers = func(string) []Provider { return provs } + return e +} + +func TestEnrichAppliesCandidate(t *testing.T) { + db := testDB(t) + itemID := seedItem(t, db, "the hobbit") + + fake := &fakeProvider{ + name: "fake", + kinds: []string{"ebooks"}, + results: []Candidate{{ + Provider: "fake", + ExternalID: "OL123W", + Title: "The Hobbit", + Authors: []string{"J.R.R. Tolkien"}, + Year: 1937, + Publisher: "Allen & Unwin", + }}, + } + e := enricherWith(t, db, fake) + + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Fatalf("EnrichItem: %v", err) + } + + ctx := context.Background() + var title string + var year sql.NullInt64 + if err := db.QueryRowContext(ctx, + `SELECT title, year FROM media_items WHERE id = ?`, itemID).Scan(&title, &year); err != nil { + t.Fatal(err) + } + if title != "The Hobbit" { + t.Errorf("title = %q, want The Hobbit", title) + } + if !year.Valid || year.Int64 != 1937 { + t.Errorf("year = %v, want 1937", year) + } + + var provider, external string + if err := db.QueryRowContext(ctx, + `SELECT provider, external_id FROM provider_ids WHERE item_id = ?`, itemID).Scan(&provider, &external); err != nil { + t.Fatalf("provider_ids not written: %v", err) + } + if provider != "fake" || external != "OL123W" { + t.Errorf("provider_ids = %s/%s", provider, external) + } + + var author string + if err := db.QueryRowContext(ctx, ` + SELECT p.name FROM item_people ip JOIN people p ON p.id = ip.person_id + WHERE ip.item_id = ? AND ip.role = 'author'`, itemID).Scan(&author); err != nil { + t.Fatalf("author not linked: %v", err) + } + if author != "J.R.R. Tolkien" { + t.Errorf("author = %q", author) + } + + var status string + if err := db.QueryRowContext(ctx, + `SELECT status FROM metadata_fetch WHERE item_id = ? AND provider = 'fake'`, itemID).Scan(&status); err != nil { + t.Fatal(err) + } + if status != "ok" { + t.Errorf("fetch status = %q, want ok", status) + } +} + +func TestEnrichFallsBackToSecondProvider(t *testing.T) { + db := testDB(t) + itemID := seedItem(t, db, "some book") + + empty := &fakeProvider{name: "empty", kinds: []string{"ebooks"}} + good := &fakeProvider{ + name: "good", + kinds: []string{"ebooks"}, + results: []Candidate{{Provider: "good", ExternalID: "g1", Title: "Some Book"}}, + } + e := enricherWith(t, db, empty, good) + + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Fatalf("EnrichItem: %v", err) + } + if empty.calls != 1 || good.calls != 1 { + t.Errorf("calls: empty=%d good=%d, want 1/1", empty.calls, good.calls) + } + + var provider string + if err := db.QueryRow( + `SELECT provider FROM provider_ids WHERE item_id = ?`, itemID).Scan(&provider); err != nil { + t.Fatal(err) + } + if provider != "good" { + t.Errorf("matched provider = %q, want good", provider) + } +} + +func TestEnrichContinuesAfterProviderError(t *testing.T) { + db := testDB(t) + itemID := seedItem(t, db, "some book") + + broken := &fakeProvider{name: "broken", kinds: []string{"ebooks"}, err: errors.New("network down")} + good := &fakeProvider{ + name: "good", + kinds: []string{"ebooks"}, + results: []Candidate{{Provider: "good", ExternalID: "g1", Title: "Some Book"}}, + } + e := enricherWith(t, db, broken, good) + + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Fatalf("a provider error must not fail the item: %v", err) + } + if good.calls != 1 { + t.Error("fallback was not consulted after an error") + } + + var status string + if err := db.QueryRow( + `SELECT status FROM metadata_fetch WHERE item_id = ? AND provider = 'broken'`, itemID).Scan(&status); err != nil { + t.Fatal(err) + } + if status != "error" { + t.Errorf("error status = %q, want error", status) + } +} + +func TestEnrichSkipsAlreadyMatched(t *testing.T) { + db := testDB(t) + itemID := seedItem(t, db, "book") + + fake := &fakeProvider{name: "fake", kinds: []string{"ebooks"}, + results: []Candidate{{Provider: "fake", ExternalID: "x", Title: "Book"}}} + e := enricherWith(t, db, fake) + + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Fatal(err) + } + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Fatal(err) + } + if fake.calls != 1 { + t.Errorf("provider called %d times, want 1 (second run must skip)", fake.calls) + } +} + +func TestEnrichDoesNotRetryRecentNotFound(t *testing.T) { + db := testDB(t) + itemID := seedItem(t, db, "unmatchable") + + empty := &fakeProvider{name: "empty", kinds: []string{"ebooks"}} + e := enricherWith(t, db, empty) + + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Fatal(err) + } + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Fatal(err) + } + if empty.calls != 1 { + t.Errorf("provider called %d times, want 1 (notfound is cached)", empty.calls) + } +} + +func TestEnrichIgnoresUnknownKind(t *testing.T) { + db := testDB(t) + itemID := seedItem(t, db, "book") + e := enricherWith(t, db) // no providers at all + + if err := e.EnrichItem(context.Background(), itemID); err != nil { + t.Errorf("EnrichItem with no providers errored: %v", err) + } +} diff --git a/internal/metadata/googlebooks.go b/internal/metadata/googlebooks.go new file mode 100644 index 0000000..fdd19f4 --- /dev/null +++ b/internal/metadata/googlebooks.go @@ -0,0 +1,130 @@ +package metadata + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +// GoogleBooks is a fallback catalogue for books. No API key for low volume. +type GoogleBooks struct { + Client *http.Client +} + +// NewGoogleBooks builds the provider with a bounded HTTP client. +func NewGoogleBooks() *GoogleBooks { + return &GoogleBooks{Client: &http.Client{Timeout: 10 * time.Second}} +} + +func (p *GoogleBooks) Name() string { return "googlebooks" } + +func (p *GoogleBooks) Kinds() []string { return []string{"ebooks"} } + +type gbResponse struct { + Items []struct { + ID string `json:"id"` + VolumeInfo struct { + Title string `json:"title"` + Authors []string `json:"authors"` + Publisher string `json:"publisher"` + Language string `json:"language"` + Description string `json:"description"` + PublishedDate string `json:"publishedDate"` + ImageLinks struct { + Thumbnail string `json:"thumbnail"` + } `json:"imageLinks"` + IndustryIdentifiers []struct { + Type string `json:"type"` + Identifier string `json:"identifier"` + } `json:"industryIdentifiers"` + } `json:"volumeInfo"` + } `json:"items"` +} + +func (p *GoogleBooks) Search(ctx context.Context, q Query) ([]Candidate, error) { + var terms []string + if q.ISBN != "" { + terms = append(terms, "isbn:"+q.ISBN) + } else { + if q.Title != "" { + terms = append(terms, "intitle:"+q.Title) + } + if q.Author != "" { + terms = append(terms, "inauthor:"+q.Author) + } + } + if len(terms) == 0 { + return nil, nil + } + + params := url.Values{} + params.Set("q", strings.Join(terms, "+")) + params.Set("maxResults", "5") + params.Set("printType", "books") + + endpoint := "https://www.googleapis.com/books/v1/volumes?" + params.Encode() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + + resp, err := p.Client.Do(req) + if err != nil { + return nil, fmt.Errorf("googlebooks: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("googlebooks: status %d", resp.StatusCode) + } + + var parsed gbResponse + if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { + return nil, fmt.Errorf("googlebooks: decode: %w", err) + } + + out := make([]Candidate, 0, len(parsed.Items)) + for _, item := range parsed.Items { + v := item.VolumeInfo + if v.Title == "" { + continue + } + c := Candidate{ + Provider: p.Name(), + ExternalID: item.ID, + Title: v.Title, + Authors: v.Authors, + Publisher: v.Publisher, + Language: v.Language, + Description: v.Description, + Year: parseYear(v.PublishedDate), + } + // Google returns http:// thumbnail URLs; https avoids mixed content. + if v.ImageLinks.Thumbnail != "" { + c.CoverURL = strings.Replace(v.ImageLinks.Thumbnail, "http://", "https://", 1) + } + for _, id := range v.IndustryIdentifiers { + if id.Type == "ISBN_13" || c.ISBN == "" { + c.ISBN = id.Identifier + } + } + out = append(out, c) + } + return out, nil +} + +// parseYear extracts a leading 4-digit year from dates like "2021-05-04". +func parseYear(date string) int { + if len(date) < 4 { + return 0 + } + n, err := strconv.Atoi(date[:4]) + if err != nil { + return 0 + } + return n +} diff --git a/internal/metadata/handlers.go b/internal/metadata/handlers.go new file mode 100644 index 0000000..e7ba5fb --- /dev/null +++ b/internal/metadata/handlers.go @@ -0,0 +1,171 @@ +package metadata + +import ( + "database/sql" + "errors" + "log/slog" + "net/http" + "strconv" + + "github.com/mithral/vellum/internal/auth" + "github.com/mithral/vellum/internal/httpx" + "github.com/mithral/vellum/internal/jobs" + "github.com/mithral/vellum/internal/media" +) + +// Handlers serves metadata endpoints. +type Handlers struct { + db *sql.DB + log *slog.Logger + auth *auth.Authenticator + queue *jobs.Queue +} + +// NewHandlers builds the metadata HTTP layer. +func NewHandlers(db *sql.DB, log *slog.Logger, a *auth.Authenticator, q *jobs.Queue) *Handlers { + return &Handlers{db: db, log: log, auth: a, queue: q} +} + +// Register mounts the routes. +func (h *Handlers) Register(mux *http.ServeMux) { + mux.Handle("POST /api/v1/libraries/{id}/enrich", h.auth.RequireAdmin(http.HandlerFunc(h.handleEnrichLibrary))) + mux.Handle("GET /api/v1/items/{id}/metadata", h.auth.Require(h.handleItemMetadata)) +} + +// handleEnrichLibrary queues enrichment for every unmatched item in a library. +func (h *Handlers) handleEnrichLibrary(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || id <= 0 { + httpx.WriteError(w, http.StatusBadRequest, "invalid library id") + return + } + + var exists int + if err := h.db.QueryRowContext(r.Context(), `SELECT count(*) FROM libraries WHERE id = ?`, id).Scan(&exists); err != nil { + h.fail(w, "check library", err) + return + } + if exists == 0 { + httpx.WriteError(w, http.StatusNotFound, "library not found") + return + } + + // Items that have no provider match yet. Already-matched items are skipped + // so a second click is cheap rather than a re-query of every provider. + rows, err := h.db.QueryContext(r.Context(), ` + SELECT i.id FROM media_items i + WHERE i.library_id = ? AND i.missing = 0 + AND NOT EXISTS (SELECT 1 FROM provider_ids p WHERE p.item_id = i.id)`, id) + if err != nil { + h.fail(w, "select items", err) + return + } + defer rows.Close() + + var ids []int64 + for rows.Next() { + var itemID int64 + if err := rows.Scan(&itemID); err != nil { + h.fail(w, "scan item id", err) + return + } + ids = append(ids, itemID) + } + if err := rows.Err(); err != nil { + h.fail(w, "iterate items", err) + return + } + + queued := 0 + for _, itemID := range ids { + if err := h.queue.Enqueue(r.Context(), JobKind, enrichPayload{ItemID: itemID}); err != nil { + h.fail(w, "enqueue enrich", err) + return + } + queued++ + } + + h.log.Info("enrichment queued", "library", id, "items", queued) + httpx.WriteJSON(w, http.StatusAccepted, map[string]any{"queued": queued}) +} + +// handleItemMetadata reports what the catalogue knows about an item. +func (h *Handlers) handleItemMetadata(w http.ResponseWriter, r *http.Request, u auth.User) { + id, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || id <= 0 { + httpx.WriteError(w, http.StatusBadRequest, "invalid item id") + return + } + + it, err := media.LoadItem(r.Context(), h.db, id) + if errors.Is(err, media.ErrItemNotFound) { + httpx.WriteError(w, http.StatusNotFound, "item not found") + return + } + if err != nil { + h.fail(w, "load item", err) + return + } + allowed, err := auth.CanAccessLibrary(r.Context(), h.db, u, it.LibraryID) + if err != nil || !allowed { + httpx.WriteError(w, http.StatusNotFound, "item not found") + return + } + + type providerID struct { + Provider string `json:"provider"` + ExternalID string `json:"external_id"` + } + type fetch struct { + Provider string `json:"provider"` + Status string `json:"status"` + Detail string `json:"detail"` + FetchedAt string `json:"fetched_at"` + } + + ids := []providerID{} + rows, err := h.db.QueryContext(r.Context(), + `SELECT provider, external_id FROM provider_ids WHERE item_id = ? ORDER BY provider`, id) + if err != nil { + h.fail(w, "select provider ids", err) + return + } + for rows.Next() { + var p providerID + if err := rows.Scan(&p.Provider, &p.ExternalID); err != nil { + rows.Close() + h.fail(w, "scan provider id", err) + return + } + ids = append(ids, p) + } + rows.Close() + + fetches := []fetch{} + frows, err := h.db.QueryContext(r.Context(), + `SELECT provider, status, detail, fetched_at FROM metadata_fetch WHERE item_id = ? ORDER BY provider`, id) + if err != nil { + h.fail(w, "select fetches", err) + return + } + defer frows.Close() + for frows.Next() { + var f fetch + if err := frows.Scan(&f.Provider, &f.Status, &f.Detail, &f.FetchedAt); err != nil { + h.fail(w, "scan fetch", err) + return + } + fetches = append(fetches, f) + } + + httpx.WriteJSON(w, http.StatusOK, map[string]any{ + "item_id": id, + "provider_ids": ids, + "fetch_history": fetches, + }) +} + +func (h *Handlers) fail(w http.ResponseWriter, what string, err error) { + h.log.Error(what, "err", err) + httpx.WriteError(w, http.StatusInternalServerError, "internal server error") +} diff --git a/internal/metadata/openlibrary.go b/internal/metadata/openlibrary.go new file mode 100644 index 0000000..28b9568 --- /dev/null +++ b/internal/metadata/openlibrary.go @@ -0,0 +1,114 @@ +package metadata + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +// OpenLibrary matches books by title and author. No API key, no registration. +type OpenLibrary struct { + Client *http.Client +} + +// NewOpenLibrary builds the provider with a bounded HTTP client. +func NewOpenLibrary() *OpenLibrary { + return &OpenLibrary{Client: &http.Client{Timeout: 10 * time.Second}} +} + +func (p *OpenLibrary) Name() string { return "openlibrary" } + +func (p *OpenLibrary) Kinds() []string { return []string{"ebooks", "audiobooks"} } + +// olSearchResponse is the subset of the search API this provider reads. +type olSearchResponse struct { + Docs []struct { + Key string `json:"key"` + Title string `json:"title"` + AuthorName []string `json:"author_name"` + FirstPublish int `json:"first_publish_year"` + Publisher []string `json:"publisher"` + Language []string `json:"language"` + CoverID int `json:"cover_i"` + ISBN []string `json:"isbn"` + } `json:"docs"` +} + +func (p *OpenLibrary) Search(ctx context.Context, q Query) ([]Candidate, error) { + params := url.Values{} + params.Set("limit", "5") + params.Set("fields", "key,title,author_name,first_publish_year,publisher,language,cover_i,isbn") + if q.ISBN != "" { + params.Set("isbn", q.ISBN) + } else { + if q.Title != "" { + params.Set("title", q.Title) + } + if q.Author != "" { + params.Set("author", q.Author) + } + } + if params.Get("isbn") == "" && params.Get("title") == "" { + return nil, nil + } + + endpoint := "https://openlibrary.org/search.json?" + params.Encode() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + // Open Library asks callers to identify themselves. + req.Header.Set("User-Agent", "Vellum/0.1 (self-hosted media server)") + + resp, err := p.Client.Do(req) + if err != nil { + return nil, fmt.Errorf("openlibrary: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("openlibrary: status %d", resp.StatusCode) + } + + var parsed olSearchResponse + if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { + return nil, fmt.Errorf("openlibrary: decode: %w", err) + } + + out := make([]Candidate, 0, len(parsed.Docs)) + for _, d := range parsed.Docs { + if d.Title == "" { + continue + } + c := Candidate{ + Provider: p.Name(), + ExternalID: d.Key, + Title: d.Title, + Authors: d.AuthorName, + Year: d.FirstPublish, + Language: first(d.Language), + } + if len(d.Publisher) > 0 { + c.Publisher = d.Publisher[0] + } + if len(d.ISBN) > 0 { + c.ISBN = d.ISBN[0] + } + if d.CoverID > 0 { + c.CoverURL = "https://covers.openlibrary.org/b/id/" + strconv.Itoa(d.CoverID) + "-L.jpg" + } + out = append(out, c) + } + return out, nil +} + +func first(s []string) string { + if len(s) == 0 { + return "" + } + return strings.TrimSpace(s[0]) +} diff --git a/internal/metadata/provider.go b/internal/metadata/provider.go new file mode 100644 index 0000000..acb7c6c --- /dev/null +++ b/internal/metadata/provider.go @@ -0,0 +1,73 @@ +// Package metadata resolves media items against external catalogues. +// +// Providers are optional by design: a library with no network still browses, +// because a failed lookup is recorded and skipped, never surfaced as a scan or +// request failure. +package metadata + +import ( + "context" + "strings" + "sync" +) + +// Query is what a provider is asked to match. +type Query struct { + Title string + Author string + ISBN string + Kind string +} + +// Candidate is one possible match returned by a provider. +type Candidate struct { + Provider string `json:"provider"` + ExternalID string `json:"external_id"` + Title string `json:"title"` + Authors []string `json:"authors"` + Year int `json:"year"` + Publisher string `json:"publisher"` + Language string `json:"language"` + Description string `json:"description"` + ISBN string `json:"isbn"` + CoverURL string `json:"cover_url"` +} + +// Provider is one external catalogue. +type Provider interface { + // Name is the stable identifier stored in provider_ids.provider. + Name() string + // Kinds lists the media kinds this provider can match. + Kinds() []string + // Search returns candidates, best first. An empty result is not an error. + Search(ctx context.Context, q Query) ([]Candidate, error) +} + +var ( + mu sync.RWMutex + providers []Provider +) + +// Register adds a provider. Later registrations are consulted later, so +// registering a primary provider before a fallback gives the intended order. +func Register(p Provider) { + mu.Lock() + defer mu.Unlock() + providers = append(providers, p) +} + +// For returns the providers that handle a media kind, in registration order. +func For(kind string) []Provider { + mu.RLock() + defer mu.RUnlock() + var out []Provider + for _, p := range providers { + for _, k := range p.Kinds() { + if strings.EqualFold(k, kind) { + out = append(out, p) + break + } + } + } + return out +} diff --git a/migrations/0004_metadata.sql b/migrations/0004_metadata.sql new file mode 100644 index 0000000..e4010cf --- /dev/null +++ b/migrations/0004_metadata.sql @@ -0,0 +1,16 @@ +-- 0004_metadata.sql — record which providers have been tried for an item. +-- +-- provider_ids records the match that was chosen. This table records the +-- attempt, including failures, so a rescan or a restart does not re-hammer a +-- rate-limited API for an item that will never match. + +CREATE TABLE metadata_fetch ( + item_id INTEGER NOT NULL REFERENCES media_items(id) ON DELETE CASCADE, + provider TEXT NOT NULL, + status TEXT NOT NULL, -- ok | notfound | error + detail TEXT NOT NULL DEFAULT '', + fetched_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + PRIMARY KEY (item_id, provider) +); + +CREATE INDEX idx_metadata_fetch_item ON metadata_fetch(item_id); diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 7d796d3..d2cae61 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -119,6 +119,16 @@ export const api = { scanLibrary: (id: number) => request('POST', `/api/v1/libraries/${id}/scan`), + enrichLibrary: (id: number) => + request<{ queued: number }>('POST', `/api/v1/libraries/${id}/enrich`), + + itemMetadata: (id: number) => + request<{ + item_id: number + provider_ids: { provider: string; external_id: string }[] + fetch_history: { provider: string; status: string; detail: string; fetched_at: string }[] + }>('GET', `/api/v1/items/${id}/metadata`), + items: (libraryId: number, limit = 200) => request<{ items: Item[]; limit: number }>('GET', `/api/v1/libraries/${libraryId}/items?limit=${limit}`), diff --git a/web/src/lib/components/Libraries.svelte b/web/src/lib/components/Libraries.svelte index 959254f..ce232d5 100644 --- a/web/src/lib/components/Libraries.svelte +++ b/web/src/lib/components/Libraries.svelte @@ -47,6 +47,20 @@ } } + async function enrich(id: number) { + busy = true + error = null + notice = null + try { + const res = await api.enrichLibrary(id) + notice = `Queued metadata lookup for ${res.queued} item${res.queued === 1 ? '' : 's'}. Titles and covers will fill in shortly.` + } catch (err) { + error = err instanceof ApiError ? err.message : 'could not queue enrichment' + } finally { + busy = false + } + } + async function remove(id: number, label: string) { if (!confirm(`Delete the library "${label}"? Items and progress for it are removed.`)) return busy = true @@ -121,6 +135,7 @@ {#if $user?.is_admin}
+