init project

apps/api/src/app.module.ts

@@ -0,0 +1,96 @@
+import { ApolloDriver, ApolloDriverConfig } from "@nestjs/apollo";
+import { Module } from "@nestjs/common";
+import { APP_GUARD } from "@nestjs/core";
+import { GraphQLISODateTime, GraphQLModule } from "@nestjs/graphql";
+import { JwtService } from "@nestjs/jwt";
+import { join } from "path";
+import { AuthModule } from "./auth/auth.module";
+import { GqlAuthGuard } from "./auth/gql-auth.guard";
+import { HealthController } from "./health.controller";
+import { MessengerResolver } from "./messenger/messenger.resolver";
+import { MessengerService } from "./messenger/messenger.service";
+import { PrismaModule } from "./prisma/prisma.module";
+import { PrismaService } from "./prisma/prisma.service";
+import { pubSubProvider } from "./pubsub.provider";
+import { SignalGateway } from "./signaling/signal.gateway";
+
+@Module({
+  imports: [
+    PrismaModule,
+    AuthModule,
+    GraphQLModule.forRootAsync<ApolloDriverConfig>({
+      driver: ApolloDriver,
+      imports: [AuthModule, PrismaModule],
+      inject: [JwtService, PrismaService],
+      useFactory: async (jwt: JwtService, prisma: PrismaService) => ({
+        typePaths: [
+          join(__dirname, "..", "..", "..", "packages", "graphql-schema", "schema.graphql"),
+        ],
+        sortSchema: true,
+        playground: true,
+        introspection: true,
+        subscriptions: {
+          "graphql-ws": true,
+        },
+        resolvers: {
+          DateTime: GraphQLISODateTime,
+          Chat: {
+            __resolveType(value: {
+              __typename?: string;
+              type?: string;
+            }) {
+              if (value.__typename) return value.__typename;
+              if (value.type === "DIRECT") return "DirectThread";
+              if (value.type === "GROUP") return "Group";
+              if (value.type === "CHANNEL") return "Channel";
+              return null;
+            },
+          },
+        },
+        context: async ({ req, connectionParams, extra }: {
+          req?: { headers?: Record<string, string> };
+          connectionParams?: Record<string, string>;
+          extra?: unknown;
+        }) => {
+          const authHeader =
+            req?.headers?.authorization ??
+            connectionParams?.authorization ??
+            connectionParams?.Authorization;
+          const base = req
+            ? { ...req, headers: { ...req.headers } }
+            : { headers: {} as Record<string, string> };
+          if (authHeader) base.headers.authorization = authHeader;
+          if (authHeader?.startsWith("Bearer ")) {
+            try {
+              const payload = (await jwt.verifyAsync(
+                authHeader.slice(7),
+              )) as { sub: string; deviceId?: string | null };
+              const user = await prisma.user.findUnique({
+                where: { id: payload.sub },
+              });
+              if (user) {
+                (base as { user?: unknown }).user = {
+                  userId: user.id,
+                  deviceId: payload.deviceId ?? null,
+                  user,
+                };
+              }
+            } catch {
+              /* invalid token */
+            }
+          }
+          return { req: base, connectionParams, extra };
+        },
+      }),
+    }),
+  ],
+  controllers: [HealthController],
+  providers: [
+    pubSubProvider,
+    MessengerService,
+    MessengerResolver,
+    SignalGateway,
+    { provide: APP_GUARD, useClass: GqlAuthGuard },
+  ],
+})
+export class AppModule {}

apps/api/src/auth/auth.module.ts

@@ -0,0 +1,18 @@
+import { Module } from "@nestjs/common";
+import { JwtModule } from "@nestjs/jwt";
+import { PassportModule } from "@nestjs/passport";
+import { AuthService } from "./auth.service";
+import { JwtStrategy } from "./jwt.strategy";
+
+@Module({
+  imports: [
+    PassportModule.register({ defaultStrategy: "jwt" }),
+    JwtModule.register({
+      secret: process.env.JWT_SECRET ?? "dev-secret-change-me",
+      signOptions: { expiresIn: "15m" },
+    }),
+  ],
+  providers: [AuthService, JwtStrategy],
+  exports: [AuthService, JwtModule],
+})
+export class AuthModule {}

apps/api/src/auth/auth.service.ts

@@ -0,0 +1,77 @@
+import { Injectable, UnauthorizedException } from "@nestjs/common";
+import { JwtService } from "@nestjs/jwt";
+import * as bcrypt from "bcryptjs";
+import { createHash, randomBytes } from "crypto";
+import { PrismaService } from "../prisma/prisma.service";
+
+@Injectable()
+export class AuthService {
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly jwt: JwtService,
+  ) {}
+
+  private hashToken(t: string) {
+    return createHash("sha256").update(t).digest("hex");
+  }
+
+  async register(email: string, password: string, displayName: string) {
+    const passwordHash = await bcrypt.hash(password, 10);
+    const user = await this.prisma.user.create({
+      data: { email, passwordHash, displayName },
+    });
+    return this.issueTokens(user.id, null);
+  }
+
+  async login(email: string, password: string) {
+    const user = await this.prisma.user.findUnique({ where: { email } });
+    if (!user || !(await bcrypt.compare(password, user.passwordHash))) {
+      throw new UnauthorizedException("Invalid credentials");
+    }
+    return this.issueTokens(user.id, null);
+  }
+
+  async refresh(refreshToken: string) {
+    const hash = this.hashToken(refreshToken);
+    const row = await this.prisma.refreshToken.findFirst({
+      where: { tokenHash: hash, expiresAt: { gt: new Date() } },
+    });
+    if (!row) throw new UnauthorizedException("Invalid refresh");
+    await this.prisma.refreshToken.delete({ where: { id: row.id } });
+    return this.issueTokens(row.userId, row.deviceId);
+  }
+
+  private async issueTokens(userId: string, deviceId: string | null) {
+    const refresh = randomBytes(48).toString("hex");
+    const expiresAt = new Date(Date.now() + 1000 * 60 * 60 * 24 * 30);
+    await this.prisma.refreshToken.create({
+      data: {
+        userId,
+        deviceId,
+        tokenHash: this.hashToken(refresh),
+        expiresAt,
+      },
+    });
+    const accessToken = await this.jwt.signAsync(
+      { sub: userId, deviceId },
+      { expiresIn: "15m" },
+    );
+    const user = await this.prisma.user.findUniqueOrThrow({
+      where: { id: userId },
+    });
+    return {
+      accessToken,
+      refreshToken: refresh,
+      user: {
+        id: user.id,
+        displayName: user.displayName,
+        email: user.email,
+        createdAt: user.createdAt,
+      },
+    };
+  }
+
+  async validateUser(userId: string) {
+    return this.prisma.user.findUnique({ where: { id: userId } });
+  }
+}

apps/api/src/auth/current-user.decorator.ts

