corpus

self-hosted listenbrainz and last.fm proxy
Log | Files | Refs | README | LICENSE

Sync.purs (14303B)

      1 module Sync
      2   ( listenBrainzUrl
      3   , lastfmTrackToListen
      4   , parseLastfmResponse
      5   , fetchLastfmPage
      6   , lbSync
      7   , lbSyncLoop
      8   , lfSync
      9   , lfSyncLoop
     10   ) where
     11 
     12 import Prelude
     13 
     14 import Control.Monad.Rec.Class (forever)
     15 import Data.Argonaut (decodeJson, parseJson)
     16 import Data.Argonaut.Core (Json, stringify)
     17 import Data.Array (length, mapMaybe, null)
     18 import Data.Either (Either(..))
     19 import Data.Foldable (foldM, for_)
     20 import Data.Maybe (Maybe(..), fromMaybe)
     21 import Data.String (Pattern(..), stripPrefix)
     22 import Data.Time.Duration (Milliseconds(..))
     23 import Db (Connection, checkExists, getOldestTs, upsertScrobble, withTransaction)
     24 import Effect.Aff (Aff, delay, launchAff_, makeAff, nonCanceler, throwError, try)
     25 import Effect.Aff.AVar (AVar)
     26 import Effect.Aff.Retry (RetryStatus(..), exponentialBackoff, limitRetries, recovering)
     27 import Effect.Class (liftEffect)
     28 import Effect.Exception (error, message)
     29 import Effect.Uncurried (mkEffectFn1)
     30 import Fetch (fetch, Method(GET))
     31 import Fetch.Argonaut.Json (fromJson)
     32 import JSURI (encodeURIComponent)
     33 import Log as Log
     34 import Metrics as Metrics
     35 import Node.EventEmitter (EventHandle(..), on_)
     36 import Node.HTTP.ClientRequest as Client
     37 import Node.HTTP.IncomingMessage as IM
     38 import Node.HTTPS as HTTPS
     39 import Node.Stream.Aff (readableToStringUtf8)
     40 import Types (Listen(..), ListenBrainzResponse(..), MbidMapping(..), Payload(..), TrackMetadata(..), LastfmResponse(..), LastfmRecentTracks(..), LastfmAttr(..), LastfmTracks(..), LastfmArtist(..), LastfmAlbum(..), LastfmDate(..), LastfmTrack(..))
     41 -- | Cast required due to a gap in the node-http FFI bindings:
     42 -- | the request object lacks an "error" event emitter type.
     43 import Unsafe.Coerce (unsafeCoerce)
     44 
     45 -- | Result of processing a batch of listens/scrobbles.
     46 type SyncResult =
     47   { added :: Int
     48   , hitCount :: Int
     49   , minTs :: Maybe Int
     50   }
     51 
     52 -- | Process a batch of listens: check existence, insert new ones, track stats.
     53 -- | When trackMinTs is true, the minimum timestamp is tracked (for cursor-based pagination).
     54 processTracks :: Connection -> Array Listen -> Boolean -> Aff SyncResult
     55 processTracks conn listens trackMinTs = do
     56   foldM step { added: 0, hitCount: 0, minTs: Nothing } listens
     57   where
     58   step :: SyncResult -> Listen -> Aff SyncResult
     59   step s l@(Listen { listenedAt: Just ts }) = do
     60     exists <- checkExists conn ts
     61     if exists then do
     62       let nextMinTs = if trackMinTs then Just ts else s.minTs
     63       pure $ s { added = s.added, hitCount = s.hitCount + 1, minTs = nextMinTs }
     64     else do
     65       upsertScrobble conn l
     66       let nextMinTs = if trackMinTs then Just ts else s.minTs
     67       pure $ s { added = s.added + 1, hitCount = s.hitCount, minTs = nextMinTs }
     68   step s (Listen { listenedAt: Nothing }) = do
     69     Log.warn "Skipping scrobble without timestamp"
     70     pure s
     71 
     72 -- | Wraps unsafeCoerce for attaching error handlers to HTTP requests.
     73 -- | This is the only legitimate use of unsafeCoerce in this module.
     74 castRequestForError :: forall a. a -> a
     75 castRequestForError = unsafeCoerce
     76 
     77 listenBrainzUrl :: String -> String
     78 listenBrainzUrl username = "https://api.listenbrainz.org/1/user/" <> username <> "/listens"
     79 
     80 -- | Retry an action with exponential backoff, but only for transient errors.
     81 -- | Parse errors and other permanent failures are not retried.
     82 withRetry :: forall a. String -> Aff a -> Aff a
     83 withRetry label action = recovering policy [ \_ err -> pure $ isTransientError err ] \(RetryStatus status) -> do
     84   when (status.iterNumber > 0)
     85     $ Log.warn
     86     $ label <> " failed, retry attempt " <> show status.iterNumber
     87   action
     88   where
     89   policy = exponentialBackoff (Milliseconds 1000.0) <> limitRetries 5
     90   isTransientError err = not isPermanent
     91     where
     92     msg = message err
     93     isPermanent =
     94       msg == "Last.fm: Failed to parse JSON response"
     95         || startsWith "Last.fm API returned status" msg
     96         || startsWith "ListenBrainz API returned status" msg
     97   startsWith prefix str = case stripPrefix (Pattern prefix) str of
     98     Just _ -> true
     99     Nothing -> false
    100 
    101 fetchListenBrainzUrl :: String -> Aff String
    102 fetchListenBrainzUrl url = withRetry "ListenBrainz fetch" $ makeAff \callback -> do
    103   req <- HTTPS.get url
    104 
    105   req # on_ Client.responseH \res -> do
    106     launchAff_ do
    107       result <- try $ readableToStringUtf8 (IM.toReadable res)
    108       liftEffect $ case result of
    109         Left err ->
    110           callback (Left err)
    111         Right body ->
    112           if IM.statusCode res == 200 then
    113             callback (Right body)
    114           else
    115             callback (Left $ error $ "ListenBrainz API returned status " <> show (IM.statusCode res))
    116 
    117   let errorH = EventHandle "error" mkEffectFn1
    118   on_ errorH (\err -> callback (Left err)) (castRequestForError req)
    119 
    120   pure nonCanceler
    121 
    122 fetchLastfmPage :: String -> String -> Int -> Maybe Int -> Aff { tracks :: Array Json, totalPages :: Int }
    123 fetchLastfmPage apiKey lfmUser page mTo = withRetry "Last.fm fetch" do
    124   let
    125     toParam = case mTo of
    126       Just ts -> "&to=" <> show ts
    127       Nothing -> ""
    128     baseUrl = "https://ws.audioscrobbler.com/2.0/?method=user.getrecenttracks&user="
    129       <> (fromMaybe lfmUser $ encodeURIComponent lfmUser)
    130       <> "&format=json&limit=200&page="
    131       <> show page
    132       <> toParam
    133     url = baseUrl <> "&api_key=" <> apiKey
    134   let headers = { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)" }
    135   fr <- fetch url { method: GET, headers }
    136   if fr.status == 200 then do
    137     json <- fromJson fr.json
    138     case parseLastfmResponse json of
    139       Just result -> do
    140         pure result
    141       Nothing -> do
    142         Log.error $ "Last.fm: Failed to parse JSON response: " <> stringify json
    143         throwError $ error "Last.fm: Failed to parse JSON response"
    144   else do
    145     throwError $ error ("Last.fm API returned status " <> show fr.status)
    146 
    147 parseLastfmResponse :: Json -> Maybe { tracks :: Array Json, totalPages :: Int }
    148 parseLastfmResponse json = case decodeJson json of
    149   Right (LastfmResponse { recenttracks: LastfmRecentTracks { track, attr: LastfmAttr { totalPages } } }) ->
    150     let
    151       tracks = case track of
    152         LastfmTrackArray' ts -> ts
    153         LastfmTrackSingle' t -> [ t ]
    154     in
    155       Just { tracks, totalPages: totalPages }
    156   Left _ ->
    157     Nothing
    158 
    159 lastfmTrackToListen :: Json -> Maybe Listen
    160 lastfmTrackToListen json = case decodeJson json of
    161   Right (LastfmTrack { name, artist: LastfmArtist { text: artistName }, album, date }) -> do
    162     ts <- date >>= \(LastfmDate { uts }) -> Just uts
    163     let
    164       releaseName = album >>= \(LastfmAlbum { text }) -> text
    165       releaseMbid = album >>= \(LastfmAlbum { mbid }) -> if mbid == "" then Nothing else Just mbid
    166     pure $ Listen
    167       { listenedAt: Just ts
    168       , trackMetadata: TrackMetadata
    169           { trackName: Just name
    170           , artistName: Just artistName
    171           , releaseName: releaseName
    172           , genre: Nothing
    173           , label: Nothing
    174           , mbidMapping: Just $ MbidMapping
    175               { releaseMbid: releaseMbid
    176               , caaReleaseMbid: releaseMbid
    177               }
    178           }
    179       }
    180   Left _ ->
    181     Nothing
    182 
    183 recordSyncSuccess :: String -> String -> Int -> Aff Unit
    184 recordSyncSuccess slug source n = do
    185   liftEffect $ Metrics.incSyncRuns slug source "success"
    186   when (n > 0) $ liftEffect $ Metrics.incSyncScrobbles slug source n
    187   liftEffect $ Metrics.setSyncLastSuccess slug source
    188 
    189 lbSync :: Connection -> String -> String -> AVar Unit -> Aff Unit
    190 lbSync conn username slug writeLock = void do
    191   Log.info $ "Starting ListenBrainz sync for " <> username
    192   result <- try $ fetchListenBrainzUrl (listenBrainzUrl username <> "?count=100")
    193   case result of
    194     Left err -> do
    195       Log.error $ "Sync fetch error: " <> message err
    196       liftEffect $ Metrics.incSyncRuns slug "listenbrainz" "error"
    197     Right body ->
    198       case parseJson body >>= decodeJson of
    199         Left err -> do
    200           Log.error $ "Sync parse error: " <> show err
    201           liftEffect $ Metrics.incSyncRuns slug "listenbrainz" "error"
    202         Right (ListenBrainzResponse { payload: Payload { listens } }) -> do
    203           syncResult <- withTransaction conn writeLock (processListens listens)
    204           let
    205             added = syncResult.added
    206             minTs = syncResult.minTs
    207             hitCount = syncResult.hitCount
    208           Log.info $ "ListenBrainz batch 1: added " <> show added <> ", " <> show hitCount <> " already present."
    209           let allExist = hitCount == length listens && not (null listens)
    210           if allExist || null listens then do
    211             when (added > 0) $ Log.info $ "ListenBrainz sync complete. Added " <> show added <> " new scrobbles."
    212             recordSyncSuccess slug "listenbrainz" added
    213           else do
    214             total <- paginateUntilDone 2 minTs added
    215             Log.info $ "ListenBrainz sync complete. Added " <> show total <> " new scrobbles."
    216             recordSyncSuccess slug "listenbrainz" total
    217   where
    218   paginateUntilDone batchNum minTs acc = case minTs of
    219     Nothing ->
    220       pure acc
    221     Just ts -> do
    222       Log.info $ "Fetching ListenBrainz batch " <> show batchNum <> " (before " <> show ts <> ")..."
    223       result <- try $ fetchListenBrainzUrl (listenBrainzUrl username <> "?count=100&max_ts=" <> show ts)
    224       case result of
    225         Left err -> do
    226           Log.error $ "Sync fetch error: " <> message err
    227           pure acc
    228         Right body ->
    229           case parseJson body >>= decodeJson of
    230             Left err -> do
    231               Log.error $ "Sync parse error: " <> show err
    232               pure acc
    233             Right (ListenBrainzResponse { payload: Payload { listens } }) -> do
    234               syncResult <- withTransaction conn writeLock (processListens listens)
    235               let
    236                 added = syncResult.added
    237                 newMinTs = syncResult.minTs
    238                 hitCount = syncResult.hitCount
    239               Log.info $ "ListenBrainz batch " <> show batchNum <> ": added " <> show added <> ", " <> show hitCount <> " already present."
    240               let allExist = hitCount == length listens && not (null listens)
    241               if allExist || null listens then
    242                 pure (acc + added)
    243               else
    244                 paginateUntilDone (batchNum + 1) newMinTs (acc + added)
    245 
    246   processListens listens = processTracks conn listens true
    247 
    248 lbSyncLoop :: Connection -> String -> String -> AVar Unit -> Aff Unit
    249 lbSyncLoop conn username slug writeLock = forever do
    250   delay (Milliseconds 60000.0)
    251   lbSync conn username slug writeLock
    252 
    253 lfSync :: Connection -> String -> String -> String -> AVar Unit -> Aff Unit
    254 lfSync conn apiKey lfmUser slug writeLock = do
    255   void performLastfmSync
    256   void performLastfmBackfill
    257   where
    258   performLastfmSync = do
    259     Log.info $ "Starting Last.fm sync for " <> lfmUser
    260     res <- try $ fetchLastfmPage apiKey lfmUser 1 Nothing
    261     case res of
    262       Left err -> do
    263         Log.error $ "Last.fm sync fetch error: " <> message err
    264         liftEffect $ Metrics.incSyncRuns slug "lastfm" "error"
    265       Right { tracks, totalPages } -> do
    266         result <- withTransaction conn writeLock (processLastfmTracks tracks)
    267         let
    268           added = result.added
    269           hitCount = result.hitCount
    270         Log.info $ "Last.fm page 1/" <> show totalPages <> ": added " <> show added <> ", " <> show hitCount <> " already present."
    271         let
    272           validTracks = mapMaybe lastfmTrackToListen tracks
    273           allExist = hitCount == length validTracks && not (null validTracks)
    274         if allExist || totalPages <= 1 then do
    275           when (added > 0) $ Log.info $ "Last.fm sync complete. Added " <> show added <> " new scrobbles."
    276           recordSyncSuccess slug "lastfm" added
    277         else do
    278           total <- paginateLastfmUntilDone 2 totalPages Nothing added
    279           Log.info $ "Last.fm sync complete. Added " <> show total <> " new scrobbles."
    280           recordSyncSuccess slug "lastfm" total
    281 
    282   paginateLastfmUntilDone page totalPages mTo acc
    283     | page > totalPages = pure acc
    284     | otherwise = do
    285         Log.info $ "Fetching Last.fm page " <> show page <> "/" <> show totalPages <> "..."
    286         res <- try $ fetchLastfmPage apiKey lfmUser page mTo
    287         case res of
    288           Left err -> do
    289             Log.error $ "Last.fm sync fetch error: " <> message err
    290             pure acc
    291           Right { tracks } -> do
    292             result <- withTransaction conn writeLock (processLastfmTracks tracks)
    293             let
    294               added = result.added
    295               hitCount = result.hitCount
    296             Log.info $ "Last.fm page" <> show page <> "/" <> show totalPages <> ": added " <> show added <> ", " <> show hitCount <> " already present."
    297             let
    298               validTracks = mapMaybe lastfmTrackToListen tracks
    299               allExist = hitCount == length validTracks && not (null validTracks)
    300             if allExist || null tracks then pure (acc + added)
    301             else paginateLastfmUntilDone (page + 1) totalPages mTo (acc + added)
    302 
    303   performLastfmBackfill = do
    304     mOldest <- getOldestTs conn
    305     for_ mOldest \oldestTs -> do
    306       Log.info $ "Checking for Last.fm history before " <> show oldestTs <> "..."
    307       res <- try $ fetchLastfmPage apiKey lfmUser 1 (Just (oldestTs - 1))
    308       case res of
    309         Left err ->
    310           Log.error $ "Last.fm backfill fetch error: " <> message err
    311         Right { tracks, totalPages } ->
    312           if null tracks then
    313             Log.info "No older Last.fm history found."
    314           else do
    315             Log.info $ "Backfilling " <> show totalPages <> " pages of Last.fm history before " <> show oldestTs
    316             result <- withTransaction conn writeLock (processLastfmTracks tracks)
    317             let
    318               added = result.added
    319               hitCount = result.hitCount
    320             Log.info $ "Last.fm backfill page 1/" <> show totalPages <> ": added " <> show added <> ", " <> show hitCount <> " already present."
    321             total <- paginateLastfmUntilDone 2 totalPages (Just (oldestTs - 1)) added
    322             Log.info $ "Last.fm backfill complete. Added " <> show total <> " older scrobbles."
    323 
    324   processLastfmTracks tracks = processTracks conn (mapMaybe lastfmTrackToListen tracks) false
    325 
    326 lfSyncLoop :: Connection -> String -> String -> String -> AVar Unit -> Aff Unit
    327 lfSyncLoop conn apiKey lfmUser slug writeLock = forever do
    328   delay (Milliseconds 60000.0)
    329   lfSync conn apiKey lfmUser slug writeLock