Db.purs (17313B)
1 module Db where 2 3 import Prelude 4 5 import Config (S3Config) 6 import Control.Monad.Error.Class (throwError) 7 import Control.Monad.Rec.Class (forever) 8 import Data.Argonaut.Core (Json, toObject, toNumber, toString) 9 import Data.Array (mapMaybe, (!!), last, length, null, replicate) 10 import Data.Either (Either(..)) 11 import Data.Foldable (for_) 12 import Data.Formatter.DateTime (formatDateTime) 13 import Data.Function.Uncurried (Fn2, Fn4, runFn2, runFn4) 14 import Data.Int as Int 15 import Data.Maybe (Maybe(..), fromMaybe) 16 import Data.Generic.Rep (class Generic) 17 import Data.Show.Generic (genericShow) 18 import Data.Nullable (Nullable, toMaybe, toNullable) 19 import Data.String (Pattern(..), split, stripSuffix) 20 import Data.String.Common (joinWith) 21 import Data.Time.Duration (Milliseconds(..)) 22 import Data.Tuple (Tuple(..)) 23 import Data.UUID (genUUID, toString) as UUID 24 import Effect (Effect) 25 import Effect.Aff (Aff, bracket, delay, makeAff, nonCanceler, try) 26 import Effect.Aff.AVar (AVar) 27 import Effect.Aff.AVar as Avar 28 import Effect.Class (liftEffect) 29 import Effect.Exception (Error, error, message) 30 import Effect.Now (nowDateTime) 31 import Foreign (Foreign) 32 import Foreign.Object as Object 33 import Log as Log 34 import Metrics as Metrics 35 import Node.FS.Aff as FSA 36 import S3 as S3 37 import Types (Listen(..), TrackMetadata(..), MbidMapping(..), Stats(..), StatsEntry(..)) 38 import Unsafe.Coerce (unsafeCoerce) 39 40 foreign import data Connection :: Type 41 42 -- | Coerces a value to Foreign for SQL parameters. 43 -- | This centralizes all unsafeCoerce usage for database operations 44 -- | into a single auditable point. 45 toParam :: forall a. a -> Foreign 46 toParam = unsafeCoerce 47 48 data FilterField = FilterArtist | FilterAlbum | FilterLabel | FilterYear | FilterGenre | FilterTrack 49 50 derive instance Eq FilterField 51 derive instance Generic FilterField _ 52 instance Show FilterField where 53 show = genericShow 54 55 fromString :: String -> Maybe FilterField 56 fromString "artist" = Just FilterArtist 57 fromString "album" = Just FilterAlbum 58 fromString "label" = Just FilterLabel 59 fromString "year" = Just FilterYear 60 fromString "genre" = Just FilterGenre 61 fromString "track" = Just FilterTrack 62 fromString _ = Nothing 63 64 foreign import connectImpl :: Fn2 String (Nullable Error -> Nullable Connection -> Effect Unit) (Effect Unit) 65 foreign import runImpl :: Fn4 Connection String (Array Foreign) (Nullable Error -> Effect Unit) (Effect Unit) 66 foreign import allImpl :: Fn4 Connection String (Array Foreign) (Nullable Error -> Nullable (Array Json) -> Effect Unit) (Effect Unit) 67 foreign import checkpointImpl :: Fn2 Connection (Nullable Error -> Effect Unit) (Effect Unit) 68 foreign import sha256 :: String -> String 69 70 connect :: String -> Aff Connection 71 connect path = makeAff \cb -> do 72 runFn2 connectImpl path \err conn -> 73 case toMaybe err of 74 Just e -> 75 cb (Left e) 76 Nothing -> case toMaybe conn of 77 Just c -> cb (Right c) 78 Nothing -> cb (Left $ error "Failed to create connection") 79 pure nonCanceler 80 81 run :: Connection -> String -> Array Foreign -> Aff Unit 82 run conn sql params = makeAff \cb -> do 83 runFn4 runImpl conn sql params \err -> 84 case toMaybe err of 85 Just e -> cb (Left e) 86 Nothing -> cb (Right unit) 87 pure nonCanceler 88 89 checkpoint :: Connection -> Aff Unit 90 checkpoint conn = makeAff \cb -> do 91 runFn2 checkpointImpl conn \err -> 92 case toMaybe err of 93 Just e -> cb (Left e) 94 Nothing -> cb (Right unit) 95 pure nonCanceler 96 97 dbBaseName :: String -> String 98 dbBaseName path = 99 let 100 parts = split (Pattern "/") path 101 name = fromMaybe path (last parts) 102 in 103 fromMaybe name (stripSuffix (Pattern ".db") name) 104 105 performBackup :: Connection -> String -> S3Config -> String -> Aff Unit 106 performBackup conn dbFile s3cfg slug = do 107 checkpoint conn 108 dt <- liftEffect nowDateTime 109 let 110 ts = case formatDateTime "YYYY-MM-DDTHH:mm:ss" dt of 111 Right s -> s 112 Left _ -> "unknown" 113 let key = "backups/" <> dbBaseName dbFile <> "-" <> ts <> ".db" 114 buf <- FSA.readFile dbFile 115 S3.uploadToS3 s3cfg key buf "application/octet-stream" 116 Log.info $ "Backup uploaded to S3: " <> key 117 liftEffect $ Metrics.setDbBackupLastSuccess slug 118 liftEffect $ Metrics.incDbBackupRun slug "success" 119 120 backupDb :: Connection -> String -> S3Config -> Number -> String -> Aff Unit 121 backupDb conn dbFile s3cfg intervalMs slug = forever do 122 delay (Milliseconds intervalMs) 123 result <- try $ performBackup conn dbFile s3cfg slug 124 case result of 125 Left err -> do 126 Log.error $ "Backup failed: " <> message err 127 liftEffect $ Metrics.incDbBackupRun slug "error" 128 Right _ -> 129 pure unit 130 131 -- Acquires the write lock, runs the action inside a transaction, then releases. 132 -- Uses bracket to guarantee lock release even on fiber kill, preventing deadlocks. 133 withTransaction :: forall a. Connection -> AVar Unit -> Aff a -> Aff a 134 withTransaction conn lock action = 135 bracket (Avar.take lock) (\_ -> Avar.put unit lock) \_ -> do 136 run conn "BEGIN TRANSACTION" [] 137 r <- try action 138 case r of 139 Left err -> do 140 void $ try $ run conn "ROLLBACK" [] 141 throwError err 142 Right v -> do 143 void $ try $ run conn "COMMIT" [] 144 pure v 145 146 queryAll :: Connection -> String -> Array Foreign -> Aff (Array Json) 147 queryAll conn sql params = makeAff \cb -> do 148 runFn4 allImpl conn sql params \err rows -> 149 case toMaybe err of 150 Just e -> cb (Left e) 151 Nothing -> cb (Right (fromMaybe [] (toMaybe rows))) 152 pure nonCanceler 153 154 initDb :: Connection -> Aff Unit 155 initDb conn = do 156 run conn "CREATE TABLE IF NOT EXISTS scrobbles (listened_at BIGINT PRIMARY KEY, track_name VARCHAR, artist_name VARCHAR, release_name VARCHAR, release_mbid VARCHAR, caa_release_mbid VARCHAR)" [] 157 run conn "CREATE TABLE IF NOT EXISTS api_tokens (slug VARCHAR PRIMARY KEY, hashed_token VARCHAR UNIQUE)" [] 158 159 getOrCreateToken :: Connection -> String -> Aff (Maybe String) 160 getOrCreateToken conn slug = do 161 rows <- queryAll conn "SELECT hashed_token FROM api_tokens WHERE slug = ?" [ toParam slug ] 162 case rows !! 0 >>= toObject >>= Object.lookup "hashed_token" >>= toString of 163 Just _ -> 164 pure Nothing 165 Nothing -> do 166 token <- liftEffect $ map UUID.toString UUID.genUUID 167 let hashedToken = sha256 token 168 run conn "INSERT INTO api_tokens (slug, hashed_token) VALUES (?, ?)" [ toParam slug, toParam hashedToken ] 169 pure $ Just token 170 171 checkExists :: Connection -> Int -> Aff Boolean 172 checkExists conn ts = do 173 rows <- queryAll conn "SELECT 1 FROM scrobbles WHERE listened_at = ?" [ toParam ts ] 174 pure case rows of 175 [] -> false 176 _ -> true 177 178 getOldestTs :: Connection -> Aff (Maybe Int) 179 getOldestTs conn = do 180 rows <- queryAll conn "SELECT MIN(listened_at) as min_ts FROM scrobbles" [] 181 pure $ do 182 row <- rows !! 0 183 obj <- toObject row 184 n <- Object.lookup "min_ts" obj >>= toNumber 185 let ts = Int.round n 186 if ts == 0 then Nothing else Just ts 187 188 upsertScrobble :: Connection -> Listen -> Aff Unit 189 upsertScrobble conn (Listen { listenedAt, trackMetadata: TrackMetadata track }) = 190 for_ listenedAt \ts -> do 191 let 192 mbid = fromMaybe (MbidMapping { releaseMbid: Nothing, caaReleaseMbid: Nothing }) track.mbidMapping 193 MbidMapping m = mbid 194 params = 195 [ toParam ts 196 , toParam (fromMaybe "" track.trackName) 197 , toParam (fromMaybe "" track.artistName) 198 , toParam (fromMaybe "" track.releaseName) 199 , toParam (fromMaybe "" m.releaseMbid) 200 , toParam (fromMaybe "" m.caaReleaseMbid) 201 ] 202 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 203 204 scrobbleCols :: String 205 scrobbleCols = "SELECT s.listened_at, s.track_name, s.artist_name, s.release_name, s.release_mbid, s.caa_release_mbid, rm.genre, rm.label" 206 207 scrobbleFromLeft :: String 208 scrobbleFromLeft = " FROM scrobbles s LEFT JOIN release_metadata rm ON s.release_mbid = rm.release_mbid" 209 210 scrobbleFromInner :: String 211 scrobbleFromInner = " FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid" 212 213 scrobbleOrderPage :: String 214 scrobbleOrderPage = " ORDER BY s.listened_at DESC LIMIT ? OFFSET ?" 215 216 getScrobbles :: Connection -> Int -> Int -> Maybe { field :: FilterField, value :: String } -> Maybe String -> Aff (Array Listen) 217 getScrobbles conn limit offset _ (Just q) = do 218 let pattern = "%" <> q <> "%" 219 rows <- queryAll conn 220 ( scrobbleCols <> scrobbleFromLeft 221 <> " WHERE (s.track_name ILIKE ? OR s.artist_name ILIKE ? OR s.release_name ILIKE ? OR rm.label ILIKE ?)" 222 <> scrobbleOrderPage 223 ) 224 [ toParam pattern, toParam pattern, toParam pattern, toParam pattern, toParam limit, toParam offset ] 225 pure $ mapMaybe rowToListen rows 226 getScrobbles conn limit offset Nothing Nothing = do 227 rows <- queryAll conn 228 (scrobbleCols <> scrobbleFromLeft <> scrobbleOrderPage) 229 [ toParam limit, toParam offset ] 230 pure $ mapMaybe rowToListen rows 231 getScrobbles conn limit offset (Just { field, value }) Nothing = do 232 rows <- queryAll conn (filterQuery field) [ toParam value, toParam limit, toParam offset ] 233 pure $ mapMaybe rowToListen rows 234 235 filterQuery :: FilterField -> String 236 filterQuery FilterArtist = 237 scrobbleCols <> scrobbleFromLeft 238 <> " WHERE s.artist_name = ?" 239 <> scrobbleOrderPage 240 filterQuery FilterAlbum = 241 scrobbleCols <> scrobbleFromLeft 242 <> " WHERE s.release_name = ?" 243 <> scrobbleOrderPage 244 filterQuery FilterLabel = 245 scrobbleCols <> scrobbleFromInner 246 <> " WHERE rm.label = ?" 247 <> scrobbleOrderPage 248 filterQuery FilterYear = 249 scrobbleCols <> scrobbleFromInner 250 <> " WHERE rm.release_year::VARCHAR = ?" 251 <> scrobbleOrderPage 252 filterQuery FilterGenre = 253 scrobbleCols <> scrobbleFromInner 254 <> " WHERE rm.genre = ?" 255 <> scrobbleOrderPage 256 filterQuery FilterTrack = 257 scrobbleCols <> scrobbleFromLeft 258 <> " WHERE s.track_name = ?" 259 <> scrobbleOrderPage 260 261 initReleaseMetadata :: Connection -> Aff Unit 262 initReleaseMetadata conn = do 263 run conn "CREATE TABLE IF NOT EXISTS release_metadata (release_mbid VARCHAR PRIMARY KEY, genre VARCHAR, label VARCHAR, release_year INTEGER, genre_checked_at INTEGER)" [] 264 -- Migration for existing databases; DuckDB supports IF NOT EXISTS for adding columns 265 run conn "ALTER TABLE release_metadata ADD COLUMN IF NOT EXISTS genre_checked_at INTEGER" [] 266 267 ping :: Connection -> Aff Unit 268 ping conn = void $ queryAll conn "SELECT 1" [] 269 270 getUnenrichedMbids :: Connection -> Int -> Aff (Array String) 271 getUnenrichedMbids conn limit = do 272 rows <- queryAll conn 273 "SELECT DISTINCT release_mbid FROM scrobbles WHERE release_mbid != '' AND release_mbid NOT IN (SELECT release_mbid FROM release_metadata) LIMIT ?" 274 [ toParam limit ] 275 pure $ mapMaybe extractMbid rows 276 where 277 extractMbid json = do 278 obj <- toObject json 279 Object.lookup "release_mbid" obj >>= toString 280 281 getEmptyGenreMbids :: Connection -> Int -> Aff (Array String) 282 getEmptyGenreMbids conn limit = do 283 rows <- queryAll conn 284 "SELECT DISTINCT s.release_mbid FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid WHERE (rm.genre IS NULL OR rm.genre = '') AND (rm.genre_checked_at IS NULL OR rm.genre_checked_at < CAST(epoch(now()) AS INTEGER) - 604800) LIMIT ?" 285 [ toParam limit ] 286 pure $ mapMaybe extractMbid rows 287 where 288 extractMbid json = do 289 obj <- toObject json 290 Object.lookup "release_mbid" obj >>= toString 291 292 upsertReleaseMetadata :: Connection -> String -> Maybe String -> Maybe String -> Maybe Int -> Aff Unit 293 upsertReleaseMetadata conn mbid genre label year = 294 run conn 295 "INSERT INTO release_metadata (release_mbid, genre, label, release_year) VALUES (?, ?, ?, ?) ON CONFLICT(release_mbid) DO UPDATE SET genre=excluded.genre, label=excluded.label, release_year=excluded.release_year" 296 [ toParam mbid 297 , toParam (toNullable genre) 298 , toParam (toNullable label) 299 , toParam (toNullable year) 300 ] 301 302 touchGenreCheckedAt :: Connection -> String -> Aff Unit 303 touchGenreCheckedAt conn mbid = do 304 run conn 305 "UPDATE release_metadata SET genre_checked_at = CAST(epoch(now()) AS INTEGER) WHERE release_mbid = ?" 306 [ toParam mbid ] 307 308 getStats :: Connection -> Maybe String -> Maybe String -> Maybe String -> Maybe String -> Aff Stats 309 getStats conn mPeriod mFrom mTo mSection = do 310 let 311 buildTimeFilterAndParams :: { timeFilter :: String, params :: Array Foreign } 312 buildTimeFilterAndParams = case mFrom, mTo of 313 Just from, Just to -> 314 { timeFilter: " AND s.listened_at >= CAST(epoch(TIMESTAMP ? || ' 00:00:00') AS INTEGER) AND s.listened_at < CAST(epoch(TIMESTAMP ? || ' 00:00:00') AS INTEGER) + 86400" 315 , params: [ toParam from, toParam to ] 316 } 317 _, _ -> case mPeriod >>= Int.fromString of 318 Just days -> 319 { timeFilter: " AND s.listened_at >= CAST(epoch(now()) AS INTEGER) - ?" 320 , params: [ toParam (days * 86400) ] 321 } 322 Nothing -> 323 { timeFilter: "", params: [] } 324 { timeFilter, params: buildTimeParams } = buildTimeFilterAndParams 325 fetch name q extraParams = case mSection of 326 Nothing -> queryAll conn (q <> " LIMIT 50") (buildTimeParams <> extraParams) 327 Just s | s == name -> queryAll conn q (buildTimeParams <> extraParams) 328 Just _ -> pure [] 329 genreRows <- fetch "genre" ("SELECT rm.genre as name, COUNT(*) as count FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid WHERE rm.genre IS NOT NULL AND rm.genre != ''" <> timeFilter <> " GROUP BY rm.genre ORDER BY count DESC") [] 330 labelRows <- fetch "label" ("SELECT rm.label as name, COUNT(*) as count FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid WHERE rm.label IS NOT NULL AND rm.label != ''" <> timeFilter <> " GROUP BY rm.label ORDER BY count DESC") [] 331 yearRows <- fetch "year" ("SELECT CAST(rm.release_year AS VARCHAR) as name, COUNT(*) as count FROM scrobbles s JOIN release_metadata rm ON s.release_mbid = rm.release_mbid WHERE rm.release_year IS NOT NULL" <> timeFilter <> " GROUP BY rm.release_year ORDER BY rm.release_year DESC") [] 332 artistRows <- fetch "artist" ("SELECT s.artist_name as name, COUNT(*) as count FROM scrobbles s WHERE s.artist_name != ''" <> timeFilter <> " GROUP BY s.artist_name ORDER BY count DESC") [] 333 trackRows <- fetch "track" ("SELECT s.artist_name || ' — ' || s.track_name as name, COUNT(*) as count FROM scrobbles s WHERE s.track_name != '' AND s.artist_name != ''" <> timeFilter <> " GROUP BY s.artist_name, s.track_name ORDER BY count DESC") [] 334 pure $ Stats 335 { genres: mapMaybe rowToEntry genreRows 336 , labels: mapMaybe rowToEntry labelRows 337 , years: mapMaybe rowToEntry yearRows 338 , artists: mapMaybe rowToEntry artistRows 339 , tracks: mapMaybe rowToEntry trackRows 340 } 341 342 rowToEntry :: Json -> Maybe StatsEntry 343 rowToEntry json = do 344 obj <- toObject json 345 name <- Object.lookup "name" obj >>= toString 346 count <- map Int.round $ Object.lookup "count" obj >>= toNumber 347 pure $ StatsEntry { name, count } 348 349 getArtistReleasesByMbids :: Connection -> Array String -> Aff (Object.Object { artist :: String, release :: String }) 350 getArtistReleasesByMbids _ mbids | null mbids = pure Object.empty 351 getArtistReleasesByMbids conn mbids = do 352 let placeholders = joinWith "," (replicate (length mbids) "?") 353 rows <- queryAll conn 354 ( "SELECT DISTINCT release_mbid, artist_name, release_name FROM scrobbles" 355 <> " WHERE release_mbid IN (" 356 <> placeholders 357 <> ") AND artist_name != '' AND release_name != ''" 358 ) 359 (map toParam mbids) 360 pure $ Object.fromFoldable (mapMaybe extractPair rows) 361 where 362 extractPair json = do 363 obj <- toObject json 364 mbid <- Object.lookup "release_mbid" obj >>= toString 365 artist <- Object.lookup "artist_name" obj >>= toString 366 release <- Object.lookup "release_name" obj >>= toString 367 pure $ Tuple mbid { artist, release } 368 369 rowToListen :: Json -> Maybe Listen 370 rowToListen json = do 371 obj <- toObject json 372 listenedAt <- map Int.round $ Object.lookup "listened_at" obj >>= toNumber 373 trackName <- Object.lookup "track_name" obj >>= toString 374 artistName <- Object.lookup "artist_name" obj >>= toString 375 releaseName <- Object.lookup "release_name" obj >>= toString 376 releaseMbid <- Object.lookup "release_mbid" obj >>= toString 377 caaReleaseMbid <- Object.lookup "caa_release_mbid" obj >>= toString 378 let genre = Object.lookup "genre" obj >>= toString 379 let label = Object.lookup "label" obj >>= toString 380 381 pure $ Listen 382 { listenedAt: Just listenedAt 383 , trackMetadata: TrackMetadata 384 { trackName: Just trackName 385 , artistName: Just artistName 386 , releaseName: Just releaseName 387 , genre 388 , label 389 , mbidMapping: Just $ MbidMapping 390 { releaseMbid: if releaseMbid == "" then Nothing else Just releaseMbid 391 , caaReleaseMbid: if caaReleaseMbid == "" then Nothing else Just caaReleaseMbid 392 } 393 } 394 } 395 396 getTokenUser :: Connection -> String -> Aff (Maybe String) 397 getTokenUser conn token = do 398 let hashedToken = sha256 token 399 rows <- queryAll conn "SELECT slug FROM api_tokens WHERE hashed_token = ?" [ toParam hashedToken ] 400 pure $ rows !! 0 >>= toObject >>= Object.lookup "slug" >>= toString