From 2da35f4e68df13cd28d62ab4bb7139040ac99be7 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Tue, 22 Sep 2026 19:35:45 +0000 Subject: [PATCH] feat(rocksky): Leverage rocksky index for dupe detection * (Re)sync agent-based RS repo on historical scrobble hydration * Define generic external existing play check * Implement external check for rocksky by using rocksky index --- src/backend/common/vendor/RockSkyApiClient.ts | 45 +++++++++++++++++-- src/backend/scrobblers/RockskyScrobbler.ts | 29 ++++++++++-- src/backend/utils/PlayComparisonUtils.ts | 19 ++++++++ 3 files changed, 87 insertions(+), 6 deletions(-) diff --git a/src/backend/common/vendor/RockSkyApiClient.ts b/src/backend/common/vendor/RockSkyApiClient.ts index c6f7041a..f047cd37 100644 --- a/src/backend/common/vendor/RockSkyApiClient.ts +++ b/src/backend/common/vendor/RockSkyApiClient.ts @@ -10,7 +10,8 @@ import type {ListenResponse, ListenType} from '../../../core/vendor/listenbrainz import { getATProtoIdentifier, identifierToAtProtoHandle, isDID } from './atproto/atUtils.ts'; import { baseFormatPlayObj } from "../../utils/PlayTransformUtils.ts"; import { AuthError, ScrobbleSubmitError, SimpleError } from "../errors/MSErrors.ts"; -import { type CreateScrobbleInput, RockskyClient, Agent, type SongViewDetailed, type ScrobbleInput, type ScrobbleViewBasic, RockskyError, type ActorTrackView } from "@rocksky/sdk"; +import { type CreateScrobbleInput, RockskyClient, Agent, artistHash, type SongViewDetailed, type ScrobbleInput, type ScrobbleViewBasic, RockskyError, type ActorTrackView } from "@rocksky/sdk"; +import { RockskyIndex } from "@rocksky/sdk/dedup"; import { getRoot } from "../../ioc.ts"; import type { MSCache } from "../Cache.ts"; import type {ATProtoUserIdentifierData, HandleData} from "../infrastructure/config/client/atproto.ts"; @@ -18,11 +19,15 @@ import { parseRegexSingle } from "@foxxmd/regex-buddy-core"; import { removeUndefinedKeys } from "../../../core/DataUtils.ts"; import { isrcNoHyphens } from '../../../core/PlayUtils.ts'; import { findCauseByFunc } from "../../utils/ErrorUtils.ts"; -import { hashObject } from "../../utils/StringUtils.ts"; +import { hashObject, normalizeStr } from "../../utils/StringUtils.ts"; import { stringSameness } from "@foxxmd/string-sameness"; import clone from "clone"; import { difference } from "../../utils.ts"; import { RockskyClientPool } from "./rocksky/RockskyClientWrapped.ts"; +import path from "node:path"; +import { ATProtoUnauthenticatedApiClient } from "./atproto/ATProtoUnauthenticatedApiClient.ts"; +import fsPromise from 'node:fs/promises'; +import { getDataDir } from "../index.ts"; interface SubmitOptions { log?: boolean @@ -54,14 +59,18 @@ export class RockSkyApiClient extends AbstractApiClient { rsClient?: RockskyClient; rsPool: RockskyClientPool; rsAgent?: Agent; + rsIndex?: RockskyIndex; - constructor(name: any, config: RockSkyData & RockSkyOptions, options: AbstractApiOptions) { + protected configDir: string; + + constructor(name: any, config: RockSkyData & RockSkyOptions, options: AbstractApiOptions & {configDir: string}) { super('RockSky', name, config, options); const { apiUrl, token, } = config; + this.configDir = options.configDir; this.cache = getRoot().items.cache(); this.apiUrl = normalizeWebAddress(apiUrl ?? 'https://api.rocksky.app/xrpc'); @@ -69,9 +78,19 @@ export class RockSkyApiClient extends AbstractApiClient { this.rsPool = new RockskyClientPool('Pool', {apis: [{enable: true}]}, {logger: this.logger}); this.rsClient = new RockskyClient(this.apiUrl.url.origin, token); + this.rsIndex = new RockskyIndex(path.resolve(getDataDir(), `${this.getSafeExternalId()}-rsindex`)); + } + + async [Symbol.asyncDispose]() { + try { + await this.rsIndex.close(); + } catch (e) { + this.logger.warn(e); + } } public async buildData() { + await this.rsIndex.open(); try { const atProtoHandleData: ATProtoUserIdentifierData = { identifier: this.config.handle @@ -111,6 +130,7 @@ export class RockSkyApiClient extends AbstractApiClient { if((this.userData !== undefined || this.config.handle !== undefined) && this.config.appPassword !== undefined) { try { this.rsAgent = await Agent.login(this.userData?.did ?? this.userData?.handle ?? this.config.handle, this.config.appPassword); + this.rsAgent.useIndex(this.rsIndex); return true; } catch (e) { throw new AuthError('Could not login using handle/did and appPassword', {cause: e, unrecoverable: true}); @@ -250,6 +270,25 @@ export class RockSkyApiClient extends AbstractApiClient { static formatPlayObj(obj: any, options: FormatPlayObjectOptions): PlayObject { return rockskyScrobbleToPlay(obj); } + + async fetchCarToFile() { + // TODO use `since` to get CAR diff instead of entire repo + // can use last import date from migrations table + const filename = path.resolve(this.configDir, `${this.getSafeExternalId()}-${dayjs().unix()}.car`); + const atClient = new ATProtoUnauthenticatedApiClient('rocksky', { handleData: this.userData, identifier: this.config.handle }, { logger: this.logger }); + await atClient.initClient(); + await fsPromise.writeFile(filename, Buffer.from(await atClient.getCAR(this.userData.did))); + return filename; + } + + async syncSdkRepo(filename: string) { + await this.rsIndex.indexCar(this.userData.did, await fsPromise.readFile(filename)); + } + + public getSafeExternalId() { + return `${this.type}-${normalizeStr(this.name, {keepSingleWhitespace: false})}`; + } + } interface RsMatchSongInput { diff --git a/src/backend/scrobblers/RockskyScrobbler.ts b/src/backend/scrobblers/RockskyScrobbler.ts index fb7d18c7..7a9ca533 100644 --- a/src/backend/scrobblers/RockskyScrobbler.ts +++ b/src/backend/scrobblers/RockskyScrobbler.ts @@ -10,7 +10,7 @@ import type {ListenPayload} from '../../core/vendor/listenbrainz/interfaces.ts'; import { isDebugMode } from "../utils.ts"; import { durationToHuman } from '../../core/TimeUtils.ts'; -import { RockSkyApiClient, rockskyScrobbleToPlay } from "../common/vendor/RockSkyApiClient.ts"; +import { playToRockskyClientRecord, RockSkyApiClient, rockskyScrobbleToPlay } from "../common/vendor/RockSkyApiClient.ts"; import type {RockSkyClientConfig} from "../common/infrastructure/config/client/rocksky.ts"; import AbstractHistoricalScrobbleClient from "./AbstractHistoricalScrobbleClient.ts"; import { fromStream } from '@atcute/repo'; @@ -23,6 +23,7 @@ import { ATProtoUnauthenticatedApiClient } from "../common/vendor/atproto/ATProt import { playToRepositoryCreatePlayHistoricalOpts, type RepositoryCreatePlayHistoricalOpts } from "../common/database/drizzle/repositories/PlayHistoricalRepository.ts"; import { isAbortError } from "abort-controller-x"; import { shouldClearNPStatus } from "./AbstractScrobbleClient.ts"; +import { removeUndefinedKeys } from "../../core/DataUtils.ts"; export default class RockskyScrobbler extends AbstractHistoricalScrobbleClient { @@ -45,7 +46,7 @@ export default class RockskyScrobbler extends AbstractHistoricalScrobbleClient { constructor(name: any, config: RockSkyClientConfig, options: InternalConfigOptional & { [key: string]: any }, emitter: EventEmitter, logger: Logger) { super('rocksky', name, config, emitter, logger); - this.api = new RockSkyApiClient(name, { ...config.data, ...config.options }, { logger: this.logger }); + this.api = new RockSkyApiClient(name, { ...config.data, ...config.options }, { logger: this.logger, configDir: options.configDir }); // https://listenbrainz.readthedocs.io/en/latest/users/api/core.html#get--1-user-(user_name)-listens // 1000 is way too high. maxing at 100 this.MAX_INITIAL_SCROBBLES_FETCH = 100; @@ -53,6 +54,24 @@ export default class RockskyScrobbler extends AbstractHistoricalScrobbleClient { // PDS rate limit for operations is ~2/sec this.scrobbleDelay = 2000; this.configDir = options.configDir; + this.existingPlayOpts = { + logger: this.dupeLogger, + transformRules: this.transformRules, + transformPlay: this.transformPlay, + existingSubmitted: this.findExistingSubmittedPlayObj, + existingExternal: async (play) => { + const createRecord = removeUndefinedKeys(playToRockskyClientRecord(play)); + const existingUri = await this.api.rsIndex.scrobbleUri(this.api.userData.did, createRecord.title, createRecord.artist, createRecord.album, createRecord.timestamp); + if(existingUri) { + return { + match: true, + reason: 'matched existing scrobble record URI', + data: existingUri + } + } + return {match: false}; + } + } } formatPlayObj = (obj: any, options: FormatPlayObjectOptions = {}) => ListenbrainzApiClient.formatPlayObj(obj, options); @@ -152,7 +171,7 @@ export default class RockskyScrobbler extends AbstractHistoricalScrobbleClient { let file: string; try { logger.verbose('Fetching scrobbles from PDS...'); - file = await this.fetchCarToFile(); + file = await this.api.fetchCarToFile() signal?.throwIfAborted(); } catch (e) { throw new Error('Failed to fetch repo CAR', {cause: e}); @@ -184,6 +203,10 @@ export default class RockskyScrobbler extends AbstractHistoricalScrobbleClient { async parseScrobblesFromCar(filename: string, batchSize: number, opts: { allowFailures?: boolean, logger?: Logger, signal?: AbortSignal } = {}) { + if(this.api.rsAgent !== undefined) { + await this.api.syncSdkRepo(filename); + } + const { allowFailures = false, logger = this.logger, diff --git a/src/backend/utils/PlayComparisonUtils.ts b/src/backend/utils/PlayComparisonUtils.ts index 0bc45bb5..cedd881f 100644 --- a/src/backend/utils/PlayComparisonUtils.ts +++ b/src/backend/utils/PlayComparisonUtils.ts @@ -13,6 +13,7 @@ import { loggerNoop } from '../common/MaybeLogger.ts'; import { statefulInvariantTransform } from "../../core/PlayUtils.ts"; import { findAsyncSequential } from "./AsyncUtils.ts"; import dayjs from "dayjs"; +import { SimpleError } from "../common/errors/MSErrors.ts"; export const metaInvariantTransform = (play: PlayObject): PlayObjectMinimal => { @@ -433,6 +434,7 @@ export const playDateWithinDurationOfAny = (play: PlayObject, plays: PlayObject export interface ExistingScrobbleOpts { transformPlay?: (play: PlayObject, hookType: TransformHook) => Promise existingSubmitted?: (play: PlayObject) => Promise<[ScrobbledPlayObject?, ScrobbledPlayObject[]?]> + existingExternal?: (play: PlayObject) => Promise<{reason?: string, match: boolean, data?: string | Record}> transformRules?: PlayTransformRules checkExistingScrobbles?: boolean logger?: Logger @@ -443,6 +445,7 @@ export const existingScrobble = async (playObjPre: PlayObject, existingScrobbles const { transformPlay = (play, hook) => play, existingSubmitted = (play) => [undefined, undefined], + existingExternal, transformRules, checkExistingScrobbles = true, logger = loggerNoop @@ -484,6 +487,22 @@ export const existingScrobble = async (playObjPre: PlayObject, existingScrobbles result.reason = 'Exact Match found in previously successfully scrobbled plays'; existingScrobble = existingExactSubmitted.scrobble; + } else if(existingExternal !== undefined) { + try { + const res = await existingExternal(playObj); + if(res.match) { + return { + match: true, + score: 1, + breakdowns: [], + reason: res.reason, + summary: res.data !== undefined ? (typeof res.data === 'string' ? res.data : JSON.stringify(res.data)) : undefined, + createdAt: dayjs().toISOString() + } + } + } catch (e) { + logger.warn(new SimpleError('External existing check failed', {cause: e})); + } } // if not though then we need to check recent scrobbles from scrobble api. // this will be less accurate than checking existing submitted (obv) but will happen if backlogging or on a fresh server start -- 2.51.2