commit 94638781984174cf89e256ee853c2c914c0deb89
parent b8b4ca3d81003f204d37b6f0c9fc01a06e7bcd76
Author: mtmn <miro@haravara.org>
Date: Sat, 18 Apr 2026 22:13:38 +0200
feat: rewrite into folds, cover/genre rewrite
Diffstat:
8 files changed, 311 insertions(+), 314 deletions(-)
diff --git a/src/Config.purs b/src/Config.purs
@@ -10,6 +10,7 @@ import Data.Maybe (Maybe(..), fromMaybe, isJust, isNothing)
import Data.String (joinWith)
import Data.Traversable (traverse)
import Effect (Effect)
+import Control.Monad.Error.Class (throwError)
import Effect.Aff (Aff, makeAff, nonCanceler)
import Effect.Class (liftEffect)
import Effect.Exception (error)
@@ -77,7 +78,7 @@ loadConfig path = do
(\j -> cb (Right j))
*> pure nonCanceler
rawUsers <- case decodeUsersJson json of
- Left msg -> makeAff \cb -> cb (Left (error msg)) *> pure nonCanceler
+ Left msg -> throwError (error msg)
Right users -> pure users
portStr <- liftEffect $ lookupEnv "PORT"
lastfmApiKey <- liftEffect $ lookupEnv "LASTFM_API_KEY"
@@ -111,7 +112,7 @@ loadConfig path = do
metricsEnabled = metricsEnabledStr == Just "true"
fullConfig = { port, metricsEnabled, users: map (\u -> u { config = fillCreds u.config }) rawUsers }
case validateConfig fullConfig of
- Left msg -> makeAff \cb -> cb (Left (error msg)) *> pure nonCanceler
+ Left msg -> throwError (error msg)
Right cfg -> pure cfg
validateConfig :: AppConfig -> Either String AppConfig
diff --git a/src/Db.js b/src/Db.js
@@ -27,20 +27,15 @@ export const checkpointImpl = (conn) => (cb) => () => {
});
};
+// DuckDB returns BIGINT as BigInt, which JSON.stringify doesn't support.
+// Pure transformation: produce new row objects rather than mutating in place.
+const convertBigInts = (row) =>
+ Object.fromEntries(
+ Object.entries(row).map(([k, v]) => [k, typeof v === "bigint" ? Number(v) : v]),
+ );
+
export const allImpl = (conn) => (sql) => (params) => (cb) => () => {
conn.all(sql, ...params, (err, rows) => {
- if (rows) {
- // DuckDB returns BIGINT as BigInt, which JSON.stringify doesn't support.
- // We convert them to Numbers here.
- for (let i = 0; i < rows.length; i++) {
- const row = rows[i];
- for (const key in row) {
- if (typeof row[key] === "bigint") {
- row[key] = Number(row[key]);
- }
- }
- }
- }
- cb(err)(rows)();
+ cb(err)(rows ? rows.map(convertBigInts) : rows)();
});
};
diff --git a/src/Db.purs b/src/Db.purs
@@ -21,6 +21,7 @@ import Types (Listen(..), TrackMetadata(..), MbidMapping(..), Stats(..), StatsEn
import Foreign.Object as Object
import Data.Nullable (Nullable, toMaybe, toNullable)
import Data.Array (mapMaybe, uncons, (!!), last, length, replicate)
+import Data.Foldable (for_)
import Data.String.Common (joinWith)
import Data.Tuple (Tuple(..))
import Data.Int (fromString)
@@ -34,6 +35,8 @@ import Metrics as Metrics
foreign import data Connection :: Type
+data FilterField = FilterArtist | FilterLabel | FilterYear | FilterGenre
+
foreign import connectImpl :: String -> (Nullable Error -> Nullable Connection -> Effect Unit) -> Effect Unit
foreign import runImpl :: Connection -> String -> Array Foreign -> (Nullable Error -> Effect Unit) -> Effect Unit
foreign import allImpl :: Connection -> String -> Array Foreign -> (Nullable Error -> Nullable (Array Json) -> Effect Unit) -> Effect Unit
@@ -149,51 +152,48 @@ getOldestTs conn = do
if ts == 0 then Nothing else Just ts
upsertScrobble :: Connection -> Listen -> Aff Unit
-upsertScrobble conn (Listen { listenedAt, trackMetadata: TrackMetadata track }) = do
- case listenedAt of
- Nothing -> pure unit
- Just ts -> do
- let
- mbid = fromMaybe (MbidMapping { releaseMbid: Nothing, caaReleaseMbid: Nothing }) track.mbidMapping
- MbidMapping m = mbid
- params =
- [ unsafeCoerce ts
- , unsafeCoerce (fromMaybe "" track.trackName)
- , unsafeCoerce (fromMaybe "" track.artistName)
- , unsafeCoerce (fromMaybe "" track.releaseName)
- , unsafeCoerce (fromMaybe "" m.releaseMbid)
- , unsafeCoerce (fromMaybe "" m.caaReleaseMbid)
- ]
- _ <- run conn "INSERT INTO scrobbles SELECT * FROM (SELECT ? as listened_at, ? as track_name, ? as artist_name, ? as release_name, ? as release_mbid, ? as caa_release_mbid) t WHERE NOT EXISTS (SELECT 1 FROM scrobbles WHERE listened_at = t.listened_at)" params
- pure unit
+upsertScrobble conn (Listen { listenedAt, trackMetadata: TrackMetadata track }) =
+ for_ listenedAt \ts -> do
+ let
+ mbid = fromMaybe (MbidMapping { releaseMbid: Nothing, caaReleaseMbid: Nothing }) track.mbidMapping
+ MbidMapping m = mbid
+ params =
+ [ unsafeCoerce ts
+ , unsafeCoerce (fromMaybe "" track.trackName)
+ , unsafeCoerce (fromMaybe "" track.artistName)
+ , unsafeCoerce (fromMaybe "" track.releaseName)
+ , unsafeCoerce (fromMaybe "" m.releaseMbid)
+ , unsafeCoerce (fromMaybe "" m.caaReleaseMbid)
+ ]
+ run conn "INSERT INTO scrobbles SELECT * FROM (SELECT ? as listened_at, ? as track_name, ? as artist_name, ? as release_name, ? as release_mbid, ? as caa_release_mbid) t WHERE NOT EXISTS (SELECT 1 FROM scrobbles WHERE listened_at = t.listened_at)" params
-getScrobbles :: Connection -> Int -> Int -> Maybe { field :: String, value :: String } -> Aff (Array Listen)
+getScrobbles :: Connection -> Int -> Int -> Maybe { field :: FilterField, value :: String } -> Aff (Array Listen)
getScrobbles conn limit offset Nothing = do
rows <- queryAll conn
"SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre FROM scrobbles s LEFT JOIN release_metadata rm ON s.release_mbid = rm.release_mbid ORDER BY s.listened_at DESC LIMIT ? OFFSET ?"
[ unsafeCoerce limit, unsafeCoerce offset ]
pure $ mapMaybe rowToListen rows
getScrobbles conn limit offset (Just { field, value }) = do
- rows <- queryAll conn query [ unsafeCoerce value, unsafeCoerce limit, unsafeCoerce offset ]
+ rows <- queryAll conn (filterQuery field) [ unsafeCoerce value, unsafeCoerce limit, unsafeCoerce offset ]
pure $ mapMaybe rowToListen rows
- where
- query = case field of
- "artist" ->
- "SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre"
- <> " FROM scrobbles s LEFT JOIN release_metadata rm ON s.release_mbid = rm.release_mbid"
- <> " WHERE s.artist_name = ? ORDER BY s.listened_at DESC LIMIT ? OFFSET ?"
- _ ->
- let
- col = case field of
- "label" -> "rm.label"
- "year" -> "rm.release_year::VARCHAR"
- _ -> "rm.genre"
- in
- "SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre"
- <> " FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid"
- <> " WHERE "
- <> col
- <> " = ? ORDER BY s.listened_at DESC LIMIT ? OFFSET ?"
+
+filterQuery :: FilterField -> String
+filterQuery FilterArtist =
+ "SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre"
+ <> " FROM scrobbles s LEFT JOIN release_metadata rm ON s.release_mbid = rm.release_mbid"
+ <> " WHERE s.artist_name = ? ORDER BY s.listened_at DESC LIMIT ? OFFSET ?"
+filterQuery FilterLabel =
+ "SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre"
+ <> " FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid"
+ <> " WHERE rm.label = ? ORDER BY s.listened_at DESC LIMIT ? OFFSET ?"
+filterQuery FilterYear =
+ "SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre"
+ <> " FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid"
+ <> " WHERE rm.release_year::VARCHAR = ? ORDER BY s.listened_at DESC LIMIT ? OFFSET ?"
+filterQuery FilterGenre =
+ "SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre"
+ <> " FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid"
+ <> " WHERE rm.genre = ? ORDER BY s.listened_at DESC LIMIT ? OFFSET ?"
initReleaseMetadata :: Connection -> Aff Unit
initReleaseMetadata conn = do
diff --git a/src/Main.purs b/src/Main.purs
@@ -38,9 +38,9 @@ import Data.Argonaut (decodeJson, encodeJson, parseJson)
import Data.Argonaut.Core (Json, toObject, toArray, toString, stringify)
import Data.Array ((!!), length, uncons, mapMaybe, find)
import Data.Tuple (Tuple(..))
-import Data.Foldable (for_)
+import Data.Foldable (for_, foldM)
import Data.Traversable (traverse)
-import Db (Connection, connect, initDb, upsertScrobble, getScrobbles, checkExists, getOldestTs, initReleaseMetadata, getUnenrichedMbids, getEmptyGenreMbids, getArtistReleasesByMbids, upsertReleaseMetadata, touchGenreCheckedAt, getStats, ping, backupDb, withTransaction)
+import Db (Connection, FilterField(..), connect, initDb, upsertScrobble, getScrobbles, checkExists, getOldestTs, initReleaseMetadata, getUnenrichedMbids, getEmptyGenreMbids, getArtistReleasesByMbids, upsertReleaseMetadata, touchGenreCheckedAt, getStats, ping, backupDb, withTransaction)
import Types (Listen(..), ListenBrainzResponse(..), MbidMapping(..), Payload(..), TrackMetadata(..))
import Control.Monad.Rec.Class (forever)
import Data.Time.Duration (Milliseconds(..))
@@ -221,20 +221,18 @@ lbSyncOnce conn username slug writeLock initialSyncEnabled = void performFullSyn
liftEffect $ Metrics.setSyncLastSuccess slug "listenbrainz"
processListens listens = do
- syncRecursive 0 Nothing 0 listens
-
- syncRecursive acc minTs hitCount listens = case uncons listens of
- Nothing -> pure $ Tuple acc (Tuple minTs hitCount)
- Just { head: l@(Listen { listenedAt: Just ts, trackMetadata: (TrackMetadata _) }), tail } -> do
+ s <- foldM step { added: 0, minTs: Nothing, hitCount: 0 } listens
+ pure $ Tuple s.added (Tuple s.minTs s.hitCount)
+ where
+ step s l@(Listen { listenedAt: Just ts }) = do
exists <- checkExists conn ts
- if exists then do
- syncRecursive acc (Just ts) (hitCount + 1) tail
+ if exists then pure s { minTs = Just ts, hitCount = s.hitCount + 1 }
else do
upsertScrobble conn l
- syncRecursive (acc + 1) (Just ts) hitCount tail
- Just { head: _, tail } -> do
+ pure s { added = s.added + 1, minTs = Just ts }
+ step s _ = do
Log.warn "Skipping scrobble without timestamp"
- syncRecursive acc minTs hitCount tail
+ pure s
lbSyncLoop :: Connection -> String -> String -> AVar Unit -> Boolean -> Aff Unit
lbSyncLoop conn username slug writeLock initialSyncEnabled = forever do
@@ -299,17 +297,17 @@ lfSyncOnce conn apiKey lfmUser slug writeLock initialSyncEnabled = do
when (n > 0) $ liftEffect $ Metrics.incSyncScrobbles slug "lastfm" n
liftEffect $ Metrics.setSyncLastSuccess slug "lastfm"
- processLastfmTracks tracks = syncLastfmRecursive 0 0 (mapMaybe lastfmTrackToListen tracks)
-
- syncLastfmRecursive acc hitCount listens = case uncons listens of
- Nothing -> pure $ Tuple acc hitCount
- Just { head: l@(Listen { listenedAt: Just ts }), tail } -> do
+ processLastfmTracks tracks = do
+ s <- foldM step { added: 0, hitCount: 0 } (mapMaybe lastfmTrackToListen tracks)
+ pure $ Tuple s.added s.hitCount
+ where
+ step s l@(Listen { listenedAt: Just ts }) = do
exists <- checkExists conn ts
- if exists then syncLastfmRecursive acc (hitCount + 1) tail
+ if exists then pure s { hitCount = s.hitCount + 1 }
else do
upsertScrobble conn l
- syncLastfmRecursive (acc + 1) hitCount tail
- Just { head: _, tail } -> syncLastfmRecursive acc hitCount tail
+ pure s { added = s.added + 1 }
+ step s _ = pure s
lfSyncLoop :: Connection -> String -> String -> String -> AVar Unit -> Boolean -> Aff Unit
lfSyncLoop conn apiKey lfmUser slug writeLock initialSyncEnabled = forever do
@@ -400,45 +398,130 @@ serveClientJs res = do
getQueryParam :: String -> URL -> Maybe String
getQueryParam name url = URLSearchParams.get name (URL.searchParams url)
+parseFilterField :: String -> Maybe FilterField
+parseFilterField "artist" = Just FilterArtist
+parseFilterField "label" = Just FilterLabel
+parseFilterField "year" = Just FilterYear
+parseFilterField "genre" = Just FilterGenre
+parseFilterField _ = Nothing
+
sanitizeKey :: String -> String
-sanitizeKey str =
+sanitizeKey = replace re1 "_" >>> replace re2 "_"
+ where
+ re1 = unsafeRegex "[^a-z0-9.-]" (parseFlags "gi")
+ re2 = unsafeRegex "_{2,}" (parseFlags "g")
+
+-- Fetch the Last.fm image URL for an artist/release (no caching, no serving).
+fetchLastfmCoverUrl :: UserConfig -> String -> String -> Aff (Maybe String)
+fetchLastfmCoverUrl cfg artist release = case cfg.lastfmApiKey of
+ Nothing -> pure Nothing
+ Just k -> do
+ let
+ searchUrl = "https://ws.audioscrobbler.com/2.0/?method=album.getinfo&artist="
+ <> (fromMaybe "" $ encodeURIComponent artist)
+ <> "&album="
+ <> (fromMaybe "" $ encodeURIComponent release)
+ <> "&format=json&api_key="
+ <> k
+ Log.info $ "Searching Last.fm for: " <> artist <> " - " <> release
+ result <- try $ fetch searchUrl { method: GET }
+ case result of
+ Right fr | fr.status == 200 -> do
+ json <- fromJson fr.json
+ pure $ do
+ obj <- toObject json
+ album <- Object.lookup "album" obj >>= toObject
+ images <- Object.lookup "image" album >>= toArray
+ imgObj <- images !! 2 >>= toObject
+ u <- Object.lookup "#text" imgObj >>= toString
+ if u == "" then Nothing else Just u
+ _ -> pure Nothing
+
+-- Fetch the Discogs image URL for an artist/release (no caching, no serving).
+fetchDiscogsCoverUrl :: UserConfig -> String -> String -> Aff (Maybe String)
+fetchDiscogsCoverUrl cfg artist release = case cfg.discogsToken of
+ Nothing -> pure Nothing
+ Just t -> do
+ let
+ queryStr = artist <> " " <> release
+ searchUrl = "https://api.discogs.com/database/search?q="
+ <> (fromMaybe "" $ encodeURIComponent queryStr)
+ <> "&type=release&per_page=1"
+ Log.info $ "Searching Discogs for: " <> queryStr
+ result <- try $ fetch searchUrl { method: GET, headers: { "User-Agent": "ScrobblerPureScript/1.0", "Authorization": "Discogs token=" <> t } }
+ case result of
+ Right fr | fr.status == 200 -> do
+ json <- fromJson fr.json
+ pure $ do
+ obj <- toObject json
+ results <- Object.lookup "results" obj >>= toArray
+ firstResult <- results !! 0 >>= toObject
+ Object.lookup "cover_image" firstResult >>= toString
+ _ -> pure Nothing
+
+type CoverSource =
+ { name :: String
+ , s3Key :: String
+ , findUrl :: Aff (Maybe String)
+ }
+
+coverSources :: String -> String -> String -> UserConfig -> Array CoverSource
+coverSources mbid artist release cfg =
let
- re1 = unsafeRegex "[^a-z0-9.-]" (parseFlags "gi")
- re2 = unsafeRegex "_{2,}" (parseFlags "g")
+ safeArtist = sanitizeKey artist
+ safeRelease = sanitizeKey release
in
- replace re2 "_" (replace re1 "_" str)
+ [ { name: "caa"
+ , s3Key: "covers/caa/" <> sanitizeKey mbid <> ".jpg"
+ , findUrl:
+ if mbid == "" then pure Nothing
+ else pure $ Just $ "https://coverartarchive.org/release/" <> mbid <> "/front-250"
+ }
+ , { name: "lastfm"
+ , s3Key: "covers/lastfm/" <> safeArtist <> "-" <> safeRelease <> ".jpg"
+ , findUrl:
+ if artist == "" || release == "" then pure Nothing
+ else fetchLastfmCoverUrl cfg artist release
+ }
+ , { name: "discogs"
+ , s3Key: "covers/discogs/" <> safeArtist <> "-" <> safeRelease <> ".jpg"
+ , findUrl:
+ if artist == "" || release == "" then pure Nothing
+ else fetchDiscogsCoverUrl cfg artist release
+ }
+ ]
serveCover :: UserConfig -> String -> URL -> Response -> Effect Unit
-serveCover cfg slug url res = do
- launchAff_ do
- let mbid = fromMaybe "" (getQueryParam "mbid" url)
- let artistStr = fromMaybe "" (getQueryParam "artist" url)
- let releaseStr = fromMaybe "" (getQueryParam "release" url)
- let s3cfg = s3ConfigFromUser cfg
-
- if mbid /= "" then do
- let safeMbid = sanitizeKey mbid
- let s3Key = "covers/caa/" <> safeMbid <> ".jpg"
- cached <- checkS3 s3cfg s3Key
- if cached then do
- Log.info $ "Serving CAA cover from S3: " <> s3Key
- liftEffect $ Metrics.incCoverRequest slug "caa" "s3_hit"
- serveS3 s3cfg s3Key res
- else do
- Log.info $ "Fetching CAA cover: " <> mbid
- let caaUrl = "https://coverartarchive.org/release/" <> mbid <> "/front-250"
- success <- tryProxyAndCache s3cfg caaUrl s3Key res
- if success then
- liftEffect $ Metrics.incCoverRequest slug "caa" "fetch"
- else do
- liftEffect $ Metrics.incCoverRequest slug "caa" "miss"
- Log.info $ "CAA cover not found for " <> mbid <> ", falling back to Last.fm"
- tryLastfm s3cfg artistStr releaseStr res
- else do
- Log.info $ "No MBID provided, trying Last.fm for: " <> artistStr <> " - " <> releaseStr
- tryLastfm s3cfg artistStr releaseStr res
+serveCover cfg slug url res = launchAff_ do
+ let
+ mbid = fromMaybe "" (getQueryParam "mbid" url)
+ artist = fromMaybe "" (getQueryParam "artist" url)
+ release = fromMaybe "" (getQueryParam "release" url)
+ s3cfg = s3ConfigFromUser cfg
+
+ served <- foldM (trySource s3cfg) false (coverSources mbid artist release cfg)
+ unless served $ liftEffect $ serveNotFound res
where
+ trySource _ true _ = pure true
+ trySource s3cfg false { name, s3Key, findUrl } = do
+ cached <- checkS3 s3cfg s3Key
+ if cached then do
+ Log.info $ "Serving " <> name <> " cover from S3: " <> s3Key
+ liftEffect $ Metrics.incCoverRequest slug name "s3_hit"
+ serveS3 s3cfg s3Key res
+ pure true
+ else do
+ mUrl <- findUrl
+ case mUrl of
+ Nothing -> pure false
+ Just urlStr -> do
+ success <- tryProxyAndCache s3cfg urlStr s3Key res
+ liftEffect $
+ if success then Metrics.incCoverRequest slug name "fetch"
+ else Metrics.incCoverRequest slug name "miss"
+ pure success
+
checkS3 s3cfg s3Key
| not cfg.coverCacheEnabled = pure false
| otherwise = do
@@ -467,7 +550,6 @@ serveCover cfg slug url res = do
nativeBuf <- fromArrayBuffer buf
void $ write writer nativeBuf
end writer
-
when cfg.coverCacheEnabled $ void $ forkAff $ do
uploadResult <- try $ uploadToS3 s3cfg s3Key (unsafeCoerce buf) contentType
case uploadResult of
@@ -476,103 +558,6 @@ serveCover cfg slug url res = do
pure true
_ -> pure false
- tryLastfm s3cfg artist release response
- | artist == "" || release == "" = do
- Log.warn "Missing artist or release for Last.fm fallback"
- liftEffect $ serveNotFound response
- | otherwise = do
- let safeArtist = sanitizeKey artist
- let safeRelease = sanitizeKey release
- let s3Key = "covers/lastfm/" <> safeArtist <> "-" <> safeRelease <> ".jpg"
- cached <- checkS3 s3cfg s3Key
- if cached then do
- Log.info $ "Serving Last.fm cover from S3: " <> s3Key
- liftEffect $ Metrics.incCoverRequest slug "lastfm" "s3_hit"
- serveS3 s3cfg s3Key response
- else case cfg.lastfmApiKey of
- Nothing -> do
- Log.warn "lastfmApiKey not configured, falling back to Discogs"
- tryDiscogs s3cfg artist release response
- Just k -> do
- let searchUrl = "https://ws.audioscrobbler.com/2.0/?method=album.getinfo&artist=" <> (fromMaybe "" $ encodeURIComponent artist) <> "&album=" <> (fromMaybe "" $ encodeURIComponent release) <> "&format=json" <> "&api_key=" <> k
- Log.info $ "Searching Last.fm for: " <> artist <> " - " <> release
- result <- try $ fetch searchUrl { method: GET }
- case result of
- Right fetchRes | fetchRes.status == 200 -> do
- json <- fromJson fetchRes.json
- let
- coverUrl = do
- obj <- toObject json
- album <- Object.lookup "album" obj >>= toObject
- images <- Object.lookup "image" album >>= toArray
- imageObj <- images !! 2 >>= toObject
- u <- Object.lookup "#text" imageObj >>= toString
- if u == "" then Nothing else Just u
- case coverUrl of
- Just urlStr -> do
- Log.info $ "Found Last.fm cover: " <> urlStr
- success <- tryProxyAndCache s3cfg urlStr s3Key response
- if success then
- liftEffect $ Metrics.incCoverRequest slug "lastfm" "fetch"
- else do
- liftEffect $ Metrics.incCoverRequest slug "lastfm" "miss"
- Log.info "Last.fm image proxy failed, falling back to Discogs"
- tryDiscogs s3cfg artist release response
- Nothing -> do
- Log.info "No cover found on Last.fm, falling back to Discogs"
- liftEffect $ Metrics.incCoverRequest slug "lastfm" "not_found"
- tryDiscogs s3cfg artist release response
- _ -> do
- Log.info "Last.fm API request failed, falling back to Discogs"
- liftEffect $ Metrics.incCoverRequest slug "lastfm" "error"
- tryDiscogs s3cfg artist release response
-
- tryDiscogs s3cfg artist release response = do
- let safeArtist = sanitizeKey artist
- let safeRelease = sanitizeKey release
- let s3Key = "covers/discogs/" <> safeArtist <> "-" <> safeRelease <> ".jpg"
- cached <- checkS3 s3cfg s3Key
- if cached then do
- Log.info $ "Serving Discogs cover from S3: " <> s3Key
- liftEffect $ Metrics.incCoverRequest slug "discogs" "s3_hit"
- serveS3 s3cfg s3Key response
- else case cfg.discogsToken of
- Nothing -> do
- Log.warn "discogsToken not configured, cannot fallback further"
- liftEffect $ serveNotFound response
- Just t -> do
- let queryStr = artist <> " " <> release
- let searchUrl = "https://api.discogs.com/database/search?q=" <> (fromMaybe "" $ encodeURIComponent queryStr) <> "&type=release&per_page=1"
- Log.info $ "Searching Discogs for: " <> queryStr
- result <- try $ fetch searchUrl { method: GET, headers: { "User-Agent": "ScrobblerPureScript/1.0", "Authorization": "Discogs token=" <> t } }
- case result of
- Right fetchRes | fetchRes.status == 200 -> do
- json <- fromJson fetchRes.json
- let
- coverUrl = do
- obj <- toObject json
- results <- Object.lookup "results" obj >>= toArray
- firstResult <- results !! 0 >>= toObject
- Object.lookup "cover_image" firstResult >>= toString
- case coverUrl of
- Just urlStr -> do
- Log.info $ "Found Discogs cover: " <> urlStr
- success <- tryProxyAndCache s3cfg urlStr s3Key response
- if success then
- liftEffect $ Metrics.incCoverRequest slug "discogs" "fetch"
- else do
- liftEffect $ Metrics.incCoverRequest slug "discogs" "miss"
- Log.info "Discogs image proxy failed"
- liftEffect $ serveNotFound response
- Nothing -> do
- Log.info "No cover found on Discogs"
- liftEffect $ Metrics.incCoverRequest slug "discogs" "not_found"
- liftEffect $ serveNotFound response
- _ -> do
- Log.info "Discogs API request failed"
- liftEffect $ Metrics.incCoverRequest slug "discogs" "error"
- liftEffect $ serveNotFound response
-
serveProxy :: Connection -> URL -> Response -> Effect Unit
serveProxy db url res = do
setHeader "Content-Type" "application/json" (toOutgoingMessage res)
@@ -582,7 +567,7 @@ serveProxy db url res = do
let offset = fromMaybe 0 (getQueryParam "offset" url >>= fromString)
let
mFilter = do
- field <- getQueryParam "filterField" url
+ field <- getQueryParam "filterField" url >>= parseFilterField
value <- getQueryParam "filterValue" url
pure { field, value }
@@ -760,6 +745,27 @@ fetchDiscogsGenre (Just t) artist release = do
Log.warn "Discogs genre API request failed"
pure Nothing
+type GenreSource =
+ { name :: String
+ , enabled :: Boolean
+ , fetch :: Aff (Maybe String)
+ }
+
+-- Try each source in order, stopping at the first that returns Just.
+-- Records found/not_found metrics only for sources that are enabled and queried.
+fetchFallbackGenre :: String -> Array GenreSource -> Aff (Maybe String)
+fetchFallbackGenre slug = foldM trySource Nothing
+ where
+ trySource (Just g) _ = pure (Just g)
+ trySource Nothing { name, enabled, fetch }
+ | not enabled = pure Nothing
+ | otherwise = do
+ result <- fetch
+ liftEffect $ case result of
+ Just _ -> Metrics.incEnrichmentFetch slug name "found"
+ Nothing -> Metrics.incEnrichmentFetch slug name "not_found"
+ pure result
+
enrichMetadata :: Connection -> UserConfig -> String -> Aff Unit
enrichMetadata conn cfg slug = forever do
unenrichedMbids <- getUnenrichedMbids conn 10
@@ -787,37 +793,29 @@ enrichMetadata conn cfg slug = forever do
touchGenreCheckedAt conn mbid
Right (Just mbdata) -> do
liftEffect $ Metrics.incEnrichmentFetch slug "musicbrainz" "success"
- if mbdata.genre == Nothing then do
- let artistRelease = Object.lookup mbid artistReleaseMap
- case artistRelease of
- Just { artist, release } -> do
- lastfmGenre <- fetchLastfmGenre cfg.lastfmApiKey artist release
- when (isJust cfg.lastfmApiKey) $ liftEffect $ case lastfmGenre of
- Just _ -> Metrics.incEnrichmentFetch slug "lastfm" "found"
- Nothing -> Metrics.incEnrichmentFetch slug "lastfm" "not_found"
- finalGenre <- case lastfmGenre of
- Just _ -> pure lastfmGenre
- Nothing -> do
- g <- fetchDiscogsGenre cfg.discogsToken artist release
- when (isJust cfg.discogsToken) $ liftEffect $ case g of
- Just _ -> Metrics.incEnrichmentFetch slug "discogs" "found"
- Nothing -> Metrics.incEnrichmentFetch slug "discogs" "not_found"
- pure g
-
- let finalMbdata = mbdata { genre = finalGenre }
- upsertReleaseMetadata conn mbid finalMbdata.genre finalMbdata.label finalMbdata.year
-
- case finalGenre of
- Just genre -> Log.info $ "Added fallback genre from " <> (if lastfmGenre /= Nothing then "Last.fm" else "Discogs") <> " for " <> mbid <> ": " <> genre
- Nothing -> do
- Log.info $ "No genre found in any source for " <> mbid
- touchGenreCheckedAt conn mbid
- Nothing -> do
- Log.warn $ "No artist/release info found for MBID " <> mbid <> ", cannot use fallback sources"
- upsertReleaseMetadata conn mbid mbdata.genre mbdata.label mbdata.year
- touchGenreCheckedAt conn mbid
- else do
- upsertReleaseMetadata conn mbid mbdata.genre mbdata.label mbdata.year
+ case mbdata.genre of
+ Just _ ->
+ upsertReleaseMetadata conn mbid mbdata.genre mbdata.label mbdata.year
+ Nothing -> do
+ let artistRelease = Object.lookup mbid artistReleaseMap
+ case artistRelease of
+ Just { artist, release } -> do
+ let
+ sources =
+ [ { name: "lastfm", enabled: isJust cfg.lastfmApiKey, fetch: fetchLastfmGenre cfg.lastfmApiKey artist release }
+ , { name: "discogs", enabled: isJust cfg.discogsToken, fetch: fetchDiscogsGenre cfg.discogsToken artist release }
+ ]
+ finalGenre <- fetchFallbackGenre slug sources
+ upsertReleaseMetadata conn mbid finalGenre mbdata.label mbdata.year
+ case finalGenre of
+ Just genre -> Log.info $ "Added fallback genre for " <> mbid <> ": " <> genre
+ Nothing -> do
+ Log.info $ "No genre found in any source for " <> mbid
+ touchGenreCheckedAt conn mbid
+ Nothing -> do
+ Log.warn $ "No artist/release info found for MBID " <> mbid <> ", cannot use fallback sources"
+ upsertReleaseMetadata conn mbid mbdata.genre mbdata.label mbdata.year
+ touchGenreCheckedAt conn mbid
-- Initialise one user: connect DB, start sync loops, return a UserContext.
startUser :: UserEntry -> Aff UserContext
diff --git a/src/Metrics.js b/src/Metrics.js
@@ -21,32 +21,18 @@ import {
// Prometheus metrics — only active when METRICS_ENABLED=true
// ---------------------------------------------------------------------------
-const metricsEnabled = process.env.METRICS_ENABLED === "true";
-
-let registry = null;
-let httpRequestsTotal = null;
-let httpRequestDurationSeconds = null;
-let syncRunsTotal = null;
-let syncScrobblesAddedTotal = null;
-let syncLastSuccessSeconds = null;
-let enrichmentFetchesTotal = null;
-let enrichmentQueueSize = null;
-let coverRequestsTotal = null;
-let dbBackupRunsTotal = null;
-let dbBackupLastSuccessSeconds = null;
-
-if (metricsEnabled) {
- registry = new Registry();
+const makeActiveMetrics = () => {
+ const registry = new Registry();
collectDefaultMetrics({ register: registry });
- httpRequestsTotal = new Counter({
+ const httpRequestsTotal = new Counter({
name: "corpus_http_requests_total",
help: "Total number of HTTP requests",
labelNames: ["method", "path", "status"],
registers: [registry],
});
- httpRequestDurationSeconds = new Histogram({
+ const httpRequestDurationSeconds = new Histogram({
name: "corpus_http_request_duration_seconds",
help: "HTTP request duration in seconds",
labelNames: ["method", "path"],
@@ -54,70 +40,105 @@ if (metricsEnabled) {
registers: [registry],
});
- syncRunsTotal = new Counter({
+ const syncRunsTotal = new Counter({
name: "corpus_sync_runs_total",
help: "Total number of sync runs",
labelNames: ["user", "source", "result"],
registers: [registry],
});
- syncScrobblesAddedTotal = new Counter({
+ const syncScrobblesAddedTotal = new Counter({
name: "corpus_sync_scrobbles_added_total",
help: "Total scrobbles added during syncs",
labelNames: ["user", "source"],
registers: [registry],
});
- syncLastSuccessSeconds = new Gauge({
+ const syncLastSuccessSeconds = new Gauge({
name: "corpus_sync_last_success_seconds",
help: "Unix timestamp of last successful sync run",
labelNames: ["user", "source"],
registers: [registry],
});
- enrichmentFetchesTotal = new Counter({
+ const enrichmentFetchesTotal = new Counter({
name: "corpus_enrichment_fetches_total",
help: "Total metadata enrichment fetches by user, source, and result",
labelNames: ["user", "source", "result"],
registers: [registry],
});
- enrichmentQueueSize = new Gauge({
+ const enrichmentQueueSize = new Gauge({
name: "corpus_enrichment_queue_size",
help: "Number of releases pending metadata enrichment per user",
labelNames: ["user", "type"],
registers: [registry],
});
- coverRequestsTotal = new Counter({
+ const coverRequestsTotal = new Counter({
name: "corpus_cover_requests_total",
help: "Total cover art requests by user, source, and result",
labelNames: ["user", "source", "result"],
registers: [registry],
});
- dbBackupRunsTotal = new Counter({
+ const dbBackupRunsTotal = new Counter({
name: "corpus_db_backup_runs_total",
help: "Total database backup runs",
labelNames: ["user", "result"],
registers: [registry],
});
- dbBackupLastSuccessSeconds = new Gauge({
+ const dbBackupLastSuccessSeconds = new Gauge({
name: "corpus_db_backup_last_success_seconds",
help: "Unix timestamp of last successful database backup",
labelNames: ["user"],
registers: [registry],
});
-}
+
+ return {
+ getMetrics: () => registry.metrics(),
+ contentType: registry.contentType,
+ recordHttpRequest: (method, path, status, durationMs) => {
+ httpRequestsTotal.inc({ method, path, status: String(status) });
+ httpRequestDurationSeconds.observe({ method, path }, durationMs / 1000);
+ },
+ incSyncRuns: (user, source, result) => syncRunsTotal.inc({ user, source, result }),
+ incSyncScrobbles: (user, source, count) => syncScrobblesAddedTotal.inc({ user, source }, count),
+ setSyncLastSuccess: (user, source) => syncLastSuccessSeconds.setToCurrentTime({ user, source }),
+ incEnrichmentFetch: (user, source, result) => enrichmentFetchesTotal.inc({ user, source, result }),
+ setEnrichmentQueueSize: (user, type, size) => enrichmentQueueSize.set({ user, type }, size),
+ incCoverRequest: (user, source, result) => coverRequestsTotal.inc({ user, source, result }),
+ incDbBackupRun: (user, result) => dbBackupRunsTotal.inc({ user, result }),
+ setDbBackupLastSuccess: (user) => dbBackupLastSuccessSeconds.setToCurrentTime({ user }),
+ };
+};
+
+const noOp = () => {};
+
+const makeNoOpMetrics = () => ({
+ getMetrics: () => Promise.resolve(""),
+ contentType: "text/plain; version=0.0.4; charset=utf-8",
+ recordHttpRequest: noOp,
+ incSyncRuns: noOp,
+ incSyncScrobbles: noOp,
+ setSyncLastSuccess: noOp,
+ incEnrichmentFetch: noOp,
+ setEnrichmentQueueSize: noOp,
+ incCoverRequest: noOp,
+ incDbBackupRun: noOp,
+ setDbBackupLastSuccess: noOp,
+});
+
+const activeMetrics =
+ process.env.METRICS_ENABLED === "true" ? makeActiveMetrics() : makeNoOpMetrics();
// ---------------------------------------------------------------------------
// OpenTelemetry — only active when OTEL_EXPORTER_OTLP_ENDPOINT is set
// ---------------------------------------------------------------------------
-let tracer = null;
-
-if (process.env.OTEL_EXPORTER_OTLP_ENDPOINT) {
+const makeTracer = () => {
+ if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT) return null;
const sdk = new NodeSDK({
traceExporter: new OTLPTraceExporter(),
instrumentations: [
@@ -129,14 +150,17 @@ if (process.env.OTEL_EXPORTER_OTLP_ENDPOINT) {
],
});
sdk.start();
- tracer = trace.getTracer(
+ const tracer = trace.getTracer(
process.env.OTEL_SERVICE_NAME || "corpus",
process.env.npm_package_version,
);
process.on("SIGTERM", () => sdk.shutdown().finally(() => process.exit(0)));
-}
+ return tracer;
+};
+
+const tracer = makeTracer();
-function startHttpSpan(method, path, headers) {
+const startHttpSpan = (method, path, headers) => {
if (!tracer) return null;
const parentCtx = propagation.extract(context.active(), headers);
return tracer.startSpan(
@@ -150,34 +174,29 @@ function startHttpSpan(method, path, headers) {
},
parentCtx,
);
-}
+};
-function endHttpSpan(span, statusCode) {
+const endHttpSpan = (span, statusCode) => {
if (!span) return;
span.setAttribute("http.status_code", statusCode);
span.setStatus({
code: statusCode >= 500 ? SpanStatusCode.ERROR : SpanStatusCode.OK,
});
span.end();
-}
+};
// ---------------------------------------------------------------------------
// Exports
// ---------------------------------------------------------------------------
export const getMetricsImpl = (onSuccess) => (onError) => () => {
- if (!registry) {
- onSuccess("")();
- return;
- }
- registry.metrics().then(
+ activeMetrics.getMetrics().then(
(s) => onSuccess(s)(),
(err) => onError(err.message)(),
);
};
-export const getContentType = () =>
- registry ? registry.contentType : "text/plain; version=0.0.4; charset=utf-8";
+export const getContentType = () => activeMetrics.contentType;
// Runs `handler` inside the active context of a server span so that any
// outbound HTTP/fetch calls made during handling automatically become child
@@ -191,8 +210,7 @@ export const wrapRequest =
res.once("finish", () => {
const durationMs = Date.now() - startMs;
const status = res.statusCode || 0;
- if (httpRequestsTotal) httpRequestsTotal.inc({ method, path, status: String(status) });
- if (httpRequestDurationSeconds) httpRequestDurationSeconds.observe({ method, path }, durationMs / 1000);
+ activeMetrics.recordHttpRequest(method, path, status, durationMs);
endHttpSpan(span, status);
logFn(`${method} ${path} ${status} ${durationMs}ms`)();
});
@@ -205,37 +223,33 @@ export const wrapRequest =
};
export const incSyncRuns = (user) => (source) => (result) => () => {
- if (syncRunsTotal) syncRunsTotal.inc({ user, source, result });
+ activeMetrics.incSyncRuns(user, source, result);
};
export const incSyncScrobbles = (user) => (source) => (count) => () => {
- if (syncScrobblesAddedTotal)
- syncScrobblesAddedTotal.inc({ user, source }, count);
+ activeMetrics.incSyncScrobbles(user, source, count);
};
export const setSyncLastSuccess = (user) => (source) => () => {
- if (syncLastSuccessSeconds)
- syncLastSuccessSeconds.setToCurrentTime({ user, source });
+ activeMetrics.setSyncLastSuccess(user, source);
};
export const incEnrichmentFetch = (user) => (source) => (result) => () => {
- if (enrichmentFetchesTotal)
- enrichmentFetchesTotal.inc({ user, source, result });
+ activeMetrics.incEnrichmentFetch(user, source, result);
};
export const setEnrichmentQueueSize = (user) => (type) => (size) => () => {
- if (enrichmentQueueSize) enrichmentQueueSize.set({ user, type }, size);
+ activeMetrics.setEnrichmentQueueSize(user, type, size);
};
export const incCoverRequest = (user) => (source) => (result) => () => {
- if (coverRequestsTotal) coverRequestsTotal.inc({ user, source, result });
+ activeMetrics.incCoverRequest(user, source, result);
};
export const incDbBackupRun = (user) => (result) => () => {
- if (dbBackupRunsTotal) dbBackupRunsTotal.inc({ user, result });
+ activeMetrics.incDbBackupRun(user, result);
};
export const setDbBackupLastSuccess = (user) => () => {
- if (dbBackupLastSuccessSeconds)
- dbBackupLastSuccessSeconds.setToCurrentTime({ user });
+ activeMetrics.setDbBackupLastSuccess(user);
};
diff --git a/src/S3.js b/src/S3.js
@@ -31,15 +31,7 @@ export const uploadToS3Impl =
client
.send(command)
.then(() => cb(null)())
- .catch((err) => {
- const msg =
- `S3: Upload failed for ${key}: ${err.message || err}`.replace(
- /\n/g,
- " ",
- );
- console.error(msg);
- cb(err)();
- });
+ .catch((err) => cb(err)());
};
export const existsInS3Impl = (cfg) => (key) => (cb) => () => {
@@ -51,18 +43,12 @@ export const existsInS3Impl = (cfg) => (key) => (cb) => () => {
client
.send(command)
- .then(() => cb(true)())
+ .then(() => cb(null)(true)())
.catch((err) => {
if (err.name === "NotFound" || err.$metadata?.httpStatusCode === 404) {
- cb(false)();
+ cb(null)(false)();
} else {
- const msg =
- `S3: exists check failed for ${key}: ${err.message || err}`.replace(
- /\n/g,
- " ",
- );
- console.error(msg);
- cb(false)();
+ cb(err)(false)();
}
});
};
diff --git a/src/S3.purs b/src/S3.purs
@@ -34,7 +34,7 @@ foreign import uploadToS3Impl
:: S3ConfigJs -> String -> Buffer -> String -> (Nullable Error -> Effect Unit) -> Effect Unit
foreign import existsInS3Impl
- :: S3ConfigJs -> String -> (Boolean -> Effect Unit) -> Effect Unit
+ :: S3ConfigJs -> String -> (Nullable Error -> Boolean -> Effect Unit) -> Effect Unit
foreign import getS3UrlImpl :: S3ConfigJs -> String -> String
@@ -48,7 +48,10 @@ uploadToS3 cfg key body contentType = makeAff \cb -> do
existsInS3 :: S3Config -> String -> Aff Boolean
existsInS3 cfg key = makeAff \cb -> do
- existsInS3Impl (toJs cfg) key \exists -> cb (Right exists)
+ existsInS3Impl (toJs cfg) key \err exists ->
+ case toMaybe err of
+ Just e -> cb (Left e)
+ Nothing -> cb (Right exists)
pure nonCanceler
getS3Url :: S3Config -> String -> String
diff --git a/test/Main.purs b/test/Main.purs
@@ -12,7 +12,7 @@ import Test.Spec.Assertions (shouldEqual, fail)
import Test.Spec.Reporter.Console (consoleReporter)
import Test.Spec.Runner.Node (runSpecAndExitProcess)
import Types (Listen(..), ListenBrainzResponse(..), MbidMapping(..), Payload(..), Stats(..), StatsEntry(..), TrackMetadata(..))
-import Db (connect, initDb, checkExists, upsertScrobble, getScrobbles, initReleaseMetadata, upsertReleaseMetadata, getStats, dbBaseName)
+import Db (FilterField(..), connect, initDb, checkExists, upsertScrobble, getScrobbles, initReleaseMetadata, upsertReleaseMetadata, getStats, dbBaseName)
import Data.Argonaut.Core (Json)
import Main (sanitizeKey, listenBrainzUrl, lastfmTrackToListen)
import S3 (getS3Url)
@@ -146,10 +146,10 @@ main = runSpecAndExitProcess [consoleReporter] do
length s.tracks `shouldEqual` 1
-- Test Filtering (as mentioned in architecture.md)
- listensFiltered <- getScrobbles conn 10 0 (Just { field: "genre", value: "Rock" })
+ listensFiltered <- getScrobbles conn 10 0 (Just { field: FilterGenre, value: "Rock" })
length listensFiltered `shouldEqual` 1
- listensEmpty <- getScrobbles conn 10 0 (Just { field: "genre", value: "Jazz" })
+ listensEmpty <- getScrobbles conn 10 0 (Just { field: FilterGenre, value: "Jazz" })
length listensEmpty `shouldEqual` 0
describe "Corpus Backup" do
