From c27db6a09e46448030a02bf6613625b622d40a29 Mon Sep 17 00:00:00 2001
From: zzstoatzz <thrast36@gmail.com>
Date: Mon, 2 Mar 2026 14:16:45 -0600
Subject: [PATCH] feat: multi-source backfill progress tracking

Change backfill_progress PK from (collection) to (collection, source)
so the same collection can be imported from multiple relays independently.
Enables backfilling from relay.waow.tech alongside bsky.network.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 src/backfill.zig  | 16 ++++++++--------
 src/event_log.zig | 20 ++++++++++++++++++--
 2 files changed, 26 insertions(+), 10 deletions(-)

diff --git a/src/backfill.zig b/src/backfill.zig
index 3c2bd7c..14b1091 100644
--- a/src/backfill.zig
+++ b/src/backfill.zig
@@ -81,7 +81,7 @@ pub const Backfiller = struct {
         // insert progress rows (skip existing)
         for (collections) |collection| {
             _ = self.db.exec(
-                "INSERT INTO backfill_progress (collection, source) VALUES ($1, $2) ON CONFLICT (collection) DO NOTHING",
+                "INSERT INTO backfill_progress (collection, source) VALUES ($1, $2) ON CONFLICT (collection, source) DO NOTHING",
                 .{ collection, self.source },
             ) catch |err| {
                 log.warn("failed to insert progress for {s}: {s}", .{ collection, @errorName(err) });
@@ -200,8 +200,8 @@ pub const Backfiller = struct {
         var imported: i64 = 0;
         {
             var row = (self.db.rowUnsafe(
-                "SELECT completed_at IS NOT NULL, cursor, imported_count FROM backfill_progress WHERE collection = $1",
-                .{collection},
+                "SELECT completed_at IS NOT NULL, cursor, imported_count FROM backfill_progress WHERE collection = $1 AND source = $2",
+                .{ collection, self.source },
             ) catch return error.DatabaseError) orelse return;
             defer row.deinit() catch {};
 
@@ -243,8 +243,8 @@ pub const Backfiller = struct {
             // update cursor in progress table
             const new_cursor = fetch_result.next_cursor orelse "";
             _ = self.db.exec(
-                "UPDATE backfill_progress SET cursor = $1, imported_count = $2 WHERE collection = $3",
-                .{ new_cursor, imported, collection },
+                "UPDATE backfill_progress SET cursor = $1, imported_count = $2 WHERE collection = $3 AND source = $4",
+                .{ new_cursor, imported, collection, self.source },
             ) catch {};
 
             if (fetch_result.next_cursor) |nc| {
@@ -257,8 +257,8 @@ pub const Backfiller = struct {
             } else {
                 // no more pages — mark complete
                 _ = self.db.exec(
-                    "UPDATE backfill_progress SET completed_at = now(), cursor = '', imported_count = $1 WHERE collection = $2",
-                    .{ imported, collection },
+                    "UPDATE backfill_progress SET completed_at = now(), cursor = '', imported_count = $1 WHERE collection = $2 AND source = $3",
+                    .{ imported, collection, self.source },
                 ) catch {};
                 log.info("{s}: complete ({d} DIDs, {d} pages)", .{ collection, imported, page_count });
                 break;
@@ -360,7 +360,7 @@ pub const Backfiller = struct {
 
         // per-collection detail
         var result = self.db.query(
-            "SELECT collection, source, cursor, imported_count, completed_at IS NOT NULL FROM backfill_progress ORDER BY collection",
+            "SELECT collection, source, cursor, imported_count, completed_at IS NOT NULL FROM backfill_progress ORDER BY collection, source",
             .{},
         ) catch return error.DatabaseError;
         defer result.deinit();
diff --git a/src/event_log.zig b/src/event_log.zig
index dc47bfd..f0e847c 100644
--- a/src/event_log.zig
+++ b/src/event_log.zig
@@ -177,15 +177,31 @@ pub const DiskPersist = struct {
 
         _ = try pool.exec(
             \\CREATE TABLE IF NOT EXISTS backfill_progress (
-            \\  collection TEXT PRIMARY KEY,
+            \\  collection TEXT NOT NULL,
             \\  source TEXT NOT NULL,
             \\  cursor TEXT NOT NULL DEFAULT '',
             \\  imported_count BIGINT NOT NULL DEFAULT 0,
             \\  completed_at TIMESTAMPTZ,
-            \\  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+            \\  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+            \\  PRIMARY KEY (collection, source)
             \\)
         , .{});
 
+        // migrate: old schema had collection as sole PK — add source to composite PK
+        _ = pool.exec(
+            \\DO $$ BEGIN
+            \\  IF EXISTS (
+            \\    SELECT 1 FROM pg_constraint
+            \\    WHERE conname = 'backfill_progress_pkey'
+            \\    AND conrelid = 'backfill_progress'::regclass
+            \\    AND array_length(conkey, 1) = 1
+            \\  ) THEN
+            \\    ALTER TABLE backfill_progress DROP CONSTRAINT backfill_progress_pkey;
+            \\    ALTER TABLE backfill_progress ADD PRIMARY KEY (collection, source);
+            \\  END IF;
+            \\END $$
+        , .{}) catch {};
+
         var self = DiskPersist{
             .allocator = allocator,
             .dir_path = try allocator.dupe(u8, dir_path),
-- 
2.51.2