diff --git a/api/src/admin/admin-stats.service.ts b/api/src/admin/admin-stats.service.ts index 038f794..7b7fb34 100644 --- a/api/src/admin/admin-stats.service.ts +++ b/api/src/admin/admin-stats.service.ts @@ -1,4 +1,12 @@ -import { Injectable } from "@nestjs/common" +import { + Inject, + Injectable, + Logger, + ServiceUnavailableException, +} from "@nestjs/common" +import { Pool } from "pg" + +import { PG_POOL } from "../database/database.module" export interface AdminStats { totalUsers: number @@ -11,13 +19,10 @@ export interface AdminStats { /** * Aggregates platform-wide stats for the admin dashboard. * - * The current implementation returns a deterministic, zero-valued - * snapshot — it exists so the endpoint, guard, and 60-second cache - * can be wired and exercised end-to-end before the Postgres data - * layer lands. - * - * When the DB layer is available the four numeric fields will be - * computed via the following aggregate queries (single round-trip): + * The four numeric fields are computed with aggregate subqueries in a + * single database round-trip. Active streams are those whose status is + * `active` at query time, and eventsLast24h covers events created within + * the preceding 24 hours. * * SELECT * (SELECT COUNT(*) FROM users) AS total_users, @@ -31,15 +36,41 @@ export interface AdminStats { */ @Injectable() export class AdminStatsService { + private readonly logger = new Logger(AdminStatsService.name) + + constructor(@Inject(PG_POOL) private readonly pool: Pool) {} + + private handleDbError(err: unknown): never { + this.logger.error("DB error in compute", (err as Error).stack) + throw new ServiceUnavailableException( + "Database is unavailable. Please try again later.", + ) + } + async compute(): Promise { - // Placeholder zero snapshot. The query above replaces this body - // once the DB module is wired. - return { - totalUsers: 0, - totalStreams: 0, - activeStreams: 0, - eventsLast24h: 0, - generatedAt: new Date().toISOString(), + try { + const { rows } = await this.pool.query<{ + total_users: number + total_streams: number + active_streams: number + events_24h: number + }>(`SELECT + (SELECT COUNT(*)::int FROM users) AS total_users, + (SELECT COUNT(*)::int FROM streams) AS total_streams, + (SELECT COUNT(*)::int FROM streams WHERE status = 'active') AS active_streams, + (SELECT COUNT(*)::int FROM stream_events + WHERE created_at > NOW() - INTERVAL '24 hours') AS events_24h`) + + const row = rows[0] + return { + totalUsers: Number(row?.total_users ?? 0), + totalStreams: Number(row?.total_streams ?? 0), + activeStreams: Number(row?.active_streams ?? 0), + eventsLast24h: Number(row?.events_24h ?? 0), + generatedAt: new Date().toISOString(), + } + } catch (err) { + this.handleDbError(err) } } } diff --git a/api/src/admin/admin.controller.ts b/api/src/admin/admin.controller.ts index 7df2a76..761f4b7 100644 --- a/api/src/admin/admin.controller.ts +++ b/api/src/admin/admin.controller.ts @@ -45,7 +45,8 @@ export class AdminController { @ApiBearerAuth("bearer") @ApiOperation({ summary: "Get platform-wide statistics", - description: "Returns cached aggregate platform metrics. Admin role required.", + description: + "Returns cached aggregate platform metrics: total users, total streams, active streams (status = 'active'), and stream events created within the last 24 hours. Admin role required.", }) @ApiOkResponse({ description: "Admin platform statistics." }) @ApiUnauthorizedResponse({ description: "Authentication required." }) diff --git a/api/src/database.integration.spec.ts b/api/src/database.integration.spec.ts index 7e2df49..de91180 100644 --- a/api/src/database.integration.spec.ts +++ b/api/src/database.integration.spec.ts @@ -6,6 +6,7 @@ import { destroyTestApp, TestAppContext, } from "./database/test-utils" +import { AdminStatsService } from "./admin/admin-stats.service" import { StreamsDbRepository } from "./streams/repository/streams-db.repository" describe("Database Integration Tests", () => { @@ -306,6 +307,43 @@ describe("Database Integration Tests", () => { }) }) + describe("Admin statistics", () => { + it("counts users, streams, active streams, and recent events", async () => { + const users = await pool.query<{ id: number }>( + `INSERT INTO users (username, email, password_hash) + VALUES ('stats-user-1', 'stats-1@test.com', 'hash'), + ('stats-user-2', 'stats-2@test.com', 'hash') + RETURNING id`, + ) + + const streams = await pool.query<{ id: number }>( + `INSERT INTO streams (user_id, name, status) + VALUES ($1, 'Active Stream 1', 'active'), + ($2, 'Active Stream 2', 'active'), + ($1, 'Inactive Stream', 'inactive') + RETURNING id`, + [users.rows[0].id, users.rows[1].id], + ) + + await pool.query( + `INSERT INTO stream_events + (stream_id, event_type, event_data, created_at) + VALUES ($1, 'recent-1', '{}'::jsonb, NOW()), + ($2, 'recent-2', '{}'::jsonb, NOW() - INTERVAL '1 hour'), + ($3, 'old', '{}'::jsonb, NOW() - INTERVAL '25 hours')`, + streams.rows.map((row) => row.id), + ) + + const stats = await new AdminStatsService(pool).compute() + + expect(stats.totalUsers).toBe(2) + expect(stats.totalStreams).toBe(3) + expect(stats.activeStreams).toBe(2) + expect(stats.eventsLast24h).toBe(2) + expect(stats.generatedAt).toEqual(expect.any(String)) + }) + }) + describe("Foreign Key Constraints", () => { it("prevents creating a stream with a non-existent user_id", async () => { await expect(