NestJS Integration

Status: Active development — Socket.IO gateway for real-time chat, REST API for room/message management, SSE for typing/presence, AsyncAPI for WebSocket contract documentation.

Deployment

Live: https://nous-mastra-nest.curlynguyen95.workers.dev Deployed via Cloudflare Containers. Container uses lite instance type with sleepAfter = 1m — container sleeps after 1 minute of inactivity and wakes on request (may take a few seconds to cold start).

Overview

The NestJS app (apps/nestjs/) serves as the backend for Nous’s real-time chat infrastructure. It provides:

WebSocket Gateway — Socket.IO-based real-time messaging via ChatRealtimeGateway
REST API — Room and message management via ChatController
AsyncAPI Documentation — Auto-generated WebSocket contract via nestjs-asyncapi
Mastra AI Integration — Vicky AI participant in group chats

Architecture

Data Flow

REST API Flow

Mobile → ChatController (/v1/chat/rooms, /v1/chat/rooms/:roomId/messages)
       → ChatService (business logic)
       → ChatRepository (data access)
       → PostgreSQL / InMemory
       → Response

WebSocket Flow

Mobile ↔ ChatRealtimeGateway (Socket.IO namespace: /chat)
       ↔ ChatRealtimeService (publishes events)
       → Room subscribers (Socket.IO rooms)

AI Reply Flow

Human Message → ChatService.sendMessage()
             → AiParticipantService.shouldVickyReply() (decision)
             → Mastra Agent.generateVickyReply() (AI generation)
             → ChatRepository.createAiMessage()
             → ChatRealtimeService.publishMessageCreated()
             → Room subscribers receive message.created event

Key Components

ChatRealtimeGateway

apps/nestjs/src/chat/realtime/chat-realtime.gateway.ts WebSocket gateway handling Socket.IO connections for real-time chat:

@WebSocketGateway({
  namespace: '/chat',
  transports: ['websocket'],
  cors: { origin: '*' },
})
export class ChatRealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect {
  @SubscribeMessage(ChatClientEventType.MessageSend)
  async handleMessageSend(
    @ConnectedSocket() client: Socket,
    @MessageBody() body: ChatSendMessageClientEventDto
  ): Promise<WsResponse<ChatAckServerEventDto | ChatErrorServerEventDto>> {
    // Process message, create in DB, return ack
  }
}

Events:

Client Event	Direction	Description
`ping`	→ Server	Health check ping
`message.send`	→ Server	Send a chat message

Server Event	Direction	Description
`pong`	→ Client	Health check response
`ack`	→ Client	Message acknowledged
`error`	→ Client	Error response
`message.created`	→ Client	New message (via ChatRealtimeService)

ChatRealtimeService

apps/nestjs/src/chat/realtime/chat-realtime.service.ts Publishes domain events to Socket.IO rooms:

@Injectable()
@AsyncApi()
export class ChatRealtimeService {
  @AsyncApiPub({
    channel: ChatRealtimeEventType.MessageCreated,
    message: { payload: ChatMessageCreatedServerEventDto },
  })
  publishMessageCreated(roomId: string, message: ChatMessageDto): ChatMessageCreatedServerEventDto {
    // Emit to room
  }
}

ChatController

apps/nestjs/src/chat/chat.controller.ts REST API for room and message management:

@ApiTags('chat')
@Controller('/v1/chat')
export class ChatController {
  @Post('rooms')                    // Create a chat room
  @Get('rooms/:roomId')             // Get room details
  @Get('rooms/:roomId/messages')    // List messages
  @Post('rooms/:roomId/messages')   // Send a message
}

Repository Pattern

apps/nestjs/src/chat/repositories/ Two implementations with runtime selection:

// Factory selects based on database availability
useFactory: (
  databaseService: DatabaseService,
  inMemoryRepository: InMemoryChatRepository,
  postgresRepository: PostgresChatRepository
) => databaseService.isConfigured ? postgresRepository : inMemoryRepository

Repository	Environment	Use Case
`PostgresChatRepository`	Production	Full persistence
`InMemoryChatRepository`	Development	No database required

DatabaseService

apps/nestjs/src/database/database.service.ts Provides PostgreSQL connection pooling with SLL support and transaction management:

@Injectable()
export class DatabaseService implements OnModuleInit, OnModuleDestroy {
  private readonly pool: pg.Pool | null;

  async transaction<T>(operation: (client: pg.PoolClient) => Promise<T>) {
    // BEGIN → operation → COMMIT/ROLLBACK
  }
}

Auto-detects DATABASE_URL — if not set, falls back to in-memory repository.

AsyncAPI Integration

Overview

AsyncAPI spec is auto-generated from @AsyncApiPub / @AsyncApiSub decorators using nestjs-asyncapi. The spec is:

