feat(tealfm): Stream repo CAR to monitor download progress

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<PagelessTimeRangeListensResult> {

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;
     }
 

src/backend/utils/DataUtils.ts

@@ -203,4 +203,24 @@ export const getCommonComponentEnvConfig = (prefix: string): Partial<CommonConfi
         name: nonEmptyStringOrDefault(process.env[`${prefix}_NAME`], undefined),
         enable: e !== undefined ? parseBoolStrict(e) : undefined
     }, false);
+}
+
+const byteSizes = ['Bytes', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB', 'EiB', 'ZiB', 'YiB']
+
+/**
+ * Convert bytes into human readable size
+ * 
+ * @see https://stackoverflow.com/a/18650828/1469797
+ */
+export const formatBytes = (bytes: number, decimals: number = 2): [string, number, string] => {
+    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];
 }

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();
     }
-}
+}
+
+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;
+  }