Submatter/apps/web/server/routers/chat.ts

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<typeof createChannelTransactionRunner>[0],
) => createChannelTransactionRunner(db);
async function channelCall<T>(operation: () => Promise<T>): Promise<T> {
  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<T>(operation: () => Promise<T>): Promise<T> {
  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<typeof createMessageMutationRunner>[0],
  actorUserId: string,
  workspaceId: string,
  command: Extract<ChatCommand, { type: "edit_message" | "delete_message" }>,
) {
  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);
      }
    }),
});