Available at /asyncapi (HTML UI)
Available at /asyncapi.json (raw JSON)
Exported to fern/generated/asyncapi.json for SDK generation

Decorator Usage

@AsyncApiSub({
  channel: ChatClientEventType.MessageSend,
  message: { payload: ChatSendMessageClientEventDto },
})
@AsyncApiPub(
  {
    channel: ChatServerEventType.Ack,
    message: { payload: ChatAckServerEventDto },
  },
  {
    channel: ChatServerEventType.Error,
    message: { payload: ChatErrorServerEventDto },
  }
)
@SubscribeMessage(ChatClientEventType.MessageSend)
handleMessageSend(...) { ... }

Export Script

apps/nestjs/scripts/export-asyncapi.ts Standalone script that generates the AsyncAPI JSON for Fern/SDK generation:

const app = await NestFactory.create(AppModule)
const document = await createAsyncApiDocument(app)
await writeFile('fern/generated/asyncapi.json', JSON.stringify(document))

Server-Sent Events (SSE)

SSE provides one-way server-to-client streaming for events like typing indicators and presence updates. It’s simpler than WebSocket and works well over HTTP/2.

SSE Endpoints

Endpoint	Method	Description
`/v1/chat/sse/events/:roomId`	GET	SSE stream for room events
`/v1/chat/sse/typing`	POST	Broadcast typing start/stop

Event Types

Event	Description
`typing.started`	User started typing
`typing.stopped`	User stopped typing (or timed out after 5s)
`presence.update`	Active users in room updated
`ai.thinking`	Vicky AI is generating a response

SseService

apps/nestjs/src/chat/sse/sse.service.ts Manages SSE subscriptions and broadcasts events to room subscribers:

@Injectable()
export class SseService {
  subscribe(userId: string, roomId: string): SseClient {
    // Returns an RxJS Subject that emits SSE events
  }

  handleTypingStart(roomId: string, userId: string) {
    // Broadcasts typing.started, auto-stops after 5s
  }

  handleTypingStop(roomId: string, userId: string) {
    // Broadcasts typing.stopped
  }

  broadcastAiThinking(roomId: string, isThinking: boolean) {
    // Broadcasts ai.thinking events
  }
}

SseController

apps/nestjs/src/chat/sse/sse.controller.ts

@ApiTags('sse')
@Controller('/v1/chat/sse')
export class SseController {
  @Sse('events/:roomId')
  events(@CurrentUser() user: AuthUser, @Param('roomId') roomId: string): Observable<MessageEvent> {
    const client = this.sseService.subscribe(user.id, roomId)
    return client.subscriber.pipe(
      map(({ event, data }) => new MessageEvent(event, { data: JSON.stringify(data) }))
    )
  }

  @Post('typing')
  broadcastTyping(@CurrentUser() user: AuthUser, @Query() query: SseTypingRequestDto) {
    // POST /v1/chat/sse/typing?type=typing.started&roomId=room_123
  }
}

Fern + SSE

Fern generates a generic Response object for SSE endpoints — there’s no native SSE support. You’d handle it manually:

// What Fern generates for GET /v1/chat/sse/events/:roomId
const response = await client.chat.sse.events({ roomId: 'room_123' })

// Manual SSE parsing required
const reader = response.body?.getReader()
const decoder = new TextDecoder()
while (true) {
  const { done, value } = await reader!.read()
  if (done) break
  const chunk = decoder.decode(value)
  // Parse SSE format: "event: typing.started\ndata: {...}\n\n"
  console.log(chunk)
}

Recommendation: For SSE in mobile apps, use a native EventSource or fetch with ReadableStream instead of the Fern-generated client. Fern is better suited for REST endpoints with JSON responses.

Auto-Stop Typing

The SseService automatically stops typing after 5 seconds to handle cases where the client doesn’t explicitly send typing.stopped:

const timer = setTimeout(() => {
  this.handleTypingStop(roomId, userId, displayName)
}, TYPING_TIMEOUT_MS) // 5000ms

DTOs and Validation

All DTOs use class-validator for runtime validation:

export class ChatSendMessageClientEventDto {
  @Equals(ChatClientEventType.MessageSend)
  type!: ChatClientEventType.MessageSend

  @IsString()
  @MinLength(1)
  @MaxLength(128)
  requestId!: string

  @IsString()
  @MinLength(1)
  @MaxLength(2000)
  content!: string
}

Validation is applied globally via ValidationPipe in main.ts.

Media / R2 Storage

apps/nestjs/src/media/ and apps/nestjs/src/storage/ Cloudflare R2 integration for image and voice message storage. Uses S3-compatible API with signed URLs for direct client uploads.

MediaController

apps/nestjs/src/media/media.controller.ts REST API for generating signed upload URLs:

