2 Commits

Author SHA1 Message Date
Jonathan 034ae40084 feat(worker): share the MusicBrainz cache from scans/discovery
The worker's MB browser (discography + artist search, used by monitor sweeps,
library scans, and Last.fm discovery) re-fetched every time. Route it through
the same ApiCache table + logical keys the web app uses, so a discography one
process fetched warms the other, and repeat lookups stop hitting MB.

- apicache.py cached_api(): read-through, serve-stale-on-outage, never caches
  null, best-effort (a cache-DB error degrades to a live fetch, never breaks
  scan/sweep/discovery). Its own DEDICATED autocommit connection so it can't
  commit the worker's in-flight transaction, and autocommit avoids freezing
  Postgres now() (which would break TTL math).
- CachingMbBrowser decorates MusicBrainzBrowser, serializing to the SAME
  camelCase JSON shape the web writes (round-trip tested). registry.build_browser
  wraps it when a DSN is present; the Last.fm source's browser rides it too.
- Hourly prune drops ApiCache rows > 90d so the table stays bounded.
- worker 213 tests / 7-skip.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-14 11:17:51 +02:00
Jonathan bdbf9d237e feat(web): shared Postgres read-through cache for MusicBrainz (ApiCache)
Reopening an artist (discography), an album (tracklist), or repeating a lookup
re-hit MusicBrainz every time with no cache and no throttle. Adds a generic
Postgres-backed read-through cache keyed on the LOGICAL operation (not the
request URL) so the worker and web app share entries.

- New ApiCache(key, json, fetchedAt) model + migration (generic name so the
  Last.fm browse cache can ride the same table).
- lib/apicache.ts cached(): read-through, TTL from fetchedAt (tunable, no
  migration), serve-stale-on-outage, never caches null.
- Wrap the high-level MB fns with logical keys + TTLs: discography &
  tracklists & artist-name 30d, searchReleaseGroup 7d, searchArtists 1d.
  Search keys normalized (trim+lowercase) to match the worker byte-for-byte.
- album-modal: a session-scoped Map<rgMbid,Track[]> so reopening is instant
  with zero network (complements the DB cache).
