prisma-store.ts

prisma-store.tsapps/realtime/src/prisma-store.tsraw ← src14,247 B← Back to the app
1/**
2 * Production seams for the realtime server (TASK-0091 notes 14/17):
3 *
4 *  - {@link PrismaDocStore} — the `DocStore` persistence seam over
5 *    `@prisma/client` (blocks table → the codec's `BlockLike`). The REST
6 *    `BlockModule` remains for non-collaborative writes; both serialize
7 *    on `pages.version` (doc §4.3 — the flush wins, the REST path sees
8 *    fresh state on retry with CONFLICT).
9 *  - {@link createPrismaMetaResolver} — `resourceId → workspaceId` +
10 *    the joining user's display name (note 14).
11 *  - {@link createPrismaPermissionResolver} — the effective-level
12 *    assembly (the API's `lib/authorize.ts` algorithm, run through the
13 *    SHARED `@factory/permissions` engine — the architecture doc's
14 *    "same engine as the API guard" requirement; the conflict between
15 *    the doc and the API's local `authorize.ts` copy is recorded in the
16 *    task file).
17 *
18 * `@prisma/client` is imported here (not in `server.ts`) so the server
19 * stays importable without a generated client (unit tests / dev).
20 */
21 
22import { PrismaClient, type Prisma, type BlockType as PrismaBlockType } from "@prisma/client";
23import {
24  resolvePermission,
25  roleFromPrisma,
26  type Grant,
27  type GranteeType,
28  type PermissionLevel,
29  type ResourceType,
30} from "@factory/permissions";
31import type { BlockLike } from "@factory/collaboration/codec";
32import { blockTypeSchema, type BlockType as CodecBlockType } from "@factory/shared";
33import type { DocStore, HydrateResult, PersistResult } from "./rooms.js";
34import type { PermissionResolver, RoomMetaResolver } from "./server.js";
35 
36type Level = PermissionLevel;
37 
38/**
39 * Block-type translation at the persistence seam.
40 *
41 * The Prisma `BlockType` enum is stored SCREAMING_SNAKE (`PARAGRAPH`)
42 * while the codec/DTO `BlockType` (PRODUCT-SPEC §4, `block.dto.ts`) is
43 * lowercase snake_case (`paragraph`) — a pre-existing cross-cutting
44 * inconsistency (the API's `block.service.ts` writes the DTO value
45 * straight to Prisma; recorded in the task file). The realtime store
46 * translates here at the one boundary where the codec doc meets the DB,
47 * keeping the Yjs codec untouched. The two value sets are 1:1 by case,
48 * so the mapping is a case fold; values outside the shared schema fail
49 * closed (the hydrating page is refused, not half-mapped).
50 */
51function dbToCodecType(raw: string): CodecBlockType {
52  const parsed = blockTypeSchema.safeParse(raw.toLowerCase());
53  if (!parsed.success) {
54    throw new Error(`PrismaDocStore: block type "${raw}" is not a known codec block type`);
55  }
56  return parsed.data;
57}
58 
59function codecToDbType(t: CodecBlockType): PrismaBlockType {
60  // The DB enum members are exactly the uppercase form of the codec
61  // values (1:1, verified against `prisma/schema.prisma` BlockType);
62  // `t` is a codec value by type, so the fold is always a valid member.
63  return t.toUpperCase() as PrismaBlockType;
64}
65 
66/** The `DocStore` seam backed by Postgres (blocks + pages tables). */
67export class PrismaDocStore implements DocStore {
68  constructor(private readonly prisma: PrismaClient) {}
69 
70  async hydrate(pageId: string): Promise<HydrateResult> {
71    const [page, rows] = await Promise.all([
72      this.prisma.page.findUnique({ where: { id: pageId }, select: { version: true } }),
73      this.prisma.block.findMany({
74        where: { pageId, deletedAt: null },
75        select: { id: true, parentId: true, type: true, position: true, content: true, properties: true },
76        orderBy: { position: "asc" },
77      }),
78    ]);
79    return {
80      blocks: rows.length === 0 ? null : this.toBlockLikes(rows),
81      // A missing page hydrates as empty at the schema default version
82      // (persist still throws — the page id is validated by the handshake
83      // room meta, this is a safety fallback, not a creation path).
84      version: page?.version ?? 1,
85    };
86  }
87 
88  /**
89   * Persist under the `pages.version` CAS (TASK-0092 R2, doc §4.3):
90   * `UPDATE pages SET version = version + 1 WHERE id = ? AND version =
91   * baseVersion`. Zero rows → a concurrent non-Yjs writer (REST batch)
92   * bumped the version; the persist is a no-op and the result carries the
93   * store's current blocks (the external state) so the room can rebase.
94   * The version bump is the FIRST statement of the transaction: if the CAS
95   * races a writer between our check and the update, the transaction
96   * aborts (nothing is written) and we report the conflict.
97   */
98  async persist(pageId: string, blocks: readonly BlockLike[], baseVersion: number): Promise<PersistResult> {
99    const page = await this.prisma.page.findUnique({
100      where: { id: pageId },
101      select: { workspaceId: true, version: true },
102    });
103    if (page === null) {
104      throw new Error(`PrismaDocStore.persist: page "${pageId}" not found`);
105    }
106    if (page.version !== baseVersion) {
107      // Fast path: the conflict is already visible before the transaction.
108      return { version: page.version, conflict: { externalBlocks: await this.readExternalBlocks(pageId) } };
109    }
110    try {
111      await this.prisma.$transaction(async (tx) => {
112        // The atomic CAS guard (the authoritative check — the pre-check
113        // above is an optimization, not the guard).
114        const cas = await tx.page.updateMany({
115          where: { id: pageId, version: baseVersion },
116          data: { version: baseVersion + 1, updatedAt: new Date() },
117        });
118        if (cas.count !== 1) {
119          // A concurrent writer won the race → abort the transaction.
120          throw new PersistConflictError();
121        }
122        const existing = await tx.block.findMany({
123          where: { pageId },
124          select: { id: true, deletedAt: true },
125        });
126        const liveIds = new Set(blocks.map((b) => b.id));
127        const removed = existing.filter((b) => b.deletedAt === null && !liveIds.has(b.id)).map((b) => b.id);
128        if (removed.length > 0) {
129          await tx.block.updateMany({
130            where: { id: { in: removed }, deletedAt: null },
131            data: { deletedAt: new Date() },
132          });
133        }
134        for (const b of blocks) {
135          await tx.block.upsert({
136            where: { id: b.id },
137            create: {
138              id: b.id,
139              pageId,
140              workspaceId: page.workspaceId,
141              parentId: b.parentId,
142              type: codecToDbType(b.type),
143              position: b.position,
144              content: b.content as Prisma.InputJsonValue,
145              properties: (b.properties ?? {}) as Prisma.InputJsonValue,
146            },
147            update: {
148              parentId: b.parentId,
149              type: codecToDbType(b.type),
150              position: b.position,
151              content: b.content as Prisma.InputJsonValue,
152              properties: (b.properties ?? {}) as Prisma.InputJsonValue,
153              deletedAt: null,
154            },
155          });
156        }
157      });
158    } catch (err) {
159      if (err instanceof PersistConflictError) {
160        return { version: await this.currentVersion(pageId), conflict: { externalBlocks: await this.readExternalBlocks(pageId) } };
161      }
162      throw err;
163    }
164    return { version: baseVersion + 1 };
165  }
166 
167  async ping(): Promise<boolean> {
168    try {
169      await this.prisma.$queryRaw`SELECT 1`;
170      return true;
171    } catch {
172      return false;
173    }
174  }
175 
176  /** The page's current version (conflict result — the room's new CAS base). */
177  private async currentVersion(pageId: string): Promise<number> {
178    const page = await this.prisma.page.findUnique({ where: { id: pageId }, select: { version: true } });
179    return page?.version ?? 1;
180  }
181 
182  /** The store's current live blocks (the external state for a rebase). */
183  private async readExternalBlocks(pageId: string): Promise<BlockLike[]> {
184    const rows = await this.prisma.block.findMany({
185      where: { pageId, deletedAt: null },
186      select: { id: true, parentId: true, type: true, position: true, content: true, properties: true },
187      orderBy: { position: "asc" },
188    });
189    return this.toBlockLikes(rows);
190  }
191 
192  private toBlockLikes(
193    rows: ReadonlyArray<{
194      id: string;
195      parentId: string | null;
196      type: PrismaBlockType | string;
197      position: string;
198      content: Prisma.JsonValue | null;
199      properties: Prisma.JsonValue | null;
200    }>,
201  ): BlockLike[] {
202    return rows.map((b) => ({
203      id: b.id,
204      // The DB `BlockType` enum is SCREAMING_SNAKE; the codec uses the
205      // PRODUCT-SPEC §4 lowercase snake_case — translated at this seam
206      // (see `dbToCodecType`).
207      type: dbToCodecType(b.type),
208      content: (b.content ?? { text: "" }) as BlockLike["content"],
209      parentId: b.parentId,
210      position: b.position,
211      ...(b.properties !== null && b.properties !== undefined && Object.keys(b.properties).length > 0
212        ? { properties: b.properties as Record<string, unknown> }
213        : {}),
214    }));
215  }
216}
217 
218/** Internal: the persist transaction's CAS guard failed (a concurrent writer). */
219class PersistConflictError extends Error {
220  constructor() {
221    super("pages.version CAS conflict (concurrent external writer)");
222  }
223}
224 
225/** `resourceId → workspaceId` + the joining user's display name. */
226export function createPrismaMetaResolver(prisma: PrismaClient): RoomMetaResolver {
227  return {
228    async resolve({ kind, id, userId }) {
229      if (kind === "user") {
230        // The control room needs no resource lookup (own-id is checked
231        // in the handshake); the name still rides the roster.
232        const user = await prisma.user.findUnique({ where: { id: userId }, select: { name: true, displayName: true } });
233        return user === null
234          ? null
235          : { workspaceId: "__own__", userName: user.displayName ?? user.name };
236      }
237      if (kind === "page" || kind === "comment") {
238        const page = await prisma.page.findUnique({
239          where: { id },
240          select: { workspaceId: true, deletedAt: true },
241        });
242        if (page === null || page.deletedAt !== null) return null;
243        const user = await prisma.user.findUnique({ where: { id: userId }, select: { name: true, displayName: true } });
244        return {
245          workspaceId: page.workspaceId,
246          userName: user === null ? undefined : (user.displayName ?? user.name),
247        };
248      }
249      // database
250      const db = await prisma.database.findUnique({
251        where: { id },
252        select: { workspaceId: true, isArchived: true },
253      });
254      if (db === null || db.isArchived) return null;
255      const user = await prisma.user.findUnique({ where: { id: userId }, select: { name: true, displayName: true } });
256      return {
257        workspaceId: db.workspaceId,
258        userName: user === null ? undefined : (user.displayName ?? user.name),
259      };
260    },
261  };
262}
263 
264/**
265 * The effective-level resolver — the API's `authorize.ts` algorithm
266 * (role baseline + the most-specific applicable grant, maxed) run
267 * THROUGH the shared `@factory/permissions` engine (the architecture
268 * doc's "same engine" rule; the conflict with the API's local copy is
269 * recorded in the task file).
270 */
271export function createPrismaPermissionResolver(prisma: PrismaClient): PermissionResolver {
272  const prismaLike = prisma;
273  return {
274    async resolveLevel({ userId, workspaceId, resourceType, resourceId }) {
275      const membership = await prismaLike.workspaceMember.findUnique({
276        where: { workspaceId_userId: { workspaceId, userId } },
277        select: { role: true },
278      });
279      // No membership at all → none (a shared/public page is Phase-14;
280      // the realtime handshake requires a token whose workspaces[]
281      // already passed, so a member is implied).
282      if (membership === null) return "none";
283 
284      // Teams FIRST (the grant query references the team ids — fetching
285      // both in one `Promise.all` would use `teams` before its
286      // declaration resolves). Then the applicable grants.
287      const teams = await prismaLike.teamMember.findMany({
288        where: { userId },
289        select: { teamId: true },
290      });
291      const grants = await prismaLike.permission.findMany({
292        where: {
293          workspaceId,
294          OR: [
295            { granteeType: "user", granteeId: userId },
296            { granteeType: "team", granteeId: { in: teams.map((t) => t.teamId) } },
297          ],
298        },
299      });
300 
301      // The grants that apply to this resource (workspace-scoped
302      // grants cover everything; page/database-scoped grants must
303      // match the target; teamspace/record/file do not apply to a
304      // room join — the API remains the source of truth for
305      // record-scoped writes).
306      const applicable: Grant[] = grants
307        .filter((g) => {
308          if (g.resourceType === "workspace") return g.resourceId === workspaceId;
309          if (g.resourceType === resourceType) return g.resourceId === resourceId;
310          return false;
311        })
312        .map((g): Grant | null => {
313          const level = grantLevel(g.level);
314          if (level === "none") return null;
315          return {
316            granteeType: g.granteeType as GranteeType,
317            granteeId: g.granteeId,
318            resourceType: g.resourceType as ResourceType,
319            resourceId: g.resourceId,
320            level,
321          };
322        })
323        .filter((g): g is Grant => g !== null);
324 
325      // The shared engine resolves role baseline + these grants +
326      // (no shares at room scope — shares are a Phase-14 sharing
327      // concern; the doc's join matrix is role + explicit grants).
328      const result = resolvePermission({
329        userId,
330        userTeams: teams.map((t) => t.teamId),
331        workspaceRole: roleFromPrisma(membership.role),
332        workspaceId,
333        resourceType: (resourceType === "page" ? "page" : "database") as ResourceType,
334        resourceId,
335        grants: applicable,
336        shares: [],
337      });
338      return result.level;
339    },
340  };
341}
342 
343function grantLevel(raw: string): Level {
344  switch (raw) {
345    case "NONE":
346      return "none";
347    case "VIEW":
348      return "view";
349    case "COMMENT":
350      return "comment";
351    case "EDIT":
352      return "edit";
353    case "FULL_ACCESS":
354      return "full_access";
355    default:
356      return "none";
357  }
358}
359 
360