feat(tealfm): Implement historical scrobble hydration · foxxmd.dev/multi-scrobbler@c4f028f

+3 -1

src/backend/common/database/drizzle/drizzleTypes.ts

··· 1 1 import { DBQueryConfig, DBQueryConfigWith, ExtractTablesFromSchema, KnownKeysOnly, RelationFieldsFilterInternals, Many, InferSelectModel, ExtractTablesWithRelations, type BuildQueryResult, RelationsFilter } from "drizzle-orm"; 2 - import { components, componentMigrations, playInputs, plays, queueStates, relations } from "./schema/schema.js"; 2 + import { components, componentMigrations, playInputs, plays, queueStates, relations, playsHistorical } from "./schema/schema.js"; 3 3 import {TSchema, TableName, Schema } from "./schema/schema.js"; 4 4 import { MarkOptional, MarkRequired } from "ts-essentials"; 5 5 ··· 21 21 export type PlayWith<K extends keyof TSchema['plays']["relations"]> = GenericRelationResult<'plays', K>; 22 22 export type PlayNew = typeof plays.$inferInsert; 23 23 24 + export type PlayHistoricalSelect = typeof playsHistorical.$inferSelect; 25 + export type PlayHistoricalNew = typeof playsHistorical.$inferInsert; 24 26 25 27 // useful references for building types 26 28 // https://github.com/drizzle-team/drizzle-orm/discussions/2596

+3 -2

src/backend/common/database/drizzle/entityUtils.ts

