From 18c520ce2f007cd8f9903400a6e69fedb2a7ab6b Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Tue, 2 Jun 2026 16:26:05 +0000 Subject: [PATCH] feat(tealfm): Stream repo CAR to monitor download progress --- .../bluesky/AbstractBlueSkyApiClient.ts | 22 +++++- src/backend/scrobblers/TealfmScrobbler.ts | 11 ++- src/backend/utils/DataUtils.ts | 20 +++++ src/backend/utils/NetworkUtils.ts | 79 ++++++++++++++++++- 4 files changed, 125 insertions(+), 7 deletions(-) diff --git a/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts b/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts index f9e0b476..02a36812 100644 --- a/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts +++ b/src/backend/common/vendor/bluesky/AbstractBlueSkyApiClient.ts @@ -16,6 +16,7 @@ import { ScrobbleSubmitError } from "../../errors/MSErrors.js"; import { UpstreamError } from "../../errors/UpstreamError.js"; import { decodeTid, generateTID } from '@ewanc26/tid'; import { Duration } from "dayjs/plugin/duration.js"; +import { streamBodyProgress } from "../../../utils/NetworkUtils.js"; export abstract class AbstractBlueSkyApiClient extends AbstractApiClient implements PagelessTimeRangeListens { @@ -81,8 +82,25 @@ 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}); + 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', + } + }); + if(resp.status !== 200) { + const text = await resp.text(); + throw new UpstreamError(`Failed to fetch repo CAR file. Response was ${resp.status} with response ${text}`, {responseBody: text}); + } + return await streamBodyProgress(resp, { + logger: this.logger, + chunkDefaultSize: 1024 * 1024 * 5, // report progress every 5 MB + fileHint: 'repo CAR' + }); } async getPagelessTimeRangeListens(params: PagelessListensTimeRangeOptions): Promise { diff --git a/src/backend/scrobblers/TealfmScrobbler.ts b/src/backend/scrobblers/TealfmScrobbler.ts index 9e82e833..91a80b58 100644 --- a/src/backend/scrobblers/TealfmScrobbler.ts +++ b/src/backend/scrobblers/TealfmScrobbler.ts @@ -13,7 +13,7 @@ import { playToListenPayload } from '../common/vendor/listenbrainz/lzUtils.js'; import { Notifiers } from "../notifier/Notifiers.js"; import AbstractScrobbleClient, { nowPlayingUpdateByPlayDuration, shouldClearNPStatus } from "./AbstractScrobbleClient.js"; -import { TealClientConfig } from "../common/infrastructure/config/client/tealfm.js"; +import { ScrobbleRecord, 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, nowPlayingExpirationDuration, playToRecord, playToStatusRecord, recordToPlay } from "../common/vendor/bluesky/AbstractBlueSkyApiClient.js"; @@ -178,20 +178,22 @@ export default class TealScrobbler extends AbstractHistoricalScrobbleClient { } protected async doHydrateHistoricalScrobbles(opts: {allowFailures?: boolean, signal?: AbortSignal } = {}) { + const logger = childLogger(this.logger, ['Historical Plays']); const { allowFailures = false, signal } = opts; let file: string; try { + logger.verbose('Fetching scrobbles from PDS...'); file = await this.fetchCarToFile(); signal?.throwIfAborted(); } catch (e) { - throw new Error('Failed to fetch CAR repo file', {cause: e}); + throw new Error('Failed to fetch repo CAR', {cause: e}); } try { - await this.parseScrobblesFromCar(file, 100, {allowFailures, logger: childLogger(this.logger, ['Historical Plays']), signal}); + await this.parseScrobblesFromCar(file, 100, {allowFailures, logger: logger, signal}); } catch (e) { throw new Error('Failed to convert CAR without any error', {cause: e}); } finally { @@ -200,8 +202,9 @@ export default class TealScrobbler extends AbstractHistoricalScrobbleClient { } async fetchCarToFile() { + const filename = path.resolve(this.configDir, `${this.getSafeExternalId()}-${dayjs().unix()}.car`); - await fsPromise.writeFile(filename, Buffer.from(((await this.client.getCAR()).data))); + await fsPromise.writeFile(filename, Buffer.from(((await this.client.getCAR())))); return filename; } diff --git a/src/backend/utils/DataUtils.ts b/src/backend/utils/DataUtils.ts index 5f1567d3..12d0ca84 100644 --- a/src/backend/utils/DataUtils.ts +++ b/src/backend/utils/DataUtils.ts @@ -203,4 +203,24 @@ export const getCommonComponentEnvConfig = (prefix: string): Partial { + if (!+bytes) return ['0 Bytes', 0, 'Byes']; + + const k = 1024 + const dm = decimals < 0 ? 0 : decimals + + const i = Math.floor(Math.log(bytes) / Math.log(k)) + + const friendlySize = parseFloat((bytes / Math.pow(k, i)).toFixed(dm)); + const friendlyUnit = byteSizes[i]; + return [`${friendlySize} ${friendlyUnit}`, friendlySize, friendlyUnit]; } \ No newline at end of file diff --git a/src/backend/utils/NetworkUtils.ts b/src/backend/utils/NetworkUtils.ts index 410ea225..9f143ab3 100644 --- a/src/backend/utils/NetworkUtils.ts +++ b/src/backend/utils/NetworkUtils.ts @@ -8,6 +8,8 @@ import { getFirstNonEmptyVal, isDebugMode} from "../utils.js"; import { URLData } from "../../core/Atomic.js"; import { CloseEvent, ErrorEvent, RetryEvent } from 'iso-websocket' import { WEBSOCKET_CLOSE_CODE_REASONS } from "../common/infrastructure/Atomic.js"; +import { loggerNoop } from "../common/MaybeLogger.js"; +import { formatBytes } from "./DataUtils.js"; export interface PortReachableOpts { host: string, @@ -265,4 +267,79 @@ export const wsReadyStateToStr = (state: number): string => { default: return state.toString(); } -} \ No newline at end of file +} + +export type StreamBodyOpts = { + logger?: Logger, + chunkDefaultSize?: number, + fileHint?: string +} + +export const streamBodyProgress = async (response: Response, opts: StreamBodyOpts = {}) => { + const { + logger = loggerNoop, + chunkDefaultSize = 1024 * 1024 * 10, // default to every 10MB, when we don't know response size + fileHint = 'file' + } = opts; + let loading = true, + chunks: any[] = []; + const reader = response.body.getReader(); + + let length: number, + chunkReportSize: number = chunkDefaultSize, + lastReportedSize: number = 0; + if(null !== response.headers.get('content-length')) { + length = +response.headers.get('content-length'); + const [summary, size, unit] = formatBytes(length); + if(unit === 'MiB' && size > 10) { + switch(true) { + case(size > 50): + chunkReportSize = length/10; + break; + case(size > 20): + chunkReportSize = length/5; + break; + default: + chunkReportSize = length/3; + break; + } + } + logger.trace(`Downloading ${summary} ${fileHint}...`); + } else { + logger.trace(`Downloading ${fileHint} of unknown size (no content-length header)...`); + } + + let received = 0; + + // Loop through the response stream and extract data chunks + while (loading) { + const { done, value } = await reader.read(); + if (done) { + // Finish loading + loading = false; + } else { + // Push values to the chunk array + chunks.push(value); + + received += value.length; + lastReportedSize += value.length; + if(lastReportedSize >= chunkReportSize) { + logger.trace(`Downloaded ${formatBytes(received)[0]}...`); + lastReportedSize = 0; + } + } + } + logger.trace(`Finished download!`); + + // Concat the chunks into a single array + let body = new Uint8Array(received); + let position = 0; + + // Order the chunks by their respective position + for (let chunk of chunks) { + body.set(chunk, position); + position += chunk.length; + } + + return body; + } \ No newline at end of file