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 }