feat(api): add stream CRUD REST endpoints (#5)

api/src/app.module.ts

@@ -5,6 +5,7 @@ import { AdminModule } from "./admin/admin.module"
 import { AuditModule } from "./audit/audit.module"
 import { GatewaysModule } from "./gateways/gateways.module"
 import { HealthModule } from "./health/health.module"
+import { StreamsModule } from "./streams/streams.module"
 import { TagsModule } from "./tags/tags.module"
 
 @Module({
@@ -19,6 +20,7 @@ import { TagsModule } from "./tags/tags.module"
     AuditModule,
     GatewaysModule,
     HealthModule,
+    StreamsModule,
     TagsModule,
   ],
   providers: [

api/src/common/guards/auth.guard.ts

@@ -0,0 +1,39 @@
+import {
+  CanActivate,
+  ExecutionContext,
+  Injectable,
+  UnauthorizedException,
+} from "@nestjs/common"
+import type { Request } from "express"
+
+/**
+ * Lightweight auth guard that extracts the authenticated user id from the
+ * `X-User-Id` header. This is a placeholder until the full JWT auth
+ * pipeline lands — at that point the guard will read `req.user.sub`
+ * instead.
+ *
+ * Apply with `@UseGuards(AuthGuard)` on controllers or individual
+ * handlers that require authentication.
+ */
+@Injectable()
+export class AuthGuard implements CanActivate {
+  canActivate(context: ExecutionContext): boolean {
+    const req = context.switchToHttp().getRequest<Request>()
+
+    const rawUserId = (req.header("x-user-id") ?? "").trim()
+    if (!rawUserId) {
+      throw new UnauthorizedException(
+        "X-User-Id header is required (placeholder until JWT auth lands)",
+      )
+    }
+    const userId = Number(rawUserId)
+    if (!Number.isInteger(userId) || userId <= 0) {
+      throw new UnauthorizedException("X-User-Id must be a positive integer")
+    }
+
+    // Stash on the request so downstream handlers / guards can access
+    // the authenticated user without re-parsing.
+    ;(req as Request & { auth?: { userId: number } }).auth = { userId }
+    return true
+  }
+}

api/src/main.ts

@@ -81,6 +81,7 @@ async function bootstrap() {
       "bearer",
     )
     .addTag("health", "Liveness and readiness probes")
+    .addTag("streams", "Stream lifecycle CRUD")
     .build()
 
   const document = SwaggerModule.createDocument(app, swaggerConfig)

api/src/streams/dto/create-stream.dto.ts

@@ -0,0 +1,19 @@
+import { IsOptional, IsString, Length, MaxLength } from "class-validator"
+
+/**
+ * Payload accepted by `POST /streams`.
+ */
+export class CreateStreamDto {
+  @IsString()
+  @Length(1, 255, {
+    message: "name must be between 1 and 255 characters",
+  })
+  name!: string
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(2000, {
+    message: "description must be at most 2000 characters",
+  })
+  description?: string
+}

api/src/streams/dto/list-streams.query.dto.ts

@@ -0,0 +1,23 @@
+import { ApiPropertyOptional } from "@nestjs/swagger"
+import { Type } from "class-transformer"
+import { IsIn, IsOptional, IsString } from "class-validator"
+import { PaginationQueryDto } from "../../common/dto/pagination.dto"
+
+/**
+ * Query parameters for `GET /streams`.
+ *
+ * Extends the shared {@link PaginationQueryDto} so paging behaviour is
+ * consistent with the rest of the API. Adds an optional `status` filter.
+ */
+export class ListStreamsQueryDto extends PaginationQueryDto {
+  @ApiPropertyOptional({
+    description: "Filter streams by status (inactive, active, error)",
+    example: "active",
+  })
+  @IsOptional()
+  @IsString()
+  @IsIn(["inactive", "active", "error"], {
+    message: "status must be one of: inactive, active, error",
+  })
+  status?: string
+}

api/src/streams/dto/update-stream.dto.ts

@@ -0,0 +1,28 @@
+import { IsOptional, IsString, Length, MaxLength } from "class-validator"
+
+/**
+ * Payload accepted by `PATCH /streams/:id`. All fields are optional;
+ * only the supplied fields are updated.
+ */
+export class UpdateStreamDto {
+  @IsOptional()
+  @IsString()
+  @Length(1, 255, {
+    message: "name must be between 1 and 255 characters",
+  })
+  name?: string
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(2000, {
+    message: "description must be at most 2000 characters",
+  })
+  description?: string
+
+  @IsOptional()
+  @IsString()
+  @Length(1, 50, {
+    message: "status must be between 1 and 50 characters",
+  })
+  status?: string
+}

api/src/streams/repository/streams.repository.ts

@@ -0,0 +1,81 @@
+import { Injectable } from "@nestjs/common"
+import { Stream } from "../stream.entity"
+
+/**
+ * In-memory streams repository.
+ *
+ * Persistence-agnostic: the controller and service depend only on the
+ * public methods exposed here. When the Postgres layer is ready this
+ * class can be swapped for a DB-backed repository without changing
+ * higher layers.
+ */
+@Injectable()
+export class StreamsRepository {
+  private readonly streamsById = new Map<number, Stream>()
+  private nextId = 1
+
+  findById(id: number): Stream | undefined {
+    return this.streamsById.get(id)
+  }
+
+  /**
+   * Returns all streams, optionally filtered by status.
+   */
+  listFiltered(filter?: { status?: string }): Stream[] {
+    let results = Array.from(this.streamsById.values())
+    if (filter?.status) {
+      results = results.filter((s) => s.status === filter.status)
+    }
+    return results.sort(
+      (a, b) => b.createdAt.getTime() - a.createdAt.getTime(),
+    )
+  }
+
+  /**
+   * Paginated listing.
+   */
+  listPaginated(
+    page: number,
+    limit: number,
+    filter?: { status?: string },
+  ): { items: Stream[]; total: number } {
+    const filtered = this.listFiltered(filter)
+    const offset = (page - 1) * limit
+    return {
+      items: filtered.slice(offset, offset + limit),
+      total: filtered.length,
+    }
+  }
+
+  create(params: { userId: number; name: string; description?: string }): Stream {
+    const stream: Stream = {
+      id: this.nextId++,
+      userId: params.userId,
+      name: params.name,
+      description: params.description ?? null,
+      status: "inactive",
+      createdAt: new Date(),
+      updatedAt: new Date(),
+    }
+    this.streamsById.set(stream.id, stream)
+    return stream
+  }
+
+  update(
+    id: number,
+    changes: { name?: string; description?: string; status?: string },
+  ): Stream {
+    const stream = this.streamsById.get(id)!
+    if (changes.name !== undefined) stream.name = changes.name
+    if (changes.description !== undefined) stream.description = changes.description
+    if (changes.status !== undefined) {
+      stream.status = changes.status as Stream["status"]
+    }
+    stream.updatedAt = new Date()
+    return stream
+  }
+
+  delete(id: number): boolean {
+    return this.streamsById.delete(id)
+  }
+}

api/src/streams/stream.entity.ts

@@ -0,0 +1,15 @@
+/**
+ * In-memory representation of a stream. Mirrors the `streams` table
+ * defined in `database/schema.sql`. The controller and service layers
+ * depend on this interface so they stay unchanged when the repository
+ * is swapped for a real DB-backed implementation.
+ */
+export interface Stream {
+  id: number
+  userId: number
+  name: string
+  description: string | null
+  status: "inactive" | "active" | "error"
+  createdAt: Date
+  updatedAt: Date
+}

api/src/streams/streams.controller.ts

@@ -0,0 +1,139 @@
+import {
+  Body,
+  Controller,
+  Delete,
+  Get,
+  HttpCode,
+  HttpStatus,
+  Param,
+  ParseIntPipe,
+  Patch,
+  Post,
+  Query,
+  Req,
+  UseGuards,
+} from "@nestjs/common"
+import {
+  ApiBearerAuth,
+  ApiConflictResponse,
+  ApiCreatedResponse,
+  ApiNoContentResponse,
+  ApiNotFoundResponse,
+  ApiOkResponse,
+  ApiOperation,
+  ApiTags,
+} from "@nestjs/swagger"
+import type { Request } from "express"
+import { AuthGuard } from "../common/guards/auth.guard"
+import { StreamOwnershipGuard } from "../common/guards/stream-ownership.guard"
+import { CreateStreamDto } from "./dto/create-stream.dto"
+import { ListStreamsQueryDto } from "./dto/list-streams.query.dto"
+import { UpdateStreamDto } from "./dto/update-stream.dto"
+import { StreamsService } from "./streams.service"
+
+/**
+ * Full CRUD for streams.
+ *
+ *   POST   /streams          Create a new stream (auth required)
+ *   GET    /streams          List streams (auth required, paginated)
+ *   GET    /streams/:id      Get a single stream (ownership required)
+ *   PATCH  /streams/:id      Update stream details (ownership required)
+ *   DELETE /streams/:id      Delete a stream (ownership required)
+ */
+@ApiTags("streams")
+@ApiBearerAuth()
+@Controller("streams")
+export class StreamsController {
+  constructor(private readonly streamsService: StreamsService) {}
+
+  /**
+   * Create a new stream. The authenticated user becomes the owner.
+   */
+  @Post()
+  @HttpCode(HttpStatus.CREATED)
+  @UseGuards(AuthGuard)
+  @ApiOperation({
+    summary: "Create a new stream",
+    description: "Creates a new stream with the authenticated user as owner.",
+  })
+  @ApiCreatedResponse({ description: "Stream created successfully." })
+  create(
+    @Body() body: CreateStreamDto,
+    @Req() req: Request & { auth?: { userId: number } },
+  ) {
+    return this.streamsService.create({
+      userId: req.auth!.userId,
+      name: body.name,
+      description: body.description,
+    })
+  }
+
+  /**
+   * List all streams with optional status filter and pagination.
+   */
+  @Get()
+  @UseGuards(AuthGuard)
+  @ApiOperation({
+    summary: "List streams",
+    description: "Returns a paginated list of streams with optional status filter.",
+  })
+  @ApiOkResponse({ description: "Paginated list of streams." })
+  list(@Query() query: ListStreamsQueryDto) {
+    const page = query.page ?? 1
+    const limit = query.limit ?? 20
+    return this.streamsService.list(page, limit, {
+      status: query.status,
+    })
+  }
+
+  /**
+   * Get a single stream by id. Requires stream ownership.
+   */
+  @Get(":id")
+  @UseGuards(StreamOwnershipGuard)
+  @ApiOperation({
+    summary: "Get a stream",
+    description: "Returns a single stream by id. Requires ownership.",
+  })
+  @ApiOkResponse({ description: "Stream found." })
+  @ApiNotFoundResponse({ description: "Stream not found." })
+  findById(@Param("id", ParseIntPipe) id: number) {
+    return this.streamsService.findById(id)
+  }
+
+  /**
+   * Update stream details (name, description, status).
+   * Requires stream ownership.
+   */
+  @Patch(":id")
+  @UseGuards(StreamOwnershipGuard)
+  @ApiOperation({
+    summary: "Update a stream",
+    description: "Partially updates a stream. Requires ownership.",
+  })
+  @ApiOkResponse({ description: "Stream updated." })
+  @ApiNotFoundResponse({ description: "Stream not found." })
+  @ApiConflictResponse({ description: "Invalid status transition." })
+  update(
+    @Param("id", ParseIntPipe) id: number,
+    @Body() body: UpdateStreamDto,
+  ) {
+    return this.streamsService.update(id, body)
+  }
+
+  /**
+   * Delete a stream. Requires stream ownership.
+   */
+  @Delete(":id")
+  @HttpCode(HttpStatus.NO_CONTENT)
+  @UseGuards(StreamOwnershipGuard)
+  @ApiOperation({
+    summary: "Delete a stream",
+    description: "Deletes a stream by id. Requires ownership.",
+  })
+  @ApiNoContentResponse({ description: "Stream deleted." })
+  @ApiNotFoundResponse({ description: "Stream not found." })
+  delete(@Param("id", ParseIntPipe) id: number): void {
+    this.streamsService.delete(id)
+  }
+}

api/src/streams/streams.module.ts

@@ -0,0 +1,20 @@
+import { Module } from "@nestjs/common"
+import { AuthGuard } from "../common/guards/auth.guard"
+import { StreamOwnershipGuard } from "../common/guards/stream-ownership.guard"
+import { StreamOwnershipService } from "../common/guards/stream-ownership.service"
+import { StreamsRepository } from "./repository/streams.repository"
+import { StreamsController } from "./streams.controller"
+import { StreamsService } from "./streams.service"
+
+@Module({
+  controllers: [StreamsController],
+  providers: [
+    StreamsService,
+    StreamsRepository,
+    AuthGuard,
+    StreamOwnershipGuard,
+    StreamOwnershipService,
+  ],
+  exports: [StreamsService],
+})
+export class StreamsModule {}