From c42fc908c73c64192d873a2b62427e490580af27 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tr=C3=A8sAbhi?= <43380238+tresabhi@users.noreply.github.com>
Date: Sun, 9 Jun 2024 17:44:38 -0500
Subject: [PATCH] feat: chunking

---
 scripts/discoverIds.ts  | 86 +++++++++++++++++++++++++++--------------
 src/core/streams/rts.ts |  6 +--
 2 files changed, 61 insertions(+), 31 deletions(-)

diff --git a/scripts/discoverIds.ts b/scripts/discoverIds.ts
index 5a8b9526a..9fadaf829 100644
--- a/scripts/discoverIds.ts
+++ b/scripts/discoverIds.ts
@@ -1,4 +1,4 @@
-import { times } from 'lodash';
+import { chunk, times } from 'lodash';
 import { compress, decompress } from 'lz4js';
 import { argv } from 'process';
 import { REGIONS, Region } from '../src/constants/regions';
@@ -9,6 +9,13 @@ import { asset } from '../src/core/blitzkit/asset';
 import { commitAssets } from '../src/core/blitzkit/commitAssets';
 import { DidsReadStream, DidsWriteStream } from '../src/core/streams/dids';
 
+interface DiscoverIdsManifest {
+  time: number;
+  chunks: number;
+  count: number;
+}
+
+const CHUNK_SIZE = 2 ** 21;
 const RUN_TIME = 1000 * 60 * 60 * 5.5;
 const PROGRESS_UPDATE_FREQUENCY = 1000 * 60;
 const MAX_REQUESTS = 10;
@@ -16,29 +23,40 @@ const ACCOUNTS_PER_CALL = 100;
 const TERMINATION_THRESHOLD = 1000;
 
 const production = argv.includes('--production');
-const startTime = Date.now();
-const indexableRegions = [...REGIONS];
-const preDiscoveredRaw = await fetch(
-  asset('definitions/ids.dids.lz4', !production),
-).then(async (response) => {
-  if (response.status === 200) {
-    const buffer = await response.arrayBuffer();
-    const decompressed = decompress(new Uint8Array(buffer)).buffer;
-    return new DidsReadStream(decompressed).dids();
-  }
-
-  return undefined;
-});
 
 console.log(`Running in ${production ? 'production' : 'development'} mode`);
 
-if (preDiscoveredRaw === undefined) {
-  console.log('No pre-discovered ids found :(');
-} else {
-  console.log(`Found ${preDiscoveredRaw.length} pre-discovered ids`);
+const startTime = Date.now();
+const indexableRegions = [...REGIONS];
+const ids: number[] = [];
+const preDiscoveredManifest = (await fetch(
+  asset('ids/manifest.json', !production),
+).then((response) => response.json())) as DiscoverIdsManifest;
+
+console.log(
+  `Fetching ${preDiscoveredManifest.chunks} pre-discovered chunks...`,
+);
+
+let chunkIndex = 0;
+while (chunkIndex < preDiscoveredManifest.chunks) {
+  const preDiscovered = await fetch(
+    asset(`ids/${chunkIndex}.dids.lz4`, !production),
+  ).then(async (response) => {
+    const buffer = await response.arrayBuffer();
+    const decompressed = decompress(new Uint8Array(buffer)).buffer;
+    return new DidsReadStream(decompressed).dids();
+  });
+
+  // no spread syntax: https://github.com/oven-sh/bun/issues/11734
+  preDiscovered.forEach((id) => ids.push(id));
+
+  console.log(
+    `Pre-discovered ${preDiscovered.length} ids (chunk ${chunkIndex})`,
+  );
+
+  chunkIndex++;
 }
 
-const ids: number[] = [...(preDiscoveredRaw ?? [])]; // TODO: re-validate these ids
 const regionalIdIndex: Record<Region, number> = {
   asia: ids.find((id) => idToRegion(id) === 'asia') ?? MIN_IDS.asia,
   com: ids.find((id) => idToRegion(id) === 'com') ?? MIN_IDS.com,
@@ -80,9 +98,7 @@ const interval = setInterval(async () => {
 
   if (Date.now() - lastProgressUpdate > PROGRESS_UPDATE_FREQUENCY) {
     lastProgressUpdate = Date.now();
-    console.log(
-      `Discovered ${verified.length} / ${idsToVerify.length} ids in ${Date.now() - startTime}ms`,
-    );
+    console.log(`Discovered ${ids.length} ids in ${Date.now() - startTime}ms`);
   }
 
   if (verified.length === 0) {
@@ -121,17 +137,31 @@ function post() {
   console.log(`Uploading ${ids.length} ids...`);
 
   const idsSorted = ids.sort((a, b) => a - b);
-  const didsWriteStream = new DidsWriteStream().dids(idsSorted);
-  const compressed = compress(didsWriteStream.uint8Array);
-  const content = Buffer.from(compressed).toString('base64');
+  const idsChunked = chunk(idsSorted, CHUNK_SIZE);
+  const files = idsChunked.map((ids, chunk) => {
+    const didsWriteStream = new DidsWriteStream().dids(idsSorted);
+    const compressed = compress(didsWriteStream.uint8Array);
+    const content = Buffer.from(compressed).toString('base64');
+
+    return {
+      content,
+      encoding: 'base64' as const,
+      path: `ids/${chunk}.dids.lz4`,
+    };
+  });
 
   commitAssets(
     'discovered ids',
     [
+      ...files,
       {
-        content,
-        encoding: 'base64',
-        path: 'definitions/ids.dids',
+        content: JSON.stringify({
+          chunks: idsChunked.length,
+          count: ids.length,
+          time: Date.now(),
+        } satisfies DiscoverIdsManifest),
+        encoding: 'utf-8',
+        path: 'ids/manifest.json',
       },
     ],
     production,
diff --git a/src/core/streams/rts.ts b/src/core/streams/rts.ts
index 2601c0bc9..8f6a54180 100644
--- a/src/core/streams/rts.ts
+++ b/src/core/streams/rts.ts
@@ -34,16 +34,16 @@ export class RtsmReadStream extends ReadStream {
   body() {
     return {
       time: this.uint32(),
-      base: this.uint8(),
+      bases: this.uint8(),
     };
   }
 }
 
 export class RtsmWriteStream extends WriteStream {
-  rtsm(base: number) {
+  rtsm(bases: number) {
     this.magic();
     this.header();
-    this.body(base);
+    this.body(bases);
 
     return this;
   }