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.
This commit is contained in:
parent
a2c2f7a7c1
commit
df740d3e1a
13 changed files with 1548 additions and 2 deletions
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
227
internal/jobs/jobs.go
Normal file
227
internal/jobs/jobs.go
Normal file
|
|
@ -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()
|
||||
}
|
||||
198
internal/jobs/jobs_test.go
Normal file
198
internal/jobs/jobs_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
307
internal/metadata/enrich.go
Normal file
307
internal/metadata/enrich.go
Normal file
|
|
@ -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
|
||||
}
|
||||
245
internal/metadata/enrich_test.go
Normal file
245
internal/metadata/enrich_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
130
internal/metadata/googlebooks.go
Normal file
130
internal/metadata/googlebooks.go
Normal file
|
|
@ -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
|
||||
}
|
||||
171
internal/metadata/handlers.go
Normal file
171
internal/metadata/handlers.go
Normal file
|
|
@ -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")
|
||||
}
|
||||
114
internal/metadata/openlibrary.go
Normal file
114
internal/metadata/openlibrary.go
Normal file
|
|
@ -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])
|
||||
}
|
||||
73
internal/metadata/provider.go
Normal file
73
internal/metadata/provider.go
Normal file
|
|
@ -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
|
||||
}
|
||||
16
migrations/0004_metadata.sql
Normal file
16
migrations/0004_metadata.sql
Normal file
|
|
@ -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);
|
||||
|
|
@ -119,6 +119,16 @@ export const api = {
|
|||
|
||||
scanLibrary: (id: number) => request<ScanResult>('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}`),
|
||||
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
<div class="actions">
|
||||
<button class="btn btn-ghost" onclick={() => scan(lib.id)} disabled={busy}>Scan</button>
|
||||
<button class="btn btn-ghost" onclick={() => enrich(lib.id)} disabled={busy}>Enrich</button>
|
||||
<button class="btn btn-ghost danger" onclick={() => remove(lib.id, lib.name)} disabled={busy}>
|
||||
Delete
|
||||
</button>
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue