feat(rocksky): Implement historical scrobbles · foxxmd.dev/multi-scrobbler@470a70f

+39 -6

src/backend/common/vendor/RockSkyApiClient.ts

··· 20 20 import { getRoot } from "../../ioc.js"; 21 21 import { MSCache } from "../Cache.js"; 22 22 import { HandleData } from "../infrastructure/config/client/atproto.js"; 23 + import { parseRegexSingle } from "@foxxmd/regex-buddy-core"; 23 24 24 25 interface SubmitOptions { 25 26 log?: boolean ··· 73 74 this.rsClient = new RockskyClient({auth: token}); 74 75 } 75 76 76 - protected isLzMode = () => this.config.key !== undefined; 77 + isLzMode = () => this.config.key !== undefined; 77 78 78 79 doCallLZApi = async <T = Response>(req: Request, retries = 0): Promise<T> => { 79 80 try { ··· 247 248 scrobbles?: RockskyScrobble[] 248 249 } 249 250 250 - export const rockskyScrobbleToPlay = (obj: RockskyScrobble): PlayObject => { 251 + export const rockskyScrobbleToPlay = (obj: RockskyScrobble, opts: {playId?: string, web?: string, user?: string} = {}): PlayObject => { 252 + const { 253 + playId, 254 + web, 255 + user 256 + } = opts; 251 257 const play: PlayObjectLifecycleless = { 252 258 data: { 253 259 track: obj.title, ··· 259 265 playDate: dayjs.utc(obj.createdAt).local() 260 266 }, 261 267 meta: { 262 - // @ts-expect-error its in the response but missing from types 263 - trackId: obj.trackId, 264 - playId: obj.id, 268 + playId, 269 + user 265 270 } 266 271 }; 272 + if(web !== undefined) { 273 + play.meta.url = {web}; 274 + } 275 + if('trackId' in obj) { 276 + play.meta.trackId = obj.trackId as string; 277 + } 278 + // if('id' in obj) { 279 + // play.meta.playId = obj.id as string; 280 + // } 267 281 if('albumArt' in obj) { 268 282 play.meta.art = {album: obj.albumArt as string} 269 283 } 270 284 285 + if(obj.uri !== undefined) { 286 + const uriRes = parseRegexSingle(ATPROTO_URI_REGEX, obj.uri); 287 + if(uriRes !== undefined) { 288 + if(web === undefined) { 289 + play.meta.url = { 290 + web: `https://atproto.at/viewer?uri=${uriRes.named.resource}` 291 + } 292 + } 293 + if(playId === undefined) { 294 + play.meta.playId = uriRes.named.tid; 295 + } 296 + if(user === undefined) { 297 + play.meta.user = uriRes.named.did; 298 + } 299 + } 300 + } 301 + 271 302 return baseFormatPlayObj(obj, play); 272 303 } 273 304 ··· 283 314 timestamp: play.data.playDate.unix() 284 315 } 285 316 return csi; 286 - } 317 + } 318 + 319 + const ATPROTO_URI_REGEX = new RegExp(/at:\/\/(?<resource>(?<did>did.*?)\/app\.rocksky\.scrobble\/(?<tid>.*))/);

+1 -1

src/backend/common/vendor/atproto/ATProtoUnauthenticatedApiClient.ts

··· 11 11 declare client: Client; 12 12 13 13 async initClient(): Promise<void> { 14 - this.userData = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth}); 14 + await this.hydrateHandleData(); 15 15 this.client = new Client({ handler: simpleFetchHandler({ service: this.userData.pds }) }); 16 16 } 17 17

+19 -7

src/backend/common/vendor/atproto/AbstractATProtoApiClient.ts

··· 5 5 import { UpstreamError } from "../../errors/UpstreamError.js"; 6 6 import { streamBodyProgress } from "../../../utils/NetworkUtils.js"; 7 7 import { ATProtoUserIdentifierData, HandleData } from "../../infrastructure/config/client/atproto.js"; 8 - import { checkPds, isDID, identifierToAtProtoHandle } from "./atUtils.js"; 8 + import { checkPds, isDID, identifierToAtProtoHandle, getATProtoIdentifier } from "./atUtils.js"; 9 9 import { Client, isXRPCErrorPayload } from '@atcute/client'; 10 10 import { ComAtprotoSyncGetRepo } from '@atcute/atproto'; 11 11 import { AtprotoDid } from "@atcute/lexicons/syntax"; ··· 20 20 21 21 cache: MSCache; 22 22 23 - constructor(name: any, config: ATProtoUserIdentifierData, options: AbstractApiOptions) { 23 + constructor(name: any, config: ATProtoUserIdentifierData & {handleData?: HandleData}, options: AbstractApiOptions) { 24 24 super('atproto', name, config, options); 25 25 this.cache = getRoot().items.cache(); 26 26 27 - const cleanIdentifier = this.config.identifier; 28 - if(isDID(cleanIdentifier)) { 29 - this.logger.debug(`Identifier ${cleanIdentifier} looks like a DID, skipping parsing as a handle.`); 30 - this.config.did = cleanIdentifier; 27 + if(config.handleData !== undefined) { 28 + this.userData = config.handleData; 29 + this.config.did = config.handleData.did; 30 + this.config.identifier = config.handleData.handle; 31 31 } else { 32 - this.config.identifier = identifierToAtProtoHandle(this.config.identifier, {logger: this.logger, defaultDomain: 'bsky.social'}); 32 + const cleanIdentifier = this.config.identifier; 33 + if(isDID(cleanIdentifier)) { 34 + this.logger.debug(`Identifier ${cleanIdentifier} looks like a DID, skipping parsing as a handle.`); 35 + this.config.did = cleanIdentifier; 36 + } else { 37 + this.config.identifier = identifierToAtProtoHandle(this.config.identifier, {logger: this.logger, defaultDomain: 'bsky.social'}); 38 + } 33 39 } 34 40 } 35 41 ··· 37 43 38 44 async checkPds(data: ATProtoUserIdentifierData): Promise<true> { 39 45 return await checkPds(data, {logger: this.logger, cache: this.cache.cacheAuth}); 46 + } 47 + 48 + async hydrateHandleData(): Promise<void> { 49 + if(this.userData === undefined) { 50 + this.userData = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth}); 51 + } 40 52 } 41 53 42 54 async getCAR(did: AtprotoDid) {

+151 -14

src/backend/scrobblers/RockskyScrobbler.ts

··· 1 - import { Logger } from "@foxxmd/logging"; 1 + import { childLogger, Logger } from "@foxxmd/logging"; 2 2 import EventEmitter from "events"; 3 3 import { PlayObject, SourcePlayerObj } from "../../core/Atomic.js"; 4 4 import { buildTrackString, capitalize } from "../../core/StringUtils.js"; 5 5 import { isNodeNetworkException } from "../common/errors/NodeErrors.js"; 6 - import { FormatPlayObjectOptions } from "../common/infrastructure/Atomic.js"; 6 + import { FormatPlayObjectOptions, InternalConfigOptional } from "../common/infrastructure/Atomic.js"; 7 7 import { ListenbrainzApiClient } from "../common/vendor/ListenbrainzApiClient.js"; 8 8 import { playToListenPayload } from '../common/vendor/listenbrainz/lzUtils.js'; 9 9 import { ListenPayload } from '../common/vendor/listenbrainz/interfaces.js'; 10 10 import { Notifiers } from "../notifier/Notifiers.js"; 11 11 12 - import AbstractScrobbleClient from "./AbstractScrobbleClient.js"; 13 - import { isDebugMode } from "../utils.js"; 14 - import { RockSkyApiClient, SubmitResponse } from "../common/vendor/RockSkyApiClient.js"; 12 + import { durationToHuman, isDebugMode } from "../utils.js"; 13 + import { RockSkyApiClient, rockskyScrobbleToPlay, SubmitResponse } from "../common/vendor/RockSkyApiClient.js"; 15 14 import { RockSkyClientConfig } from "../common/infrastructure/config/client/rocksky.js"; 16 15 import { ScrobbleSubmitError } from "../common/errors/MSErrors.js"; 16 + import AbstractHistoricalScrobbleClient from "./AbstractHistoricalScrobbleClient.js"; 17 + import { fromStream } from '@atcute/repo'; 18 + import fsPromise from 'node:fs/promises'; 19 + import fs from 'node:fs'; 20 + import path from 'path'; 21 + import dayjs from "dayjs"; 22 + import { Readable } from 'stream'; 23 + import { ATProtoUnauthenticatedApiClient } from "../common/vendor/atproto/ATProtoUnauthenticatedApiClient.js"; 24 + import { playToRepositoryCreatePlayHistoricalOpts, RepositoryCreatePlayHistoricalOpts } from "../common/database/drizzle/repositories/PlayHistoricalRepository.js"; 25 + import { isAbortError } from "abort-controller-x"; 17 26 18 - export default class RockskyScrobbler extends AbstractScrobbleClient { 27 + export default class RockskyScrobbler extends AbstractHistoricalScrobbleClient { 19 28 20 29 api: RockSkyApiClient; 21 30 requiresAuth = true; ··· 23 32 24 33 declare config: RockSkyClientConfig; 25 34 26 - constructor(name: any, config: RockSkyClientConfig, options = {}, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { 35 + protected configDir: string; 36 + 37 + constructor(name: any, config: RockSkyClientConfig, options: InternalConfigOptional & { [key: string]: any }, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { 27 38 super('rocksky', name, config, notifier, emitter, logger); 28 - this.api = new RockSkyApiClient(name, {...config.data, ...config.options}, {logger: this.logger}); 39 + this.api = new RockSkyApiClient(name, { ...config.data, ...config.options }, { logger: this.logger }); 29 40 // https://listenbrainz.readthedocs.io/en/latest/users/api/core.html#get--1-user-(user_name)-listens 30 41 // 1000 is way too high. maxing at 100 31 42 this.MAX_INITIAL_SCROBBLES_FETCH = 100; 32 43 this.supportsNowPlaying = false; 33 44 // PDS rate limit for operations is ~2/sec 34 45 this.scrobbleDelay = 2000; 46 + this.configDir = options.configDir; 35 47 } 36 48 37 49 formatPlayObj = (obj: any, options: FormatPlayObjectOptions = {}) => ListenbrainzApiClient.formatPlayObj(obj, options); ··· 59 71 try { 60 72 return await this.api.testAuth(); 61 73 } catch (e) { 62 - if(isNodeNetworkException(e)) { 74 + if (isNodeNetworkException(e)) { 63 75 this.logger.error('Could not communicate with Rocksky API'); 64 76 } 65 77 throw e; ··· 83 95 } = playObj; 84 96 85 97 try { 86 - const result = await this.api.submitListen(playObj, { log: isDebugMode()}); 98 + const result = await this.api.submitListen(playObj, { log: isDebugMode() }); 87 99 88 - if(((result.response as SubmitResponse).payload?.ignored_listens ?? 0) > 0) { 89 - throw new ScrobbleSubmitError('Scrobble was successfully submitted but Rocksky ignored it', {showStopper: false, responseBody: result.response, payload: result.payload}); 100 + if (this.api.isLzMode() && ((result.response as SubmitResponse).payload?.ignored_listens ?? 0) > 0) { 101 + throw new ScrobbleSubmitError('Scrobble was successfully submitted but Rocksky ignored it', { showStopper: false, responseBody: result.response, payload: result.payload }); 90 102 } 91 103 92 104 if (newFromSource) { ··· 96 108 } 97 109 return result; 98 110 } catch (e) { 99 - await this.notifier.notify({title: `Client - ${capitalize(this.type)} - ${this.name} - Scrobble Error`, message: `Failed to scrobble => ${buildTrackString(playObj)} | Error: ${e.message}`, priority: 'error'}); 111 + await this.notifier.notify({ title: `Client - ${capitalize(this.type)} - ${this.name} - Scrobble Error`, message: `Failed to scrobble => ${buildTrackString(playObj)} | Error: ${e.message}`, priority: 'error' }); 100 112 throw e; 101 113 } 102 114 } 103 115 104 116 doPlayingNow = async (data: SourcePlayerObj) => { 105 117 try { 106 - await this.api.submitListen(data.play, { listenType: 'playing_now'}); 118 + await this.api.submitListen(data.play, { listenType: 'playing_now' }); 107 119 } catch (e) { 108 120 throw e; 109 121 } 122 + } 123 + 124 + protected async doHydrateHistoricalScrobbles(opts: {allowFailures?: boolean, signal?: AbortSignal } = {}) { 125 + const logger = childLogger(this.logger, ['Historical Plays']); 126 + const { 127 + allowFailures = false, 128 + signal 129 + } = opts; 130 + let file: string; 131 + try { 132 + logger.verbose('Fetching scrobbles from PDS...'); 133 + file = await this.fetchCarToFile(); 134 + signal?.throwIfAborted(); 135 + } catch (e) { 136 + throw new Error('Failed to fetch repo CAR', {cause: e}); 137 + } 138 + 139 + try { 140 + await this.parseScrobblesFromCar(file, 100, {allowFailures, logger: logger, signal}); 141 + } catch (e) { 142 + throw new Error('Failed to convert CAR without any error', {cause: e}); 143 + } finally { 144 + await fsPromise.rm(file); 145 + } 146 + } 147 + 148 + async fetchCarToFile() { 149 + // TODO use `since` to get CAR diff instead of entire repo 150 + // can use last import date from migrations table 151 + const filename = path.resolve(this.configDir, `${this.getSafeExternalId()}-${dayjs().unix()}.car`); 152 + const atClient = new ATProtoUnauthenticatedApiClient('rocksky', { handleData: this.api.userData, identifier: this.api.userData.handle }, { logger: this.logger }); 153 + await atClient.initClient(); 154 + await fsPromise.writeFile(filename, Buffer.from(await atClient.getCAR(this.api.userData.did))); 155 + return filename; 156 + } 157 + 158 + async parseScrobblesFromCar(filename: string, batchSize: number, opts: { allowFailures?: boolean, logger?: Logger, signal?: AbortSignal } = {}) { 159 + 160 + const { 161 + allowFailures = false, 162 + logger = this.logger, 163 + signal 164 + } = opts; 165 + 166 + const stream = Readable.toWeb(fs.createReadStream(filename)); 167 + 168 + await using repo = fromStream(stream); 169 + 170 + let batch: RepositoryCreatePlayHistoricalOpts[] = []; 171 + let allGood = true; 172 + let count = 0; 173 + let persisted = 0; 174 + const start = dayjs(); 175 + 176 + logger.info('Starting CAR conversion to historical plays...'); 177 + 178 + for await (const entry of repo) { 179 + if (entry.collection === 'app.rocksky.scrobble') { 180 + let play: PlayObject; 181 + try { 182 + play = rockskyScrobbleToPlay(entry.record, {user: this.api.userData.did, playId: entry.rkey, web: `${this.api.userData.did}/app.rocksky.scrobble/${entry.rkey}`}) 183 + if (isDebugMode()) { 184 + logger.trace(`(${count}) rKey ${entry.rkey} => ${buildTrackString(play)}`); 185 + } 186 + count++; 187 + if (count % (batchSize * 5) === 0) { 188 + logger.debug(`Processed ${count} records`); 189 + signal?.throwIfAborted(); 190 + } 191 + } catch (e) { 192 + if (isAbortError(e)) { 193 + throw e; 194 + } 195 + if (allowFailures) { 196 + this.logger.warn(new Error(`Failed to convert record ${entry.rkey} to Play but will continue`, { cause: e })); 197 + continue; 198 + } else { 199 + throw new Error(`Failed to convert record ${entry.rkey} to Play`, { cause: e }); 200 + } 201 + } 202 + 203 + const existing = await this.playsHistoricalRepo.hasByUid(entry.rkey); 204 + if (!existing) { 205 + batch.push(playToRepositoryCreatePlayHistoricalOpts({ play })); 206 + } 207 + if (batch.length >= batchSize) { 208 + try { 209 + const [res, valid] = await this.createHistoricalPlays(batch, opts); 210 + persisted += valid; 211 + if (!res) { 212 + allGood = false; 213 + } 214 + } catch (e) { 215 + throw e; 216 + } 217 + batch = []; 218 + } 219 + } 220 + } 221 + 222 + logger.debug('Reached end of CAR file'); 223 + if (batch.length > 0) { 224 + logger.debug(`Persisting remaining ${batch.length} records...`); 225 + try { 226 + const [res, valid] = await this.createHistoricalPlays(batch, opts); 227 + persisted += valid; 228 + if (!res) { 229 + allGood = false; 230 + } 231 + } catch (e) { 232 + throw e; 233 + } 234 + } 235 + logger.info(`Completed CAR conversion: Result ${allGood ? 'OK' : 'Some Errors'} in ${durationToHuman(dayjs.duration(dayjs().diff(start)))} | Records ${count} | Persisted ${persisted}`) 236 + } 237 + 238 + protected async syncRecentHistoricalScrobbles(): Promise<PlayObject[]> { 239 + const recentPlays = await this.getScrobblesForTimeRange(undefined); 240 + const unseenPlays: PlayObject[] = []; 241 + for (const p of recentPlays) { 242 + if(!(await this.playsHistoricalRepo.hasByUid(p.meta.playId))) { 243 + unseenPlays.push(p); 244 + } 245 + } 246 + return unseenPlays; 110 247 } 111 248 }

+1 -1

src/backend/scrobblers/ScrobbleClients.ts

··· 468 468 break; 469 469 case 'rocksky': 470 470 const RockskyScrobbler = (await import('./RockskyScrobbler.js')).default; 471 - newClient = new RockskyScrobbler(name, {...clientConfig, data: {configDir: this.internalConfig.configDir, ...d}, options: compositeOptions } as unknown as RockSkyClientConfig, {}, notifier, this.emitter, this.logger); 471 + newClient = new RockskyScrobbler(name, {...clientConfig, data: {configDir: this.internalConfig.configDir, ...d}, options: compositeOptions } as unknown as RockSkyClientConfig, this.internalConfig, notifier, this.emitter, this.logger); 472 472 break; 473 473 case 'discord': 474 474 const DiscordScrobbler = (await import('./DiscordScrobbler.js')).default;

Configure Feed

Configure Feed