feat(service): the service's held reads and waits on a row are woken by Neon Realtime

apps/web/server/api-rewrites.json

@@ -395,10 +395,18 @@
     "src": "/api/issues",
     "dest": "/api/default.js?route=issues"
   },
+  {
+    "src": "/api/live/agent-inbox",
+    "dest": "/api/default.js?route=live/agent-inbox"
+  },
   {
     "src": "/api/live/inbox",
     "dest": "/api/default.js?route=live/inbox"
   },
+  {
+    "src": "/api/live/task-inbox",
+    "dest": "/api/default.js?route=live/task-inbox"
+  },
   {
     "src": "/api/mac-commands",
     "dest": "/api/default.js?route=mac-commands"

apps/web/server/routes/live/agent-inbox.ts

@@ -0,0 +1,6 @@
+import { resolveHostedUserId } from "@luke/service/auth/bearer";
+import { liveApp } from "@luke/service/live/live-app";
+import { routeFromHttpRouter } from "../../route-effect.js";
+
+/** One agent's inbox under `?id=`, sealed for following over Neon Realtime (GET). The logic lives in `packages/service/src/live/live-app.ts`; this file only hands it the deployment's real seams. */
+export default routeFromHttpRouter(liveApp({ resolveUserId: resolveHostedUserId }));

apps/web/server/routes/live/task-inbox.ts

@@ -0,0 +1,6 @@
+import { resolveHostedUserId } from "@luke/service/auth/bearer";
+import { liveApp } from "@luke/service/live/live-app";
+import { routeFromHttpRouter } from "../../route-effect.js";
+
+/** One task's inbox under `?id=`, sealed for following over Neon Realtime (GET). The logic lives in `packages/service/src/live/live-app.ts`; this file only hands it the deployment's real seams. */
+export default routeFromHttpRouter(liveApp({ resolveUserId: resolveHostedUserId }));

apps/web/tests/plan-comments.test.ts

@@ -41,6 +41,7 @@ import {
   CONNECTOR_ASKED_NOTHING,
   integrationConnectOf,
 } from "@luke/service/testing/integration-connect";
+import { quietLiveWake } from "@luke/service/testing/live-wake";
 import { unparsedWire } from "@luke/wire";
 import { readEither } from "@luke/wire/effect";
 import { Clock, type Context, Effect, Layer, Option, Result, Stream } from "effect";
@@ -125,6 +126,7 @@ const answer = (routes: "comments" | "plans", request: Request) =>
       NO_GITHUB,
       modelCatalogOf([]),
       integrationConnectOf(() => CONNECTOR_ASKED_NOTHING),
+      quietLiveWake,
     );
     let response: Response;
     if (routes === "comments") {

packages/database/src/core/live-statement.ts

@@ -10,8 +10,10 @@ import { Option, Schema } from "effect";
  * selects every column under its wire key (`liveSelection`), and an instant
  * as the epoch milliseconds the wire carries, so a row as Realtime delivers
  * it decodes under the same schema the held read answers with. Realtime takes
- * raw SQL with `$n` placeholders and its values beside it, which is what
- * Drizzle renders a statement to (`liveStatementOf`).
+ * the Drizzle select itself through its Drizzle adapter, which renders it to
+ * SQL with `$n` placeholders and its values beside it and refuses one whose
+ * selected key is not the name Postgres answers the column under
+ * (`liveStatementOf`).
  *
  * Note that a live read is deterministic, as Realtime requires: no `now()`,
  * no window measured from the clock. A read that needs one bounds it on the
@@ -21,15 +23,25 @@ import { Option, Schema } from "effect";
 /** One value of a live statement, as Realtime takes it: Postgres infers its type from where it stands. */
 type LiveParameter = string | number | boolean | null;
 
-/** A live statement: its SQL and the values its placeholders name, in order. */
+/**
+ * A Drizzle select as Realtime's Drizzle adapter seals it: the rows it is
+ * typed to answer, and the SQL and values it renders to.
+ */
+export interface LiveSelect {
+  readonly _: { readonly result: readonly unknown[] };
+  toSQL(): { readonly sql: string; readonly params: readonly unknown[] };
+}
+
+/**
+ * A live statement: its SQL and the values its placeholders name, in order,
+ * and the Drizzle select it was rendered from, which is what is sealed
+ * (`drizzleAdapter()`); a statement written as text alone, as a test's
+ * stand-in is, carries none and is sealed as that text.
+ */
 export interface LiveStatement {
   readonly sql: string;
   readonly params: readonly LiveParameter[];
-}
-
-/** What a Drizzle statement renders to, whichever builder made it. */
-interface RenderableStatement {
-  toSQL(): { readonly sql: string; readonly params: readonly unknown[] };
+  readonly query?: LiveSelect;
 }
 
 /**
@@ -80,10 +92,12 @@ const LiveValuesSchema = Schema.Array(
  * The statement as Realtime takes it; none where it names a value Realtime
  * cannot carry, which a caller answers as the read not being live.
  */
-export function liveStatementOf(statement: RenderableStatement): Option.Option<LiveStatement> {
+export function liveStatementOf(statement: LiveSelect): Option.Option<LiveStatement> {
   const { sql: text, params } = statement.toSQL();
   return Option.map(Schema.decodeUnknownOption(LiveValuesSchema)(params), (values) => ({
     sql: text,
     params: values.map((value) => (value instanceof Date ? value.toISOString() : value)),
+    // Note that a select binding an instant is sealed as its text, because the adapter encodes no `Date`.
+    ...(values.some((value) => value instanceof Date) ? undefined : { query: statement }),
   }));
 }

packages/database/src/models/__tests__/mac-command-store.test.ts

@@ -3,12 +3,14 @@ import { randomUUID } from "node:crypto";
 import { it } from "@effect/vitest";
 import type { MacCommandResult } from "@luke/features/mac-command-wire";
 import { Clock, Effect, Option } from "effect";
+import { SqlClient } from "effect/sql";
 import { Database } from "../../core/query.js";
 import { user } from "../../schemas/auth.js";
 import { getTestDB } from "../../testing/index.js";
 import {
   claimMacCommands,
   macCommandResult,
+  macCommandSettledLiveStatement,
   queueMacCommand,
   settleMacCommand,
 } from "../mac-command-store.js";
@@ -77,4 +79,25 @@ it.layer(getTestDB())("the Mac command store", (it) => {
         assert.deepEqual(yield* macCommandResult(other, queued.id), Option.none());
       }),
   );
+  it.effect(
+    "the statement Realtime wakes a waiting call on says the command settled once it does, and reads nothing of another account's",
+    () =>
+      Effect.gen(function* () {
+        const sql = yield* SqlClient.SqlClient;
+        const { owner, other } = yield* openAccounts;
+        const since = yield* now;
+        const queued = command();
+        yield* queueMacCommand(owner, queued);
+        const rows = (userId: string) => {
+          const statement = Option.getOrThrow(macCommandSettledLiveStatement(userId, queued.id));
+          return sql.unsafe(statement.sql, [...statement.params]);
+        };
+
+        assert.deepEqual(yield* rows(owner), [{ id: queued.id, settled: false }]);
+        yield* claimMacCommands(owner, since);
+        assert.ok(yield* settleMacCommand(owner, queued.id, RESULT));
+        assert.deepEqual(yield* rows(owner), [{ id: queued.id, settled: true }]);
+        assert.deepEqual(yield* rows(other), []);
+      }),
+  );
 });

