From 082602677d613f2762b27ef22c82d1f741623064 Mon Sep 17 00:00:00 2001
From: Florian <45694132+flo-bit@users.noreply.github.com>
Date: Mon, 23 Mar 2026 16:59:35 +0100
Subject: [PATCH] split records into tables

---
 .../lexicon/calendar/event/listRecords.json   |   4 +
 src/config.ts                                 |   1 +
 src/core/db/records.ts                        | 139 +++++++++---------
 src/core/db/schema.ts                         |  76 ++++++----
 src/core/router/admin.ts                      |  22 ++-
 src/core/router/collection.ts                 |  11 +-
 src/core/router/feed.ts                       |  17 ++-
 src/core/router/hydrate.ts                    |  33 ++---
 src/core/router/notify.ts                     |   6 +-
 src/core/router/profiles.ts                   |   8 +-
 src/core/types.ts                             |   4 +
 src/generate.ts                               |   3 +-
 tests/hydrate.test.ts                         |  14 +-
 tests/records.test.ts                         |  11 +-
 tests/schema.test.ts                          |   3 +-
 15 files changed, 197 insertions(+), 155 deletions(-)

diff --git a/lexicons-generated/community/lexicon/calendar/event/listRecords.json b/lexicons-generated/community/lexicon/calendar/event/listRecords.json
index 5e92721..9f38cab 100644
--- a/lexicons-generated/community/lexicon/calendar/event/listRecords.json
+++ b/lexicons-generated/community/lexicon/calendar/event/listRecords.json
@@ -26,6 +26,10 @@
             "type": "boolean",
             "description": "Include profile + identity info keyed by DID"
           },
