refactor: split monolithic db.zig into focused modules

- store/db.zig: sqlite connection and schema only
- store/posts.zig: post CRUD with cursor pagination
- store/follows.zig: follows cache operations
- store/backfill.zig: backfill completion tracking
- bsky/auth.zig: JWT verification (from atproto.zig)
- bsky/api.zig: bluesky API calls (from atproto.zig)
- feed/skeleton.zig: feed response JSON building
- feed/backfill.zig: backfill orchestration
- jetstream.zig: moved to top level (was stream/)

removes 800+ lines of tangled code from stream/db.zig

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[e-mail]>

src/bsky/api.zig

···
+const std = @import("std");
+const Allocator = std.mem.Allocator;
+const zat = @import("zat");
+
+const log = std.log.scoped(.bsky_api);
+
+const BSKY_PUBLIC_API = "https://public.api.bsky.app";
+
+/// fetch all DIDs that a user follows (with pagination)
+pub fn getFollows(alloc: Allocator, actor_did: []const u8) ![][]const u8 {
+    var client = zat.XrpcClient.init(alloc, BSKY_PUBLIC_API);
+    defer client.deinit();
+
+    var result: std.ArrayList([]const u8) = .empty;
+    errdefer {
+        for (result.items) |item| alloc.free(item);
+        result.deinit(alloc);
+    }
+
+    var cursor: ?[]const u8 = null;
+    defer if (cursor) |c| alloc.free(c);
+
+    const nsid = zat.Nsid.parse("app.bsky.graph.getFollows") orelse return error.InvalidNsid;
+
+    while (true) {
+        var params = std.StringHashMap([]const u8).init(alloc);
+        defer params.deinit();
+
+        try params.put("actor", actor_did);
+        try params.put("limit", "100");
+        if (cursor) |c| {
+            try params.put("cursor", c);
+        }
+
+        var response = client.query(nsid, params) catch |err| {
+            log.err("getFollows API error: {}", .{err});
+            return error.ApiFailed;
+        };
+        defer response.deinit();
+
+        if (!response.ok()) {
+            return error.ApiFailed;
+        }
+
+        var parsed = response.json() catch return error.InvalidJson;
+        defer parsed.deinit();
+
+        const follows = zat.json.getArray(parsed.value, "follows") orelse break;
+
+        for (follows) |item| {
+            const did = zat.json.getString(item, "did") orelse continue;
+            try result.append(alloc, try alloc.dupe(u8, did));
+        }
+
+        // check for next page
+        if (cursor) |c| alloc.free(c);
+        cursor = null;
+
+        if (zat.json.getString(parsed.value, "cursor")) |c| {
+            if (c.len > 0) {
+                cursor = try alloc.dupe(u8, c);
+            }
+        }
+
+        if (cursor == null) break;
+    }
+
+    return try result.toOwnedSlice(alloc);
+}
+
+pub fn freeFollows(alloc: Allocator, follows: [][]const u8) void {
+    for (follows) |did| alloc.free(did);
+    alloc.free(follows);
+}
+
+/// post from getAuthorFeed
+pub const AuthorPost = struct {
+    uri: []const u8,
+    cid: []const u8,
+    text: []const u8,
+    embed_uri: ?[]const u8,
+};
+
+/// struct for extracting post data from feed items
+const FeedPost = struct {
+    uri: []const u8,
+    cid: []const u8,
+    record: struct {
+        text: []const u8 = "",
+        embed: ?struct {
+            external: ?struct { uri: []const u8 } = null,
+        } = null,
+    },
+};
+
+/// fetch recent posts from an author's feed
+pub fn getAuthorFeed(alloc: Allocator, actor_did: []const u8, limit: usize) ![]AuthorPost {
+    var client = zat.XrpcClient.init(alloc, BSKY_PUBLIC_API);
+    defer client.deinit();
+
+    var post_list: std.ArrayList(AuthorPost) = .empty;
+    errdefer {
+        for (post_list.items) |p| freeAuthorPost(alloc, p);
+        post_list.deinit(alloc);
+    }
+
+    const nsid = zat.Nsid.parse("app.bsky.feed.getAuthorFeed") orelse return error.InvalidNsid;
+
+    var params = std.StringHashMap([]const u8).init(alloc);
+    defer params.deinit();
+
+    var limit_buf: [8]u8 = undefined;
+    const limit_str = std.fmt.bufPrint(&limit_buf, "{d}", .{@min(limit, 100)}) catch return error.FormatError;
+
+    try params.put("actor", actor_did);
+    try params.put("limit", limit_str);
+    try params.put("filter", "posts_no_replies");
+
+    var response = client.query(nsid, params) catch |err| {
+        log.err("getAuthorFeed API error for {s}: {}", .{ actor_did, err });
+        return error.ApiFailed;
+    };
+    defer response.deinit();
+
+    if (!response.ok()) {
+        return error.ApiFailed;
+    }
+
+    var parsed = response.json() catch return error.InvalidJson;
+    defer parsed.deinit();
+
+    const feed = zat.json.getArray(parsed.value, "feed") orelse return try post_list.toOwnedSlice(alloc);
+
+    for (feed) |item| {
+        const post = zat.json.extractAt(FeedPost, alloc, item, .{"post"}) catch continue;
+
+        const embed_uri = if (post.record.embed) |e| if (e.external) |ext| ext.uri else null else null;
+
+        try post_list.append(alloc, .{
+            .uri = try alloc.dupe(u8, post.uri),
+            .cid = try alloc.dupe(u8, post.cid),
+            .text = try alloc.dupe(u8, post.record.text),
+            .embed_uri = if (embed_uri) |eu| try alloc.dupe(u8, eu) else null,
+        });
+    }
+
+    return try post_list.toOwnedSlice(alloc);
+}
+
+pub fn freeAuthorPost(alloc: Allocator, post: AuthorPost) void {
+    alloc.free(post.uri);
+    alloc.free(post.cid);
+    alloc.free(post.text);
+    if (post.embed_uri) |eu| alloc.free(eu);
+}
+
+pub fn freeAuthorFeed(alloc: Allocator, posts: []AuthorPost) void {
+    for (posts) |p| freeAuthorPost(alloc, p);
+    alloc.free(posts);
+}

src/bsky/auth.zig

···
+const std = @import("std");
+const mem = std.mem;
+const Allocator = mem.Allocator;
+const zat = @import("zat");
+
+const log = std.log.scoped(.bsky_auth);
+
+/// extract requester DID from Authorization header (claims-only, no signature verification)
+pub fn getRequesterDid(alloc: Allocator, auth_header: ?[]const u8, service_did: []const u8) ?[]const u8 {
+    const auth = auth_header orelse return null;
+    if (!mem.startsWith(u8, auth, "Bearer ")) return null;
+
+    const token = auth["Bearer ".len..];
+
+    var jwt = zat.Jwt.parse(alloc, token) catch return null;
+    defer jwt.deinit();
+
+    if (jwt.isExpired()) return null;
+    if (!mem.eql(u8, jwt.payload.aud, service_did)) return null;
+
+    return alloc.dupe(u8, jwt.payload.iss) catch null;
+}
+
+/// extract requester DID with full signature verification
+pub fn getRequesterDidVerified(alloc: Allocator, auth_header: ?[]const u8, service_did: []const u8) ?[]const u8 {
+    const auth = auth_header orelse return null;
+    if (!mem.startsWith(u8, auth, "Bearer ")) return null;
+
+    const token = auth["Bearer ".len..];
+
+    var jwt = zat.Jwt.parse(alloc, token) catch |err| {
+        log.debug("jwt parse failed: {}", .{err});
+        return null;
+    };
+    defer jwt.deinit();
+
+    if (jwt.isExpired()) {
+        log.debug("jwt expired", .{});
+        return null;
+    }
+    if (!mem.eql(u8, jwt.payload.aud, service_did)) {
+        log.debug("jwt audience mismatch", .{});
+        return null;
+    }
+
+    // resolve issuer's DID document to get signing key
+    const did = zat.Did.parse(jwt.payload.iss) orelse {
+        log.debug("invalid issuer DID: {s}", .{jwt.payload.iss});
+        return null;
+    };
+
+    var resolver = zat.DidResolver.init(alloc);
+    defer resolver.deinit();
+
+    var doc = resolver.resolve(did) catch |err| {
+        log.debug("DID resolution failed: {}", .{err});
+        return null;
+    };
+    defer doc.deinit();
+
+    const signing_key = doc.signingKey() orelse {
+        log.debug("no signing key in DID document", .{});
+        return null;
+    };
+
+    jwt.verify(signing_key.public_key_multibase) catch |err| {
+        log.debug("jwt signature verification failed: {}", .{err});
+        return null;
+    };
+
+    return alloc.dupe(u8, jwt.payload.iss) catch null;
+}

src/feed/backfill.zig

···
+const std = @import("std");
+const Allocator = std.mem.Allocator;
+const api = @import("../bsky/api.zig");
+const posts = @import("../store/posts.zig");
+const backfill_store = @import("../store/backfill.zig");
+const filter = @import("filter.zig");
+
+const log = std.log.scoped(.backfill);
+
+/// backfill posts from a single account
+fn backfillAccount(alloc: Allocator, did: []const u8) void {
+    if (backfill_store.isComplete(did)) return;
+
+    log.debug("backfilling {s}...", .{did});
+
+    const author_posts = api.getAuthorFeed(alloc, did, 50) catch |err| {
+        log.debug("backfill failed for {s}: {}", .{ did, err });
+        return;
+    };
+    defer api.freeAuthorFeed(alloc, author_posts);
+
+    var added: usize = 0;
+    for (author_posts) |post| {
+        const text_match = filter.textContainsMusicLink(post.text);
+        const embed_match = if (post.embed_uri) |eu| filter.textContainsMusicLink(eu) else false;
+        if (text_match or embed_match) {
+            posts.add(post.uri, post.cid) catch continue;
+            added += 1;
+        }
+    }
+
+    backfill_store.markComplete(did);
+    log.debug("backfilled {s}: {d} music posts from {d} total", .{ did, added, author_posts.len });
+}
+
+/// trigger backfill for a list of DIDs (called after fetching follows)
+pub fn trigger(alloc: Allocator, dids: []const []const u8) void {
+    const max_per_request = 10;
+    var backfilled: usize = 0;
+
+    for (dids) |did| {
+        if (backfilled >= max_per_request) break;
+        if (!backfill_store.isComplete(did)) {
+            backfillAccount(alloc, did);
+            backfilled += 1;
+        }
+    }
+}

src/feed/skeleton.zig

···
+const std = @import("std");
+const Allocator = std.mem.Allocator;
+const posts = @import("../store/posts.zig");
+
+/// build a feed skeleton response JSON
+pub fn build(
+    alloc: Allocator,
+    post_list: []const posts.Post,
+    next_cursor: ?posts.Cursor,
+) ![]const u8 {
+    var buf: std.ArrayList(u8) = .empty;
+    errdefer buf.deinit(alloc);
+    const w = buf.writer(alloc);
+
+    try w.writeAll("{\"cursor\":");
+
+    if (next_cursor) |c| {
+        var cursor_buf: [128]u8 = undefined;
+        const cursor_str = c.format(&cursor_buf);
+        try w.print("\"{s}\"", .{cursor_str});
+    } else {
+        try w.writeAll("\"eof\"");
+    }
+
+    try w.writeAll(",\"feed\":[");
+
+    for (post_list, 0..) |post, i| {
+        if (i > 0) try w.writeAll(",");
+        try w.print("{{\"post\":\"{s}\"}}", .{post.uri});
+    }
+
+    try w.writeAll("]}");
+
+    return try buf.toOwnedSlice(alloc);
+}
+
+/// build an empty/EOF response
+pub fn empty(alloc: Allocator) ![]const u8 {
+    return try alloc.dupe(u8, "{\"cursor\":\"eof\",\"feed\":[]}");
+}
+
+test "build empty feed" {
+    const alloc = std.testing.allocator;
+    const result = try build(alloc, &.{}, null);
+    defer alloc.free(result);
+    try std.testing.expectEqualStrings("{\"cursor\":\"eof\",\"feed\":[]}", result);
+}
+
+test "build feed with posts" {
+    const alloc = std.testing.allocator;
+
+    const post_list = [_]posts.Post{
+        .{ .uri = "at://did:plc:abc/app.bsky.feed.post/123", .cid = "cid1", .indexed_at = 1000 },
+        .{ .uri = "at://did:plc:xyz/app.bsky.feed.post/456", .cid = "cid2", .indexed_at = 900 },
+    };
+
+    const cursor = posts.Cursor{ .timestamp = 900, .cid = "cid2" };
+    const result = try build(alloc, &post_list, cursor);
+    defer alloc.free(result);
+
+    try std.testing.expect(std.mem.indexOf(u8, result, "\"cursor\":\"900::cid2\"") != null);
+    try std.testing.expect(std.mem.indexOf(u8, result, "at://did:plc:abc/app.bsky.feed.post/123") != null);
+    try std.testing.expect(std.mem.indexOf(u8, result, "at://did:plc:xyz/app.bsky.feed.post/456") != null);
+}

src/main.zig

···
 const Thread = std.Thread;
 
 const http = @import("server/http.zig");
-const db = @import("stream/db.zig");
-const jetstream = @import("stream/jetstream.zig");
+const db = @import("store/db.zig");
+const jetstream = @import("jetstream.zig");
 const stats = @import("server/stats.zig");
 
 const log = std.log.scoped(.main);
···
 
 test {
     _ = @import("feed/filter.zig");
+    _ = @import("feed/skeleton.zig");
     _ = @import("server/stats.zig");
+    _ = @import("store/posts.zig");
 }

src/server/dashboard.zig

···
 const std = @import("std");
 const stats = @import("stats.zig");
-const db = @import("../stream/db.zig");
+const posts = @import("../store/posts.zig");
 
 fn formatNumber(buf: []u8, n: u64) []const u8 {
     var temp: [32]u8 = undefined;
···
 
 pub fn render(alloc: std.mem.Allocator) ![]const u8 {
     const s = stats.get();
-    const post_count = db.getPostCount();
+    const post_count = posts.count();
     const lag_ms = s.getPostLagMs();
     const lag_sec = @divTrunc(lag_ms, 1000);
     const status = s.getStatus();

src/server/http.zig

···
 const mem = std.mem;
 const config = @import("../feed/config.zig");
 const dashboard = @import("dashboard.zig");
-const atproto = @import("../stream/atproto.zig");
+const auth = @import("../bsky/auth.zig");
+const api = @import("../bsky/api.zig");
 const stats = @import("stats.zig");
-const db = @import("../stream/db.zig");
+const posts = @import("../store/posts.zig");
+const skeleton = @import("../feed/skeleton.zig");
+const backfill = @import("../feed/backfill.zig");
 
 const HTTP_BUF_SIZE = 8192;
 
···
 
     // use verified version for full signature verification
     // falls back to claims-only if DID resolution fails
-    var requester_did = atproto.getRequesterDidVerified(alloc, auth_header, service_did);
+    var requester_did = auth.getRequesterDidVerified(alloc, auth_header, service_did);
     if (requester_did == null) {
         // fallback to claims-only verification
-        requester_did = atproto.getRequesterDid(alloc, auth_header, service_did);
+        requester_did = auth.getRequesterDid(alloc, auth_header, service_did);
     }
 
     // parse query params
···
         std.debug.print("feed request: {s} (anonymous)\n", .{@tagName(feed_type)});
     }
 
-    const cursor = parseQueryParam(alloc, target, "cursor") catch null;
+    const cursor_str = parseQueryParam(alloc, target, "cursor") catch null;
     const limit_str = parseQueryParam(alloc, target, "limit") catch "20";
     const limit = std.fmt.parseInt(usize, limit_str, 10) catch 20;
 
-    const result = db.getPosts(alloc, feed_type, requester_did, cursor, limit) catch |err| {
+    // handle eof cursor
+    if (cursor_str) |c| {
+        if (mem.eql(u8, c, "eof")) {
+            const empty_response = skeleton.empty(alloc) catch {
+                try sendJson(request, .internal_server_error, "{\"error\":\"internal error\"}");
+                return;
+            };
+            try sendJson(request, .ok, empty_response);
+            return;
+        }
+    }
+
+    // parse cursor
+    const cursor = if (cursor_str) |c| posts.Cursor.parse(c) else null;
+
+    // get author filter for following feed
+    var author_filter: ?[][]const u8 = null;
+    defer if (author_filter) |f| api.freeFollows(alloc, f);
+
+    if (feed_type == .following) {
+        if (requester_did) |did| {
+            // fetch follows
+            author_filter = api.getFollows(alloc, did) catch |err| blk: {
+                std.debug.print("failed to fetch follows: {}\n", .{err});
+                break :blk null;
+            };
+
+            // trigger backfill for followed accounts
+            if (author_filter) |follows| {
+                backfill.trigger(alloc, follows);
+            }
+        } else {
+            // anonymous request to following feed - return empty
+            const empty_response = skeleton.empty(alloc) catch {
+                try sendJson(request, .internal_server_error, "{\"error\":\"internal error\"}");
+                return;
+            };
+            try sendJson(request, .ok, empty_response);
+            return;
+        }
+    }
+
+    // query posts
+    const query_result = posts.query(alloc, cursor, limit, author_filter) catch |err| {
         std.debug.print("feed error: {}\n", .{err});
-        try sendJson(request, .internal_server_error,
-            \\{"error":"internal error"}
-        );
+        try sendJson(request, .internal_server_error, "{\"error\":\"internal error\"}");
+        return;
+    };
+    defer posts.freeQueryResult(alloc, query_result.posts);
+
+    // build response
+    const result = skeleton.build(alloc, query_result.posts, query_result.next_cursor) catch |err| {
+        std.debug.print("skeleton build error: {}\n", .{err});
+        try sendJson(request, .internal_server_error, "{\"error\":\"internal error\"}");
         return;
     };
 
···
     const last_event_ago = if (last_event_at > 0) @divTrunc(now_ms - last_event_at, 1000) else 0;
     const last_match_ago = if (last_match_at > 0) @divTrunc(now_ms - last_match_at, 1000) else 0;
 
-    const post_count = db.getPostCount();
+    const post_count = posts.count();
     const status = s.getStatus();
 
     var buf: std.ArrayList(u8) = .{};

src/store/backfill.zig

···
+const std = @import("std");
+const db = @import("db.zig");
+
+/// check if an account has been backfilled
+pub fn isComplete(did: []const u8) bool {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return false;
+    var rows = conn.rows(
+        "SELECT 1 FROM backfilled_accounts WHERE did = ?",
+        .{did},
+    ) catch return false;
+    defer rows.deinit();
+
+    return rows.next() != null;
+}
+
+/// mark an account as backfilled
+pub fn markComplete(did: []const u8) void {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return;
+    const now = std.time.milliTimestamp();
+    conn.exec(
+        "INSERT OR REPLACE INTO backfilled_accounts (did, backfilled_at) VALUES (?, ?)",
+        .{ did, now },
+    ) catch {};
+}

src/store/db.zig

···
+const std = @import("std");
+const posix = std.posix;
+const zqlite = @import("zqlite");
+
+var conn: ?zqlite.Conn = null;
+var mutex: std.Thread.Mutex = .{};
+
+var path_buf: [256]u8 = undefined;
+
+pub fn init() !void {
+    const path_env = posix.getenv("DATABASE_PATH") orelse "/data/feed.db";
+
+    @memcpy(path_buf[0..path_env.len], path_env);
+    path_buf[path_env.len] = 0;
+    const path: [*:0]const u8 = path_buf[0..path_env.len :0];
+
+    std.debug.print("opening database: {s}\n", .{path_env});
+
+    const flags = zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite;
+    conn = zqlite.open(path, flags) catch |err| {
+        std.debug.print("failed to open database: {}\n", .{err});
+        return err;
+    };
+
+    // enable WAL mode for better concurrency
+    _ = conn.?.exec("PRAGMA journal_mode=WAL", .{}) catch {};
+    _ = conn.?.exec("PRAGMA busy_timeout=5000", .{}) catch {};
+
+    // create tables
+    try initSchema();
+
+    std.debug.print("database initialized\n", .{});
+}
+
+fn initSchema() !void {
+    const c = conn orelse return error.NotInitialized;
+
+    c.exec(
+        \\CREATE TABLE IF NOT EXISTS posts (
+        \\  uri TEXT PRIMARY KEY,
+        \\  cid TEXT NOT NULL,
+        \\  indexed_at INTEGER NOT NULL,
+        \\  author_did TEXT
+        \\)
+    , .{}) catch |err| {
+        std.debug.print("failed to create posts table: {}\n", .{err});
+        return err;
+    };
+
+    c.exec(
+        "CREATE INDEX IF NOT EXISTS idx_posts_indexed_at ON posts(indexed_at DESC)",
+        .{},
+    ) catch {};
+
+    c.exec(
+        "CREATE INDEX IF NOT EXISTS idx_posts_author ON posts(author_did)",
+        .{},
+    ) catch {};
+
+    c.exec(
+        \\CREATE TABLE IF NOT EXISTS follows_cache (
+        \\  requester_did TEXT NOT NULL,
+        \\  follows_did TEXT NOT NULL,
+        \\  cached_at INTEGER NOT NULL,
+        \\  PRIMARY KEY (requester_did, follows_did)
+        \\)
+    , .{}) catch |err| {
+        std.debug.print("failed to create follows_cache table: {}\n", .{err});
+        return err;
+    };
+
+    c.exec(
+        "CREATE INDEX IF NOT EXISTS idx_follows_cache_requester ON follows_cache(requester_did, cached_at)",
+        .{},
+    ) catch {};
+
+    c.exec(
+        \\CREATE TABLE IF NOT EXISTS backfilled_accounts (
+        \\  did TEXT PRIMARY KEY,
+        \\  backfilled_at INTEGER NOT NULL
+        \\)
+    , .{}) catch |err| {
+        std.debug.print("failed to create backfilled_accounts table: {}\n", .{err});
+        return err;
+    };
+}
+
+pub fn getConn() ?zqlite.Conn {
+    return conn;
+}
+
+pub fn lock() void {
+    mutex.lock();
+}
+
+pub fn unlock() void {
+    mutex.unlock();
+}
+
+/// execute with mutex held
+pub fn withLock(comptime func: fn (?zqlite.Conn) anyerror!void) !void {
+    mutex.lock();
+    defer mutex.unlock();
+    try func(conn);
+}

src/store/follows.zig

···
+const std = @import("std");
+const db = @import("db.zig");
+
+/// cache follows for a requester (replaces existing)
+pub fn cache(requester_did: []const u8, follows: []const []const u8) !void {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return error.NotInitialized;
+    const now = std.time.milliTimestamp();
+
+    // clear old entries for this requester
+    conn.exec(
+        "DELETE FROM follows_cache WHERE requester_did = ?",
+        .{requester_did},
+    ) catch {};
+
+    // insert new follows
+    for (follows) |did| {
+        conn.exec(
+            "INSERT INTO follows_cache (requester_did, follows_did, cached_at) VALUES (?, ?, ?)",
+            .{ requester_did, did, now },
+        ) catch {};
+    }
+}
+
+/// get cached follows as slice of DIDs (caller owns memory)
+pub fn get(alloc: std.mem.Allocator, requester_did: []const u8) ![][]const u8 {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return error.NotInitialized;
+
+    var result = std.ArrayList([]const u8).init(alloc);
+    errdefer {
+        for (result.items) |item| alloc.free(item);
+        result.deinit();
+    }
+
+    var rows = conn.rows(
+        "SELECT follows_did FROM follows_cache WHERE requester_did = ?",
+        .{requester_did},
+    ) catch return error.QueryFailed;
+    defer rows.deinit();
+
+    while (rows.next()) |row| {
+        const did = try alloc.dupe(u8, row.text(0));
+        try result.append(did);
+    }
+
+    return try result.toOwnedSlice();
+}
+
+pub fn freeFollows(alloc: std.mem.Allocator, follows: [][]const u8) void {
+    for (follows) |did| alloc.free(did);
+    alloc.free(follows);
+}
+
+/// check if we have cached follows for a requester
+pub fn hasCached(requester_did: []const u8) bool {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return false;
+    var rows = conn.rows(
+        "SELECT 1 FROM follows_cache WHERE requester_did = ? LIMIT 1",
+        .{requester_did},
+    ) catch return false;
+    defer rows.deinit();
+
+    return rows.next() != null;
+}

src/store/posts.zig

···
+const std = @import("std");
+const zat = @import("zat");
+const db = @import("db.zig");
+
+pub fn add(uri: []const u8, cid: []const u8) !void {
+    try addWithTimestamp(uri, cid, null);
+}
+
+pub fn addWithTimestamp(uri: []const u8, cid: []const u8, timestamp: ?i64) !void {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return error.NotInitialized;
+
+    const ts = timestamp orelse parseTimestampFromUri(uri) orelse std.time.milliTimestamp();
+    const author_did = extractDidFromUri(uri);
+
+    conn.exec(
+        "INSERT OR IGNORE INTO posts (uri, cid, indexed_at, author_did) VALUES (?, ?, ?, ?)",
+        .{ uri, cid, ts, author_did },
+    ) catch |err| {
+        std.debug.print("failed to insert post: {}\n", .{err});
+        return err;
+    };
+}
+
+pub fn count() usize {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return 0;
+    var rows = conn.rows("SELECT COUNT(*) FROM posts", .{}) catch return 0;
+    defer rows.deinit();
+
+    if (rows.next()) |row| {
+        return @intCast(row.int(0));
+    }
+    return 0;
+}
+
+/// row from posts table
+pub const Post = struct {
+    uri: []const u8,
+    cid: []const u8,
+    indexed_at: i64,
+};
+
+/// cursor for pagination
+pub const Cursor = struct {
+    timestamp: i64,
+    cid: []const u8,
+
+    pub fn parse(cursor_str: []const u8) ?Cursor {
+        const sep = std.mem.indexOf(u8, cursor_str, "::") orelse return null;
+        const ts = std.fmt.parseInt(i64, cursor_str[0..sep], 10) catch return null;
+        return .{
+            .timestamp = ts,
+            .cid = cursor_str[sep + 2 ..],
+        };
+    }
+
+    pub fn format(self: Cursor, buf: []u8) []const u8 {
+        return std.fmt.bufPrint(buf, "{d}::{s}", .{ self.timestamp, self.cid }) catch buf[0..0];
+    }
+};
+
+/// query posts with cursor-based pagination
+pub fn query(
+    alloc: std.mem.Allocator,
+    cursor: ?Cursor,
+    limit: usize,
+    author_filter: ?[]const []const u8,
+) !struct { posts: []Post, next_cursor: ?Cursor } {
+    db.lock();
+    defer db.unlock();
+
+    const conn = db.getConn() orelse return error.NotInitialized;
+
+    const cursor_ts = if (cursor) |c| c.timestamp else std.math.maxInt(i64);
+    const cursor_cid = if (cursor) |c| c.cid else "";
+
+    var post_list: std.ArrayList(Post) = .empty;
+    errdefer post_list.deinit(alloc);
+
+    var next_cursor: ?Cursor = null;
+
+    if (author_filter) |authors| {
+        if (authors.len == 0) {
+            return .{ .posts = try post_list.toOwnedSlice(alloc), .next_cursor = null };
+        }
+
+        // build IN clause query
+        var query_buf: std.ArrayList(u8) = .empty;
+        defer query_buf.deinit(alloc);
+        const qw = query_buf.writer(alloc);
+
+        try qw.writeAll("SELECT uri, cid, indexed_at FROM posts WHERE author_did IN (");
+
+        for (authors, 0..) |did, i| {
+            if (i > 0) try qw.writeAll(",");
+            try qw.print("'{s}'", .{did});
+        }
+
+        try qw.print(
+            ") AND (indexed_at < {d} OR (indexed_at = {d} AND cid < '{s}')) ORDER BY indexed_at DESC, cid DESC LIMIT {d}",
+            .{ cursor_ts, cursor_ts, cursor_cid, limit },
+        );
+
+        try query_buf.append(alloc, 0);
+        const query_z: [:0]const u8 = query_buf.items[0 .. query_buf.items.len - 1 :0];
+
+        var rows = conn.rows(query_z, .{}) catch return error.QueryFailed;
+        defer rows.deinit();
+
+        while (rows.next()) |row| {
+            const uri = try alloc.dupe(u8, row.text(0));
+            const cid = try alloc.dupe(u8, row.text(1));
+            const ts = row.int(2);
+
+            try post_list.append(alloc, .{ .uri = uri, .cid = cid, .indexed_at = ts });
+            next_cursor = .{ .timestamp = ts, .cid = cid };
+        }
+    } else {
+        var rows = conn.rows(
+            \\SELECT uri, cid, indexed_at FROM posts
+            \\WHERE indexed_at < ? OR (indexed_at = ? AND cid < ?)
+            \\ORDER BY indexed_at DESC, cid DESC
+            \\LIMIT ?
+        , .{ cursor_ts, cursor_ts, cursor_cid, limit }) catch return error.QueryFailed;
+        defer rows.deinit();
+
+        while (rows.next()) |row| {
+            const uri = try alloc.dupe(u8, row.text(0));
+            const cid = try alloc.dupe(u8, row.text(1));
+            const ts = row.int(2);
+
+            try post_list.append(alloc, .{ .uri = uri, .cid = cid, .indexed_at = ts });
+            next_cursor = .{ .timestamp = ts, .cid = cid };
+        }
+    }
+
+    return .{ .posts = try post_list.toOwnedSlice(alloc), .next_cursor = next_cursor };
+}
+
+pub fn freeQueryResult(alloc: std.mem.Allocator, posts: []Post) void {
+    for (posts) |p| {
+        alloc.free(p.uri);
+        alloc.free(p.cid);
+    }
+    alloc.free(posts);
+}
+
+// helpers
+
+/// parse TID (rkey) to get timestamp in milliseconds
+pub fn parseTidTimestamp(rkey: []const u8) i64 {
+    const tid = zat.Tid.parse(rkey) orelse return 0;
+    return @intCast(tid.timestamp() / 1000); // microseconds -> milliseconds
+}
+
+fn parseTimestampFromUri(uri: []const u8) ?i64 {
+    const at_uri = zat.AtUri.parse(uri) orelse return null;
+    const rkey = at_uri.rkey() orelse return null;
+    const tid = zat.Tid.parse(rkey) orelse return null;
+    const ts: i64 = @intCast(tid.timestamp() / 1000);
+    return if (ts > 0) ts else null;
+}
+
+fn extractDidFromUri(uri: []const u8) ?[]const u8 {
+    const at_uri = zat.AtUri.parse(uri) orelse return null;
+    return at_uri.authority();
+}
+
+// tests
+
+test "Cursor.parse" {
+    const c = Cursor.parse("1234567890::abc123") orelse unreachable;
+    try std.testing.expectEqual(1234567890, c.timestamp);
+    try std.testing.expectEqualStrings("abc123", c.cid);
+
+    try std.testing.expect(Cursor.parse("invalid") == null);
+    try std.testing.expect(Cursor.parse("notanumber::cid") == null);
+}
+
+test "Cursor.format" {
+    var buf: [64]u8 = undefined;
+    const c = Cursor{ .timestamp = 1234567890, .cid = "abc123" };
+    const formatted = c.format(&buf);
+    try std.testing.expectEqualStrings("1234567890::abc123", formatted);
+}

src/stream/atproto.zig

···
-const std = @import("std");
-const mem = std.mem;
-const Allocator = mem.Allocator;
-const zat = @import("zat");
-
-// =============================================================================
-// atproto utilities
-//
-// JWT verification and API helpers using zat SDK.
-// =============================================================================
-
-const log = std.log.scoped(.atproto);
-
-/// extract requester DID from an HTTP Authorization header (claims-only verification).
-pub fn getRequesterDid(allocator: Allocator, auth_header: ?[]const u8, service_did: []const u8) ?[]const u8 {
-    const auth = auth_header orelse return null;
-    if (!mem.startsWith(u8, auth, "Bearer ")) return null;
-
-    const token = auth["Bearer ".len..];
-
-    var jwt = zat.Jwt.parse(allocator, token) catch return null;
-    defer jwt.deinit();
-
-    // check claims
-    if (jwt.isExpired()) return null;
-    if (!mem.eql(u8, jwt.payload.aud, service_did)) return null;
-
-    return allocator.dupe(u8, jwt.payload.iss) catch null;
-}
-
-/// extract requester DID with full signature verification.
-pub fn getRequesterDidVerified(allocator: Allocator, auth_header: ?[]const u8, service_did: []const u8) ?[]const u8 {
-    const auth = auth_header orelse return null;
-    if (!mem.startsWith(u8, auth, "Bearer ")) return null;
-
-    const token = auth["Bearer ".len..];
-
-    var jwt = zat.Jwt.parse(allocator, token) catch |err| {
-        log.debug("jwt parse failed: {}", .{err});
-        return null;
-    };
-    defer jwt.deinit();
-
-    // check claims
-    if (jwt.isExpired()) {
-        log.debug("jwt expired", .{});
-        return null;
-    }
-    if (!mem.eql(u8, jwt.payload.aud, service_did)) {
-        log.debug("jwt audience mismatch", .{});
-        return null;
-    }
-
-    // resolve issuer's DID document to get signing key
-    const did = zat.Did.parse(jwt.payload.iss) orelse {
-        log.debug("invalid issuer DID: {s}", .{jwt.payload.iss});
-        return null;
-    };
-
-    var resolver = zat.DidResolver.init(allocator);
-    defer resolver.deinit();
-
-    var doc = resolver.resolve(did) catch |err| {
-        log.debug("DID resolution failed: {}", .{err});
-        return null;
-    };
-    defer doc.deinit();
-
-    const signing_key = doc.signingKey() orelse {
-        log.debug("no signing key in DID document", .{});
-        return null;
-    };
-
-    // verify signature
-    jwt.verify(signing_key.public_key_multibase) catch |err| {
-        log.debug("jwt signature verification failed: {}", .{err});
-        return null;
-    };
-
-    return allocator.dupe(u8, jwt.payload.iss) catch null;
-}
-
-// -----------------------------------------------------------------------------
-// social graph API (using zat.XrpcClient)
-// -----------------------------------------------------------------------------
-
-const BSKY_PUBLIC_API = "https://public.api.bsky.app";
-
-/// fetch all DIDs that a user follows (with pagination).
-/// returns newline-separated DIDs.
-pub fn getFollows(allocator: Allocator, actor_did: []const u8) ![]u8 {
-    var client = zat.XrpcClient.init(allocator, BSKY_PUBLIC_API);
-    defer client.deinit();
-
-    var result: std.ArrayList(u8) = .empty;
-    errdefer result.deinit(allocator);
-
-    var cursor: ?[]const u8 = null;
-    defer if (cursor) |c| allocator.free(c);
-
-    const nsid = zat.Nsid.parse("app.bsky.graph.getFollows") orelse return error.InvalidNsid;
-
-    while (true) {
-        var params = std.StringHashMap([]const u8).init(allocator);
-        defer params.deinit();
-
-        try params.put("actor", actor_did);
-        try params.put("limit", "100");
-        if (cursor) |c| {
-            try params.put("cursor", c);
-        }
-
-        var response = client.query(nsid, params) catch |err| {
-            log.err("getFollows API error: {}", .{err});
-            return error.ApiFailed;
-        };
-        defer response.deinit();
-
-        if (!response.ok()) {
-            return error.ApiFailed;
-        }
-
-        var parsed = response.json() catch return error.InvalidJson;
-        defer parsed.deinit();
-
-        // extract follows array using zat.json helpers
-        const follows = zat.json.getArray(parsed.value, "follows") orelse break;
-
-        for (follows) |item| {
-            const did = zat.json.getString(item, "did") orelse continue;
-            try result.appendSlice(allocator, did);
-            try result.append(allocator, '\n');
-        }
-
-        // check for next page
-        if (cursor) |c| allocator.free(c);
-        cursor = null;
-
-        if (zat.json.getString(parsed.value, "cursor")) |c| {
-            if (c.len > 0) {
-                cursor = try allocator.dupe(u8, c);
-            }
-        }
-
-        if (cursor == null) break;
-    }
-
-    return try result.toOwnedSlice(allocator);
-}
-
-/// a post from getAuthorFeed
-pub const AuthorPost = struct {
-    uri: []const u8,
-    cid: []const u8,
-    text: []const u8,
-    embed_uri: ?[]const u8,
-};
-
-/// struct for extracting post data from feed items
-const FeedPost = struct {
-    uri: []const u8,
-    cid: []const u8,
-    record: struct {
-        text: []const u8 = "",
-        embed: ?struct {
-            external: ?struct { uri: []const u8 } = null,
-        } = null,
-    },
-};
-
-/// fetch recent posts from an author's feed.
-pub fn getAuthorFeed(allocator: Allocator, actor_did: []const u8, limit: usize) ![]AuthorPost {
-    var client = zat.XrpcClient.init(allocator, BSKY_PUBLIC_API);
-    defer client.deinit();
-
-    var posts: std.ArrayList(AuthorPost) = .empty;
-    errdefer {
-        for (posts.items) |p| {
-            allocator.free(p.uri);
-            allocator.free(p.cid);
-            allocator.free(p.text);
-            if (p.embed_uri) |eu| allocator.free(eu);
-        }
-        posts.deinit(allocator);
-    }
-
-    const nsid = zat.Nsid.parse("app.bsky.feed.getAuthorFeed") orelse return error.InvalidNsid;
-
-    var params = std.StringHashMap([]const u8).init(allocator);
-    defer params.deinit();
-
-    var limit_buf: [8]u8 = undefined;
-    const limit_str = std.fmt.bufPrint(&limit_buf, "{d}", .{@min(limit, 100)}) catch return error.FormatError;
-
-    try params.put("actor", actor_did);
-    try params.put("limit", limit_str);
-    try params.put("filter", "posts_no_replies");
-
-    var response = client.query(nsid, params) catch |err| {
-        log.err("getAuthorFeed API error for {s}: {}", .{ actor_did, err });
-        return error.ApiFailed;
-    };
-    defer response.deinit();
-
-    if (!response.ok()) {
-        return error.ApiFailed;
-    }
-
-    var parsed = response.json() catch return error.InvalidJson;
-    defer parsed.deinit();
-
-    const feed = zat.json.getArray(parsed.value, "feed") orelse return try posts.toOwnedSlice(allocator);
-
-    for (feed) |item| {
-        const post = zat.json.extractAt(FeedPost, allocator, item, .{"post"}) catch continue;
-
-        const embed_uri = if (post.record.embed) |e| if (e.external) |ext| ext.uri else null else null;
-
-        try posts.append(allocator, .{
-            .uri = try allocator.dupe(u8, post.uri),
-            .cid = try allocator.dupe(u8, post.cid),
-            .text = try allocator.dupe(u8, post.record.text),
-            .embed_uri = if (embed_uri) |eu| try allocator.dupe(u8, eu) else null,
-        });
-    }
-
-    return try posts.toOwnedSlice(allocator);
-}

src/stream/db.zig

···
-const std = @import("std");
-const posix = std.posix;
-const zqlite = @import("zqlite");
-const zat = @import("zat");
-const config = @import("../feed/config.zig");
-const atproto = @import("atproto.zig");
-
-var db: ?zqlite.Conn = null;
-var mutex: std.Thread.Mutex = .{};
-
-// static buffer for db path (env vars return slices, zqlite needs sentinel)
-var db_path_buf: [256]u8 = undefined;
-
-pub fn init() !void {
-    const db_path_env = posix.getenv("DATABASE_PATH") orelse "/data/feed.db";
-
-    // copy to null-terminated buffer
-    @memcpy(db_path_buf[0..db_path_env.len], db_path_env);
-    db_path_buf[db_path_env.len] = 0;
-    const db_path: [*:0]const u8 = db_path_buf[0..db_path_env.len :0];
-
-    std.debug.print("opening database: {s}\n", .{db_path_env});
-
-    const flags = zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite;
-    db = zqlite.open(db_path, flags) catch |err| {
-        std.debug.print("failed to open database: {}\n", .{err});
-        return err;
-    };
-
-    // enable WAL mode for better concurrency
-    _ = db.?.exec("PRAGMA journal_mode=WAL", .{}) catch {};
-    _ = db.?.exec("PRAGMA busy_timeout=5000", .{}) catch {};
-
-    // create tables
-    db.?.exec(
-        \\CREATE TABLE IF NOT EXISTS posts (
-        \\  uri TEXT PRIMARY KEY,
-        \\  cid TEXT NOT NULL,
-        \\  indexed_at INTEGER NOT NULL
-        \\)
-    , .{}) catch |err| {
-        std.debug.print("failed to create posts table: {}\n", .{err});
-        return err;
-    };
-
-    db.?.exec(
-        "CREATE INDEX IF NOT EXISTS idx_posts_indexed_at ON posts(indexed_at DESC)",
-        .{},
-    ) catch {};
-
-    // follows cache table
-    db.?.exec(
-        \\CREATE TABLE IF NOT EXISTS follows_cache (
-        \\  requester_did TEXT NOT NULL,
-        \\  follows_did TEXT NOT NULL,
-        \\  cached_at INTEGER NOT NULL,
-        \\  PRIMARY KEY (requester_did, follows_did)
-        \\)
-    , .{}) catch |err| {
-        std.debug.print("failed to create follows_cache table: {}\n", .{err});
-        return err;
-    };
-
-    db.?.exec(
-        "CREATE INDEX IF NOT EXISTS idx_follows_cache_requester ON follows_cache(requester_did, cached_at)",
-        .{},
-    ) catch {};
-
-    // backfilled accounts table - track which accounts we've fetched historical posts from
-    db.?.exec(
-        \\CREATE TABLE IF NOT EXISTS backfilled_accounts (
-        \\  did TEXT PRIMARY KEY,
-        \\  backfilled_at INTEGER NOT NULL
-        \\)
-    , .{}) catch |err| {
-        std.debug.print("failed to create backfilled_accounts table: {}\n", .{err});
-        return err;
-    };
-
-    std.debug.print("database initialized\n", .{});
-}
-
-pub fn addPost(uri: []const u8, cid: []const u8) !void {
-    try addPostWithTimestamp(uri, cid, null);
-}
-
-/// add a post with explicit timestamp (for backfill) or current time
-pub fn addPostWithTimestamp(uri: []const u8, cid: []const u8, timestamp: ?i64) !void {
-    mutex.lock();
-    defer mutex.unlock();
-
-    const conn = db orelse return error.NotInitialized;
-
-    // use provided timestamp, or try to parse from URI, or fall back to now
-    const ts = timestamp orelse parseTimestampFromUri(uri) orelse std.time.milliTimestamp();
-
-    // extract author DID from URI
-    const author_did = extractDidFromUri(uri);
-
-    conn.exec(
-        "INSERT OR IGNORE INTO posts (uri, cid, indexed_at, author_did) VALUES (?, ?, ?, ?)",
-        .{ uri, cid, ts, author_did },
-    ) catch |err| {
-        std.debug.print("failed to insert post: {}\n", .{err});
-        return err;
-    };
-}
-
-/// parse TID to get timestamp in milliseconds
-pub fn parseTidTimestamp(rkey: []const u8) i64 {
-    const tid = zat.Tid.parse(rkey) orelse return 0;
-    return @intCast(tid.timestamp() / 1000); // microseconds -> milliseconds
-}
-
-/// parse TID from AT URI to get post timestamp
-fn parseTimestampFromUri(uri: []const u8) ?i64 {
-    const at_uri = zat.AtUri.parse(uri) orelse return null;
-    const rkey = at_uri.rkey() orelse return null;
-    const ts = parseTidTimestamp(rkey);
-    return if (ts > 0) ts else null;
-}
-
-pub fn getPosts(
-    alloc: std.mem.Allocator,
-    feed_type: config.FeedType,
-    requester_did: ?[]const u8,
-    cursor: ?[]const u8,
-    limit: usize,
-) ![]const u8 {
-    // for following feed, fetch follows first (before mutex)
-    var follows_list: ?[]const u8 = null;
-    defer if (follows_list) |f| alloc.free(f);
-
-    if (feed_type == .following) {
-        if (requester_did) |did| {
-            follows_list = getFollowsList(alloc, did) catch null;
-        } else {
-            return try alloc.dupe(u8, "{\"cursor\":\"eof\",\"feed\":[]}");
-        }
-    }
-
-    mutex.lock();
-    defer mutex.unlock();
-
-    const conn = db orelse return error.NotInitialized;
-
-    var buf: std.ArrayList(u8) = .{};
-    defer buf.deinit(alloc);
-    const w = buf.writer(alloc);
-
-    const CURSOR_EOF = "eof";
-
-    if (cursor) |c| {
-        if (std.mem.eql(u8, c, CURSOR_EOF)) {
-            try w.writeAll("{\"cursor\":\"eof\",\"feed\":[]}");
-            return try alloc.dupe(u8, buf.items);
-        }
-    }
-
-    // parse cursor (format: timestamp::cid)
-    var cursor_ts: i64 = std.math.maxInt(i64);
-    var cursor_cid: []const u8 = "";
-    if (cursor) |c| {
-        if (std.mem.indexOf(u8, c, "::")) |sep| {
-            cursor_ts = std.fmt.parseInt(i64, c[0..sep], 10) catch std.math.maxInt(i64);
-            cursor_cid = c[sep + 2 ..];
-        }
-    }
-
-    // build and execute query
-    var posts_json: std.ArrayList(u8) = .{};
-    defer posts_json.deinit(alloc);
-    const pw = posts_json.writer(alloc);
-
-    var cursor_buf: [128]u8 = undefined;
-    var cursor_len: usize = 0;
-    var count: usize = 0;
-
-    if (feed_type == .following) {
-        // for following feed: query by author_did using IN clause
-        if (follows_list) |follows| {
-            if (follows.len == 0) {
-                try w.writeAll("{\"cursor\":\"eof\",\"feed\":[]}");
-                return try alloc.dupe(u8, buf.items);
-            }
-
-            // build query with IN clause
-            var query_buf: std.ArrayList(u8) = .{};
-            defer query_buf.deinit(alloc);
-            const qw = query_buf.writer(alloc);
-
-            try qw.writeAll("SELECT uri, cid, indexed_at FROM posts WHERE author_did IN (");
-
-            var first = true;
-            var it = std.mem.splitScalar(u8, follows, '\n');
-            while (it.next()) |did| {
-                if (did.len == 0) continue;
-                if (!first) try qw.writeAll(",");
-                try qw.print("'{s}'", .{did});
-                first = false;
-            }
-
-            try qw.print(") AND (indexed_at < {d} OR (indexed_at = {d} AND cid < '{s}')) ORDER BY indexed_at DESC, cid DESC LIMIT {d}", .{ cursor_ts, cursor_ts, cursor_cid, limit });
-
-            // need null-terminated string for zqlite
-            try query_buf.append(alloc, 0);
-            const query_z: [:0]const u8 = query_buf.items[0 .. query_buf.items.len - 1 :0];
-
-            var rows = conn.rows(query_z, .{}) catch |err| {
-                std.debug.print("following query failed: {}\n", .{err});
-                return error.QueryFailed;
-            };
-            defer rows.deinit();
-
-            while (rows.next()) |row| {
-                const uri = row.text(0);
-                const cid = row.text(1);
-                const ts = row.int(2);
-
-                if (count > 0) try pw.writeAll(",");
-                try pw.print("{{\"post\":\"{s}\"}}", .{uri});
-                cursor_len = (std.fmt.bufPrint(&cursor_buf, "{d}::{s}", .{ ts, cid }) catch &cursor_buf).len;
-                count += 1;
-            }
-
-            std.debug.print("following feed: returning {d} posts\n", .{count});
-        }
-    } else {
-        // for all feed: simple query
-        var rows = conn.rows(
-            \\SELECT uri, cid, indexed_at FROM posts
-            \\WHERE indexed_at < ? OR (indexed_at = ? AND cid < ?)
-            \\ORDER BY indexed_at DESC, cid DESC
-            \\LIMIT ?
-        , .{ cursor_ts, cursor_ts, cursor_cid, limit }) catch |err| {
-            std.debug.print("query failed: {}\n", .{err});
-            return error.QueryFailed;
-        };
-        defer rows.deinit();
-
-        while (rows.next()) |row| {
-            const uri = row.text(0);
-            const cid = row.text(1);
-            const ts = row.int(2);
-
-            if (count > 0) try pw.writeAll(",");
-            try pw.print("{{\"post\":\"{s}\"}}", .{uri});
-            cursor_len = (std.fmt.bufPrint(&cursor_buf, "{d}::{s}", .{ ts, cid }) catch &cursor_buf).len;
-            count += 1;
-        }
-    }
-
-    // build final response
-    try w.writeAll("{\"cursor\":");
-    if (count > 0) {
-        try w.print("\"{s}\"", .{cursor_buf[0..cursor_len]});
-    } else {
-        try w.writeAll("\"eof\"");
-    }
-
-    try w.writeAll(",\"feed\":[");
-    try w.writeAll(posts_json.items);
-    try w.writeAll("]}");
-
-    return try alloc.dupe(u8, buf.items);
-}
-
-/// extract DID from post URI (at://did:plc:xxx/app.bsky.feed.post/yyy)
-fn extractDidFromUri(uri: []const u8) ?[]const u8 {
-    const at_uri = zat.AtUri.parse(uri) orelse return null;
-    return at_uri.authority();
-}
-
-/// get follows as newline-separated string (caller owns returned memory)
-fn getFollowsList(alloc: std.mem.Allocator, requester_did: []const u8) ![]const u8 {
-    std.debug.print("fetching follows for {s}...\n", .{requester_did});
-    const follows = atproto.getFollows(alloc, requester_did) catch |err| {
-        std.debug.print("failed to fetch follows: {}\n", .{err});
-        return error.FetchFailed;
-    };
-
-    // count follows for logging
-    var count: usize = 0;
-    var it = std.mem.splitScalar(u8, follows, '\n');
-    while (it.next()) |did| {
-        if (did.len > 0) count += 1;
-    }
-    std.debug.print("fetched {d} follows for {s}\n", .{ count, requester_did });
-
-    // trigger backfill in background
-    var dids_to_backfill: std.ArrayList([]const u8) = .empty;
-    defer dids_to_backfill.deinit(alloc);
-    var it2 = std.mem.splitScalar(u8, follows, '\n');
-    while (it2.next()) |did| {
-        if (did.len > 0) {
-            dids_to_backfill.append(alloc, did) catch continue;
-        }
-    }
-    if (dids_to_backfill.items.len > 0) {
-        triggerBackfill(alloc, dids_to_backfill.items);
-    }
-
-    return follows;
-}
-
-pub fn getPostCount() usize {
-    mutex.lock();
-    defer mutex.unlock();
-
-    const conn = db orelse return 0;
-    var rows = conn.rows("SELECT COUNT(*) FROM posts", .{}) catch return 0;
-    defer rows.deinit();
-
-    if (rows.next()) |row| {
-        return @intCast(row.int(0));
-    }
-    return 0;
-}
-
-// -----------------------------------------------------------------------------
-// lazy backfill
-// -----------------------------------------------------------------------------
-
-const filter = @import("../feed/filter.zig");
-
-/// check if an account has been backfilled
-fn isBackfilled(did: []const u8) bool {
-    mutex.lock();
-    defer mutex.unlock();
-
-    const conn = db orelse return false;
-    var rows = conn.rows(
-        "SELECT 1 FROM backfilled_accounts WHERE did = ?",
-        .{did},
-    ) catch return false;
-    defer rows.deinit();
-
-    return rows.next() != null;
-}
-
-/// mark an account as backfilled
-fn markBackfilled(did: []const u8) void {
-    mutex.lock();
-    defer mutex.unlock();
-
-    const conn = db orelse return;
-    const now = std.time.milliTimestamp();
-    conn.exec(
-        "INSERT OR REPLACE INTO backfilled_accounts (did, backfilled_at) VALUES (?, ?)",
-        .{ did, now },
-    ) catch {};
-}
-
-/// backfill posts from a single account
-fn backfillAccount(alloc: std.mem.Allocator, did: []const u8) void {
-    if (isBackfilled(did)) return;
-
-    std.debug.print("backfilling {s}...\n", .{did});
-
-    const posts = atproto.getAuthorFeed(alloc, did, 50) catch |err| {
-        std.debug.print("backfill failed for {s}: {}\n", .{ did, err });
-        return;
-    };
-    defer {
-        for (posts) |p| {
-            alloc.free(p.uri);
-            alloc.free(p.cid);
-            alloc.free(p.text);
-            if (p.embed_uri) |eu| alloc.free(eu);
-        }
-        alloc.free(posts);
-    }
-
-    var added: usize = 0;
-    for (posts) |post| {
-        const text_match = filter.textContainsMusicLink(post.text);
-        const embed_match = if (post.embed_uri) |eu| filter.textContainsMusicLink(eu) else false;
-        if (text_match or embed_match) {
-            addPost(post.uri, post.cid) catch continue;
-            added += 1;
-        }
-    }
-
-    markBackfilled(did);
-    std.debug.print("backfilled {s}: {d} music posts from {d} total\n", .{ did, added, posts.len });
-}
-
-/// trigger backfill for a list of DIDs (called after fetching follows)
-pub fn triggerBackfill(alloc: std.mem.Allocator, dids: []const []const u8) void {
-    var backfilled: usize = 0;
-    const max_per_request = 10; // limit how many we backfill per request
-
-    for (dids) |did| {
-        if (backfilled >= max_per_request) break;
-        if (!isBackfilled(did)) {
-            backfillAccount(alloc, did);
-            backfilled += 1;
-        }
-    }
-}

src/stream/jetstream.zig src/jetstream.zig

···
 const posix = std.posix;
 const Allocator = mem.Allocator;
 const websocket = @import("websocket");
-const filter = @import("../feed/filter.zig");
-const stats = @import("../server/stats.zig");
-const db = @import("db.zig");
+const filter = @import("feed/filter.zig");
+const stats = @import("server/stats.zig");
+const posts = @import("store/posts.zig");
 
 // turbostream: hydrated jetstream from graze.social
 const TURBOSTREAM_HOST = "api.graze.social";
···
         if (rkey_val != .string) return error.NotAPost;
         const rkey = rkey_val.string;
 
-        const post_time_ms = db.parseTidTimestamp(rkey);
+        const post_time_ms = posts.parseTidTimestamp(rkey);
 
         // get time_us from message
         const time_us: i64 = if (msg.get("time_us")) |t| (if (t == .integer) t.integer else 0) else 0;
···
         if (at_uri != .string) return error.NotAPost;
 
         // add to feed
-        db.addPost(at_uri.string, cid_val.string) catch |err| {
+        posts.add(at_uri.string, cid_val.string) catch |err| {
             std.debug.print("failed to add post: {}\n", .{err});
             return;
         };