packages/database/src/models/__tests__/plan-comments.test.ts

@@ -12,7 +12,8 @@ import {
   type PlanCommentStartRequest,
   type PlanCommentThread,
 } from "@luke/features/plan-comment-wire";
-import { Effect, Option } from "effect";
+import { Effect, Option, Schema } from "effect";
+import { SqlClient } from "effect/sql";
 import { Database } from "../../core/query.js";
 import { user } from "../../schemas/auth.js";
 import { getTestDB, joinWorkspace, openWorkspace } from "../../testing/index.js";
@@ -22,6 +23,8 @@ import {
   PLAN_COMMENT_OUTCOME,
   type PlanCommentOutcome,
   type PlanCommentWriterRef,
+  planCommentStamps,
+  planCommentStampsLiveStatement,
   readPlanCommentThreads,
   startPlanCommentThread,
 } from "../plan-comments.js";
@@ -158,6 +161,37 @@ it.layer(getTestDB())("a plan's comment threads", (it) => {
       }),
   );
 
+  it.effect(
+    "the statement Realtime wakes a held read of the threads on delivers where each thread stands as the held read's stamps say, to a member and to no one else",
+    () =>
+      Effect.gen(function* () {
+        const sql = yield* SqlClient.SqlClient;
+        const { owner, teammate, outsider, taskId } = yield* openSharedTask;
+        threadOf(
+          yield* startPlanCommentThread(
+            member(owner),
+            taskId,
+            startOn(PLAN_TEXT, "teammate", "Or a guest?"),
+          ),
+        );
+        const StampRows = Schema.Array(
+          Schema.Struct({
+            id: Schema.String,
+            revision: Schema.Number,
+            status: Schema.String,
+            updatedAt: Schema.Number,
+          }),
+        );
+        for (const reader of [owner, teammate, outsider]) {
+          const statement = Option.getOrThrow(planCommentStampsLiveStatement(reader, taskId));
+          const rows = yield* Schema.decodeUnknownEffect(StampRows)(
+            yield* sql.unsafe(statement.sql, [...statement.params]),
+          );
+          assert.deepEqual(rows, yield* planCommentStamps(reader, taskId));
+        }
+      }),
+  );
+
   it.effect("a thread sent again under its key is started once", () =>
     Effect.gen(function* () {
       const { owner, taskId } = yield* openSharedTask;

packages/database/src/models/board-store.ts

@@ -5,10 +5,11 @@ import {
   type DrawingElement,
   drawingElementsSchema,
 } from "@luke/features/board-wire";
-import { eq, sql } from "drizzle-orm";
+import { eq, isNotNull, sql } from "drizzle-orm";
 import { DateTime, Effect, Option, Schema } from "effect";
 import { SqlClient, SqlSchema } from "effect/sql";
-import { Database, type DatabaseServices } from "../core/query.js";
+import { type LiveStatement, liveSelection, liveStatementOf } from "../core/live-statement.js";
+import { Database, type DatabaseServices, queryBuilder } from "../core/query.js";
 import { task, taskBoard } from "../schemas/task.js";
 import { TaskArchived, type TaskStoreEffect, type TaskStoreFailure } from "./task-store.js";
 import { reachableTask } from "./workspace-store.js";
@@ -134,6 +135,32 @@ const findBoardImage = SqlSchema.findOneOption({
     ),
 });
 
+/**
+ * Where the plan's board stands for `readBoardImage`, written for Neon
+ * Realtime to wake a look on: the drawing numbers and whether an image is
+ * kept, never the image itself. None where it names a value Realtime cannot
+ * carry.
+ */
+export function boardImageLiveStatement(
+  userId: string,
+  taskId: string,
+): Option.Option<LiveStatement> {
+  return liveStatementOf(
+    queryBuilder
+      .select(
+        liveSelection({
+          taskId: task.id,
+          appliedDrawing: taskBoard.appliedDrawing,
+          drawingNumber: taskBoard.drawingNumber,
+          imaged: isNotNull(taskBoard.image),
+        }),
+      )
+      .from(task)
+      .leftJoin(taskBoard, eq(taskBoard.taskId, task.id))
+      .where(reachableTask(userId, taskId)),
+  );
+}
+
 /** Luke's drawing written whole, as the next number. */
 function upsertDrawing(drawing: readonly DrawingElement[]) {
   return SqlSchema.findOne({

packages/database/src/models/mac-command-store.ts

@@ -6,7 +6,8 @@ import {
 import { and, asc, eq, gte, isNotNull, isNull, type SQL } from "drizzle-orm";
 import { DateTime, Effect, Option, Schema } from "effect";
 import { SqlSchema } from "effect/sql";
-import { Database } from "../core/query.js";
+import { type LiveStatement, liveSelection, liveStatementOf } from "../core/live-statement.js";
+import { Database, queryBuilder } from "../core/query.js";
 import { macCommand, macCommandListener } from "../schemas/mac-command.js";
 import type { TaskStoreEffect } from "./task-store.js";
 
@@ -142,6 +143,23 @@ const findResult = SqlSchema.findOneOption({
     ),
 });
 
+/**
+ * Whether the account's command is settled, written for Neon Realtime to
+ * wake the call waiting on its result: its id and whether a result is kept,
+ * never the result itself. None where it names a value Realtime cannot carry.
+ */
+export function macCommandSettledLiveStatement(
+  userId: string,
+  id: string,
+): Option.Option<LiveStatement> {
+  return liveStatementOf(
+    queryBuilder
+      .select(liveSelection({ id: macCommand.id, settled: isNotNull(macCommand.result) }))
+      .from(macCommand)
+      .where(and(eq(macCommand.id, id), eq(macCommand.userId, userId))),
+  );
+}
+
 const deleteCommand = SqlSchema.void({
   Request: CommandKeySchema,
   execute: ({ userId, id }) =>

packages/database/src/models/plan-comments.ts

@@ -23,7 +23,8 @@ import { planSections } from "@luke/features/plan-document";
 import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
 import { DateTime, Effect, Option, Schema } from "effect";
 import { SqlClient, SqlSchema } from "effect/sql";
-import { Database } from "../core/query.js";
+import { type LiveStatement, liveSelection, liveStatementOf } from "../core/live-statement.js";
+import { Database, queryBuilder } from "../core/query.js";
 import { user } from "../schemas/auth.js";
 import { planComment, planCommentThread } from "../schemas/plan-comment.js";
 import { task } from "../schemas/task.js";
@@ -223,6 +224,33 @@ const findStamps = SqlSchema.findAll({
     ),
 });
 
+/**
+ * Where each of the plan's threads stands, as `planCommentStamps` reads it,
+ * written for Neon Realtime to wake a held read on. None where it names a
+ * value Realtime cannot carry.
+ */
+export function planCommentStampsLiveStatement(
+  userId: string,
+  taskId: string,
+): Option.Option<LiveStatement> {
+  return liveStatementOf(
+    queryBuilder
+      .select(
+        liveSelection(
+          {
+            id: planCommentThread.id,
+            revision: planCommentThread.revision,
+            status: planCommentThread.status,
+          },
+          { updatedAt: planCommentThread.updatedAt },
+        ),
+      )
+      .from(planCommentThread)
+      .innerJoin(task, eq(task.id, planCommentThread.taskId))
+      .where(and(eq(planCommentThread.taskId, taskId), taskReachableBy(userId))),
+  );
+}
+
 const countThreads = SqlSchema.findOne({
   Request: Schema.Struct({ taskId: Schema.String }),
   Result: Schema.Struct({ count: Schema.Number }),