@@ -0,0 +1,15 @@
+import { createParamDecorator, ExecutionContext } from "@nestjs/common";
+import { GqlExecutionContext } from "@nestjs/graphql";
+
+export type AuthUserCtx = {
+  userId: string;
+  deviceId: string | null;
+  user: { id: string; email: string; displayName: string; createdAt: Date };
+};
+
+export const CurrentUser = createParamDecorator(
+  (_data: unknown, context: ExecutionContext): AuthUserCtx => {
+    const ctx = GqlExecutionContext.create(context).getContext();
+    return ctx.req.user;
+  },
+);

apps/api/src/auth/gql-auth.guard.ts

@@ -0,0 +1,42 @@
+import { ExecutionContext, Injectable } from "@nestjs/common";
+import { Reflector } from "@nestjs/core";
+import { GqlExecutionContext } from "@nestjs/graphql";
+import { AuthGuard } from "@nestjs/passport";
+import { IS_PUBLIC_KEY } from "./public.decorator";
+
+@Injectable()
+export class GqlAuthGuard extends AuthGuard("jwt") {
+  constructor(private readonly reflector: Reflector) {
+    super();
+  }
+
+  canActivate(context: ExecutionContext) {
+    const isPublic = this.reflector.getAllAndOverride<boolean>(IS_PUBLIC_KEY, [
+      context.getHandler(),
+      context.getClass(),
+    ]);
+    if (isPublic) return true;
+    return super.canActivate(context);
+  }
+
+  getRequest(context: ExecutionContext) {
+    if (context.getType<string>() === "http") {
+      return context.switchToHttp().getRequest();
+    }
+    const ctx = GqlExecutionContext.create(context);
+    const { req, connectionParams, extra } = ctx.getContext();
+    if (req?.headers?.authorization) return req;
+    const auth =
+      connectionParams?.authorization ??
+      connectionParams?.Authorization ??
+      extra?.request?.headers?.authorization;
+    if (auth && req) {
+      req.headers.authorization = auth;
+    }
+    if (auth && !req?.headers?.authorization && extra?.request) {
+      extra.request.headers.authorization = auth;
+      return extra.request;
+    }
+    return req;
+  }
+}

apps/api/src/auth/jwt.strategy.ts

@@ -0,0 +1,27 @@
+import { Injectable, UnauthorizedException } from "@nestjs/common";
+import { PassportStrategy } from "@nestjs/passport";
+import { ExtractJwt, Strategy } from "passport-jwt";
+import { AuthService } from "./auth.service";
+
+export type JwtPayload = { sub: string; deviceId?: string | null };
+
+@Injectable()
+export class JwtStrategy extends PassportStrategy(Strategy, "jwt") {
+  constructor(private readonly auth: AuthService) {
+    super({
+      jwtFromRequest: ExtractJwt.fromAuthHeaderAsBearerToken(),
+      ignoreExpiration: false,
+      secretOrKey: process.env.JWT_SECRET ?? "dev-secret-change-me",
+    });
+  }
+
+  async validate(payload: JwtPayload) {
+    const user = await this.auth.validateUser(payload.sub);
+    if (!user) throw new UnauthorizedException();
+    return {
+      userId: user.id,
+      deviceId: payload.deviceId ?? null,
+      user,
+    };
+  }
+}

apps/api/src/auth/public.decorator.ts

@@ -0,0 +1,4 @@
+import { SetMetadata } from "@nestjs/common";
+
+export const IS_PUBLIC_KEY = "isPublic";
+export const Public = () => SetMetadata(IS_PUBLIC_KEY, true);

apps/api/src/health.controller.ts

@@ -0,0 +1,11 @@
+import { Controller, Get } from "@nestjs/common";
+import { Public } from "./auth/public.decorator";
+
+@Controller()
+export class HealthController {
+  @Public()
+  @Get("health")
+  health() {
+    return { ok: true };
+  }
+}

apps/api/src/main.ts

@@ -0,0 +1,14 @@
+import { NestFactory } from "@nestjs/core";
+import { WsAdapter } from "@nestjs/platform-ws";
+import { AppModule } from "./app.module";
+
+async function bootstrap() {
+  const app = await NestFactory.create(AppModule);
+  app.useWebSocketAdapter(new WsAdapter(app));
+  app.enableCors({ origin: true, credentials: true });
+  const port = process.env.PORT ? Number(process.env.PORT) : 4000;
+  await app.listen(port);
+  console.log(`API http://localhost:${port}/graphql  ws subscriptions graphql-ws`);
+}
+
+bootstrap();

apps/api/src/messenger/messenger.resolver.ts

@@ -0,0 +1,412 @@
+import { Inject, UseGuards } from "@nestjs/common";
+import { Args, Mutation, Query, Resolver, Subscription } from "@nestjs/graphql";
+import { RedisPubSub } from "graphql-redis-subscriptions";
+import { AuthService } from "../auth/auth.service";
+import { CurrentUser, AuthUserCtx } from "../auth/current-user.decorator";
+import { GqlAuthGuard } from "../auth/gql-auth.guard";
+import { Public } from "../auth/public.decorator";
+import { PUBSUB } from "../pubsub.provider";
+import { MessengerService } from "./messenger.service";
+
+const chatTopic = (id: string) => `CHAT_${id}`;
+const accountTopic = (userId: string) => `ACCOUNT_${userId}`;
+
+@Resolver()
+export class MessengerResolver {
+  constructor(
+    private readonly messenger: MessengerService,
+    private readonly auth: AuthService,
+    @Inject(PUBSUB) private readonly pubsub: RedisPubSub,
+  ) {}
+
+  @Public()
+  @Mutation("register")
+  async register(
+    @Args("email") email: string,
+    @Args("password") password: string,
+    @Args("displayName") displayName: string,
+  ) {
+    return this.auth.register(email, password, displayName);
+  }
+
+  @Public()
+  @Mutation("login")
+  async login(@Args("email") email: string, @Args("password") password: string) {
+    return this.auth.login(email, password);
+  }
+
+  @Public()
+  @Mutation("refresh")
+  async refresh(@Args("refreshToken") refreshToken: string) {
+    return this.auth.refresh(refreshToken);
+  }
+
+  @Query("me")
+  @UseGuards(GqlAuthGuard)
+  async me(@CurrentUser() u: AuthUserCtx) {
+    return {
+      id: u.user.id,
+      displayName: u.user.displayName,
+      email: u.user.email,
+      createdAt: u.user.createdAt.toISOString(),
+    };
+  }
+
+  @Query("myDevices")
+  @UseGuards(GqlAuthGuard)
+  async myDevices(@CurrentUser() u: AuthUserCtx) {
+    const rows = await this.messenger.listDevices(u.userId);
+    return rows.map((d) => ({
+      id: d.id,
+      userId: d.userId,
+      name: d.name,
+      lastSeenAt: d.lastSeenAt?.toISOString() ?? null,
+      createdAt: d.createdAt.toISOString(),
+    }));
+  }
+
+  @Query("chats")
+  @UseGuards(GqlAuthGuard)
+  async chats(@CurrentUser() u: AuthUserCtx) {
+    return this.messenger.listChats(u.userId);
+  }
+
+  @Query("messages")
+  @UseGuards(GqlAuthGuard)
+  async messages(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("after", { nullable: true }) after?: string,
+    @Args("first", { nullable: true }) first?: number,
+  ) {
+    return this.messenger.messagesConnection(chatId, u.userId, after, first ?? 30);
+  }
+
+  @Query("bot")
+  @UseGuards(GqlAuthGuard)
+  async bot(@CurrentUser() u: AuthUserCtx, @Args("id") id: string) {
+    const b = await this.messenger.findBotForOwner(u.userId, id);
+    if (!b) return null;
+    return {
+      id: b.id,
+      displayName: b.displayName,
+      tokenHint: this.messenger.botTokenHint(b.id),
+      webhookUrl: b.webhookUrl,
+      createdAt: b.createdAt.toISOString(),
+    };
+  }
+
+  @Query("myBots")
+  @UseGuards(GqlAuthGuard)
+  async myBots(@CurrentUser() u: AuthUserCtx) {
+    const rows = await this.messenger.listBots(u.userId);
+    return rows.map((b) => ({
+      id: b.id,
+      displayName: b.displayName,
+      tokenHint: this.messenger.botTokenHint(b.id),
+      webhookUrl: b.webhookUrl,
+      createdAt: b.createdAt.toISOString(),
+    }));
+  }
+
+  @Mutation("registerDevice")
+  @UseGuards(GqlAuthGuard)
+  async registerDevice(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("name", { nullable: true }) name?: string,
+  ) {
+    const d = await this.messenger.registerDevice(u.userId, name ?? null);
+    await this.messenger.publishAccount(u.userId, {
+      type: "DEVICE_ADDED",
+      deviceId: d.id,
+    });
+    return {
+      id: d.id,
+      userId: d.userId,
+      name: d.name,
+      lastSeenAt: d.lastSeenAt?.toISOString() ?? null,
+      createdAt: d.createdAt.toISOString(),
+    };
+  }
+
+  @Mutation("revokeDevice")
+  @UseGuards(GqlAuthGuard)
+  async revokeDevice(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("deviceId") deviceId: string,
+  ) {
+    return this.messenger.revokeDevice(u.userId, deviceId);
+  }
+
+  @Mutation("createDirectThread")
+  @UseGuards(GqlAuthGuard)
+  async createDirectThread(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("peerUserId") peerUserId: string,
+  ) {
+    return this.messenger.findOrCreateDirect(u.userId, peerUserId);
+  }
+
+  @Mutation("createGroup")
+  @UseGuards(GqlAuthGuard)
+  async createGroup(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("title") title: string,
+    @Args("memberIds", { type: () => [String], nullable: true }) memberIds?: string[],
+    @Args("isBroadcast", { nullable: true }) isBroadcast?: boolean,
+  ) {
+    return this.messenger.createGroup(u.userId, title, memberIds ?? [], isBroadcast ?? false);
+  }
+
+  @Mutation("createChannel")
+  @UseGuards(GqlAuthGuard)
+  async createChannel(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("title") title: string,
+    @Args("isPublic", { nullable: true }) isPublic?: boolean,
+  ) {
+    return this.messenger.createChannel(u.userId, title, isPublic ?? false);
+  }
+
+  @Mutation("joinChannel")
+  @UseGuards(GqlAuthGuard)
+  async joinChannel(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("channelId") channelId: string,
+  ) {
+    return this.messenger.joinChannel(u.userId, channelId);
+  }
+
+  @Mutation("inviteToGroup")
+  @UseGuards(GqlAuthGuard)
+  async inviteToGroup(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("groupId") groupId: string,
+    @Args("userIds", { type: () => [String] }) userIds: string[],
+  ) {
+    return this.messenger.inviteToGroup(u.userId, groupId, userIds);
+  }
+
+  @Mutation("setMemberRole")
+  @UseGuards(GqlAuthGuard)
+  async setMemberRole(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("userId") userId: string,
+    @Args("role") role: string,
+  ) {
+    return this.messenger.setMemberRole(u.userId, chatId, userId, role);
+  }
+
+  @Mutation("pinMessage")
+  @UseGuards(GqlAuthGuard)
+  async pinMessage(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("messageId") messageId: string,
+  ) {
+    return this.messenger.pinMessage(u.userId, chatId, messageId);
+  }
+
+  @Mutation("unpinMessage")
+  @UseGuards(GqlAuthGuard)
+  async unpinMessage(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("messageId") messageId: string,
+  ) {
+    return this.messenger.unpinMessage(u.userId, chatId, messageId);
+  }
+
+  @Mutation("sendMessage")
+  @UseGuards(GqlAuthGuard)
+  async sendMessage(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("body", { nullable: true }) body?: string,
+    @Args("replyToId", { nullable: true }) replyToId?: string,
+  ) {
+    return this.messenger.sendText(u.userId, chatId, body, replyToId ?? null);
+  }
+
+  @Mutation("editMessage")
+  @UseGuards(GqlAuthGuard)
+  async editMessage(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("messageId") messageId: string,
+    @Args("body") body: string,
+  ) {
+    return this.messenger.editMessage(u.userId, messageId, body);
+  }
+
+  @Mutation("deleteMessage")
+  @UseGuards(GqlAuthGuard)
+  async deleteMessage(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("messageId") messageId: string,
+    @Args("forEveryone", { nullable: true }) forEveryone?: boolean,
+  ) {
+    return this.messenger.deleteMessage(u.userId, messageId, forEveryone ?? false);
+  }
+
+  @Mutation("markDelivered")
+  @UseGuards(GqlAuthGuard)
+  async markDelivered(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("upToSeq") upToSeq: number,
+  ) {
+    return this.messenger.markDelivered(u.userId, chatId, upToSeq);
+  }
+
+  @Mutation("markRead")
+  @UseGuards(GqlAuthGuard)
+  async markRead(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("upToSeq") upToSeq: number,
+  ) {
+    return this.messenger.markRead(u.userId, chatId, upToSeq);
+  }
+
+  @Mutation("setTyping")
+  @UseGuards(GqlAuthGuard)
+  async setTyping(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("isTyping") isTyping: boolean,
+  ) {
+    return this.messenger.setTyping(u.userId, chatId, isTyping);
+  }
+
+  @Mutation("requestMediaUpload")
+  @UseGuards(GqlAuthGuard)
+  async requestMediaUpload(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("mimeType") mimeType: string,
+    @Args("bytesHint", { nullable: true }) _bytesHint?: number,
+  ) {
+    return this.messenger.requestUpload(u.userId, mimeType);
+  }
+
+  @Mutation("finalizeMediaUpload")
+  @UseGuards(GqlAuthGuard)
+  async finalizeMediaUpload(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("mediaId") mediaId: string,
+    @Args("mimeType") mimeType: string,
+    @Args("sizeBytes") sizeBytes: number,
+  ) {
+    return this.messenger.finalizeUpload(u.userId, mediaId, mimeType, sizeBytes);
+  }
+
+  @Mutation("sendMediaMessage")
+  @UseGuards(GqlAuthGuard)
+  async sendMediaMessage(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("mediaId") mediaId: string,
+    @Args("caption", { nullable: true }) caption?: string,
+    @Args("replyToId", { nullable: true }) replyToId?: string,
+  ) {
+    return this.messenger.sendMediaMessage(
+      u.userId,
+      chatId,
+      mediaId,
+      caption ?? null,
+      replyToId ?? null,
+    );
+  }
+
+  @Mutation("startCall")
+  @UseGuards(GqlAuthGuard)
+  async startCall(@CurrentUser() u: AuthUserCtx, @Args("chatId") chatId: string) {
+    return this.messenger.startCall(u.userId, chatId);
+  }
+
+  @Mutation("endCall")
+  @UseGuards(GqlAuthGuard)
+  async endCall(@CurrentUser() u: AuthUserCtx, @Args("callId") callId: string) {
+    return this.messenger.endCall(u.userId, callId);
+  }
+
+  @Mutation("createBot")
+  @UseGuards(GqlAuthGuard)
+  async createBot(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("displayName") displayName: string,
+    @Args("webhookUrl", { nullable: true }) webhookUrl?: string,
+  ) {
+    const { bot, token } = await this.messenger.createBot(
+      u.userId,
+      displayName,
+      webhookUrl ?? null,
+    );
+    return {
+      bot: {
+        id: bot.id,
+        displayName: bot.displayName,
+        tokenHint: this.messenger.botTokenHint(bot.id),
+        webhookUrl: bot.webhookUrl,
+        createdAt: bot.createdAt.toISOString(),
+      },
+      token,
+    };
+  }
+
+  @Mutation("rotateBotToken")
+  @UseGuards(GqlAuthGuard)
+  async rotateBotToken(@CurrentUser() u: AuthUserCtx, @Args("botId") botId: string) {
+    const { bot, token } = await this.messenger.rotateBotToken(u.userId, botId);
+    return {
+      bot: {
+        id: bot.id,
+        displayName: bot.displayName,
+        tokenHint: this.messenger.botTokenHint(bot.id),
+        webhookUrl: bot.webhookUrl,
+        createdAt: bot.createdAt.toISOString(),
+      },
+      token,
+    };
+  }
+
+  @Mutation("setBotWebhook")
+  @UseGuards(GqlAuthGuard)
+  async setBotWebhook(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("botId") botId: string,
+    @Args("webhookUrl") webhookUrl: string,
+  ) {
+    return this.messenger.setBotWebhook(u.userId, botId, webhookUrl);
+  }
+
+  @Mutation("addBotToChat")
+  @UseGuards(GqlAuthGuard)
+  async addBotToChat(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+    @Args("botId") botId: string,
+  ) {
+    return this.messenger.addBotToChat(u.userId, chatId, botId);
+  }
+
+  @Subscription("onChatEvent", {
+    resolve: (payload: { onChatEvent: unknown }) => payload.onChatEvent,
+  })
+  @UseGuards(GqlAuthGuard)
+  async onChatEvent(
+    @CurrentUser() u: AuthUserCtx,
+    @Args("chatId") chatId: string,
+  ) {
+    await this.messenger.assertMember(chatId, u.userId);
+    return this.pubsub.asyncIterator(chatTopic(chatId));
+  }
+
+  @Subscription("onAccountEvent", {
+    resolve: (payload: { onAccountEvent: unknown }) => payload.onAccountEvent,
+  })
+  @UseGuards(GqlAuthGuard)
+  async onAccountEvent(@CurrentUser() u: AuthUserCtx) {
+    return this.pubsub.asyncIterator(accountTopic(u.userId));
+  }
+}

