remanso-jetstream/jetstream.ts

import { Jetstream } from "@skyware/jetstream";
import {
  deleteNote,
  getCursor,
  getWebhooksByDidAndVerb,
  saveCursor,
  upsertNote,
} from "./src/data/db.ts";
import { Note } from "./src/data/note.ts";
import { log } from "./src/log.ts";
import { ensureIndex, indexNote, removeNote } from "./src/search/opensearch.ts";
import { dispatchAll, fireWebhooks } from "./src/jetstream/webhooks.ts";
import { type BulkDeps, queueBulkCreate } from "./src/jetstream/bulk.ts";

const safeIndex = async (note: Note): Promise<void> => {
  try {
    await indexNote(note);
  } catch (error) {
    log(`[opensearch] index ${note.did}/${note.rkey} failed:`, error);
  }
};

const safeRemove = async (did: string, rkey: string): Promise<void> => {
  try {
    await removeNote(did, rkey);
  } catch (error) {
    log(`[opensearch] delete ${did}/${rkey} failed:`, error);
  }
};

const bulkDeps: BulkDeps = {
  getSubs: getWebhooksByDidAndVerb,
  dispatch: dispatchAll,
};

if (import.meta.main) {
  globalThis.addEventListener("unhandledrejection", (e) => {
    log("[jetstream] unhandled rejection:", e.reason);
  });

  globalThis.addEventListener("error", (e) => {
    log("[jetstream] uncaught error:", e.error);
  });

  const cursor = getCursor();
  log(`[jetstream] starting with cursor: ${cursor ?? "none"}`);

  const jetstream = new Jetstream({
    wantedCollections: ["space.remanso.note"],
    cursor: cursor ? Number(cursor) : undefined,
    endpoint: "https://jetstream2.fr.hose.cam/subscribe",
  });

  jetstream.onCreate("space.remanso.note", async (event) => {
    try {
      const {
        did,
        commit: { rkey, record },
      } = event;
      log(`[jetstream] creating ${did}/${rkey}...`);
      const note = record as unknown as Omit<Note, "did" | "rkey">;
      upsertNote({ did, rkey, ...note });
      log(`[jetstream] create ${did}/${rkey}: ${note.title}`);
      await Promise.allSettled([
        fireWebhooks(
          did,
          "create",
          { event: "create", did, rkey, ...note },
          getWebhooksByDidAndVerb,
        ),
        safeIndex({ did, rkey, ...note }),
      ]);
      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
    } catch (error) {
      log(`[jetstream] error on create:`, error);
    }
  });

  jetstream.onUpdate("space.remanso.note", async (event) => {
    try {
      const {
        did,
        commit: { rkey, record },
      } = event;
      log(`[jetstream] updating ${did}/${rkey}...`);
      const note = record as unknown as Omit<Note, "did" | "rkey">;
      upsertNote({ did, rkey, ...note });
      log(`[jetstream] update ${did}/${rkey}: ${note.title}`);
      // Updates fold into the `create` verb — subscribers reconcile by (did, rkey).
      await Promise.allSettled([
        fireWebhooks(
          did,
          "create",
          { event: "create", did, rkey, ...note },
          getWebhooksByDidAndVerb,
        ),
        safeIndex({ did, rkey, ...note }),
      ]);
      queueBulkCreate(did, { rkey, ...note }, bulkDeps);
    } catch (error) {
      log(`[jetstream] error on update:`, error);
    }
  });

  jetstream.onDelete("space.remanso.note", async (event) => {
    try {
      const {
        did,
        commit: { rkey },
      } = event;
      log(`[jetstream] deleting ${did}/${rkey}...`);
      deleteNote({ did, rkey });
      log(`[jetstream] delete ${did}/${rkey}`);
      await Promise.allSettled([
        fireWebhooks(
          did,
          "delete",
          { event: "delete", did, rkey },
          getWebhooksByDidAndVerb,
        ),
        safeRemove(did, rkey),
      ]);
    } catch (error) {
      log(`[jetstream] error on delete:`, error);
    }
  });

  jetstream.on("open", () => {
    log("[jetstream] connected");
  });

  jetstream.on("close", () => {
    log("[jetstream] connection closed");
  });

  jetstream.on("error", (error) => {
    log("[jetstream] connection closed with error", error);
  });

  try {
    await ensureIndex();
  } catch (error) {
    log("[opensearch] ensureIndex failed at startup:", error);
  }

  log("[jetstream] launching");
  jetstream.start();

  setInterval(() => {
    if (jetstream.cursor) {
      saveCursor(jetstream.cursor);
    }
  }, 5000);
}