+          "search": {
+            "type": "string",
+            "description": "Full-text search across: name, description"
+          },
           "mode": {
             "type": "string",
             "description": "Filter by mode"
diff --git a/src/config.ts b/src/config.ts
index de061dd..3a0f5fb 100644
--- a/src/config.ts
+++ b/src/config.ts
@@ -4,6 +4,7 @@ export const config: ContrailConfig = {
   namespace: "rsvp.atmo",
   collections: {
     "community.lexicon.calendar.event": {
+      searchable: ["name", "description"],
       relations: {
         rsvps: {
           collection: "community.lexicon.calendar.rsvp",
diff --git a/src/core/db/records.ts b/src/core/db/records.ts
index f87b52a..1a03ad5 100644
--- a/src/core/db/records.ts
+++ b/src/core/db/records.ts
@@ -7,7 +7,7 @@ import type {
   RecordRow,
   RecordSource,
 } from "../types";
-import { getNestedValue, getRelationField, countColumnName, getFeedFollowCollections } from "../types";
+import { getNestedValue, getRelationField, countColumnName, getFeedFollowCollections, recordsTableName } from "../types";
 import { resolvedRelationsMap } from "../queryable.generated";
 import { getSearchableFields, ftsTableName, buildFtsContent } from "../search";
 
@@ -59,7 +59,8 @@ function buildCountStatements(
 
     const field = getRelationField(rel);
     const matchColumn = rel.match === "did" ? "did" : "uri";
-    const childCollection = rel.collection;
+    const childTable = recordsTableName(rel.collection);
+    const parentTable = recordsTableName(parentCollection);
 
     // Collect target URIs/DIDs that need recounting (current + old if changed)
     const targets: string[] = [];
@@ -83,20 +84,20 @@ function buildCountStatements(
       // Total count
       const totalCol = countColumnName(rel.collection);
       setClauses.push(
-        `${totalCol} = (SELECT COUNT(*) FROM records WHERE collection = ? AND json_extract(record, '$.${field}') = ?)`
+        `${totalCol} = (SELECT COUNT(*) FROM ${childTable} WHERE json_extract(record, '$.${field}') = ?)`
       );
-      setBindings.push(childCollection, targetValue);
+      setBindings.push(targetValue);
 
       // Grouped counts
       if (rel.groupBy) {
         const mapping = (resolvedRelationsMap as Record<string, any>)[parentCollection]?.[relationName];
         if (mapping?.groups) {
-          for (const [groupValue, fullToken] of Object.entries(mapping.groups as Record<string, string>)) {
+          for (const [, fullToken] of Object.entries(mapping.groups as Record<string, string>)) {
             const groupCol = countColumnName(fullToken);
             setClauses.push(
-              `${groupCol} = (SELECT COUNT(*) FROM records WHERE collection = ? AND json_extract(record, '$.${field}') = ? AND json_extract(record, '$.${rel.groupBy}') = ?)`
+              `${groupCol} = (SELECT COUNT(*) FROM ${childTable} WHERE json_extract(record, '$.${field}') = ? AND json_extract(record, '$.${rel.groupBy}') = ?)`
             );
-            setBindings.push(childCollection, targetValue, groupValue);
+            setBindings.push(targetValue, fullToken);
           }
         }
       }
@@ -105,7 +106,7 @@ function buildCountStatements(
         statements.push(
           db
             .prepare(
-              `UPDATE records SET ${setClauses.join(", ")} WHERE ${matchColumn} = ?`
+              `UPDATE ${parentTable} SET ${setClauses.join(", ")} WHERE ${matchColumn} = ?`
             )
             .bind(...setBindings, targetValue)
         );
@@ -164,21 +165,20 @@ function buildFeedStatements(
   const stmts: Statement[] = [];
 
   for (const [, feedConfig] of Object.entries(config.feeds)) {
+    const followTable = recordsTableName(feedConfig.follow);
+
     // Target collection: fan out to followers
     if (feedConfig.targets.includes(event.collection)) {
       if (event.operation === "create" || event.operation === "update") {
-        // Insert feed items for all followers of the event creator.
-        // Follow records have: did = follower, record.subject = followed person.
         stmts.push(
           db
             .prepare(
               `INSERT OR IGNORE INTO feed_items (actor, uri, collection, time_us)
                SELECT r.did, ?, ?, ?
-               FROM records r
-               WHERE r.collection = ?
-                 AND json_extract(r.record, '$.subject') = ?`
+               FROM ${followTable} r
+               WHERE json_extract(r.record, '$.subject') = ?`
             )
-            .bind(event.uri, event.collection, event.time_us, feedConfig.follow, event.did)
+            .bind(event.uri, event.collection, event.time_us, event.did)
         );
       } else if (event.operation === "delete") {
         stmts.push(
@@ -193,15 +193,15 @@ function buildFeedStatements(
         const record = event.record ? JSON.parse(event.record) : null;
         const subject = record?.subject;
         if (subject) {
-          // New follow: backfill recent items from the followed user
           for (const targetCol of feedConfig.targets) {
+            const targetTable = recordsTableName(targetCol);
             stmts.push(
               db
                 .prepare(
                   `INSERT OR IGNORE INTO feed_items (actor, uri, collection, time_us)
-                   SELECT ?, r.uri, r.collection, r.time_us
-                   FROM records r
-                   WHERE r.collection = ? AND r.did = ?
+                   SELECT ?, r.uri, ?, r.time_us
+                   FROM ${targetTable} r
+                   WHERE r.did = ?
                    ORDER BY r.time_us DESC
                    LIMIT 100`
                 )
@@ -210,23 +210,23 @@ function buildFeedStatements(
           }
         }
       } else if (event.operation === "delete") {
-        // Unfollow: remove feed items from the unfollowed user.
-        // The record field is null for deletes, so we look up the existing record.
         const existingRecord = existingRecords.get(event.uri);
         if (existingRecord) {
           const parsed = JSON.parse(existingRecord);
           const subject = parsed?.subject;
           if (subject) {
-            const targetPlaceholders = feedConfig.targets.map(() => "?").join(",");
-            stmts.push(
-              db
-                .prepare(
-                  `DELETE FROM feed_items WHERE actor = ? AND uri IN (
-                     SELECT uri FROM records WHERE did = ? AND collection IN (${targetPlaceholders})
-                   )`
-                )
-                .bind(event.did, subject, ...feedConfig.targets)
-            );
+            for (const targetCol of feedConfig.targets) {
+              const targetTable = recordsTableName(targetCol);
+              stmts.push(
+                db
+                  .prepare(
+                    `DELETE FROM feed_items WHERE actor = ? AND uri IN (
+                       SELECT uri FROM ${targetTable} WHERE did = ?
+                     )`
+                  )
+                  .bind(event.did, subject)
+              );
+            }
           }
         }
       }
@@ -287,10 +287,7 @@ export async function applyEvents(
 ): Promise<void> {
   if (events.length === 0) return;
 
-  // Look up existing records so we can skip duplicate count updates on replayed events.
-  // A create/update with the same CID is a replay; a delete for a missing URI is a replay.
-  // Can be skipped during backfill where records are known to be fresh inserts.
-  // Also fetches record content for follow-delete events (needed for unfollow feed cleanup).
+  // Look up existing records for replay detection and count recounts.
   const existingCids = new Map<string, string | null>();
   const existingRecords = new Map<string, string | null>();
   const followCollections = config ? getFeedFollowCollections(config) : [];
@@ -300,40 +297,48 @@ export async function applyEvents(
   const needRecordContent = followCollections.length > 0 || hasCountingRelations;
 
   if (config && !options?.skipReplayDetection) {
-    const uris = events.map((e) => e.uri);
+    // Group events by collection to query the correct tables
+    const byCollection = new Map<string, string[]>();
+    for (const e of events) {
+      const uris = byCollection.get(e.collection) ?? [];
+      uris.push(e.uri);
+      byCollection.set(e.collection, uris);
+    }
+
     const selectCols = needRecordContent ? "uri, cid, record" : "uri, cid";
-    for (let i = 0; i < uris.length; i += 50) {
-      const chunk = uris.slice(i, i + 50);
-      const placeholders = chunk.map(() => "?").join(",");
-      const rows = await db
-        .prepare(`SELECT ${selectCols} FROM records WHERE uri IN (${placeholders})`)
-        .bind(...chunk)
-        .all<{ uri: string; cid: string | null; record?: string | null }>();
-      for (const row of rows.results ?? []) {
-        existingCids.set(row.uri, row.cid);
-        if (needRecordContent && row.record) {
-          existingRecords.set(row.uri, row.record);
+    for (const [collection, uris] of byCollection) {
+      const table = recordsTableName(collection);
+      for (let i = 0; i < uris.length; i += 50) {
+        const chunk = uris.slice(i, i + 50);
+        const placeholders = chunk.map(() => "?").join(",");
+        const rows = await db
+          .prepare(`SELECT ${selectCols} FROM ${table} WHERE uri IN (${placeholders})`)
+          .bind(...chunk)
+          .all<{ uri: string; cid: string | null; record?: string | null }>();
+        for (const row of rows.results ?? []) {
+          existingCids.set(row.uri, row.cid);
+          if (needRecordContent && row.record) {
+            existingRecords.set(row.uri, row.record);
+          }
         }
       }
     }
   }
 
-  const upsertStmt = db.prepare(
-    "INSERT INTO records (uri, did, collection, rkey, cid, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(uri) DO UPDATE SET cid = excluded.cid, record = excluded.record, time_us = excluded.time_us, indexed_at = excluded.indexed_at"
-  );
-  const deleteStmt = db.prepare("DELETE FROM records WHERE uri = ?");
-
   const batch: Statement[] = [];
 
   for (const e of events) {
+    const table = recordsTableName(e.collection);
+
     if (e.operation === "delete") {
-      batch.push(deleteStmt.bind(e.uri));
+      batch.push(db.prepare(`DELETE FROM ${table} WHERE uri = ?`).bind(e.uri));
     } else {
       batch.push(
-        upsertStmt.bind(
+        db.prepare(
+          `INSERT INTO ${table} (uri, did, rkey, cid, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(uri) DO UPDATE SET cid = excluded.cid, record = excluded.record, time_us = excluded.time_us, indexed_at = excluded.indexed_at`
+        ).bind(
           e.uri,
           e.did,
-          e.collection,
           e.rkey,
           e.cid,
           e.record,
@@ -345,7 +350,6 @@ export async function applyEvents(
 
     if (config) {
       // Recount is idempotent — always run it for create/update/delete.
-      // Pass existing record so deletes and updates that change the target can recount the old parent.
       const existingRecordJson = existingRecords.get(e.uri) ?? null;
       batch.push(...buildCountStatements(db, e, config, existingRecordJson));
 
@@ -390,8 +394,8 @@ function getCountColumns(config: ContrailConfig, collection: string): { type: st
 // --- Query ---
 
 export interface SortOption {
-  recordField?: string;  // json path, e.g. "startsAt" — sorts by json_extract
-  countType?: string;    // count type, e.g. collection NSID — sorts by aggregated count
+  recordField?: string;
+  countType?: string;
   direction: "asc" | "desc";
 }
 
@@ -426,9 +430,10 @@ export async function queryRecords(
     source,
   } = options;
 
+  const table = recordsTableName(collection);
   const limit = Math.min(Math.max(1, rawLimit ?? 50), 200);
-  const conditions: string[] = ["r.collection = ?"];
-  const bindings: (string | number)[] = [collection];
+  const conditions: string[] = [];
+  const bindings: (string | number)[] = [];
 
   if (source?.conditions) conditions.push(...source.conditions);
   if (source?.params) bindings.push(...source.params);
@@ -443,7 +448,7 @@ export async function queryRecords(
   // Cursor = AT URI of last seen record. Look it up to get keyset values.
   if (cursor) {
     const cursorRow = await db
-      .prepare("SELECT * FROM records WHERE uri = ?")
+      .prepare(`SELECT * FROM ${table} WHERE uri = ?`)
       .bind(cursor)
       .first<any>();
 
@@ -484,7 +489,6 @@ export async function queryRecords(
     }
   }
 
-  // Count filters — direct column comparison instead of HAVING
   for (const [type, minCount] of Object.entries(countFilters)) {
     const col = countColumnName(type);
     conditions.push(`r.${col} >= ?`);
@@ -497,20 +501,19 @@ export async function queryRecords(
     const colConfig2 = config.collections[collection];
     const fields = colConfig2 ? getSearchableFields(collection, colConfig2) : null;
     if (fields && fields.length > 0) {
-      const table = ftsTableName(collection);
-      ftsJoin = `JOIN ${table} fts ON fts.uri = r.uri`;
+      const ftsTable = ftsTableName(collection);
+      ftsJoin = `JOIN ${ftsTable} fts ON fts.uri = r.uri`;
       conditions.push("fts.content MATCH ?");
       bindings.push(search);
     }
   }
 
-  const where = conditions.join(" AND ");
+  const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
 
-  // Select count columns directly
   const countSelect = countCols.length > 0
     ? ", " + countCols.map(({ column }) => `r.${column}`).join(", ")
     : "";
-  const select = `r.uri, r.did, r.collection, r.rkey, r.cid, r.record, r.time_us, r.indexed_at${countSelect}`;
+  const select = `r.uri, r.did, r.rkey, r.cid, r.record, r.time_us, r.indexed_at${countSelect}`;
 
   const join = [source?.joins, ftsJoin].filter(Boolean).join(" ");
 
@@ -530,7 +533,7 @@ export async function queryRecords(
 
   bindings.push(limit);
 
-  const query = `SELECT ${select} FROM records r ${join} WHERE ${where} ORDER BY ${orderBy} LIMIT ?`;
+  const query = `SELECT ${select} FROM ${table} r ${join} ${where} ORDER BY ${orderBy} LIMIT ?`;
 
   const result = await db
     .prepare(query)
@@ -541,7 +544,7 @@ export async function queryRecords(
     const rec: RecordRow & { counts?: Record<string, number> } = {
       uri: row.uri,
       did: row.did,
-      collection: row.collection,
+      collection,
       rkey: row.rkey,
       cid: row.cid,
       record: row.record,
diff --git a/src/core/db/schema.ts b/src/core/db/schema.ts
index cf56768..63b3044 100644
--- a/src/core/db/schema.ts
+++ b/src/core/db/schema.ts
@@ -1,21 +1,9 @@
 import type { ContrailConfig, Database } from "../types";
-import { getRelationField, countColumnName } from "../types";
+import { getRelationField, countColumnName, recordsTableName } from "../types";
 import { resolvedQueryable, resolvedRelationsMap } from "../queryable.generated";
 import { getSearchableFields, ftsTableName } from "../search";
 
 const BASE_SCHEMA = `
-CREATE TABLE IF NOT EXISTS records (
-  uri TEXT PRIMARY KEY,
-  did TEXT NOT NULL,
-  collection TEXT NOT NULL,
-  rkey TEXT NOT NULL,
-  cid TEXT,
-  record TEXT,
-  time_us INTEGER NOT NULL,
-  indexed_at INTEGER NOT NULL
-);
-CREATE INDEX IF NOT EXISTS idx_records_collection_time ON records(collection, time_us DESC);
-CREATE INDEX IF NOT EXISTS idx_records_collection_did ON records(collection, did);
 CREATE TABLE IF NOT EXISTS backfills (
   did TEXT NOT NULL,
   collection TEXT NOT NULL,
@@ -49,22 +37,46 @@ function sanitizeName(name: string): string {
   return name.replace(/[^a-zA-Z0-9]/g, "_");
 }
 
+function buildCollectionTables(config: ContrailConfig): string[] {
+  const stmts: string[] = [];
+  for (const collection of Object.keys(config.collections)) {
+    const table = recordsTableName(collection);
+    stmts.push(
+      `CREATE TABLE IF NOT EXISTS ${table} (
+        uri TEXT PRIMARY KEY,
+        did TEXT NOT NULL,
+        rkey TEXT NOT NULL,
+        cid TEXT,
+        record TEXT,
+        time_us INTEGER NOT NULL,
+        indexed_at INTEGER NOT NULL
+      )`
+    );
+    stmts.push(`CREATE INDEX IF NOT EXISTS idx_${sanitizeName(collection)}_did ON ${table}(did)`);
+    stmts.push(`CREATE INDEX IF NOT EXISTS idx_${sanitizeName(collection)}_time ON ${table}(time_us DESC)`);
+  }
+  return stmts;
+}
+
 function buildDynamicIndexes(config: ContrailConfig): string[] {
   const indexes: string[] = [];
   for (const [collection, colConfig] of Object.entries(config.collections)) {
+    const table = recordsTableName(collection);
     const queryable = resolvedQueryable[collection] ?? colConfig.queryable ?? {};
     for (const field of Object.keys(queryable)) {
       const idxName = `idx_${sanitizeName(collection)}_${sanitizeName(field)}`;
       indexes.push(
-        `CREATE INDEX IF NOT EXISTS ${idxName} ON records(collection, json_extract(record, '$.${field}'))`
+        `CREATE INDEX IF NOT EXISTS ${idxName} ON ${table}(json_extract(record, '$.${field}'))`
       );
     }
 
+    // Relation field indexes go on the CHILD collection's table
     for (const [, rel] of Object.entries(colConfig.relations ?? {})) {
       const on = getRelationField(rel);
+      const childTable = recordsTableName(rel.collection);
       const idxName = `idx_${sanitizeName(rel.collection)}_${sanitizeName(on)}`;
       indexes.push(
-        `CREATE INDEX IF NOT EXISTS ${idxName} ON records(collection, json_extract(record, '$.${on}'))`
+        `CREATE INDEX IF NOT EXISTS ${idxName} ON ${childTable}(json_extract(record, '$.${on}'))`
       );
     }
   }
@@ -73,23 +85,27 @@ function buildDynamicIndexes(config: ContrailConfig): string[] {
 
 function buildCountColumns(config: ContrailConfig): string[] {
   const stmts: string[] = [];
-  const addedColumns = new Set<string>();
+  const addedColumns = new Map<string, Set<string>>(); // table → columns
 
   for (const [collection, colConfig] of Object.entries(config.collections)) {
+    const table = recordsTableName(collection);
     const relMap = resolvedRelationsMap[collection] ?? {};
+
+    if (!addedColumns.has(table)) addedColumns.set(table, new Set());
+    const tableColumns = addedColumns.get(table)!;
+
     for (const [relName, rel] of Object.entries(colConfig.relations ?? {})) {
       if (rel.count === false) continue;
-      // Total count column
+      // Total count column — on the PARENT collection's table
       const totalCol = countColumnName(rel.collection);
-      if (!addedColumns.has(totalCol)) {
-        addedColumns.add(totalCol);
+      if (!tableColumns.has(totalCol)) {
+        tableColumns.add(totalCol);
         stmts.push(
-          `ALTER TABLE records ADD COLUMN ${totalCol} INTEGER NOT NULL DEFAULT 0`
+          `ALTER TABLE ${table} ADD COLUMN ${totalCol} INTEGER NOT NULL DEFAULT 0`
         );
       }
-      // Index for sorting by this count within the parent collection
       stmts.push(
-        `CREATE INDEX IF NOT EXISTS idx_${sanitizeName(collection)}_${totalCol} ON records(collection, ${totalCol} DESC, time_us DESC)`
+        `CREATE INDEX IF NOT EXISTS idx_${sanitizeName(collection)}_${totalCol} ON ${table}(${totalCol} DESC, time_us DESC)`
       );
 
       // Grouped count columns
@@ -97,14 +113,14 @@ function buildCountColumns(config: ContrailConfig): string[] {
       if (mapping) {
         for (const [, fullToken] of Object.entries(mapping.groups)) {
           const groupCol = countColumnName(fullToken);
-          if (!addedColumns.has(groupCol)) {
-            addedColumns.add(groupCol);
+          if (!tableColumns.has(groupCol)) {
+            tableColumns.add(groupCol);
             stmts.push(
-              `ALTER TABLE records ADD COLUMN ${groupCol} INTEGER NOT NULL DEFAULT 0`
+              `ALTER TABLE ${table} ADD COLUMN ${groupCol} INTEGER NOT NULL DEFAULT 0`
             );
           }
           stmts.push(
-            `CREATE INDEX IF NOT EXISTS idx_${sanitizeName(collection)}_${groupCol} ON records(collection, ${groupCol} DESC, time_us DESC)`
+            `CREATE INDEX IF NOT EXISTS idx_${sanitizeName(collection)}_${groupCol} ON ${table}(${groupCol} DESC, time_us DESC)`
           );
         }
       }
@@ -136,9 +152,10 @@ function buildFeedTables(config: ContrailConfig): string[] {
   // Index follow collections on subject for efficient fan-out lookups
   const followCollections = new Set(Object.values(config.feeds).map((f) => f.follow));
   for (const col of followCollections) {
-    const safe = col.replace(/[^a-zA-Z0-9]/g, "_");
+    const table = recordsTableName(col);
+    const safe = sanitizeName(col);
     stmts.push(
-      `CREATE INDEX IF NOT EXISTS idx_${safe}_subject ON records(collection, json_extract(record, '$.subject')) WHERE collection = '${col}'`
+      `CREATE INDEX IF NOT EXISTS idx_${safe}_subject ON ${table}(json_extract(record, '$.subject'))`
     );
   }
 
@@ -181,10 +198,11 @@ export async function initSchema(
     .map((s) => s.trim())
     .filter((s) => s.length > 0);
 
+  const collectionStatements = buildCollectionTables(config);
   const indexStatements = buildDynamicIndexes(config);
   const ftsStatements = buildFtsTables(config);
   const feedStatements = buildFeedTables(config);
-  const all = [...baseStatements, ...indexStatements, ...ftsStatements, ...feedStatements];
+  const all = [...baseStatements, ...collectionStatements, ...indexStatements, ...ftsStatements, ...feedStatements];
 
   await db.batch(all.map((s) => db.prepare(s)));
   await runMigrations(db);
diff --git a/src/core/router/admin.ts b/src/core/router/admin.ts
index f76c24c..4dd4a28 100644
--- a/src/core/router/admin.ts
+++ b/src/core/router/admin.ts
@@ -1,7 +1,7 @@
 import type { Hono, Context, Next } from "hono";
 import type { ContrailConfig, Database } from "../types";
+import { getCollectionNames, recordsTableName } from "../types";
 import { getLastCursor } from "../db";
-import { initSchema } from "../db/schema";
 
 export function registerAdminRoutes(
   app: Hono,
@@ -37,13 +37,18 @@ export function registerAdminRoutes(
   });
 
   app.get(`/xrpc/${ns}.admin.getOverview`, async (c) => {
-    const result = await db
-      .prepare(
-        "SELECT collection, COUNT(*) as records, COUNT(DISTINCT did) as unique_users FROM records GROUP BY collection"
-      )
-      .all<{ collection: string; records: number; unique_users: number }>();
+    const collections: { collection: string; records: number; unique_users: number }[] = [];
+
+    for (const collection of getCollectionNames(config)) {
+      const table = recordsTableName(collection);
+      const row = await db
+        .prepare(`SELECT COUNT(*) as records, COUNT(DISTINCT did) as unique_users FROM ${table}`)
+        .first<{ records: number; unique_users: number }>();
+      if (row) {
+        collections.push({ collection, records: row.records, unique_users: row.unique_users });
+      }
+    }
 
-    const collections = result.results ?? [];
     return c.json({
       total_records: collections.reduce((sum, col) => sum + col.records, 0),
       collections,
@@ -51,7 +56,8 @@ export function registerAdminRoutes(
   });
 
   app.get(`/xrpc/${ns}.admin.reset`, requireAdmin, async (c) => {
-    const tables = ["records", "backfills", "discovery", "cursor", "identities"];
+    const collectionTables = getCollectionNames(config).map(recordsTableName);
+    const tables = [...collectionTables, "backfills", "discovery", "cursor", "identities"];
     await db.batch(tables.map((t) => db.prepare(`DELETE FROM ${t}`)));
     return c.json({ ok: true });
   });
diff --git a/src/core/router/collection.ts b/src/core/router/collection.ts
index 1ed9892..8bcbcf3 100644
--- a/src/core/router/collection.ts
+++ b/src/core/router/collection.ts
@@ -1,6 +1,6 @@
 import type { Hono } from "hono";
 import type { ContrailConfig, Database, RecordRow, QueryableField, RecordSource } from "../types";
-import { getCollectionNames, countColumnName } from "../types";
+import { getCollectionNames, countColumnName, recordsTableName } from "../types";
 import { resolvedQueryable, resolvedRelationsMap } from "../queryable.generated";
 import { queryRecords } from "../db";
 import type { SortOption } from "../db/records";
@@ -194,16 +194,15 @@ export function registerCollectionRoutes(
       const relations = colConfig.relations ?? {};
       const references = colConfig.references ?? {};
 
+      const table = recordsTableName(collection);
       const row = await db
-        .prepare(
-          "SELECT * FROM records WHERE uri = ? AND collection = ?"
-        )
-        .bind(uri, collection)
+        .prepare(`SELECT * FROM ${table} WHERE uri = ?`)
+        .bind(uri)
         .first<any>();
 
       if (!row) return c.json({ error: "Record not found" }, 404);
 
-      const formatted = formatRecord(row);
+      const formatted = formatRecord({ ...row, collection });
       const counts = extractCounts(row, collection, relations);
       if (counts) flattenCounts(formatted, counts, collection, relations);
 
diff --git a/src/core/router/feed.ts b/src/core/router/feed.ts
index f009384..fba7995 100644
--- a/src/core/router/feed.ts
+++ b/src/core/router/feed.ts
@@ -1,6 +1,6 @@
 import type { Hono } from "hono";
 import type { ContrailConfig, Database, FeedConfig } from "../types";
-import { DEFAULT_FEED_MAX_ITEMS } from "../types";
+import { DEFAULT_FEED_MAX_ITEMS, recordsTableName } from "../types";
 import { resolveActor } from "../identity";
 import { backfillUser } from "../backfill";
 import { runPipeline } from "./collection";
@@ -33,22 +33,23 @@ async function maybeBackfillFeed(
   const maxItems = feedConfig.maxItems ?? DEFAULT_FEED_MAX_ITEMS;
 
   // Populate feed from existing records by followed users
+  const followTable = recordsTableName(feedConfig.follow);
   for (const targetCol of feedConfig.targets) {
+    const targetTable = recordsTableName(targetCol);
     await db
       .prepare(
         `INSERT OR IGNORE INTO feed_items (actor, uri, collection, time_us)
-         SELECT ?, r.uri, r.collection, r.time_us
-         FROM records r
-         WHERE r.collection = ?
-           AND r.did IN (
+         SELECT ?, r.uri, ?, r.time_us
+         FROM ${targetTable} r
+         WHERE r.did IN (
              SELECT json_extract(f.record, '$.subject')
-             FROM records f
-             WHERE f.collection = ? AND f.did = ?
+             FROM ${followTable} f
+             WHERE f.did = ?
            )
          ORDER BY r.time_us DESC
          LIMIT ?`
       )
-      .bind(actor, targetCol, feedConfig.follow, actor, maxItems)
+      .bind(actor, targetCol, actor, maxItems)
       .run();
   }
 
diff --git a/src/core/router/hydrate.ts b/src/core/router/hydrate.ts
index 0cde478..005b615 100644
--- a/src/core/router/hydrate.ts
+++ b/src/core/router/hydrate.ts
@@ -1,5 +1,5 @@
 import type { RelationConfig, ReferenceConfig, RecordRow, Database } from "../types";
-import { getNestedValue, getRelationField } from "../types";
+import { getNestedValue, getRelationField, recordsTableName } from "../types";
 import { batchedInQuery, formatRecord } from "./helpers";
 
 // --- Hydration: embed related records ---
@@ -44,13 +44,13 @@ export async function resolveHydrates(
 ): Promise<HydrateResult> {
   if (Object.keys(requested).length === 0 || records.length === 0) return {};
 
-  // Accumulate grouped results first, then flatten ungrouped ones
   const grouped: Record<string, Record<string, Record<string, any[]>>> = {};
 
   for (const [relName, hydrateLimit] of Object.entries(requested)) {
     const rel = relations[relName];
     const field = getRelationField(rel);
     const matchMode = rel.match ?? "uri";
+    const table = recordsTableName(rel.collection);
 
     const matchValues = matchMode === "did"
       ? [...new Set(records.map((r) => r.did))]
@@ -58,16 +58,15 @@ export async function resolveHydrates(
 
     if (matchValues.length === 0) continue;
 
-    // Fetch more rows than needed since the limit applies per-group, not total
-    const groupCount = rel.groupBy ? 10 : 1; // estimate; overfetch is fine
+    const groupCount = rel.groupBy ? 10 : 1;
     const maxRows = matchValues.length * hydrateLimit * groupCount;
-    const relatedRows = await batchedInQuery<RecordRow>(
+    const relatedRows = await batchedInQuery<Omit<RecordRow, "collection">>(
       db,
-      `SELECT uri, did, collection, rkey, record, time_us FROM records
-       WHERE collection = ? AND json_extract(record, '$.${field}') IN (__IN__)
+      `SELECT uri, did, rkey, record, time_us FROM ${table}
+       WHERE json_extract(record, '$.${field}') IN (__IN__)
        ORDER BY time_us DESC
        LIMIT ${maxRows}`,
-      [rel.collection],
+      [],
       matchValues
     );
 
@@ -93,13 +92,12 @@ export async function resolveHydrates(
 
         const group = grouped[targetUri][relName][groupValue];
         if (group.length < hydrateLimit) {
-          group.push(formatRecord(row));
+          group.push(formatRecord({ ...row, collection: rel.collection }));
         }
       }
     }
   }
 
-  // Convert to final shape: ungrouped relations become flat arrays
   const result: HydrateResult = {};
   for (const [uri, rels] of Object.entries(grouped)) {
     result[uri] = {};
@@ -133,8 +131,9 @@ export async function resolveReferences(
     const ref = references[refName];
     if (!ref) continue;
 
-    // Extract target URIs from our records
-    const targetMap = new Map<string, string[]>(); // targetUri → parentUris[]
+    const table = recordsTableName(ref.collection);
+
+    const targetMap = new Map<string, string[]>();
     for (const r of records) {
       const parsed = r.record ? JSON.parse(r.record) : null;
       const targetValue = parsed ? getNestedValue(parsed, ref.field) : null;
@@ -146,11 +145,11 @@ export async function resolveReferences(
     const targetUris = [...targetMap.keys()];
     if (targetUris.length === 0) continue;
 
-    const rows = await batchedInQuery<RecordRow>(
+    const rows = await batchedInQuery<Omit<RecordRow, "collection">>(
       db,
-      `SELECT uri, did, collection, rkey, record, time_us FROM records
-       WHERE collection = ? AND uri IN (__IN__)`,
-      [ref.collection],
+      `SELECT uri, did, rkey, record, time_us FROM ${table}
+       WHERE uri IN (__IN__)`,
+      [],
       targetUris
     );
 
@@ -158,7 +157,7 @@ export async function resolveReferences(
       const parentUris = targetMap.get(row.uri) ?? [];
       for (const parentUri of parentUris) {
         if (!result[parentUri]) result[parentUri] = {};
-        result[parentUri][refName] = formatRecord(row);
+        result[parentUri][refName] = formatRecord({ ...row, collection: ref.collection });
       }
     }
   }
diff --git a/src/core/router/notify.ts b/src/core/router/notify.ts
index 72e4672..e8f681f 100644
--- a/src/core/router/notify.ts
+++ b/src/core/router/notify.ts
@@ -1,5 +1,6 @@
 import type { Hono } from "hono";
 import type { Database, ContrailConfig, IngestEvent } from "../types";
+import { recordsTableName } from "../types";
 import { applyEvents } from "../db/records";
 import { getPDS } from "../client";
 import type { Did } from "@atcute/lexicons";
@@ -89,8 +90,9 @@ export function registerNotifyRoute(
       const now = Date.now() * 1000; // microseconds
 
       // Check if this record already exists locally
+      const table = recordsTableName(parsed.collection);
       const existing = await db
-        .prepare("SELECT cid FROM records WHERE uri = ?")
+        .prepare(`SELECT cid FROM ${table} WHERE uri = ?`)
         .bind(uri)
         .first<{ cid: string | null }>();
 
@@ -116,7 +118,7 @@ export function registerNotifyRoute(
         // Record gone from PDS but exists locally — delete it.
         // We need the old record data so buildCountStatements can decrement counts.
         const existingRecord = await db
-          .prepare("SELECT record FROM records WHERE uri = ?")
+          .prepare(`SELECT record FROM ${table} WHERE uri = ?`)
           .bind(uri)
           .first<{ record: string | null }>();
 
diff --git a/src/core/router/profiles.ts b/src/core/router/profiles.ts
index f926bb8..4551c5b 100644
--- a/src/core/router/profiles.ts
+++ b/src/core/router/profiles.ts
@@ -1,4 +1,5 @@
 import type { Database, ContrailConfig, RecordRow } from "../types";
+import { recordsTableName } from "../types";
 import { resolveIdentities } from "../identity";
 import { batchedInQuery } from "./helpers";
 
@@ -46,11 +47,12 @@ export async function resolveProfiles(
     const remaining = dids.filter((d) => !result[d]);
     if (remaining.length === 0) break;
 
+    const table = recordsTableName(collection);
     const uris = remaining.map((did) => `at://${did}/${collection}/self`);
 
-    const rows = await batchedInQuery<RecordRow>(
+    const rows = await batchedInQuery<Omit<RecordRow, "collection">>(
       db,
-      `SELECT uri, did, collection, rkey, cid, record FROM records WHERE uri IN (__IN__)`,
+      `SELECT uri, did, rkey, cid, record FROM ${table} WHERE uri IN (__IN__)`,
       [],
       uris
     );
@@ -68,7 +70,7 @@ export async function resolveProfiles(
         did: row.did,
         handle: null, // filled below
         uri: row.uri,
-        collection: row.collection,
+        collection,
         rkey: row.rkey,
         cid: row.cid,
         record,
diff --git a/src/core/types.ts b/src/core/types.ts
index 8d79a48..b1fb76c 100644
--- a/src/core/types.ts
+++ b/src/core/types.ts
@@ -203,6 +203,10 @@ export function countColumnName(type: string): string {
   return "count_" + type.replace(/[^a-zA-Z0-9]/g, "_");
 }
 
+export function recordsTableName(collection: string): string {
+  return "records_" + collection.replace(/[^a-zA-Z0-9]/g, "_");
+}
+
 export function getCollectionNames(config: ContrailConfig): string[] {
   return Object.keys(config.collections);
 }
diff --git a/src/generate.ts b/src/generate.ts
index f4b8264..21a43cf 100644
--- a/src/generate.ts
+++ b/src/generate.ts
@@ -601,7 +601,8 @@ export function generateLexicons(options: GenerateOptions): Record<string, objec
       for (const ref of findRefsInLexicon(file)) allRefs.add(ref);
     }
     const profileNsids = config.profiles ?? ["app.bsky.actor.profile"];
-    const pullNsids = new Set([...collectionNsids, ...profileNsids]);
+    const feedFollowNsids = config.feeds ? Object.values(config.feeds).map((f) => f.follow) : [];
+    const pullNsids = new Set([...collectionNsids, ...profileNsids, ...feedFollowNsids]);
     for (const ref of allRefs) {
       if (!ref.startsWith("com.atproto.")) pullNsids.add(ref);
     }
diff --git a/tests/hydrate.test.ts b/tests/hydrate.test.ts
index b9ca899..5dc95de 100644
--- a/tests/hydrate.test.ts
+++ b/tests/hydrate.test.ts
@@ -1,5 +1,6 @@
 import { describe, it, expect, beforeEach } from "vitest";
 import type { Database, RecordRow, RelationConfig, ReferenceConfig } from "../src/core/types";
+import { recordsTableName } from "../src/core/types";
 import { parseHydrateParams, resolveHydrates, resolveReferences } from "../src/core/router/hydrate";
 import { createTestDbWithSchema, makeEvent, TEST_CONFIG } from "./helpers";
 import { applyEvents } from "../src/core/db/records";
@@ -99,7 +100,7 @@ describe("resolveHydrates", () => {
     }
 
     const eventRow = await db
-      .prepare("SELECT * FROM records WHERE uri = ?")
+      .prepare(`SELECT * FROM ${recordsTableName("community.lexicon.calendar.event")} WHERE uri = ?`)
       .bind(eventUri)
       .first<RecordRow>();
 
@@ -129,7 +130,7 @@ describe("resolveHydrates", () => {
     }
 
     const eventRow = await db
-      .prepare("SELECT * FROM records WHERE uri = ?")
+      .prepare(`SELECT * FROM ${recordsTableName("community.lexicon.calendar.event")} WHERE uri = ?`)
       .bind(eventUri)
       .first<RecordRow>();
 
@@ -159,7 +160,7 @@ describe("resolveHydrates", () => {
     ]);
 
     const eventRow = await db
-      .prepare("SELECT * FROM records WHERE uri = ?")
+      .prepare(`SELECT * FROM ${recordsTableName("community.lexicon.calendar.event")} WHERE uri = ?`)
       .bind(eventUri)
       .first<RecordRow>();
 
@@ -196,7 +197,7 @@ describe("resolveHydrates", () => {
     ]);
 
     const rsvpRow = await db
-      .prepare("SELECT * FROM records WHERE uri = ?")
+      .prepare(`SELECT * FROM ${recordsTableName("community.lexicon.calendar.rsvp")} WHERE uri = ?`)
       .bind(rsvpUri)
       .first<RecordRow>();
 
@@ -242,8 +243,7 @@ describe("resolveHydrates", () => {
     ]);
 
     const rsvpRows = await db
-      .prepare("SELECT * FROM records WHERE collection = ? ORDER BY time_us DESC")
-      .bind("community.lexicon.calendar.rsvp")
+      .prepare(`SELECT * FROM ${recordsTableName("community.lexicon.calendar.rsvp")} ORDER BY time_us DESC`)
       .all<RecordRow>();
 
     const references = TEST_CONFIG.collections["community.lexicon.calendar.rsvp"].references!;
@@ -274,7 +274,7 @@ describe("resolveHydrates", () => {
     ]);
 
     const eventRow = await db
-      .prepare("SELECT * FROM records WHERE uri = ?")
+      .prepare(`SELECT * FROM ${recordsTableName("community.lexicon.calendar.event")} WHERE uri = ?`)
       .bind(eventUri)
       .first<RecordRow>();
 
diff --git a/tests/records.test.ts b/tests/records.test.ts
index 662b127..7e3e401 100644
--- a/tests/records.test.ts
+++ b/tests/records.test.ts
@@ -277,11 +277,12 @@ describe("queryRecords", () => {
     expect(result.cursor).toBeUndefined();
   });
 
-  it("returns empty for unknown collection", async () => {
-    const result = await queryRecords(db, TEST_CONFIG, {
-      collection: "nonexistent.collection",
-    });
-    expect(result.records).toHaveLength(0);
+  it("throws for unknown collection (table does not exist)", async () => {
+    await expect(
+      queryRecords(db, TEST_CONFIG, {
+        collection: "nonexistent.collection",
+      })
+    ).rejects.toThrow();
   });
 
   it("filter keys are interpolated into SQL — only trusted input should be passed", async () => {
diff --git a/tests/schema.test.ts b/tests/schema.test.ts
index 5b28f16..54829ff 100644
--- a/tests/schema.test.ts
+++ b/tests/schema.test.ts
@@ -12,7 +12,8 @@ describe("initSchema", () => {
       .all<{ name: string }>();
     const names = tables.results.map((t) => t.name);
 
-    expect(names).toContain("records");
+    expect(names).toContain("records_community_lexicon_calendar_event");
+    expect(names).toContain("records_community_lexicon_calendar_rsvp");
     expect(names).toContain("backfills");
     expect(names).toContain("discovery");
     expect(names).toContain("cursor");
-- 
2.51.2