apps/api/src/messenger/messenger.service.ts

@@ -0,0 +1,753 @@
+import {
+  ForbiddenException,
+  Inject,
+  Injectable,
+  NotFoundException,
+} from "@nestjs/common";
+import { ChatTypeEnum, Prisma } from "@prisma/client";
+import { createHash, randomBytes } from "crypto";
+import { PrismaService } from "../prisma/prisma.service";
+import { PUBSUB } from "../pubsub.provider";
+import { RedisPubSub } from "graphql-redis-subscriptions";
+import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3";
+import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
+
+const chatTopic = (id: string) => `CHAT_${id}`;
+const accountTopic = (userId: string) => `ACCOUNT_${userId}`;
+
+@Injectable()
+export class MessengerService {
+  private s3: S3Client | null = null;
+
+  constructor(
+    private readonly prisma: PrismaService,
+    @Inject(PUBSUB) private readonly pubsub: RedisPubSub,
+  ) {
+    const endpoint = process.env.S3_ENDPOINT;
+    if (endpoint) {
+      this.s3 = new S3Client({
+        region: process.env.S3_REGION ?? "us-east-1",
+        endpoint,
+        forcePathStyle: true,
+        credentials: {
+          accessKeyId: process.env.S3_ACCESS_KEY ?? "minio",
+          secretAccessKey: process.env.S3_SECRET_KEY ?? "minio12345",
+        },
+      });
+    }
+  }
+
+  async notifyBots(chatId: string, messagePayload: Record<string, unknown>) {
+    const members = await this.prisma.botMember.findMany({
+      where: { chatId },
+      include: { bot: true },
+    });
+    for (const m of members) {
+      const url = m.bot.webhookUrl;
+      if (!url) continue;
+      const body = JSON.stringify({
+        type: "message",
+        chatId,
+        message: messagePayload,
+      });
+      void fetch(url, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body,
+      }).catch(() => undefined);
+    }
+  }
+
+  private async member(chatId: string, userId: string) {
+    return this.prisma.chatMember.findUnique({
+      where: { chatId_userId: { chatId, userId } },
+    });
+  }
+
+  async assertMember(chatId: string, userId: string) {
+    const m = await this.member(chatId, userId);
+    if (!m) throw new ForbiddenException("Not a member");
+    return m;
+  }
+
+  private roleRank(role: string) {
+    const order = ["SUBSCRIBER", "MEMBER", "ADMIN", "OWNER"];
+    return order.indexOf(role);
+  }
+
+  private canPost(
+    chat: { type: ChatTypeEnum; isBroadcast: boolean },
+    role: string,
+  ) {
+    if (chat.type === ChatTypeEnum.CHANNEL) {
+      if (chat.isBroadcast) return this.roleRank(role) >= this.roleRank("ADMIN");
+      return this.roleRank(role) >= this.roleRank("MEMBER");
+    }
+    if (chat.type === ChatTypeEnum.GROUP && chat.isBroadcast) {
+      return this.roleRank(role) >= this.roleRank("ADMIN");
+    }
+    return this.roleRank(role) >= this.roleRank("MEMBER");
+  }
+
+  async publishChat(chatId: string, payload: Record<string, unknown>) {
+    await this.pubsub.publish(chatTopic(chatId), {
+      onChatEvent: { ...payload, at: new Date().toISOString() },
+    });
+  }
+
+  async publishAccount(userId: string, payload: Record<string, unknown>) {
+    await this.pubsub.publish(accountTopic(userId), {
+      onAccountEvent: { ...payload, at: new Date().toISOString() },
+    });
+  }
+
+  async listDevices(userId: string) {
+    return this.prisma.device.findMany({
+      where: { userId },
+      orderBy: { createdAt: "desc" },
+    });
+  }
+
+  async registerDevice(userId: string, name?: string | null) {
+    return this.prisma.device.create({
+      data: { userId, name: name ?? null },
+    });
+  }
+
+  async revokeDevice(userId: string, deviceId: string) {
+    const d = await this.prisma.device.findFirst({
+      where: { id: deviceId, userId },
+    });
+    if (!d) throw new NotFoundException();
+    await this.prisma.device.delete({ where: { id: deviceId } });
+    await this.publishAccount(userId, {
+      type: "DEVICE_REVOKED",
+      deviceId,
+    });
+    return true;
+  }
+
+  async findOrCreateDirect(userId: string, peerUserId: string) {
+    if (peerUserId === userId) throw new ForbiddenException("No self DM");
+    const peer = await this.prisma.user.findUnique({ where: { id: peerUserId } });
+    if (!peer) throw new NotFoundException("Peer not found");
+
+    const match = await this.prisma.chat.findFirst({
+      where: {
+        type: ChatTypeEnum.DIRECT,
+        AND: [
+          { members: { some: { userId } } },
+          { members: { some: { userId: peerUserId } } },
+        ],
+      },
+      include: { members: true },
+    });
+    if (match && match.members.length === 2) {
+      return this.toDirectGql(match.id, userId);
+    }
+
+    const chat = await this.prisma.chat.create({
+      data: {
+        type: ChatTypeEnum.DIRECT,
+        members: {
+          create: [
+            { userId, role: "MEMBER" },
+            { userId: peerUserId, role: "MEMBER" },
+          ],
+        },
+      },
+    });
+    return this.toDirectGql(chat.id, userId);
+  }
+
+  async createGroup(userId: string, title: string, memberIds: string[], isBroadcast: boolean) {
+    const ids = [...new Set([userId, ...memberIds])];
+    const chat = await this.prisma.chat.create({
+      data: {
+        type: ChatTypeEnum.GROUP,
+        title,
+        isBroadcast,
+        members: {
+          create: ids.map((uid) => ({
+            userId: uid,
+            role: uid === userId ? "OWNER" : "MEMBER",
+          })),
+        },
+      },
+    });
+    return this.toGroupGql(chat.id, userId);
+  }
+
+  async createChannel(userId: string, title: string, isPublic: boolean) {
+    const chat = await this.prisma.chat.create({
+      data: {
+        type: ChatTypeEnum.CHANNEL,
+        title,
+        isPublic,
+        isBroadcast: true,
+        members: {
+          create: [{ userId, role: "OWNER" }],
+        },
+      },
+    });
+    return this.toChannelGql(chat.id, userId);
+  }
+
+  async joinChannel(userId: string, channelId: string) {
+    const chat = await this.prisma.chat.findUnique({ where: { id: channelId } });
+    if (!chat || chat.type !== ChatTypeEnum.CHANNEL) throw new NotFoundException();
+    if (!chat.isPublic) throw new ForbiddenException("Private channel");
+    await this.prisma.chatMember.upsert({
+      where: { chatId_userId: { chatId: channelId, userId } },
+      create: { chatId: channelId, userId, role: "SUBSCRIBER" },
+      update: {},
+    });
+    return true;
+  }
+
+  async inviteToGroup(actorId: string, groupId: string, userIds: string[]) {
+    const chat = await this.prisma.chat.findUnique({ where: { id: groupId } });
+    if (!chat || chat.type !== ChatTypeEnum.GROUP) throw new NotFoundException();
+    const actor = await this.assertMember(groupId, actorId);
+    if (this.roleRank(actor.role) < this.roleRank("ADMIN")) {
+      throw new ForbiddenException();
+    }
+    for (const uid of userIds) {
+      await this.prisma.chatMember.upsert({
+        where: { chatId_userId: { chatId: groupId, userId: uid } },
+        create: { chatId: groupId, userId: uid, role: "MEMBER" },
+        update: {},
+      });
+    }
+    return true;
+  }
+
+  async setMemberRole(
+    actorId: string,
+    chatId: string,
+    targetUserId: string,
+    role: string,
+  ) {
+    const chat = await this.prisma.chat.findUnique({ where: { id: chatId } });
+    if (!chat) throw new NotFoundException();
+    const actor = await this.assertMember(chatId, actorId);
+    if (this.roleRank(actor.role) < this.roleRank("OWNER")) {
+      throw new ForbiddenException();
+    }
+    await this.prisma.chatMember.update({
+      where: { chatId_userId: { chatId, userId: targetUserId } },
+      data: { role },
+    });
+    return true;
+  }
+
+  async pinMessage(actorId: string, chatId: string, messageId: string) {
+    const actor = await this.assertMember(chatId, actorId);
+    if (this.roleRank(actor.role) < this.roleRank("ADMIN")) {
+      throw new ForbiddenException();
+    }
+    await this.prisma.pinnedMessage.create({
+      data: { chatId, messageId },
+    });
+    return true;
+  }
+
+  async unpinMessage(actorId: string, chatId: string, messageId: string) {
+    const actor = await this.assertMember(chatId, actorId);
+    if (this.roleRank(actor.role) < this.roleRank("ADMIN")) {
+      throw new ForbiddenException();
+    }
+    await this.prisma.pinnedMessage.deleteMany({
+      where: { chatId, messageId },
+    });
+    return true;
+  }
+
+  async listChats(userId: string) {
+    const memberships = await this.prisma.chatMember.findMany({
+      where: { userId },
+      include: { chat: true },
+    });
+    const out: unknown[] = [];
+    for (const m of memberships) {
+      if (m.chat.type === ChatTypeEnum.DIRECT) {
+        out.push(await this.toDirectGql(m.chatId, userId));
+      } else if (m.chat.type === ChatTypeEnum.GROUP) {
+        out.push(await this.toGroupGql(m.chatId, userId));
+      } else {
+        out.push(await this.toChannelGql(m.chatId, userId));
+      }
+    }
+    return out;
+  }
+
+  private async lastMessage(chatId: string) {
+    const msg = await this.prisma.message.findFirst({
+      where: { chatId, deletedAt: null },
+      orderBy: { seq: "desc" },
+      include: { sender: true, media: { include: { media: true } } },
+    });
+    if (!msg) return null;
+    return this.toMessageGql(msg);
+  }
+
+  private unreadCount(chatId: string, userId: string, lastReadSeq: number) {
+    return this.prisma.message.count({
+      where: {
+        chatId,
+        deletedAt: null,
+        seq: { gt: lastReadSeq },
+        NOT: { senderId: userId },
+      },
+    });
+  }
+
+  async toDirectGql(chatId: string, viewerId: string) {
+    const chat = await this.prisma.chat.findUniqueOrThrow({
+      where: { id: chatId },
+      include: { members: { include: { user: true } } },
+    });
+    const other = chat.members.find((m) => m.userId !== viewerId)?.user;
+    if (!other) throw new Error("Invalid DM");
+    const mem = await this.member(chatId, viewerId);
+    const ur = mem ? await this.unreadCount(chatId, viewerId, mem.lastReadSeq) : 0;
+    return {
+      __typename: "DirectThread" as const,
+      id: chat.id,
+      type: "DIRECT",
+      title: null,
+      peer: {
+        id: other.id,
+        displayName: other.displayName,
+        email: other.email,
+        createdAt: other.createdAt.toISOString(),
+      },
+      lastMessage: await this.lastMessage(chatId),
+      unreadCount: ur,
+      updatedAt: chat.updatedAt.toISOString(),
+    };
+  }
+
+  async toGroupGql(chatId: string, viewerId: string) {
+    const chat = await this.prisma.chat.findUniqueOrThrow({ where: { id: chatId } });
+    const mem = await this.member(chatId, viewerId);
+    const ur = mem ? await this.unreadCount(chatId, viewerId, mem.lastReadSeq) : 0;
+    const memberCount = await this.prisma.chatMember.count({ where: { chatId } });
+    return {
+      __typename: "Group" as const,
+      id: chat.id,
+      type: "GROUP",
+      title: chat.title,
+      lastMessage: await this.lastMessage(chatId),
+      unreadCount: ur,
+      updatedAt: chat.updatedAt.toISOString(),
+      memberCount,
+      isBroadcast: chat.isBroadcast,
+    };
+  }
+
+  async toChannelGql(chatId: string, viewerId: string) {
+    const chat = await this.prisma.chat.findUniqueOrThrow({ where: { id: chatId } });
+    const mem = await this.member(chatId, viewerId);
+    const ur = mem ? await this.unreadCount(chatId, viewerId, mem.lastReadSeq) : 0;
+    const subscriberCount = await this.prisma.chatMember.count({ where: { chatId } });
+    const pins = await this.prisma.pinnedMessage.findMany({
+      where: { chatId },
+      include: { message: { include: { sender: true, media: { include: { media: true } } } } },
+    });
+    const pinnedMessages = await Promise.all(
+      pins.map((p) => this.toMessageGql(p.message)),
+    );
+    return {
+      __typename: "Channel" as const,
+      id: chat.id,
+      type: "CHANNEL",
+      title: chat.title,
+      lastMessage: await this.lastMessage(chatId),
+      unreadCount: ur,
+      updatedAt: chat.updatedAt.toISOString(),
+      subscriberCount,
+      pinnedMessages,
+    };
+  }
+
+  async toMessageGql(
+    msg: Prisma.MessageGetPayload<{
+      include: { sender: true; media: { include: { media: true } } };
+    }>,
+  ) {
+    const mediaRefs = await Promise.all(
+      msg.media.map(async (mm) => {
+        const url = await this.publicUrlForMedia(mm.media);
+        const thumb = mm.media.thumbnailKey
+          ? await this.publicUrlForKey(mm.media.thumbnailKey)
+          : null;
+        return {
+          id: mm.media.id,
+          mimeType: mm.media.mimeType,
+          url,
+          thumbnailUrl: thumb,
+        };
+      }),
+    );
+    return {
+      id: msg.id,
+      chatId: msg.chatId,
+      senderId: msg.senderId,
+      sender: msg.sender
+        ? {
+            id: msg.sender.id,
+            displayName: msg.sender.displayName,
+            email: msg.sender.email,
+            createdAt: msg.sender.createdAt.toISOString(),
+          }
+        : null,
+      kind: msg.kind,
+      body: msg.body,
+      mediaRefs,
+      replyToId: msg.replyToId,
+      seq: msg.seq,
+      createdAt: msg.createdAt.toISOString(),
+      editedAt: msg.editedAt?.toISOString() ?? null,
+      deletedAt: msg.deletedAt?.toISOString() ?? null,
+    };
+  }
+
+  private async publicUrlForMedia(m: { s3Key: string }) {
+    return this.publicUrlForKey(m.s3Key);
+  }
+
+  private async publicUrlForKey(key: string) {
+    const base = process.env.PUBLIC_MEDIA_BASE_URL;
+    const bucket = process.env.S3_BUCKET ?? "tlm-media";
+    if (base) return `${base.replace(/\/$/, "")}/${key}`;
+    if (this.s3) {
+      const endpoint = process.env.S3_ENDPOINT ?? "";
+      return `${endpoint.replace(/\/$/, "")}/${bucket}/${key}`;
+    }
+    return `media://${key}`;
+  }
+
+  async messagesConnection(chatId: string, userId: string, after?: string | null, first = 30) {
+    await this.assertMember(chatId, userId);
+    const take = Math.min(first, 100);
+    const cursorSeq = after ? Number(Buffer.from(after, "base64").toString("utf8")) : null;
+    const where: Prisma.MessageWhereInput = {
+      chatId,
+      deletedAt: null,
+      ...(cursorSeq !== null && !Number.isNaN(cursorSeq) ? { seq: { lt: cursorSeq } } : {}),
+    };
+    const rows = await this.prisma.message.findMany({
+      where,
+      orderBy: { seq: "desc" },
+      take: take + 1,
+      include: { sender: true, media: { include: { media: true } } },
+    });
+    const hasNextPage = rows.length > take;
+    const slice = hasNextPage ? rows.slice(0, take) : rows;
+    const edges = await Promise.all(
+      slice.map(async (m) => ({
+        cursor: Buffer.from(String(m.seq), "utf8").toString("base64"),
+        node: await this.toMessageGql(m),
+      })),
+    );
+    const endCursor = edges.length ? edges[edges.length - 1].cursor : null;
+    return {
+      edges,
+      pageInfo: { endCursor, hasNextPage },
+    };
+  }
+
+  async sendText(userId: string, chatId: string, body?: string | null, replyToId?: string | null) {
+    const chat = await this.prisma.chat.findUniqueOrThrow({ where: { id: chatId } });
+    const mem = await this.assertMember(chatId, userId);
+    if (!this.canPost(chat, mem.role)) throw new ForbiddenException("Cannot post");
+    const msg = await this.prisma.$transaction(async (tx) => {
+      const agg = await tx.message.aggregate({
+        where: { chatId },
+        _max: { seq: true },
+      });
+      const seq = (agg._max.seq ?? 0) + 1;
+      return tx.message.create({
+        data: {
+          chatId,
+          senderId: userId,
+          kind: "TEXT",
+          body: body ?? "",
+          replyToId: replyToId ?? undefined,
+          seq,
+        },
+        include: { sender: true, media: { include: { media: true } } },
+      });
+    });
+    const gql = await this.toMessageGql(msg);
+    await this.publishChat(chatId, {
+      type: "MESSAGE_CREATED",
+      chatId,
+      message: gql,
+    });
+    void this.notifyBots(chatId, gql);
+    return gql;
+  }
+
+  async sendMediaMessage(
+    userId: string,
+    chatId: string,
+    mediaId: string,
+    caption?: string | null,
+    replyToId?: string | null,
+  ) {
+    const chat = await this.prisma.chat.findUniqueOrThrow({ where: { id: chatId } });
+    const mem = await this.assertMember(chatId, userId);
+    if (!this.canPost(chat, mem.role)) throw new ForbiddenException();
+    const media = await this.prisma.mediaObject.findFirst({
+      where: { id: mediaId, userId, finalized: true },
+    });
+    if (!media) throw new NotFoundException("Media not ready");
+    const msg = await this.prisma.$transaction(async (tx) => {
+      const agg = await tx.message.aggregate({
+        where: { chatId },
+        _max: { seq: true },
+      });
+      const seq = (agg._max.seq ?? 0) + 1;
+      const m = await tx.message.create({
+        data: {
+          chatId,
+          senderId: userId,
+          kind: "MEDIA",
+          body: caption ?? "",
+          replyToId: replyToId ?? undefined,
+          seq,
+          media: { create: [{ mediaId: media.id }] },
+        },
+        include: { sender: true, media: { include: { media: true } } },
+      });
+      return m;
+    });
+    const gql = await this.toMessageGql(msg);
+    await this.publishChat(chatId, {
+      type: "MESSAGE_CREATED",
+      chatId,
+      message: gql,
+    });
+    void this.notifyBots(chatId, gql);
+    return gql;
+  }
+
+  async editMessage(userId: string, messageId: string, body: string) {
+    const msg = await this.prisma.message.findUniqueOrThrow({ where: { id: messageId } });
+    if (msg.senderId !== userId) throw new ForbiddenException();
+    const updated = await this.prisma.message.update({
+      where: { id: messageId },
+      data: { body, editedAt: new Date() },
+      include: { sender: true, media: { include: { media: true } } },
+    });
+    const gql = await this.toMessageGql(updated);
+    await this.publishChat(msg.chatId, {
+      type: "MESSAGE_UPDATED",
+      chatId: msg.chatId,
+      message: gql,
+    });
+    return gql;
+  }
+
+  async deleteMessage(userId: string, messageId: string, forEveryone: boolean) {
+    const msg = await this.prisma.message.findUniqueOrThrow({ where: { id: messageId } });
+    const mem = await this.assertMember(msg.chatId, userId);
+    const isOwner = msg.senderId === userId;
+    if (!forEveryone && !isOwner) throw new ForbiddenException();
+    if (forEveryone && this.roleRank(mem.role) < this.roleRank("ADMIN")) {
+      throw new ForbiddenException();
+    }
+    const deleted = await this.prisma.message.update({
+      where: { id: messageId },
+      data: { deletedAt: new Date(), body: null },
+      include: { sender: true, media: { include: { media: true } } },
+    });
+    const gql = await this.toMessageGql(deleted);
+    await this.publishChat(msg.chatId, {
+      type: "MESSAGE_DELETED",
+      chatId: msg.chatId,
+      message: gql,
+    });
+    return true;
+  }
+
+  async markDelivered(userId: string, chatId: string, upToSeq: number) {
+    await this.assertMember(chatId, userId);
+    await this.prisma.chatMember.update({
+      where: { chatId_userId: { chatId, userId } },
+      data: { lastDeliveredSeq: upToSeq },
+    });
+    await this.publishChat(chatId, {
+      type: "DELIVERED",
+      chatId,
+      userId,
+      upToSeq,
+    });
+    return true;
+  }
+
+  async markRead(userId: string, chatId: string, upToSeq: number) {
+    await this.assertMember(chatId, userId);
+    await this.prisma.chatMember.update({
+      where: { chatId_userId: { chatId, userId } },
+      data: { lastReadSeq: upToSeq },
+    });
+    await this.publishChat(chatId, {
+      type: "READ_RECEIPT",
+      chatId,
+      userId,
+      upToSeq,
+    });
+    return true;
+  }
+
+  async setTyping(userId: string, chatId: string, isTyping: boolean) {
+    await this.assertMember(chatId, userId);
+    await this.publishChat(chatId, {
+      type: "TYPING",
+      chatId,
+      userId,
+      isTyping,
+    });
+    return true;
+  }
+
+  async requestUpload(userId: string, mimeType: string) {
+    if (!this.s3) throw new ForbiddenException("S3 not configured");
+    const id = randomBytes(8).toString("hex");
+    const key = `uploads/${userId}/${id}`;
+    await this.prisma.mediaObject.create({
+      data: {
+        id,
+        userId,
+        mimeType,
+        s3Key: key,
+        finalized: false,
+      },
+    });
+    const bucket = process.env.S3_BUCKET ?? "tlm-media";
+    const cmd = new PutObjectCommand({
+      Bucket: bucket,
+      Key: key,
+      ContentType: mimeType,
+    });
+    const uploadUrl = await getSignedUrl(this.s3, cmd, { expiresIn: 3600 });
+    return { uploadUrl, mediaId: id, headers: [] as { name: string; value: string }[] };
+  }
+
+  async finalizeUpload(userId: string, mediaId: string, mimeType: string, sizeBytes: number) {
+    const m = await this.prisma.mediaObject.updateMany({
+      where: { id: mediaId, userId },
+      data: { finalized: true, mimeType, sizeBytes },
+    });
+    if (m.count === 0) throw new NotFoundException();
+    const row = await this.prisma.mediaObject.findUniqueOrThrow({ where: { id: mediaId } });
+    return {
+      id: row.id,
+      mimeType: row.mimeType,
+      url: await this.publicUrlForMedia(row),
+      thumbnailUrl: row.thumbnailKey
+        ? await this.publicUrlForKey(row.thumbnailKey)
+        : null,
+    };
+  }
+
+  async startCall(userId: string, chatId: string) {
+    await this.assertMember(chatId, userId);
+    const session = await this.prisma.callSession.create({
+      data: { chatId, createdBy: userId, status: "RINGING" },
+    });
+    return this.formatCall(session);
+  }
+
+  async endCall(userId: string, callId: string) {
+    const c = await this.prisma.callSession.findUnique({ where: { id: callId } });
+    if (!c) throw new NotFoundException();
+    await this.assertMember(c.chatId, userId);
+    await this.prisma.callSession.update({
+      where: { id: callId },
+      data: { status: "ENDED" },
+    });
+    return true;
+  }
+
+  formatCall(session: { id: string; chatId: string; createdBy: string; status: string; createdAt: Date }) {
+    const urls = (process.env.ICE_URLS ?? "stun:stun.l.google.com:19302").split(",");
+    return {
+      id: session.id,
+      chatId: session.chatId,
+      createdBy: session.createdBy,
+      status: session.status,
+      createdAt: session.createdAt.toISOString(),
+      iceServers: urls.map((u) => ({ urls: [u.trim()] })),
+    };
+  }
+
+  async createBot(ownerId: string, displayName: string, webhookUrl?: string | null) {
+    const token = `bot_${randomBytes(24).toString("hex")}`;
+    const tokenHash = createHash("sha256").update(token).digest("hex");
+    const bot = await this.prisma.bot.create({
+      data: {
+        ownerUserId: ownerId,
+        displayName,
+        tokenHash,
+        webhookUrl: webhookUrl ?? null,
+      },
+    });
+    return { bot, token };
+  }
+
+  async rotateBotToken(ownerId: string, botId: string) {
+    const bot = await this.prisma.bot.findFirst({
+      where: { id: botId, ownerUserId: ownerId },
+    });
+    if (!bot) throw new NotFoundException();
+    const token = `bot_${randomBytes(24).toString("hex")}`;
+    const tokenHash = createHash("sha256").update(token).digest("hex");
+    const updated = await this.prisma.bot.update({
+      where: { id: botId },
+      data: { tokenHash },
+    });
+    return { bot: updated, token };
+  }
+
+  async setBotWebhook(ownerId: string, botId: string, webhookUrl: string) {
+    const bot = await this.prisma.bot.findFirst({
+      where: { id: botId, ownerUserId: ownerId },
+    });
+    if (!bot) throw new NotFoundException();
+    await this.prisma.bot.update({
+      where: { id: botId },
+      data: { webhookUrl },
+    });
+    return true;
+  }
+
+  async addBotToChat(ownerId: string, chatId: string, botId: string) {
+    const bot = await this.prisma.bot.findFirst({
+      where: { id: botId, ownerUserId: ownerId },
+    });
+    if (!bot) throw new ForbiddenException();
+    await this.assertMember(chatId, ownerId);
+    await this.prisma.botMember.upsert({
+      where: { botId_chatId: { botId, chatId } },
+      create: { botId, chatId },
+      update: {},
+    });
+    return true;
+  }
+
+  botTokenHint(botId: string) {
+    return `${botId.slice(0, 4)}…`;
+  }
+
+  async findBotForOwner(ownerId: string, botId: string) {
+    return this.prisma.bot.findFirst({ where: { id: botId, ownerUserId: ownerId } });
+  }
+
+  async listBots(ownerId: string) {
+    return this.prisma.bot.findMany({ where: { ownerUserId: ownerId } });
+  }
+}

