diff --git a/api/src/app.module.ts b/api/src/app.module.ts index 32817d2..e330700 100644 --- a/api/src/app.module.ts +++ b/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: [ diff --git a/api/src/common/guards/auth.guard.ts b/api/src/common/guards/auth.guard.ts new file mode 100644 index 0000000..fb0befb --- /dev/null +++ b/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() + + 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 + } +} diff --git a/api/src/main.ts b/api/src/main.ts index 81eb472..42ab0a3 100644 --- a/api/src/main.ts +++ b/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) diff --git a/api/src/streams/dto/create-stream.dto.ts b/api/src/streams/dto/create-stream.dto.ts new file mode 100644 index 0000000..e79e217 --- /dev/null +++ b/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 +} diff --git a/api/src/streams/dto/list-streams.query.dto.ts b/api/src/streams/dto/list-streams.query.dto.ts new file mode 100644 index 0000000..6cd19dd --- /dev/null +++ b/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 +} diff --git a/api/src/streams/dto/update-stream.dto.ts b/api/src/streams/dto/update-stream.dto.ts new file mode 100644 index 0000000..b874bcf --- /dev/null +++ b/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 +} diff --git a/api/src/streams/repository/streams.repository.ts b/api/src/streams/repository/streams.repository.ts new file mode 100644 index 0000000..83bcfa6 --- /dev/null +++ b/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() + 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) + } +} diff --git a/api/src/streams/stream.entity.ts b/api/src/streams/stream.entity.ts new file mode 100644 index 0000000..e0625bb --- /dev/null +++ b/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 +} diff --git a/api/src/streams/streams.controller.ts b/api/src/streams/streams.controller.ts new file mode 100644 index 0000000..c5d0718 --- /dev/null +++ b/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) + } +} diff --git a/api/src/streams/streams.module.ts b/api/src/streams/streams.module.ts new file mode 100644 index 0000000..2503322 --- /dev/null +++ b/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 {} diff --git a/api/src/streams/streams.service.ts b/api/src/streams/streams.service.ts new file mode 100644 index 0000000..1c8cc56 --- /dev/null +++ b/api/src/streams/streams.service.ts @@ -0,0 +1,98 @@ +import { + ConflictException, + Injectable, + NotFoundException, +} from "@nestjs/common" +import { PaginatedResult } from "../common/dto/pagination.dto" +import { Stream } from "./stream.entity" +import { StreamsRepository } from "./repository/streams.repository" + +export interface PagedStreams extends PaginatedResult { + hasMore: boolean +} + +@Injectable() +export class StreamsService { + constructor(private readonly repo: StreamsRepository) {} + + create(dto: { userId: number; name: string; description?: string }): Stream { + return this.repo.create({ + userId: dto.userId, + name: dto.name.trim(), + description: dto.description?.trim(), + }) + } + + list( + page: number, + limit: number, + filter?: { status?: string }, + ): PagedStreams { + const { items, total } = this.repo.listPaginated(page, limit, filter) + return { + data: items, + page, + limit, + total, + hasMore: page * limit < total, + } + } + + findById(id: number): Stream { + const stream = this.repo.findById(id) + if (!stream) { + throw new NotFoundException(`stream ${id} not found`) + } + return stream + } + + update( + id: number, + changes: { name?: string; description?: string; status?: string }, + ): Stream { + const stream = this.findById(id) + + // Validate status transitions + if (changes.status !== undefined) { + this.validateStatusTransition(stream.status, changes.status) + } + + return this.repo.update(id, { + name: changes.name?.trim(), + description: changes.description?.trim(), + status: changes.status, + }) + } + + delete(id: number): void { + const exists = this.repo.delete(id) + if (!exists) { + throw new NotFoundException(`stream ${id} not found`) + } + } + + /** + * Enforces valid status transitions: + * inactive → active (start streaming) + * active → inactive (stop streaming) + * * → error (any status can transition to error) + * error → inactive (recover from error) + */ + private validateStatusTransition( + current: string, + next: string, + ): void { + const allowed: Record = { + inactive: ["active", "error"], + active: ["inactive", "error"], + error: ["inactive"], + } + + const allowedTransitions = allowed[current] + if (!allowedTransitions?.includes(next)) { + throw new ConflictException( + `cannot transition stream from "${current}" to "${next}"`, + ) + } + } +}