path: root/packages/meshbay-node/src/meshbay_node/media_cache.py

"""
MeshBay Node — TMDB/MusicBrainz metadata and thumbnail cache, shared by the
Videos and Music group apps.

Node-wide (not per-group, `data_dir/media_cache.db`), same rationale as
`tmdb_enabled`/`tmdb_api_token` (and `musicbrainz_enabled`,
docs/MESHBAY_DESIGN.md §9.8) living in `group_settings` under the `group_id=""`
sentinel (docs/MESHBAY_DESIGN.md §9.7): the credential/budget is one
operator's, and a thumbnail or cover image is the same bytes
regardless of which group happens to share the file. The `file_mbid`/
`mbid_meta` tables below are the Music app's equivalent of `file_tmdb`/
`tmdb_meta`, sharing the same `thumbs` table for cover art (a MusicBrainz
release's cover is cached under a synthetic `musicbrainz:{mbid}` file_id,
the same trick `_fetch_and_cache_poster` uses for a TMDB poster_path).

Disposable and rebuildable, like the rest of the file index
(docs/MESHBAY_DESIGN.md §6.5) — never
a second identity for a file. Every row here is keyed off a value the node
can already derive (a file's own blake3 id, or a TMDB id), so losing this
database costs re-probing/re-fetching, not data.
"""

import json
import logging
import time
from pathlib import Path

import aiosqlite

log = logging.getLogger(__name__)

_SCHEMA = """
CREATE TABLE IF NOT EXISTS file_tmdb (
    file_id     TEXT PRIMARY KEY,
    tmdb_id     TEXT NOT NULL,
    media_type  TEXT NOT NULL
);
-- Files whose match was set by an explicit operator "Fix match"
-- correction, not by the automatic matcher. `ops.rematch_video` and a
-- rename's re-enrichment wipe the *auto-resolved* file->tmdb mappings so
-- they re-resolve against the current matcher; a manual correction must
-- survive that, so it is recorded here and skipped.
CREATE TABLE IF NOT EXISTS tmdb_override (
    file_id     TEXT PRIMARY KEY
);
CREATE TABLE IF NOT EXISTS tmdb_meta (
    tmdb_id     TEXT NOT NULL,
    media_type  TEXT NOT NULL,
    json        TEXT NOT NULL,
    fetched_at  REAL NOT NULL,
    PRIMARY KEY (tmdb_id, media_type)
);
CREATE TABLE IF NOT EXISTS thumbs (
    thumb_hash  TEXT PRIMARY KEY,
    file_id     TEXT NOT NULL,
    jpeg        BLOB NOT NULL,
    -- Last time these bytes were served or written. The only thing that makes
    -- eviction possible: without it the cache had no notion of "least useful"
    -- and so no way to have a ceiling at all.
    used_at     REAL NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_thumbs_file ON thumbs(file_id);
-- idx_thumbs_used is NOT here: on a database that predates `used_at`, this
-- script runs before the ALTER TABLE that adds the column, and CREATE INDEX on
-- a column that does not exist yet fails -- which would have been every
-- existing node refusing to open its cache on the first start after upgrading.
-- It is created in _migrate(), after the column is guaranteed to be there.
CREATE TABLE IF NOT EXISTS season_meta (
    tmdb_id     TEXT NOT NULL,
    season      INTEGER NOT NULL,
    json        TEXT NOT NULL,
    fetched_at  REAL NOT NULL,
    PRIMARY KEY (tmdb_id, season)
);
-- Videos app: the ffprobe fields enrich.py derives alongside the
-- thumbnail — duration/width/height only, deliberately *not*
-- display_title/season/episode. Those come from guessit against the
-- filename, which is exactly what a rename needs re-derived
-- (_reenrich_renamed_video_entries exists for precisely that); caching
-- them here would silently defeat that mechanism by handing back the old
-- name's parse under the new name. ffprobe's own output has no such
-- concern — the same bytes probe the same regardless of what the file is
-- called — so only the content-only fields are safe to skip recomputing.
-- thumb_hash is not duplicated here either, for the same reason it isn't
-- in photo_meta — get_thumb_hash_by_file_id(file_id) already answers that,
-- and it too is content-only (a frame grab doesn't depend on the name).
CREATE TABLE IF NOT EXISTS video_meta (
    file_id       TEXT PRIMARY KEY,
    duration      INTEGER,
    width         INTEGER,
    height        INTEGER
);
CREATE TABLE IF NOT EXISTS file_mbid (
    file_id     TEXT PRIMARY KEY,
    mbid        TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS mbid_meta (
    mbid        TEXT PRIMARY KEY,
    json        TEXT NOT NULL,
    fetched_at  REAL NOT NULL
);
-- Photos app (docs/MESHBAY_DESIGN.md §9.9): the technical/EXIF fields
-- enrich_photo.py reads alongside the thumbnail. Durable for the same reason
-- `thumbs` is — without this, only the thumbnail bytes survived a restart, and
-- every image was still fully re-decoded through Pillow just to re-derive
-- width/height/taken_at/camera, which get_thumb_hash_by_file_id's own cache
-- hit had already proven unnecessary. thumb_hash is not duplicated here —
-- get_thumb_hash_by_file_id(file_id) already answers that, and a second copy
-- would just be one more place for the two to drift.
CREATE TABLE IF NOT EXISTS photo_meta (
    file_id     TEXT PRIMARY KEY,
    width       INTEGER,
    height      INTEGER,
    taken_at    INTEGER,
    camera      TEXT
);
"""

