Compare commits
2 Commits
d275a0ddd6
...
034ae40084
| Author | SHA1 | Date | |
|---|---|---|---|
| 034ae40084 | |||
| bdbf9d237e |
@@ -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");
|
||||
@@ -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])
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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
@@ -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;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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
|
||||
});
|
||||
|
||||
@@ -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)."""
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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"}
|
||||
Reference in New Issue
Block a user