@ApiTags('media')
@Controller('/v1/media')
export class MediaController {
  @Post('upload-url')         // Get signed URL for image upload
  @Post('voice/upload-url')   // Get signed URL for voice message upload
}

R2Service

apps/nestjs/src/storage/r2.service.ts S3-compatible R2 operations:

Method	Description
`generateKey()`	Generates unique key: `images/{roomId}/{userId}/{timestamp}-{random}.{ext}`
`getSignedUploadUrl()`	Returns signed URL for direct client upload
`uploadFile()`	Server-side upload via HTTP
`deleteFile()`	Remove file from R2
`getPublicUrl()`	Get public URL for a file

Local development returns mock URLs pointing to localhost:3001/r2/upload/* when R2 credentials are not configured.

R2 Environment Variables

Variable	Description
`R2_ACCOUNT_ID`	Cloudflare account ID
`R2_ACCESS_KEY_ID`	R2 API access key
`R2_SECRET_ACCESS_KEY`	R2 API secret
`R2_BUCKET_NAME`	Bucket name (default: `nous-media`)
`R2_PUBLIC_URL`	Public URL base (e.g., `http://localhost:3001/r2`)

Authentication

Temporary auth via x-user-id header (JWT pending):

@ApiHeader({ name: 'x-user-id', required: true })
@UseGuards(AuthUserGuard)
@Controller('/v1/chat')
export class ChatController { ... }

WebSocket auth extracts from query params or headers:

async handleConnection(client: Socket) {
  const userId = this.readQueryValue(client, 'userId')
    ?? this.readHeaderValue(client, 'x-user-id')
  // ...
}

API Documentation

OpenAPI (REST)

UI: /docs
JSON: /openapi.json

AsyncAPI (WebSocket)

UI: /asyncapi
JSON: /asyncapi.json
Exported: fern/generated/asyncapi.json

Environment Variables

Variable	Default	Description
`PORT`	`3000`	Server port
`HOST`	`0.0.0.0`	Server host
`DATABASE_URL`	—	PostgreSQL connection string
`DATABASE_SSL`	`true`	Require SSL for PostgreSQL
`DATABASE_POOL_MAX`	`5`	Max pool size (1-20)
`MASTRA_PREFIX`	`/api`	Mastra route prefix
`MASTRA_MODEL`	`MiniMax-M2.7`	AI model name
`THROTTLE_TTL_MS`	`60000`	Throttle window (ms)
`THROTTLE_LIMIT`	`120`	Max requests per window
`CORS_ORIGINS`	`*`	Comma-separated allowed origins
`AI_TOKEN_MX2`	—	MiniMax API token
`AI_URL_MX2`	`https://api.minimax.io/anthropic/v1`	MiniMax API URL
`R2_ACCOUNT_ID`	—	Cloudflare R2 account ID
`R2_ACCESS_KEY_ID`	—	R2 API access key
`R2_SECRET_ACCESS_KEY`	—	R2 API secret
`R2_BUCKET_NAME`	`nous-media`	R2 bucket name
`R2_PUBLIC_URL`	—	Public URL base for local dev

Project Structure

apps/nestjs/
├── src/
│   ├── main.ts                    # Bootstrap, global pipes, asyncapi setup
│   ├── app.module.ts              # Root module with Mastra, Throttler
│   ├── asyncapi.ts                # AsyncAPI document builder
│   ├── openapi.ts                 # OpenAPI document builder
│   ├── chat/
│   │   ├── chat.module.ts         # Chat feature module
│   │   ├── chat.controller.ts     # REST API controller
│   │   ├── chat.service.ts        # Business logic
│   │   ├── chat.mapper.ts         # Entity → DTO mapping
│   │   ├── chat.constants.ts      # Constants (VICKY_PARTICIPANT_ID)
│   │   ├── dto/
│   │   │   ├── chat.dto.ts        # All DTOs
│   │   │   └── chat-params.dto.ts # Route params
│   │   ├── entities/
│   │   │   └── chat.entities.ts   # Domain entities
│   │   ├── realtime/
│   │   │   ├── chat-realtime.gateway.ts      # Socket.IO gateway
│   │   │   ├── chat-realtime.service.ts      # Event publisher
│   │   │   └── chat-realtime.gateway.test.ts
│   │   ├── sse/
│   │   │   ├── sse.controller.ts              # SSE endpoints
│   │   │   ├── sse.service.ts                # Subscription management
│   │   │   └── sse.dto.ts                    # SSE event DTOs
│   │   ├── repositories/
│   │   │   ├── chat.repository.ts            # Repository interface
│   │   │   ├── in-memory-chat.repository.ts  # Dev implementation
│   │   │   └── postgres-chat.repository.ts   # Prod implementation
│   │   └── ai/
│   │       └── ai-participant.service.ts     # Vicky AI integration
│   ├── media/
│   │   ├── media.module.ts         # R2 storage module
│   │   ├── media.controller.ts    # Signed URL endpoints
│   │   └── media.dto.ts            # Media DTOs
│   ├── storage/
│   │   └── r2.service.ts          # Cloudflare R2 S3-compatible API
│   ├── auth/
│   │   ├── auth-user.guard.ts    # Auth guard
│   │   ├── current-user.decorator.ts
│   │   └── auth-user.ts          # Auth user type
│   ├── database/
│   │   ├── database.module.ts
│   │   └── database.service.ts   # DB connection management
│   ├── mastra/
│   │   ├── index.ts              # Mastra instance
│   │   └── agents/
│   │       ├── vicky-agent.ts
│   │       └── weather-agent.ts
│   └── config/
│       └── env.ts                # Env validation
├── scripts/
│   └── export-asyncapi.ts        # Export AsyncAPI to fern/
└── package.json

Running Locally

cd apps/nestjs
pnpm install
pnpm start:dev

The server starts on http://localhost:3000 with:

REST API: http://localhost:3000/v1/chat/*
WebSocket: ws://localhost:3000/chat
SSE: http://localhost:3000/v1/chat/sse/events/:roomId
OpenAPI UI: http://localhost:3000/docs
AsyncAPI UI: http://localhost:3000/asyncapi

Testing

REST API

# Create a room
curl -X POST http://localhost:3000/v1/chat/rooms \
  -H "Content-Type: application/json" \
  -H "x-user-id: user_alice" \
  -d '{"vickyMode":"off","humanParticipants":[{"userId":"user_alice","displayName":"Alice"},{"userId":"user_bob","displayName":"Bob"}]}'

# Get room
curl http://localhost:3000/v1/chat/rooms/:roomId \
  -H "x-user-id: user_alice"

SSE (Server-Sent Events)

# Subscribe to room events
curl -N http://localhost:3000/v1/chat/sse/events/:roomId \
  -H "x-user-id: user_alice"

# Broadcast typing
curl -X POST "http://localhost:3000/v1/chat/sse/typing?roomId=:roomId&type=typing.started" \
  -H "x-user-id: user_alice"

SSE events: typing.started, typing.stopped, presence.update, ai.thinking

WebSocket (Socket.IO)

Use socket.io-client in your app:

import { io } from 'socket.io-client'

const socket = io('http://localhost:3000/chat', {
  transports: ['websocket'],
  query: { roomId: 'room_id', userId: 'user_alice' }
})

socket.on('connect', () => {
  // Send ping
  socket.emit('ping', {
    type: 'ping',
    requestId: 'test1',
    sentAt: new Date().toISOString()
  })
})

socket.on('pong', (data) => console.log('Pong:', data))
socket.on('ack', (data) => console.log('Ack:', data))
socket.on('error', (data) => console.error('Error:', data))
socket.on('message.created', (data) => console.log('New message:', data))

Last Updated: 2026-05-14

​NestJS Integration

​Deployment

​Overview

​Architecture

​Data Flow

​REST API Flow

​WebSocket Flow

​AI Reply Flow

​Key Components

​ChatRealtimeGateway

​ChatRealtimeService

​ChatController

​Repository Pattern

​DatabaseService

​AsyncAPI Integration

​Overview

​Decorator Usage

​Export Script

​Server-Sent Events (SSE)

​SSE Endpoints

​Event Types

​SseService

​SseController

​Fern + SSE

​Auto-Stop Typing

​DTOs and Validation

​Media / R2 Storage

​MediaController

​R2Service

​R2 Environment Variables

​Authentication

​API Documentation

​OpenAPI (REST)

​AsyncAPI (WebSocket)

​Environment Variables

​Project Structure

​Running Locally

​Testing

​REST API

​SSE (Server-Sent Events)

​WebSocket (Socket.IO)

NestJS Integration

Deployment

Overview

Architecture

Data Flow

REST API Flow

WebSocket Flow

AI Reply Flow

Key Components

ChatRealtimeGateway

ChatRealtimeService

ChatController

Repository Pattern

DatabaseService

AsyncAPI Integration

Overview

Decorator Usage

Export Script

Server-Sent Events (SSE)

SSE Endpoints

Event Types

SseService

SseController

Fern + SSE

Auto-Stop Typing

DTOs and Validation

Media / R2 Storage

MediaController

R2Service

R2 Environment Variables

Authentication

API Documentation

OpenAPI (REST)

AsyncAPI (WebSocket)

Environment Variables

Project Structure

Running Locally

Testing

REST API

SSE (Server-Sent Events)

WebSocket (Socket.IO)