diff --git a/src/backend/common/database/drizzle/drizzleTypes.ts b/src/backend/common/database/drizzle/drizzleTypes.ts index 8872023d..8d7f7abe 100644 --- a/src/backend/common/database/drizzle/drizzleTypes.ts +++ b/src/backend/common/database/drizzle/drizzleTypes.ts @@ -1,5 +1,5 @@ import { DBQueryConfig, DBQueryConfigWith, ExtractTablesFromSchema, KnownKeysOnly, RelationFieldsFilterInternals, Many, InferSelectModel, ExtractTablesWithRelations, type BuildQueryResult, RelationsFilter } from "drizzle-orm"; -import { components, componentMigrations, playInputs, plays, queueStates, relations } from "./schema/schema.js"; +import { components, componentMigrations, playInputs, plays, queueStates, relations, playsHistorical } from "./schema/schema.js"; import {TSchema, TableName, Schema } from "./schema/schema.js"; import { MarkOptional, MarkRequired } from "ts-essentials"; @@ -21,6 +21,8 @@ export type PlaySelectWithQueueStates = GenericRelationResult<'plays', 'queueSta export type PlayWith = GenericRelationResult<'plays', K>; export type PlayNew = typeof plays.$inferInsert; +export type PlayHistoricalSelect = typeof playsHistorical.$inferSelect; +export type PlayHistoricalNew = typeof playsHistorical.$inferInsert; // useful references for building types // https://github.com/drizzle-team/drizzle-orm/discussions/2596 diff --git a/src/backend/common/database/drizzle/entityUtils.ts b/src/backend/common/database/drizzle/entityUtils.ts index 9d4241fb..d5833e2f 100644 --- a/src/backend/common/database/drizzle/entityUtils.ts +++ b/src/backend/common/database/drizzle/entityUtils.ts @@ -1,5 +1,5 @@ import assert from "node:assert"; -import { PlayNew, PlaySelect, PlaySelectWithQueueStates } from "./drizzleTypes.js"; +import { PlayHistoricalNew, PlayHistoricalSelect, PlayNew, PlaySelect, PlaySelectWithQueueStates } from "./drizzleTypes.js"; import { PlayInputNew } from "./drizzleTypes.js"; import { QueueStateNew } from "./drizzleTypes.js"; import { ComponentNew } from "./drizzleTypes.js"; @@ -20,6 +20,7 @@ export const generateComponentEntity = (data: MarkOptional) } export type PlayEntityOpts = Partial> & { error?: ErrorLike }; +export type PlayHistoricalEntityOpts = Partial>; export const generatePlayEntity = (play: PlayObject, opts: PlayEntityOpts = {}): PlayNew => { const { @@ -51,7 +52,7 @@ export const generatePlayEntity = (play: PlayObject, opts: PlayEntityOpts = {}): export type PlayHydateOptions = 'asPlay' | 'id' | 'uid'; -export const hydratePlaySelect = (select: PlaySelect, opts: PlayHydateOptions[] = ['id','uid']): PlayObject => { +export const hydratePlaySelect = (select: T, opts: PlayHydateOptions[] = ['id','uid']): PlayObject => { if(opts.length === 0) { return select.play; } diff --git a/src/backend/common/database/drizzle/repositories/PlayHistoricalRepository.ts b/src/backend/common/database/drizzle/repositories/PlayHistoricalRepository.ts new file mode 100644 index 00000000..fef147f5 --- /dev/null +++ b/src/backend/common/database/drizzle/repositories/PlayHistoricalRepository.ts @@ -0,0 +1,340 @@ +import { childLogger, Logger, LoggerAppExtras } from "@foxxmd/logging"; +import { DbConcrete, runTransaction } from "../drizzleUtils.js"; +import { loggerNoop } from "../../../MaybeLogger.js"; +import { ErrorLike, PlayObject, TA_CLOSE, TA_DEFAULT_ACCURACY, TA_EXACT, TemporalAccuracy } from "../../../../../core/Atomic.js"; +import { generateInputEntity, generatePlayEntity, PlayEntityOpts, hydratePlaySelect, PlayHydateOptions, PlayHistoricalEntityOpts } from "../entityUtils.js"; +import { playInputs, plays, playsHistorical, queueStates, relations } from "../schema/schema.js"; +import { PlayNew, PlaySelect, PlayInputNew, FindWhere, FindMany, QueueStateSelect, FindWith, PlaySelectWithQueueStates, WhereClause, PlayWith, PlayHistoricalSelect, PlayHistoricalNew } from "../drizzleTypes.js";; +import { MarkOptional, MarkRequired, PathValue } from "ts-essentials"; +import { genGroupIdStrFromPlay, removeEmptyArrays, removeUndefinedKeys } from "../../../../utils.js"; +import dayjs, { Dayjs } from "dayjs"; +import { RelationsFieldFilter, eq, inArray, ne, notInArray, desc, asc, and, sql, Placeholder } from "drizzle-orm"; +import { CompactableProperty, RetentionOptions, retentionPlayTypes } from "../../../infrastructure/config/database.js"; +import { shortTodayAwareFormat } from "../../../../../core/TimeUtils.js"; +import { buildDateCompare, CompareDateOp, ComponentConstrainedRepoOpts, DrizzleBaseRepository, DrizzleRepositoryOpts, PaginatedQueryResponse, PaginatedResponse } from "./BaseRepository.js"; +import { asPlay } from "../../../../../core/PlayMarshalUtils.js"; +import assert, { Assert } from "node:assert"; +import { hashObject, parseArrayFromMaybeString } from "../../../../utils/StringUtils.js"; +import { playContentBasicInvariantTransform, playMbidIdentifier } from "../../../../utils/PlayComparisonUtils.js"; +import { comparePlayTemporally, getScrobbleTsSOCDate, getScrobbleTsSOCDateWithContext, getTemporalAccuracyCloseVal, hasAcceptableTemporalAccuracy } from "../../../../utils/TimeUtils.js"; +import { SourceType } from "../../../infrastructure/config/source/sources.js"; +import { getTemporallyCloseDateCompareOp } from "./PlayRepository.js"; + +// https://github.com/drizzle-team/drizzle-orm/issues/695 may be useful for typing models with relations? + +export interface PlayWhereOpts { + componentId?: number + seenAt?: CompareDateOp + playedAt?: CompareDateOp + uid?: string[] +} + +export interface QueryPlaysOpts extends PlayWhereOpts { + sort?: 'seenAt' | 'playedAt' + order?: 'asc' | 'desc' + limit?: number + offset?: number +} + +export interface HydrateOpts { + hydrate?: PlayHydateOptions[] +} + +export type RepositoryCreatePlayHistoricalOpts = PlayHistoricalEntityOpts + & Pick; + +type PlayIdentifierPrimitiveMap = { + uid: string; + id: number; +}; + +const identifierExtractor: { [K in keyof PlayIdentifierPrimitiveMap]: (play: {id: number, uid: string}) => PlayIdentifierPrimitiveMap[K] } = { + id: (play) => play.id, + uid: (play) => play.uid, +}; +export class DrizzlePlayHistoricalRepository extends DrizzleBaseRepository<'playsHistorical'> { + + constructor(db: DbConcrete, opts: DrizzleRepositoryOpts = {}) { + super(db, 'plays', 'Plays', opts); + } + + findByUid = async (uid: string, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise => { + const res = await this.db.query.playsHistorical.findFirst({ + where: { + uid, + componentId: opts.componentId ?? this.componentId + } + }); + res.play = hydratePlaySelect(res, opts.hydrate); + return res; + } + + hasByUid = async (uid: string, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise => { + const res = await this.db.query.playsHistorical.findFirst({ + columns: {id: true}, + where: { + uid, + componentId: opts.componentId ?? this.componentId + } + }); + return res !== undefined; + } + + createPlays = async (entitiesOpts: RepositoryCreatePlayHistoricalOpts[], opts: HydrateOpts = {}) => { + + const { + hydrate + } = opts; + let playRows: PlayHistoricalSelect[]; + + await runTransaction(this.db, async () => { + + const entitiesData = entitiesOpts.map((data) => { + const { + play, + ...rest + } = data; + return generatePlayEntity(play, { componentId: this.componentId, ...rest }); + }); + + playRows = await this.db.insert(playsHistorical).values(entitiesData).returning(); + }); + + return playRows.map(x => ({...x, play: hydratePlaySelect(x, hydrate)})); + } + + findPlays = async (args: QueryPlaysOpts, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise => { + const { + hydrate, + componentId = this.componentId + } = opts; + // this does not work as type for query variable + // it erases the result type for some reason + // + // Parameters[0] + + // this does work but it is also integrated into FindWith + //let withQuery: Parameters[0]['with'] = undefined; + + let query: FindMany<'playsHistorical'> = { + limit: args.limit, + offset: args.offset + }; + + query.where = buildPlayHistoricalWhere({componentId: componentId, ...args}); + + if (args.sort !== undefined) { + query.orderBy = { + [args.sort]: args.order ?? 'desc' + } + } else { + query.orderBy = { + id: 'asc' + } + } + + query = removeUndefinedKeys(query); + const results = await this.db.query.playsHistorical.findMany(query); + return results.map((x) => ({...x, play: hydratePlaySelect(x, hydrate)})); + } + + findPlayIds = async (args: QueryPlaysOpts, opts: ComponentConstrainedRepoOpts = {}): Promise => { + const { + componentId = this.componentId + } = opts; + + let query: FindMany<'playsHistorical'> = { + limit: args.limit, + offset: args.offset, + }; + + query.where = buildPlayHistoricalWhere({componentId: componentId, ...args}); + + if (args.sort !== undefined) { + query.orderBy = { + [args.sort]: args.order ?? 'desc' + } + } else { + query.orderBy = { + id: 'asc' + } + } + + query = removeUndefinedKeys(query); + const results = await this.db.query.plays.findMany({ + limit: args.limit, + offset: args.offset, + columns: {id: true}, + orderBy: args.sort !== undefined ? {[args.sort]: args.order ?? 'desc'} : {id: 'asc'}, + }); + return results.map((x) => x.id); + } + + findPlayIdentifiers = async (args: QueryPlaysOpts, identifier: T, opts: ComponentConstrainedRepoOpts = {}): Promise => { + const { + componentId = this.componentId, + } = opts; + + const results = await this.db.query.playsHistorical.findMany({ + limit: args.limit, + offset: args.offset, + columns: {id: true, uid: true}, + orderBy: args.sort !== undefined ? {[args.sort]: args.order ?? 'desc'} : {id: 'asc'}, + where: buildPlayHistoricalWhere({componentId: componentId, ...args}) + }); + + // we getting fancy now + return results.map(identifierExtractor[identifier]); + } + + findPlaysPaginated = async (args: QueryPlaysOpts, opts: HydrateOpts & ComponentConstrainedRepoOpts = {}): Promise> => { + const { + limit = 100, + offset = 0, + ...rest + } = args; + const clampedLimit = Math.min(limit, 100); + const res = await this.findPlays({limit: clampedLimit, offset, ...rest}, opts); + return {data: res, meta: {limit: clampedLimit, offset}}; + } + + // async updateById(id: number, data: Partial): Promise { + // if(data.play !== undefined) { + // data.play = withoutDbAwareness(data.play); + // } + // super.updateById(id, data); + // } + + deletePlays = async (playsData: (Pick | number)[]) => { + const ids = playsData.map(x => typeof x === 'number' ? x : x.id); + await this.db.delete(playsHistorical).where(inArray(plays.id, ids)); + } + + public checkExisting = async (play: PlayObject, opts: {taAccuracy?: TemporalAccuracy[]} & ComponentConstrainedRepoOpts = {}): Promise => { + const { + componentId = this.componentId, + taAccuracy = TA_DEFAULT_ACCURACY, + } = opts; + const hash = hashObject(playContentBasicInvariantTransform(play).data); + + // we get all plays with a play date between playdate - (source accuracy) AND (playDateCompleted or playDate) + (source accuracy) + // which we can then use with temporal comparison to make sure we are comparing the correct dates + // + // this isn't as fast as just comparing playDate directly but its still much faster/cheaper than paginating plays and doing everything in-memory + const dateGranularity = getTemporalAccuracyCloseVal(play.meta.source as SourceType); + let endRange: Dayjs; + if(play.data.playDateCompleted !== undefined) { + // this will be present if source reports it + // or we tracked it live with MemorySource + endRange = play.data.playDateCompleted.add(dateGranularity, 's'); + } else { + endRange = play.data.playDate.add(dateGranularity, 's'); + } + let where: FindWhere<'playsHistorical'> = { + componentId, + playedAt: buildDateCompare(getTemporallyCloseDateCompareOp(play)), + }; + + const mbidId = playMbidIdentifier(play); + if(mbidId !== undefined) { + where.AND = [ + { + OR: [ + { + playHash: hash + }, + { + mbidIdentifier: mbidId + } + ] + } + ] + } else { + where.playHash = hash; + } + + const res = await this.db.query.playsHistorical.findMany({ + where + }); + if(res.length === 0) { + return undefined; + } + return res.map(x => ({...x, play: hydratePlaySelect(x)})).find(x => { + const temporalComparison = comparePlayTemporally(x.play, play); + return hasAcceptableTemporalAccuracy(temporalComparison.match, taAccuracy) + }) + } + + public getTemporallyClosePlays = async (play: PlayObject, opts: {states?: PlaySelect['state'][], bufferTime?: number} & ComponentConstrainedRepoOpts = {}): Promise => { + const { + componentId = this.componentId, + bufferTime, + states + } = opts; + + let query: FindMany<'plays'> = {}; + + let where: FindWhere<'plays'> = { + componentId, + playedAt: buildDateCompare(getTemporallyCloseDateCompareOp(play, {bufferTime})), + }; + if(states !== undefined) { + where.state = { + in: states + } + } + query.where = where; + + return ((await this.db.query.plays.findMany({ + where, + })) as PlayHistoricalSelect[]).map(x => ({...x, play: hydratePlaySelect(x)})); + } +} + +export const buildPlayHistoricalWhere = (args: PlayWhereOpts): WhereClause<'playsHistorical'> => { + // old way + // let where: Parameters<(ReturnType)['query']['plays']['findMany']>[0]['where'] = { + // }; + let where: FindWhere<'playsHistorical'> = { + componentId: args.componentId + }; + if (args.seenAt !== undefined) { + where.seenAt = buildDateCompare(args.seenAt); + } + if (args.playedAt !== undefined) { + where.playedAt = buildDateCompare(args.playedAt); + } + if(args.uid !== undefined) { + where.uid = { + in: args.uid + } + } + return where; +} + +export const playToRepositoryCreatePlayHistoricalOpts = (data: MarkOptional): RepositoryCreatePlayHistoricalOpts => { + const { + play: { + meta: { + lifecycle, + ...metaRest + }, + ...playRest + }, + ...rest + } = data; + + return { + play: { + ...playRest, + meta: { + ...metaRest, + // @ts-expect-error + lifecycle: { + } + } + }, + uid: data.play.meta?.playId, + ...rest + } +} \ No newline at end of file diff --git a/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts b/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts index ce4310b4..f9e0b476 100644 --- a/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts +++ b/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts @@ -80,6 +80,11 @@ export abstract class AbstractBlueSkyApiClient extends AbstractApiClient impleme } } + async getCAR() { + // wish there was a way stream this... + return await this.agent.com.atproto.sync.getRepo({did: this.agent.sessionManager.did}); + } + async getPagelessTimeRangeListens(params: PagelessListensTimeRangeOptions): Promise { const {to, limit} = params; diff --git a/src/backend/scrobblers/AbstractHistoricalScrobbleClient.ts b/src/backend/scrobblers/AbstractHistoricalScrobbleClient.ts index 3187081c..02df02e3 100644 --- a/src/backend/scrobblers/AbstractHistoricalScrobbleClient.ts +++ b/src/backend/scrobblers/AbstractHistoricalScrobbleClient.ts @@ -3,34 +3,58 @@ import { sortByNewestDate } from "../../core/PlayUtils.js"; import AbstractScrobbleClient from "./AbstractScrobbleClient.js"; import { ComponentMigrationSelect } from "../common/database/drizzle/drizzleTypes.js"; import { ErrorIsh } from "../../core/ErrorUtils.js"; +import { DrizzlePlayHistoricalRepository } from "../common/database/drizzle/repositories/PlayHistoricalRepository.js"; +import { spawn, isAbortError } from 'abort-controller-x'; +import { generateLoggableAbortReason } from "../common/errors/MSErrors.js"; export default abstract class AbstractHistoricalScrobbleClient extends AbstractScrobbleClient { protected importAbortController: AbortController | undefined; protected importPromise: Promise | undefined; + protected playsHistoricalRepo!: DrizzlePlayHistoricalRepository; lastImport?: Dayjs; lastImportSuccess?: Dayjs; synced: boolean; syncedReason?: string; syncError?: ErrorIsh; - protected abstract doHydrateHistoricalScrobbles(): Promise; + protected abstract doHydrateHistoricalScrobbles(opts: {allowFailures?: boolean, signal?: AbortSignal }): Promise; - protected async hydrateHistoricalScrobbles(): Promise { - const newImport: ComponentMigrationSelect = await this.migrationRepo.create({name: 'historicalImport', componentId: this.dbComponent.id}) as ComponentMigrationSelect; - try { - await this.doHydrateHistoricalScrobbles(); - await this.migrationRepo.updateById(newImport.id, {success: true}); - this.synced = true; - this.lastImportSuccess = dayjs(); - } catch (e) { - await this.migrationRepo.updateById(newImport.id, {success: false, error: e}); - this.syncError = e; - this.synced = false; - } finally { - this.lastImport = dayjs(); + hydrateHistoricalScrobbles(allowFailures: boolean = false): void { + if(this.importAbortController !== undefined) { + throw new Error('Cannot start a new import while one is already running'); } - this.dbComponent.migrations.push(newImport); + this.importAbortController = new AbortController(); + this.importPromise = spawn(this.importAbortController.signal, async (signal, {defer, fork}) => { + + defer(async () => { + this.importAbortController = undefined; + this.importPromise = undefined; + }); + + const newImport: ComponentMigrationSelect = await this.migrationRepo.create({name: 'historicalImport', componentId: this.dbComponent.id}) as ComponentMigrationSelect; + try { + await this.doHydrateHistoricalScrobbles({signal, allowFailures}); + await this.migrationRepo.updateById(newImport.id, {success: true}); + this.synced = true; + this.lastImportSuccess = dayjs(); + } catch (e) { + await this.migrationRepo.updateById(newImport.id, {success: false, error: e}); + this.syncError = e; + this.synced = false; + } finally { + this.lastImport = dayjs(); + } + this.dbComponent.migrations.push(newImport); + }).catch((e) => { + if (isAbortError(e)) { + const err = generateLoggableAbortReason('Import processing stopped', this.importAbortController.signal); + this.logger.info(err); + this.logger.trace(e) + } else { + this.logger.warn(new Error('Uncaught error during import processing', { cause: e })); + } + }); } async getHistoricalScrobblesAreSynced(): Promise<[boolean, string?]> { @@ -53,6 +77,7 @@ export default abstract class AbstractHistoricalScrobbleClient extends AbstractS protected async postDatabase(): Promise { await super.postDatabase(); + this.playsHistoricalRepo = new DrizzlePlayHistoricalRepository(this.db, {componentId: this.dbComponent.id, logger: this.logger}); const imports = this.dbComponent.migrations.filter(x => x.name === 'historicalImport'); const [synced, reason] = await this.getHistoricalScrobblesAreSynced(); this.synced = synced; diff --git a/src/backend/scrobblers/ScrobbleClients.ts b/src/backend/scrobblers/ScrobbleClients.ts index cc81b898..36a0daa7 100644 --- a/src/backend/scrobblers/ScrobbleClients.ts +++ b/src/backend/scrobblers/ScrobbleClients.ts @@ -464,7 +464,7 @@ ${sources.join('\n')}`); break; case 'tealfm': const TealScrobbler = (await import('./TealfmScrobbler.js')).default; - newClient = new TealScrobbler(name, {...clientConfig, data: d, options: compositeOptions} as unknown as TealClientConfig, {}, notifier, this.emitter, this.logger); + newClient = new TealScrobbler(name, {...clientConfig, data: d, options: compositeOptions} as unknown as TealClientConfig, this.internalConfig, notifier, this.emitter, this.logger); break; case 'rocksky': const RockskyScrobbler = (await import('./RockskyScrobbler.js')).default; diff --git a/src/backend/scrobblers/TealfmScrobbler.ts b/src/backend/scrobblers/TealfmScrobbler.ts index 698d6f07..433dddc1 100644 --- a/src/backend/scrobblers/TealfmScrobbler.ts +++ b/src/backend/scrobblers/TealfmScrobbler.ts @@ -1,9 +1,14 @@ -import { Logger, LogLevel } from "@foxxmd/logging"; +import { childLogger, Logger, LogLevel } from "@foxxmd/logging"; import EventEmitter from "events"; +import fsPromise from 'node:fs/promises'; +import fs from 'node:fs'; +import path from 'path'; + +import { Readable } from 'stream'; import { PlayObject, SourcePlayerObj } from "../../core/Atomic.js"; import { buildTrackString, capitalize } from "../../core/StringUtils.js"; import { isNodeNetworkException } from "../common/errors/NodeErrors.js"; -import { FormatPlayObjectOptions, CALCULATED_PLAYER_STATUSES, ReportedPlayerStatus } from "../common/infrastructure/Atomic.js"; +import { FormatPlayObjectOptions, CALCULATED_PLAYER_STATUSES, ReportedPlayerStatus, InternalConfigOptional } from "../common/infrastructure/Atomic.js"; import { playToListenPayload } from '../common/vendor/listenbrainz/lzUtils.js'; import { Notifiers } from "../notifier/Notifiers.js"; @@ -11,11 +16,17 @@ import AbstractScrobbleClient, { nowPlayingUpdateByPlayDuration, shouldClearNPSt import { TealClientConfig } from "../common/infrastructure/config/client/tealfm.js"; import { BlueSkyAppApiClient } from "../common/vendor/bluesky/BlueSkyAppApiClient.js"; import { BlueSkyOauthApiClient } from "../common/vendor/bluesky/BlueSkyOauthApiClient.js"; -import { AbstractBlueSkyApiClient, listRecordToPlay, nowPlayingExpirationDuration, playToRecord, playToStatusRecord, recordToPlay } from "../common/vendor/bluesky/AbstractBlueSkyApiClient.js"; +import { AbstractBlueSkyApiClient, nowPlayingExpirationDuration, playToRecord, playToStatusRecord, recordToPlay } from "../common/vendor/bluesky/AbstractBlueSkyApiClient.js"; import dayjs, { Dayjs } from "dayjs"; import { durationToHuman } from "../utils.js"; +import AbstractHistoricalScrobbleClient from "./AbstractHistoricalScrobbleClient.js"; +import dayjs from "dayjs"; +import { fromStream } from '@atcute/repo'; +import { playToRepositoryCreatePlayHistoricalOpts, RepositoryCreatePlayHistoricalOpts } from "../common/database/drizzle/repositories/PlayHistoricalRepository.js"; +import { durationToHuman, isDebugMode } from "../utils.js"; +import { isAbortError } from "abort-controller-x"; -export default class TealScrobbler extends AbstractScrobbleClient { +export default class TealScrobbler extends AbstractHistoricalScrobbleClient { requiresAuth = true; requiresAuthInteraction = false; @@ -24,9 +35,11 @@ export default class TealScrobbler extends AbstractScrobbleClient { declare config: TealClientConfig; + protected configDir: string; + client: AbstractBlueSkyApiClient; - constructor(name: any, config: TealClientConfig, options = {}, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { + constructor(name: any, config: TealClientConfig, options: InternalConfigOptional & {[key: string]: any}, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { super('tealfm', name, config, notifier, emitter, logger); this.MAX_INITIAL_SCROBBLES_FETCH = 20; this.scrobbleDelay = 1500; @@ -41,6 +54,7 @@ export default class TealScrobbler extends AbstractScrobbleClient { } this.nowPlayingMaxThreshold = nowPlayingUpdateByPlayDuration; this.nowPlayingMinThreshold = (_) => 20; + this.configDir = options.configDir; } formatPlayObj = (obj: any, options: FormatPlayObjectOptions = {}) => recordToPlay(obj); @@ -87,7 +101,8 @@ export default class TealScrobbler extends AbstractScrobbleClient { return true; } if(this.client instanceof BlueSkyAppApiClient) { - return await this.client.appLogin(); + const res = await this.client.appLogin(); + return res; } } catch (e) { if(isNodeNetworkException(e)) { @@ -163,5 +178,153 @@ export default class TealScrobbler extends AbstractScrobbleClient { } return dayjs().isAfter(this.lastExpirationDate); } + + protected async doHydrateHistoricalScrobbles(opts: {allowFailures?: boolean, signal?: AbortSignal } = {}) { + const { + allowFailures = false, + signal + } = opts; + let file: string; + try { + file = await this.fetchCarToFile(); + signal?.throwIfAborted(); + } catch (e) { + throw new Error('Failed to fetch CAR repo file', {cause: e}); + } + + try { + await this.parseScrobblesFromCar(file, 100, {allowFailures, logger: childLogger(this.logger, ['Historical Plays']), signal}); + } catch (e) { + throw new Error('Failed to convert CAR without any error', {cause: e}); + } finally { + await fsPromise.rm(file); + } + } + + async fetchCarToFile() { + const filename = path.resolve(this.configDir, `${this.getSafeExternalId()}-${dayjs().unix()}.car`); + await fsPromise.writeFile(filename, Buffer.from(((await this.client.getCAR()).data))); + return filename; + } + + async parseScrobblesFromCar(filename: string, batchSize: number, opts: {allowFailures?: boolean, logger?: Logger, signal?: AbortSignal} = {}) { + + const { + allowFailures = false, + logger = this.logger, + signal + } = opts; + + const stream = Readable.toWeb(fs.createReadStream(filename)); + + await using repo = fromStream(stream); + + const did = this.client?.agent?.sessionManager?.did; + + let batch: RepositoryCreatePlayHistoricalOpts[] = []; + let allGood = true; + let count = 0; + let persisted = 0; + const start = dayjs(); + + logger.info('Starting CAR conversion to historical plays...'); + + for await (const entry of repo) { + if(entry.collection === 'fm.teal.alpha.feed.play') { + let play: PlayObject; + try { + play = recordToPlay(entry.record as ScrobbleRecord, { + web: did !== undefined ? `at://did:plc:${did}/fm.teal.alpha.feed.play/${entry.rkey}` : undefined, + playId: entry.rkey, + user: did + }); + if(isDebugMode()) { + logger.trace(`(${count}) rKey ${entry.rkey} => ${buildTrackString(play)}`); + } + count++; + if(count % (batchSize * 5) === 0) { + logger.debug(`Processed ${count} records`); + signal?.throwIfAborted(); + } + } catch (e) { + if(isAbortError(e)) { + throw e; + } + if(allowFailures) { + this.logger.warn(new Error(`Failed to convert record ${entry.rkey} to Play but will continue`, {cause: e})); + continue; + } else { + throw new Error(`Failed to convert record ${entry.rkey} to Play`, {cause: e}); + } + } + + const existing = await this.playsHistoricalRepo.hasByUid(entry.rkey); + if(!existing) { + batch.push(playToRepositoryCreatePlayHistoricalOpts({play})); + } + if(batch.length >= batchSize) { + try { + const [res, valid] = await this.createHistoricalPlays(batch, opts); + persisted += valid; + if(!res) { + allGood = false; + } + } catch (e) { + throw e; + } + batch = []; + } + } + } + + logger.debug('Reached end of CAR file'); + if(batch.length > 0) { + logger.debug(`Persisting remaining ${batch.length} records...`); + try { + const [res, valid] = await this.createHistoricalPlays(batch, opts); + persisted += valid; + if(!res) { + allGood = false; + } + } catch (e) { + throw e; + } + } + logger.info(`Completed CAR conversion: Result ${allGood ? 'OK' : 'Some Errors'} in ${durationToHuman(dayjs.duration(dayjs().diff(start)))} | Records ${count} | Persisted ${persisted}`) + } + + async createHistoricalPlays(batch: RepositoryCreatePlayHistoricalOpts[], opts: {allowFailures?: boolean, logger?: Logger, signal?: AbortSignal} = {}): Promise<[boolean, number]> { + const { + allowFailures = false, + logger = this.logger, + signal + } = opts; + try { + await this.playsHistoricalRepo.createPlays(batch); + return [true, batch.length]; + } catch (e) { + logger.warn(`Failed to persist batch of ${batch} plays, trying individually...`); + } + signal?.throwIfAborted(); + + let valid = 0; + for(const p of batch) { + try { + await this.playsHistoricalRepo.createPlays([p]); + valid++; + } catch (e) { + if(allowFailures) { + logger.warn(p.play,`Failed to persist play from record with rKey ${p.play.meta.playId} => ${buildTrackString(p.play)}`); + logger.warn(e); + } else { + logger.error(p.play,`Failed to persist play from record with rKey ${p.play.meta.playId} => ${buildTrackString(p.play)}`); + throw e; + } + } + signal?.throwIfAborted(); + } + + return [false, valid]; + } } diff --git a/src/backend/server/api.ts b/src/backend/server/api.ts index 1ff206b5..336d3c3d 100644 --- a/src/backend/server/api.ts +++ b/src/backend/server/api.ts @@ -37,6 +37,7 @@ import prom from 'prom-client'; import { SimpleError } from "../common/errors/MSErrors.js"; import { QueryPlaysOpts } from "../common/database/drizzle/repositories/PlayRepository.js"; import { playSelectToDeadScrobble } from "../common/database/drizzle/entityUtils.js"; +import AbstractHistoricalScrobbleClient from "../scrobblers/AbstractHistoricalScrobbleClient.js"; const maxBufferSize = 300; const output: Record> = {}; @@ -527,6 +528,19 @@ export const setupApi = (app: Express, logger: Logger, appLoggerStream: PassThro res.status(200).send('OK'); }); + app.post('/api/client/historical', clientRequiredMiddle, async (req, res) => { + // @ts-expect-error TS(2339): Property 'scrobbleSource' does not exist on type '... Remove this comment to see the full error message + const client = req.scrobbleClient as AbstractScrobbleClient; + if(client instanceof AbstractHistoricalScrobbleClient) { + client.logger.info('User requested historical play hydration'); + client.hydrateHistoricalScrobbles(); + res.status(200).send('OK'); + } else { + client.logger.warn('This client does not have historical play capabilities'); + return res.status(400).json({error: 'This client does not have historical play capabilities'}); + } + }); + app.get('/health', async (req, res) => res.redirect(307, `/api/${req.url.slice(1)}`)); app.get('/api/health', async (req, res) => { const { diff --git a/src/backend/tests/tealfm/tealfm.test.ts b/src/backend/tests/tealfm/tealfm.test.ts index 46e3e422..fcbaf827 100644 --- a/src/backend/tests/tealfm/tealfm.test.ts +++ b/src/backend/tests/tealfm/tealfm.test.ts @@ -5,6 +5,13 @@ import { generateArtistCredits, generatePlay, generateTealPlayRecord, withBrainz import { listRecordToPlay, playToRecord } from '../../common/vendor/bluesky/AbstractBlueSkyApiClient.js'; import dayjs from 'dayjs'; import { artistCreditsToNames } from '../../../core/StringUtils.js'; +import TealScrobbler from '../../scrobblers/TealfmScrobbler.js'; +import { Notifiers } from '../../notifier/Notifiers.js'; +import { EventEmitter } from "events"; +import { loggerNoop } from '../../common/MaybeLogger.js'; +import path from 'node:path'; +import { configDir } from '../../common/index.js'; +import { loggerDebug, loggerTrace } from '@foxxmd/logging'; chai.use(asPromised); @@ -71,4 +78,45 @@ describe('#tealfm Play To Record', function () { expect(record.artists[0].artistMbId).eq(`mbid:${play.data.artists[0].mbid}`); }); +}); + +describe('#tealfm Play To Record', function () { + + it('Adds mbids with uri format', function () { + + const play = withBrainz(generatePlay({artists: generateArtistCredits(2)}), {include: ['recording']}); + const record = playToRecord(play); + + expect(record.recordingMbId).to.eq(`mbid:${play.data.meta.brainz.recording}`); + expect(record.releaseMbId).is.undefined; + expect(record.artists).length(2); + expect(record.artists[0].artistName).eq(play.data.artists[0].name); + expect(record.artists[0].artistMbId).eq(`mbid:${play.data.artists[0].mbid}`); + }); + +}); + +describe('#tealfmCar', function() { + + before(function () { + if (process.env.TEAL_CAR_TEST !== 'true') { + this.skip(); + } + }); + + it('Parses car file', async function() { + + this.timeout(100000); + + const tfm = new TealScrobbler('test', + {name: 'test', data: {identifier: 'test', appPassword: 'test'}}, + {configDir: 'test', localUrl: new URL('https://example.com'), version: 'test'}, + new Notifiers(new EventEmitter(), new EventEmitter(), new EventEmitter(), loggerNoop), + new EventEmitter(), + loggerDebug + ); + await tfm.buildDatabase(); + + await tfm.parseScrobblesFromCar(path.resolve(configDir, 'tealfm-myteal-1778870858.car'), 100); + }); }); \ No newline at end of file