server: start backfilling new records

will need to refactor the backfill directory bc Ho Boy Thats Bad

server/deno.json

···
     "dk": "deno run -A --node-modules-dir npm:drizzle-kit"
   },
   "imports": {
+    "@atcute/cbor": "npm:@atcute/cbor@^2.2.6",
     "@atcute/identity-resolver": "npm:@atcute/identity-resolver@^1.1.3",
     "@atcute/atproto": "npm:@atcute/atproto@^3.1.4",
     "@atcute/client": "npm:@atcute/client@^4.0.3",

server/deno.lock

···
   "version": "5",
   "specifiers": {
     "npm:@atcute/atproto@^3.1.4": "3.1.4",
+    "npm:@atcute/cbor@^2.2.6": "2.2.6",
     "npm:@atcute/client@^4.0.3": "4.0.3",
     "npm:@atcute/identity-resolver@^1.1.3": "1.1.3_@atcute+identity@1.1.0",
     "npm:@atcute/lex-cli@^2.2.0": "2.2.0_@badrap+valita@0.4.6",
···
       "integrity": "sha512-v0/ue7mZYtjYw4vWbtda51bLwW88mqsUQB8F/UZNO18ANAQWmKq1HDceVqjvruaLe2QPqE43XM3WkEyZ2FhOrA==",
       "dependencies": [
         "@atcute/lexicons"
+      ]
+    },
+    "@atcute/cbor@2.2.6": {
+      "integrity": "sha512-pDfsn/vPTmgeXZiZdyc5vCGCPSxWlfTUIGFMCd5SroAgoLk1v9xxF7R/8+gt1lj1OKAwCwhS0doVmtLjqqzdbA==",
+      "dependencies": [
+        "@atcute/cid",
+        "@atcute/multibase",
+        "@atcute/uint8array"
+      ]
+    },
+    "@atcute/cid@2.2.4": {
+      "integrity": "sha512-6RUMyt7rp6KOSb4TWwifOZURnFrGgKqYyjVkYjiAcscZWgJpJxwoCUCdonxCfxhQtB0yJ+WlfqNXicGB+Pe94A==",
+      "dependencies": [
+        "@atcute/multibase",
+        "@atcute/uint8array"
       ]
     },
     "@atcute/client@4.0.3": {
···
       "dependencies": [
         "esm-env"
       ]
+    },
+    "@atcute/multibase@1.1.6": {
+      "integrity": "sha512-HBxuCgYLKPPxETV0Rot4VP9e24vKl8JdzGCZOVsDaOXJgbRZoRIF67Lp0H/OgnJeH/Xpva8Z5ReoTNJE5dn3kg==",
+      "dependencies": [
+        "@atcute/uint8array"
+      ]
+    },
+    "@atcute/uint8array@1.0.5": {
+      "integrity": "sha512-XLWWxoR2HNl2qU+FCr0rp1APwJXci7HnzbOQLxK55OaMNBXZ19+xNC5ii4QCsThsDxa4JS/JTzuiQLziITWf2Q=="
     },
     "@atcute/util-fetch@1.0.1": {
       "integrity": "sha512-Clc0E/5ufyGBVfYBUwWNlHONlZCoblSr4Ho50l1LhmRPGB1Wu/AQ9Sz+rsBg7fdaW/auve8ulmwhRhnX2cGRow==",
···
   "workspace": {
     "dependencies": [
       "npm:@atcute/atproto@^3.1.4",
+      "npm:@atcute/cbor@^2.2.6",
       "npm:@atcute/client@^4.0.3",
       "npm:@atcute/identity-resolver@^1.1.3",
       "npm:@atcute/lex-cli@^2.2.0",

server/src/backfill/index.ts

···
 import { db as db_type } from "../utils.ts";
 import { Client, simpleFetchHandler } from "@atcute/client";
 import oldRecords from "./old-records.ts";
+import newRecords from "./new-records.ts";
 
-const db: db_type = drizzle<typeof schema>(Deno.env.get("DB_FILE_NAME")!);
+const db: db_type = drizzle<typeof schema>(
+  Deno.env.get("DB_FILE_NAME")! ||
+    (() => {
+      throw "DB_FILE_NAME not set";
+    })(),
+);
 
 // we block access to the database till we've set up listeners etc
 // set an initial value for ts sake; this will set asap
···
 // so we should just nuke them
 await db.delete(routes);
 
-const backfillClient = new Client({
+const relay = new Client({
   handler: simpleFetchHandler({
-    service:
-      Deno.env.get("ATPROTO_RELAY") ||
+    service: Deno.env.get("ATPROTO_RELAY") ||
       (() => {
         throw "ATPROTO_RELAY not set";
       })(),
   }),
 });
 
-oldRecords(backfillClient, db);
+oldRecords(db, relay);
+newRecords(db);
 
 res(db);

server/src/backfill/new-records.ts

···
+import { db, getPds, isDid, rkeyToUrl } from "../utils.ts";
+import { decodeFirst } from "@atcute/cbor";
+import {
+  ComAtprotoSyncSubscribeRepos,
+  DevAtcitiesRoute,
+} from "../lexicons/index.ts";
+import { is } from "@atcute/lexicons";
+import { Client, simpleFetchHandler } from "@atcute/client";
+import { routes } from "../db/schema.ts";
+import { and, eq } from "drizzle-orm";
+
+const ErrorEvent = Symbol("error event");
+export default async function newRecords(db: db) {
+  // https://pdsls.dev/at://did:plc:6msi3pj7krzih5qxqtryxlzw/com.atproto.lexicon.schema/com.atproto.sync.subscribeRepos
+  const ws = new WebSocket(
+    `${
+      Deno.env.get("ATPROTO_RELAY") ||
+      (() => {
+        throw "ATPROTO_RELAY not set";
+      })()
+    }/xrpc/com.atproto.sync.subscribeRepos`
+  );
+
+  ws.addEventListener("close", (ev) => {
+    throw ev;
+  });
+  ws.addEventListener("error", (ev) => {
+    throw ev;
+  });
+
+  try {
+    ws.addEventListener("message", async (ev) => {
+      if (ev.data instanceof Blob) {
+        try {
+          const [header, remainder] = decodeFirst(await ev.data.bytes());
+          const [payload] = decodeFirst(remainder);
+          // https://atproto.com/specs/event-stream#framing
+          if (typeof header !== "object" || !header) return;
+          if (!("op" in header) || typeof header.op !== "number") return;
+          if (header.op === -1) {
+            console.error(header);
+            ws.close();
+            throw ErrorEvent;
+          }
+
+          if (!is(ComAtprotoSyncSubscribeRepos.commitSchema, payload)) {
+            return;
+          }
+          if (!isDid(payload.repo)) {
+            console.warn("Invalid did:", payload.repo);
+            return;
+          }
+
+          const pdsClient = (() => {
+            let client: Client | undefined;
+            return async () => {
+              if (client) {
+                return client;
+              }
+              if (!isDid(payload.repo)) {
+                console.warn("Invalid did:", payload.repo);
+                return;
+              }
+
+              const pds = await getPds(payload.repo);
+              if (!pds) {
+                return;
+              }
+              client = new Client({
+                handler: simpleFetchHandler({
+                  service: pds,
+                }),
+              });
+
+              return client;
+            };
+          })();
+
+          for (const op of payload.ops) {
+            const [collection, rkey] = op.path.split("/");
+            const path = rkeyToUrl(rkey);
+            if (!path) {
+              console.warn("rkey not valid!");
+              continue;
+            }
+            if (collection !== "dev.atcities.route") continue;
+            switch (op.action) {
+              case "create":
+              case "update": {
+                const client = await pdsClient();
+                if (!client) {
+                  console.warn("could not resolve pds for", payload.repo);
+                  return;
+                }
+                const { data, ok } = await client.get(
+                  "com.atproto.repo.getRecord",
+                  {
+                    params: {
+                      repo: payload.repo,
+                      collection: "dev.atcities.route",
+                      rkey,
+                    },
+                  }
+                );
+
+                if (!ok) {
+                  console.warn(data);
+                  continue;
+                }
+
+                if (!is(DevAtcitiesRoute.mainSchema, data.value)) {
+                  console.warn("Invalid record");
+                  continue;
+                }
+
+                if (data.value.page.$type !== "dev.atcities.route#blob") {
+                  console.warn("Unknown page type");
+                  continue;
+                }
+
+                const id = (
+                  await db
+                    .select({
+                      id: routes.id,
+                    })
+                    .from(routes)
+                    .where(
+                      and(
+                        eq(routes.did, payload.repo),
+                        eq(routes.url_route, path)
+                      )
+                    )
+                ).at(0);
+
+                await db.insert(routes).values({
+                  did: payload.repo,
+                  url_route: path,
+                  id: id ? id.id : undefined,
+                  blob_cid:
+                    "ref" in data.value.page.blob
+                      ? data.value.page.blob.ref.$link
+                      : data.value.page.blob.cid,
+                  mime: data.value.page.blob.mimeType,
+                });
+
+                console.log("wrote route");
+
+                continue;
+              }
+              case "delete": {
+                const id = (
+                  await db
+                    .select({
+                      id: routes.id,
+                    })
+                    .from(routes)
+                    .where(
+                      and(
+                        eq(routes.did, payload.repo),
+                        eq(routes.url_route, path)
+                      )
+                    )
+                ).at(0);
+                if (!id) continue;
+
+                await db.delete(routes).where(eq(routes.id, id.id));
+                console.log("deleted route");
+
+                continue;
+              }
+            }
+          }
+        } catch (e) {
+          if (Error.isError(e)) {
+            console.warn(
+              "Invalid DAG-CBOR data:",
+              e.name,
+              e.message,
+              e.cause,
+              e.stack
+            );
+          } else throw e;
+        }
+      }
+    });
+  } catch (e) {
+    // if and only if an error message was recived, reopen the connection
+    if (e === ErrorEvent) newRecords(db);
+    else throw e;
+  }
+  ws.addEventListener("open", () => console.log("subscribed!!"));
+}

server/src/backfill/old-records.ts

···
       break;
     }
 
-    for (const record of data.records)
+    for (const record of data.records) {
       if (is(DevAtcitiesRoute.mainSchema, record.value)) {
         if (record.value.page.$type === "dev.atcities.route#blob") {
           const url = rkeyToUrl(record.uri.split("/")[4]);
···
                 : record.value.page.blob.cid,
             mime: record.value.page.blob.mimeType,
           };
-          db.insert(routes).values(data);
+          await db.insert(routes).values(data);
         }
       } else console.warn("Invalid record:", record.uri);
+    }
 
     cursor = data.cursor;
     if (!cursor) break;
   }
 }
 
-export default async function (backfillClient: Client, db: db) {
+export default async function (db: db, relay: Client) {
   let repos: `did:${string}:${string}`[] = [];
   let cursor: string | undefined;
   while (true) {
-    const { data, ok } = await backfillClient.get(
+    const { data, ok } = await relay.get(
       "com.atproto.sync.listReposByCollection",
       {
         params: {