Fixed some poor choices in jetstream consumption

.claude/settings.local.json

···
       "Read(//Users/scarndp/dev/skywatch/skywatch-tail/src/blobs/**)",
       "Read(//Users/scarndp/dev/skywatch/skywatch-tail/**)",
       "Bash(curl:*)",
-      "Bash(cat:*)"
+      "Bash(cat:*)",
+      "Bash(git push:*)",
+      "Bash(bun scripts/compute-phash.ts:*)",
+      "Bash(bun test:*)"
     ],
     "deny": [],
     "ask": []

src/config/index.ts

···
   jetstream: {
     url: string;
     wantedCollections: string[];
+    cursorUpdateInterval: number;
   };
   redis: {
     url: string;
···
   jetstream: {
     url: getEnv("JETSTREAM_URL", "wss://jetstream.atproto.tools/subscribe"),
     wantedCollections: ["app.bsky.feed.post"],
+    cursorUpdateInterval: getEnvNumber("CURSOR_UPDATE_INTERVAL", 10000),
   },
   redis: {
     url: getEnv("REDIS_URL", "redis://localhost:6379"),

src/main.ts

···
-import { readFile, writeFile } from "node:fs/promises";
-import { existsSync } from "node:fs";
+import fs from "node:fs";
 import type { CommitCreateEvent } from "@skyware/jetstream";
 import Redis from "ioredis";
 import { Jetstream } from "@skyware/jetstream";
···
 
 logger.info("Starting skywatch-phash service");
 
-// Cursor persistence
-const CURSOR_FILE = "/app/cursor.txt";
+let cursor = 0;
+let cursorUpdateInterval: NodeJS.Timeout;
 
-async function loadCursor(): Promise<number | undefined> {
-  try {
-    if (existsSync(CURSOR_FILE)) {
-      const content = await readFile(CURSOR_FILE, "utf-8");
-      const cursor = Number.parseInt(content.trim(), 10);
-      if (!Number.isNaN(cursor) && cursor > 0) {
-        logger.info({ cursor, cursorDate: new Date(cursor / 1000).toISOString() }, "Loaded cursor from file");
-        return cursor;
-      }
-    }
-  } catch (error) {
-    logger.warn({ error }, "Failed to load cursor from file");
-  }
-  return undefined;
+function epochUsToDateTime(cursor: number): string {
+  return new Date(cursor / 1000).toISOString();
 }
 
-async function saveCursor(cursor: number): Promise<void> {
-  try {
-    await writeFile(CURSOR_FILE, cursor.toString(), "utf-8");
-  } catch (error) {
-    logger.error({ error }, "Failed to save cursor to file");
+try {
+  logger.info({ process: "MAIN" }, "Trying to read cursor from cursor.txt");
+  cursor = Number(fs.readFileSync("cursor.txt", "utf8"));
+  logger.info(
+    { process: "MAIN", cursor, datetime: epochUsToDateTime(cursor) },
+    "Cursor found",
+  );
+} catch (error) {
+  if (error instanceof Error && "code" in error && error.code === "ENOENT") {
+    cursor = Math.floor(Date.now() * 1000);
+    logger.info(
+      { process: "MAIN", cursor, datetime: epochUsToDateTime(cursor) },
+      "Cursor not found in cursor.txt, setting cursor",
+    );
+    fs.writeFileSync("cursor.txt", cursor.toString(), "utf8");
+  } else {
+    logger.error({ process: "MAIN", error }, "Failed to read cursor");
+    process.exit(1);
   }
 }
 
-// Load cursor before creating Jetstream instance
-const savedCursor = await loadCursor();
-
 // Create Jetstream instance at module level
 const jetstream = new Jetstream({
   endpoint: config.jetstream.url,
   wantedCollections: ["app.bsky.feed.post"],
-  cursor: savedCursor,
+  cursor,
 });
 
 // Module-level variables for queue and worker
···
 
 // Register Jetstream event handlers at module level
 jetstream.on("open", () => {
-  logger.info(
-    {
-      url: config.jetstream.url,
-      cursor: jetstream.cursor,
-    },
-    "Connected to Jetstream"
-  );
+  if (jetstream.cursor) {
+    logger.info(
+      {
+        process: "MAIN",
+        url: config.jetstream.url,
+        cursor: jetstream.cursor,
+        datetime: epochUsToDateTime(jetstream.cursor),
+      },
+      "Connected to Jetstream with cursor",
+    );
+  } else {
+    logger.info(
+      { process: "MAIN", url: config.jetstream.url },
+      "Connected to Jetstream, waiting for cursor",
+    );
+  }
+  cursorUpdateInterval = setInterval(() => {
+    if (jetstream.cursor) {
+      logger.info(
+        {
+          process: "MAIN",
+          cursor: jetstream.cursor,
+          datetime: epochUsToDateTime(jetstream.cursor),
+        },
+        "Cursor updated",
+      );
+      fs.writeFile("cursor.txt", jetstream.cursor.toString(), (err) => {
+        if (err)
+          logger.error(
+            { process: "MAIN", error: err },
+            "Failed to write cursor",
+          );
+      });
+    }
+  }, config.jetstream.cursorUpdateInterval);
 });
 
 jetstream.on("close", () => {
-  logger.info("Jetstream connection closed");
+  clearInterval(cursorUpdateInterval);
+  logger.info({ process: "MAIN" }, "Jetstream connection closed");
 });
 
 jetstream.on("error", (error) => {
···
 });
 
 // Register onCreate handler for posts
-jetstream.onCreate("app.bsky.feed.post", async (event: CommitCreateEvent<"app.bsky.feed.post">) => {
-  try {
-    const record = event.commit.record as Record<string, unknown>;
-    const embed = record.embed;
+jetstream.onCreate(
+  "app.bsky.feed.post",
+  async (event: CommitCreateEvent<"app.bsky.feed.post">) => {
+    const atURI = `at://${event.did}/app.bsky.feed.post/${event.commit.rkey}`;
+    const hasEmbed = Object.prototype.hasOwnProperty.call(
+      event.commit.record,
+      "embed",
+    );
 
-    // Extract blob references
-    const blobs: Array<{ cid: string; mimeType?: string }> = [];
+    if (!hasEmbed) {
+      return;
+    }
 
-    if (embed && typeof embed === "object") {
-      const embedObj = embed as Record<string, unknown>;
+    try {
+      const record = event.commit.record as Record<string, unknown>;
+      const embed = record.embed;
 
-      if (Array.isArray(embedObj.images)) {
-        for (const img of embedObj.images) {
-          if (typeof img === "object" && img !== null) {
-            const image = img as Record<string, unknown>;
-            if (image.image && typeof image.image === "object" && image.image !== null) {
-              const imageObj = image.image as Record<string, unknown>;
-              const ref = imageObj.ref as Record<string, unknown> | undefined;
-              if (ref && typeof ref.$link === "string") {
-                blobs.push({
-                  cid: ref.$link,
-                  mimeType: typeof imageObj.mimeType === "string" ? imageObj.mimeType : undefined,
-                });
-              }
-            }
-          }
-        }
-      }
+      // Extract blob references
+      const blobs: Array<{ cid: string; mimeType?: string }> = [];
+
+      if (embed && typeof embed === "object") {
+        const embedObj = embed as Record<string, unknown>;
 
-      if (embedObj.media && typeof embedObj.media === "object" && embedObj.media !== null) {
-        const media = embedObj.media as Record<string, unknown>;
-        if (Array.isArray(media.images)) {
-          for (const img of media.images) {
+        if (Array.isArray(embedObj.images)) {
+          for (const img of embedObj.images) {
             if (typeof img === "object" && img !== null) {
               const image = img as Record<string, unknown>;
-              if (image.image && typeof image.image === "object" && image.image !== null) {
+              if (
+                image.image &&
+                typeof image.image === "object" &&
+                image.image !== null
+              ) {
                 const imageObj = image.image as Record<string, unknown>;
                 const ref = imageObj.ref as Record<string, unknown> | undefined;
                 if (ref && typeof ref.$link === "string") {
                   blobs.push({
                     cid: ref.$link,
-                    mimeType: typeof imageObj.mimeType === "string" ? imageObj.mimeType : undefined,
+                    mimeType:
+                      typeof imageObj.mimeType === "string"
+                        ? imageObj.mimeType
+                        : undefined,
                   });
                 }
               }
             }
           }
         }
+
+        if (
+          embedObj.media &&
+          typeof embedObj.media === "object" &&
+          embedObj.media !== null
+        ) {
+          const media = embedObj.media as Record<string, unknown>;
+          if (Array.isArray(media.images)) {
+            for (const img of media.images) {
+              if (typeof img === "object" && img !== null) {
+                const image = img as Record<string, unknown>;
+                if (
+                  image.image &&
+                  typeof image.image === "object" &&
+                  image.image !== null
+                ) {
+                  const imageObj = image.image as Record<string, unknown>;
+                  const ref = imageObj.ref as
+                    | Record<string, unknown>
+                    | undefined;
+                  if (ref && typeof ref.$link === "string") {
+                    blobs.push({
+                      cid: ref.$link,
+                      mimeType:
+                        typeof imageObj.mimeType === "string"
+                          ? imageObj.mimeType
+                          : undefined,
+                    });
+                  }
+                }
+              }
+            }
+          }
+        }
       }
-    }
 
-    if (blobs.length === 0) {
-      return;
-    }
+      if (blobs.length === 0) {
+        return;
+      }
 
-    const postUri = `at://${event.did}/${event.commit.collection}/${event.commit.rkey}`;
+      const postUri = `at://${event.did}/${event.commit.collection}/${event.commit.rkey}`;
 
-    logger.debug({ uri: postUri, blobCount: blobs.length }, "Post with blobs detected");
+      logger.debug(
+        { uri: postUri, blobCount: blobs.length },
+        "Post with blobs detected",
+      );
 
-    const job: ImageJob = {
-      postUri,
-      postCid: event.commit.cid,
-      postDid: event.did,
-      blobs,
-      timestamp: Date.now(),
-      attempts: 0,
-    };
+      const job: ImageJob = {
+        postUri,
+        postCid: event.commit.cid,
+        postDid: event.did,
+        blobs,
+        timestamp: Date.now(),
+        attempts: 0,
+      };
 
-    await queue.enqueue(job);
-  } catch (error) {
-    logger.error({ error, event }, "Error processing jetstream event");
-  }
-});
+      await queue.enqueue(job);
+    } catch (error) {
+      logger.error({ error, event }, "Error processing jetstream event");
+    }
+  },
+);
 
 // Async setup
 logger.info("Authenticating labeler");
···
   logger.error({ error }, "Redis error");
 });
 
-cache = config.cache.enabled ? new PhashCache(redis, config.cache.ttl) : undefined;
+cache = config.cache.enabled
+  ? new PhashCache(redis, config.cache.ttl)
+  : undefined;
 claims = new ModerationClaims(redis);
 metrics = new MetricsCollector();
 
···
   },
   agent,
   cache,
-  metrics
+  metrics,
 );
 
 worker.onMatchFound(async (postUri, postCid, postDid, match) => {
···
       matchedPhash: match.matchedPhash,
       hammingDistance: match.hammingDistance,
     },
-    "Match found - executing moderation actions"
+    "Match found - executing moderation actions",
   );
 
   try {
     if (check.toLabel) {
-      await createPostLabel(postUri, postCid, check.label, check.comment, match.phash, claims, metrics);
+      await createPostLabel(
+        postUri,
+        postCid,
+        check.label,
+        check.comment,
+        match.phash,
+        claims,
+        metrics,
+      );
     }
 
     if (check.reportPost) {
-      await createPostReport(postUri, postCid, check.comment, match.phash, metrics);
+      await createPostReport(
+        postUri,
+        postCid,
+        check.comment,
+        match.phash,
+        metrics,
+      );
     }
 
     if (check.labelAcct) {
-      await createAccountLabel(postDid, check.label, check.comment, postUri, match.phash, claims, metrics);
+      await createAccountLabel(
+        postDid,
+        check.label,
+        check.comment,
+        postUri,
+        match.phash,
+        claims,
+        metrics,
+      );
     }
 
     if (check.reportAcct) {
-      await createAccountReport(postDid, check.comment, postUri, match.phash, metrics);
+      await createAccountReport(
+        postDid,
+        check.comment,
+        postUri,
+        match.phash,
+        metrics,
+      );
     }
   } catch (error) {
-    logger.error({ error, postUri, postDid }, "Failed to execute moderation actions");
+    logger.error(
+      { error, postUri, postDid },
+      "Failed to execute moderation actions",
+    );
   }
 });
 
-const shutdown = async () => {
-  try {
-    logger.info("Shutting down gracefully");
-    jetstream.close();
-
-    // Save cursor one last time before shutdown
-    if (jetstream.cursor) {
-      await saveCursor(jetstream.cursor);
-      logger.info({ cursor: jetstream.cursor }, "Saved final cursor position");
-    }
-
-    await worker.stop();
-    await queue.disconnect();
-    await redis.quit();
-    clearInterval(statsInterval);
-    clearInterval(cursorInterval);
-    logger.info("Service stopped");
-    process.exit(0);
-  } catch (error) {
-    logger.error({ error }, "Error shutting down gracefully");
-    process.exit(1);
-  }
-};
-
-process.on("SIGTERM", () => void shutdown());
-process.on("SIGINT", () => void shutdown());
-
 logger.info("Starting queue worker");
 await worker.start();
 
···
 
 logger.info("Service started and ready");
 
-// Save cursor position every 10 seconds
-cursorInterval = setInterval(async () => {
-  if (jetstream.cursor) {
-    const cursorDate = new Date(jetstream.cursor / 1000).toISOString();
-    logger.info(
-      {
-        cursor: jetstream.cursor,
-        cursorDate,
-      },
-      "Jetstream cursor position"
-    );
-    await saveCursor(jetstream.cursor);
-  }
-}, 10_000);
-
 statsInterval = setInterval(async () => {
   const workerStats = await worker.getStats();
   const cacheStats = cache ? await cache.getStats() : null;
···
       cache: cacheStats,
       metrics: metricsData,
     },
-    "Service stats"
+    "Service stats",
   );
 }, 60000);
+
+async function shutdown() {
+  try {
+    logger.info({ process: "MAIN" }, "Shutting down gracefully");
+    if (jetstream.cursor !== undefined) {
+      fs.writeFileSync("cursor.txt", jetstream.cursor.toString(), "utf8");
+    }
+    jetstream.close();
+    await worker.stop();
+    await queue.disconnect();
+    await redis.quit();
+    clearInterval(statsInterval);
+    clearInterval(cursorUpdateInterval);
+    logger.info({ process: "MAIN" }, "Service stopped");
+    process.exit(0);
+  } catch (error) {
+    logger.error({ process: "MAIN", error }, "Error shutting down gracefully");
+    process.exit(1);
+  }
+}
+
+process.on("SIGINT", () => void shutdown());
+process.on("SIGTERM", () => void shutdown());

src/processor/image-processor.ts

···
         return null;
       }
 
