internal.service.ts

internal.service.tsapps/api/src/modules/internal/internal.service.tsraw ← internal13,906 B← Back to the app
1import { Inject, Injectable } from "@nestjs/common";
2import { Prisma, type BlockType as PrismaBlockType } from "@prisma/client";
3 
4/** JSON column input type (the `content` / `properties` Json fields). */
5type PrismaJsonValue = Prisma.InputJsonValue;
6import { PrismaService } from "../../common/prisma.service.js";
7import { AuditService } from "../../common/audit.service.js";
8import { EventBus } from "../../common/event-bus.js";
9import { ApiError } from "../../common/errors.js";
10import { EVENTS } from "../../constants.js";
11import {
12  type InternalBlock,
13  type InternalFlushInput,
14  type InternalReadQueryInput
15} from "./internal.dto.js";
16 
17/**
18 * Internal block endpoints (TASK-0092 slice 2).
19 *
20 * The service-to-service surface the realtime gateway uses to (a) hydrate a
21 * cold room (`internalRead`) and (b) persist — flush — a CRDT doc under the
22 * `pages.version` CAS (`internalFlush`). These are the **internal twin** of
23 * the public block endpoints: same tables, service-token auth, no
24 * permission re-check (the room's join matrix already gated the edits —
25 * the flush is a system write, attributed `actor: 'realtime-flush'`).
26 *
27 * They live behind `InternalServiceGuard` (shared `INTERNAL_SERVICE_TOKEN`,
28 * fail-closed) and are EXCLUDED from the global `api/v1` prefix in
29 * `main.ts` — the documented contract: `/internal/blocks/*`, never on the
30 * public route table.
31 *
32 * Block-type translation: the codec/DTO `type` is lowercase snake_case
33 * (PRODUCT-SPEC §4), the DB enum SCREAMING_SNAKE — the 1:1 case fold at
34 * this boundary (same resolution as the realtime `PrismaDocStore` seam).
35 */
36 
37/** The codec block type (lowercase) → the Prisma enum (uppercase). */
38function codecToDbType(t: string): PrismaBlockType {
39  return t.toUpperCase() as PrismaBlockType;
40}
41 
42/** The Prisma enum (uppercase) → the codec block type (lowercase). */
43function dbToCodecType(t: PrismaBlockType | string): string {
44  return t.toLowerCase();
45}
46 
47/** Canonical JSON equality (key-sorted stringify) — the blockDelta's
48 *  content/properties comparison, the codec `diffBlockSets` semantics. */
49function jsonEqual(a: unknown, b: unknown): boolean {
50  return canonicalJson(a) === canonicalJson(b);
51}
52 
53function canonicalJson(value: unknown): string {
54  if (value === null || typeof value !== "object") return JSON.stringify(value) ?? "null";
55  if (Array.isArray(value)) return `[${value.map((v) => canonicalJson(v)).join(",")}]`;
56  const entries = Object.entries(value as Record<string, unknown>)
57    .filter(([, v]) => v !== undefined)
58    .sort(([ka], [kb]) => (ka < kb ? -1 : ka > kb ? 1 : 0));
59  return `{${entries.map(([k, v]) => `${JSON.stringify(k)}:${canonicalJson(v)}`).join(",")}}`;
60}
61 
62interface BlockRow {
63  id: string;
64  parentId: string | null;
65  type: PrismaBlockType | string;
66  position: string;
67  content: PrismaJsonValue | null;
68  properties: PrismaJsonValue | null;
69}
70 
71/** The wire block shape (codec convention) for read responses / conflicts. */
72interface WireBlock {
73  id: string;
74  parentId: string | null;
75  type: string;
76  position: string;
77  content: Record<string, unknown>;
78  properties?: Record<string, unknown>;
79}
80 
81function toWireBlock(row: BlockRow): WireBlock {
82  const properties = row.properties as Record<string, unknown> | null;
83  return {
84    id: row.id,
85    parentId: row.parentId,
86    type: dbToCodecType(row.type),
87    position: row.position,
88    content: (row.content ?? {}) as Record<string, unknown>,
89    ...(properties !== null && properties !== undefined && Object.keys(properties).length > 0
90      ? { properties }
91      : {})
92  };
93}
94 
95@Injectable()
96export class InternalBlocksService {
97  // Plain parameter injection — matches PageService (the sibling that boots).
98  // The earlier explicit `@Inject(PrismaService)` token failed at boot
99  // ("Nest can't resolve dependencies of the InternalBlocksService (?…)")
100  // while the identical plain injection in PageService resolved fine, so this
101  // class now uses the same proven path. The unit test instantiates the ctor
102  // positionally, so it is unaffected. (TASK-0100 ledger: pre-existing API
103  // boot regression recorded in TASK-0095 S9; see also TestSeamService,
104  // which shares the root-level PrismaService/AuditService providers.)
105  constructor(
106    @Inject(PrismaService) private readonly prisma: PrismaService,
107    @Inject(AuditService) private readonly audit: AuditService,
108    @Inject(EventBus) private readonly events: EventBus
109  ) {}
110 
111  /**
112   * `GET /internal/blocks/:pageId` — the hydrator's read. Position-ordered
113   * blocks + the `pages.version` (the flush's CAS base), cursor-paginated
114   * by `position` (the 100k-block page streams per page — doc §4.2).
115   */
116  async internalRead(pageId: string, query: InternalReadQueryInput): Promise<{
117    pageId: string;
118    version: number;
119    blocks: WireBlock[];
120    nextCursor: string | null;
121    hasMore: boolean;
122  }> {
123    const pageSize = query.pageSize ?? 500;
124    const page = await this.prisma.page.findUnique({
125      where: { id: pageId },
126      select: { workspaceId: true, version: true, deletedAt: true }
127    });
128    if (page === null || page.deletedAt !== null) {
129      throw ApiError.notFound(`Page ${pageId} not found.`);
130    }
131    const where: Prisma.BlockWhereInput = {
132      pageId,
133      deletedAt: null,
134      ...(query.cursor ? { position: { gt: query.cursor } } : {})
135    };
136    const rows = await this.prisma.block.findMany({
137      where,
138      select: { id: true, parentId: true, type: true, position: true, content: true, properties: true },
139      orderBy: { position: "asc" },
140      take: pageSize + 1 // +1: the hasMore probe (no COUNT query)
141    });
142    const hasMore = rows.length > pageSize;
143    const pageRows = hasMore ? rows.slice(0, pageSize) : rows;
144    return {
145      pageId,
146      version: page.version,
147      blocks: pageRows.map(toWireBlock),
148      nextCursor: hasMore ? pageRows[pageRows.length - 1].position : null,
149      hasMore
150    };
151  }
152 
153  /**
154   * `POST /internal/blocks/flush` — the persister's write (doc §4.3).
155   *
156   * The full block set of the page (the CRDT doc re-encoded) persisted
157   * under the `pages.version` CAS:
158   *  - `baseVersion` === current version → the version bumps, the set is
159   *    upserted (removed rows soft-deleted), `page.updated` is emitted
160   *    (the `version` + `blockDelta` additive fields), the flush is
161   *    audited `actor: 'realtime-flush'`;
162   *  - otherwise a concurrent non-Yjs writer (REST batch) won — the flush
163   *    is a NO-OP (nothing is written) and the response carries the
164   *    store's current state as `conflict.externalBlocks` for the room to
165   *    rebase on (the "flush wins" rule, the 3-way merge in
166   *    `apps/realtime/src/rooms.ts`).
167   */
168  async internalFlush(input: InternalFlushInput): Promise<
169    | { version: number }
170    | { version: number; conflict: { externalBlocks: WireBlock[] } }
171  > {
172    const { pageId, baseVersion, blocks } = input;
173    const page = await this.prisma.page.findUnique({
174      where: { id: pageId },
175      select: { workspaceId: true, version: true, deletedAt: true }
176    });
177    if (page === null || page.deletedAt !== null) {
178      throw ApiError.notFound(`Page ${pageId} not found.`);
179    }
180    if (page.version !== baseVersion) {
181      // Fast path: the conflict is already visible before the transaction
182      // (the CAS below is the authoritative guard).
183      return {
184        version: page.version,
185        conflict: { externalBlocks: await this.readLiveBlocks(pageId) }
186      };
187    }
188    // The flush's minimal-diff basis: the state AT baseVersion (valid —
189    // the version equality means no writer landed since baseVersion).
190    const previous = await this.readLiveBlocks(pageId);
191 
192    try {
193      const newVersion = await this.prisma.$transaction(async (tx) => {
194        // The atomic CAS guard (first statement — a racy concurrent writer
195        // aborts the whole transaction, nothing is written).
196        const cas = await tx.page.updateMany({
197          where: { id: pageId, version: baseVersion },
198          data: { version: baseVersion + 1, updatedAt: new Date() }
199        });
200        if (cas.count !== 1) throw new FlushCasConflict();
201 
202        const existing = await tx.block.findMany({
203          where: { pageId },
204          select: { id: true, deletedAt: true }
205        });
206        const liveIds = new Set(blocks.map((b) => b.id));
207        const removed = existing
208          .filter((b) => b.deletedAt === null && !liveIds.has(b.id))
209          .map((b) => b.id);
210        if (removed.length > 0) {
211          await tx.block.updateMany({
212            where: { id: { in: removed }, deletedAt: null },
213            data: { deletedAt: new Date() }
214          });
215        }
216        // TASK-0135 N+1 fix (hot path #2 — the editor autosave flush, the
217        // profiled per-block `upsert` loop). All blocks are now upserted in
218        // ONE multi-row `INSERT … ON CONFLICT (id) DO UPDATE`, so the flush
219        // transaction's query count is CONSTANT (1 CAS + 1 read + ≤1
220        // soft-delete + 1 upsert) regardless of block count — flat across 10/
221        // 100/1000 blocks (asserted by the query-shape test). Semantics are
222        // identical to the per-block upsert: an absent id is created, a
223        // present id has its fields updated, a soft-deleted row is REVIVED
224        // (`deletedAt = NULL`). `updatedAt` is set explicitly — Prisma's
225        // `@updatedAt` does not apply to raw SQL, and the column has no DB
226        // default (NOT NULL).
227        const now = new Date();
228        const upsertRows = blocks.map(
229          (b) =>
230            Prisma.sql`(
231              ${b.id}, ${pageId}, ${page.workspaceId}, ${b.parentId ?? null},
232              ${codecToDbType(b.type)}::"BlockType", ${b.position},
233              ${JSON.stringify(b.content)}::jsonb, ${JSON.stringify(b.properties ?? {})}::jsonb, ${now}
234            )`
235        );
236        if (upsertRows.length > 0) {
237          await tx.$executeRaw`
238            INSERT INTO "blocks"
239              ("id", "pageId", "workspaceId", "parentId", "type", "position", "content", "properties", "updatedAt")
240            VALUES ${Prisma.join(upsertRows, ", ")}
241            ON CONFLICT ("id") DO UPDATE SET
242              "parentId" = EXCLUDED."parentId",
243              "type" = EXCLUDED."type",
244              "position" = EXCLUDED."position",
245              "content" = EXCLUDED."content",
246              "properties" = EXCLUDED."properties",
247              "deletedAt" = NULL,
248              "updatedAt" = ${now}
249          `;
250        }
251        return baseVersion + 1;
252      });
253 
254      const blockDelta = this.diffCounts(previous, blocks);
255      // Audit: the flush is a system write — attributed, not impersonable
256      // (the public surface has no path to set this actor; the service
257      // token is env-only).
258      await this.audit.write({
259        action: "block.flush",
260        resourceType: "block",
261        resourceId: pageId,
262        workspaceId: page.workspaceId,
263        pageId,
264        metadata: { actor: "realtime-flush", version: newVersion, blocks: blocks.length, ...blockDelta }
265      });
266      // The Phase-3 event, the `version` + `blockDelta` additive fields
267      // (packages/shared PageUpdatedPayload) — the Phase-8 indexer + the
268      // computed-recompute consumer both listen.
269      await this.events.publish(EVENTS.PAGE_UPDATED, {
270        workspaceId: page.workspaceId,
271        pageId,
272        version: newVersion,
273        blockDelta
274      });
275      return { version: newVersion };
276    } catch (err) {
277      if (err instanceof FlushCasConflict) {
278        // The racy CAS: read back the concurrent writer's state for the
279        // room's rebase (the fast path above misses this window).
280        const version = await this.prisma.page.findUnique({
281          where: { id: pageId },
282          select: { version: true }
283        });
284        return {
285          version: version?.version ?? baseVersion + 1,
286          conflict: { externalBlocks: await this.readLiveBlocks(pageId) }
287        };
288      }
289      throw err;
290    }
291  }
292 
293  /** The page's current live blocks (position-ordered, codec shape). */
294  private async readLiveBlocks(pageId: string): Promise<WireBlock[]> {
295    const rows = await this.prisma.block.findMany({
296      where: { pageId, deletedAt: null },
297      select: { id: true, parentId: true, type: true, position: true, content: true, properties: true },
298      orderBy: { position: "asc" }
299    });
300    return rows.map(toWireBlock);
301  }
302 
303  /** The flush's minimal op counts vs the last persisted state — the
304   *  codec `diffBlockSets` semantics (insert/update/move/delete counts). */
305  private diffCounts(
306    previous: WireBlock[],
307    next: readonly InternalBlock[]
308  ): { inserted: number; updated: number; moved: number; deleted: number } {
309    const prevById = new Map(previous.map((b) => [b.id, b] as const));
310    const nextById = new Map(next.map((b) => [b.id, b] as const));
311    let inserted = 0;
312    let updated = 0;
313    let moved = 0;
314    let deleted = 0;
315    for (const nb of next) {
316      const pb = prevById.get(nb.id);
317      if (pb === undefined) {
318        inserted += 1;
319        continue;
320      }
321      const contentChanged = !jsonEqual(pb.content, nb.content);
322      const prevProps = pb.properties ?? {};
323      const nextProps = nb.properties ?? {};
324      const propertiesChanged = !jsonEqual(prevProps, nextProps);
325      if (contentChanged || propertiesChanged) updated += 1;
326      if (pb.position !== nb.position || pb.parentId !== (nb.parentId ?? null)) moved += 1;
327    }
328    for (const pb of previous) {
329      if (!nextById.has(pb.id)) deleted += 1;
330    }
331    return { inserted, updated, moved, deleted };
332  }
333}
334 
335/** The flush's `pages.version` CAS guard failed inside the transaction. */
336class FlushCasConflict extends Error {
337  constructor() {
338    super("pages.version CAS conflict (concurrent external writer)");
339  }
340}
341