import { TRPCError } from "@trpc/server"; import { chatChannelNameSchema, chatClientMutationIdSchema, chatCommandSchema, chatCreateChannelInputSchema, chatCreateDirectInputSchema, type ChatCommand, } from "@tasks/shared"; import { addPrivateMember, archiveChannel, ChannelPersistenceError, ChatPersistenceError, ChatMutationError, ChatReadError, chatHistory, chatThreadHistory, chatCatchUp, chatMutationCatchUp, chatMarkChannelRead, chatUnreadSummary, chatSequenceSchema, chatCursorSchema, chatPageSizeSchema, chatCatchUpSizeSchema, createChannel, createChannelTransactionRunner, createChatTransactionRunner, createDirectChannel, listVisibleChannels, createMessageMutationRunner, mutateChatMessage, removePrivateMember, renameChannel, sendChatMessage, } from "@tasks/database/chat"; import { z } from "zod"; import { router, workspaceProcedure } from "@/server/trpc"; const sendCommandSchema = chatCommandSchema.refine( (command) => command.type === "send_message", { message: "Expected a send_message command" }, ); const mutationCommandSchema = chatCommandSchema.refine( (command) => command.type === "edit_message" || command.type === "delete_message", { message: "Expected a message mutation" }, ); const mutationInputSchema = z .object({ workspace: z.string().min(1), command: mutationCommandSchema, }) .strict(); const inputSchema = z .object({ workspace: z.string().min(1), command: sendCommandSchema, mentionedUserIds: z.array(z.string().uuid()).max(100).optional(), }) .strict(); function mapPersistenceError(error: unknown): never { if (error instanceof ChatPersistenceError) { if (error.code === "CHAT_FORBIDDEN") { throw new TRPCError({ code: "NOT_FOUND", message: "Chat channel not found", }); } if (error.code === "CHAT_IDEMPOTENCY_CONFLICT") { throw new TRPCError({ code: "CONFLICT", message: "Message request conflicts with an existing request", }); } throw new TRPCError({ code: "BAD_REQUEST", message: "Invalid chat message request", }); } throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Unable to send chat message", }); } const workspaceInput = z.object({ workspace: z.string().min(1) }).strict(); const channelInput = workspaceInput .extend({ channelId: z.string().uuid(), clientMutationId: chatClientMutationIdSchema, }) .strict(); const runChannels = ( db: Parameters[0], ) => createChannelTransactionRunner(db); async function channelCall(operation: () => Promise): Promise { try { return await operation(); } catch (error) { if (error instanceof ChannelPersistenceError) { if (error.code === "NOT_FOUND") throw new TRPCError({ code: "NOT_FOUND", message: "Chat channel not found", }); if (error.code === "CONFLICT") throw new TRPCError({ code: "CONFLICT", message: "Channel conflicts with existing state", }); throw new TRPCError({ code: "BAD_REQUEST", message: error.code === "READ_ONLY" ? "Channel is archived" : "Invalid channel request", }); } throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Unable to update chat channel", }); } } const readScope = workspaceInput .extend({ channelId: z.string().uuid() }) .strict(); const catchUpInput = readScope .extend({ after: chatCursorSchema, through: chatCursorSchema.optional(), limit: chatCatchUpSizeSchema.default(100), }) .strict() .refine( (input) => input.through === undefined || BigInt(input.after) <= BigInt(input.through), { path: ["through"], message: "After exceeds through" }, ); const pageInput = readScope .extend({ before: chatSequenceSchema.optional(), limit: chatPageSizeSchema.default(50), }) .strict(); async function readCall(operation: () => Promise): Promise { try { return await operation(); } catch (error) { if (error instanceof ChatReadError) throw new TRPCError({ code: "NOT_FOUND", message: "Chat channel not found", }); throw error; } } async function mutationCall( db: Parameters[0], actorUserId: string, workspaceId: string, command: Extract, ) { if ( command.workspaceId !== workspaceId || command.context.workspaceId !== workspaceId ) throw new TRPCError({ code: "BAD_REQUEST", message: "Workspace mismatch" }); try { return await mutateChatMessage(createMessageMutationRunner(db), { actorUserId, command, }); } catch (error) { if (error instanceof ChatMutationError) { if (error.code === "CHAT_FORBIDDEN" || error.code === "CHAT_NOT_FOUND") throw new TRPCError({ code: "NOT_FOUND", message: "Chat message not found", }); if (error.code === "CHAT_REVISION_CONFLICT") throw new TRPCError({ code: "CONFLICT", message: "Message changed. Refresh before trying again.", }); throw new TRPCError({ code: "CONFLICT", message: "Message request conflicts with an existing request", }); } throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Unable to update chat message", }); } } export const chatRouter = router({ viewer: workspaceProcedure .input(workspaceInput) .query(({ ctx }) => ({ role: ctx.workspace.role })), // History and reconnect endpoints use one workspace-resolved, membership-checked read seam. history: workspaceProcedure.input(pageInput).query(({ ctx, input }) => readCall(() => chatHistory(ctx.db, { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, before: input.before, limit: input.limit, }), ), ), threadHistory: workspaceProcedure .input(pageInput.extend({ rootMessageId: z.string().uuid() }).strict()) .query(({ ctx, input }) => readCall(() => chatThreadHistory(ctx.db, { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, rootMessageId: input.rootMessageId, before: input.before, limit: input.limit, }), ), ), catchUp: workspaceProcedure .input(catchUpInput) .query(({ ctx, input }) => readCall(() => chatCatchUp(ctx.db, { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, after: input.after, through: input.through, limit: input.limit, }), ), ), mutationCatchUp: workspaceProcedure .input(catchUpInput) .query(({ ctx, input }) => readCall(() => chatMutationCatchUp(ctx.db, { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, after: input.after, through: input.through, limit: input.limit, }), ), ), markChannelRead: workspaceProcedure .input(readScope.extend({ sequence: chatSequenceSchema }).strict()) .mutation(({ ctx, input }) => readCall(() => chatMarkChannelRead(ctx.db, { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, sequence: input.sequence, }), ), ), unreadSummary: workspaceProcedure.input(workspaceInput).query(({ ctx }) => readCall(() => chatUnreadSummary(ctx.db, { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, }), ), ), listChannels: workspaceProcedure.input(workspaceInput).query(({ ctx }) => channelCall(() => listVisibleChannels(runChannels(ctx.db), { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, }), ), ), createChannel: workspaceProcedure .input(chatCreateChannelInputSchema) .mutation(({ ctx, input }) => channelCall(() => createChannel(runChannels(ctx.db), { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, kind: input.kind, name: input.name, clientMutationId: input.clientMutationId, }), ), ), renameChannel: workspaceProcedure .input(channelInput.extend({ name: chatChannelNameSchema }).strict()) .mutation(({ ctx, input }) => channelCall(() => renameChannel(runChannels(ctx.db), { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, name: input.name, clientMutationId: input.clientMutationId, }), ), ), archiveChannel: workspaceProcedure .input(channelInput) .mutation(({ ctx, input }) => channelCall(() => archiveChannel(runChannels(ctx.db), { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, clientMutationId: input.clientMutationId, }), ), ), addPrivateMember: workspaceProcedure .input(channelInput.extend({ userId: z.string().uuid() }).strict()) .mutation(({ ctx, input }) => channelCall(() => addPrivateMember(runChannels(ctx.db), { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, userId: input.userId, clientMutationId: input.clientMutationId, }), ), ), removePrivateMember: workspaceProcedure .input(channelInput.extend({ userId: z.string().uuid() }).strict()) .mutation(({ ctx, input }) => channelCall(() => removePrivateMember(runChannels(ctx.db), { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, channelId: input.channelId, userId: input.userId, clientMutationId: input.clientMutationId, }), ), ), createDirect: workspaceProcedure .input(chatCreateDirectInputSchema) .mutation(({ ctx, input }) => channelCall(() => createDirectChannel(runChannels(ctx.db), { workspaceId: ctx.workspace.id, actorUserId: ctx.session.user.id, participantUserIds: input.participantUserIds, clientMutationId: input.clientMutationId, }), ), ), editMessage: workspaceProcedure .input(mutationInputSchema) .mutation(({ ctx, input }) => { if (input.command.type !== "edit_message") throw new TRPCError({ code: "BAD_REQUEST", message: "Expected edit_message command", }); return mutationCall( ctx.db, ctx.session.user.id, ctx.workspace.id, input.command, ); }), deleteMessage: workspaceProcedure .input(mutationInputSchema) .mutation(({ ctx, input }) => { if (input.command.type !== "delete_message") throw new TRPCError({ code: "BAD_REQUEST", message: "Expected delete_message command", }); return mutationCall( ctx.db, ctx.session.user.id, ctx.workspace.id, input.command, ); }), send: workspaceProcedure .input(inputSchema) .mutation(async ({ ctx, input }) => { const command = { ...input.command, workspaceId: ctx.workspace.id, context: { type: "workspace" as const, workspaceId: ctx.workspace.id }, }; try { const result = await sendChatMessage( createChatTransactionRunner(ctx.db), { actorUserId: ctx.session.user.id, command, mentionedUserIds: input.mentionedUserIds ?? [], }, ); return { created: result.created, message: { ...result.message, sequence: result.message.sequence.toString(), }, }; } catch (error) { mapPersistenceError(error); } }), });