-      const didDoc = await didDocResponse.json();
-      const pdsService = didDoc.service?.find((s: any) =>
+      const didDoc = await didDocResponse.json() as { service?: Array<{ id: string; type: string; serviceEndpoint?: string }> };
+      const pdsService = didDoc.service?.find((s) =>
         s.id === "#atproto_pds" && s.type === "AtprotoPersonalDataServer"
       );
 
···
       );
 
       if (!response.ok) {
-        const data = await response.json().catch(() => ({}));
+        const data = await response.json().catch(() => ({})) as { error?: string; message?: string };
         if (data.error === "RepoTakendown") {
           logger.info({ did, message: data.message }, "Repo has been taken down, skipping");
           this.repoTakendownCache.set(did, true);

tests/unit/image-processor.test.ts

···
   return ab;
 }
 
+// Mock PDS resolution and blob fetch
+function createMockFetch(blobResponse: Response | ((callCount: number) => Response)) {
+  let callCount = 0;
+  return mock((url: string) => {
+    // Mock PLC directory lookup
+    if (url.includes('plc.directory') || url.includes('plc.wtf')) {
+      return Promise.resolve({
+        ok: true,
+        json: async () => ({
+          service: [{
+            id: '#atproto_pds',
+            type: 'AtprotoPersonalDataServer',
+            serviceEndpoint: 'https://test.pds'
+          }]
+        })
+      } as Response);
+    }
+
+    // Mock describeRepo for takedown check
+    if (url.includes('describeRepo')) {
+      return Promise.resolve({
+        ok: true,
+        json: async () => ({})
+      } as Response);
+    }
+
+    // Mock blob fetch
+    callCount++;
+    if (typeof blobResponse === 'function') {
+      return Promise.resolve(blobResponse(callCount));
+    }
+    return Promise.resolve(blobResponse);
+  });
+}
+
 describe("ImageProcessor", () => {
   let processor: ImageProcessor;
   const testChecks: BlobCheck[] = [
···
     },
   ];
 
+  const mockAgent = {} as any; // Mock AtpAgent
+
   beforeEach(() => {
-    processor = new ImageProcessor(testChecks, "https://test.pds");
+    processor = new ImageProcessor(testChecks, mockAgent);
   });
 
   test("should skip non-image blobs", async () => {
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: () => Promise.resolve(new ArrayBuffer(0)),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: () => Promise.resolve(new ArrayBuffer(0)),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
   });
 
   test("should skip SVG images", async () => {
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: () => Promise.resolve(new ArrayBuffer(0)),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: () => Promise.resolve(new ArrayBuffer(0)),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
   });
 
   test("should return null when blob fetch fails", async () => {
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: false,
-        status: 404,
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: false,
+      status: 404,
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
       .png()
       .toBuffer();
 
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: async () => bufferToArrayBuffer(testImage),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: async () => bufferToArrayBuffer(testImage),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
       .png()
       .toBuffer();
 
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: async () => bufferToArrayBuffer(testImage),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: async () => bufferToArrayBuffer(testImage),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
       },
     ];
 
-    const matchingProcessor = new ImageProcessor(checksWithActualHash, "https://test.pds");
+    const matchingProcessor = new ImageProcessor(checksWithActualHash, mockAgent);
 
     const result = await matchingProcessor.processBlob("did:test:123", blob);
 
···
       .png()
       .toBuffer();
 
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: async () => bufferToArrayBuffer(testImage),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: async () => bufferToArrayBuffer(testImage),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
       .png()
       .toBuffer();
 
-    let callCount = 0;
-    const mockFetch = mock(() => {
-      callCount++;
+    const mockFetch = createMockFetch((callCount: number) => {
+      // First blob fetch fails (after PLC/describeRepo calls)
       if (callCount === 1) {
-        return Promise.resolve({
+        return {
           ok: false,
           status: 404,
-        } as Response);
+        } as Response;
       }
-      return Promise.resolve({
+      // Second blob fetch succeeds
+      return {
         ok: true,
         arrayBuffer: async () => bufferToArrayBuffer(testImage),
-      } as Response);
+      } as Response;
     });
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
···
       },
     ];
 
-    const processorWithIgnore = new ImageProcessor(checksWithIgnore, "https://test.pds");
+    const processorWithIgnore = new ImageProcessor(checksWithIgnore, mockAgent);
 
     const testImage = await sharp({
       create: {
···
       .png()
       .toBuffer();
 
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: async () => bufferToArrayBuffer(testImage),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: async () => bufferToArrayBuffer(testImage),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
       },
     ];
 
-    const processorWithIgnore = new ImageProcessor(checksWithIgnore, "https://test.pds");
+    const processorWithIgnore = new ImageProcessor(checksWithIgnore, mockAgent);
 
     const testImage = await sharp({
       create: {
···
       .png()
       .toBuffer();
 
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: async () => bufferToArrayBuffer(testImage),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: async () => bufferToArrayBuffer(testImage),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;
 
···
       },
     ];
 
-    const processorMixed = new ImageProcessor(mixedChecks, "https://test.pds");
+    const processorMixed = new ImageProcessor(mixedChecks, mockAgent);
 
     const testImage = await sharp({
       create: {
···
       .png()
       .toBuffer();
 
-    const mockFetch = mock(() =>
-      Promise.resolve({
-        ok: true,
-        arrayBuffer: async () => bufferToArrayBuffer(testImage),
-      } as Response)
-    );
+    const mockFetch = createMockFetch({
+      ok: true,
+      arrayBuffer: async () => bufferToArrayBuffer(testImage),
+    } as Response);
 
     globalThis.fetch = mockFetch as unknown as typeof fetch;