apps/api/src/prisma/prisma.module.ts

@@ -0,0 +1,9 @@
+import { Global, Module } from "@nestjs/common";
+import { PrismaService } from "./prisma.service";
+
+@Global()
+@Module({
+  providers: [PrismaService],
+  exports: [PrismaService],
+})
+export class PrismaModule {}

apps/api/src/prisma/prisma.service.ts

@@ -0,0 +1,16 @@
+import { Injectable, OnModuleDestroy, OnModuleInit } from "@nestjs/common";
+import { PrismaClient } from "@prisma/client";
+
+@Injectable()
+export class PrismaService
+  extends PrismaClient
+  implements OnModuleInit, OnModuleDestroy
+{
+  async onModuleInit() {
+    await this.$connect();
+  }
+
+  async onModuleDestroy() {
+    await this.$disconnect();
+  }
+}

apps/api/src/pubsub.provider.ts

@@ -0,0 +1,22 @@
+import { Provider } from "@nestjs/common";
+import { RedisPubSub } from "graphql-redis-subscriptions";
+import Redis from "ioredis";
+
+export const PUBSUB = "PUBSUB";
+
+/**
+ * For higher fan-out, swap `RedisPubSub` for NATS/Kafka-backed fan-out and keep the
+ * same topic naming (`CHAT_${chatId}`, `ACCOUNT_${userId}`) so resolvers stay stable.
+ */
+export const pubSubProvider: Provider = {
+  provide: PUBSUB,
+  useFactory: () => {
+    const host = process.env.REDIS_HOST ?? "127.0.0.1";
+    const port = Number(process.env.REDIS_PORT ?? "6379");
+    const options = { host, port };
+    return new RedisPubSub({
+      publisher: new Redis(options),
+      subscriber: new Redis(options),
+    });
+  },
+};

