Vellum/internal/metadata/handlers.go
mithral df740d3e1a 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.
2026-10-09 21:16:07 -04:00

171 lines
4.7 KiB
Go

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")
}