[READ-ONLY] Mirror of https://github.com/FoxxMD/multi-scrobbler. Scrobble plays from multiple sources to multiple clients
docs.multi-scrobbler.app
deezer
docker
jellyfin
koito
lastfm
listenbrainz
maloja
mopidy
mpris
music
music-assistant
plex
scrobble
self-hosted
spotify
subsonic
tautulli
youtube-music
refactor: Replace @atproto with @atcute for abstract/app api implementation
+175
-112
+51
-39
src/backend/common/vendor/atproto/ATProtoAppApiClient.ts
+51
-39
src/backend/common/vendor/atproto/ATProtoAppApiClient.ts
···
1
1
import { AbstractApiOptions } from "../../infrastructure/Atomic.js";
2
2
import { TealClientData } from "../../infrastructure/config/client/tealfm.js";
3
-
import { Agent, CredentialSession, AtpSessionEvent, AtpSessionData } from "@atproto/api";
4
-
import { AbstractATProtoApiClient } from "./AbstractATProtoApiClient.js";
5
-
import { getATProtoIdentifier, identifierToAtProtoHandle, isDID } from "./atUtils.js";
3
+
import { getATProtoIdentifier } from "./atUtils.js";
6
4
import { ATProtoAppData, ATProtoUserIdentifierData } from "../../infrastructure/config/client/atproto.js";
5
+
import { ATProtoAuthenticatedApiClient } from "./ATProtoAuthenticatedApiClient.js";
6
+
import { PasswordSession, PasswordSessionData } from '@atcute/password-session';
7
+
import { Client } from "@atcute/client";
7
8
8
-
export class ATProtoAppApiClient extends AbstractATProtoApiClient {
9
+
export class ATProtoAppApiClient extends ATProtoAuthenticatedApiClient {
9
10
10
11
declare config: ATProtoUserIdentifierData & ATProtoAppData;
11
-
appSession?: CredentialSession;
12
-
appPwAuth: boolean
13
-
14
12
15
13
constructor(name: any, config: TealClientData, options: AbstractApiOptions) {
16
14
super(name, config, options);
17
-
this.logger.verbose(`Using App Password auth for session`);
18
-
const cleanIdentifier = this.config.identifier;
19
-
if(isDID(cleanIdentifier)) {
20
-
this.logger.debug(`Identifier ${cleanIdentifier} looks like a DID, skipping parsing as a handle.`);
21
-
this.config.did = cleanIdentifier;
22
-
} else {
23
-
this.config.identifier = identifierToAtProtoHandle(this.config.identifier, {logger: this.logger, defaultDomain: 'bsky.social'});
24
-
}
25
-
if(this.config.appPassword === undefined) {
15
+
if (this.config.appPassword === undefined) {
26
16
throw new Error('Must provide app password');
27
17
}
18
+
this.logger.verbose(`Using App Password auth for session`);
28
19
}
29
20
30
21
async initClient(): Promise<void> {
31
-
const hd = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth});
32
-
this.logger.verbose(`Using ${hd.did} on PDS ${hd.pds}`);
33
-
this.appSession = new CredentialSession(new URL(hd.pds), undefined, (evt: AtpSessionEvent, sess?: AtpSessionData) => {
34
-
this.cache.cacheAuth.set(`appPwSession-${this.name}-${hd.did}`, sess, '1000h');
35
-
});
36
-
this.agent = new Agent(this.appSession);
22
+
this.userData = await getATProtoIdentifier(this.config, { logger: this.logger, cache: this.cache.cacheAuth });
23
+
this.logger.verbose(`Using ${this.userData.did} on PDS ${this.userData.pds}`);
37
24
}
38
25
39
26
restoreSession = async (): Promise<boolean> => {
40
-
const hd = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth});
41
-
const savedSession = await this.cache.cacheAuth.get<AtpSessionData>(`appPwSession-${this.name}-${hd.did}`);
42
-
if (savedSession !== undefined) {
27
+
const savedSessionCute = await this.getSession();
28
+
if (savedSessionCute !== undefined) {
29
+
const that = this;
43
30
try {
44
-
this.logger.debug('Found existing session, trying to resume...');
45
-
await this.appSession.resumeSession(savedSession);
46
-
this.logger.debug('Resumed session!');
47
-
return true;
31
+
const session = await PasswordSession.resume(savedSessionCute, {
32
+
async onUpdate(data) {
33
+
// called on login and token refresh — persist the session
34
+
await that.saveSession(data);
35
+
},
36
+
async onDelete(data) {
37
+
// called on logout or session invalidation — clean up
38
+
await that.deleteSession();
39
+
},
40
+
});
41
+
this.client = new Client({ handler: session });
48
42
} catch (e) {
49
43
this.logger.warn(new Error('Could not resume app password session from data', { cause: e }));
50
44
return false;
51
45
}
52
46
}
53
-
this.logger.debug('No app password session data to restore');
47
+
}
48
+
49
+
protected async saveSession(data: PasswordSessionData): Promise<void> {
50
+
await this.cache.cacheAuth.set(`appPwSessionCute-${this.name}-${this.userData.did}`, data, '1000h');
51
+
}
52
+
53
+
protected async getSession(): Promise<PasswordSessionData> {
54
+
return await this.cache.cacheAuth.get<PasswordSessionData>(`appPwSessionCute-${this.name}-${this.userData.did}`);
55
+
}
56
+
57
+
protected async deleteSession(): Promise<void> {
58
+
await this.cache.cacheAuth.delete(`appPwSessionCute-${this.name}-${this.userData.did}`);
54
59
}
55
60
56
61
appLogin = async (): Promise<boolean> => {
62
+
const that = this;
57
63
try {
64
+
const session = await PasswordSession.login(
65
+
{ service: this.userData.pds, identifier: this.userData.handle, password: this.config.appPassword },
66
+
{
67
+
//session: savedSession,
68
+
async onUpdate(data) {
69
+
// called on login and token refresh — persist the session
70
+
await that.saveSession(data);
71
+
},
72
+
async onDelete(data) {
73
+
// called on logout or session invalidation — clean up
74
+
await that.deleteSession();
75
+
},
76
+
},
77
+
);
58
78
59
-
const f = await this.appSession.login({
60
-
identifier: this.config.identifier,
61
-
password: this.config.appPassword
62
-
});
63
-
if (!f.success) {
64
-
this.logger.error('Login was not successful with app password');
65
-
return false;
66
-
}
67
-
this.logger.debug('Logged in.');
79
+
this.client = new Client({ handler: session });
68
80
return true;
69
81
} catch (e) {
70
82
this.logger.error(new Error('Could not login using app password', { cause: e }));
+5
src/backend/common/vendor/atproto/ATProtoAuthenticatedApiClient.ts
+5
src/backend/common/vendor/atproto/ATProtoAuthenticatedApiClient.ts
+2
-3
src/backend/common/vendor/atproto/ATProtoOauthApiClient.ts
+2
-3
src/backend/common/vendor/atproto/ATProtoOauthApiClient.ts
···
8
8
OAuthSession,
9
9
} from "@atproto/oauth-client-node";
10
10
import { Agent } from "@atproto/api";
11
-
import { AbstractATProtoApiClient } from "./AbstractATProtoApiClient.js";
11
+
import { ATProtoAuthenticatedApiClient } from "./ATProtoAuthenticatedApiClient.js";
12
12
13
-
14
-
export class ATProtoOauthApiClient extends AbstractATProtoApiClient {
13
+
export class ATProtoOauthApiClient extends ATProtoAuthenticatedApiClient {
15
14
16
15
declare config: TealClientData;
17
16
+35
src/backend/common/vendor/atproto/ATProtoUnauthenticatedApiClient.ts
+35
src/backend/common/vendor/atproto/ATProtoUnauthenticatedApiClient.ts
···
1
+
import { UpstreamError } from "../../errors/UpstreamError.js";
2
+
import { HandleData } from "../../infrastructure/config/client/atproto.js";
3
+
import { AbstractATProtoApiClient } from "./AbstractATProtoApiClient.js";
4
+
import { getATProtoIdentifier, checkPds } from "./atUtils.js";
5
+
import { Client, simpleFetchHandler } from '@atcute/client';
6
+
import type {} from '@atcute/atproto';
7
+
import { Nsid } from "@atcute/lexicons";
8
+
9
+
export class ATProtoUnauthenticatedApiClient extends AbstractATProtoApiClient {
10
+
11
+
declare client: Client;
12
+
13
+
async initClient(): Promise<void> {
14
+
this.userData = await getATProtoIdentifier(this.config, {logger: this.logger, cache: this.cache.cacheAuth});
15
+
this.client = new Client({ handler: simpleFetchHandler({ service: this.userData.pds }) });
16
+
}
17
+
18
+
async listRecords(collection: string, options: {limit?: number, cursor?: string} = {}) {
19
+
const {limit = 20, cursor} = options;
20
+
try {
21
+
// records are returned newest to oldest
22
+
const response = await this.client.get('com.atproto.repo.listRecords', {
23
+
params: {
24
+
repo: this.userData.did,
25
+
collection: collection as Nsid,
26
+
limit,
27
+
cursor
28
+
}
29
+
});
30
+
return response;
31
+
} catch (e) {
32
+
throw new UpstreamError(`Failed to list scrobble record`, { cause: e, response: 'response' in e ? e.response : undefined });
33
+
}
34
+
}
35
+
}
+35
-37
src/backend/common/vendor/atproto/AbstractATProtoApiClient.ts
+35
-37
src/backend/common/vendor/atproto/AbstractATProtoApiClient.ts
···
1
1
import { getRoot } from "../../../ioc.js";
2
2
import { AbstractApiOptions } from "../../infrastructure/Atomic.js";
3
-
import { TealClientData } from "../../infrastructure/config/client/tealfm.js";
4
3
import AbstractApiClient from "../AbstractApiClient.js";
5
-
import { Agent, ComAtprotoRepoListRecords } from "@atproto/api";
4
+
import { Agent } from "@atproto/api";
6
5
import { MSCache } from "../../Cache.js";
7
6
import { UpstreamError } from "../../errors/UpstreamError.js";
8
7
import { streamBodyProgress } from "../../../utils/NetworkUtils.js";
9
-
import { ATProtoUserIdentifierData } from "../../infrastructure/config/client/atproto.js";
10
-
import { getATProtoIdentifier, checkPds } from "./atUtils.js";
8
+
import { ATProtoUserIdentifierData, HandleData } from "../../infrastructure/config/client/atproto.js";
9
+
import { checkPds, isDID, identifierToAtProtoHandle } from "./atUtils.js";
10
+
import { Client, isXRPCErrorPayload } from '@atcute/client';
11
+
import { ComAtprotoSyncGetRepo } from '@atcute/atproto';
12
+
import { AtprotoDid } from "@atcute/lexicons/syntax";
11
13
12
14
export abstract class AbstractATProtoApiClient extends AbstractApiClient {
13
15
14
16
agent!: Agent;
15
17
16
-
cache: MSCache;
18
+
declare config: ATProtoUserIdentifierData;
17
19
18
-
constructor(name: any, config: TealClientData, options: AbstractApiOptions) {
19
-
super('atproto', name, config, options);
20
+
declare client: Client;
20
21
21
-
this.cache = getRoot().items.cache();
22
-
}
22
+
userData!: HandleData
23
23
24
-
abstract initClient(): Promise<void>;
24
+
cache: MSCache;
25
25
26
-
abstract restoreSession(): Promise<boolean>;
26
+
constructor(name: any, config: ATProtoUserIdentifierData, options: AbstractApiOptions) {
27
+
super('atproto', name, config, options);
28
+
this.cache = getRoot().items.cache();
27
29
28
-
async listRecord(collection: string, options: {limit?: number, cursor?: string} = {}): Promise<ComAtprotoRepoListRecords.Response> {
29
-
const {limit = 20, cursor} = options;
30
-
try {
31
-
// records are returned newest to oldest
32
-
const response = await this.agent.com.atproto.repo.listRecords({
33
-
repo: this.agent.sessionManager.did,
34
-
collection,
35
-
limit,
36
-
cursor // cursor TID is EXCLUSIVE IE first record returned will be the first older than cursor
37
-
});
38
-
return response;
39
-
} catch (e) {
40
-
throw new UpstreamError(`Failed to list scrobble record`, { cause: e, response: 'response' in e ? e.response : undefined });
30
+
const cleanIdentifier = this.config.identifier;
31
+
if(isDID(cleanIdentifier)) {
32
+
this.logger.debug(`Identifier ${cleanIdentifier} looks like a DID, skipping parsing as a handle.`);
33
+
this.config.did = cleanIdentifier;
34
+
} else {
35
+
this.config.identifier = identifierToAtProtoHandle(this.config.identifier, {logger: this.logger, defaultDomain: 'bsky.social'});
41
36
}
42
37
}
38
+
39
+
abstract initClient(): Promise<void>;
43
40
44
41
async checkPds(data: ATProtoUserIdentifierData): Promise<true> {
45
42
return await checkPds(data, {logger: this.logger, cache: this.cache.cacheAuth});
46
43
}
47
44
48
-
async getCAR() {
49
-
const resp = await this.agent.sessionManager.fetchHandler(`/xrpc/com.atproto.sync.getRepo?did=${encodeURIComponent(this.agent.sessionManager.did)}`, {
50
-
method: 'GET',
51
-
// @ts-expect-error
52
-
duplex: 'half',
53
-
redirect: 'follow',
54
-
headers: {
55
-
...(Object.fromEntries(this.agent.headers.entries())),
56
-
Accept: 'application/vnd.ipld.car',
57
-
}
45
+
async getCAR(did: AtprotoDid) {
46
+
const resp = await this.client.call(ComAtprotoSyncGetRepo, {
47
+
params: {
48
+
did
49
+
},
50
+
as: 'stream'
58
51
});
59
-
if(resp.status !== 200) {
60
-
const text = await resp.text();
52
+
if(!resp.ok) {
53
+
let text: string;
54
+
if(isXRPCErrorPayload(resp.data)) {
55
+
text = resp.data.error;
56
+
}
61
57
throw new UpstreamError(`Failed to fetch repo CAR file. Response was ${resp.status} with response ${text}`, {responseBody: text});
62
58
}
63
-
return await streamBodyProgress(resp, {
59
+
60
+
resp.headers
61
+
return await streamBodyProgress(resp.data, {
64
62
logger: this.logger,
65
63
chunkDefaultSize: 1024 * 1024 * 5, // report progress every 5 MB
66
64
fileHint: 'repo CAR'
+5
-3
src/backend/common/vendor/atproto/atUtils.ts
+5
-3
src/backend/common/vendor/atproto/atUtils.ts
···
105
105
identifier
106
106
} = data;
107
107
108
-
assert(isAtprotoDid(givenDid), `Given DID is not an ATProto DID: ${givenDid}`);
109
-
let did: AtprotoDid = givenDid;
110
-
if (did === undefined) {
108
+
let did: AtprotoDid;
109
+
if (givenDid === undefined) {
111
110
try {
112
111
did = await handleResolver.resolve(identifier as `${string}.${string}`);
113
112
logger.debug(`Resolved ${did}`);
114
113
} catch (e) {
115
114
throw new Error('Unable to resolve handle', { cause: e });
116
115
}
116
+
} else {
117
+
assert(isAtprotoDid(givenDid), `Given DID is not an ATProto DID: ${givenDid}`);
118
+
did = givenDid;
117
119
}
118
120
119
121
const docResolver = new CompositeDidDocumentResolver({
+32
-13
src/backend/common/vendor/teal/TealApiClient.ts
+32
-13
src/backend/common/vendor/teal/TealApiClient.ts
···
7
7
import { AbstractApiOptions, PagelessListensTimeRangeOptions, PagelessTimeRangeListens, PagelessTimeRangeListensResult } from "../../infrastructure/Atomic.js";
8
8
import { ListRecord, RecordOptions, TealClientData } from "../../infrastructure/config/client/tealfm.js";
9
9
import AbstractApiClient from "../AbstractApiClient.js";
10
-
import { AbstractATProtoApiClient } from "../atproto/AbstractATProtoApiClient.js";
11
10
import { ATProtoAppApiClient } from "../atproto/ATProtoAppApiClient.js";
12
11
import { ATProtoOauthApiClient } from "../atproto/ATProtoOauthApiClient.js";
13
12
import { Duration } from "dayjs/plugin/duration.js";
14
13
import { FmTealAlphaActorStatus, FmTealAlphaFeedPlay } from "./lexicons/index.js";
15
14
import { ScrobbleSubmitError } from "../../errors/MSErrors.js";
16
-
import { ComAtprotoRepoCreateRecord, ComAtprotoRepoPutRecord } from "@atproto/api";
17
15
import { getScrobbleTsSOCDateWithContext, usecToUnix } from "../../../utils/TimeUtils.js";
18
16
import { musicServiceToCononical } from "../listenbrainz/lzUtils.js";
19
17
import { parseRegexSingle } from "@foxxmd/regex-buddy-core";
20
18
import { decodeTid, generateTID } from "@ewanc26/tid";
19
+
import { ATProtoAuthenticatedApiClient } from "../atproto/ATProtoAuthenticatedApiClient.js";
20
+
import { UpstreamError } from "../../errors/UpstreamError.js";
21
+
import { ComAtprotoRepoCreateRecord, ComAtprotoRepoPutRecord } from '@atcute/atproto';
21
22
22
23
export class TealApiClient extends AbstractApiClient implements PagelessTimeRangeListens {
23
24
24
25
declare config: TealClientData;
25
26
26
-
declare client: AbstractATProtoApiClient;
27
+
declare client: ATProtoAuthenticatedApiClient;
27
28
28
29
cache: MSCache;
29
30
···
43
44
44
45
45
46
async createScrobbleRecord(record: FmTealAlphaFeedPlay.Main): Promise<ScrobbleActionResult> {
46
-
const input: ComAtprotoRepoCreateRecord.InputSchema = {
47
-
repo: this.client.agent.sessionManager.did,
48
-
collection: "fm.teal.alpha.feed.play",
47
+
const input: ComAtprotoRepoCreateRecord.$input = {
48
+
repo: this.client.userData.did,
49
+
collection: 'fm.teal.alpha.feed.play',
49
50
record
50
51
};
51
52
try {
52
-
const resp = await this.client.agent.com.atproto.repo.createRecord(input);
53
-
return {payload: input, response: resp.data};
53
+
const res = await this.client.client.post('com.atproto.repo.createRecord', {
54
+
input,
55
+
params: {}
56
+
});
57
+
return {payload: input, response: res.data};
54
58
} catch (e) {
55
59
throw new ScrobbleSubmitError(`Failed to create record for scrobble`, { cause: e, payload: input, response: 'response' in e ? e.response : undefined });
56
60
}
57
61
}
58
62
59
63
async updateStatusRecord(record: FmTealAlphaActorStatus.Main): Promise<ScrobbleActionResult> {
60
-
const input: ComAtprotoRepoPutRecord.InputSchema = {
61
-
repo: this.client.agent.sessionManager.did,
64
+
const input: ComAtprotoRepoPutRecord.$input = {
65
+
repo: this.client.userData.did,
62
66
collection: "fm.teal.alpha.actor.status",
63
67
rkey: "self",
64
68
record
65
69
};
66
70
try {
67
-
const resp = await this.client.agent.com.atproto.repo.putRecord(input);
68
-
return {payload: input, response: resp.data};
71
+
const res = await this.client.client.post('com.atproto.repo.putRecord', {
72
+
input,
73
+
params: {}
74
+
});
75
+
return {payload: input, response: res.data};
69
76
} catch (e) {
70
77
throw new ScrobbleSubmitError(`Failed to update status record for scrobble`, { cause: e, payload: input, response: 'response' in e ? e.response : undefined });
71
78
}
···
83
90
cursor = generateTID(dayjs.unix(to).toISOString());
84
91
}
85
92
86
-
const resp = await this.client.listRecord("fm.teal.alpha.feed.play", {cursor, limit});
93
+
const resp = await this.client.client.get('com.atproto.repo.listRecords', {
94
+
params: {
95
+
repo: this.client.userData.did,
96
+
collection: "fm.teal.alpha.feed.play",
97
+
limit,
98
+
cursor
99
+
}
100
+
});
101
+
102
+
if(!resp.ok) {
103
+
throw new UpstreamError('Fetching records from PDS failed', {cause: resp.data});
104
+
}
105
+
87
106
let fromTS: UnixTimestamp;
88
107
if(resp.data.cursor !== undefined) {
89
108
const { timestampUs } = decodeTid(resp.data.cursor);
+2
-11
src/backend/scrobblers/TealfmScrobbler.ts
+2
-11
src/backend/scrobblers/TealfmScrobbler.ts
···
16
16
import { TealClientConfig } from "../common/infrastructure/config/client/tealfm.js";
17
17
import { ATProtoAppApiClient } from "../common/vendor/atproto/ATProtoAppApiClient.js";
18
18
import { ATProtoOauthApiClient } from "../common/vendor/atproto/ATProtoOauthApiClient.js";
19
-
import { AbstractATProtoApiClient } from "../common/vendor/atproto/AbstractATProtoApiClient.js";
20
19
import { playToRecord, TealApiClient } from "../common/vendor/teal/TealApiClient.js";
21
20
import { playToStatusRecord } from "../common/vendor/teal/TealApiClient.js";
22
21
import { nowPlayingExpirationDuration } from "../common/vendor/teal/TealApiClient.js";
···
48
47
this.scrobbleDelay = 1500;
49
48
this.supportsNowPlaying = true;
50
49
this.client = new TealApiClient(name, config.data, {...options, logger});
51
-
// if(config.data.appPassword !== undefined) {
52
-
// this.client = new BlueSkyAppApiClient(name, config.data, {...options, logger});
53
-
// this.requiresAuthInteraction = false;
54
-
// } else if(config.data.baseUri !== undefined) {
55
-
// this.client = new BlueSkyOauthApiClient(name, config.data, {...options, logger});
56
-
// } else {
57
-
// throw new Error(`Must define either 'baseUri' or 'appPassword' in configuration!`);
58
-
// }
59
50
this.nowPlayingMaxThreshold = nowPlayingUpdateByPlayDuration;
60
51
this.nowPlayingMinThreshold = (_) => 20;
61
52
this.configDir = options.configDir;
···
211
202
// TODO use `since` to get CAR diff instead of entire repo
212
203
// can use last import date from migrations table
213
204
const filename = path.resolve(this.configDir, `${this.getSafeExternalId()}-${dayjs().unix()}.car`);
214
-
await fsPromise.writeFile(filename, Buffer.from(((await this.client.client.getCAR()))));
205
+
await fsPromise.writeFile(filename, Buffer.from(((await this.client.client.getCAR(this.client.client.userData.did)))));
215
206
return filename;
216
207
}
217
208
···
227
218
228
219
await using repo = fromStream(stream);
229
220
230
-
const did = this.client?.client?.agent?.sessionManager?.did;
221
+
const did = this.client.client.userData.did;
231
222
232
223
let batch: RepositoryCreatePlayHistoricalOpts[] = [];
233
224
let allGood = true;
+8
-6
src/backend/utils/NetworkUtils.ts
+8
-6
src/backend/utils/NetworkUtils.ts
···
272
272
export type StreamBodyOpts = {
273
273
logger?: Logger,
274
274
chunkDefaultSize?: number,
275
-
fileHint?: string
275
+
fileHint?: string,
276
+
headers?: Headers
276
277
}
277
278
278
-
export const streamBodyProgress = async (response: Response, opts: StreamBodyOpts = {}) => {
279
+
export const streamBodyProgress = async (stream: ReadableStream<Uint8Array<ArrayBufferLike>>, opts: StreamBodyOpts = {}) => {
279
280
const {
280
281
logger = loggerNoop,
281
282
chunkDefaultSize = 1024 * 1024 * 10, // default to every 10MB, when we don't know response size
282
-
fileHint = 'file'
283
+
fileHint = 'file',
284
+
headers
283
285
} = opts;
284
286
let loading = true,
285
287
chunks: any[] = [];
286
-
const reader = response.body.getReader();
288
+
const reader = stream.getReader();
287
289
288
290
let length: number,
289
291
chunkReportSize: number = chunkDefaultSize,
290
292
lastReportedSize: number = 0;
291
-
if(null !== response.headers.get('content-length')) {
292
-
length = +response.headers.get('content-length');
293
+
if(headers !== undefined && null !== headers.get('content-length')) {
294
+
length = +headers.get('content-length');
293
295
const [summary, size, unit] = formatBytes(length);
294
296
if(unit === 'MiB' && size > 10) {
295
297
switch(true) {