Vellum/internal/metadata/enrich.go

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
}