- Test setup clears ApiCache between tests. web 147 tests, tsc + build clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-14 11:17:51 +02:00
14 changed files with 486 additions and 55 deletions
+1
View File
@@ -42,3 +42,4 @@ Thumbs.db
*.swp
slskd-downloads/
backups/
.playwright-mcp/
@@ -0,0 +1,11 @@
-- CreateTable
CREATE TABLE "ApiCache" (
"key" TEXT NOT NULL,
"json" JSONB NOT NULL,
"fetchedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "ApiCache_pkey" PRIMARY KEY ("key")
);
-- CreateIndex
CREATE INDEX "ApiCache_fetchedAt_idx" ON "ApiCache"("fetchedAt");
+12
View File
@@ -182,3 +182,15 @@ model ScanWorkItem {
@@unique([scanId, path])
@@index([scanId, done, artist, album])
}
// Generic read-through cache for external API responses shared by web + worker (keyed on
// the LOGICAL operation, not the request URL, so both processes share entries). TTL is
// applied at read time from fetchedAt in code (no expiresAt column → tunable, no migration).
// Named generically so the Last.fm browse cache can ride the same table.
model ApiCache {
key String @id
json Json
fetchedAt DateTime @default(now())
@@index([fetchedAt])
}
+17 -3
View File
@@ -6,6 +6,10 @@ import { CoverArt } from "./cover-art";
type Track = { position: number; title: string; lengthMs: number | null };
// Session-scoped, per-release-group tracklist cache: reopening the same album modal is
// instant with zero network. Complements the server-side ApiCache (which spans sessions).
const trackCache = new Map<string, Track[]>();
function fmt(ms: number | null): string {
if (ms == null) return "";
const s = Math.round(ms / 1000);
@@ -39,15 +43,25 @@ export function AlbumModal({
useEffect(() => {
if (!open) return;
if (!album.rgMbid) {
const rgMbid = album.rgMbid;
if (!rgMbid) {
setTracks("error");
return;
}
const hit = trackCache.get(rgMbid);
if (hit) {
setTracks(hit);
return;
}
setTracks("loading");
let cancelled = false;
fetch(`/api/mb/release-groups/${album.rgMbid}/tracks`)
fetch(`/api/mb/release-groups/${rgMbid}/tracks`)
.then((r) => (r.ok ? r.json() : Promise.reject()))
.then((d) => !cancelled && setTracks(d.tracks))
.then((d) => {
if (cancelled) return;
trackCache.set(rgMbid, d.tracks);
setTracks(d.tracks);
})
.catch(() => !cancelled && setTracks("error"));
return () => {
cancelled = true;
+54
View File
@@ -0,0 +1,54 @@
import { describe, it, expect, vi } from "vitest";
import { cached, normKey } from "./apicache";
import { prisma } from "@/lib/db";
describe("cached (read-through)", () => {
it("fetches once, then serves the cached value without re-fetching", async () => {
const fetchFresh = vi.fn(async () => ({ v: 1 }));
const first = await cached("k1", 3600, fetchFresh);
const second = await cached("k1", 3600, fetchFresh);
expect(first).toEqual({ v: 1 });
expect(second).toEqual({ v: 1 });
expect(fetchFresh).toHaveBeenCalledTimes(1); // second served from cache
});
it("re-fetches once the TTL has elapsed", async () => {
const fetchFresh = vi.fn(async () => ({ v: 2 }));
await cached("k2", 3600, fetchFresh);
// backdate the row beyond the ttl
await prisma.apiCache.update({ where: { key: "k2" }, data: { fetchedAt: new Date(Date.now() - 7200_000) } });
await cached("k2", 3600, fetchFresh);
expect(fetchFresh).toHaveBeenCalledTimes(2);
});
it("serves a stale value when the fresh fetch throws", async () => {
await cached("k3", 3600, async () => ({ v: "old" }));
await prisma.apiCache.update({ where: { key: "k3" }, data: { fetchedAt: new Date(Date.now() - 7200_000) } });
const result = await cached("k3", 3600, async () => {
throw new Error("upstream down");
});
expect(result).toEqual({ v: "old" }); // stale-on-outage
});
it("rethrows when the fetch fails and there is no cached row", async () => {
await expect(
cached("k4", 3600, async () => {
throw new Error("boom");
}),
).rejects.toThrow("boom");
// nothing was cached
expect(await prisma.apiCache.findUnique({ where: { key: "k4" } })).toBeNull();
});
it("does not cache null results", async () => {
const fetchFresh = vi.fn(async () => null);
await cached("k5", 3600, fetchFresh);
await cached("k5", 3600, fetchFresh);
expect(fetchFresh).toHaveBeenCalledTimes(2); // null never cached → always re-fetched
expect(await prisma.apiCache.findUnique({ where: { key: "k5" } })).toBeNull();
});
it("normKey trims and lowercases", () => {
expect(normKey(" Radiohead ")).toBe("radiohead");
});
});
+39
View File
@@ -0,0 +1,39 @@
import { prisma } from "@/lib/db";
// Read-through cache over the shared ApiCache table. Keyed on the LOGICAL operation (not the
// request URL) so the worker and web app share entries. TTL is applied here from fetchedAt,
// so TTLs are tunable in code with no migration. On a fetch failure a stale row is served.
// Null/undefined results are NOT cached (keeps negative caching out + avoids Prisma JsonNull).
export const CACHE_TTL = {
DAY: 86_400,
WEEK: 7 * 86_400,
MONTH: 30 * 86_400,
};
export async function cached<T>(key: string, ttlSec: number, fetchFresh: () => Promise<T>): Promise<T> {
const hit = await prisma.apiCache.findUnique({ where: { key } });
if (hit && (Date.now() - hit.fetchedAt.getTime()) / 1000 < ttlSec) {
return hit.json as T;
}
try {
const fresh = await fetchFresh();
if (fresh !== null && fresh !== undefined) {
await prisma.apiCache.upsert({
where: { key },
create: { key, json: fresh as object },
update: { json: fresh as object, fetchedAt: new Date() },
});
}
return fresh;
} catch (e) {
if (hit) return hit.json as T; // serve stale on an upstream outage
throw e;
}
}
/** Normalize a free-text search term so casing/whitespace don't fragment the cache. MUST
* match the worker's normalization byte-for-byte (both: trim + lowercase). */
export function normKey(s: string): string {
return s.trim().toLowerCase();
}
+61 -43
View File
@@ -1,3 +1,5 @@
import { cached, normKey, CACHE_TTL } from "@/lib/apicache";
const MB_BASE = "https://musicbrainz.org/ws/2";
const USER_AGENT = "Lyra/0.1 ( https://git.jger.nl/Jonathan/Lyra )";
@@ -30,28 +32,34 @@ function toReleaseGroup(g: any): ReleaseGroupInfo {
}
export async function searchArtists(query: string): Promise<ArtistHit[]> {
const data = await mbGet(`/artist?query=${encodeURIComponent(query)}&fmt=json&limit=8`);
return (data.artists ?? []).map((a: any) => ({
mbid: a.id,
name: a.name ?? "",
disambiguation: a.disambiguation ?? "",
}));
const q = normKey(query);
if (!q) return [];
return cached(`artist-search:${q}`, CACHE_TTL.DAY, async () => {
const data = await mbGet(`/artist?query=${encodeURIComponent(query)}&fmt=json&limit=8`);
return (data.artists ?? []).map((a: any) => ({
mbid: a.id,
name: a.name ?? "",
disambiguation: a.disambiguation ?? "",
}));
});
}
export async function browseReleaseGroups(artistMbid: string): Promise<ReleaseGroupInfo[]> {
const out: ReleaseGroupInfo[] = [];
let offset = 0;
for (;;) {
const data = await mbGet(
`/release-group?artist=${encodeURIComponent(artistMbid)}&fmt=json&limit=100&offset=${offset}`,
);
const groups: any[] = data["release-groups"] ?? [];
for (const g of groups) out.push(toReleaseGroup(g));
offset += groups.length;
const total = data["release-group-count"] ?? offset;
if (groups.length === 0 || offset >= total) break;
}
return out;
return cached(`artist-rgs:${artistMbid}`, CACHE_TTL.MONTH, async () => {
const out: ReleaseGroupInfo[] = [];
let offset = 0;
for (;;) {
const data = await mbGet(
`/release-group?artist=${encodeURIComponent(artistMbid)}&fmt=json&limit=100&offset=${offset}`,
);
const groups: any[] = data["release-groups"] ?? [];
for (const g of groups) out.push(toReleaseGroup(g));
offset += groups.length;
const total = data["release-group-count"] ?? offset;
if (groups.length === 0 || offset >= total) break;
}
return out;
});
}
function escapeLucenePhrase(s: string): string {
@@ -94,6 +102,12 @@ function albumPreferenceRank(g: any): number {
}
export async function searchReleaseGroup(artist: string, album: string): Promise<ReleaseGroupMatch | null> {
return cached(`rg-search:${normKey(artist)}|${normKey(album)}`, CACHE_TTL.WEEK, () =>
_searchReleaseGroup(artist, album),
);
}
async function _searchReleaseGroup(artist: string, album: string): Promise<ReleaseGroupMatch | null> {
const art = escapeLucenePhrase(artist);
const stripped = stripEditionQualifiers(album);
// Decreasing precision: exact phrase on the given title, then on the edition-stripped
@@ -121,32 +135,36 @@ export async function searchReleaseGroup(artist: string, album: string): Promise
export type Track = { position: number; title: string; lengthMs: number | null };
export async function getArtistName(mbid: string): Promise<string> {
const data = await mbGet(`/artist/${encodeURIComponent(mbid)}?fmt=json`);
return data.name ?? "";
return cached(`artist-name:${mbid}`, CACHE_TTL.MONTH, async () => {
const data = await mbGet(`/artist/${encodeURIComponent(mbid)}?fmt=json`);
return data.name ?? "";
});
}
export async function browseReleaseGroupTracks(rgMbid: string): Promise<Track[]> {
const data = await mbGet(
`/release?release-group=${encodeURIComponent(rgMbid)}&inc=recordings&fmt=json&limit=25`,
);
const releases: any[] = data.releases ?? [];
if (releases.length === 0) return [];
// Prefer an Official release, then the earliest date.
const pick = [...releases].sort((a, b) => {
const ao = a.status === "Official" ? 0 : 1;
const bo = b.status === "Official" ? 0 : 1;
if (ao !== bo) return ao - bo;
return (a.date ?? "9999").localeCompare(b.date ?? "9999");
})[0];
const tracks: Track[] = [];
for (const medium of pick.media ?? []) {
for (const t of medium.tracks ?? []) {
tracks.push({
position: Number(t.position ?? tracks.length + 1),
title: t.title ?? t.recording?.title ?? "",
lengthMs: t.length ?? t.recording?.length ?? null,
});
return cached(`rg-tracks:${rgMbid}`, CACHE_TTL.MONTH, async () => {
const data = await mbGet(
`/release?release-group=${encodeURIComponent(rgMbid)}&inc=recordings&fmt=json&limit=25`,
);
const releases: any[] = data.releases ?? [];
if (releases.length === 0) return [];
// Prefer an Official release, then the earliest date.
const pick = [...releases].sort((a, b) => {
const ao = a.status === "Official" ? 0 : 1;
const bo = b.status === "Official" ? 0 : 1;
if (ao !== bo) return ao - bo;
return (a.date ?? "9999").localeCompare(b.date ?? "9999");
})[0];
const tracks: Track[] = [];
for (const medium of pick.media ?? []) {
for (const t of medium.tracks ?? []) {
tracks.push({
position: Number(t.position ?? tracks.length + 1),
title: t.title ?? t.recording?.title ?? "",
lengthMs: t.length ?? t.recording?.length ?? null,
});
}
}
}
return tracks;
return tracks;
});
}
+1
View File
@@ -17,4 +17,5 @@ beforeEach(async () => {
await prisma.watchedArtist.deleteMany();
await prisma.discoverySuggestion.deleteMany();
await prisma.config.deleteMany();
await prisma.apiCache.deleteMany(); // read-through cache must not leak across tests
});
+66
View File
@@ -1,6 +1,72 @@
from lyra_worker.apicache import DAY, MONTH, cached_api, norm_key, open_cache_conn
from lyra_worker.browser import ArtistHit, ReleaseGroupInfo
# Serialize to the SAME camelCase JSON shape the web app stores (web/src/lib/musicbrainz.ts),
# so a discography one process fetched warms the cache for the other.
def _artist_hit_to_json(a: ArtistHit) -> dict:
return {"mbid": a.mbid, "name": a.name, "disambiguation": a.disambiguation}
def _artist_hit_from_json(d: dict) -> ArtistHit:
return ArtistHit(mbid=d["mbid"], name=d.get("name", ""), disambiguation=d.get("disambiguation", ""))
def _rg_to_json(r: ReleaseGroupInfo) -> dict:
return {
"rgMbid": r.rg_mbid,
"title": r.title,
"primaryType": r.primary_type or None,
"secondaryTypes": list(r.secondary_types),
"firstReleaseDate": r.first_release_date or None,
}
def _rg_from_json(d: dict) -> ReleaseGroupInfo:
return ReleaseGroupInfo(
rg_mbid=d["rgMbid"],
title=d.get("title", "") or "",
primary_type=d.get("primaryType") or "",
secondary_types=tuple(d.get("secondaryTypes") or ()),
first_release_date=d.get("firstReleaseDate") or "",
)
class CachingMbBrowser:
"""Wraps an MbBrowser with the shared ApiCache read-through, keyed on the same logical
strings the web side uses (`artist-search:{q}`, `artist-rgs:{mbid}`). Uses a dedicated,
lazily (re)opened cache connection; if it can't connect, it just calls through uncached."""
def __init__(self, inner, dsn: str):
self._inner = inner
self._dsn = dsn
self._conn = None
def _cache_conn(self):
if self._conn is not None and not self._conn.closed:
return self._conn
self._conn = open_cache_conn(self._dsn)
return self._conn
def search_artist(self, name: str) -> list[ArtistHit]:
conn = self._cache_conn()
if conn is None:
return self._inner.search_artist(name)
key = f"artist-search:{norm_key(name)}"
rows = cached_api(conn, key, DAY,
lambda: [_artist_hit_to_json(a) for a in self._inner.search_artist(name)])
return [_artist_hit_from_json(d) for d in rows]
def browse_release_groups(self, artist_mbid: str) -> list[ReleaseGroupInfo]:
conn = self._cache_conn()
if conn is None:
return self._inner.browse_release_groups(artist_mbid)
key = f"artist-rgs:{artist_mbid}"
rows = cached_api(conn, key, MONTH,
lambda: [_rg_to_json(r) for r in self._inner.browse_release_groups(artist_mbid)])
return [_rg_from_json(d) for d in rows]
class MusicBrainzBrowser:
"""Real MbBrowser via musicbrainzngs (imported lazily; not unit-tested offline)."""
+82
View File
@@ -0,0 +1,82 @@
"""Read-through cache over the shared ApiCache table (same table + logical keys the web app
uses, so web and worker share entries). TTL applied here from fetchedAt; serves a stale row
if the fresh fetch fails; never caches None.
Best-effort: all DB access is guarded so a cache-layer problem NEVER breaks the caller
(scan/sweep/discovery) — on a DB error it just degrades to a live fetch. It is given a
DEDICATED connection (not the worker's main conn) so its commits can't commit the worker's
in-flight transaction."""
from typing import Callable, TypeVar
import psycopg
from psycopg.types.json import Jsonb
T = TypeVar("T")
DAY = 86_400
WEEK = 7 * 86_400
MONTH = 30 * 86_400
def _read(conn, key: str):
try:
with conn.cursor() as cur:
cur.execute(
'SELECT json, extract(epoch from (now() - "fetchedAt")) FROM "ApiCache" WHERE key = %s',
(key,),
)
return cur.fetchone()
except Exception:
try:
conn.rollback()
except Exception:
pass
return None
def _write(conn, key: str, value) -> None:
try:
with conn.cursor() as cur:
cur.execute(
'INSERT INTO "ApiCache" (key, json, "fetchedAt") VALUES (%s, %s, now()) '
'ON CONFLICT (key) DO UPDATE SET json = EXCLUDED.json, "fetchedAt" = now()',
(key, Jsonb(value)),
)
conn.commit()
except Exception:
try:
conn.rollback()
except Exception:
pass
def cached_api(conn, key: str, ttl_sec: int, fetch_fresh: Callable[[], T]) -> T:
row = _read(conn, key)
if row is not None and row[1] < ttl_sec:
return row[0]
try:
fresh = fetch_fresh()
except Exception:
if row is not None:
return row[0] # serve stale on an upstream outage
raise
if fresh is not None:
_write(conn, key, fresh)
return fresh
def norm_key(s: str) -> str:
"""Match the web side's normKey (trim + lowercase) so search keys don't fragment."""
return s.strip().lower()
def open_cache_conn(dsn: str):
"""A dedicated AUTOCOMMIT connection for cache I/O, or None if it can't be opened (the
cache then simply no-ops). Kept separate from the worker's main connection. Autocommit is
essential: a read that hits the TTL returns without committing, and a non-autocommit conn
would leave that transaction open — freezing Postgres `now()` at the transaction start
(now() = transaction timestamp), which would break every subsequent TTL comparison."""
try:
return psycopg.connect(dsn, autocommit=True)
except Exception:
return None
+25 -2
View File
@@ -23,6 +23,7 @@ IDLE_SLEEP = 2.0
HEARTBEAT_SECONDS = 15.0 # how often to stamp worker.heartbeat (its updatedAt = liveness)
MONITOR_TICK_SECONDS = 60.0
DISCOVER_TICK_SECONDS = 300.0 # check the discovery interval every 5 min (interval itself is hours)
CACHE_PRUNE_SECONDS = 3600.0 # prune the shared ApiCache at most hourly (drops rows > 90d old)
DEST_ROOT = "/music" # must match run_pipeline's default library root
# Per-job download staging. Defaults inside the library (/music/.staging) to preserve prior
# behavior; set STAGING_ROOT (mounted as a separate volume) to keep temp download I/O off a
@@ -167,6 +168,23 @@ def maybe_run_discovery(conn, sources, browser, config, dcfg, now, last_discover
return now if due else last_discover_tick
def prune_api_cache(conn) -> None:
"""Drop ApiCache rows older than 90 days so the shared cache stays bounded. Best-effort:
a failure here (e.g. a transient error) must never take down the worker loop."""
try:
with conn.cursor() as cur:
cur.execute('DELETE FROM "ApiCache" WHERE "fetchedAt" < now() - interval \'90 days\'')
conn.commit()
except psycopg.OperationalError:
raise # a real disconnect — let the loop's reconnect handler deal with it
except Exception as e:
print(f"worker: ApiCache prune failed: {e}", flush=True)
try:
conn.rollback()
except Exception:
pass
def _require_secret_key() -> None:
"""Fail fast (or warn loudly) on a missing/placeholder/public LYRA_SECRET_KEY before
the worker touches any encrypted credential."""
@@ -192,9 +210,10 @@ def run_forever() -> None:
adapters = build_adapters(get_config(conn))
resolver = build_resolver()
tagger = build_tagger()
browser = build_browser()
_dsn = os.environ.get("DATABASE_URL")
browser = build_browser(_dsn) # shared ApiCache read-through when a DSN is present
probe = build_probe()
similarity_sources = build_similarity_sources(get_config(conn))
similarity_sources = build_similarity_sources(get_config(conn), _dsn)
# Backfill: releases pressed while the monitor was off never got reconciled, so they linger
# on the wanted list. Fix any already-owned ones once at startup.
fixed = fulfill_owned_releases(conn, MonitorConfig.from_config(get_config(conn)))
@@ -205,6 +224,7 @@ def run_forever() -> None:
last_tick = 0.0
last_discover_tick = 0.0
last_heartbeat = 0.0
last_prune = 0.0
try:
while True:
# A transient DB connection loss (e.g. AdminShutdown when Postgres restarts) raises
@@ -218,6 +238,9 @@ def run_forever() -> None:
if now - last_heartbeat >= HEARTBEAT_SECONDS:
_set_config(conn, "worker.heartbeat", "1") # value irrelevant; updatedAt is the signal
last_heartbeat = now
if now - last_prune >= CACHE_PRUNE_SECONDS:
prune_api_cache(conn) # keep the shared ApiCache bounded
last_prune = now
if mcfg.enabled and now - last_tick >= MONITOR_TICK_SECONDS:
try:
sweep(conn, browser, mcfg)
+11 -7
View File
@@ -1,4 +1,4 @@
from lyra_worker._mbbrowser import MusicBrainzBrowser
from lyra_worker._mbbrowser import CachingMbBrowser, MusicBrainzBrowser
from lyra_worker._musicbrainz import MusicBrainzResolver
from lyra_worker._mutagen import MutagenTagger
from lyra_worker._probe import MutagenProbe
@@ -42,9 +42,12 @@ def build_tagger() -> Tagger:
return MutagenTagger()
def build_browser() -> MbBrowser:
"""The MusicBrainz browser used by the background monitor."""
return MusicBrainzBrowser()
def build_browser(dsn: str | None = None) -> MbBrowser:
"""The MusicBrainz browser used by the background monitor. When a DSN is given, wrap it
in the shared ApiCache read-through so scans/discovery stop re-fetching discographies
(and warm the same cache the web modals read)."""
inner = MusicBrainzBrowser()
return CachingMbBrowser(inner, dsn) if dsn else inner
def build_probe() -> AudioProbe:
@@ -52,14 +55,15 @@ def build_probe() -> AudioProbe:
return MutagenProbe()
def build_similarity_sources(config: dict) -> list[SimilaritySource]:
def build_similarity_sources(config: dict, dsn: str | None = None) -> list[SimilaritySource]:
"""Similarity sources for discovery. ListenBrainz needs no credentials, so it is
always constructed; per-run reachability is checked via each source's health().
Last.fm is added only when an API key is configured."""
Last.fm is added only when an API key is configured. Its MB name→MBID lookups ride the
shared ApiCache when a DSN is given."""
sources: list[SimilaritySource] = [
ListenBrainzSource(base_url=config.get("discover.listenBrainzUrl") or "")
]
key = config.get("lastfm.api_key")
if key:
sources.append(LastfmSource(api_key=key, browser=MusicBrainzBrowser()))
sources.append(LastfmSource(api_key=key, browser=build_browser(dsn)))
return sources
+2
View File
@@ -30,6 +30,7 @@ def conn():
cur.execute('DELETE FROM "DiscoverySuggestion"')
cur.execute('DELETE FROM "DiscoverySeedContribution"')
cur.execute('DELETE FROM "ScanWorkItem"')
cur.execute('DELETE FROM "ApiCache"')
cur.execute('DELETE FROM "Config"')
connection.commit()
yield connection
@@ -47,6 +48,7 @@ def conn():
cur.execute('DELETE FROM "DiscoverySuggestion"')
cur.execute('DELETE FROM "DiscoverySeedContribution"')
cur.execute('DELETE FROM "ScanWorkItem"')
cur.execute('DELETE FROM "ApiCache"')
cur.execute('DELETE FROM "Config"')
connection.commit()
connection.close()
+104
View File
@@ -0,0 +1,104 @@
"""Read-through + shared-shape tests for the worker cache. cached_api uses the real
lyra_test DB (via the conn fixture); MB is never hit (fetch_fresh is a fake)."""
from lyra_worker.apicache import cached_api, norm_key
from lyra_worker._mbbrowser import (
_artist_hit_from_json, _artist_hit_to_json, _rg_from_json, _rg_to_json,
)
from lyra_worker.browser import ArtistHit, ReleaseGroupInfo
def _backdate(conn, key, seconds):
with conn.cursor() as cur:
cur.execute(
'UPDATE "ApiCache" SET "fetchedAt" = now() - make_interval(secs => %s) WHERE key = %s',
(seconds, key),
)
conn.commit()
def test_read_through_fetches_once(conn):
calls = []
def fresh():
calls.append(1)
return {"v": 1}
assert cached_api(conn, "k1", 3600, fresh) == {"v": 1}
assert cached_api(conn, "k1", 3600, fresh) == {"v": 1}
assert len(calls) == 1 # second served from cache
def test_refetches_after_ttl(conn):
calls = []
def fresh():
calls.append(1)
return {"v": len(calls)}
cached_api(conn, "k2", 3600, fresh)
_backdate(conn, "k2", 7200) # older than the ttl
cached_api(conn, "k2", 3600, fresh)
assert len(calls) == 2
def test_serves_stale_on_fetch_error(conn):
cached_api(conn, "k3", 3600, lambda: {"v": "old"})
_backdate(conn, "k3", 7200)
def boom():
raise RuntimeError("mb down")
assert cached_api(conn, "k3", 3600, boom) == {"v": "old"}
def test_reraises_when_no_cached_row(conn):
def boom():
raise RuntimeError("mb down")
try:
cached_api(conn, "k4-missing", 3600, boom)
except RuntimeError as e:
assert "mb down" in str(e)
else:
raise AssertionError("expected the fetch error to propagate")
def test_does_not_cache_none(conn):
calls = []
def fresh():
calls.append(1)
return None
cached_api(conn, "k5", 3600, fresh)
cached_api(conn, "k5", 3600, fresh)
assert len(calls) == 2 # None never cached
with conn.cursor() as cur:
cur.execute('SELECT 1 FROM "ApiCache" WHERE key = %s', ("k5",))
assert cur.fetchone() is None
def test_norm_key():
assert norm_key(" Radiohead ") == "radiohead"
def test_release_group_json_roundtrip():
r = ReleaseGroupInfo(rg_mbid="rg1", title="OK Computer", primary_type="Album",
secondary_types=("Live",), first_release_date="1997-05-21")
assert _rg_from_json(_rg_to_json(r)) == r
# a bare group (no type/date) round-trips to the worker's empty-string defaults
bare = ReleaseGroupInfo(rg_mbid="rg2", title="X")
assert _rg_from_json(_rg_to_json(bare)) == bare
def test_artist_hit_json_roundtrip():
a = ArtistHit(mbid="a1", name="Radiohead", disambiguation="UK band")
assert _artist_hit_from_json(_artist_hit_to_json(a)) == a
def test_worker_shape_matches_web_camelcase():
# the stored dict MUST use the same camelCase keys the web app writes/reads
d = _rg_to_json(ReleaseGroupInfo(rg_mbid="rg", title="T", primary_type="Album",
secondary_types=(), first_release_date="2000"))
assert set(d) == {"rgMbid", "title", "primaryType", "secondaryTypes", "firstReleaseDate"}
assert set(_artist_hit_to_json(ArtistHit(mbid="a", name="N"))) == {"mbid", "name", "disambiguation"}