··· 1 1 import assert from "node:assert"; 2 - import { PlayNew, PlaySelect, PlaySelectWithQueueStates } from "./drizzleTypes.js"; 2 + import { PlayHistoricalNew, PlayHistoricalSelect, PlayNew, PlaySelect, PlaySelectWithQueueStates } from "./drizzleTypes.js"; 3 3 import { PlayInputNew } from "./drizzleTypes.js"; 4 4 import { QueueStateNew } from "./drizzleTypes.js"; 5 5 import { ComponentNew } from "./drizzleTypes.js"; ··· 20 20 } 21 21 22 22 export type PlayEntityOpts = Partial<Pick<PlayNew, 'seenAt' | 'playedAt' | 'uid' | 'state' | 'parentId' | 'componentId'>> & { error?: ErrorLike }; 23 + export type PlayHistoricalEntityOpts = Partial<Pick<PlayHistoricalNew, 'seenAt' | 'playedAt' | 'uid' | 'componentId'>>; 23 24 24 25 export const generatePlayEntity = (play: PlayObject, opts: PlayEntityOpts = {}): PlayNew => { 25 26 const { ··· 51 52 52 53 export type PlayHydateOptions = 'asPlay' | 'id' | 'uid'; 53 54 54 - export const hydratePlaySelect = (select: PlaySelect, opts: PlayHydateOptions[] = ['id','uid']): PlayObject => { 55 + export const hydratePlaySelect = <T extends PlaySelect | PlayHistoricalSelect>(select: T, opts: PlayHydateOptions[] = ['id','uid']): PlayObject => { 55 56 if(opts.length === 0) { 56 57 return select.play; 57 58 }

+340

src/backend/common/database/drizzle/repositories/PlayHistoricalRepository.ts

··· 1 + import { childLogger, Logger, LoggerAppExtras } from "@foxxmd/logging"; 2 + import { DbConcrete, runTransaction } from "../drizzleUtils.js"; 3 + import { loggerNoop } from "../../../MaybeLogger.js"; 4 + import { ErrorLike, PlayObject, TA_CLOSE, TA_DEFAULT_ACCURACY, TA_EXACT, TemporalAccuracy } from "../../../../../core/Atomic.js"; 5 + import { generateInputEntity, generatePlayEntity, PlayEntityOpts, hydratePlaySelect, PlayHydateOptions, PlayHistoricalEntityOpts } from "../entityUtils.js"; 6 + import { playInputs, plays, playsHistorical, queueStates, relations } from "../schema/schema.js"; 7 + import { PlayNew, PlaySelect, PlayInputNew, FindWhere, FindMany, QueueStateSelect, FindWith, PlaySelectWithQueueStates, WhereClause, PlayWith, PlayHistoricalSelect, PlayHistoricalNew } from "../drizzleTypes.js";; 8 + import { MarkOptional, MarkRequired, PathValue } from "ts-essentials"; 9 + import { genGroupIdStrFromPlay, removeEmptyArrays, removeUndefinedKeys } from "../../../../utils.js"; 10 + import dayjs, { Dayjs } from "dayjs"; 11 + import { RelationsFieldFilter, eq, inArray, ne, notInArray, desc, asc, and, sql, Placeholder } from "drizzle-orm"; 12 + import { CompactableProperty, RetentionOptions, retentionPlayTypes } from "../../../infrastructure/config/database.js"; 13 + import { shortTodayAwareFormat } from "../../../../../core/TimeUtils.js"; 14 + import { buildDateCompare, CompareDateOp, ComponentConstrainedRepoOpts, DrizzleBaseRepository, DrizzleRepositoryOpts, PaginatedQueryResponse, PaginatedResponse } from "./BaseRepository.js"; 15 + import { asPlay } from "../../../../../core/PlayMarshalUtils.js"; 16 + import assert, { Assert } from "node:assert"; 17 + import { hashObject, parseArrayFromMaybeString } from "../../../../utils/StringUtils.js"; 18 + import { playContentBasicInvariantTransform, playMbidIdentifier } from "../../../../utils/PlayComparisonUtils.js"; 19 + import { comparePlayTemporally, getScrobbleTsSOCDate, getScrobbleTsSOCDateWithContext, getTemporalAccuracyCloseVal, hasAcceptableTemporalAccuracy } from "../../../../utils/TimeUtils.js"; 20 + import { SourceType } from "../../../infrastructure/config/source/sources.js"; 21 + import { getTemporallyCloseDateCompareOp } from "./PlayRepository.js"; 22 + 23 + // https://github.com/drizzle-team/drizzle-orm/issues/695 may be useful for typing models with relations? 24 + 25 + export interface PlayWhereOpts { 26 + componentId?: number 27 + seenAt?: CompareDateOp 28 + playedAt?: CompareDateOp 29 + uid?: string[] 30 + } 31 + 32 + export interface QueryPlaysOpts extends PlayWhereOpts { 33 + sort?: 'seenAt' | 'playedAt' 34 + order?: 'asc' | 'desc' 35 + limit?: number 36 + offset?: number 37 + } 38 + 39 + export interface HydrateOpts { 40 + hydrate?: PlayHydateOptions[] 41 + } 42 + 43 + export type RepositoryCreatePlayHistoricalOpts = PlayHistoricalEntityOpts 44 + & Pick<PlayHistoricalNew, 'play' | 'componentId'>; 45 + 46 + type PlayIdentifierPrimitiveMap = { 47 + uid: string; 48 + id: number; 49 + }; 50 + 51 + const identifierExtractor: { [K in keyof PlayIdentifierPrimitiveMap]: (play: {id: number, uid: string}) => PlayIdentifierPrimitiveMap[K] } = { 52 + id: (play) => play.id, 53 + uid: (play) => play.uid, 54 + }; 55 + export class DrizzlePlayHistoricalRepository extends DrizzleBaseRepository<'playsHistorical'> { 56 + 57 + constructor(db: DbConcrete, opts: DrizzleRepositoryOpts = {}) { 58 + super(db, 'plays', 'Plays', opts); 59 + } 60 + 61 + findByUid = async (uid: string, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise<PlayHistoricalSelect | undefined> => { 62 + const res = await this.db.query.playsHistorical.findFirst({ 63 + where: { 64 + uid, 65 + componentId: opts.componentId ?? this.componentId 66 + } 67 + }); 68 + res.play = hydratePlaySelect(res, opts.hydrate); 69 + return res; 70 + } 71 + 72 + hasByUid = async (uid: string, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise<boolean> => { 73 + const res = await this.db.query.playsHistorical.findFirst({ 74 + columns: {id: true}, 75 + where: { 76 + uid, 77 + componentId: opts.componentId ?? this.componentId 78 + } 79 + }); 80 + return res !== undefined; 81 + } 82 + 83 + createPlays = async (entitiesOpts: RepositoryCreatePlayHistoricalOpts[], opts: HydrateOpts = {}) => { 84 + 85 + const { 86 + hydrate 87 + } = opts; 88 + let playRows: PlayHistoricalSelect[]; 89 + 90 + await runTransaction(this.db, async () => { 91 + 92 + const entitiesData = entitiesOpts.map((data) => { 93 + const { 94 + play, 95 + ...rest 96 + } = data; 97 + return generatePlayEntity(play, { componentId: this.componentId, ...rest }); 98 + }); 99 + 100 + playRows = await this.db.insert(playsHistorical).values(entitiesData).returning(); 101 + }); 102 + 103 + return playRows.map(x => ({...x, play: hydratePlaySelect(x, hydrate)})); 104 + } 105 + 106 + findPlays = async (args: QueryPlaysOpts, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise<PlayHistoricalSelect[]> => { 107 + const { 108 + hydrate, 109 + componentId = this.componentId 110 + } = opts; 111 + // this does not work as type for query variable 112 + // it erases the result type for some reason 113 + // 114 + // Parameters<typeof this.db.query.plays.findMany>[0] 115 + 116 + // this does work but it is also integrated into FindWith 117 + //let withQuery: Parameters<typeof this.db.query.plays.findMany>[0]['with'] = undefined; 118 + 119 + let query: FindMany<'playsHistorical'> = { 120 + limit: args.limit, 121 + offset: args.offset 122 + }; 123 + 124 + query.where = buildPlayHistoricalWhere({componentId: componentId, ...args}); 125 + 126 + if (args.sort !== undefined) { 127 + query.orderBy = { 128 + [args.sort]: args.order ?? 'desc' 129 + } 130 + } else { 131 + query.orderBy = { 132 + id: 'asc' 133 + } 134 + } 135 + 136 + query = removeUndefinedKeys(query); 137 + const results = await this.db.query.playsHistorical.findMany(query); 138 + return results.map((x) => ({...x, play: hydratePlaySelect(x, hydrate)})); 139 + } 140 + 141 + findPlayIds = async (args: QueryPlaysOpts, opts: ComponentConstrainedRepoOpts = {}): Promise<number[]> => { 142 + const { 143 + componentId = this.componentId 144 + } = opts; 145 + 146 + let query: FindMany<'playsHistorical'> = { 147 + limit: args.limit, 148 + offset: args.offset, 149 + }; 150 + 151 + query.where = buildPlayHistoricalWhere({componentId: componentId, ...args}); 152 + 153 + if (args.sort !== undefined) { 154 + query.orderBy = { 155 + [args.sort]: args.order ?? 'desc' 156 + } 157 + } else { 158 + query.orderBy = { 159 + id: 'asc' 160 + } 161 + } 162 + 163 + query = removeUndefinedKeys(query); 164 + const results = await this.db.query.plays.findMany({ 165 + limit: args.limit, 166 + offset: args.offset, 167 + columns: {id: true}, 168 + orderBy: args.sort !== undefined ? {[args.sort]: args.order ?? 'desc'} : {id: 'asc'}, 169 + }); 170 + return results.map((x) => x.id); 171 + } 172 + 173 + findPlayIdentifiers = async <T extends keyof PlayIdentifierPrimitiveMap>(args: QueryPlaysOpts, identifier: T, opts: ComponentConstrainedRepoOpts = {}): Promise<PlayIdentifierPrimitiveMap[T][]> => { 174 + const { 175 + componentId = this.componentId, 176 + } = opts; 177 + 178 + const results = await this.db.query.playsHistorical.findMany({ 179 + limit: args.limit, 180 + offset: args.offset, 181 + columns: {id: true, uid: true}, 182 + orderBy: args.sort !== undefined ? {[args.sort]: args.order ?? 'desc'} : {id: 'asc'}, 183 + where: buildPlayHistoricalWhere({componentId: componentId, ...args}) 184 + }); 185 + 186 + // we getting fancy now 187 + return results.map(identifierExtractor[identifier]); 188 + } 189 + 190 + findPlaysPaginated = async (args: QueryPlaysOpts, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise<PaginatedResponse<PlayHistoricalSelect>> => { 191 + const { 192 + limit = 100, 193 + offset = 0, 194 + ...rest 195 + } = args; 196 + const clampedLimit = Math.min(limit, 100); 197 + const res = await this.findPlays({limit: clampedLimit, offset, ...rest}, opts); 198 + return {data: res, meta: {limit: clampedLimit, offset}}; 199 + } 200 + 201 + // async updateById(id: number, data: Partial<PlayNew>): Promise<void> { 202 + // if(data.play !== undefined) { 203 + // data.play = withoutDbAwareness(data.play); 204 + // } 205 + // super.updateById(id, data); 206 + // } 207 + 208 + deletePlays = async (playsData: (Pick<PlayHistoricalSelect, 'id'> | number)[]) => { 209 + const ids = playsData.map(x => typeof x === 'number' ? x : x.id); 210 + await this.db.delete(playsHistorical).where(inArray(plays.id, ids)); 211 + } 212 + 213 + public checkExisting = async (play: PlayObject, opts: {taAccuracy?: TemporalAccuracy[]} & ComponentConstrainedRepoOpts = {}): Promise<PlayHistoricalSelect | undefined> => { 214 + const { 215 + componentId = this.componentId, 216 + taAccuracy = TA_DEFAULT_ACCURACY, 217 + } = opts; 218 + const hash = hashObject(playContentBasicInvariantTransform(play).data); 219 + 220 + // we get all plays with a play date between playdate - (source accuracy) AND (playDateCompleted or playDate) + (source accuracy) 221 + // which we can then use with temporal comparison to make sure we are comparing the correct dates 222 + // 223 + // this isn't as fast as just comparing playDate directly but its still much faster/cheaper than paginating plays and doing everything in-memory 224 + const dateGranularity = getTemporalAccuracyCloseVal(play.meta.source as SourceType); 225 + let endRange: Dayjs; 226 + if(play.data.playDateCompleted !== undefined) { 227 + // this will be present if source reports it 228 + // or we tracked it live with MemorySource 229 + endRange = play.data.playDateCompleted.add(dateGranularity, 's'); 230 + } else { 231 + endRange = play.data.playDate.add(dateGranularity, 's'); 232 + } 233 + let where: FindWhere<'playsHistorical'> = { 234 + componentId, 235 + playedAt: buildDateCompare(getTemporallyCloseDateCompareOp(play)), 236 + }; 237 + 238 + const mbidId = playMbidIdentifier(play); 239 + if(mbidId !== undefined) { 240 + where.AND = [ 241 + { 242 + OR: [ 243 + { 244 + playHash: hash 245 + }, 246 + { 247 + mbidIdentifier: mbidId 248 + } 249 + ] 250 + } 251 + ] 252 + } else { 253 + where.playHash = hash; 254 + } 255 + 256 + const res = await this.db.query.playsHistorical.findMany({ 257 + where 258 + }); 259 + if(res.length === 0) { 260 + return undefined; 261 + } 262 + return res.map(x => ({...x, play: hydratePlaySelect(x)})).find(x => { 263 + const temporalComparison = comparePlayTemporally(x.play, play); 264 + return hasAcceptableTemporalAccuracy(temporalComparison.match, taAccuracy) 265 + }) 266 + } 267 + 268 + public getTemporallyClosePlays = async (play: PlayObject, opts: {states?: PlaySelect['state'][], bufferTime?: number} & ComponentConstrainedRepoOpts = {}): Promise<PlayHistoricalSelect[]> => { 269 + const { 270 + componentId = this.componentId, 271 + bufferTime, 272 + states 273 + } = opts; 274 + 275 + let query: FindMany<'plays'> = {}; 276 + 277 + let where: FindWhere<'plays'> = { 278 + componentId, 279 + playedAt: buildDateCompare(getTemporallyCloseDateCompareOp(play, {bufferTime})), 280 + }; 281 + if(states !== undefined) { 282 + where.state = { 283 + in: states 284 + } 285 + } 286 + query.where = where; 287 + 288 + return ((await this.db.query.plays.findMany({ 289 + where, 290 + })) as PlayHistoricalSelect[]).map(x => ({...x, play: hydratePlaySelect(x)})); 291 + } 292 + } 293 + 294 + export const buildPlayHistoricalWhere = (args: PlayWhereOpts): WhereClause<'playsHistorical'> => { 295 + // old way 296 + // let where: Parameters<(ReturnType<typeof getDb>)['query']['plays']['findMany']>[0]['where'] = { 297 + // }; 298 + let where: FindWhere<'playsHistorical'> = { 299 + componentId: args.componentId 300 + }; 301 + if (args.seenAt !== undefined) { 302 + where.seenAt = buildDateCompare(args.seenAt); 303 + } 304 + if (args.playedAt !== undefined) { 305 + where.playedAt = buildDateCompare(args.playedAt); 306 + } 307 + if(args.uid !== undefined) { 308 + where.uid = { 309 + in: args.uid 310 + } 311 + } 312 + return where; 313 + } 314 + 315 + export const playToRepositoryCreatePlayHistoricalOpts = (data: MarkOptional<RepositoryCreatePlayHistoricalOpts, 'componentId'>): RepositoryCreatePlayHistoricalOpts => { 316 + const { 317 + play: { 318 + meta: { 319 + lifecycle, 320 + ...metaRest 321 + }, 322 + ...playRest 323 + }, 324 + ...rest 325 + } = data; 326 + 327 + return { 328 + play: { 329 + ...playRest, 330 + meta: { 331 + ...metaRest, 332 + // @ts-expect-error 333 + lifecycle: { 334 + } 335 + } 336 + }, 337 + uid: data.play.meta?.playId, 338 + ...rest 339 + } 340 + }

+5

src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts

··· 80 80 } 81 81 } 82 82 83 + async getCAR() { 84 + // wish there was a way stream this... 85 + return await this.agent.com.atproto.sync.getRepo({did: this.agent.sessionManager.did}); 86 + } 87 + 83 88 async getPagelessTimeRangeListens(params: PagelessListensTimeRangeOptions): Promise<PagelessTimeRangeListensResult> { 84 89 const {to, limit} = params; 85 90

+40 -15

src/backend/scrobblers/AbstractHistoricalScrobbleClient.ts

··· 3 3 import AbstractScrobbleClient from "./AbstractScrobbleClient.js"; 4 4 import { ComponentMigrationSelect } from "../common/database/drizzle/drizzleTypes.js"; 5 5 import { ErrorIsh } from "../../core/ErrorUtils.js"; 6 + import { DrizzlePlayHistoricalRepository } from "../common/database/drizzle/repositories/PlayHistoricalRepository.js"; 7 + import { spawn, isAbortError } from 'abort-controller-x'; 8 + import { generateLoggableAbortReason } from "../common/errors/MSErrors.js"; 6 9 7 10 export default abstract class AbstractHistoricalScrobbleClient extends AbstractScrobbleClient { 8 11 9 12 protected importAbortController: AbortController | undefined; 10 13 protected importPromise: Promise<void> | undefined; 14 + protected playsHistoricalRepo!: DrizzlePlayHistoricalRepository; 11 15 lastImport?: Dayjs; 12 16 lastImportSuccess?: Dayjs; 13 17 synced: boolean; 14 18 syncedReason?: string; 15 19 syncError?: ErrorIsh; 16 20 17 - protected abstract doHydrateHistoricalScrobbles(): Promise<void>; 21 + protected abstract doHydrateHistoricalScrobbles(opts: {allowFailures?: boolean, signal?: AbortSignal }): Promise<void>; 18 22 19 - protected async hydrateHistoricalScrobbles(): Promise<void> { 20 - const newImport: ComponentMigrationSelect = await this.migrationRepo.create({name: 'historicalImport', componentId: this.dbComponent.id}) as ComponentMigrationSelect; 21 - try { 22 - await this.doHydrateHistoricalScrobbles(); 23 - await this.migrationRepo.updateById(newImport.id, {success: true}); 24 - this.synced = true; 25 - this.lastImportSuccess = dayjs(); 26 - } catch (e) { 27 - await this.migrationRepo.updateById(newImport.id, {success: false, error: e}); 28 - this.syncError = e; 29 - this.synced = false; 30 - } finally { 31 - this.lastImport = dayjs(); 23 + hydrateHistoricalScrobbles(allowFailures: boolean = false): void { 24 + if(this.importAbortController !== undefined) { 25 + throw new Error('Cannot start a new import while one is already running'); 32 26 } 33 - this.dbComponent.migrations.push(newImport); 27 + this.importAbortController = new AbortController(); 28 + this.importPromise = spawn(this.importAbortController.signal, async (signal, {defer, fork}) => { 29 + 30 + defer(async () => { 31 + this.importAbortController = undefined; 32 + this.importPromise = undefined; 33 + }); 34 + 35 + const newImport: ComponentMigrationSelect = await this.migrationRepo.create({name: 'historicalImport', componentId: this.dbComponent.id}) as ComponentMigrationSelect; 36 + try { 37 + await this.doHydrateHistoricalScrobbles({signal, allowFailures}); 38 + await this.migrationRepo.updateById(newImport.id, {success: true}); 39 + this.synced = true; 40 + this.lastImportSuccess = dayjs(); 41 + } catch (e) { 42 + await this.migrationRepo.updateById(newImport.id, {success: false, error: e}); 43 + this.syncError = e; 44 + this.synced = false; 45 + } finally { 46 + this.lastImport = dayjs(); 47 + } 48 + this.dbComponent.migrations.push(newImport); 49 + }).catch((e) => { 50 + if (isAbortError(e)) { 51 + const err = generateLoggableAbortReason('Import processing stopped', this.importAbortController.signal); 52 + this.logger.info(err); 53 + this.logger.trace(e) 54 + } else { 55 + this.logger.warn(new Error('Uncaught error during import processing', { cause: e })); 56 + } 57 + }); 34 58 } 35 59 36 60 async getHistoricalScrobblesAreSynced(): Promise<[boolean, string?]> { ··· 53 77 54 78 protected async postDatabase(): Promise<void> { 55 79 await super.postDatabase(); 80 + this.playsHistoricalRepo = new DrizzlePlayHistoricalRepository(this.db, {componentId: this.dbComponent.id, logger: this.logger}); 56 81 const imports = this.dbComponent.migrations.filter(x => x.name === 'historicalImport'); 57 82 const [synced, reason] = await this.getHistoricalScrobblesAreSynced(); 58 83 this.synced = synced;

+1 -1

src/backend/scrobblers/ScrobbleClients.ts

··· 464 464 break; 465 465 case 'tealfm': 466 466 const TealScrobbler = (await import('./TealfmScrobbler.js')).default; 467 - newClient = new TealScrobbler(name, {...clientConfig, data: d, options: compositeOptions} as unknown as TealClientConfig, {}, notifier, this.emitter, this.logger); 467 + newClient = new TealScrobbler(name, {...clientConfig, data: d, options: compositeOptions} as unknown as TealClientConfig, this.internalConfig, notifier, this.emitter, this.logger); 468 468 break; 469 469 case 'rocksky': 470 470 const RockskyScrobbler = (await import('./RockskyScrobbler.js')).default;

+169 -6

src/backend/scrobblers/TealfmScrobbler.ts

··· 1 - import { Logger, LogLevel } from "@foxxmd/logging"; 1 + import { childLogger, Logger, LogLevel } from "@foxxmd/logging"; 2 2 import EventEmitter from "events"; 3 + import fsPromise from 'node:fs/promises'; 4 + import fs from 'node:fs'; 5 + import path from 'path'; 6 + 7 + import { Readable } from 'stream'; 3 8 import { PlayObject, SourcePlayerObj } from "../../core/Atomic.js"; 4 9 import { buildTrackString, capitalize } from "../../core/StringUtils.js"; 5 10 import { isNodeNetworkException } from "../common/errors/NodeErrors.js"; 6 - import { FormatPlayObjectOptions, CALCULATED_PLAYER_STATUSES, ReportedPlayerStatus } from "../common/infrastructure/Atomic.js"; 11 + import { FormatPlayObjectOptions, CALCULATED_PLAYER_STATUSES, ReportedPlayerStatus, InternalConfigOptional } from "../common/infrastructure/Atomic.js"; 7 12 import { playToListenPayload } from '../common/vendor/listenbrainz/lzUtils.js'; 8 13 import { Notifiers } from "../notifier/Notifiers.js"; 9 14 ··· 11 16 import { TealClientConfig } from "../common/infrastructure/config/client/tealfm.js"; 12 17 import { BlueSkyAppApiClient } from "../common/vendor/bluesky/BlueSkyAppApiClient.js"; 13 18 import { BlueSkyOauthApiClient } from "../common/vendor/bluesky/BlueSkyOauthApiClient.js"; 14 - import { AbstractBlueSkyApiClient, listRecordToPlay, nowPlayingExpirationDuration, playToRecord, playToStatusRecord, recordToPlay } from "../common/vendor/bluesky/AbstractBlueSkyApiClient.js"; 19 + import { AbstractBlueSkyApiClient, nowPlayingExpirationDuration, playToRecord, playToStatusRecord, recordToPlay } from "../common/vendor/bluesky/AbstractBlueSkyApiClient.js"; 15 20 import dayjs, { Dayjs } from "dayjs"; 16 21 import { durationToHuman } from "../utils.js"; 22 + import AbstractHistoricalScrobbleClient from "./AbstractHistoricalScrobbleClient.js"; 23 + import dayjs from "dayjs"; 24 + import { fromStream } from '@atcute/repo'; 25 + import { playToRepositoryCreatePlayHistoricalOpts, RepositoryCreatePlayHistoricalOpts } from "../common/database/drizzle/repositories/PlayHistoricalRepository.js"; 26 + import { durationToHuman, isDebugMode } from "../utils.js"; 27 + import { isAbortError } from "abort-controller-x"; 17 28 18 - export default class TealScrobbler extends AbstractScrobbleClient { 29 + export default class TealScrobbler extends AbstractHistoricalScrobbleClient { 19 30 20 31 requiresAuth = true; 21 32 requiresAuthInteraction = false; ··· 24 35 25 36 declare config: TealClientConfig; 26 37 38 + protected configDir: string; 39 + 27 40 client: AbstractBlueSkyApiClient; 28 41 29 - constructor(name: any, config: TealClientConfig, options = {}, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { 42 + constructor(name: any, config: TealClientConfig, options: InternalConfigOptional & {[key: string]: any}, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { 30 43 super('tealfm', name, config, notifier, emitter, logger); 31 44 this.MAX_INITIAL_SCROBBLES_FETCH = 20; 32 45 this.scrobbleDelay = 1500; ··· 41 54 } 42 55 this.nowPlayingMaxThreshold = nowPlayingUpdateByPlayDuration; 43 56 this.nowPlayingMinThreshold = (_) => 20; 57 + this.configDir = options.configDir; 44 58 } 45 59 46 60 formatPlayObj = (obj: any, options: FormatPlayObjectOptions = {}) => recordToPlay(obj); ··· 87 101 return true; 88 102 } 89 103 if(this.client instanceof BlueSkyAppApiClient) { 90 - return await this.client.appLogin(); 104 + const res = await this.client.appLogin(); 105 + return res; 91 106 } 92 107 } catch (e) { 93 108 if(isNodeNetworkException(e)) { ··· 162 177 return false; 163 178 } 164 179 return dayjs().isAfter(this.lastExpirationDate); 180 + } 181 + 182 + protected async doHydrateHistoricalScrobbles(opts: {allowFailures?: boolean, signal?: AbortSignal } = {}) { 183 + const { 184 + allowFailures = false, 185 + signal 186 + } = opts; 187 + let file: string; 188 + try { 189 + file = await this.fetchCarToFile(); 190 + signal?.throwIfAborted(); 191 + } catch (e) { 192 + throw new Error('Failed to fetch CAR repo file', {cause: e}); 193 + } 194 + 195 + try { 196 + await this.parseScrobblesFromCar(file, 100, {allowFailures, logger: childLogger(this.logger, ['Historical Plays']), signal}); 197 + } catch (e) { 198 + throw new Error('Failed to convert CAR without any error', {cause: e}); 199 + } finally { 200 + await fsPromise.rm(file); 201 + } 202 + } 203 + 204 + async fetchCarToFile() { 205 + const filename = path.resolve(this.configDir, `${this.getSafeExternalId()}-${dayjs().unix()}.car`); 206 + await fsPromise.writeFile(filename, Buffer.from(((await this.client.getCAR()).data))); 207 + return filename; 208 + } 209 + 210 + async parseScrobblesFromCar(filename: string, batchSize: number, opts: {allowFailures?: boolean, logger?: Logger, signal?: AbortSignal} = {}) { 211 + 212 + const { 213 + allowFailures = false, 214 + logger = this.logger, 215 + signal 216 + } = opts; 217 + 218 + const stream = Readable.toWeb(fs.createReadStream(filename)); 219 + 220 + await using repo = fromStream(stream); 221 + 222 + const did = this.client?.agent?.sessionManager?.did; 223 + 224 + let batch: RepositoryCreatePlayHistoricalOpts[] = []; 225 + let allGood = true; 226 + let count = 0; 227 + let persisted = 0; 228 + const start = dayjs(); 229 + 230 + logger.info('Starting CAR conversion to historical plays...'); 231 + 232 + for await (const entry of repo) { 233 + if(entry.collection === 'fm.teal.alpha.feed.play') { 234 + let play: PlayObject; 235 + try { 236 + play = recordToPlay(entry.record as ScrobbleRecord, { 237 + web: did !== undefined ? `at://did:plc:${did}/fm.teal.alpha.feed.play/${entry.rkey}` : undefined, 238 + playId: entry.rkey, 239 + user: did 240 + }); 241 + if(isDebugMode()) { 242 + logger.trace(`(${count}) rKey ${entry.rkey} => ${buildTrackString(play)}`); 243 + } 244 + count++; 245 + if(count % (batchSize * 5) === 0) { 246 + logger.debug(`Processed ${count} records`); 247 + signal?.throwIfAborted(); 248 + } 249 + } catch (e) { 250 + if(isAbortError(e)) { 251 + throw e; 252 + } 253 + if(allowFailures) { 254 + this.logger.warn(new Error(`Failed to convert record ${entry.rkey} to Play but will continue`, {cause: e})); 255 + continue; 256 + } else { 257 + throw new Error(`Failed to convert record ${entry.rkey} to Play`, {cause: e}); 258 + } 259 + } 260 + 261 + const existing = await this.playsHistoricalRepo.hasByUid(entry.rkey); 262 + if(!existing) { 263 + batch.push(playToRepositoryCreatePlayHistoricalOpts({play})); 264 + } 265 + if(batch.length >= batchSize) { 266 + try { 267 + const [res, valid] = await this.createHistoricalPlays(batch, opts); 268 + persisted += valid; 269 + if(!res) { 270 + allGood = false; 271 + } 272 + } catch (e) { 273 + throw e; 274 + } 275 + batch = []; 276 + } 277 + } 278 + } 279 + 280 + logger.debug('Reached end of CAR file'); 281 + if(batch.length > 0) { 282 + logger.debug(`Persisting remaining ${batch.length} records...`); 283 + try { 284 + const [res, valid] = await this.createHistoricalPlays(batch, opts); 285 + persisted += valid; 286 + if(!res) { 287 + allGood = false; 288 + } 289 + } catch (e) { 290 + throw e; 291 + } 292 + } 293 + logger.info(`Completed CAR conversion: Result ${allGood ? 'OK' : 'Some Errors'} in ${durationToHuman(dayjs.duration(dayjs().diff(start)))} | Records ${count} | Persisted ${persisted}`) 294 + } 295 + 296 + async createHistoricalPlays(batch: RepositoryCreatePlayHistoricalOpts[], opts: {allowFailures?: boolean, logger?: Logger, signal?: AbortSignal} = {}): Promise<[boolean, number]> { 297 + const { 298 + allowFailures = false, 299 + logger = this.logger, 300 + signal 301 + } = opts; 302 + try { 303 + await this.playsHistoricalRepo.createPlays(batch); 304 + return [true, batch.length]; 305 + } catch (e) { 306 + logger.warn(`Failed to persist batch of ${batch} plays, trying individually...`); 307 + } 308 + signal?.throwIfAborted(); 309 + 310 + let valid = 0; 311 + for(const p of batch) { 312 + try { 313 + await this.playsHistoricalRepo.createPlays([p]); 314 + valid++; 315 + } catch (e) { 316 + if(allowFailures) { 317 + logger.warn(p.play,`Failed to persist play from record with rKey ${p.play.meta.playId} => ${buildTrackString(p.play)}`); 318 + logger.warn(e); 319 + } else { 320 + logger.error(p.play,`Failed to persist play from record with rKey ${p.play.meta.playId} => ${buildTrackString(p.play)}`); 321 + throw e; 322 + } 323 + } 324 + signal?.throwIfAborted(); 325 + } 326 + 327 + return [false, valid]; 165 328 } 166 329 } 167 330

+14

src/backend/server/api.ts

··· 37 37 import { SimpleError } from "../common/errors/MSErrors.js"; 38 38 import { QueryPlaysOpts } from "../common/database/drizzle/repositories/PlayRepository.js"; 39 39 import { playSelectToDeadScrobble } from "../common/database/drizzle/entityUtils.js"; 40 + import AbstractHistoricalScrobbleClient from "../scrobblers/AbstractHistoricalScrobbleClient.js"; 40 41 41 42 const maxBufferSize = 300; 42 43 const output: Record<number, FixedSizeList<LogDataPretty>> = {}; ··· 525 526 client.logger.info('Trying to start scrobbler...'); 526 527 client.initScrobbleMonitoring({force, notify: false}).catch(e => client.logger.error(e)); 527 528 res.status(200).send('OK'); 529 + }); 530 + 531 + app.post('/api/client/historical', clientRequiredMiddle, async (req, res) => { 532 + // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message 533 + const client = req.scrobbleClient as AbstractScrobbleClient; 534 + if(client instanceof AbstractHistoricalScrobbleClient) { 535 + client.logger.info('User requested historical play hydration'); 536 + client.hydrateHistoricalScrobbles(); 537 + res.status(200).send('OK'); 538 + } else { 539 + client.logger.warn('This client does not have historical play capabilities'); 540 + return res.status(400).json({error: 'This client does not have historical play capabilities'}); 541 + } 528 542 }); 529 543 530 544 app.get('/health', async (req, res) => res.redirect(307, `/api/${req.url.slice(1)}`));

+48

src/backend/tests/tealfm/tealfm.test.ts

··· 5 5 import { listRecordToPlay, playToRecord } from '../../common/vendor/bluesky/AbstractBlueSkyApiClient.js'; 6 6 import dayjs from 'dayjs'; 7 7 import { artistCreditsToNames } from '../../../core/StringUtils.js'; 8 + import TealScrobbler from '../../scrobblers/TealfmScrobbler.js'; 9 + import { Notifiers } from '../../notifier/Notifiers.js'; 10 + import { EventEmitter } from "events"; 11 + import { loggerNoop } from '../../common/MaybeLogger.js'; 12 + import path from 'node:path'; 13 + import { configDir } from '../../common/index.js'; 14 + import { loggerDebug, loggerTrace } from '@foxxmd/logging'; 8 15 9 16 chai.use(asPromised); 10 17 ··· 71 78 expect(record.artists[0].artistMbId).eq(`mbid:${play.data.artists[0].mbid}`); 72 79 }); 73 80 81 + }); 82 + 83 + describe('#tealfm Play To Record', function () { 84 + 85 + it('Adds mbids with uri format', function () { 86 + 87 + const play = withBrainz(generatePlay({artists: generateArtistCredits(2)}), {include: ['recording']}); 88 + const record = playToRecord(play); 89 + 90 + expect(record.recordingMbId).to.eq(`mbid:${play.data.meta.brainz.recording}`); 91 + expect(record.releaseMbId).is.undefined; 92 + expect(record.artists).length(2); 93 + expect(record.artists[0].artistName).eq(play.data.artists[0].name); 94 + expect(record.artists[0].artistMbId).eq(`mbid:${play.data.artists[0].mbid}`); 95 + }); 96 + 97 + }); 98 + 99 + describe('#tealfmCar', function() { 100 + 101 + before(function () { 102 + if (process.env.TEAL_CAR_TEST !== 'true') { 103 + this.skip(); 104 + } 105 + }); 106 + 107 + it('Parses car file', async function() { 108 + 109 + this.timeout(100000); 110 + 111 + const tfm = new TealScrobbler('test', 112 + {name: 'test', data: {identifier: 'test', appPassword: 'test'}}, 113 + {configDir: 'test', localUrl: new URL('https://example.com'), version: 'test'}, 114 + new Notifiers(new EventEmitter(), new EventEmitter(), new EventEmitter(), loggerNoop), 115 + new EventEmitter(), 116 + loggerDebug 117 + ); 118 + await tfm.buildDatabase(); 119 + 120 + await tfm.parseScrobblesFromCar(path.resolve(configDir, 'tealfm-myteal-1778870858.car'), 100); 121 + }); 74 122 });

Configure Feed

Configure Feed