corpus

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

Main.purs (16384B)

      1 module Main where
      2 
      3 import Prelude
      4 
      5 import Config (AppConfig, UserConfig, UserEntry, loadConfig, s3ConfigFromUser)
      6 import Cover (serveCover)
      7 import Cosine (serveSimilar)
      8 import Data.Argonaut (decodeJson, encodeJson, parseJson, stringify)
      9 import Data.Array (find, length)
     10 import Data.Array as Data.Array
     11 import Data.Either (Either(..), fromRight')
     12 import Data.Foldable (for_, traverse_)
     13 import Data.Int as Int
     14 import Data.Maybe (Maybe(..), fromMaybe)
     15 import Data.String (Pattern(..), stripPrefix)
     16 import Data.String.Regex (regex, replace, parseFlags)
     17 import Partial.Unsafe (unsafeCrashWith)
     18 import Data.Traversable (traverse)
     19 import Db (Connection, backupDb, connect, fromString, getOrCreateToken, getScrobbles, getStats, getTokenUser, initDb, initReleaseMetadata, ping, upsertScrobble, withTransaction)
     20 import Effect (Effect)
     21 import Effect.Aff (Aff, Fiber, forkAff, joinFiber, killFiber, launchAff_, try)
     22 import Effect.Aff.AVar (AVar)
     23 import Effect.Aff.AVar as Avar
     24 import Effect.Class (liftEffect)
     25 import Effect.Exception as Exception
     26 import Node.Encoding (Encoding(UTF8))
     27 import Log as Log
     28 import Node.EventEmitter (on_)
     29 import Metadata (enrichMetadata)
     30 import Metrics as Metrics
     31 import Node.FS.Aff as FSA
     32 import Node.HTTP (createServer)
     33 import Node.HTTP.IncomingMessage as IM
     34 import Node.HTTP.OutgoingMessage (setHeader, toWriteable)
     35 import Node.HTTP.ServerResponse (setStatusCode, toOutgoingMessage)
     36 import Node.HTTP.Server as Server
     37 import Node.Net.Server (listenTcp, listeningH)
     38 import Node.Process (argv, lookupEnv)
     39 import Node.Stream (end, write, writeString)
     40 import Node.Stream.Aff (readableToStringUtf8)
     41 import Sync (lbSync, lbSyncLoop, lfSync, lfSyncLoop)
     42 import Web.URL (URL)
     43 import Web.URL as URL
     44 import Web.URL.URLSearchParams as URLSearchParams
     45 import Command as Command
     46 import Handler (Request, Response, serveBadRequest, serveError, serveInternalError, serveNotFound, serveUnauthorized)
     47 import Templates (indexHtml)
     48 import Types (Listen(..), ListenBrainzAdditionalInfo(..), ListenBrainzSubmitListen(..), ListenBrainzSubmitPayload(..), ListenBrainzSubmitTrackMetadata(..), MbidMapping(..), TrackMetadata(..))
     49 import Foreign.Object as Object
     50 
     51 type UserContext =
     52   { conn :: Connection
     53   , writeLock :: AVar Unit
     54   , config :: UserConfig
     55   , slug :: String
     56   , displayName :: String
     57   , enrichMetadataFiber :: Maybe (Fiber Unit)
     58   , backupFiber :: Maybe (Fiber Unit)
     59   }
     60 
     61 normalizePath :: String -> String
     62 normalizePath path = case stripPrefix (Pattern "/u/") path of
     63   Just _ -> "/u/:slug"
     64   Nothing -> path
     65 
     66 -- Request handler
     67 -- API endpoints (/proxy, /stats, /cover, /healthz) select the user via ?user=<slug>.
     68 -- Index pages are served at / (root user) and /u/<slug> (named users).
     69 handleRequest :: Boolean -> String -> Array UserContext -> Request -> Response -> Effect Unit
     70 handleRequest metricsEnabled corsOrigin contexts req res = do
     71   let method = IM.method req
     72   let rawUrl = IM.url req
     73   let allUsers = map (\ctx -> { slug: ctx.slug, name: ctx.displayName }) contexts
     74   case URL.fromRelative rawUrl "http://localhost" of
     75     Nothing ->
     76       serveNotFound res
     77     Just url -> do
     78       let path = URL.pathname url
     79       Metrics.wrapRequest method (normalizePath path) Log.info req res do
     80         launchAff_ $ do
     81           result <- try $ routeRequest metricsEnabled corsOrigin contexts req url path allUsers res
     82           case result of
     83             Left err -> do
     84               Log.error $ "Internal server error: " <> Exception.message err
     85               liftEffect $ serveInternalError res
     86             Right _ ->
     87               pure unit
     88 
     89 routeRequest :: Boolean -> String -> Array UserContext -> Request -> URL -> String -> Array { slug :: String, name :: String } -> Response -> Aff Unit
     90 routeRequest metricsEnabled corsOrigin contexts req url path allUsers res = liftEffect $ case path of
     91   "/client.js" ->
     92     serveClientJs res
     93   "/favicon.png" ->
     94     serveAsset "image/png" "assets/favicon.png" res
     95   "/cover.webp" ->
     96     serveAsset "image/webp" "assets/cover.webp" res
     97   "/" ->
     98     serveIndex allUsers "" res
     99   "/metrics" ->
    100     if metricsEnabled then serveMetrics res
    101     else do
    102       Log.warn "Path not found: /metrics"
    103       serveNotFound res
    104   "/proxy" ->
    105     withUser url \ctx -> serveProxy corsOrigin ctx.conn url res
    106   "/stats" ->
    107     withUser url \ctx -> serveStats ctx.conn url res
    108   "/cover" ->
    109     withUser url \ctx -> launchAff_ $ serveCover serveNotFound ctx.config ctx.slug url res
    110   "/similar" ->
    111     withUser url \ctx -> serveSimilar serveBadRequest serveError ctx.slug ctx.config url res
    112   "/healthz" ->
    113     withUser url \ctx -> serveHealthz ctx.conn res
    114   "/1/submit-listens" ->
    115     if IM.method req == "POST" then
    116       launchAff_ $ withUserFromToken contexts req res \ctx ->
    117         serveSubmitListens ctx.slug ctx.conn ctx.writeLock req res
    118     else
    119       liftEffect $ serveBadRequest res "Method not allowed"
    120   _ ->
    121     case stripPrefix (Pattern "/u/") path of
    122       Just slug ->
    123         serveIndex allUsers slug res
    124       Nothing -> do
    125         Log.warn $ "Path not found: " <> path
    126         serveNotFound res
    127   where
    128   withUser urlParam f =
    129     let
    130       slug = fromMaybe "" (getQueryParam "user" urlParam)
    131     in
    132       case find (\c -> c.slug == slug) contexts of
    133         Nothing -> do
    134           Log.warn $ "Unknown user: " <> show slug
    135           serveNotFound res
    136         Just ctx ->
    137           f ctx
    138 
    139 withUserFromToken :: Array UserContext -> Request -> Response -> (UserContext -> Aff Unit) -> Aff Unit
    140 withUserFromToken contextsParam reqParam resParam f = do
    141   let
    142     headers = IM.headers reqParam
    143     mAuth = Object.lookup "authorization" headers
    144     mToken = mAuth >>= stripPrefix (Pattern "Token ")
    145   case mToken of
    146     Nothing -> liftEffect $ do
    147       Log.warn "Missing or invalid Authorization header"
    148       serveUnauthorized resParam
    149     Just tokenValue -> do
    150       -- We need to check which connection to use to verify the token.
    151       -- Since initDb creates the table in every user's database, but usually tokens are user-specific.
    152       -- The current structure assumes each user has their own DB.
    153       -- We'll try to find the user by checking each context's DB for the token.
    154       mCtx <- findUserByToken contextsParam tokenValue
    155       case mCtx of
    156         Nothing -> liftEffect $ do
    157           Log.warn $ "Unknown API token provided"
    158           serveUnauthorized resParam
    159         Just ctx ->
    160           f ctx
    161 
    162 findUserByToken :: Array UserContext -> String -> Aff (Maybe UserContext)
    163 findUserByToken contexts tokenValue = case Data.Array.uncons contexts of
    164   Nothing ->
    165     pure Nothing
    166   Just { head: ctx, tail: rest } -> do
    167     mSlug <- getTokenUser ctx.conn tokenValue
    168     case mSlug of
    169       Just slug | slug == ctx.slug ->
    170         pure (Just ctx)
    171       _ ->
    172         findUserByToken rest tokenValue
    173 
    174 serveIndex :: Array { slug :: String, name :: String } -> String -> Response -> Effect Unit
    175 serveIndex allUsers slug res = do
    176   setHeader "Content-Type" "text/html" (toOutgoingMessage res)
    177   setStatusCode 200 res
    178   let w = toWriteable (toOutgoingMessage res)
    179   void $ writeString w UTF8 (indexHtml slug allUsers)
    180   end w
    181 
    182 serveMetrics :: Response -> Effect Unit
    183 serveMetrics res = do
    184   launchAff_ do
    185     contentType <- liftEffect Metrics.getContentType
    186     metricsText <- Metrics.getMetrics
    187     liftEffect $ do
    188       setHeader "Content-Type" contentType (toOutgoingMessage res)
    189       setStatusCode 200 res
    190       let w = toWriteable (toOutgoingMessage res)
    191       void $ writeString w UTF8 metricsText
    192       end w
    193 
    194 serveClientJs :: Response -> Effect Unit
    195 serveClientJs res = do
    196   setHeader "Content-Type" "application/javascript" (toOutgoingMessage res)
    197   launchAff_ do
    198     result <- try $ FSA.readTextFile UTF8 "client.js"
    199     liftEffect $ case result of
    200       Right content -> do
    201         setStatusCode 200 res
    202         let w = toWriteable (toOutgoingMessage res)
    203         void $ writeString w UTF8 content
    204         end w
    205       Left err -> do
    206         Log.error $ "Failed to read client.js: " <> Exception.message err
    207         serveNotFound res
    208 
    209 getQueryParam :: String -> URL -> Maybe String
    210 getQueryParam name url = URLSearchParams.get name (URL.searchParams url)
    211 
    212 serveProxy :: String -> Connection -> URL -> Response -> Effect Unit
    213 serveProxy corsOrigin db url res = do
    214   setHeader "Content-Type" "application/json" (toOutgoingMessage res)
    215   setHeader "Access-Control-Allow-Origin" corsOrigin (toOutgoingMessage res)
    216 
    217   launchAff_ do
    218     let limit = fromMaybe 25 (getQueryParam "limit" url >>= Int.fromString)
    219     let offset = fromMaybe 0 (getQueryParam "offset" url >>= Int.fromString)
    220     let
    221       mFilter = do
    222         field <- getQueryParam "filterField" url >>= fromString
    223         value <- getQueryParam "filterValue" url
    224         pure { field, value }
    225 
    226     let mSearch = getQueryParam "search" url
    227     listens <- getScrobbles db limit offset mFilter mSearch
    228     let responseBody = stringify $ encodeJson { payload: { listens: listens } }
    229 
    230     liftEffect $ do
    231       setStatusCode 200 res
    232       let w = toWriteable (toOutgoingMessage res)
    233       void $ writeString w UTF8 responseBody
    234       end w
    235 
    236 sanitizeDate :: String -> String
    237 sanitizeDate = replace (fromRight' (\_ -> unsafeCrashWith "invalid regex") $ regex "[^0-9\\-]" (parseFlags "g")) ""
    238 
    239 serveStats :: Connection -> URL -> Response -> Effect Unit
    240 serveStats db url res = do
    241   setHeader "Content-Type" "application/json" (toOutgoingMessage res)
    242   launchAff_ do
    243     let period = getQueryParam "period" url
    244     let section = getQueryParam "section" url
    245     let safeDate = map sanitizeDate
    246     let mFrom = safeDate (getQueryParam "from" url)
    247     let mTo = safeDate (getQueryParam "to" url)
    248     stats <- getStats db period mFrom mTo section
    249     let responseBody = stringify $ encodeJson stats
    250     liftEffect $ do
    251       setStatusCode 200 res
    252       let w = toWriteable (toOutgoingMessage res)
    253       void $ writeString w UTF8 responseBody
    254       end w
    255 
    256 serveAsset :: String -> String -> Response -> Effect Unit
    257 serveAsset contentType path res = do
    258   setHeader "Content-Type" contentType (toOutgoingMessage res)
    259   setHeader "Cache-Control" "public, max-age=86400" (toOutgoingMessage res)
    260   launchAff_ do
    261     result <- try $ FSA.readFile path
    262     liftEffect $ case result of
    263       Right buf -> do
    264         setStatusCode 200 res
    265         let w = toWriteable (toOutgoingMessage res)
    266         void $ write w buf
    267         end w
    268       Left _ ->
    269         serveNotFound res
    270 
    271 serveHealthz :: Connection -> Response -> Effect Unit
    272 serveHealthz db res = do
    273   setHeader "Content-Type" "application/json" (toOutgoingMessage res)
    274   launchAff_ do
    275     result <- try $ ping db
    276     liftEffect $ do
    277       let w = toWriteable (toOutgoingMessage res)
    278       case result of
    279         Right _ -> do
    280           setStatusCode 200 res
    281           void $ writeString w UTF8 """{"status":"ok"}"""
    282         Left err -> do
    283           setStatusCode 503 res
    284           void $ writeString w UTF8 $ """{"status":"error","message":""" <> show (Exception.message err) <> "}"
    285       end w
    286 
    287 serveSubmitListens :: String -> Connection -> AVar Unit -> Request -> Response -> Aff Unit
    288 serveSubmitListens slug db lock req res = do
    289   body <- readableToStringUtf8 (IM.toReadable req)
    290   case parseJson body >>= decodeJson of
    291     Left err ->
    292       liftEffect $ serveBadRequest res $ "Invalid JSON: " <> show err
    293     Right (ListenBrainzSubmitPayload { listenType, payload }) -> do
    294       let listens = Data.Array.mapMaybe (submitListenToListen listenType) payload
    295       withTransaction db lock $ traverse_ (upsertScrobble db) listens
    296       liftEffect $ do
    297         Metrics.incSyncScrobbles slug "api" (length listens)
    298         setHeader "Content-Type" "application/json" (toOutgoingMessage res)
    299         setStatusCode 200 res
    300         let w = toWriteable (toOutgoingMessage res)
    301         void $ writeString w UTF8 """{"status":"ok"}"""
    302         end w
    303 
    304 submitListenToListen :: String -> ListenBrainzSubmitListen -> Maybe Listen
    305 submitListenToListen "single" (ListenBrainzSubmitListen { listenedAt, trackMetadata }) =
    306   Just $ Listen
    307     { listenedAt
    308     , trackMetadata: submitTrackMetadataToTrackMetadata trackMetadata
    309     }
    310 submitListenToListen "import" (ListenBrainzSubmitListen { listenedAt, trackMetadata }) =
    311   Just $ Listen
    312     { listenedAt
    313     , trackMetadata: submitTrackMetadataToTrackMetadata trackMetadata
    314     }
    315 submitListenToListen "playing_now" _ = Nothing
    316 submitListenToListen _ _ = Nothing
    317 
    318 submitTrackMetadataToTrackMetadata :: ListenBrainzSubmitTrackMetadata -> TrackMetadata
    319 submitTrackMetadataToTrackMetadata (ListenBrainzSubmitTrackMetadata { trackName, artistName, releaseName, additionalInfo }) =
    320   TrackMetadata
    321     { trackName: Just trackName
    322     , artistName: Just artistName
    323     , releaseName
    324     , genre: Nothing
    325     , label: Nothing
    326     , mbidMapping: do
    327         info <- additionalInfo
    328         let ListenBrainzAdditionalInfo i = info
    329         pure $ MbidMapping
    330           { releaseMbid: i.releaseMbid
    331           , caaReleaseMbid: i.releaseMbid
    332           }
    333     }
    334 
    335 startUser :: UserEntry -> Aff UserContext
    336 startUser { slug, name, config } = do
    337   Log.info $ "Starting user: " <> if slug == "" then "(root)" else slug
    338   conn <- connect config.databaseFile
    339   initDb conn
    340   initReleaseMetadata conn
    341   writeLock <- Avar.new unit
    342 
    343   mToken <- getOrCreateToken conn slug
    344   for_ mToken \token ->
    345     Log.info $ "User '" <> slug <> "' token: " <> token
    346 
    347   case config.lastfmUser, config.lastfmApiKey of
    348     Just _, Nothing -> Log.warn $ "User '" <> slug <> "': Last.fm sync disabled (missing API key)"
    349     _, _ -> pure unit
    350 
    351   lbFiber <- case config.listenbrainzUser of
    352     Just username -> Just <$> forkAff (lbSync conn username slug writeLock)
    353     Nothing -> pure Nothing
    354   lfFiber <- case config.lastfmUser, config.lastfmApiKey of
    355     Just lfmUser, Just apiKey -> Just <$> forkAff (lfSync conn apiKey lfmUser slug writeLock)
    356     _, _ -> pure Nothing
    357 
    358   void $ forkAff do
    359     for_ lbFiber joinFiber
    360     for_ lfFiber joinFiber
    361     for_ config.listenbrainzUser \username ->
    362       void $ forkAff $ lbSyncLoop conn username slug writeLock
    363     case config.lastfmUser, config.lastfmApiKey of
    364       Just lfmUser, Just apiKey -> void $ forkAff $ lfSyncLoop conn apiKey lfmUser slug writeLock
    365       _, _ -> pure unit
    366 
    367   enrichMetadataFiber <- forkAff $ enrichMetadata conn config slug
    368   backupFiber <-
    369     if config.backupEnabled then Just <$> forkAff (backupDb conn config.databaseFile (s3ConfigFromUser config) (Int.toNumber config.backupIntervalHours * 3600000.0) slug)
    370     else pure Nothing
    371 
    372   let displayName = fromMaybe (if slug == "" then "root" else slug) name
    373   pure { conn, writeLock, config, slug, displayName, enrichMetadataFiber: Just enrichMetadataFiber, backupFiber }
    374 
    375 cleanupUser :: UserContext -> Aff Unit
    376 cleanupUser ctx = do
    377   let label = if ctx.slug == "" then "(root)" else ctx.slug
    378   Log.info $ "Shutting down user: " <> label
    379   for_ ctx.enrichMetadataFiber \fiber ->
    380     void $ try $ killFiber (Exception.error "Server shutting down") fiber
    381   for_ ctx.backupFiber \fiber ->
    382     void $ try $ killFiber (Exception.error "Server shutting down") fiber
    383   Log.info $ "Closed user: " <> label
    384 
    385 foreign import dotenvConfig :: Effect Unit
    386 
    387 main :: Effect Unit
    388 main = do
    389   dotenvConfig
    390   launchAff_ do
    391     configFile <- liftEffect $ map (fromMaybe "users.json") $ lookupEnv "CORPUS_USERS_FILE"
    392     args <- liftEffect argv
    393     let commandArgs = Data.Array.drop 2 args
    394     case Data.Array.uncons commandArgs of
    395       Just _ ->
    396         Command.run configFile commandArgs
    397       Nothing -> do
    398         result <- try $ loadConfig configFile
    399         case result of
    400           Left err -> do
    401             Log.error $ "Failed to load " <> configFile <> ": " <> Exception.message err
    402             liftEffect $ Exception.throwException err
    403           Right (appConfig :: AppConfig) -> do
    404             Log.info $ "Loaded " <> show (length appConfig.users) <> " user(s) from " <> configFile
    405             contexts <- traverse startUser appConfig.users
    406             liftEffect $ do
    407               server <- createServer
    408               server # on_ Server.requestH (handleRequest appConfig.metricsEnabled appConfig.corsOrigin contexts)
    409               let netServer = Server.toNetServer server
    410 
    411               netServer # on_ listeningH do
    412                 Log.info $ "Server is running on " <> appConfig.host <> ":" <> show appConfig.port
    413 
    414               listenTcp netServer { host: appConfig.host, port: appConfig.port, backlog: 128 }