M2: the full job lifecycle — chat, hiring, geocoding, quotes, bookings, reviews

Closes the funnel. Before this the product could match two people and then stopped: `quotes`, `bookings` and `reviews` had tables and state machines and nothing that wrote a row, the entry deck's right swipe was wired to an empty handler, and every address resolved to the city centre. Jobs tab and chat - message router: thread, send, markRead, unreadTotal. A thread is a MATCH, not a job — one job with three interested pros is three private conversations. - Current/Past segments derived from ACTIVE_JOB_STATUSES, job detail listing the pros who accepted, and the conversation itself with attachments. Hiring from the deck - A right swipe on the entry deck opened nothing. It now resolves "which job?" through a sheet — sign in, pick an open job, or post one — and calls the same deck.swipe the per-job deck does, so the open-request cap and row lock apply exactly once. Swipes are vetoable so closing the sheet returns the card. Geocoding - ST_Distance and ST_DWithin rank and filter every deck, and both operands were placeholders. Addresses now resolve through Mapbox (permanent=true, which is what licenses storing the coordinates), the server resolves points rather than trusting client-supplied lat/lng, and every stored point records how it was obtained. A `city`-precision base cannot reach the verification queue. Quote -> booking -> review - The commercial chain, minus payments. Accepting a quote is the only place a booking is created; confirming completion is what unlocks reviews and moves the pro's completed_jobs. - Reviews publish double-blind with no sweeper: each is written with published_at already set to its embargo deadline and every read filters published_at <= now(), so it publishes itself. The second review pulls both forward. A silent counterparty cannot bury a bad review by never replying. State machine changes, both deliberate - booked -> matched: a cancelled booking is not a cancelled job. - scheduled -> awaiting_confirmation: in_progress is optional, so a pro who never tapped Start can still say the work is done. Test suite - api tests ran files in parallel against one database and failed roughly one run in three on whichever file lost the race. Serialised, and three fixtures that grabbed "the first client" pinned to the seeded accounts. Also includes work from a parallel session: admin verification queue, pro public profile and reviews read path, notification sending, denormalised stats recompute, search, and observability. 318 tests passing; typecheck and lint clean across 7 packages. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-21 06:29:59 -04:00
parent 8f3509d1dd
commit 974e312534
115 changed files with 19994 additions and 569 deletions
@@ -0,0 +1,280 @@
+import { TRPCError } from '@trpc/server';
+import { and, desc, eq, isNull, ne, or, sql } from 'drizzle-orm';
+import { z } from 'zod';
+import { schema, type Db } from '@linkder/db';
+import { PAST_JOB_STATUSES, sendMessageSchema, type JobStatus } from '@linkder/shared';
+import { protectedProcedure, router } from '../trpc';
+
+/**
+ * Chat between the two parties on a job.
+ *
+ * A thread is a MATCH, not a job. One job can have several pros accept it, and
+ * each of those is a separate private conversation — keying chat on the job
+ * would put three tradespeople in one room with the customer and each other.
+ *
+ * `matches` already carries `lastMessageAt` and `messages` is already indexed
+ * for both "this thread, newest last" and the unread badge. This router is the
+ * first thing to read or write either.
+ */
+
+/** A page of history. Thirty is about two phone screens. */
+const PAGE_SIZE = 30;
+
+/**
+ * A crude per-process send throttle.
+ *
+ * It is deliberately not a real rate limiter: it resets on deploy and does not
+ * span instances. What it does buy is that a runaway client or a held-down send
+ * button cannot write ten thousand rows before anyone notices. The real limiter
+ * belongs with the shared Redis in M4 — this is the floor until then.
+ */
+const SEND_WINDOW_MS = 60_000;
+const SEND_LIMIT = 30;
+const recentSends = new Map<string, number[]>();
+
+function assertSendRate(userId: string): void {
+  const now = Date.now();
+  const window = (recentSends.get(userId) ?? []).filter((at) => now - at < SEND_WINDOW_MS);
+  if (window.length >= SEND_LIMIT) {
+    throw new TRPCError({ code: 'TOO_MANY_REQUESTS', message: 'Slow down a moment.' });
+  }
+  window.push(now);
+  recentSends.set(userId, window);
+}
+
+type Tx = Parameters<Parameters<Db['transaction']>[0]>[0];
+type Executor = Db | Tx;
+
+export interface MatchContext {
+  matchId: string;
+  clientId: string;
+  proId: string;
+  jobId: string;
+  jobTitle: string;
+  jobStatus: JobStatus;
+}
+
+/**
+ * The authorization gate for everything in this file.
+ *
+ * Throws NOT_FOUND rather than FORBIDDEN for a match the caller is not part of —
+ * same rule as `job.byId` and `request.accept`: a stranger must not be able to
+ * probe whether a conversation exists.
+ *
+ * There is no admin bypass, unlike `job.byId`. An admin can already see the job,
+ * the booking and the money; reading a private conversation is a different kind
+ * of access and belongs behind a support flow that leaves an audit trail, not
+ * behind the same procedure the participants use.
+ */
+export async function requireMatchParticipant(
+  exec: Executor,
+  matchId: string,
+  userId: string,
+): Promise<MatchContext> {
+  const [row] = await exec
+    .select({
+      matchId: schema.matches.id,
+      clientId: schema.matches.clientId,
+      proId: schema.matches.proId,
+      jobId: schema.jobs.id,
+      jobTitle: schema.jobs.title,
+      jobStatus: schema.jobs.status,
+    })
+    .from(schema.matches)
+    .innerJoin(schema.jobs, eq(schema.jobs.id, schema.matches.jobId))
+    .where(eq(schema.matches.id, matchId));
+
+  if (!row || (row.clientId !== userId && row.proId !== userId)) {
+    throw new TRPCError({ code: 'NOT_FOUND', message: 'Conversation not found' });
+  }
+  return row;
+}
+
+/** History is closed once the job is. See PAST_JOB_STATUSES. */
+function canReply(status: JobStatus): boolean {
+  return !PAST_JOB_STATUSES.includes(status);
+}
+
+/**
+ * Keyset pagination, oldest-ward.
+ *
+ * The cursor is the id of the oldest message on the page, and the comparison
+ * resolves that row's `created_at` inside the query. Two reasons it is not a
+ * timestamp the caller carries:
+ *
+ *  - Postgres stores microseconds and a JS `Date` holds milliseconds, so a
+ *    round-tripped timestamp is truncated — and every message that landed later
+ *    in the same millisecond then falls the wrong side of `<` and is skipped.
+ *  - A cursor from another conversation resolves to NULL here, which yields an
+ *    empty page rather than a row from a thread the caller cannot see.
+ *
+ * `(created_at, id)` rather than `created_at` alone because a tie on the
+ * timestamp would otherwise drop a message or repeat one.
+ */
+const cursorSchema = z.string().uuid();
+
+export const messageRouter = router({
+  /**
+   * One thread: the header the screen needs, plus a page of messages oldest-first.
+   *
+   * Header and page come back together because the first render needs both, and
+   * a second round trip to learn whose conversation this is would leave the
+   * screen titleless for a beat.
+   */
+  thread: protectedProcedure
+    .input(z.object({ matchId: z.string().uuid(), cursor: cursorSchema.optional() }))
+    .query(async ({ ctx, input }) => {
+      const uid = ctx.session.userId;
+      const match = await requireMatchParticipant(ctx.db, input.matchId, uid);
+
+      const peerId = match.clientId === uid ? match.proId : match.clientId;
+      const [peer] = await ctx.db
+        .select({
+          id: schema.users.id,
+          name: schema.users.name,
+          image: schema.users.image,
+          role: schema.users.role,
+        })
+        .from(schema.users)
+        .where(eq(schema.users.id, peerId));
+
+      const rows = await ctx.db
+        .select({
+          id: schema.messages.id,
+          senderId: schema.messages.senderId,
+          body: schema.messages.body,
+          attachments: schema.messages.attachments,
+          readAt: schema.messages.readAt,
+          createdAt: schema.messages.createdAt,
+        })
+        .from(schema.messages)
+        .where(
+          and(
+            eq(schema.messages.matchId, input.matchId),
+            input.cursor
+              ? sql`(${schema.messages.createdAt}, ${schema.messages.id}) < (
+                  SELECT anchor.created_at, anchor.id FROM messages anchor
+                   WHERE anchor.id = ${input.cursor}::uuid
+                     AND anchor.match_id = ${input.matchId}::uuid
+                )`
+              : undefined,
+          ),
+        )
+        .orderBy(desc(schema.messages.createdAt), desc(schema.messages.id))
+        .limit(PAGE_SIZE + 1);
+
+      const hasMore = rows.length > PAGE_SIZE;
+      const page = hasMore ? rows.slice(0, PAGE_SIZE) : rows;
+      const oldest = page[page.length - 1];
+
+      return {
+        match: {
+          id: match.matchId,
+          jobId: match.jobId,
+          jobTitle: match.jobTitle,
+          jobStatus: match.jobStatus,
+          canReply: canReply(match.jobStatus),
+          peer: peer ?? null,
+        },
+        // Newest last, the way a chat reads.
+        messages: page.reverse().map((m) => ({ ...m, isMine: m.senderId === uid })),
+        nextCursor: hasMore && oldest ? oldest.id : null,
+      };
+    }),
+
+  /**
+   * Say something.
+   *
+   * Insert and `lastMessageAt` move together in one transaction: the jobs list
+   * sorts and previews on that column, so a message that lands without it is a
+   * conversation that silently stops surfacing.
+   */
+  send: protectedProcedure.input(sendMessageSchema).mutation(async ({ ctx, input }) => {
+    const uid = ctx.session.userId;
+    assertSendRate(uid);
+
+    return await ctx.db.transaction(async (tx) => {
+      const match = await requireMatchParticipant(tx, input.matchId, uid);
+
+      if (!canReply(match.jobStatus)) {
+        throw new TRPCError({
+          code: 'PRECONDITION_FAILED',
+          message:
+            match.jobStatus === 'cancelled'
+              ? 'This job was cancelled. The conversation is closed.'
+              : 'This job is finished. The conversation is closed.',
+        });
+      }
+
+      const [message] = await tx
+        .insert(schema.messages)
+        .values({
+          matchId: input.matchId,
+          senderId: uid,
+          body: input.body,
+          attachments: input.attachments,
+        })
+        .returning();
+
+      if (!message) {
+        throw new TRPCError({ code: 'INTERNAL_SERVER_ERROR', message: 'Could not send' });
+      }
+
+      await tx
+        .update(schema.matches)
+        .set({ lastMessageAt: message.createdAt })
+        .where(eq(schema.matches.id, input.matchId));
+
+      return { ...message, isMine: true as const };
+    });
+  }),
+
+  /**
+   * Mark everything the other side sent as read.
+   *
+   * Idempotent by construction — the `read_at IS NULL` predicate is also the
+   * partial index (`messages_unread_idx`), so a second call touches no rows.
+   */
+  markRead: protectedProcedure
+    .input(z.object({ matchId: z.string().uuid() }))
+    .mutation(async ({ ctx, input }) => {
+      const uid = ctx.session.userId;
+      await requireMatchParticipant(ctx.db, input.matchId, uid);
+
+      const updated = await ctx.db
+        .update(schema.messages)
+        .set({ readAt: new Date() })
+        .where(
+          and(
+            eq(schema.messages.matchId, input.matchId),
+            ne(schema.messages.senderId, uid),
+            isNull(schema.messages.readAt),
+          ),
+        )
+        .returning({ id: schema.messages.id });
+
+      return { read: updated.length };
+    }),
+
+  /**
+   * One number for the tab badge: everything unread across every conversation
+   * this person is part of, on either side of the market.
+   */
+  unreadTotal: protectedProcedure.query(async ({ ctx }) => {
+    const uid = ctx.session.userId;
+
+    const [row] = await ctx.db
+      .select({ n: sql<number>`count(*)::int` })
+      .from(schema.messages)
+      .innerJoin(schema.matches, eq(schema.matches.id, schema.messages.matchId))
+      .where(
+        and(
+          or(eq(schema.matches.clientId, uid), eq(schema.matches.proId, uid)),
+          ne(schema.messages.senderId, uid),
+          isNull(schema.messages.readAt),
+        ),
+      );
+
+    return { unread: row?.n ?? 0 };
+  }),
+});