apps/api/src/signaling/signal.gateway.ts

@@ -0,0 +1,67 @@
+import { Logger } from "@nestjs/common";
+import {
+  OnGatewayConnection,
+  OnGatewayDisconnect,
+  WebSocketGateway,
+  WebSocketServer,
+} from "@nestjs/websockets";
+import { IncomingMessage } from "http";
+import { WebSocket, Server } from "ws";
+
+type Client = WebSocket & { callId?: string };
+
+@WebSocketGateway({ path: "/signal" })
+export class SignalGateway implements OnGatewayConnection, OnGatewayDisconnect {
+  private readonly logger = new Logger(SignalGateway.name);
+  private readonly rooms = new Map<string, Set<Client>>();
+
+  @WebSocketServer()
+  server!: Server;
+
+  handleConnection(client: Client, req: IncomingMessage) {
+    client.on("message", (raw) => {
+      try {
+        const msg = JSON.parse(String(raw)) as {
+          type?: string;
+          callId?: string;
+          payload?: unknown;
+        };
+        if (msg.type === "join" && msg.callId) {
+          this.leave(client);
+          client.callId = msg.callId;
+          if (!this.rooms.has(msg.callId)) this.rooms.set(msg.callId, new Set());
+          this.rooms.get(msg.callId)!.add(client);
+          return;
+        }
+        if (msg.type === "signal" && client.callId) {
+          const room = this.rooms.get(client.callId);
+          if (!room) return;
+          const data = JSON.stringify({
+            type: "signal",
+            payload: msg.payload,
+          });
+          for (const peer of room) {
+            if (peer !== client && peer.readyState === WebSocket.OPEN) {
+              peer.send(data);
+            }
+          }
+        }
+      } catch (e) {
+        this.logger.warn(`Bad WS message: ${e}`);
+      }
+    });
+  }
+
+  handleDisconnect(client: Client) {
+    this.leave(client);
+  }
+
+  private leave(client: Client) {
+    if (!client.callId) return;
+    const room = this.rooms.get(client.callId);
+    if (!room) return;
+    room.delete(client);
+    if (room.size === 0) this.rooms.delete(client.callId);
+    client.callId = undefined;
+  }
+}