From 14b33393f6b1429fa453f35ffcbae814a227a103 Mon Sep 17 00:00:00 2001
From: Jared Pereira <jared@awarm.space>
Date: Sat, 31 Jan 2026 23:50:37 -0500
Subject: [PATCH] add inngest function to write records

---
 app/api/inngest/client.ts                     | 10 +++
 .../inngest/functions/write_records_to_pds.ts | 70 +++++++++++++++++++
 app/api/inngest/route.tsx                     |  2 +
 3 files changed, 82 insertions(+)
 create mode 100644 app/api/inngest/functions/write_records_to_pds.ts

diff --git a/app/api/inngest/client.ts b/app/api/inngest/client.ts
index 2c635edb..4dceb80f 100644
--- a/app/api/inngest/client.ts
+++ b/app/api/inngest/client.ts
@@ -51,6 +51,16 @@ export type Events = {
       documentUris?: string[];
     };
   };
+  "user/write-records-to-pds": {
+    data: {
+      did: string;
+      records: Array<{
+        collection: string;
+        rkey: string;
+        record: unknown;
+      }>;
+    };
+  };
 };
 
 // Create a client to send and receive events
diff --git a/app/api/inngest/functions/write_records_to_pds.ts b/app/api/inngest/functions/write_records_to_pds.ts
new file mode 100644
index 00000000..c03b67f7
--- /dev/null
+++ b/app/api/inngest/functions/write_records_to_pds.ts
@@ -0,0 +1,70 @@
+import { inngest } from "../client";
+import { restoreOAuthSession } from "src/atproto-oauth";
+import { AtpBaseClient } from "lexicons/api";
+
+// Batch size to avoid Inngest payload limits and PDS rate limits
+const BATCH_SIZE = 50;
+
+// Helper to create authenticated agent - must be called fresh in each step
+// (OAuth sessions cannot be serialized across Inngest steps)
+async function createAuthenticatedAgent(did: string): Promise<AtpBaseClient> {
+  const result = await restoreOAuthSession(did);
+  if (!result.ok) {
+    throw new Error(`Failed to restore OAuth session: ${result.error.message}`);
+  }
+  return new AtpBaseClient(result.value.fetchHandler.bind(result.value));
+}
+
+export const write_records_to_pds = inngest.createFunction(
+  { id: "write-records-to-pds" },
+  { event: "user/write-records-to-pds" },
+  async ({ event, step }) => {
+    const { did, records } = event.data;
+
+    // Step 1: Verify OAuth session is valid before proceeding
+    await step.run("verify-oauth-session", async () => {
+      const result = await restoreOAuthSession(did);
+      if (!result.ok) {
+        throw new Error(`OAuth restore failed: ${result.error.message}`);
+      }
+      return { success: true };
+    });
+
+    // Step 2: Write records to PDS in batches
+    // Split records into batches to avoid payload limits and rate limiting
+    const batches: typeof records[] = [];
+    for (let i = 0; i < records.length; i += BATCH_SIZE) {
+      batches.push(records.slice(i, i + BATCH_SIZE));
+    }
+
+    let totalWritten = 0;
+    for (let batchIndex = 0; batchIndex < batches.length; batchIndex++) {
+      const batch = batches[batchIndex];
+      const batchWritten = await step.run(
+        `write-batch-${batchIndex}`,
+        async () => {
+          const agent = await createAuthenticatedAgent(did);
+          let written = 0;
+          for (const rec of batch) {
+            await agent.com.atproto.repo.putRecord({
+              repo: did,
+              collection: rec.collection,
+              rkey: rec.rkey,
+              record: rec.record as Record<string, unknown>,
+              validate: false,
+            });
+            written++;
+          }
+          return written;
+        },
+      );
+      totalWritten += batchWritten;
+    }
+
+    return {
+      success: true,
+      recordsWritten: totalWritten,
+      batchCount: batches.length,
+    };
+  },
+);
diff --git a/app/api/inngest/route.tsx b/app/api/inngest/route.tsx
index 7e32e2c1..ce86472d 100644
--- a/app/api/inngest/route.tsx
+++ b/app/api/inngest/route.tsx
@@ -12,6 +12,7 @@ import {
   cleanup_expired_oauth_sessions,
   check_oauth_session,
 } from "./functions/cleanup_expired_oauth_sessions";
+import { write_records_to_pds } from "./functions/write_records_to_pds";
 
 export const { GET, POST, PUT } = serve({
   client: inngest,
@@ -26,5 +27,6 @@ export const { GET, POST, PUT } = serve({
     fix_standard_document_postref,
     cleanup_expired_oauth_sessions,
     check_oauth_session,
+    write_records_to_pds,
   ],
 });
-- 
2.51.2