From f07527a4b5035703fa748ac5ff5f15c513012dfd Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Fri, 14 Aug 2026 16:32:04 +0000 Subject: [PATCH 1/2] feat: Add Postgres schema, indexes, and feed queries Users, follows, and posts sit in 3NF with a composite author timeline index and keyset SQL. PostgresStore implements ActivityStore and runs the shared contract suite against PGlite. --- README.md | 17 ++- package.json | 3 +- pnpm-lock.yaml | 8 ++ src/index.ts | 4 +- src/postgres/schema.ts | 56 +++++++++ src/postgres/store.ts | 233 ++++++++++++++++++++++++++++++++++++ test/postgres-store.test.ts | 90 ++++++++++++++ 7 files changed, 405 insertions(+), 6 deletions(-) create mode 100644 src/postgres/schema.ts create mode 100644 src/postgres/store.ts create mode 100644 test/postgres-store.test.ts diff --git a/README.md b/README.md index e1b6683..8f98f29 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ One domain, many stores: the same activity feed modeled on relational, document, ## What this demonstrates -Polyglot persistence is the idea that one product rarely has one ideal database. A follow graph wants adjacency lookups, a home timeline wants a clustered time key, and a unique handle wants a relational constraint. This repo keeps one domain and one `ActivityStore` port, then implements that port on each paradigm so query shape, consistency, and scaling stay comparable. The first backend is in-memory fan-in-on-read: per-author timelines merged into a home feed. Later backends (Postgres, Redis, MongoDB, Cassandra, CockroachDB, Neo4j) keep the same interface and the same contract tests. +Polyglot persistence is the idea that one product rarely has one ideal database. A follow graph wants adjacency lookups, a home timeline wants a clustered time key, and a unique handle wants a relational constraint. This repo keeps one domain and one `ActivityStore` port, then implements that port on each paradigm so query shape, consistency, and scaling stay comparable. The in-memory backend is fan-in on read: per-author timelines merged into a home feed. The Postgres backend is the same port over a 3NF schema, composite B-tree indexes, and keyset SQL. Later backends (Redis, MongoDB, Cassandra, CockroachDB, Neo4j) keep the same interface and the same contract tests. ## Concepts demonstrated @@ -16,18 +16,27 @@ Polyglot persistence is the idea that one product rarely has one ideal database. - Compound cursor pagination over a total order `(createdAt DESC, id DESC)` - Unique constraints on user id and handle - Directed follow graph with idempotent edges and live (not snapshot) feeds +- Third normal form: users, a composite follow edge, posts +- Composite B-tree indexes `(author_id, created_at DESC, id DESC)` for a newest-first timeline +- Keyset pagination with a row comparison `(created_at, id) < cursor` +- Join-based fan-in: home feed is `posts INNER JOIN follows` +- Table-level `CHECK` (no self-follow) and `UNIQUE` (handle) as the source of integrity +- Query planner: `EXPLAIN` should pick `posts_author_timeline_idx` for the feed ## What's implemented - Project scaffold with TypeScript strict mode, Vitest, and CI - Activity-feed domain, `ActivityStore` port, in-memory backend, and a shared contract suite +- Relational backend (Postgres): schema, indexes, the SQL queries for the domain ## Usage ```ts -import { MemoryStore } from 'polyglot-persistence' +import { PGlite } from '@electric-sql/pglite' +import { PostgresStore } from 'polyglot-persistence' -const store = new MemoryStore() +const db = new PGlite() +const store = await PostgresStore.create(db) await store.createUser({ id: 'ada', handle: 'ada' }) await store.createUser({ id: 'bob', handle: 'bob' }) @@ -42,7 +51,7 @@ const page = await store.feed('ada', { limit: 20 }) const next = await store.feed('ada', { limit: 20, before: post }) ``` -A new backend implements `ActivityStore` and calls `defineStoreContract(name, factory)` from `test/contract.ts`. If the contract is green, the store matches the domain. +`PostgresStore` takes any `{ query(sql, params) }` client (PGlite in tests, `pg` against a server). A new backend implements `ActivityStore` and calls `defineStoreContract(name, factory)` from `test/contract.ts`. ## Running the tests diff --git a/package.json b/package.json index 0fbd730..5e928d7 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "polyglot-persistence", - "version": "0.1.0", + "version": "0.2.0", "description": "One domain modeled across relational, document, key-value, wide-column, distributed SQL, and graph stores. A reference for when to reach for which database.", "type": "module", "license": "MIT", @@ -11,6 +11,7 @@ "build": "tsc" }, "devDependencies": { + "@electric-sql/pglite": "^0.5.5", "typescript": "^5.6.0", "vitest": "^2.1.0" } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c1a394e..bfbc9cf 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -8,6 +8,9 @@ importers: .: devDependencies: + '@electric-sql/pglite': + specifier: ^0.5.5 + version: 0.5.5 typescript: specifier: ^5.6.0 version: 5.9.3 @@ -17,6 +20,9 @@ importers: packages: + '@electric-sql/pglite@0.5.5': + resolution: {integrity: sha512-QeZ+oB7QaU4EJj2awrB8hwFZ5TXxLRBQ9Gfq0dQauYDdWsRtuWRCERgtri35vRfL07KQHlVlc/4Qxvn7a5AXeA==} + '@esbuild/aix-ppc64@0.21.5': resolution: {integrity: sha512-1SDgH6ZSPTlggy1yI6+Dbkiz8xzpHJEVAlF/AM1tHPLsf5STom9rwtjE4hKAF20FfXXNTFqEYXyJNWh1GiZedQ==} engines: {node: '>=12'} @@ -521,6 +527,8 @@ packages: snapshots: + '@electric-sql/pglite@0.5.5': {} + '@esbuild/aix-ppc64@0.21.5': optional: true diff --git a/src/index.ts b/src/index.ts index 9cb1c2e..f32c174 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,4 @@ -export const VERSION = '0.1.0' +export const VERSION = '0.2.0' export { BODY_MAX, StoreError } from './domain' export type { FeedCursor, @@ -11,3 +11,5 @@ export type { } from './domain' export type { ActivityStore, CreateUserInput, PublishInput } from './store' export { MemoryStore } from './memory-store' +export { PostgresStore, SCHEMA_STATEMENTS, SQL } from './postgres/store' +export type { SqlQuery } from './postgres/store' diff --git a/src/postgres/schema.ts b/src/postgres/schema.ts new file mode 100644 index 0000000..d23c48d --- /dev/null +++ b/src/postgres/schema.ts @@ -0,0 +1,56 @@ +export const SCHEMA_STATEMENTS = [ + `CREATE TABLE IF NOT EXISTS users ( + id TEXT PRIMARY KEY, + handle TEXT NOT NULL, + created_at BIGINT NOT NULL + )`, + `CREATE UNIQUE INDEX IF NOT EXISTS users_handle_uidx ON users (handle)`, + `CREATE TABLE IF NOT EXISTS follows ( + follower_id TEXT NOT NULL REFERENCES users (id) ON DELETE CASCADE, + followee_id TEXT NOT NULL REFERENCES users (id) ON DELETE CASCADE, + PRIMARY KEY (follower_id, followee_id), + CONSTRAINT follows_no_self CHECK (follower_id <> followee_id) + )`, + `CREATE INDEX IF NOT EXISTS follows_inbound_idx ON follows (followee_id)`, + `CREATE TABLE IF NOT EXISTS posts ( + id TEXT PRIMARY KEY, + author_id TEXT NOT NULL REFERENCES users (id) ON DELETE CASCADE, + body TEXT NOT NULL, + created_at BIGINT NOT NULL, + CONSTRAINT posts_body_len CHECK (char_length(body) BETWEEN 1 AND 280) + )`, + `CREATE INDEX IF NOT EXISTS posts_author_timeline_idx + ON posts (author_id, created_at DESC, id DESC)`, +] as const + +export const SQL = { + insertUser: `INSERT INTO users (id, handle, created_at) + VALUES ($1, $2, $3) + RETURNING id, handle, created_at`, + selectUser: `SELECT id, handle, created_at FROM users WHERE id = $1`, + selectUserByHandle: `SELECT id, handle, created_at FROM users WHERE handle = $1`, + insertFollow: `INSERT INTO follows (follower_id, followee_id) + VALUES ($1, $2) + ON CONFLICT (follower_id, followee_id) DO NOTHING`, + deleteFollow: `DELETE FROM follows WHERE follower_id = $1 AND followee_id = $2`, + hasFollow: `SELECT 1 FROM follows WHERE follower_id = $1 AND followee_id = $2`, + following: `SELECT followee_id FROM follows WHERE follower_id = $1 ORDER BY followee_id`, + followers: `SELECT follower_id FROM follows WHERE followee_id = $1 ORDER BY follower_id`, + insertPost: `INSERT INTO posts (id, author_id, body, created_at) + VALUES ($1, $2, $3, $4) + RETURNING id, author_id, body, created_at`, + selectPost: `SELECT id, author_id, body, created_at FROM posts WHERE id = $1`, + authorTimeline: `SELECT id, author_id, body, created_at + FROM posts + WHERE author_id = $1 + AND ($2::bigint IS NULL OR (created_at, id) < ($2::bigint, $3::text)) + ORDER BY created_at DESC, id DESC + LIMIT $4`, + feed: `SELECT p.id, p.author_id, p.body, p.created_at + FROM posts p + INNER JOIN follows f ON f.followee_id = p.author_id + WHERE f.follower_id = $1 + AND ($2::bigint IS NULL OR (p.created_at, p.id) < ($2::bigint, $3::text)) + ORDER BY p.created_at DESC, p.id DESC + LIMIT $4`, +} as const diff --git a/src/postgres/store.ts b/src/postgres/store.ts new file mode 100644 index 0000000..f78829f --- /dev/null +++ b/src/postgres/store.ts @@ -0,0 +1,233 @@ +import { + normalizeBody, + normalizeHandle, + normalizeId, + pageLimit, + StoreError, + tryNormalizeHandle, + tryNormalizeId, + type Page, + type Post, + type PostId, + type User, + type UserId, +} from '../domain' +import type { ActivityStore, CreateUserInput, PublishInput } from '../store' +import { SCHEMA_STATEMENTS, SQL } from './schema' + +export { SCHEMA_STATEMENTS, SQL } from './schema' + +export interface SqlQuery { + query>( + sql: string, + params?: unknown[], + ): Promise<{ rows: T[] }> +} + +interface UserRow { + id: string + handle: string + created_at: string | number +} + +interface PostRow { + id: string + author_id: string + body: string + created_at: string | number +} + +interface FolloweeRow { + followee_id: string +} + +interface FollowerRow { + follower_id: string +} + +export class PostgresStore implements ActivityStore { + private constructor(private readonly sql: SqlQuery) {} + + static async migrate(sql: SqlQuery): Promise { + for (const statement of SCHEMA_STATEMENTS) { + await sql.query(statement) + } + } + + static attach(sql: SqlQuery): PostgresStore { + return new PostgresStore(sql) + } + + static async create(sql: SqlQuery): Promise { + await PostgresStore.migrate(sql) + return PostgresStore.attach(sql) + } + + async createUser(input: CreateUserInput, now = Date.now()): Promise { + const id = normalizeId(input.id) + const handle = normalizeHandle(input.handle) + try { + const { rows } = await this.sql.query(SQL.insertUser, [id, handle, now]) + const row = rows[0] + if (!row) throw new StoreError('user_exists') + return toUser(row) + } catch (err) { + throwMapped(err) + } + } + + async getUser(id: UserId): Promise { + const key = tryNormalizeId(id) + if (!key) return null + const { rows } = await this.sql.query(SQL.selectUser, [key]) + const row = rows[0] + return row ? toUser(row) : null + } + + async getUserByHandle(handle: string): Promise { + const key = tryNormalizeHandle(handle) + if (!key) return null + const { rows } = await this.sql.query(SQL.selectUserByHandle, [key]) + const row = rows[0] + return row ? toUser(row) : null + } + + async follow(followerId: UserId, followeeId: UserId): Promise { + const from = await this.requireUser(followerId) + const to = await this.requireUser(followeeId) + if (from === to) throw new StoreError('self_follow') + try { + await this.sql.query(SQL.insertFollow, [from, to]) + } catch (err) { + throwMapped(err) + } + } + + async unfollow(followerId: UserId, followeeId: UserId): Promise { + const from = await this.requireUser(followerId) + const to = await this.requireUser(followeeId) + await this.sql.query(SQL.deleteFollow, [from, to]) + } + + async isFollowing(followerId: UserId, followeeId: UserId): Promise { + const from = await this.requireUser(followerId) + const to = await this.requireUser(followeeId) + const { rows } = await this.sql.query(SQL.hasFollow, [from, to]) + return rows.length > 0 + } + + async following(userId: UserId): Promise { + const id = await this.requireUser(userId) + const { rows } = await this.sql.query(SQL.following, [id]) + return rows.map((row) => row.followee_id) + } + + async followers(userId: UserId): Promise { + const id = await this.requireUser(userId) + const { rows } = await this.sql.query(SQL.followers, [id]) + return rows.map((row) => row.follower_id) + } + + async publish(input: PublishInput, now = Date.now()): Promise { + const id = normalizeId(input.id) + const authorId = await this.requireUser(input.authorId) + const body = normalizeBody(input.body) + try { + const { rows } = await this.sql.query(SQL.insertPost, [ + id, + authorId, + body, + now, + ]) + const row = rows[0] + if (!row) throw new StoreError('post_exists') + return toPost(row) + } catch (err) { + throwMapped(err) + } + } + + async getPost(id: PostId): Promise { + const key = tryNormalizeId(id) + if (!key) return null + const { rows } = await this.sql.query(SQL.selectPost, [key]) + const row = rows[0] + return row ? toPost(row) : null + } + + async postsByAuthor(authorId: UserId, page?: Page): Promise { + const id = await this.requireUser(authorId) + const [at, cursorId] = cursorParams(page) + const { rows } = await this.sql.query(SQL.authorTimeline, [ + id, + at, + cursorId, + pageLimit(page), + ]) + return rows.map(toPost) + } + + async feed(userId: UserId, page?: Page): Promise { + const id = await this.requireUser(userId) + const [at, cursorId] = cursorParams(page) + const { rows } = await this.sql.query(SQL.feed, [ + id, + at, + cursorId, + pageLimit(page), + ]) + return rows.map(toPost) + } + + private async requireUser(id: UserId): Promise { + const key = tryNormalizeId(id) + if (!key) throw new StoreError('user_not_found') + const { rows } = await this.sql.query(SQL.selectUser, [key]) + if (!rows[0]) throw new StoreError('user_not_found') + return key + } +} + +function cursorParams(page?: Page): [number | null, string | null] { + const before = page?.before + if (!before) return [null, null] + return [before.createdAt, before.id] +} + +function epoch(value: string | number): number { + return typeof value === 'number' ? value : Number(value) +} + +function toUser(row: UserRow): User { + return { id: row.id, handle: row.handle, createdAt: epoch(row.created_at) } +} + +function toPost(row: PostRow): Post { + return { + id: row.id, + authorId: row.author_id, + body: row.body, + createdAt: epoch(row.created_at), + } +} + +function pgField(err: unknown, key: 'code' | 'constraint'): string | undefined { + if (typeof err !== 'object' || err === null || !(key in err)) return undefined + const value = (err as Record)[key] + return typeof value === 'string' ? value : undefined +} + +function throwMapped(err: unknown): never { + const code = pgField(err, 'code') + const constraint = pgField(err, 'constraint') + if (code === '23505') { + if (constraint === 'users_pkey') throw new StoreError('user_exists') + if (constraint === 'users_handle_uidx') throw new StoreError('handle_taken') + if (constraint === 'posts_pkey') throw new StoreError('post_exists') + } + if (code === '23514' && constraint === 'follows_no_self') { + throw new StoreError('self_follow') + } + if (code === '23503') throw new StoreError('user_not_found') + throw err +} diff --git a/test/postgres-store.test.ts b/test/postgres-store.test.ts new file mode 100644 index 0000000..29b21e5 --- /dev/null +++ b/test/postgres-store.test.ts @@ -0,0 +1,90 @@ +import { PGlite } from '@electric-sql/pglite' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { PostgresStore, SQL } from '../src/postgres/store' +import { defineStoreContract } from './contract' + +const db = new PGlite() + +beforeAll(async () => { + await db.waitReady + await PostgresStore.migrate(db) +}, 30_000) + +afterAll(async () => { + await db.close() +}) + +defineStoreContract('postgres', async () => { + await db.query('TRUNCATE TABLE users CASCADE') + return PostgresStore.attach(db) +}) + +describe('postgres schema and planner', () => { + it('creates the unique handle index and the author timeline index', async () => { + const { rows } = await db.query<{ indexname: string }>( + `SELECT indexname FROM pg_indexes + WHERE schemaname = 'public' + AND indexname IN ('users_handle_uidx', 'posts_author_timeline_idx', 'follows_inbound_idx') + ORDER BY indexname`, + ) + expect(rows.map((row) => row.indexname)).toEqual([ + 'follows_inbound_idx', + 'posts_author_timeline_idx', + 'users_handle_uidx', + ]) + }) + + it('rejects a self-follow at the table even when the store is skipped', async () => { + await db.query('TRUNCATE TABLE users CASCADE') + await db.query(`INSERT INTO users (id, handle, created_at) VALUES ('ada', 'ada', 1)`) + await expect( + db.query(`INSERT INTO follows (follower_id, followee_id) VALUES ('ada', 'ada')`), + ).rejects.toMatchObject({ code: '23514', constraint: 'follows_no_self' }) + }) + + it('enforces handle uniqueness at the table', async () => { + await db.query('TRUNCATE TABLE users CASCADE') + await db.query(`INSERT INTO users (id, handle, created_at) VALUES ('u1', 'ada', 1)`) + await expect( + db.query(`INSERT INTO users (id, handle, created_at) VALUES ('u2', 'ada', 2)`), + ).rejects.toMatchObject({ code: '23505', constraint: 'users_handle_uidx' }) + }) + + it('pages equal timestamps with a row-comparison keyset', async () => { + await db.query('TRUNCATE TABLE users CASCADE') + await db.query( + `INSERT INTO users (id, handle, created_at) VALUES ('ada', 'ada', 1), ('bob', 'bob', 1)`, + ) + await db.query(`INSERT INTO follows (follower_id, followee_id) VALUES ('ada', 'bob')`) + await db.query( + `INSERT INTO posts (id, author_id, body, created_at) VALUES + ('p1', 'bob', 'older-id', 10), + ('p2', 'bob', 'newer-id', 10), + ('p3', 'bob', 'earlier', 4)`, + ) + const first = await db.query<{ id: string }>(SQL.feed, ['ada', null, null, 1]) + expect(first.rows.map((row) => row.id)).toEqual(['p2']) + const second = await db.query<{ id: string }>(SQL.feed, ['ada', 10, 'p2', 2]) + expect(second.rows.map((row) => row.id)).toEqual(['p1', 'p3']) + }) + + it('uses the author timeline index for the feed join when seq scans are off', async () => { + await db.query('TRUNCATE TABLE users CASCADE') + await db.query( + `INSERT INTO users (id, handle, created_at) VALUES ('ada', 'ada', 1), ('bob', 'bob', 1)`, + ) + await db.query(`INSERT INTO follows (follower_id, followee_id) VALUES ('ada', 'bob')`) + await db.query( + `INSERT INTO posts (id, author_id, body, created_at) + SELECT 'p' || g, 'bob', 'body', g FROM generate_series(1, 40) AS g`, + ) + await db.query('SET enable_seqscan = off') + const { rows } = await db.query<{ 'QUERY PLAN': string }>( + `EXPLAIN ${SQL.feed}`, + ['ada', null, null, 20], + ) + await db.query('SET enable_seqscan = on') + const plan = rows.map((row) => row['QUERY PLAN']).join('\n') + expect(plan).toMatch(/posts_author_timeline_idx/) + }) +}) From 580ab6d8c678d01b383cff6bd9917d4a4074a653 Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Fri, 14 Aug 2026 16:42:56 +0000 Subject: [PATCH 2/2] fix: Walk the author timeline index and split keyset SQL Seed enough posts, ANALYZE, and EXPLAIN the author timeline as an index scan with no sort. Walk each followee through the same index with LATERAL. First page and keyed page are separate statements. --- README.md | 14 ++++++------- src/postgres/schema.ts | 31 ++++++++++++++++++++++++---- src/postgres/store.ts | 40 ++++++++++++++++++------------------- test/postgres-store.test.ts | 37 +++++++++++++++++++++++----------- 4 files changed, 78 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index 8f98f29..6e7992b 100644 --- a/README.md +++ b/README.md @@ -18,10 +18,10 @@ Polyglot persistence is the idea that one product rarely has one ideal database. - Directed follow graph with idempotent edges and live (not snapshot) feeds - Third normal form: users, a composite follow edge, posts - Composite B-tree indexes `(author_id, created_at DESC, id DESC)` for a newest-first timeline -- Keyset pagination with a row comparison `(created_at, id) < cursor` -- Join-based fan-in: home feed is `posts INNER JOIN follows` +- Keyset pagination with a row comparison `(created_at, id) < cursor` (first page and keyed page are separate statements) +- Fan-in via `CROSS JOIN LATERAL`: each followee's newest-N posts walk `posts_author_timeline_idx`, then the outer query merges - Table-level `CHECK` (no self-follow) and `UNIQUE` (handle) as the source of integrity -- Query planner: `EXPLAIN` should pick `posts_author_timeline_idx` for the feed +- Query planner: after `ANALYZE`, `EXPLAIN` on the author timeline is `Index Scan using posts_author_timeline_idx` with no Sort. The lateral feed uses that same scan per followee. ## What's implemented @@ -32,11 +32,9 @@ Polyglot persistence is the idea that one product rarely has one ideal database. ## Usage ```ts -import { PGlite } from '@electric-sql/pglite' -import { PostgresStore } from 'polyglot-persistence' +import { MemoryStore } from 'polyglot-persistence' -const db = new PGlite() -const store = await PostgresStore.create(db) +const store = new MemoryStore() await store.createUser({ id: 'ada', handle: 'ada' }) await store.createUser({ id: 'bob', handle: 'bob' }) @@ -51,7 +49,7 @@ const page = await store.feed('ada', { limit: 20 }) const next = await store.feed('ada', { limit: 20, before: post }) ``` -`PostgresStore` takes any `{ query(sql, params) }` client (PGlite in tests, `pg` against a server). A new backend implements `ActivityStore` and calls `defineStoreContract(name, factory)` from `test/contract.ts`. +`PostgresStore` takes any `{ query(sql, params) }` client. Tests drive it with PGlite (`@electric-sql/pglite` is a devDependency, not a runtime dep). A server `pg` Pool works the same way. A new backend implements `ActivityStore` and calls `defineStoreContract(name, factory)` from `test/contract.ts`. ## Running the tests diff --git a/src/postgres/schema.ts b/src/postgres/schema.ts index d23c48d..610462a 100644 --- a/src/postgres/schema.ts +++ b/src/postgres/schema.ts @@ -43,14 +43,37 @@ export const SQL = { authorTimeline: `SELECT id, author_id, body, created_at FROM posts WHERE author_id = $1 - AND ($2::bigint IS NULL OR (created_at, id) < ($2::bigint, $3::text)) + ORDER BY created_at DESC, id DESC + LIMIT $2`, + authorTimelineBefore: `SELECT id, author_id, body, created_at + FROM posts + WHERE author_id = $1 + AND (created_at, id) < ($2::bigint, $3::text) ORDER BY created_at DESC, id DESC LIMIT $4`, feed: `SELECT p.id, p.author_id, p.body, p.created_at - FROM posts p - INNER JOIN follows f ON f.followee_id = p.author_id + FROM follows f + CROSS JOIN LATERAL ( + SELECT p.id, p.author_id, p.body, p.created_at + FROM posts p + WHERE p.author_id = f.followee_id + ORDER BY p.created_at DESC, p.id DESC + LIMIT $2 + ) p + WHERE f.follower_id = $1 + ORDER BY p.created_at DESC, p.id DESC + LIMIT $2`, + feedBefore: `SELECT p.id, p.author_id, p.body, p.created_at + FROM follows f + CROSS JOIN LATERAL ( + SELECT p.id, p.author_id, p.body, p.created_at + FROM posts p + WHERE p.author_id = f.followee_id + AND (p.created_at, p.id) < ($2::bigint, $3::text) + ORDER BY p.created_at DESC, p.id DESC + LIMIT $4 + ) p WHERE f.follower_id = $1 - AND ($2::bigint IS NULL OR (p.created_at, p.id) < ($2::bigint, $3::text)) ORDER BY p.created_at DESC, p.id DESC LIMIT $4`, } as const diff --git a/src/postgres/store.ts b/src/postgres/store.ts index f78829f..f21fa94 100644 --- a/src/postgres/store.ts +++ b/src/postgres/store.ts @@ -157,25 +157,31 @@ export class PostgresStore implements ActivityStore { async postsByAuthor(authorId: UserId, page?: Page): Promise { const id = await this.requireUser(authorId) - const [at, cursorId] = cursorParams(page) - const { rows } = await this.sql.query(SQL.authorTimeline, [ - id, - at, - cursorId, - pageLimit(page), - ]) + const limit = pageLimit(page) + const before = page?.before + const { rows } = before + ? await this.sql.query(SQL.authorTimelineBefore, [ + id, + before.createdAt, + before.id, + limit, + ]) + : await this.sql.query(SQL.authorTimeline, [id, limit]) return rows.map(toPost) } async feed(userId: UserId, page?: Page): Promise { const id = await this.requireUser(userId) - const [at, cursorId] = cursorParams(page) - const { rows } = await this.sql.query(SQL.feed, [ - id, - at, - cursorId, - pageLimit(page), - ]) + const limit = pageLimit(page) + const before = page?.before + const { rows } = before + ? await this.sql.query(SQL.feedBefore, [ + id, + before.createdAt, + before.id, + limit, + ]) + : await this.sql.query(SQL.feed, [id, limit]) return rows.map(toPost) } @@ -188,12 +194,6 @@ export class PostgresStore implements ActivityStore { } } -function cursorParams(page?: Page): [number | null, string | null] { - const before = page?.before - if (!before) return [null, null] - return [before.createdAt, before.id] -} - function epoch(value: string | number): number { return typeof value === 'number' ? value : Number(value) } diff --git a/test/postgres-store.test.ts b/test/postgres-store.test.ts index 29b21e5..03fd87d 100644 --- a/test/postgres-store.test.ts +++ b/test/postgres-store.test.ts @@ -62,13 +62,13 @@ describe('postgres schema and planner', () => { ('p2', 'bob', 'newer-id', 10), ('p3', 'bob', 'earlier', 4)`, ) - const first = await db.query<{ id: string }>(SQL.feed, ['ada', null, null, 1]) + const first = await db.query<{ id: string }>(SQL.feed, ['ada', 1]) expect(first.rows.map((row) => row.id)).toEqual(['p2']) - const second = await db.query<{ id: string }>(SQL.feed, ['ada', 10, 'p2', 2]) + const second = await db.query<{ id: string }>(SQL.feedBefore, ['ada', 10, 'p2', 2]) expect(second.rows.map((row) => row.id)).toEqual(['p1', 'p3']) }) - it('uses the author timeline index for the feed join when seq scans are off', async () => { + it('walks posts_author_timeline_idx as an index-ordered scan after ANALYZE', async () => { await db.query('TRUNCATE TABLE users CASCADE') await db.query( `INSERT INTO users (id, handle, created_at) VALUES ('ada', 'ada', 1), ('bob', 'bob', 1)`, @@ -76,15 +76,28 @@ describe('postgres schema and planner', () => { await db.query(`INSERT INTO follows (follower_id, followee_id) VALUES ('ada', 'bob')`) await db.query( `INSERT INTO posts (id, author_id, body, created_at) - SELECT 'p' || g, 'bob', 'body', g FROM generate_series(1, 40) AS g`, + SELECT 'p' || g, 'bob', 'body', g FROM generate_series(1, 500) AS g`, ) - await db.query('SET enable_seqscan = off') - const { rows } = await db.query<{ 'QUERY PLAN': string }>( - `EXPLAIN ${SQL.feed}`, - ['ada', null, null, 20], - ) - await db.query('SET enable_seqscan = on') - const plan = rows.map((row) => row['QUERY PLAN']).join('\n') - expect(plan).toMatch(/posts_author_timeline_idx/) + await db.query('ANALYZE') + + const timeline = await explain(SQL.authorTimeline, ['bob', 20]) + expect(timeline).toContain('Index Scan using posts_author_timeline_idx') + expect(timeline).not.toContain('Bitmap Index Scan') + expect(timeline).not.toMatch(/\bSort\b/) + + const keyed = await explain(SQL.authorTimelineBefore, ['bob', 250, 'p250', 20]) + expect(keyed).toContain('Index Scan using posts_author_timeline_idx') + expect(keyed).toContain('Index Cond:') + expect(keyed).not.toContain('Bitmap Index Scan') + expect(keyed).not.toMatch(/\bSort\b/) + + const feed = await explain(SQL.feed, ['ada', 20]) + expect(feed).toContain('Index Scan using posts_author_timeline_idx') + expect(feed).not.toContain('Bitmap Index Scan') }) }) + +async function explain(sql: string, params: unknown[]): Promise { + const { rows } = await db.query<{ 'QUERY PLAN': string }>(`EXPLAIN ${sql}`, params) + return rows.map((row) => row['QUERY PLAN']).join('\n') +}