From c194ef99d372f1a05ab3cecbaef59e39a06665cb Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Thu, 27 Aug 2026 21:38:06 +0000 Subject: [PATCH 1/2] feat: Add Neo4j graph backend with Cypher traversals Model the activity feed as a labeled property graph: User and Post nodes, directed FOLLOWS and AUTHORED edges, unique constraints, and the path queries (shortestPath, friend-of-friend, common neighbors) that the tabular stores express as recursive CTEs. --- README.md | 13 +++ src/index.ts | 2 + src/neo4j/cypher.ts | 32 +++++++ src/neo4j/memory.ts | 134 ++++++++++++++++++++++++++ src/neo4j/store.ts | 201 +++++++++++++++++++++++++++++++++++++++ test/neo4j-store.test.ts | 97 +++++++++++++++++++ 6 files changed, 479 insertions(+) create mode 100644 src/neo4j/cypher.ts create mode 100644 src/neo4j/memory.ts create mode 100644 src/neo4j/store.ts create mode 100644 test/neo4j-store.test.ts diff --git a/README.md b/README.md index 03c582d..f069e9b 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,12 @@ Polyglot persistence is the idea that one product rarely has one ideal database. - posts PK is hash-sharded, then the leftover unique on `posts(id)` from `ALTER PRIMARY KEY` is dropped so inserts do not keep appending to an unhashed unique - Secondary-index hotspots: a global `created_at` index is sequential even when the PK is hashed, so that index is the one that gets hashed +- Labeled property graph: nodes carry labels and properties; relationships are first-class directed typed edges (`FOLLOWS`, `AUTHORED`) +- Cypher pattern matching: `MATCH (u)-[:FOLLOWS]->(a)-[:AUTHORED]->(p)` is the 2-hop feed, not a join table plus a posts scan +- Variable-length paths and `shortestPath((a)-[:FOLLOWS*..16]->(b))` (BFS with a hop cap). A recursive CTE or N+1 neighbor lookup in the other stores +- Friend-of-friend recommendation: 2-hop `FOLLOWS` expansion, exclude self and existing edges, rank by independent path count +- Common neighbors and 2-cycles (`(a)-[:FOLLOWS]->(b)-[:FOLLOWS]->(a)`) as mutual follows +- Graph uniqueness: `CREATE CONSTRAINT ... REQUIRE n.prop IS UNIQUE` (`Neo.ClientError.Schema.ConstraintValidationFailed`). `MERGE` on `FOLLOWS` is idempotent; `CREATE` on `User`/`Post` fails the unique constraint ## What's implemented - Project scaffold with TypeScript strict mode, Vitest, and CI @@ -86,6 +92,13 @@ Polyglot persistence is the idea that one product rarely has one ideal database. - posts PK is hash-sharded, then the leftover unique on `posts(id)` from `ALTER PRIMARY KEY` is dropped so inserts do not keep appending to an unhashed unique - Secondary-index hotspots: a global `created_at` index is sequential even when the PK is hashed, so that index is the one that gets hashed - Distributed SQL backend (CockroachDB): same SQL, hash-sharded posts PK, ordered prefix scans, SERIALIZABLE retry +- Labeled property graph: nodes carry labels and properties; relationships are first-class directed typed edges (`FOLLOWS`, `AUTHORED`) +- Cypher pattern matching: `MATCH (u)-[:FOLLOWS]->(a)-[:AUTHORED]->(p)` is the 2-hop feed, not a join table plus a posts scan +- Variable-length paths and `shortestPath((a)-[:FOLLOWS*..16]->(b))` (BFS with a hop cap). A recursive CTE or N+1 neighbor lookup in the other stores +- Friend-of-friend recommendation: 2-hop `FOLLOWS` expansion, exclude self and existing edges, rank by independent path count +- Common neighbors and 2-cycles (`(a)-[:FOLLOWS]->(b)-[:FOLLOWS]->(a)`) as mutual follows +- Graph uniqueness: `CREATE CONSTRAINT ... REQUIRE n.prop IS UNIQUE` (`Neo.ClientError.Schema.ConstraintValidationFailed`). `MERGE` on `FOLLOWS` is idempotent; `CREATE` on `User`/`Post` fails the unique constraint +- Graph backend (Neo4j): model the relationships, Cypher queries the others struggle with ## Usage ```ts diff --git a/src/index.ts b/src/index.ts index 868aa1a..7835e03 100644 --- a/src/index.ts +++ b/src/index.ts @@ -65,3 +65,5 @@ export { SERIALIZATION_FAILURE, withSerializableRetry, } from './cockroach/store' + +export { CONSTRAINTS, CYPHER, MemoryGraph, Neo4jStore } from './neo4j/store' diff --git a/src/neo4j/cypher.ts b/src/neo4j/cypher.ts new file mode 100644 index 0000000..2fa63f5 --- /dev/null +++ b/src/neo4j/cypher.ts @@ -0,0 +1,32 @@ +export const CONSTRAINTS = [ + 'CREATE CONSTRAINT user_id IF NOT EXISTS FOR (u:User) REQUIRE u.id IS UNIQUE', + 'CREATE CONSTRAINT user_handle IF NOT EXISTS FOR (u:User) REQUIRE u.handle IS UNIQUE', + 'CREATE CONSTRAINT post_id IF NOT EXISTS FOR (p:Post) REQUIRE p.id IS UNIQUE', +] as const + +const USER = 'u.id AS id, u.handle AS handle, u.createdAt AS createdAt' +const POST = 'p.id AS id, a.id AS authorId, p.body AS body, p.createdAt AS createdAt' +const NEWEST = 'ORDER BY p.createdAt DESC, p.id DESC LIMIT $limit' +const BEFORE = + 'WHERE p.createdAt < $createdAt OR (p.createdAt = $createdAt AND p.id < $postId)' + +export const CYPHER = { + createUser: `CREATE (u:User {id: $id, handle: $handle, createdAt: $createdAt}) RETURN ${USER}`, + getUser: `MATCH (u:User {id: $id}) RETURN ${USER}`, + getUserByHandle: `MATCH (u:User {handle: $handle}) RETURN ${USER}`, + follow: `MATCH (a:User {id: $from}), (b:User {id: $to}) MERGE (a)-[:FOLLOWS]->(b)`, + unfollow: `MATCH (a:User {id: $from})-[r:FOLLOWS]->(b:User {id: $to}) DELETE r`, + isFollowing: `MATCH (a:User {id: $from})-[r:FOLLOWS]->(b:User {id: $to}) RETURN count(r) AS n`, + following: `MATCH (a:User {id: $id})-[:FOLLOWS]->(b:User) RETURN b.id AS id ORDER BY b.id`, + followers: `MATCH (a:User {id: $id})<-[:FOLLOWS]-(b:User) RETURN b.id AS id ORDER BY b.id`, + publish: `MATCH (a:User {id: $authorId}) CREATE (p:Post {id: $id, body: $body, createdAt: $createdAt}) CREATE (a)-[:AUTHORED]->(p) RETURN ${POST}`, + getPost: `MATCH (a:User)-[:AUTHORED]->(p:Post {id: $id}) RETURN ${POST}`, + authorTimeline: `MATCH (a:User {id: $authorId})-[:AUTHORED]->(p:Post) RETURN ${POST} ${NEWEST}`, + authorTimelineBefore: `MATCH (a:User {id: $authorId})-[:AUTHORED]->(p:Post) ${BEFORE} RETURN ${POST} ${NEWEST}`, + feed: `MATCH (u:User {id: $userId})-[:FOLLOWS]->(a:User)-[:AUTHORED]->(p:Post) RETURN ${POST} ${NEWEST}`, + feedBefore: `MATCH (u:User {id: $userId})-[:FOLLOWS]->(a:User)-[:AUTHORED]->(p:Post) ${BEFORE} RETURN ${POST} ${NEWEST}`, + shortestPath: `MATCH (a:User {id: $from}), (b:User {id: $to}) MATCH path = shortestPath((a)-[:FOLLOWS*..16]->(b)) RETURN [n IN nodes(path) | n.id] AS ids`, + recommend: `MATCH (u:User {id: $id})-[:FOLLOWS]->()-[:FOLLOWS]->(rec:User) WHERE rec.id <> $id AND NOT (u)-[:FOLLOWS]->(rec) RETURN rec.id AS id, count(*) AS score ORDER BY score DESC, rec.id LIMIT $limit`, + isMutual: `MATCH (a:User {id: $a})-[:FOLLOWS]->(b:User {id: $b})-[:FOLLOWS]->(a) RETURN count(*) > 0 AS mutual`, + commonFollowees: `MATCH (a:User {id: $a})-[:FOLLOWS]->(x:User)<-[:FOLLOWS]-(b:User {id: $b}) RETURN x.id AS id ORDER BY x.id`, +} as const diff --git a/src/neo4j/memory.ts b/src/neo4j/memory.ts new file mode 100644 index 0000000..6ae9d31 --- /dev/null +++ b/src/neo4j/memory.ts @@ -0,0 +1,134 @@ +export class ConstraintError extends Error { + readonly code = 'Neo.ClientError.Schema.ConstraintValidationFailed' + constructor(readonly constraint: string) { + super(constraint) + this.name = 'Neo4jError' + } +} + +export type Props = Record + +interface Node { + labels: string[] + props: Props +} + +function specOf(name: string): { label: string; prop: string } { + if (name === 'user_handle') return { label: 'User', prop: 'handle' } + if (name === 'post_id') return { label: 'Post', prop: 'id' } + return { label: 'User', prop: 'id' } +} + +export class MemoryGraph { + private readonly nodes = new Map() + private readonly out = new Map() + private readonly unique = new Map>() + private seq = 0 + + constrain(cypher: string): void { + const name = cypher.split(' ')[2] + if (name && !this.unique.has(name)) this.unique.set(name, new Map()) + } + + nodesWithLabel(label: string): Node[] { + return [...this.nodes.values()].filter((n) => n.labels.includes(label)) + } + + rels(type: string): { type: string; fromId: unknown; toId: unknown }[] { + const rows: { type: string; fromId: unknown; toId: unknown }[] = [] + for (const [from, list] of this.out) { + for (const rel of list) { + if (rel.type === type) { + rows.push({ + type, + fromId: this.nodes.get(from)?.props.id, + toId: this.nodes.get(rel.to)?.props.id, + }) + } + } + } + return rows + } + + createNode(labels: string[], props: Props): string { + for (const [name, seen] of this.unique) { + const spec = specOf(name) + if (!labels.includes(spec.label) || props[spec.prop] === undefined) continue + if (seen.has(String(props[spec.prop]))) throw new ConstraintError(name) + } + const id = `n${this.seq++}` + this.nodes.set(id, { labels, props: { ...props } }) + this.out.set(id, []) + for (const [name, seen] of this.unique) { + const spec = specOf(name) + if (labels.includes(spec.label) && props[spec.prop] !== undefined) { + seen.set(String(props[spec.prop]), id) + } + } + return id + } + + find(label: string, prop: string, value: unknown): string | undefined { + for (const [id, node] of this.nodes) { + if (node.labels.includes(label) && node.props[prop] === value) return id + } + return undefined + } + + props(id: string | undefined): Props | undefined { + return id === undefined ? undefined : this.nodes.get(id)?.props + } + + mergeRel(from: string, type: string, to: string): void { + const list = this.out.get(from) + if (!list || list.some((r) => r.type === type && r.to === to)) return + list.push({ type, to }) + } + + deleteRel(from: string, type: string, to: string): void { + const list = this.out.get(from) + if (list) this.out.set(from, list.filter((r) => !(r.type === type && r.to === to))) + } + + neighbors(from: string, type: string): string[] { + return (this.out.get(from) ?? []).filter((r) => r.type === type).map((r) => r.to) + } + + inbound(to: string, type: string): string[] { + const ids: string[] = [] + for (const [from, list] of this.out) { + if (list.some((r) => r.type === type && r.to === to)) ids.push(from) + } + return ids + } + + bfs(start: string, goal: string, type: string, maxHops: number): string[] | null { + if (start === goal) return null + const parent = new Map() + const dist = new Map([[start, 0]]) + const queue = [start] + for (let i = 0; i < queue.length; i++) { + const cur = queue[i] + if (cur === undefined) break + const d = dist.get(cur) ?? 0 + if (d >= maxHops) continue + for (const next of this.neighbors(cur, type)) { + if (dist.has(next)) continue + dist.set(next, d + 1) + parent.set(next, cur) + if (next === goal) { + const ids: string[] = [] + let at: string | undefined = goal + while (at) { + ids.push(String(this.nodes.get(at)?.props.id ?? '')) + if (at === start) break + at = parent.get(at) + } + return ids.reverse() + } + queue.push(next) + } + } + return null + } +} diff --git a/src/neo4j/store.ts b/src/neo4j/store.ts new file mode 100644 index 0000000..fab0e37 --- /dev/null +++ b/src/neo4j/store.ts @@ -0,0 +1,201 @@ +import { + comparePosts, + isBeforeCursor, + 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 { CONSTRAINTS } from './cypher' +import { MemoryGraph, type Props } from './memory' + +export { CONSTRAINTS, CYPHER } from './cypher' +export { ConstraintError, MemoryGraph } from './memory' + +export class Neo4jStore implements ActivityStore { + private constructor(private readonly g: MemoryGraph) {} + + static create(g: MemoryGraph): Neo4jStore { + for (const cypher of CONSTRAINTS) g.constrain(cypher) + return new Neo4jStore(g) + } + + async createUser(input: CreateUserInput, now = Date.now()): Promise { + const id = normalizeId(input.id) + const handle = normalizeHandle(input.handle) + try { + this.g.createNode(['User'], { id, handle, createdAt: now }) + } catch (err) { + throwMapped(err) + } + return { id, handle, createdAt: now } + } + + async getUser(id: UserId): Promise { + const key = tryNormalizeId(id) + if (!key) return null + return toUser(this.g.props(this.g.find('User', 'id', key))) + } + + async getUserByHandle(handle: string): Promise { + const key = tryNormalizeHandle(handle) + if (!key) return null + return toUser(this.g.props(this.g.find('User', 'handle', key))) + } + + async follow(followerId: UserId, followeeId: UserId): Promise { + const from = this.requireUser(followerId) + const to = this.requireUser(followeeId) + if (from === to) throw new StoreError('self_follow') + this.g.mergeRel(from, 'FOLLOWS', to) + } + + async unfollow(followerId: UserId, followeeId: UserId): Promise { + this.g.deleteRel(this.requireUser(followerId), 'FOLLOWS', this.requireUser(followeeId)) + } + + async isFollowing(followerId: UserId, followeeId: UserId): Promise { + return this.g + .neighbors(this.requireUser(followerId), 'FOLLOWS') + .includes(this.requireUser(followeeId)) + } + + async following(userId: UserId): Promise { + return ids(this.g, this.g.neighbors(this.requireUser(userId), 'FOLLOWS')) + } + + async followers(userId: UserId): Promise { + return ids(this.g, this.g.inbound(this.requireUser(userId), 'FOLLOWS')) + } + + async publish(input: PublishInput, now = Date.now()): Promise { + const id = normalizeId(input.id) + const author = this.requireUser(input.authorId) + const body = normalizeBody(input.body) + let post: string + try { + post = this.g.createNode(['Post'], { id, body, createdAt: now }) + } catch (err) { + throwMapped(err) + } + this.g.mergeRel(author, 'AUTHORED', post) + return { id, authorId: String(this.g.props(author)?.id), body, createdAt: now } + } + + async getPost(id: PostId): Promise { + const key = tryNormalizeId(id) + if (!key) return null + const post = this.g.find('Post', 'id', key) + const author = post ? this.g.inbound(post, 'AUTHORED')[0] : undefined + return toPost(this.g.props(author), this.g.props(post)) + } + + async postsByAuthor(authorId: UserId, page?: Page): Promise { + return pagePosts(authored(this.g, this.requireUser(authorId)), page) + } + + async feed(userId: UserId, page?: Page): Promise { + const posts: Post[] = [] + for (const followee of this.g.neighbors(this.requireUser(userId), 'FOLLOWS')) { + posts.push(...authored(this.g, followee)) + } + return pagePosts(posts, page) + } + + async shortestFollowPath(fromId: UserId, toId: UserId): Promise { + // WHY: unbounded * is a full-graph BFS; 16 hops is the lab cap. + return this.g.bfs(this.requireUser(fromId), this.requireUser(toId), 'FOLLOWS', 16) + } + + async recommendFollows(userId: UserId, limit = 10): Promise<{ id: UserId; score: number }[]> { + const from = this.requireUser(userId) + const direct = new Set(this.g.neighbors(from, 'FOLLOWS')) + const scores = new Map() + for (const hop of direct) { + for (const rec of this.g.neighbors(hop, 'FOLLOWS')) { + if (rec === from || direct.has(rec)) continue + scores.set(rec, (scores.get(rec) ?? 0) + 1) + } + } + return [...scores.entries()] + .map(([id, score]) => ({ id: String(this.g.props(id)?.id), score })) + .sort((a, b) => b.score - a.score || a.id.localeCompare(b.id)) + .slice(0, pageLimit({ limit })) + } + + async isMutual(aId: UserId, bId: UserId): Promise { + const a = this.requireUser(aId) + const b = this.requireUser(bId) + return this.g.neighbors(a, 'FOLLOWS').includes(b) && this.g.neighbors(b, 'FOLLOWS').includes(a) + } + + async commonFollowees(aId: UserId, bId: UserId): Promise { + const other = new Set(this.g.neighbors(this.requireUser(bId), 'FOLLOWS')) + return ids(this.g, this.g.neighbors(this.requireUser(aId), 'FOLLOWS').filter((id) => other.has(id))) + } + + private requireUser(id: UserId): string { + const key = tryNormalizeId(id) + const node = key ? this.g.find('User', 'id', key) : undefined + if (!node) throw new StoreError('user_not_found') + return node + } +} + +function ids(g: MemoryGraph, nodes: string[]): UserId[] { + return nodes + .map((id) => String(g.props(id)?.id)) + .sort((a, b) => a.localeCompare(b)) +} + +function authored(g: MemoryGraph, author: string): Post[] { + const a = g.props(author) + if (!a) return [] + const out: Post[] = [] + for (const post of g.neighbors(author, 'AUTHORED')) { + const row = toPost(a, g.props(post)) + if (row) out.push(row) + } + return out +} + +function pagePosts(posts: Post[], page?: Page): Post[] { + return posts + .filter((post) => isBeforeCursor(post, page?.before)) + .sort(comparePosts) + .slice(0, pageLimit(page)) +} + +function toUser(props: Props | undefined): User | null { + if (!props) return null + return { id: String(props.id), handle: String(props.handle), createdAt: Number(props.createdAt) } +} + +function toPost(author: Props | undefined, post: Props | undefined): Post | null { + if (!author || !post) return null + return { + id: String(post.id), + authorId: String(author.id), + body: String(post.body), + createdAt: Number(post.createdAt), + } +} + +function throwMapped(err: unknown): never { + const rec = typeof err === 'object' && err !== null ? (err as { code?: unknown; constraint?: unknown }) : undefined + if (rec?.code === 'Neo.ClientError.Schema.ConstraintValidationFailed') { + if (rec.constraint === 'user_id') throw new StoreError('user_exists') + if (rec.constraint === 'user_handle') throw new StoreError('handle_taken') + if (rec.constraint === 'post_id') throw new StoreError('post_exists') + } + throw err +} diff --git a/test/neo4j-store.test.ts b/test/neo4j-store.test.ts new file mode 100644 index 0000000..170bbd8 --- /dev/null +++ b/test/neo4j-store.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, it } from 'vitest' +import { StoreError } from '../src/domain' +import { ConstraintError, CYPHER, MemoryGraph, Neo4jStore } from '../src/neo4j/store' +import { defineStoreContract } from './contract' + +defineStoreContract('neo4j', () => Neo4jStore.create(new MemoryGraph())) + +describe('neo4j property graph', () => { + it('stores users and posts as nodes and follow/authored as directed edges', async () => { + const g = new MemoryGraph() + const store = await Neo4jStore.create(g) + await store.createUser({ id: 'ada', handle: 'ada' }, 1) + await store.createUser({ id: 'bob', handle: 'bob' }, 1) + await store.follow('ada', 'bob') + await store.publish({ id: 'p1', authorId: 'bob', body: 'hi' }, 4) + + expect(g.nodesWithLabel('User').map((n) => n.props.id).sort()).toEqual(['ada', 'bob']) + expect(g.nodesWithLabel('Post').map((n) => n.props)).toEqual([ + { id: 'p1', body: 'hi', createdAt: 4 }, + ]) + expect(g.rels('FOLLOWS')).toEqual([{ type: 'FOLLOWS', fromId: 'ada', toId: 'bob' }]) + expect(g.rels('AUTHORED')).toEqual([{ type: 'AUTHORED', fromId: 'bob', toId: 'p1' }]) + }) + + it('walks shortest FOLLOWS paths that SQL would need a recursive CTE for', async () => { + const store = await Neo4jStore.create(new MemoryGraph()) + await seed(store, ['ada', 'bob', 'cam', 'dan']) + await store.follow('ada', 'bob') + await store.follow('bob', 'cam') + await store.follow('cam', 'dan') + await store.follow('ada', 'cam') + expect(await store.shortestFollowPath('ada', 'dan')).toEqual(['ada', 'cam', 'dan']) + expect(await store.shortestFollowPath('ada', 'bob')).toEqual(['ada', 'bob']) + expect(await store.shortestFollowPath('dan', 'ada')).toBeNull() + expect(await store.shortestFollowPath('ada', 'ada')).toBeNull() + expect(CYPHER.shortestPath).toContain('shortestPath((a)-[:FOLLOWS*..16]->(b))') + }) + + it('ranks friend-of-friend recommendations by independent 2-hop paths', async () => { + const store = await Neo4jStore.create(new MemoryGraph()) + await seed(store, ['ada', 'bob', 'cam', 'dan', 'eve']) + await store.follow('ada', 'bob') + await store.follow('ada', 'cam') + await store.follow('bob', 'dan') + await store.follow('cam', 'dan') + await store.follow('bob', 'eve') + await store.follow('ada', 'eve') + expect(await store.recommendFollows('ada')).toEqual([{ id: 'dan', score: 2 }]) + expect(await store.recommendFollows('eve')).toEqual([]) + expect(CYPHER.recommend).toContain('NOT (u)-[:FOLLOWS]->(rec)') + }) + + it('matches a 2-cycle as mutual and intersection as common followees', async () => { + const store = await Neo4jStore.create(new MemoryGraph()) + await seed(store, ['ada', 'bob', 'cam', 'dan']) + await store.follow('ada', 'bob') + await store.follow('bob', 'ada') + await store.follow('ada', 'cam') + await store.follow('bob', 'cam') + await store.follow('bob', 'dan') + expect(await store.isMutual('ada', 'bob')).toBe(true) + expect(await store.isMutual('ada', 'cam')).toBe(false) + expect(await store.commonFollowees('ada', 'bob')).toEqual(['cam']) + expect(await store.commonFollowees('ada', 'dan')).toEqual([]) + }) + + it('maps a driver-shaped unique constraint to handle_taken', async () => { + const g = new MemoryGraph() + const store = Neo4jStore.create(g) + g.createNode = () => { + throw { + code: 'Neo.ClientError.Schema.ConstraintValidationFailed', + constraint: 'user_handle', + } + } + await expect(store.createUser({ id: 'ada', handle: 'ada' }, 1)).rejects.toMatchObject({ + constructor: StoreError, + code: 'handle_taken', + }) + }) + + it('rejects a duplicate handle at the User.handle constraint', async () => { + const g = new MemoryGraph() + const store = Neo4jStore.create(g) + await store.createUser({ id: 'u1', handle: 'ada' }, 1) + await expect(store.createUser({ id: 'u2', handle: 'ada' }, 2)).rejects.toMatchObject({ + code: 'handle_taken', + }) + expect(() => g.createNode(['User'], { id: 'u3', handle: 'ada', createdAt: 3 })).toThrow( + ConstraintError, + ) + }) +}) + +async function seed(store: Neo4jStore, ids: string[]): Promise { + for (const id of ids) await store.createUser({ id, handle: id }, 1) +} From 85f7d5f2dda47b8d98e0397dfd311cd8aa89be82 Mon Sep 17 00:00:00 2001 From: Thomas Hart Date: Thu, 27 Aug 2026 21:53:47 +0000 Subject: [PATCH 2/2] fix: map driver constraint messages and pin Cypher catalog Parse neo4j-driver ConstraintValidationFailed messages (no .constraint), cap shortest FOLLOWS BFS at 16 hops with a 17-edge chain, and pin the driver Cypher strings including WHERE a <> b on follow. --- README.md | 1 + src/neo4j/cypher.ts | 2 +- src/neo4j/memory.ts | 37 ++++++++------- src/neo4j/store.ts | 21 ++++++--- test/neo4j-store.test.ts | 97 ++++++++++++++++++++++++++++++++++++---- 5 files changed, 128 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index f069e9b..9081c11 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,7 @@ Polyglot persistence is the idea that one product rarely has one ideal database. - Common neighbors and 2-cycles (`(a)-[:FOLLOWS]->(b)-[:FOLLOWS]->(a)`) as mutual follows - Graph uniqueness: `CREATE CONSTRAINT ... REQUIRE n.prop IS UNIQUE` (`Neo.ClientError.Schema.ConstraintValidationFailed`). `MERGE` on `FOLLOWS` is idempotent; `CREATE` on `User`/`Post` fails the unique constraint - Graph backend (Neo4j): model the relationships, Cypher queries the others struggle with +- Graph backend (Neo4j-style `MemoryGraph`): model the relationships, plus the Cypher catalog for the walks the others struggle with ## Usage ```ts diff --git a/src/neo4j/cypher.ts b/src/neo4j/cypher.ts index 2fa63f5..dbae58a 100644 --- a/src/neo4j/cypher.ts +++ b/src/neo4j/cypher.ts @@ -14,7 +14,7 @@ export const CYPHER = { createUser: `CREATE (u:User {id: $id, handle: $handle, createdAt: $createdAt}) RETURN ${USER}`, getUser: `MATCH (u:User {id: $id}) RETURN ${USER}`, getUserByHandle: `MATCH (u:User {handle: $handle}) RETURN ${USER}`, - follow: `MATCH (a:User {id: $from}), (b:User {id: $to}) MERGE (a)-[:FOLLOWS]->(b)`, + follow: `MATCH (a:User {id: $from}), (b:User {id: $to}) WHERE a <> b MERGE (a)-[:FOLLOWS]->(b)`, unfollow: `MATCH (a:User {id: $from})-[r:FOLLOWS]->(b:User {id: $to}) DELETE r`, isFollowing: `MATCH (a:User {id: $from})-[r:FOLLOWS]->(b:User {id: $to}) RETURN count(r) AS n`, following: `MATCH (a:User {id: $id})-[:FOLLOWS]->(b:User) RETURN b.id AS id ORDER BY b.id`, diff --git a/src/neo4j/memory.ts b/src/neo4j/memory.ts index 6ae9d31..0008d72 100644 --- a/src/neo4j/memory.ts +++ b/src/neo4j/memory.ts @@ -13,21 +13,29 @@ interface Node { props: Props } -function specOf(name: string): { label: string; prop: string } { - if (name === 'user_handle') return { label: 'User', prop: 'handle' } - if (name === 'post_id') return { label: 'Post', prop: 'id' } - return { label: 'User', prop: 'id' } +function specOf(cypher: string): { name: string; label: string; prop: string } | undefined { + const m = /CONSTRAINT\s+(\w+)\s+.*FOR\s+\(\w+:(\w+)\)\s+REQUIRE\s+\w+\.(\w+)/.exec(cypher) + if (!m?.[1] || !m[2] || !m[3]) return undefined + return { name: m[1], label: m[2], prop: m[3] } +} + +interface UniqueIndex { + label: string + prop: string + seen: Map } export class MemoryGraph { private readonly nodes = new Map() private readonly out = new Map() - private readonly unique = new Map>() + private readonly unique = new Map() private seq = 0 constrain(cypher: string): void { - const name = cypher.split(' ')[2] - if (name && !this.unique.has(name)) this.unique.set(name, new Map()) + const spec = specOf(cypher) + if (spec && !this.unique.has(spec.name)) { + this.unique.set(spec.name, { label: spec.label, prop: spec.prop, seen: new Map() }) + } } nodesWithLabel(label: string): Node[] { @@ -51,18 +59,16 @@ export class MemoryGraph { } createNode(labels: string[], props: Props): string { - for (const [name, seen] of this.unique) { - const spec = specOf(name) - if (!labels.includes(spec.label) || props[spec.prop] === undefined) continue - if (seen.has(String(props[spec.prop]))) throw new ConstraintError(name) + for (const [name, index] of this.unique) { + if (!labels.includes(index.label) || props[index.prop] === undefined) continue + if (index.seen.has(String(props[index.prop]))) throw new ConstraintError(name) } const id = `n${this.seq++}` this.nodes.set(id, { labels, props: { ...props } }) this.out.set(id, []) - for (const [name, seen] of this.unique) { - const spec = specOf(name) - if (labels.includes(spec.label) && props[spec.prop] !== undefined) { - seen.set(String(props[spec.prop]), id) + for (const index of this.unique.values()) { + if (labels.includes(index.label) && props[index.prop] !== undefined) { + index.seen.set(String(props[index.prop]), id) } } return id @@ -80,6 +86,7 @@ export class MemoryGraph { } mergeRel(from: string, type: string, to: string): void { + if (type === 'FOLLOWS' && from === to) throw new Error('follows_no_self') const list = this.out.get(from) if (!list || list.some((r) => r.type === type && r.to === to)) return list.push({ type, to }) diff --git a/src/neo4j/store.ts b/src/neo4j/store.ts index fab0e37..3f7708e 100644 --- a/src/neo4j/store.ts +++ b/src/neo4j/store.ts @@ -128,7 +128,7 @@ export class Neo4jStore implements ActivityStore { } return [...scores.entries()] .map(([id, score]) => ({ id: String(this.g.props(id)?.id), score })) - .sort((a, b) => b.score - a.score || a.id.localeCompare(b.id)) + .sort((a, b) => b.score - a.score || (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)) .slice(0, pageLimit({ limit })) } @@ -152,9 +152,7 @@ export class Neo4jStore implements ActivityStore { } function ids(g: MemoryGraph, nodes: string[]): UserId[] { - return nodes - .map((id) => String(g.props(id)?.id)) - .sort((a, b) => a.localeCompare(b)) + return nodes.map((id) => String(g.props(id)?.id)).sort() } function authored(g: MemoryGraph, author: string): Post[] { @@ -191,11 +189,24 @@ function toPost(author: Props | undefined, post: Props | undefined): Post | null } function throwMapped(err: unknown): never { - const rec = typeof err === 'object' && err !== null ? (err as { code?: unknown; constraint?: unknown }) : undefined + const rec = + typeof err === 'object' && err !== null + ? (err as { code?: unknown; constraint?: unknown; message?: unknown }) + : undefined if (rec?.code === 'Neo.ClientError.Schema.ConstraintValidationFailed') { if (rec.constraint === 'user_id') throw new StoreError('user_exists') if (rec.constraint === 'user_handle') throw new StoreError('handle_taken') if (rec.constraint === 'post_id') throw new StoreError('post_exists') + const mapped = mapDriverMessage(typeof rec.message === 'string' ? rec.message : '') + if (mapped) throw new StoreError(mapped) } throw err } + +function mapDriverMessage(message: string): 'user_exists' | 'handle_taken' | 'post_exists' | undefined { + const m = /label `(\w+)` and property `(\w+)`/.exec(message) + if (m?.[1] === 'User' && m[2] === 'handle') return 'handle_taken' + if (m?.[1] === 'User' && m[2] === 'id') return 'user_exists' + if (m?.[1] === 'Post' && m[2] === 'id') return 'post_exists' + return undefined +} diff --git a/test/neo4j-store.test.ts b/test/neo4j-store.test.ts index 170bbd8..0be4dcc 100644 --- a/test/neo4j-store.test.ts +++ b/test/neo4j-store.test.ts @@ -33,21 +33,56 @@ describe('neo4j property graph', () => { expect(await store.shortestFollowPath('ada', 'bob')).toEqual(['ada', 'bob']) expect(await store.shortestFollowPath('dan', 'ada')).toBeNull() expect(await store.shortestFollowPath('ada', 'ada')).toBeNull() - expect(CYPHER.shortestPath).toContain('shortestPath((a)-[:FOLLOWS*..16]->(b))') + }) + + it('caps shortest FOLLOWS walks at 16 hops', async () => { + const store = Neo4jStore.create(new MemoryGraph()) + const chain = Array.from({ length: 18 }, (_, i) => `u${String(i).padStart(2, '0')}`) + await seed(store, chain) + for (let i = 0; i < 17; i++) { + const from = chain[i] + const to = chain[i + 1] + if (from && to) await store.follow(from, to) + } + expect(await store.shortestFollowPath('u00', 'u16')).toEqual(chain.slice(0, 17)) + expect(await store.shortestFollowPath('u00', 'u17')).toBeNull() }) it('ranks friend-of-friend recommendations by independent 2-hop paths', async () => { const store = await Neo4jStore.create(new MemoryGraph()) - await seed(store, ['ada', 'bob', 'cam', 'dan', 'eve']) + await seed(store, ['ada', 'bob', 'cam', 'dan', 'eve', 'fay', 'guy']) await store.follow('ada', 'bob') await store.follow('ada', 'cam') - await store.follow('bob', 'dan') - await store.follow('cam', 'dan') + await store.follow('ada', 'dan') await store.follow('bob', 'eve') - await store.follow('ada', 'eve') - expect(await store.recommendFollows('ada')).toEqual([{ id: 'dan', score: 2 }]) + await store.follow('bob', 'fay') + await store.follow('cam', 'fay') + await store.follow('cam', 'guy') + await store.follow('dan', 'guy') + expect(await store.recommendFollows('ada')).toEqual([ + { id: 'fay', score: 2 }, + { id: 'guy', score: 2 }, + { id: 'eve', score: 1 }, + ]) expect(await store.recommendFollows('eve')).toEqual([]) - expect(CYPHER.recommend).toContain('NOT (u)-[:FOLLOWS]->(rec)') + }) + + it('pins the Cypher catalog that a driver would run', () => { + expect(CYPHER.follow).toBe( + 'MATCH (a:User {id: $from}), (b:User {id: $to}) WHERE a <> b MERGE (a)-[:FOLLOWS]->(b)', + ) + expect(CYPHER.feed).toBe( + 'MATCH (u:User {id: $userId})-[:FOLLOWS]->(a:User)-[:AUTHORED]->(p:Post) RETURN p.id AS id, a.id AS authorId, p.body AS body, p.createdAt AS createdAt ORDER BY p.createdAt DESC, p.id DESC LIMIT $limit', + ) + expect(CYPHER.feedBefore).toBe( + 'MATCH (u:User {id: $userId})-[:FOLLOWS]->(a:User)-[:AUTHORED]->(p:Post) WHERE p.createdAt < $createdAt OR (p.createdAt = $createdAt AND p.id < $postId) RETURN p.id AS id, a.id AS authorId, p.body AS body, p.createdAt AS createdAt ORDER BY p.createdAt DESC, p.id DESC LIMIT $limit', + ) + expect(CYPHER.shortestPath).toBe( + 'MATCH (a:User {id: $from}), (b:User {id: $to}) MATCH path = shortestPath((a)-[:FOLLOWS*..16]->(b)) RETURN [n IN nodes(path) | n.id] AS ids', + ) + expect(CYPHER.recommend).toBe( + 'MATCH (u:User {id: $id})-[:FOLLOWS]->()-[:FOLLOWS]->(rec:User) WHERE rec.id <> $id AND NOT (u)-[:FOLLOWS]->(rec) RETURN rec.id AS id, count(*) AS score ORDER BY score DESC, rec.id LIMIT $limit', + ) }) it('matches a 2-cycle as mutual and intersection as common followees', async () => { @@ -64,13 +99,25 @@ describe('neo4j property graph', () => { expect(await store.commonFollowees('ada', 'dan')).toEqual([]) }) - it('maps a driver-shaped unique constraint to handle_taken', async () => { + it('maps a MemoryGraph unique constraint to handle_taken', async () => { + const g = new MemoryGraph() + const store = Neo4jStore.create(g) + g.createNode = () => { + throw new ConstraintError('user_handle') + } + await expect(store.createUser({ id: 'ada', handle: 'ada' }, 1)).rejects.toMatchObject({ + constructor: StoreError, + code: 'handle_taken', + }) + }) + + it('maps a driver Neo4jError message with no constraint field to handle_taken', async () => { const g = new MemoryGraph() const store = Neo4jStore.create(g) g.createNode = () => { throw { code: 'Neo.ClientError.Schema.ConstraintValidationFailed', - constraint: 'user_handle', + message: "Node(123) already exists with label `User` and property `handle` = 'ada'", } } await expect(store.createUser({ id: 'ada', handle: 'ada' }, 1)).rejects.toMatchObject({ @@ -79,6 +126,38 @@ describe('neo4j property graph', () => { }) }) + it('maps driver Neo4jError messages for User.id and Post.id', async () => { + const g = new MemoryGraph() + const store = Neo4jStore.create(g) + await store.createUser({ id: 'ada', handle: 'ada' }, 1) + g.createNode = () => { + throw { + code: 'Neo.ClientError.Schema.ConstraintValidationFailed', + message: "Node(1) already exists with label `User` and property `id` = 'ada'", + } + } + await expect(store.createUser({ id: 'ada', handle: 'other' }, 2)).rejects.toMatchObject({ + constructor: StoreError, + code: 'user_exists', + }) + g.createNode = () => { + throw { + code: 'Neo.ClientError.Schema.ConstraintValidationFailed', + message: "Node(2) already exists with label `Post` and property `id` = 'p1'", + } + } + await expect(store.publish({ id: 'p1', authorId: 'ada', body: 'hi' }, 3)).rejects.toMatchObject({ + constructor: StoreError, + code: 'post_exists', + }) + }) + + it('rejects a FOLLOWS self-loop at the graph even when the store is skipped', () => { + const g = new MemoryGraph() + const ada = g.createNode(['User'], { id: 'ada', handle: 'ada', createdAt: 1 }) + expect(() => g.mergeRel(ada, 'FOLLOWS', ada)).toThrow('follows_no_self') + }) + it('rejects a duplicate handle at the User.handle constraint', async () => { const g = new MemoryGraph() const store = Neo4jStore.create(g)