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(); 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[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 { 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); } /** * A conversation is either a MATCH or an ENQUIRY. * * A match is "a pro agreed to this job"; an enquiry is "somebody asked a * question before there was a job". They differ in what they hang off and what * closes them, and in nothing else — so they share one message table, one chat * screen, and the normalised context below. */ export const threadRefSchema = z.union([ z.object({ matchId: z.string().uuid() }), z.object({ enquiryId: z.string().uuid() }), ]); export type ThreadRef = z.infer; export interface ThreadContext { kind: 'match' | 'enquiry'; matchId: string | null; enquiryId: string | null; clientId: string; proId: string; /** Null on an enquiry — that is the whole point of one. */ jobId: string | null; /** Null on an enquiry. Kept because "cancelled" and "finished" are not the * same thing to say to somebody, and only the job knows which it was. */ jobStatus: JobStatus | null; title: string; canReply: boolean; } /** * The gate for anything that reads or writes a conversation. * * NOT_FOUND rather than FORBIDDEN for a thread the caller is not part of — the * same rule as `job.byId` and `request.accept`: a stranger must not be able to * probe whether a conversation exists. */ export async function requireThreadParticipant( exec: Executor, ref: ThreadRef, userId: string, ): Promise { if ('matchId' in ref) { const match = await requireMatchParticipant(exec, ref.matchId, userId); return { kind: 'match', matchId: match.matchId, enquiryId: null, clientId: match.clientId, proId: match.proId, jobId: match.jobId, jobStatus: match.jobStatus, title: match.jobTitle, canReply: canReply(match.jobStatus), }; } const [row] = await exec .select({ id: schema.enquiries.id, clientId: schema.enquiries.clientId, proId: schema.enquiries.proId, closedAt: schema.enquiries.closedAt, headline: schema.proProfiles.headline, }) .from(schema.enquiries) .innerJoin(schema.proProfiles, eq(schema.proProfiles.userId, schema.enquiries.proId)) .where(eq(schema.enquiries.id, ref.enquiryId)); if (!row || (row.clientId !== userId && row.proId !== userId)) { throw new TRPCError({ code: 'NOT_FOUND', message: 'Conversation not found' }); } return { kind: 'enquiry', matchId: null, enquiryId: row.id, clientId: row.clientId, proId: row.proId, jobId: null, jobStatus: null, title: row.headline, // A pro can end an enquiry. The history stays readable; nothing more lands. canReply: row.closedAt === null, }; } /** Scopes a message query to one thread, whichever kind it is. */ function inThread(ctx: ThreadContext) { return ctx.kind === 'match' ? eq(schema.messages.matchId, ctx.matchId!) : eq(schema.messages.enquiryId, ctx.enquiryId!); } /** * 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({ ref: threadRefSchema, cursor: cursorSchema.optional() })) .query(async ({ ctx, input }) => { const uid = ctx.session.userId; const thread = await requireThreadParticipant(ctx.db, input.ref, uid); const peerId = thread.clientId === uid ? thread.proId : thread.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( inThread(thread), input.cursor ? sql`(${schema.messages.createdAt}, ${schema.messages.id}) < ( SELECT anchor.created_at, anchor.id FROM messages anchor WHERE anchor.id = ${input.cursor}::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: { kind: thread.kind, id: thread.matchId ?? thread.enquiryId!, matchId: thread.matchId, enquiryId: thread.enquiryId, jobId: thread.jobId, jobStatus: thread.jobStatus, jobTitle: thread.title, canReply: thread.canReply, 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 thread = await requireThreadParticipant(tx, input.ref, uid); if (!thread.canReply) { throw new TRPCError({ code: 'PRECONDITION_FAILED', message: thread.kind === 'enquiry' ? 'This enquiry was closed.' : thread.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: thread.matchId, enquiryId: thread.enquiryId, senderId: uid, body: input.body, attachments: input.attachments, }) .returning(); if (!message) { throw new TRPCError({ code: 'INTERNAL_SERVER_ERROR', message: 'Could not send' }); } // Both parents carry lastMessageAt: the jobs list and the pro's enquiry // inbox both sort on it, so a message that lands without it is a // conversation that silently stops surfacing. if (thread.kind === 'match') { await tx .update(schema.matches) .set({ lastMessageAt: message.createdAt }) .where(eq(schema.matches.id, thread.matchId!)); } else { await tx .update(schema.enquiries) .set({ lastMessageAt: message.createdAt, // The pro answering is what takes this off the client's open cap. ...(uid === thread.proId ? { respondedAt: message.createdAt } : {}), }) .where(eq(schema.enquiries.id, thread.enquiryId!)); } 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({ ref: threadRefSchema })) .mutation(async ({ ctx, input }) => { const uid = ctx.session.userId; const thread = await requireThreadParticipant(ctx.db, input.ref, uid); const updated = await ctx.db .update(schema.messages) .set({ readAt: new Date() }) .where( and( inThread(thread), 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; // Both kinds of thread, in one scan. Written as a single predicate rather // than two queries because the badge is one number and reading it twice // would let the halves disagree. const [row] = await ctx.db .select({ n: sql`count(*)::int` }) .from(schema.messages) .where( and( ne(schema.messages.senderId, uid), isNull(schema.messages.readAt), or( sql`EXISTS ( SELECT 1 FROM matches m WHERE m.id = ${schema.messages.matchId} AND (m.client_id = ${uid} OR m.pro_id = ${uid}) )`, sql`EXISTS ( SELECT 1 FROM enquiries e WHERE e.id = ${schema.messages.enquiryId} AND (e.client_id = ${uid} OR e.pro_id = ${uid}) )`, ), ), ); return { unread: row?.n ?? 0 }; }), });