diff --git a/Dockerfile.runtime b/Dockerfile.runtime
index 61a21be..1c4865b 100644
--- a/Dockerfile.runtime
+++ b/Dockerfile.runtime
@@ -1,7 +1,6 @@
 FROM debian:bookworm-slim
 RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates && rm -rf /var/lib/apt/lists/*
 COPY zig-out/bin/zlay /usr/local/bin/zlay
-COPY zig-out/bin/collectiondir /usr/local/bin/collectiondir
 RUN mkdir -p /data/events /data/collection-index
 ENV RELAY_DATA_DIR=/data/events
 EXPOSE 3000 3001
diff --git a/build.zig b/build.zig
index ce7a4e6..44c6f03 100644
--- a/build.zig
+++ b/build.zig
@@ -64,27 +64,6 @@ pub fn build(b: *std.Build) void {
     const run_step = b.step("run", "run the relay");
     run_step.dependOn(&run_relay.step);
 
-    // collectiondir executable (shadow collection directory service)
-    const collectiondir_mod = b.createModule(.{
-        .root_source_file = b.path("src/collectiondir.zig"),
-        .target = target,
-        .optimize = optimize,
-        .imports = imports,
-    });
-    collectiondir_mod.addImport("build_options", build_options.createModule());
-    const collectiondir = b.addExecutable(.{
-        .name = "collectiondir",
-        .root_module = collectiondir_mod,
-    });
-    collectiondir.linkLibC();
-    collectiondir.linkLibCpp();
-    b.installArtifact(collectiondir);
-
-    const run_collectiondir = b.addRunArtifact(collectiondir);
-    if (b.args) |args| run_collectiondir.addArgs(args);
-    const run_collectiondir_step = b.step("run-collectiondir", "run the collection directory service");
-    run_collectiondir_step.dependOn(&run_collectiondir.step);
-
     // tests
     const test_step = b.step("test", "run unit tests");
     const test_files = .{
diff --git a/src/collection_index.zig b/src/collection_index.zig
index 86aa98d..5c69be7 100644
--- a/src/collection_index.zig
+++ b/src/collection_index.zig
@@ -19,7 +19,6 @@ const separator = '\x00';
 
 pub const CollectionIndex = struct {
     db: rocksdb.DB,
-    default: rocksdb.ColumnFamilyHandle,
     rbc: rocksdb.ColumnFamilyHandle,
     cbr: rocksdb.ColumnFamilyHandle,
     allocator: Allocator,
@@ -51,11 +50,9 @@ pub const CollectionIndex = struct {
         defer allocator.free(families);
 
         // find column family handles by name
-        var default: ?rocksdb.ColumnFamilyHandle = null;
         var rbc: ?rocksdb.ColumnFamilyHandle = null;
         var cbr: ?rocksdb.ColumnFamilyHandle = null;
         for (families) |cf| {
-            if (std.mem.eql(u8, cf.name, "default")) default = cf.handle;
             if (std.mem.eql(u8, cf.name, "rbc")) rbc = cf.handle;
             if (std.mem.eql(u8, cf.name, "cbr")) cbr = cf.handle;
         }
@@ -64,57 +61,12 @@ pub const CollectionIndex = struct {
 
         return .{
             .db = db,
-            .default = default orelse return error.MissingColumnFamily,
             .rbc = rbc orelse return error.MissingColumnFamily,
             .cbr = cbr orelse return error.MissingColumnFamily,
             .allocator = allocator,
         };
     }
 
-    // --- metadata helpers (default CF) ---
-
-    /// read a metadata value from the default column family.
-    /// caller must call .deinit() on the returned Data to free rocksdb memory.
-    pub fn getMeta(self: *CollectionIndex, key: []const u8) ?rocksdb.Data {
-        var err_str: ?rocksdb.Data = null;
-        const val = self.db.get(self.default, key, &err_str) catch {
-            if (err_str) |e| {
-                log.debug("getMeta({s}) failed: {s}", .{ key, e.data });
-                e.deinit();
-            }
-            return null;
-        };
-        return val;
-    }
-
-    /// write a metadata value to the default column family
-    pub fn putMeta(self: *CollectionIndex, key: []const u8, value: []const u8) void {
-        var err_str: ?rocksdb.Data = null;
-        var batch = rocksdb.WriteBatch.init();
-        defer batch.deinit();
-        batch.put(self.default, key, value);
-        self.db.write(batch, &err_str) catch {
-            if (err_str) |e| {
-                log.warn("putMeta({s}) failed: {s}", .{ key, e.data });
-                e.deinit();
-            }
-        };
-    }
-
-    /// delete a metadata key from the default column family
-    pub fn deleteMeta(self: *CollectionIndex, key: []const u8) void {
-        var err_str: ?rocksdb.Data = null;
-        var batch = rocksdb.WriteBatch.init();
-        defer batch.deinit();
-        batch.delete(self.default, key);
-        self.db.write(batch, &err_str) catch {
-            if (err_str) |e| {
-                log.warn("deleteMeta({s}) failed: {s}", .{ key, e.data });
-                e.deinit();
-            }
-        };
-    }
-
     pub fn deinit(self: *CollectionIndex) void {
         self.db.deinit();
     }
diff --git a/src/collectiondir.zig b/src/collectiondir.zig
deleted file mode 100644
index cfe6750..0000000
--- a/src/collectiondir.zig
+++ /dev/null
@@ -1,407 +0,0 @@
-//! collectiondir — standalone collection directory service
-//!
-//! subscribes to the relay's firehose and maintains an independent RocksDB
-//! collection index. no postgres — all state (cursor, backfill progress,
-//! collection index) lives in RocksDB.
-//!
-//! reuses collection_index.zig (unchanged) and resync.zig (unchanged, stage 2).
-//! backfill uses collectiondir_backfill.zig (RocksDB-backed progress).
-//!
-//! LISTEN_PORT (default 2510): HTTP API
-//!   GET /xrpc/com.atproto.sync.listReposByCollection
-//!   GET /_health    — readiness (subscriber connected + cursor fresh)
-//!   GET /_healthz   — liveness
-//!   GET /metrics
-//!   GET /admin/backfill-collections  — backfill status
-//!   POST /admin/backfill-collections — trigger backfill
-//!
-//! env:
-//!   RELAY_URL — relay firehose host (e.g. "zlay" for in-cluster)
-//!   RELAY_PORT — relay firehose port (default 3000)
-//!   RELAY_TLS — "true" for wss:// (default "false")
-//!   COLLECTION_INDEX_DIR — RocksDB data directory
-//!   LISTEN_PORT — HTTP listen port (default 2510)
-//!   RELAY_ADMIN_PASSWORD — admin auth token
-
-const std = @import("std");
-const http = std.http;
-const collection_index_mod = @import("collection_index.zig");
-const backfill_mod = @import("collectiondir_backfill.zig");
-const collectiondir_sub = @import("collectiondir_subscriber.zig");
-
-const log = std.log.scoped(.collectiondir);
-
-/// 8 MiB stacks — matches relay's default_stack_size for ReleaseSafe
-pub const default_stack_size = 8 * 1024 * 1024;
-
-var shutdown_flag: std.atomic.Value(bool) = .{ .raw = false };
-
-pub fn main() !void {
-    const allocator = std.heap.c_allocator;
-
-    // parse config from env
-    const listen_port = parseEnvInt(u16, "LISTEN_PORT", 2510);
-    const relay_url = std.posix.getenv("RELAY_URL") orelse "zlay";
-    const relay_port = parseEnvInt(u16, "RELAY_PORT", 3000);
-    const use_tls = if (std.posix.getenv("RELAY_TLS")) |v| std.mem.eql(u8, v, "true") else false;
-    const ci_dir = std.posix.getenv("COLLECTION_INDEX_DIR") orelse "data/collection-index";
-
-    // install signal handlers
-    installSignalHandlers();
-
-    // init RocksDB collection index (all state lives here)
-    var ci = collection_index_mod.CollectionIndex.open(allocator, ci_dir) catch |err| {
-        log.err("failed to open collection index at {s}: {s}", .{ ci_dir, @errorName(err) });
-        return err;
-    };
-    defer ci.deinit();
-
-    // init backfiller (RocksDB-backed progress)
-    var backfiller = backfill_mod.Backfiller.init(allocator, &ci);
-
-    // init firehose subscriber (cursor in RocksDB)
-    var subscriber = collectiondir_sub.Subscriber.init(
-        allocator,
-        relay_url,
-        relay_port,
-        use_tls,
-        &ci,
-        &shutdown_flag,
-    );
-
-    // start subscriber thread
-    const sub_thread = try std.Thread.spawn(.{ .stack_size = default_stack_size }, collectiondir_sub.Subscriber.run, .{&subscriber});
-
-    // start HTTP server
-    log.info("collectiondir listening on :{d}", .{listen_port});
-    log.info("relay upstream: {s}:{d} (tls={s})", .{ relay_url, relay_port, if (use_tls) "true" else "false" });
-    log.info("collection index: {s}", .{ci_dir});
-
-    var http_ctx = HttpContext{
-        .collection_index = &ci,
-        .backfiller = &backfiller,
-        .subscriber = &subscriber,
-    };
-
-    const address = std.net.Address.initIp4(.{ 0, 0, 0, 0 }, listen_port);
-    var server = address.listen(.{ .reuse_address = true }) catch |err| {
-        log.err("failed to listen on :{d}: {s}", .{ listen_port, @errorName(err) });
-        return err;
-    };
-
-    while (!shutdown_flag.load(.acquire)) {
-        const conn = server.accept() catch |err| {
-            if (shutdown_flag.load(.acquire)) break;
-            log.debug("accept error: {s}", .{@errorName(err)});
-            continue;
-        };
-        const timeout = std.posix.timeval{ .sec = 5, .usec = 0 };
-        std.posix.setsockopt(conn.stream.handle, std.posix.SOL.SOCKET, std.posix.SO.RCVTIMEO, std.mem.asBytes(&timeout)) catch {};
-        handleConnection(conn.stream, &http_ctx);
-    }
-
-    log.info("shutdown signal received, stopping...", .{});
-    server.stream.close();
-    sub_thread.join();
-    log.info("collectiondir stopped cleanly", .{});
-}
-
-const HttpContext = struct {
-    collection_index: *collection_index_mod.CollectionIndex,
-    backfiller: *backfill_mod.Backfiller,
-    subscriber: *collectiondir_sub.Subscriber,
-};
-
-fn handleConnection(stream: std.net.Stream, ctx: *HttpContext) void {
-    defer stream.close();
-
-    var recv_buf: [4096]u8 = undefined;
-    var send_buf: [4096]u8 = undefined;
-    var connection_reader = stream.reader(&recv_buf);
-    var connection_writer = stream.writer(&send_buf);
-    var server = http.Server.init(connection_reader.interface(), &connection_writer.interface);
-
-    var request = server.receiveHead() catch return;
-    const target = request.head.target;
-    const method = request.head.method;
-
-    const qmark = std.mem.indexOfScalar(u8, target, '?');
-    const path = target[0..(qmark orelse target.len)];
-    const query = if (qmark) |q| target[q + 1 ..] else "";
-
-    if (method == .GET) {
-        if (std.mem.eql(u8, path, "/_healthz")) {
-            respond(&request, .ok, "application/json", "{\"status\":\"ok\"}");
-        } else if (std.mem.eql(u8, path, "/_health")) {
-            handleReadiness(&request, ctx);
-        } else if (std.mem.eql(u8, path, "/metrics")) {
-            var metrics_buf: [8192]u8 = undefined;
-            const body = ctx.subscriber.formatMetrics(&metrics_buf);
-            respond(&request, .ok, "text/plain; version=0.0.4; charset=utf-8", body);
-        } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listReposByCollection")) {
-            handleListReposByCollection(&request, query, ctx.collection_index);
-        } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) {
-            if (!checkAdmin(&request)) return;
-            handleBackfillStatus(&request, ctx.backfiller);
-        } else {
-            respond(&request, .not_found, "text/plain", "not found");
-        }
-    } else if (method == .POST) {
-        if (std.mem.eql(u8, path, "/admin/backfill-collections")) {
-            if (!checkAdmin(&request)) return;
-            handleBackfillTrigger(&request, query, ctx.backfiller);
-        } else {
-            respond(&request, .not_found, "text/plain", "not found");
-        }
-    } else {
-        respond(&request, .method_not_allowed, "text/plain", "method not allowed");
-    }
-}
-
-/// readiness: subscriber connected + cursor advancing
-fn handleReadiness(request: *http.Server.Request, ctx: *HttpContext) void {
-    const stats = &ctx.subscriber.stats;
-    const connected = stats.connected.load(.monotonic);
-    const last_event_time = stats.last_event_time.load(.monotonic);
-    const now = std.time.timestamp();
-
-    // subscriber must be connected and have received an event within the last 60s
-    // (allows for startup grace — last_event_time == 0 means never received)
-    const stale_threshold = 60;
-    const cursor_fresh = connected and (last_event_time == 0 or (now - last_event_time) < stale_threshold);
-
-    if (cursor_fresh) {
-        var buf: [256]u8 = undefined;
-        const body = std.fmt.bufPrint(&buf, "{{\"status\":\"ok\",\"connected\":{},\"last_seq\":{d}}}", .{
-            connected, stats.last_seq.load(.monotonic),
-        }) catch "{\"status\":\"ok\"}";
-        respond(request, .ok, "application/json", body);
-    } else {
-        var buf: [256]u8 = undefined;
-        const body = std.fmt.bufPrint(&buf, "{{\"status\":\"unhealthy\",\"connected\":{},\"cursor_fresh\":{}}}", .{
-            connected, cursor_fresh,
-        }) catch "{\"status\":\"unhealthy\"}";
-        respond(request, .service_unavailable, "application/json", body);
-    }
-}
-
-fn handleListReposByCollection(
-    request: *http.Server.Request,
-    query: []const u8,
-    ci: *collection_index_mod.CollectionIndex,
-) void {
-    const collection = queryParam(query, "collection") orelse {
-        respond(request, .bad_request, "application/json", "{\"error\":\"BadRequest\",\"message\":\"collection parameter required\"}");
-        return;
-    };
-
-    if (collection.len == 0 or !std.mem.containsAtLeast(u8, collection, 1, ".")) {
-        respond(request, .bad_request, "application/json", "{\"error\":\"BadRequest\",\"message\":\"invalid collection NSID\"}");
-        return;
-    }
-
-    const limit_str = queryParam(query, "limit") orelse "500";
-    const limit = std.fmt.parseInt(usize, limit_str, 10) catch {
-        respond(request, .bad_request, "application/json", "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}");
-        return;
-    };
-    if (limit < 1 or limit > 2000) {
-        respond(request, .bad_request, "application/json", "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..2000\"}");
-        return;
-    }
-
-    var cursor_buf: [256]u8 = undefined;
-    const cursor_did = queryParamDecoded(query, "cursor", &cursor_buf);
-
-    var did_buf: [65536]u8 = undefined;
-    const ci_result = ci.listReposByCollection(collection, limit, cursor_did, &did_buf) catch {
-        respond(request, .internal_server_error, "application/json", "{\"error\":\"InternalError\",\"message\":\"index scan failed\"}");
-        return;
-    };
-
-    var buf: [65536]u8 = undefined;
-    var fbs = std.io.fixedBufferStream(&buf);
-    const w = fbs.writer();
-
-    w.writeAll("{\"repos\":[") catch return;
-    for (0..ci_result.count) |i| {
-        if (i > 0) w.writeByte(',') catch return;
-        w.writeAll("{\"did\":\"") catch return;
-        w.writeAll(ci_result.getDid(i)) catch return;
-        w.writeAll("\"}") catch return;
-    }
-    w.writeByte(']') catch return;
-
-    if (ci_result.last_did) |last| {
-        if (ci_result.count >= limit) {
-            w.writeAll(",\"cursor\":\"") catch return;
-            w.writeAll(last) catch return;
-            w.writeAll("\"") catch return;
-        }
-    }
-
-    w.writeByte('}') catch return;
-    respond(request, .ok, "application/json", fbs.getWritten());
-}
-
-fn handleBackfillStatus(request: *http.Server.Request, backfiller: *backfill_mod.Backfiller) void {
-    const body = backfiller.getStatus(backfiller.allocator) catch {
-        respond(request, .internal_server_error, "application/json", "{\"error\":\"failed to query backfill status\"}");
-        return;
-    };
-    defer backfiller.allocator.free(body);
-    respond(request, .ok, "application/json", body);
-}
-
-fn handleBackfillTrigger(request: *http.Server.Request, query: []const u8, backfiller: *backfill_mod.Backfiller) void {
-    const source = queryParam(query, "source") orelse "bsky.network";
-    backfiller.start(source) catch |err| {
-        switch (err) {
-            error.AlreadyRunning => {
-                respond(request, .conflict, "application/json", "{\"error\":\"backfill already in progress\"}");
-            },
-            else => {
-                respond(request, .internal_server_error, "application/json", "{\"error\":\"failed to start backfill\"}");
-            },
-        }
-        return;
-    };
-    var buf: [256]u8 = undefined;
-    const resp_body = std.fmt.bufPrint(&buf, "{{\"status\":\"started\",\"source\":\"{s}\"}}", .{source}) catch {
-        respond(request, .ok, "application/json", "{\"status\":\"started\"}");
-        return;
-    };
-    respond(request, .ok, "application/json", resp_body);
-}
-
-// --- auth ---
-
-fn checkAdmin(request: *http.Server.Request) bool {
-    const admin_pw = std.posix.getenv("RELAY_ADMIN_PASSWORD") orelse {
-        respond(request, .forbidden, "application/json", "{\"error\":\"admin endpoint not configured\"}");
-        return false;
-    };
-
-    var it = request.iterateHeaders();
-    while (it.next()) |header| {
-        if (std.ascii.eqlIgnoreCase(header.name, "authorization")) {
-            const bearer_prefix = "Bearer ";
-            if (!std.mem.startsWith(u8, header.value, bearer_prefix)) {
-                respond(request, .unauthorized, "application/json", "{\"error\":\"invalid authorization scheme\"}");
-                return false;
-            }
-            const token = header.value[bearer_prefix.len..];
-            if (!std.mem.eql(u8, token, admin_pw)) {
-                respond(request, .unauthorized, "application/json", "{\"error\":\"invalid token\"}");
-                return false;
-            }
-            return true;
-        }
-    }
-
-    respond(request, .unauthorized, "application/json", "{\"error\":\"missing authorization header\"}");
-    return false;
-}
-
-// --- response helpers ---
-
-fn respond(request: *http.Server.Request, status: http.Status, content_type: []const u8, body: []const u8) void {
-    request.respond(body, .{
-        .status = status,
-        .keep_alive = false,
-        .extra_headers = &.{
-            .{ .name = "content-type", .value = content_type },
-            .{ .name = "server", .value = "collectiondir (zlay)" },
-        },
-    }) catch {};
-}
-
-// --- query string helpers ---
-
-fn queryParam(query: []const u8, name: []const u8) ?[]const u8 {
-    if (query.len == 0) return null;
-    var iter = std.mem.splitScalar(u8, query, '&');
-    while (iter.next()) |pair| {
-        const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue;
-        if (std.mem.eql(u8, pair[0..eq], name)) {
-            return pair[eq + 1 ..];
-        }
-    }
-    return null;
-}
-
-fn queryParamDecoded(query: []const u8, name: []const u8, buf: []u8) ?[]const u8 {
-    const raw = queryParam(query, name) orelse return null;
-    var i: usize = 0;
-    var out: usize = 0;
-    while (i < raw.len) {
-        if (raw[i] == '%' and i + 2 < raw.len) {
-            const hi = hexVal(raw[i + 1]) orelse {
-                if (out >= buf.len) return null;
-                buf[out] = raw[i];
-                out += 1;
-                i += 1;
-                continue;
-            };
-            const lo = hexVal(raw[i + 2]) orelse {
-                if (out >= buf.len) return null;
-                buf[out] = raw[i];
-                out += 1;
-                i += 1;
-                continue;
-            };
-            if (out >= buf.len) return null;
-            buf[out] = (@as(u8, hi) << 4) | @as(u8, lo);
-            out += 1;
-            i += 3;
-        } else if (raw[i] == '+') {
-            if (out >= buf.len) return null;
-            buf[out] = ' ';
-            out += 1;
-            i += 1;
-        } else {
-            if (out >= buf.len) return null;
-            buf[out] = raw[i];
-            out += 1;
-            i += 1;
-        }
-    }
-    return buf[0..out];
-}
-
-fn hexVal(c: u8) ?u4 {
-    return switch (c) {
-        '0'...'9' => @intCast(c - '0'),
-        'a'...'f' => @intCast(c - 'a' + 10),
-        'A'...'F' => @intCast(c - 'A' + 10),
-        else => null,
-    };
-}
-
-// --- helpers ---
-
-fn signalHandler(_: c_int) callconv(.c) void {
-    shutdown_flag.store(true, .release);
-}
-
-fn installSignalHandlers() void {
-    const act: std.posix.Sigaction = .{
-        .handler = .{ .handler = signalHandler },
-        .mask = std.posix.sigemptyset(),
-        .flags = 0,
-    };
-    std.posix.sigaction(std.posix.SIG.INT, &act, null);
-    std.posix.sigaction(std.posix.SIG.TERM, &act, null);
-
-    const ignore_act: std.posix.Sigaction = .{
-        .handler = .{ .handler = std.posix.SIG.IGN },
-        .mask = std.posix.sigemptyset(),
-        .flags = 0,
-    };
-    std.posix.sigaction(std.posix.SIG.PIPE, &ignore_act, null);
-}
-
-fn parseEnvInt(comptime T: type, key: []const u8, default: T) T {
-    const val = std.posix.getenv(key) orelse return default;
-    return std.fmt.parseInt(T, val, 10) catch default;
-}
diff --git a/src/collectiondir_backfill.zig b/src/collectiondir_backfill.zig
deleted file mode 100644
index 7f6cde4..0000000
--- a/src/collectiondir_backfill.zig
+++ /dev/null
@@ -1,423 +0,0 @@
-//! collectiondir backfill — discovers collections and imports DIDs from a source relay
-//!
-//! same logic as backfill.zig but stores progress in RocksDB default CF instead
-//! of postgres. no external database dependency.
-//!
-//! progress keys: "bf:{collection}\0{source}" → "{cursor}\0{count}\0{0|1}"
-//! (cursor string, imported count, completed flag)
-
-const std = @import("std");
-const http = std.http;
-const collection_index_mod = @import("collection_index.zig");
-
-const Allocator = std.mem.Allocator;
-const log = std.log.scoped(.backfill);
-
-pub const Backfiller = struct {
-    allocator: Allocator,
-    collection_index: *collection_index_mod.CollectionIndex,
-    running: std.atomic.Value(bool),
-    thread: ?std.Thread,
-    source: []const u8,
-
-    pub fn init(
-        allocator: Allocator,
-        collection_index: *collection_index_mod.CollectionIndex,
-    ) Backfiller {
-        return .{
-            .allocator = allocator,
-            .collection_index = collection_index,
-            .running = .{ .raw = false },
-            .thread = null,
-            .source = "",
-        };
-    }
-
-    pub fn isRunning(self: *Backfiller) bool {
-        return self.running.load(.acquire);
-    }
-
-    pub fn start(self: *Backfiller, source: []const u8) !void {
-        if (self.running.cmpxchgStrong(false, true, .acq_rel, .acquire) != null) {
-            return error.AlreadyRunning;
-        }
-        errdefer self.running.store(false, .release);
-
-        self.source = try self.allocator.dupe(u8, source);
-        self.thread = try std.Thread.spawn(.{ .stack_size = 8 * 1024 * 1024 }, run, .{self});
-    }
-
-    fn run(self: *Backfiller) void {
-        defer {
-            self.allocator.free(self.source);
-            self.source = "";
-            self.thread = null;
-            self.running.store(false, .release);
-        }
-
-        const collections = self.discoverCollections() catch |err| {
-            log.err("collection discovery failed: {s}", .{@errorName(err)});
-            return;
-        };
-        defer {
-            for (collections) |c| self.allocator.free(c);
-            self.allocator.free(collections);
-        }
-
-        log.info("discovered {d} collections to backfill from {s}", .{ collections.len, self.source });
-
-        for (collections) |collection| {
-            self.backfillCollection(collection) catch |err| {
-                log.warn("backfill failed for {s}: {s}", .{ collection, @errorName(err) });
-            };
-        }
-
-        log.info("backfill complete", .{});
-    }
-
-    fn discoverCollections(self: *Backfiller) ![][]const u8 {
-        var seen: std.StringHashMapUnmanaged(void) = .{};
-        defer seen.deinit(self.allocator);
-
-        // source 1: lexicon garden
-        const garden = self.fetchLexiconGarden() catch |err| blk: {
-            log.warn("lexicon garden fetch failed: {s}", .{@errorName(err)});
-            break :blk &[_][]const u8{};
-        };
-        defer {
-            for (garden) |c| self.allocator.free(c);
-            self.allocator.free(garden);
-        }
-        for (garden) |c| {
-            if (!seen.contains(c)) {
-                const duped = try self.allocator.dupe(u8, c);
-                errdefer self.allocator.free(duped);
-                try seen.put(self.allocator, duped, {});
-            }
-        }
-
-        // source 2: observed collections from RBC scan
-        const observed = self.collection_index.listKnownCollections(self.allocator) catch |err| blk: {
-            log.warn("RBC scan failed: {s}", .{@errorName(err)});
-            break :blk &[_][]const u8{};
-        };
-        defer {
-            for (observed) |c| self.allocator.free(c);
-            self.allocator.free(observed);
-        }
-        for (observed) |c| {
-            if (!seen.contains(c)) {
-                const duped = try self.allocator.dupe(u8, c);
-                errdefer self.allocator.free(duped);
-                try seen.put(self.allocator, duped, {});
-            }
-        }
-
-        const result = try self.allocator.alloc([]const u8, seen.count());
-        var i: usize = 0;
-        var key_iter = seen.keyIterator();
-        while (key_iter.next()) |key| {
-            result[i] = key.*;
-            i += 1;
-        }
-        return result;
-    }
-
-    fn fetchLexiconGarden(self: *Backfiller) ![][]const u8 {
-        var client: http.Client = .{ .allocator = self.allocator };
-        defer client.deinit();
-
-        var aw: std.Io.Writer.Allocating = .init(self.allocator);
-        defer aw.deinit();
-
-        const result = client.fetch(.{
-            .location = .{ .url = "https://lexicon.garden/llms.txt" },
-            .response_writer = &aw.writer,
-            .method = .GET,
-        }) catch return error.FetchFailed;
-
-        if (result.status != .ok) return error.FetchFailed;
-
-        const body = aw.written();
-
-        var nsids: std.ArrayListUnmanaged([]const u8) = .{};
-        defer nsids.deinit(self.allocator);
-
-        var lines = std.mem.splitScalar(u8, body, '\n');
-        while (lines.next()) |line| {
-            const backtick_start = std.mem.indexOf(u8, line, "- [`") orelse continue;
-            const nsid_start = backtick_start + 4;
-            const rest = line[nsid_start..];
-            const backtick_end = std.mem.indexOf(u8, rest, "`](") orelse continue;
-            const nsid = rest[0..backtick_end];
-            if (!std.mem.containsAtLeast(u8, nsid, 1, ".")) continue;
-
-            const duped = try self.allocator.dupe(u8, nsid);
-            errdefer self.allocator.free(duped);
-            try nsids.append(self.allocator, duped);
-        }
-
-        log.info("lexicon garden: found {d} NSIDs", .{nsids.items.len});
-        return try nsids.toOwnedSlice(self.allocator);
-    }
-
-    // --- progress storage in RocksDB default CF ---
-    // key: "bf:{collection}\0{source}"
-    // value: "{cursor}\0{count}\0{0|1}" (cursor, imported count, completed)
-
-    fn progressKey(self: *const Backfiller, collection: []const u8, buf: []u8) ?[]const u8 {
-        const prefix = "bf:";
-        const needed = prefix.len + collection.len + 1 + self.source.len;
-        if (needed > buf.len) return null;
-        @memcpy(buf[0..prefix.len], prefix);
-        @memcpy(buf[prefix.len..][0..collection.len], collection);
-        buf[prefix.len + collection.len] = 0;
-        @memcpy(buf[prefix.len + collection.len + 1 ..][0..self.source.len], self.source);
-        return buf[0..needed];
-    }
-
-    const Progress = struct {
-        cursor: []const u8,
-        imported: i64,
-        completed: bool,
-        /// rocksdb-owned memory — caller must call deinit() when done
-        _data: @import("rocksdb").Data,
-
-        pub fn deinit(self: Progress) void {
-            self._data.deinit();
-        }
-    };
-
-    fn loadProgress(self: *Backfiller, collection: []const u8) ?Progress {
-        var key_buf: [512]u8 = undefined;
-        const key = self.progressKey(collection, &key_buf) orelse return null;
-        const data = self.collection_index.getMeta(key) orelse return null;
-        const val = data.data;
-
-        // parse: "{cursor}\0{count}\0{0|1}"
-        const sep1 = std.mem.indexOfScalar(u8, val, 0) orelse {
-            data.deinit();
-            return null;
-        };
-        const rest = val[sep1 + 1 ..];
-        const sep2 = std.mem.indexOfScalar(u8, rest, 0) orelse {
-            data.deinit();
-            return null;
-        };
-
-        return .{
-            .cursor = val[0..sep1],
-            .imported = std.fmt.parseInt(i64, rest[0..sep2], 10) catch 0,
-            .completed = rest.len > sep2 + 1 and rest[sep2 + 1] == '1',
-            ._data = data,
-        };
-    }
-
-    fn saveProgress(self: *Backfiller, collection: []const u8, cursor: []const u8, imported: i64, completed: bool) void {
-        var key_buf: [512]u8 = undefined;
-        const key = self.progressKey(collection, &key_buf) orelse return;
-
-        var val_buf: [512]u8 = undefined;
-        var pos: usize = 0;
-        if (cursor.len > 0) {
-            @memcpy(val_buf[0..cursor.len], cursor);
-            pos = cursor.len;
-        }
-        val_buf[pos] = 0;
-        pos += 1;
-        const count_str = std.fmt.bufPrint(val_buf[pos..], "{d}", .{imported}) catch return;
-        pos += count_str.len;
-        val_buf[pos] = 0;
-        pos += 1;
-        val_buf[pos] = if (completed) '1' else '0';
-        pos += 1;
-
-        self.collection_index.putMeta(key, val_buf[0..pos]);
-    }
-
-    fn backfillCollection(self: *Backfiller, collection: []const u8) !void {
-        // check saved progress
-        var cursor: ?[]const u8 = null;
-        defer if (cursor) |c| self.allocator.free(c);
-        var imported: i64 = 0;
-
-        if (self.loadProgress(collection)) |progress| {
-            defer progress.deinit();
-            if (progress.completed) return; // already done
-            imported = progress.imported;
-            if (progress.cursor.len > 0) {
-                cursor = try self.allocator.dupe(u8, progress.cursor);
-                log.info("{s}: resuming from cursor (imported {d} so far)", .{ collection, imported });
-            }
-        }
-
-        var client: http.Client = .{ .allocator = self.allocator };
-        defer client.deinit();
-
-        var page_count: usize = 0;
-        while (true) {
-            const fetch_result = self.fetchPage(&client, collection, cursor) catch |err| {
-                log.warn("{s}: fetch page failed: {s}", .{ collection, @errorName(err) });
-                break;
-            };
-            defer {
-                for (fetch_result.dids) |d| self.allocator.free(d);
-                self.allocator.free(fetch_result.dids);
-                if (fetch_result.next_cursor) |nc| self.allocator.free(nc);
-            }
-
-            for (fetch_result.dids) |did| {
-                self.collection_index.addCollection(did, collection) catch {};
-                imported += 1;
-            }
-
-            page_count += 1;
-
-            const new_cursor = fetch_result.next_cursor orelse "";
-            self.saveProgress(collection, new_cursor, imported, false);
-
-            if (fetch_result.next_cursor) |nc| {
-                if (cursor) |old| self.allocator.free(old);
-                cursor = self.allocator.dupe(u8, nc) catch break;
-                std.posix.nanosleep(0, 100 * std.time.ns_per_ms);
-            } else {
-                self.saveProgress(collection, "", imported, true);
-                log.info("{s}: complete ({d} DIDs, {d} pages)", .{ collection, imported, page_count });
-                break;
-            }
-        }
-    }
-
-    fn fetchPage(self: *Backfiller, client: *http.Client, collection: []const u8, cursor: ?[]const u8) !FetchResult {
-        var url_buf: [1024]u8 = undefined;
-        const url = if (cursor) |c|
-            std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.sync.listReposByCollection?collection={s}&limit=1000&cursor={s}", .{ self.source, collection, c }) catch return error.UrlTooLong
-        else
-            std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.sync.listReposByCollection?collection={s}&limit=1000", .{ self.source, collection }) catch return error.UrlTooLong;
-
-        var aw: std.Io.Writer.Allocating = .init(self.allocator);
-        defer aw.deinit();
-
-        const result = client.fetch(.{
-            .location = .{ .url = url },
-            .response_writer = &aw.writer,
-            .method = .GET,
-        }) catch return error.FetchFailed;
-
-        if (result.status != .ok) return error.FetchFailed;
-
-        const body = aw.written();
-
-        const parsed = std.json.parseFromSlice(ListReposResponse, self.allocator, body, .{ .ignore_unknown_fields = true }) catch return error.ParseFailed;
-        defer parsed.deinit();
-
-        const repos = parsed.value.repos orelse return .{
-            .dids = try self.allocator.alloc([]const u8, 0),
-            .next_cursor = null,
-        };
-
-        var dids: std.ArrayListUnmanaged([]const u8) = .{};
-        defer dids.deinit(self.allocator);
-
-        for (repos) |repo| {
-            const duped = self.allocator.dupe(u8, repo.did) catch continue;
-            dids.append(self.allocator, duped) catch {
-                self.allocator.free(duped);
-                continue;
-            };
-        }
-
-        const next_cursor = if (parsed.value.cursor) |c|
-            self.allocator.dupe(u8, c) catch null
-        else
-            null;
-
-        return .{
-            .dids = dids.toOwnedSlice(self.allocator) catch return error.OutOfMemory,
-            .next_cursor = next_cursor,
-        };
-    }
-
-    const FetchResult = struct {
-        dids: [][]const u8,
-        next_cursor: ?[]const u8,
-    };
-
-    const ListReposResponse = struct {
-        repos: ?[]const RepoEntry = null,
-        cursor: ?[]const u8 = null,
-    };
-
-    const RepoEntry = struct {
-        did: []const u8,
-    };
-
-    /// return status summary for the admin endpoint
-    pub fn getStatus(self: *Backfiller, allocator: Allocator) ![]u8 {
-        var list: std.ArrayListUnmanaged(u8) = .{};
-        defer list.deinit(allocator);
-        const w = list.writer(allocator);
-
-        var total: usize = 0;
-        var completed: usize = 0;
-        var total_imported: i64 = 0;
-
-        // scan all bf: keys in the default CF
-        const ci = self.collection_index;
-        var err_str: ?@import("rocksdb").Data = null;
-        var iter = ci.db.iterator(ci.default, .forward, "bf:");
-        defer iter.deinit();
-
-        var first_detail = true;
-        var details: std.ArrayListUnmanaged(u8) = .{};
-        defer details.deinit(allocator);
-        const dw = details.writer(allocator);
-
-        while (iter.next(&err_str) catch null) |entry| {
-            const key_data = entry[0].data;
-            if (!std.mem.startsWith(u8, key_data, "bf:")) break;
-
-            const val_data = entry[1].data;
-
-            // parse key: "bf:{collection}\0{source}"
-            const key_rest = key_data[3..]; // skip "bf:"
-            const sep = std.mem.indexOfScalar(u8, key_rest, 0) orelse continue;
-            const collection = key_rest[0..sep];
-            const source = key_rest[sep + 1 ..];
-
-            // parse value: "{cursor}\0{count}\0{0|1}"
-            const val_sep1 = std.mem.indexOfScalar(u8, val_data, 0) orelse continue;
-            const val_rest = val_data[val_sep1 + 1 ..];
-            const val_sep2 = std.mem.indexOfScalar(u8, val_rest, 0) orelse continue;
-            const count = std.fmt.parseInt(i64, val_rest[0..val_sep2], 10) catch 0;
-            const is_completed = val_rest.len > val_sep2 + 1 and val_rest[val_sep2 + 1] == '1';
-
-            total += 1;
-            if (is_completed) completed += 1;
-            total_imported += count;
-
-            if (!first_detail) dw.writeByte(',') catch {};
-            first_detail = false;
-
-            std.fmt.format(dw, "{{\"collection\":\"{s}\",\"source\":\"{s}\",\"imported\":{d},\"completed\":{}}}", .{
-                collection,
-                source,
-                count,
-                is_completed,
-            }) catch {};
-        }
-
-        std.fmt.format(w, "{{\"running\":{},\"total\":{d},\"completed\":{d},\"in_progress\":{d},\"total_imported\":{d},\"collections\":[", .{
-            self.isRunning(),
-            total,
-            completed,
-            total - completed,
-            total_imported,
-        }) catch return error.FormatError;
-
-        w.writeAll(details.items) catch {};
-        w.writeAll("]}") catch {};
-
-        return try list.toOwnedSlice(allocator);
-    }
-};
diff --git a/src/collectiondir_subscriber.zig b/src/collectiondir_subscriber.zig
deleted file mode 100644
index 72bfcba..0000000
--- a/src/collectiondir_subscriber.zig
+++ /dev/null
@@ -1,314 +0,0 @@
-//! collectiondir firehose subscriber — consumes relay's subscribeRepos stream
-//!
-//! simpler than the relay's subscriber.zig — no validation, no host management,
-//! no rate limiting. just CBOR frame decode → collection index updates.
-//!
-//! on #commit: extract ops, call trackCommitOps
-//! on #account: if inactive, removeAll from index
-//! on #sync: count only (resync needs PDS hostname — requires DID doc resolution, stage 2)
-//!
-//! cursor persisted to RocksDB default CF (key "cursor"), not postgres.
-
-const std = @import("std");
-const websocket = @import("websocket");
-const zat = @import("zat");
-const collection_index_mod = @import("collection_index.zig");
-
-const Allocator = std.mem.Allocator;
-const log = std.log.scoped(.collectiondir);
-
-const cursor_flush_interval_sec = 4;
-
-pub const Stats = struct {
-    commits: std.atomic.Value(u64) = .{ .raw = 0 },
-    accounts: std.atomic.Value(u64) = .{ .raw = 0 },
-    syncs: std.atomic.Value(u64) = .{ .raw = 0 },
-    identities: std.atomic.Value(u64) = .{ .raw = 0 },
-    decode_errors: std.atomic.Value(u64) = .{ .raw = 0 },
-    ops_tracked: std.atomic.Value(u64) = .{ .raw = 0 },
-    accounts_removed: std.atomic.Value(u64) = .{ .raw = 0 },
-    reconnects: std.atomic.Value(u64) = .{ .raw = 0 },
-    last_seq: std.atomic.Value(u64) = .{ .raw = 0 },
-    connected: std.atomic.Value(bool) = .{ .raw = false },
-    last_event_time: std.atomic.Value(i64) = .{ .raw = 0 },
-};
-
-pub const Subscriber = struct {
-    allocator: Allocator,
-    relay_url: []const u8,
-    relay_port: u16,
-    use_tls: bool,
-    collection_index: *collection_index_mod.CollectionIndex,
-    shutdown: *std.atomic.Value(bool),
-    stats: Stats = .{},
-    last_seq: ?u64 = null,
-    last_cursor_flush: i64 = 0,
-
-    pub fn init(
-        allocator: Allocator,
-        relay_url: []const u8,
-        relay_port: u16,
-        use_tls: bool,
-        collection_index: *collection_index_mod.CollectionIndex,
-        shutdown: *std.atomic.Value(bool),
-    ) Subscriber {
-        return .{
-            .allocator = allocator,
-            .relay_url = relay_url,
-            .relay_port = relay_port,
-            .use_tls = use_tls,
-            .collection_index = collection_index,
-            .shutdown = shutdown,
-        };
-    }
-
-    /// load saved cursor from RocksDB default CF
-    pub fn loadCursor(self: *Subscriber) void {
-        const val = self.collection_index.getMeta("cursor") orelse return;
-        defer val.deinit();
-        const seq = std.fmt.parseInt(u64, val.data, 10) catch return;
-        if (seq > 0) {
-            self.last_seq = seq;
-            log.info("resuming from cursor {d}", .{seq});
-        }
-    }
-
-    /// save cursor to RocksDB default CF
-    fn flushCursor(self: *Subscriber) void {
-        const seq = self.last_seq orelse return;
-        var buf: [20]u8 = undefined;
-        const val = std.fmt.bufPrint(&buf, "{d}", .{seq}) catch return;
-        self.collection_index.putMeta("cursor", val);
-    }
-
-    /// run the subscriber loop with reconnect + backoff
-    pub fn run(self: *Subscriber) void {
-        self.loadCursor();
-
-        var backoff: u64 = 1;
-        const max_backoff: u64 = 60;
-
-        while (!self.shutdown.load(.acquire)) {
-            log.info("connecting to {s}:{d}...", .{ self.relay_url, self.relay_port });
-
-            if (self.connectAndRead()) {
-                // clean disconnect (EOF) — reset backoff
-                backoff = 1;
-            } else |err| {
-                if (self.shutdown.load(.acquire)) return;
-                log.err("connection error: {s}, reconnecting in {d}s...", .{ @errorName(err), backoff });
-                _ = self.stats.reconnects.fetchAdd(1, .monotonic);
-            }
-
-            if (self.shutdown.load(.acquire)) return;
-
-            // flush cursor before backoff
-            self.flushCursor();
-
-            // backoff sleep in 1s chunks
-            var remaining: u64 = backoff;
-            while (remaining > 0 and !self.shutdown.load(.acquire)) {
-                std.posix.nanosleep(1, 0);
-                remaining -= 1;
-            }
-            backoff = @min(backoff * 2, max_backoff);
-        }
-
-        // final cursor flush
-        self.flushCursor();
-    }
-
-    fn connectAndRead(self: *Subscriber) !void {
-        var path_buf: [256]u8 = undefined;
-        var w: std.Io.Writer = .fixed(&path_buf);
-
-        try w.writeAll("/xrpc/com.atproto.sync.subscribeRepos");
-        if (self.last_seq) |cursor| {
-            try w.print("?cursor={d}", .{cursor});
-        }
-        const path = w.buffered();
-
-        var client = try websocket.Client.init(self.allocator, .{
-            .host = self.relay_url,
-            .port = self.relay_port,
-            .tls = self.use_tls,
-            .max_size = 5 * 1024 * 1024,
-        });
-        defer client.deinit();
-
-        var host_header_buf: [256]u8 = undefined;
-        const host_header = std.fmt.bufPrint(
-            &host_header_buf,
-            "Host: {s}\r\n",
-            .{self.relay_url},
-        ) catch self.relay_url;
-
-        try client.handshake(path, .{ .headers = host_header });
-        log.info("connected to {s}", .{self.relay_url});
-        self.stats.connected.store(true, .release);
-        defer self.stats.connected.store(false, .release);
-
-        var handler = FrameHandler{ .sub = self };
-        try client.readLoop(&handler);
-    }
-
-    pub fn formatMetrics(self: *Subscriber, buf: []u8) []const u8 {
-        var fbs = std.io.fixedBufferStream(buf);
-        const w = fbs.writer();
-
-        const commits = self.stats.commits.load(.monotonic);
-        const accounts = self.stats.accounts.load(.monotonic);
-        const syncs = self.stats.syncs.load(.monotonic);
-        const identities = self.stats.identities.load(.monotonic);
-        const decode_errors = self.stats.decode_errors.load(.monotonic);
-        const ops_tracked = self.stats.ops_tracked.load(.monotonic);
-        const accounts_removed = self.stats.accounts_removed.load(.monotonic);
-        const reconnects = self.stats.reconnects.load(.monotonic);
-        const last_seq = self.stats.last_seq.load(.monotonic);
-        const connected: u64 = if (self.stats.connected.load(.monotonic)) 1 else 0;
-        const last_event_time = self.stats.last_event_time.load(.monotonic);
-
-        std.fmt.format(w,
-            \\# HELP collectiondir_events_total events processed by type
-            \\# TYPE collectiondir_events_total counter
-            \\collectiondir_events_total{{type="commit"}} {d}
-            \\collectiondir_events_total{{type="account"}} {d}
-            \\collectiondir_events_total{{type="sync"}} {d}
-            \\collectiondir_events_total{{type="identity"}} {d}
-            \\# HELP collectiondir_decode_errors_total CBOR decode failures
-            \\# TYPE collectiondir_decode_errors_total counter
-            \\collectiondir_decode_errors_total {d}
-            \\# HELP collectiondir_ops_tracked_total collection index ops processed
-            \\# TYPE collectiondir_ops_tracked_total counter
-            \\collectiondir_ops_tracked_total {d}
-            \\# HELP collectiondir_accounts_removed_total accounts removed from index (inactive)
-            \\# TYPE collectiondir_accounts_removed_total counter
-            \\collectiondir_accounts_removed_total {d}
-            \\# HELP collectiondir_reconnects_total WebSocket reconnect count
-            \\# TYPE collectiondir_reconnects_total counter
-            \\collectiondir_reconnects_total {d}
-            \\# HELP collectiondir_last_seq last processed firehose sequence number
-            \\# TYPE collectiondir_last_seq gauge
-            \\collectiondir_last_seq {d}
-            \\# HELP collectiondir_connected whether firehose subscriber is connected
-            \\# TYPE collectiondir_connected gauge
-            \\collectiondir_connected {d}
-            \\# HELP collectiondir_last_event_time_seconds unix timestamp of last processed event
-            \\# TYPE collectiondir_last_event_time_seconds gauge
-            \\collectiondir_last_event_time_seconds {d}
-            \\
-        , .{
-            commits,
-            accounts,
-            syncs,
-            identities,
-            decode_errors,
-            ops_tracked,
-            accounts_removed,
-            reconnects,
-            last_seq,
-            connected,
-            last_event_time,
-        }) catch {};
-
-        return fbs.getWritten();
-    }
-};
-
-const FrameHandler = struct {
-    sub: *Subscriber,
-
-    pub fn serverMessage(self: *FrameHandler, data: []const u8) !void {
-        const sub = self.sub;
-
-        var arena = std.heap.ArenaAllocator.init(sub.allocator);
-        defer arena.deinit();
-        const alloc = arena.allocator();
-
-        // decode header
-        const header_result = zat.cbor.decode(alloc, data) catch |err| {
-            log.debug("frame header decode failed: {s}", .{@errorName(err)});
-            _ = sub.stats.decode_errors.fetchAdd(1, .monotonic);
-            return;
-        };
-        const header = header_result.value;
-        const payload_data = data[header_result.consumed..];
-
-        const op = header.getInt("op") orelse return;
-        if (op != 1) return; // skip error frames
-
-        const frame_type = header.getString("t") orelse return;
-        const payload = zat.cbor.decodeAll(alloc, payload_data) catch |err| {
-            log.debug("frame payload decode failed: {s} (type={s})", .{ @errorName(err), frame_type });
-            _ = sub.stats.decode_errors.fetchAdd(1, .monotonic);
-            return;
-        };
-
-        const upstream_seq = payload.getUint("seq");
-
-        // route by frame type
-        const is_commit = std.mem.eql(u8, frame_type, "#commit");
-        const is_account = std.mem.eql(u8, frame_type, "#account");
-        const is_sync = std.mem.eql(u8, frame_type, "#sync");
-        const is_identity = std.mem.eql(u8, frame_type, "#identity");
-
-        if (is_commit) {
-            _ = sub.stats.commits.fetchAdd(1, .monotonic);
-
-            const did = payload.getString("repo") orelse {
-                if (upstream_seq) |s| sub.last_seq = s;
-                return;
-            };
-
-            // track collection ops — the core purpose of this service
-            if (payload.get("ops")) |ops| {
-                sub.collection_index.trackCommitOps(did, ops);
-
-                // count individual ops for metrics
-                switch (ops) {
-                    .array => |items| {
-                        _ = sub.stats.ops_tracked.fetchAdd(items.len, .monotonic);
-                    },
-                    else => {},
-                }
-            }
-        } else if (is_account) {
-            _ = sub.stats.accounts.fetchAdd(1, .monotonic);
-
-            const did = payload.getString("did") orelse {
-                if (upstream_seq) |s| sub.last_seq = s;
-                return;
-            };
-
-            const is_active = payload.getBool("active") orelse false;
-            if (!is_active) {
-                sub.collection_index.removeAll(did) catch |err| {
-                    log.debug("removeAll failed for {s}: {s}", .{ did, @errorName(err) });
-                };
-                _ = sub.stats.accounts_removed.fetchAdd(1, .monotonic);
-            }
-        } else if (is_sync) {
-            _ = sub.stats.syncs.fetchAdd(1, .monotonic);
-            // stage 1: count only — resync requires PDS hostname via DID doc resolution (stage 2)
-        } else if (is_identity) {
-            _ = sub.stats.identities.fetchAdd(1, .monotonic);
-        } else {
-            if (upstream_seq) |s| sub.last_seq = s;
-            return;
-        }
-
-        // advance cursor
-        if (upstream_seq) |s| {
-            sub.last_seq = s;
-            sub.stats.last_seq.store(s, .monotonic);
-            sub.stats.last_event_time.store(std.time.timestamp(), .monotonic);
-        }
-
-        // periodic cursor flush
-        const now = std.time.timestamp();
-        if (now - sub.last_cursor_flush >= cursor_flush_interval_sec) {
-            sub.flushCursor();
-            sub.last_cursor_flush = now;
-        }
-    }
-};