feat(database,service,features): a sidebar section's or repository pin's write answers the transaction it committed in

packages/database/src/models/__tests__/task-section-store.test.ts

@@ -58,11 +58,17 @@ const openSharedWorkspace = Effect.gen(function* () {
   return { dean, charles, workspace, first, second };
 });
 
+/** A write the test expects taken, answering the sections as they then stand and the transaction it wrote them in. */
+const writtenWithTxid = (userId: string, write: TaskSectionWrite) =>
+  Effect.map(writeTaskSections(userId, write), (answer) =>
+    Option.getOrElse(answer, () => assert.fail("the write was refused")),
+  );
+
 /** A write the test expects taken, answering the sections as they then stand. */
 const written = (userId: string, write: TaskSectionWrite) =>
   Effect.map(
-    writeTaskSections(userId, write),
-    (answer): TaskSections => Option.getOrElse(answer, () => assert.fail("the write was refused")),
+    writtenWithTxid(userId, write),
+    ({ txid: _txid, ...sections }): TaskSections => sections,
   );
 
 const refused = (userId: string, write: TaskSectionWrite) =>
@@ -313,4 +319,77 @@ it.layer(getTestDB())("an account's task sections", (it) => {
       assert.deepEqual(yield* listTaskSections(dean), { sections: [], entries: [] });
     }),
   );
+
+  it.effect(
+    "a write that changes a row answers the transaction it wrote in, which the live sidebar waits for, and one that changes nothing answers none",
+    () =>
+      Effect.gen(function* () {
+        const { dean, workspace, first } = yield* openSharedWorkspace;
+        const txidOf = (write: TaskSectionWrite) =>
+          Effect.map(writtenWithTxid(dean, write), (answer) => answer.txid);
+        const isTxid = (txid: string | undefined) => txid !== undefined && /^\d+$/u.test(txid);
+        const launch = randomUUID();
+        const icon = { kind: TASK_SECTION_ICON_KIND.GLYPH, glyph: TASK_SECTION_GLYPH.ROCKET };
+        const edit: TaskSectionWrite = {
+          kind: TASK_SECTION_WRITE.EDIT,
+          id: launch,
+          title: "Ship",
+          icon,
+        };
+        const move: TaskSectionWrite = {
+          kind: TASK_SECTION_WRITE.MOVE,
+          id: launch,
+          position: "a0",
+        };
+        const remove: TaskSectionWrite = { kind: TASK_SECTION_WRITE.DELETE, id: launch };
+
+        const made = yield* txidOf(create(workspace, "Launch", launch));
+        const madeAgain = yield* txidOf(create(workspace, "Launch", launch));
+        const edited = yield* txidOf(edit);
+        const editedAgain = yield* txidOf(edit);
+        const moved = yield* txidOf(move);
+        const movedAgain = yield* txidOf(move);
+        const placed = yield* txidOf(place(first, launch));
+        const placedAgain = yield* txidOf(place(first, launch));
+        const placedAt = yield* txidOf(place(first, launch, "a1"));
+        const placedAtAgain = yield* txidOf(place(first, launch, "a1"));
+        const unplaced = yield* txidOf(place(first, null));
+        const unplacedAgain = yield* txidOf(place(first, null));
+        const deleted = yield* txidOf(remove);
+        const deletedAgain = yield* txidOf(remove);
+
+        const changed = [made, edited, moved, placed, placedAt, unplaced, deleted];
+        assert.ok(
+          changed.every(isTxid),
+          `every write that changed a row answers a txid: ${changed}`,
+        );
+        assert.equal(new Set(changed).size, changed.length, "each write is its own transaction");
+        assert.deepEqual(
+          [
+            madeAgain,
+            editedAgain,
+            movedAgain,
+            placedAgain,
+            placedAtAgain,
+            unplacedAgain,
+            deletedAgain,
+          ],
+          Array.from({ length: 7 }, () => undefined),
+        );
+
+        const pinned = yield* setRepositoryPin(dean, "acme/relay", true);
+        const pinnedAgain = yield* setRepositoryPin(dean, "acme/relay", true);
+        const unpinned = yield* setRepositoryPin(dean, "acme/relay", false);
+        const unpinnedAgain = yield* setRepositoryPin(dean, "acme/relay", false);
+        const txids = [pinned, pinnedAgain, unpinned, unpinnedAgain].map(
+          (answer) => Option.getOrThrow(answer).txid,
+        );
+        assert.ok(isTxid(txids[0]) && isTxid(txids[2]) && txids[0] !== txids[2]);
+        assert.deepEqual([txids[1], txids[3]], [undefined, undefined]);
+        assert.deepEqual(
+          Option.getOrThrow(pinnedAgain).pins.map((pin) => pin.repository),
+          ["acme/relay"],
+        );
+      }),
+  );
 });

packages/database/src/models/repository-pin-store.ts

