From 41d181e02fe38cf376004f97d52dbdaaf2331a30 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Thu, 4 Jun 2026 17:50:40 +0000 Subject: [PATCH] refactor: Replace @atproto with @atcute for abstract/app api implementation --- .../vendor/atproto/ATProtoAppApiClient.ts | 90 +++++++++++-------- .../atproto/ATProtoAuthenticatedApiClient.ts | 5 ++ .../vendor/atproto/ATProtoOauthApiClient.ts | 5 +- .../ATProtoUnauthenticatedApiClient.ts | 35 ++++++++ .../atproto/AbstractATProtoApiClient.ts | 72 ++++++++------- src/backend/common/vendor/atproto/atUtils.ts | 8 +- .../common/vendor/teal/TealApiClient.ts | 45 +++++++--- src/backend/scrobblers/TealfmScrobbler.ts | 13 +-- src/backend/utils/NetworkUtils.ts | 14 +-- 9 files changed, 175 insertions(+), 112 deletions(-) create mode 100644 src/backend/common/vendor/atproto/ATProtoAuthenticatedApiClient.ts create mode 100644 src/backend/common/vendor/atproto/ATProtoUnauthenticatedApiClient.ts diff --git a/src/backend/common/vendor/atproto/ATProtoAppApiClient.ts b/src/backend/common/vendor/atproto/ATProtoAppApiClient.ts index f88706ec..3cd67aa9 100644 --- a/src/backend/common/vendor/atproto/ATProtoAppApiClient.ts +++ b/src/backend/common/vendor/atproto/ATProtoAppApiClient.ts @@ -1,70 +1,82 @@ import { AbstractApiOptions } from "../../infrastructure/Atomic.js"; import { TealClientData } from "../../infrastructure/config/client/tealfm.js"; -import { Agent, CredentialSession, AtpSessionEvent, AtpSessionData } from "@atproto/api"; -import { AbstractATProtoApiClient } from "./AbstractATProtoApiClient.js"; -import { getATProtoIdentifier, identifierToAtProtoHandle, isDID } from "./atUtils.js"; +import { getATProtoIdentifier } from "./atUtils.js"; import { ATProtoAppData, ATProtoUserIdentifierData } from "../../infrastructure/config/client/atproto.js"; +import { ATProtoAuthenticatedApiClient } from "./ATProtoAuthenticatedApiClient.js"; +import { PasswordSession, PasswordSessionData } from '@atcute/password-session'; +import { Client } from "@atcute/client"; -export class ATProtoAppApiClient extends AbstractATProtoApiClient { +export class ATProtoAppApiClient extends ATProtoAuthenticatedApiClient { declare config: ATProtoUserIdentifierData & ATProtoAppData; - appSession?: CredentialSession; - appPwAuth: boolean - constructor(name: any, config: TealClientData, options: AbstractApiOptions) { super(name, config, options); - this.logger.verbose(`Using App Password auth for session`); - const cleanIdentifier = this.config.identifier; - if(isDID(cleanIdentifier)) { - this.logger.debug(`Identifier ${cleanIdentifier} looks like a DID, skipping parsing as a handle.`); - this.config.did = cleanIdentifier; - } else { - this.config.identifier = identifierToAtProtoHandle(this.config.identifier, {logger: this.logger, defaultDomain: 'bsky.social'}); - } - if(this.config.appPassword === undefined) { + if (this.config.appPassword === undefined) { throw new Error('Must provide app password'); } + this.logger.verbose(`Using App Password auth for session`); } async initClient(): Promise { - const hd = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth}); - this.logger.verbose(`Using ${hd.did} on PDS ${hd.pds}`); - this.appSession = new CredentialSession(new URL(hd.pds), undefined, (evt: AtpSessionEvent, sess?: AtpSessionData) => { - this.cache.cacheAuth.set(`appPwSession-${this.name}-${hd.did}`, sess, '1000h'); - }); - this.agent = new Agent(this.appSession); + this.userData = await getATProtoIdentifier(this.config, { logger: this.logger, cache: this.cache.cacheAuth }); + this.logger.verbose(`Using ${this.userData.did} on PDS ${this.userData.pds}`); } restoreSession = async (): Promise => { - const hd = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth}); - const savedSession = await this.cache.cacheAuth.get(`appPwSession-${this.name}-${hd.did}`); - if (savedSession !== undefined) { + const savedSessionCute = await this.getSession(); + if (savedSessionCute !== undefined) { + const that = this; try { - this.logger.debug('Found existing session, trying to resume...'); - await this.appSession.resumeSession(savedSession); - this.logger.debug('Resumed session!'); - return true; + const session = await PasswordSession.resume(savedSessionCute, { + async onUpdate(data) { + // called on login and token refresh — persist the session + await that.saveSession(data); + }, + async onDelete(data) { + // called on logout or session invalidation — clean up + await that.deleteSession(); + }, + }); + this.client = new Client({ handler: session }); } catch (e) { this.logger.warn(new Error('Could not resume app password session from data', { cause: e })); return false; } } - this.logger.debug('No app password session data to restore'); + } + + protected async saveSession(data: PasswordSessionData): Promise { + await this.cache.cacheAuth.set(`appPwSessionCute-${this.name}-${this.userData.did}`, data, '1000h'); + } + + protected async getSession(): Promise { + return await this.cache.cacheAuth.get(`appPwSessionCute-${this.name}-${this.userData.did}`); + } + + protected async deleteSession(): Promise { + await this.cache.cacheAuth.delete(`appPwSessionCute-${this.name}-${this.userData.did}`); } appLogin = async (): Promise => { + const that = this; try { + const session = await PasswordSession.login( + { service: this.userData.pds, identifier: this.userData.handle, password: this.config.appPassword }, + { + //session: savedSession, + async onUpdate(data) { + // called on login and token refresh — persist the session + await that.saveSession(data); + }, + async onDelete(data) { + // called on logout or session invalidation — clean up + await that.deleteSession(); + }, + }, + ); - const f = await this.appSession.login({ - identifier: this.config.identifier, - password: this.config.appPassword - }); - if (!f.success) { - this.logger.error('Login was not successful with app password'); - return false; - } - this.logger.debug('Logged in.'); + this.client = new Client({ handler: session }); return true; } catch (e) { this.logger.error(new Error('Could not login using app password', { cause: e })); diff --git a/src/backend/common/vendor/atproto/ATProtoAuthenticatedApiClient.ts b/src/backend/common/vendor/atproto/ATProtoAuthenticatedApiClient.ts new file mode 100644 index 00000000..4f82b770 --- /dev/null +++ b/src/backend/common/vendor/atproto/ATProtoAuthenticatedApiClient.ts @@ -0,0 +1,5 @@ +import { AbstractATProtoApiClient } from "./AbstractATProtoApiClient.js"; + +export abstract class ATProtoAuthenticatedApiClient extends AbstractATProtoApiClient { + abstract restoreSession(): Promise; +} \ No newline at end of file diff --git a/src/backend/common/vendor/atproto/ATProtoOauthApiClient.ts b/src/backend/common/vendor/atproto/ATProtoOauthApiClient.ts index ff675858..57a7d856 100644 --- a/src/backend/common/vendor/atproto/ATProtoOauthApiClient.ts +++ b/src/backend/common/vendor/atproto/ATProtoOauthApiClient.ts @@ -8,10 +8,9 @@ import { OAuthSession, } from "@atproto/oauth-client-node"; import { Agent } from "@atproto/api"; -import { AbstractATProtoApiClient } from "./AbstractATProtoApiClient.js"; +import { ATProtoAuthenticatedApiClient } from "./ATProtoAuthenticatedApiClient.js"; - -export class ATProtoOauthApiClient extends AbstractATProtoApiClient { +export class ATProtoOauthApiClient extends ATProtoAuthenticatedApiClient { declare config: TealClientData; diff --git a/src/backend/common/vendor/atproto/ATProtoUnauthenticatedApiClient.ts b/src/backend/common/vendor/atproto/ATProtoUnauthenticatedApiClient.ts new file mode 100644 index 00000000..95386a6e --- /dev/null +++ b/src/backend/common/vendor/atproto/ATProtoUnauthenticatedApiClient.ts @@ -0,0 +1,35 @@ +import { UpstreamError } from "../../errors/UpstreamError.js"; +import { HandleData } from "../../infrastructure/config/client/atproto.js"; +import { AbstractATProtoApiClient } from "./AbstractATProtoApiClient.js"; +import { getATProtoIdentifier, checkPds } from "./atUtils.js"; +import { Client, simpleFetchHandler } from '@atcute/client'; +import type {} from '@atcute/atproto'; +import { Nsid } from "@atcute/lexicons"; + +export class ATProtoUnauthenticatedApiClient extends AbstractATProtoApiClient { + + declare client: Client; + + async initClient(): Promise { + this.userData = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth}); + this.client = new Client({ handler: simpleFetchHandler({ service: this.userData.pds }) }); + } + + async listRecords(collection: string, options: {limit?: number, cursor?: string} = {}) { + const {limit = 20, cursor} = options; + try { + // records are returned newest to oldest + const response = await this.client.get('com.atproto.repo.listRecords', { + params: { + repo: this.userData.did, + collection: collection as Nsid, + limit, + cursor + } + }); + return response; + } catch (e) { + throw new UpstreamError(`Failed to list scrobble record`, { cause: e, response: 'response' in e ? e.response : undefined }); + } + } +} \ No newline at end of file diff --git a/src/backend/common/vendor/atproto/AbstractATProtoApiClient.ts b/src/backend/common/vendor/atproto/AbstractATProtoApiClient.ts index 082a2419..6e5fe326 100644 --- a/src/backend/common/vendor/atproto/AbstractATProtoApiClient.ts +++ b/src/backend/common/vendor/atproto/AbstractATProtoApiClient.ts @@ -1,66 +1,64 @@ import { getRoot } from "../../../ioc.js"; import { AbstractApiOptions } from "../../infrastructure/Atomic.js"; -import { TealClientData } from "../../infrastructure/config/client/tealfm.js"; import AbstractApiClient from "../AbstractApiClient.js"; -import { Agent, ComAtprotoRepoListRecords } from "@atproto/api"; +import { Agent } from "@atproto/api"; import { MSCache } from "../../Cache.js"; import { UpstreamError } from "../../errors/UpstreamError.js"; import { streamBodyProgress } from "../../../utils/NetworkUtils.js"; -import { ATProtoUserIdentifierData } from "../../infrastructure/config/client/atproto.js"; -import { getATProtoIdentifier, checkPds } from "./atUtils.js"; +import { ATProtoUserIdentifierData, HandleData } from "../../infrastructure/config/client/atproto.js"; +import { checkPds, isDID, identifierToAtProtoHandle } from "./atUtils.js"; +import { Client, isXRPCErrorPayload } from '@atcute/client'; +import { ComAtprotoSyncGetRepo } from '@atcute/atproto'; +import { AtprotoDid } from "@atcute/lexicons/syntax"; export abstract class AbstractATProtoApiClient extends AbstractApiClient { agent!: Agent; - cache: MSCache; + declare config: ATProtoUserIdentifierData; - constructor(name: any, config: TealClientData, options: AbstractApiOptions) { - super('atproto', name, config, options); + declare client: Client; - this.cache = getRoot().items.cache(); - } + userData!: HandleData - abstract initClient(): Promise; + cache: MSCache; - abstract restoreSession(): Promise; + constructor(name: any, config: ATProtoUserIdentifierData, options: AbstractApiOptions) { + super('atproto', name, config, options); + this.cache = getRoot().items.cache(); - async listRecord(collection: string, options: {limit?: number, cursor?: string} = {}): Promise { - const {limit = 20, cursor} = options; - try { - // records are returned newest to oldest - const response = await this.agent.com.atproto.repo.listRecords({ - repo: this.agent.sessionManager.did, - collection, - limit, - cursor // cursor TID is EXCLUSIVE IE first record returned will be the first older than cursor - }); - return response; - } catch (e) { - throw new UpstreamError(`Failed to list scrobble record`, { cause: e, response: 'response' in e ? e.response : undefined }); + const cleanIdentifier = this.config.identifier; + if(isDID(cleanIdentifier)) { + this.logger.debug(`Identifier ${cleanIdentifier} looks like a DID, skipping parsing as a handle.`); + this.config.did = cleanIdentifier; + } else { + this.config.identifier = identifierToAtProtoHandle(this.config.identifier, {logger: this.logger, defaultDomain: 'bsky.social'}); } } + abstract initClient(): Promise; + async checkPds(data: ATProtoUserIdentifierData): Promise { return await checkPds(data, {logger: this.logger, cache: this.cache.cacheAuth}); } - async getCAR() { - const resp = await this.agent.sessionManager.fetchHandler(`/xrpc/com.atproto.sync.getRepo?did=${encodeURIComponent(this.agent.sessionManager.did)}`, { - method: 'GET', - // @ts-expect-error - duplex: 'half', - redirect: 'follow', - headers: { - ...(Object.fromEntries(this.agent.headers.entries())), - Accept: 'application/vnd.ipld.car', - } + async getCAR(did: AtprotoDid) { + const resp = await this.client.call(ComAtprotoSyncGetRepo, { + params: { + did + }, + as: 'stream' }); - if(resp.status !== 200) { - const text = await resp.text(); + if(!resp.ok) { + let text: string; + if(isXRPCErrorPayload(resp.data)) { + text = resp.data.error; + } throw new UpstreamError(`Failed to fetch repo CAR file. Response was ${resp.status} with response ${text}`, {responseBody: text}); } - return await streamBodyProgress(resp, { + + resp.headers + return await streamBodyProgress(resp.data, { logger: this.logger, chunkDefaultSize: 1024 * 1024 * 5, // report progress every 5 MB fileHint: 'repo CAR' diff --git a/src/backend/common/vendor/atproto/atUtils.ts b/src/backend/common/vendor/atproto/atUtils.ts index 1464e46d..12a0cbd4 100644 --- a/src/backend/common/vendor/atproto/atUtils.ts +++ b/src/backend/common/vendor/atproto/atUtils.ts @@ -105,15 +105,17 @@ export const getATProtoIdentifier = async (data: ATProtoUserIdentifierData, opts identifier } = data; - assert(isAtprotoDid(givenDid), `Given DID is not an ATProto DID: ${givenDid}`); - let did: AtprotoDid = givenDid; - if (did === undefined) { + let did: AtprotoDid; + if (givenDid === undefined) { try { did = await handleResolver.resolve(identifier as `${string}.${string}`); logger.debug(`Resolved ${did}`); } catch (e) { throw new Error('Unable to resolve handle', { cause: e }); } + } else { + assert(isAtprotoDid(givenDid), `Given DID is not an ATProto DID: ${givenDid}`); + did = givenDid; } const docResolver = new CompositeDidDocumentResolver({ diff --git a/src/backend/common/vendor/teal/TealApiClient.ts b/src/backend/common/vendor/teal/TealApiClient.ts index e39b08df..debe4e04 100644 --- a/src/backend/common/vendor/teal/TealApiClient.ts +++ b/src/backend/common/vendor/teal/TealApiClient.ts @@ -7,23 +7,24 @@ import { MSCache } from "../../Cache.js"; import { AbstractApiOptions, PagelessListensTimeRangeOptions, PagelessTimeRangeListens, PagelessTimeRangeListensResult } from "../../infrastructure/Atomic.js"; import { ListRecord, RecordOptions, TealClientData } from "../../infrastructure/config/client/tealfm.js"; import AbstractApiClient from "../AbstractApiClient.js"; -import { AbstractATProtoApiClient } from "../atproto/AbstractATProtoApiClient.js"; import { ATProtoAppApiClient } from "../atproto/ATProtoAppApiClient.js"; import { ATProtoOauthApiClient } from "../atproto/ATProtoOauthApiClient.js"; import { Duration } from "dayjs/plugin/duration.js"; import { FmTealAlphaActorStatus, FmTealAlphaFeedPlay } from "./lexicons/index.js"; import { ScrobbleSubmitError } from "../../errors/MSErrors.js"; -import { ComAtprotoRepoCreateRecord, ComAtprotoRepoPutRecord } from "@atproto/api"; import { getScrobbleTsSOCDateWithContext, usecToUnix } from "../../../utils/TimeUtils.js"; import { musicServiceToCononical } from "../listenbrainz/lzUtils.js"; import { parseRegexSingle } from "@foxxmd/regex-buddy-core"; import { decodeTid, generateTID } from "@ewanc26/tid"; +import { ATProtoAuthenticatedApiClient } from "../atproto/ATProtoAuthenticatedApiClient.js"; +import { UpstreamError } from "../../errors/UpstreamError.js"; +import { ComAtprotoRepoCreateRecord, ComAtprotoRepoPutRecord } from '@atcute/atproto'; export class TealApiClient extends AbstractApiClient implements PagelessTimeRangeListens { declare config: TealClientData; - declare client: AbstractATProtoApiClient; + declare client: ATProtoAuthenticatedApiClient; cache: MSCache; @@ -43,29 +44,35 @@ export class TealApiClient extends AbstractApiClient implements PagelessTimeRang async createScrobbleRecord(record: FmTealAlphaFeedPlay.Main): Promise { - const input: ComAtprotoRepoCreateRecord.InputSchema = { - repo: this.client.agent.sessionManager.did, - collection: "fm.teal.alpha.feed.play", + const input: ComAtprotoRepoCreateRecord.$input = { + repo: this.client.userData.did, + collection: 'fm.teal.alpha.feed.play', record }; try { - const resp = await this.client.agent.com.atproto.repo.createRecord(input); - return {payload: input, response: resp.data}; + const res = await this.client.client.post('com.atproto.repo.createRecord', { + input, + params: {} + }); + return {payload: input, response: res.data}; } catch (e) { throw new ScrobbleSubmitError(`Failed to create record for scrobble`, { cause: e, payload: input, response: 'response' in e ? e.response : undefined }); } } async updateStatusRecord(record: FmTealAlphaActorStatus.Main): Promise { - const input: ComAtprotoRepoPutRecord.InputSchema = { - repo: this.client.agent.sessionManager.did, + const input: ComAtprotoRepoPutRecord.$input = { + repo: this.client.userData.did, collection: "fm.teal.alpha.actor.status", rkey: "self", record }; try { - const resp = await this.client.agent.com.atproto.repo.putRecord(input); - return {payload: input, response: resp.data}; + const res = await this.client.client.post('com.atproto.repo.putRecord', { + input, + params: {} + }); + return {payload: input, response: res.data}; } catch (e) { throw new ScrobbleSubmitError(`Failed to update status record for scrobble`, { cause: e, payload: input, response: 'response' in e ? e.response : undefined }); } @@ -83,7 +90,19 @@ export class TealApiClient extends AbstractApiClient implements PagelessTimeRang cursor = generateTID(dayjs.unix(to).toISOString()); } - const resp = await this.client.listRecord("fm.teal.alpha.feed.play", {cursor, limit}); + const resp = await this.client.client.get('com.atproto.repo.listRecords', { + params: { + repo: this.client.userData.did, + collection: "fm.teal.alpha.feed.play", + limit, + cursor + } + }); + + if(!resp.ok) { + throw new UpstreamError('Fetching records from PDS failed', {cause: resp.data}); + } + let fromTS: UnixTimestamp; if(resp.data.cursor !== undefined) { const { timestampUs } = decodeTid(resp.data.cursor); diff --git a/src/backend/scrobblers/TealfmScrobbler.ts b/src/backend/scrobblers/TealfmScrobbler.ts index cb563e7b..de38ab5d 100644 --- a/src/backend/scrobblers/TealfmScrobbler.ts +++ b/src/backend/scrobblers/TealfmScrobbler.ts @@ -16,7 +16,6 @@ import { nowPlayingUpdateByPlayDuration, shouldClearNPStatus } from "./AbstractS import { TealClientConfig } from "../common/infrastructure/config/client/tealfm.js"; import { ATProtoAppApiClient } from "../common/vendor/atproto/ATProtoAppApiClient.js"; import { ATProtoOauthApiClient } from "../common/vendor/atproto/ATProtoOauthApiClient.js"; -import { AbstractATProtoApiClient } from "../common/vendor/atproto/AbstractATProtoApiClient.js"; import { playToRecord, TealApiClient } from "../common/vendor/teal/TealApiClient.js"; import { playToStatusRecord } from "../common/vendor/teal/TealApiClient.js"; import { nowPlayingExpirationDuration } from "../common/vendor/teal/TealApiClient.js"; @@ -48,14 +47,6 @@ export default class TealScrobbler extends AbstractHistoricalScrobbleClient { this.scrobbleDelay = 1500; this.supportsNowPlaying = true; this.client = new TealApiClient(name, config.data, {...options, logger}); - // if(config.data.appPassword !== undefined) { - // this.client = new BlueSkyAppApiClient(name, config.data, {...options, logger}); - // this.requiresAuthInteraction = false; - // } else if(config.data.baseUri !== undefined) { - // this.client = new BlueSkyOauthApiClient(name, config.data, {...options, logger}); - // } else { - // throw new Error(`Must define either 'baseUri' or 'appPassword' in configuration!`); - // } this.nowPlayingMaxThreshold = nowPlayingUpdateByPlayDuration; this.nowPlayingMinThreshold = (_) => 20; this.configDir = options.configDir; @@ -211,7 +202,7 @@ export default class TealScrobbler extends AbstractHistoricalScrobbleClient { // 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`); - await fsPromise.writeFile(filename, Buffer.from(((await this.client.client.getCAR())))); + await fsPromise.writeFile(filename, Buffer.from(((await this.client.client.getCAR(this.client.client.userData.did))))); return filename; } @@ -227,7 +218,7 @@ export default class TealScrobbler extends AbstractHistoricalScrobbleClient { await using repo = fromStream(stream); - const did = this.client?.client?.agent?.sessionManager?.did; + const did = this.client.client.userData.did; let batch: RepositoryCreatePlayHistoricalOpts[] = []; let allGood = true; diff --git a/src/backend/utils/NetworkUtils.ts b/src/backend/utils/NetworkUtils.ts index 9f143ab3..f8f26d4a 100644 --- a/src/backend/utils/NetworkUtils.ts +++ b/src/backend/utils/NetworkUtils.ts @@ -272,24 +272,26 @@ export const wsReadyStateToStr = (state: number): string => { export type StreamBodyOpts = { logger?: Logger, chunkDefaultSize?: number, - fileHint?: string + fileHint?: string, + headers?: Headers } -export const streamBodyProgress = async (response: Response, opts: StreamBodyOpts = {}) => { +export const streamBodyProgress = async (stream: ReadableStream>, opts: StreamBodyOpts = {}) => { const { logger = loggerNoop, chunkDefaultSize = 1024 * 1024 * 10, // default to every 10MB, when we don't know response size - fileHint = 'file' + fileHint = 'file', + headers } = opts; let loading = true, chunks: any[] = []; - const reader = response.body.getReader(); + const reader = stream.getReader(); let length: number, chunkReportSize: number = chunkDefaultSize, lastReportedSize: number = 0; - if(null !== response.headers.get('content-length')) { - length = +response.headers.get('content-length'); + if(headers !== undefined && null !== headers.get('content-length')) { + length = +headers.get('content-length'); const [summary, size, unit] = formatBytes(length); if(unit === 'MiB' && size > 10) { switch(true) { -- 2.51.2