# TMDB overviews/ratings do drift; a file's own resolved tmdb_id does not
# need re-checking on this schedule, only the metadata blob (V3).
TMDB_META_TTL_SECS = 30 * 86400

# Same default as TMDB (docs/MESHBAY_DESIGN.md §9.8) — MusicBrainz release data is
# not expected to drift faster; revisit if that proves wrong in practice.
MUSICBRAINZ_META_TTL_SECS = 30 * 86400


# The blob store's ceiling.
#
# `thumbs` holds every generated thumbnail, every TMDB poster and backdrop,
# every Cover Art Archive image and every cached audio transcode. Rows were only
# ever removed when their source file left every group's index, so a library
# that merely *changes* over years — films watched once, albums added and
# removed, posters re-fetched after a rename — grew this database without any
# bound. Nothing here is precious: every row is keyed off a value the node can
# re-derive, which is what makes evicting the least recently used ones safe.
#
# 512 MB holds many thousands of posters and thumbnails; the audio transcodes
# are what actually consume it, at a few MB apiece.
MAX_THUMB_CACHE_BYTES = 512 * 1024 * 1024


class MediaCache:
    """Async SQLite cache for TMDB/MusicBrainz lookups and generated thumbnails/cover art."""

    def __init__(self, db_path: Path):
        self._db_path = db_path
        self._db: aiosqlite.Connection | None = None

    async def open(self) -> None:
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        self._db = await aiosqlite.connect(str(self._db_path))
        await self._db.executescript(_SCHEMA)
        await self._migrate()
        await self._db.commit()

    async def _migrate(self) -> None:
        """Add columns to databases that predate them.

        `CREATE TABLE IF NOT EXISTS` creates missing *tables* and never a
        missing *column*, so a new column reaches a fresh test database and
        never reaches a deployed node — the lesson `CLAUDE.md` records against
        `create_all()`. Every existing node has a `thumbs` table without
        `used_at`, and the eviction below reads it on every write.
        """
        async with self._db.execute("PRAGMA table_info(thumbs)") as cur:
            columns = {row[1] for row in await cur.fetchall()}
        if "used_at" not in columns:
            await self._db.execute(
                "ALTER TABLE thumbs ADD COLUMN used_at REAL NOT NULL DEFAULT 0")
            # Existing rows get "now" rather than 0: the alternative is that the
            # first write after an upgrade evicts the entire cache at once,
            # which is a correct-but-hostile reading of "least recently used"
            # for rows whose real age nothing recorded.
            await self._db.execute("UPDATE thumbs SET used_at = ?", (time.time(),))
            log.info("media_cache: added thumbs.used_at and seeded it")
        # Unconditional, and after the column is certain to exist: this is also
        # where a brand-new database gets the index, since _SCHEMA deliberately
        # does not carry it.
        await self._db.execute(
            "CREATE INDEX IF NOT EXISTS idx_thumbs_used ON thumbs(used_at)")

    async def close(self) -> None:
        if self._db:
            await self._db.close()
            self._db = None

    # ── file -> tmdb id mapping ──────────────────────────────────────────────

    async def get_file_tmdb(self, file_id: str) -> tuple[str, str] | None:
        """Returns (tmdb_id, media_type), or None if this file was never resolved."""
        async with self._db.execute(
            "SELECT tmdb_id, media_type FROM file_tmdb WHERE file_id = ?",
            (file_id,),
        ) as cur:
            row = await cur.fetchone()
        return (row[0], row[1]) if row else None

    async def set_file_tmdb(self, file_id: str, tmdb_id: str, media_type: str) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO file_tmdb (file_id, tmdb_id, media_type) "
            "VALUES (?, ?, ?)",
            (file_id, tmdb_id, media_type),
        )
        await self._db.commit()

    # ── manual "Fix match" corrections vs auto-resolved matches ──────────────

    async def mark_tmdb_override(self, file_id: str) -> None:
        """Record that this file's current match is an explicit operator
        correction — `clear_file_tmdb` / `clear_tmdb_matches` skip it."""
        await self._db.execute(
            "INSERT OR IGNORE INTO tmdb_override (file_id) VALUES (?)", (file_id,))
        await self._db.commit()

    async def clear_file_tmdb(self, file_id: str) -> None:
        """
        Drop one file's *auto-resolved* match. Used on rename: the new name
        re-derives the title, so the old name's match no longer applies —
        but file_tmdb is keyed by content hash, unchanged by a rename, so
        nothing else would ever dislodge it. A manual "Fix match"
        correction is kept: the content, hence what the operator corrected,
        is the same.
        """
        await self._db.execute(
            "DELETE FROM file_tmdb WHERE file_id = ? AND file_id NOT IN "
            "(SELECT file_id FROM tmdb_override)", (file_id,))
        await self._db.commit()

    async def drop_tmdb_match(self, file_id: str) -> None:
        """
        Full per-file reset: forget the match *and* any manual override
        marker, so the next `media_meta_req` re-resolves from scratch with
        the current matcher. This is the explicit operator "re-match this
        one" action (V13) — deliberately stronger than
        `clear_file_tmdb`, which spares an override.
        """
        await self._db.execute("DELETE FROM file_tmdb WHERE file_id = ?", (file_id,))
        await self._db.execute("DELETE FROM tmdb_override WHERE file_id = ?", (file_id,))
        await self._db.commit()

    async def clear_tmdb_matches(self, file_ids: list[str]) -> int:
        """
        Drop the auto-resolved file->tmdb mappings for these files so the
        next `media_meta_req` re-resolves each against the current matcher
        (`ops.rematch_video`, run by the operator after a matcher/parser
        fix). Manual "Fix match" corrections (`tmdb_override`) are left in
        place. Returns the number of rows removed.
        """
        if not self._db or not file_ids:
            return 0
        marks = ",".join("?" * len(file_ids))
        cur = await self._db.execute(
            f"DELETE FROM file_tmdb WHERE file_id IN ({marks}) AND file_id NOT IN "
            "(SELECT file_id FROM tmdb_override)", file_ids)
        await self._db.commit()
        return cur.rowcount

    # ── tmdb id -> metadata json ─────────────────────────────────────────────

    async def get_tmdb_meta(self, tmdb_id: str, media_type: str) -> dict | None:
        """Returns None on a miss or on an entry older than TMDB_META_TTL_SECS."""
        async with self._db.execute(
            "SELECT json, fetched_at FROM tmdb_meta WHERE tmdb_id = ? AND media_type = ?",
            (tmdb_id, media_type),
        ) as cur:
            row = await cur.fetchone()
        if not row:
            return None
        raw_json, fetched_at = row
        if time.time() - fetched_at > TMDB_META_TTL_SECS:
            return None
        return json.loads(raw_json)

    async def set_tmdb_meta(self, tmdb_id: str, media_type: str, meta: dict) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO tmdb_meta (tmdb_id, media_type, json, fetched_at) "
            "VALUES (?, ?, ?, ?)",
            (tmdb_id, media_type, json.dumps(meta), time.time()),
        )
        await self._db.commit()

    # ── tmdb id + season number -> season-level metadata json ────────────────
    #
    # A show's own overview (tmdb_meta above) is one static field an operator
    # found does not necessarily describe every season alike
    # (docs/MESHBAY_DESIGN.md §9.7) — this is TMDB's per-season
    # `overview`/`air_date`/`poster_path`, fetched and cached independently,
    # on the same staleness schedule.

    async def get_season_meta(self, tmdb_id: str, season: int) -> dict | None:
        async with self._db.execute(
            "SELECT json, fetched_at FROM season_meta WHERE tmdb_id = ? AND season = ?",
            (tmdb_id, season),
        ) as cur:
            row = await cur.fetchone()
        if not row:
            return None
        raw_json, fetched_at = row
        if time.time() - fetched_at > TMDB_META_TTL_SECS:
            return None
        return json.loads(raw_json)

    async def set_season_meta(self, tmdb_id: str, season: int, meta: dict) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO season_meta (tmdb_id, season, json, fetched_at) "
            "VALUES (?, ?, ?, ?)",
            (tmdb_id, season, json.dumps(meta), time.time()),
        )
        await self._db.commit()

    # ── file -> musicbrainz release id mapping (Music app) ───────────────────

    async def get_file_mbid(self, file_id: str) -> str | None:
        async with self._db.execute(
            "SELECT mbid FROM file_mbid WHERE file_id = ?", (file_id,),
        ) as cur:
            row = await cur.fetchone()
        return row[0] if row else None

    async def set_file_mbid(self, file_id: str, mbid: str) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO file_mbid (file_id, mbid) VALUES (?, ?)",
            (file_id, mbid),
        )
        await self._db.commit()

    # ── musicbrainz release id -> metadata json (Music app) ───────────────────

    async def get_mbid_meta(self, mbid: str) -> dict | None:
        """Returns None on a miss or on an entry older than MUSICBRAINZ_META_TTL_SECS."""
        async with self._db.execute(
            "SELECT json, fetched_at FROM mbid_meta WHERE mbid = ?", (mbid,),
        ) as cur:
            row = await cur.fetchone()
        if not row:
            return None
        raw_json, fetched_at = row
        if time.time() - fetched_at > MUSICBRAINZ_META_TTL_SECS:
            return None
        return json.loads(raw_json)

    async def set_mbid_meta(self, mbid: str, meta: dict) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO mbid_meta (mbid, json, fetched_at) VALUES (?, ?, ?)",
            (mbid, json.dumps(meta), time.time()),
        )
        await self._db.commit()

    # ── thumbnails ────────────────────────────────────────────────────────────

    async def get_thumb(self, thumb_hash: str) -> bytes | None:
        async with self._db.execute(
            "SELECT jpeg FROM thumbs WHERE thumb_hash = ?", (thumb_hash,),
        ) as cur:
            row = await cur.fetchone()
        if row is None:
            return None
        await self._touch_thumb(thumb_hash)
        return bytes(row[0])

    async def _touch_thumb(self, thumb_hash: str) -> None:
        """Record that these bytes were wanted, so eviction can tell what is
        still in use from what was cached once and never looked at again."""
        await self._db.execute(
            "UPDATE thumbs SET used_at = ? WHERE thumb_hash = ?",
            (time.time(), thumb_hash))
        await self._db.commit()

    async def get_thumb_hash_by_file_id(self, file_id: str) -> str | None:
        """
        A TMDB poster/backdrop is stored under a synthetic file_id
        (`tmdb:{poster_path}`, stable across requests for the same image) —
        this is how `_fetch_and_cache_poster` recognizes "already fetched"
        without knowing the content hash up front (that's only known once
        the bytes are downloaded).
        """
        async with self._db.execute(
            "SELECT thumb_hash FROM thumbs WHERE file_id = ?", (file_id,),
        ) as cur:
            row = await cur.fetchone()
        if row is None:
            return None
        # A poster resolved through its synthetic id is in use just as much as
        # one fetched by hash — this is the lookup `_fetch_and_cache_poster`
        # makes on every visit to a grid, and missing it would let the images a
        # busy library shows most often look like the coldest rows here.
        await self._touch_thumb(row[0])
        return row[0]

    async def put_thumb(self, thumb_hash: str, file_id: str, jpeg: bytes) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO thumbs (thumb_hash, file_id, jpeg, used_at) "
            "VALUES (?, ?, ?, ?)",
            (thumb_hash, file_id, jpeg, time.time()),
        )
        await self._db.commit()
        await self._evict_thumbs()

    async def thumb_bytes(self) -> int:
        """Total size of the blob store, as SQLite reports it."""
        async with self._db.execute(
            "SELECT COALESCE(SUM(LENGTH(jpeg)), 0) FROM thumbs") as cur:
            return int((await cur.fetchone())[0])

    async def _evict_thumbs(self, cap: int = MAX_THUMB_CACHE_BYTES) -> int:
        """Drop least-recently-used rows until the store is back under `cap`.

        Run on write rather than on a timer: a cache only grows when something
        is written to it, and a timer is one more thing to own and to get wrong.
        Writes are rare — one per new thumbnail, poster or transcode.

        The row just written is never the one evicted: it carries the newest
        `used_at` by construction. A single blob larger than the whole cap would
        otherwise evict everything and then itself, so the loop stops when only
        it is left rather than emptying the table for nothing.

        Note the database file does not shrink; SQLite reuses the freed pages.
        The point is the plateau, not the file size.
        """
        total = await self.thumb_bytes()
        if total <= cap:
            return 0
        removed = 0
        async with self._db.execute(
            "SELECT thumb_hash, LENGTH(jpeg) FROM thumbs ORDER BY used_at ASC"
        ) as cur:
            rows = await cur.fetchall()
        for thumb_hash, size in rows:
            if total <= cap or len(rows) - removed <= 1:
                break
            await self._db.execute(
                "DELETE FROM thumbs WHERE thumb_hash = ?", (thumb_hash,))
            total -= int(size)
            removed += 1
        if removed:
            await self._db.commit()
            log.info("media_cache: evicted %d cached image(s), now %.1f MB",
                     removed, total / 1048576)
        return removed

    # ── photo technical/EXIF fields (Photos app) ─────────────────────────────

    async def get_photo_meta(self, file_id: str) -> dict | None:
        async with self._db.execute(
            "SELECT width, height, taken_at, camera FROM photo_meta WHERE file_id = ?",
            (file_id,),
        ) as cur:
            row = await cur.fetchone()
        if not row:
            return None
        return {"width": row[0], "height": row[1], "taken_at": row[2], "camera": row[3]}

    async def put_photo_meta(self, file_id: str, width: int | None, height: int | None,
                             taken_at: int | None, camera: str | None) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO photo_meta (file_id, width, height, taken_at, camera) "
            "VALUES (?, ?, ?, ?, ?)",
            (file_id, width, height, taken_at, camera),
        )
        await self._db.commit()

    # ── video technical fields (Videos app) ──────────────────────────────────
    #
    # duration/width/height only — see the schema comment on why
    # display_title/season/episode are deliberately not cached here.

    async def get_video_meta(self, file_id: str) -> dict | None:
        async with self._db.execute(
            "SELECT duration, width, height FROM video_meta WHERE file_id = ?",
            (file_id,),
        ) as cur:
            row = await cur.fetchone()
        if not row:
            return None
        return {"duration": row[0], "width": row[1], "height": row[2]}

    async def put_video_meta(self, file_id: str, duration: int | None,
                             width: int | None, height: int | None) -> None:
        await self._db.execute(
            "INSERT OR REPLACE INTO video_meta (file_id, duration, width, height) "
            "VALUES (?, ?, ?, ?)",
            (file_id, duration, width, height),
        )
        await self._db.commit()

    # ── pruning ───────────────────────────────────────────────────────────────

    async def prune_file(self, file_id: str) -> None:
        """
        Called when a file leaves the index (deletion, unshared root). Removes
        its thumbnail, its per-app technical fields, and its file->tmdb/
        file->mbid mappings. `tmdb_meta`/`mbid_meta` rows are left alone —
        they're keyed by tmdb_id/mbid, not file_id, and other files (other
        episodes of the same show, other tracks of the same release) may
        still reference the same entry.
        """
        await self._db.execute("DELETE FROM thumbs WHERE file_id = ?", (file_id,))
        await self._db.execute("DELETE FROM photo_meta WHERE file_id = ?", (file_id,))
        await self._db.execute("DELETE FROM video_meta WHERE file_id = ?", (file_id,))
        await self._db.execute("DELETE FROM file_tmdb WHERE file_id = ?", (file_id,))
        await self._db.execute("DELETE FROM tmdb_override WHERE file_id = ?", (file_id,))
        await self._db.execute("DELETE FROM file_mbid WHERE file_id = ?", (file_id,))
        await self._db.commit()