@@ -1,5 +1,9 @@
-import { REPOSITORY_PIN_BOUNDS, type RepositoryPin } from "@luke/features/task-wire";
-import { and, asc, desc, eq } from "drizzle-orm";
+import {
+  REPOSITORY_PIN_BOUNDS,
+  type RepositoryPin,
+  type RepositoryPinsWritten,
+} from "@luke/features/task-wire";
+import { and, asc, desc, eq, sql } from "drizzle-orm";
 import { DateTime, Effect, Option, Schema } from "effect";
 import { SqlClient, SqlSchema } from "effect/sql";
 import { type LiveStatement, liveSelection, liveStatementOf } from "../core/live-statement.js";
@@ -27,6 +31,17 @@ const PinRowSchema = Schema.Struct({
 
 const PinKeySchema = Schema.Struct({ userId: Schema.String, repository: Schema.String });
 
+/**
+ * The id of the transaction that made a pin, read in the write's own
+ * `RETURNING`, as Neon's mutations guide does it: a follower of the sidebar
+ * waits for it to arrive on its live query. A write that changed no row
+ * returns none, so there is nothing to wait for.
+ */
+const TXID = { txid: sql<string>`pg_current_xact_id()::text` };
+
+/** The transaction a statement wrote its row in; no row where it changed none. */
+const WrittenSchema = Schema.Struct({ txid: Schema.String });
+
 const findPins = SqlSchema.findAll({
   Request: Schema.String,
   Result: PinRowSchema,
@@ -50,26 +65,30 @@ const lockAccount = SqlSchema.findOneOption({
     ),
 });
 
-const insertPin = SqlSchema.void({
+const insertPin = SqlSchema.findAll({
   Request: Schema.Struct({ ...PinKeySchema.fields, now: Schema.Date }),
+  Result: WrittenSchema,
   execute: (write) =>
     Database.use((db) =>
       db
         .insert(repositoryPin)
         .values({ userId: write.userId, repository: write.repository, pinnedAt: write.now })
-        .onConflictDoNothing(),
+        .onConflictDoNothing()
+        .returning(TXID),
     ),
 });
 
-const deletePin = SqlSchema.void({
+const deletePin = SqlSchema.findAll({
   Request: PinKeySchema,
+  Result: WrittenSchema,
   execute: (key) =>
     Database.use((db) =>
       db
         .delete(repositoryPin)
         .where(
           and(eq(repositoryPin.userId, key.userId), eq(repositoryPin.repository, key.repository)),
-        ),
+        )
+        .returning(TXID),
     ),
 });
 
@@ -101,30 +120,35 @@ export function repositoryPinsLiveStatement(userId: string): Option.Option<LiveS
 
 /**
  * Pins one repository or unpins it, and answers the account's pins as they
- * then stand; nothing where a new pin would hold more than the bound, or the
- * account is gone, with nothing written. A repository already pinned keeps
- * its instant and counts against no bound.
+ * then stand, with the transaction the pin changed a row in; nothing where
+ * a new pin would hold more than the bound, or the account is gone, with
+ * nothing written. A repository already pinned keeps its instant and counts
+ * against no bound, and a pin or unpin that changed no row answers no
+ * transaction, so a follower has nothing to wait for.
  */
 export function setRepositoryPin(
   userId: string,
   repository: string,
   pinned: boolean,
-): TaskStoreEffect<Option.Option<readonly RepositoryPin[]>> {
+): TaskStoreEffect<Option.Option<RepositoryPinsWritten>> {
   return Effect.flatMap(SqlClient.SqlClient, (client) =>
     client.withTransaction(
       Effect.gen(function* () {
         const account = yield* lockAccount(userId);
         if (Option.isNone(account)) return Option.none();
-        if (!pinned) {
-          yield* deletePin({ userId, repository });
-          return Option.some(yield* listRepositoryPins(userId));
-        }
+        const answer = (rows: readonly (typeof WrittenSchema.Type)[]) =>
+          Effect.map(listRepositoryPins(userId), (pins) => {
+            const txid = rows[0]?.txid;
+            return Option.some(txid === undefined ? { pins } : { pins, txid });
+          });
+        if (!pinned) return yield* answer(yield* deletePin({ userId, repository }));
         const standing = yield* listRepositoryPins(userId);
-        if (standing.some((pin) => pin.repository === repository)) return Option.some(standing);
+        if (standing.some((pin) => pin.repository === repository)) {
+          return Option.some({ pins: standing });
+        }
         if (standing.length >= REPOSITORY_PIN_BOUNDS.MAX_PINS) return Option.none();
         const now = yield* DateTime.nowAsDate;
-        yield* insertPin({ userId, repository, now });
-        return Option.some(yield* listRepositoryPins(userId));
+        return yield* answer(yield* insertPin({ userId, repository, now }));
       }),
     ),
   );