fix(processing): add per-step timeout enforcement to gracefulshutdown

api/src/audit/audit-action.enum.ts

@@ -24,4 +24,8 @@ export enum AuditAction {
   STREAM_DELETE = "stream_delete",
   ROLE_CHANGE = "role_change",
   PROFILE_UPDATE = "profile_update",
+
+  // ── User lifecycle ────────────────────────────────────────────────
+  /** Captured when a user soft-deletes their account (issue #344). */
+  USER_DELETED = "user_deleted",
 }

api/src/auth/users.repository.ts

@@ -1,5 +1,6 @@
 import { Inject, Injectable } from "@nestjs/common"
 import { Pool } from "pg"
+
 import { PG_POOL } from "../database/database.module"
 
 export interface User {
@@ -9,6 +10,7 @@ export interface User {
   password_hash: string
   created_at: Date
   password_changed_at?: Date
+  deleted_at?: Date | null
 }
 
 /**
@@ -23,23 +25,23 @@ export class UsersRepository {
 
   async findByEmail(email: string): Promise<User | null> {
     const { rows } = await this.pool.query(
-      "SELECT id, username, email, password_hash, created_at, password_changed_at FROM users WHERE email = $1",
+      "SELECT id, username, email, password_hash, created_at, password_changed_at, deleted_at FROM users WHERE email = $1 AND deleted_at IS NULL",
       [email],
     )
     return rows[0] ?? null
   }
 
   async findByUsername(username: string): Promise<User | null> {
     const { rows } = await this.pool.query(
-      "SELECT id, username, email, password_hash, created_at, password_changed_at FROM users WHERE username = $1",
+      "SELECT id, username, email, password_hash, created_at, password_changed_at, deleted_at FROM users WHERE username = $1 AND deleted_at IS NULL",
       [username],
     )
     return rows[0] ?? null
   }
 
   async findById(id: number): Promise<User | null> {
     const { rows } = await this.pool.query(
-      "SELECT id, username, email, password_hash, created_at, password_changed_at FROM users WHERE id = $1",
+      "SELECT id, username, email, password_hash, created_at, password_changed_at, deleted_at FROM users WHERE id = $1 AND deleted_at IS NULL",
       [id],
     )
     return rows[0] ?? null
@@ -100,10 +102,26 @@ export class UsersRepository {
       `UPDATE users
        SET password_hash = $1,
            password_changed_at = $2
-       WHERE id = $3
-       RETURNING id, username, email, password_hash, created_at, password_changed_at`,
+       WHERE id = $3 AND deleted_at IS NULL
+       RETURNING id, username, email, password_hash, created_at, password_changed_at, deleted_at`,
       [passwordHash, passwordChangedAt, id],
     )
     return rows[0]
   }
+
+  /**
+   * Soft-delete a user by setting deleted_at (issue #344).
+   * Returns the soft-deleted user row or null if the user was not
+   * found or was already deleted.
+   */
+  async softDelete(id: number): Promise<User | null> {
+    const { rows } = await this.pool.query(
+      `UPDATE users
+       SET deleted_at = NOW()
+       WHERE id = $1 AND deleted_at IS NULL
+       RETURNING id, username, email, password_hash, created_at, password_changed_at, deleted_at`,
+      [id],
+    )
+    return rows[0] ?? null
+  }
 }

api/src/users/users.controller.ts

@@ -1,6 +1,7 @@
 import {
   Body,
   Controller,
+  Delete,
   Get,
   HttpCode,
   HttpStatus,
@@ -9,17 +10,21 @@ import {
   Req,
   UseGuards,
 } from "@nestjs/common"
-import { Throttle } from "@nestjs/throttler"
 import {
+  ApiNoContentResponse,
   ApiOkResponse,
   ApiOperation,
   ApiTags,
 } from "@nestjs/swagger"
-import type { Request } from "express"
-import { AuthGuard } from "../common/guards/auth.guard"
+import { Throttle } from "@nestjs/throttler"
+
+
 import { ChangePasswordDto } from "./dto/change-password.dto"
 import { UpdateProfileDto } from "./dto/update-profile.dto"
 import { ProfileResponse, UsersService } from "./users.service"
+import { AuthGuard } from "../common/guards/auth.guard"
+
+import type { Request } from "express"
 
 @ApiTags("users")
 @UseGuards(AuthGuard)
@@ -88,4 +93,39 @@ export class UsersController {
       req.header("authorization") ?? "",
     )
   }
+
+  @Delete("me")
+  @HttpCode(HttpStatus.NO_CONTENT)
+  @Throttle({ default: { limit: 3, ttl: 60000 } })
+  @ApiOperation({
+    summary: "Delete user account",
+    description:
+      "Soft-deletes the authenticated user's account. " +
+      "Sets deleted_at on the user row and revokes the current token. " +
+      "Audit logs and other historical data are preserved with the " +
+      "user_id set to NULL.",
+  })
+  @ApiNoContentResponse({
+    description: "Account soft-deleted successfully.",
+  })
+  async deleteAccount(@Req() req: Request): Promise<void> {
+    const { userId } = (req as Request & { auth: { userId: number } }).auth
+    const ip = this.extractClientIp(req)
+    await this.usersService.deleteAccount(
+      userId,
+      req.header("authorization") ?? "",
+      ip,
+    )
+  }
+
+  private extractClientIp(req: Request): string {
+    const xForwardedFor = req.headers["x-forwarded-for"]
+    if (typeof xForwardedFor === "string") {
+      return xForwardedFor.split(",")[0].trim()
+    }
+    if (Array.isArray(xForwardedFor)) {
+      return xForwardedFor[0]
+    }
+    return req.ip ?? "unknown"
+  }
 }

api/src/users/users.service.ts

@@ -1,14 +1,17 @@
 import {
   ConflictException,
   Injectable,
+  NotFoundException,
   UnauthorizedException,
 } from "@nestjs/common"
 import { JwtService } from "@nestjs/jwt"
 import * as bcrypt from "bcrypt"
+
+import { AuditAction } from "../audit/audit-action.enum"
+import { AuditService } from "../audit/audit.service"
 import { SafeUser, toSafeUser } from "../auth/auth.service"
 import { TokenDenylistService } from "../auth/token-denylist.service"
 import { User, UsersRepository } from "../auth/users.repository"
-import { AuditService } from "../audit/audit.service"
 import { ChangePasswordDto } from "./dto/change-password.dto"
 import { UpdateProfileDto } from "./dto/update-profile.dto"
 
@@ -119,6 +122,37 @@ export class UsersService {
     })
   }
 
+  /**
+   * Soft-delete the authenticated user's account (issue #344).
+   * Revokes their current token and logs the deletion to the audit trail.
+   */
+  async deleteAccount(
+    userId: number,
+    authorizationHeader: string,
+    ip: string,
+  ): Promise<void> {
+    const user = await this.usersRepository.findById(userId)
+    if (!user) {
+      throw new UnauthorizedException("user not found")
+    }
+
+    const deleted = await this.usersRepository.softDelete(userId)
+    if (!deleted) {
+      throw new NotFoundException("user not found")
+    }
+
+    // Revoke the current access token so it cannot be reused
+    const token = this.extractBearerToken(authorizationHeader)
+    await this.tokenDenylistService.revoke(token, 3600)
+
+    await this.auditService.log(
+      userId,
+      AuditAction.USER_DELETED,
+      { email: user.email, username: user.username },
+      ip,
+    )
+  }
+
   private extractBearerToken(header: string): string {
     const match = header.trim().match(/^Bearer\s+(.+)$/i)
     if (!match) {

database/audit_logs.sql

@@ -1,6 +1,6 @@
 CREATE TABLE IF NOT EXISTS audit_logs (
     id SERIAL PRIMARY KEY,
-    user_id INTEGER REFERENCES users(id),
+    user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
     action VARCHAR(100) NOT NULL,
     ip VARCHAR(45),
     created_at TIMESTAMPTZ DEFAULT NOW()

database/migrations/2026072901_add_users_soft_delete.down.sql

@@ -0,0 +1,22 @@
+-- Migration: 2026072901_add_users_soft_delete (DOWN)
+--
+-- Rollback: revert soft-delete column and restore the original FK.
+
+BEGIN;
+
+-- 1. Drop the partial index
+DROP INDEX IF EXISTS idx_users_active;
+
+-- 2. Remove the soft-delete column
+ALTER TABLE users
+    DROP COLUMN IF EXISTS deleted_at;
+
+-- 3. Restore the original FK constraint (no ON DELETE SET NULL)
+ALTER TABLE audit_logs
+    DROP CONSTRAINT IF EXISTS audit_logs_user_id_fkey;
+
+ALTER TABLE audit_logs
+    ADD CONSTRAINT audit_logs_user_id_fkey
+        FOREIGN KEY (user_id) REFERENCES users(id);
+
+COMMIT;

database/migrations/2026072901_add_users_soft_delete.up.sql

@@ -0,0 +1,30 @@
+-- Migration: 2026072901_add_users_soft_delete (UP)
+--
+-- Soft-delete for the users table (issue #344):
+--   1. Drop the existing audit_logs.user_id FK constraint and recreate it
+--      with ON DELETE SET NULL so deleting a user does not cascade-fail.
+--   2. Add a deleted_at TIMESTAMPTZ column to the users table so the
+--      application can soft-delete instead of issuing a hard DELETE.
+--   3. Add an index on deleted_at so WHERE deleted_at IS NULL queries
+--      are efficient (partial index, nulls are excluded from the index).
+
+BEGIN;
+
+-- 1. Fix the audit_logs FK constraint
+ALTER TABLE audit_logs
+    DROP CONSTRAINT IF EXISTS audit_logs_user_id_fkey;
+
+ALTER TABLE audit_logs
+    ADD CONSTRAINT audit_logs_user_id_fkey
+        FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL;
+
+-- 2. Add soft-delete column to users
+ALTER TABLE users
+    ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
+
+-- 3. Partial index for active-user queries
+CREATE INDEX IF NOT EXISTS idx_users_active
+    ON users(id)
+    WHERE deleted_at IS NULL;
+
+COMMIT;

database/schema.sql

@@ -6,9 +6,12 @@ CREATE TABLE IF NOT EXISTS users (
     username VARCHAR(255) UNIQUE NOT NULL,
     email VARCHAR(255) UNIQUE NOT NULL,
     password_hash VARCHAR(255) NOT NULL,
-    created_at TIMESTAMPTZ DEFAULT NOW()
+    created_at TIMESTAMPTZ DEFAULT NOW(),
+    deleted_at TIMESTAMPTZ
 );
 
+CREATE INDEX IF NOT EXISTS idx_users_active ON users(id) WHERE deleted_at IS NULL;
+
 -- Streams table
 CREATE TABLE IF NOT EXISTS streams (
     id SERIAL PRIMARY KEY,
@@ -184,10 +187,10 @@ CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_pending_next_attempt
 
 CREATE TABLE IF NOT EXISTS audit_logs (
     id SERIAL PRIMARY KEY,
-    user_id INTEGER REFERENCES users(id),
+    user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
     action VARCHAR(100) NOT NULL,
     ip VARCHAR(45),
-    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+    created_at TIMESTAMPTZ DEFAULT NOW()
 );
 
 CREATE INDEX IF NOT EXISTS idx_audit_logs_user_id ON audit_logs(user_id);

k8s/README.md

@@ -381,6 +381,97 @@ kubectl -n xstreamroll exec deploy/processing -- wget -q -O - http://localhost:3
 curl -vI https://xstreamroll.example.com/api/health
 ```
 
+## Monitoring (issue #349)
+
+Example Prometheus alerting rules and a Grafana dashboard are shipped in the
+`monitoring/` directory at the repo root. Operators can import these into
+their observability stack to get production-ready alerting and dashboards
+without starting from scratch.
+
+### Prometheus alerting rules
+
+`monitoring/prometheus-rules.yaml` ships Prometheus alerting rules covering:
+
+| Alert                  | Metric(s)                              | Threshold                            |
+| ---------------------- | -------------------------------------- | ------------------------------------ |
+| HighErrorRate          | `http_requests_total` (5xx)           | > 5% of all requests over 5m        |
+| HighAPILatency         | `http_request_duration_seconds` (p99)  | > 1s over 5m                        |
+| DatabasePoolExhaustion | `db_pool_waiting_requests`            | > 0 for 1m                          |
+| WorkerRestart          | `xstreamroll_uptime_seconds` (reset)  | any reset within 5m                 |
+| HighWorkerErrorRate    | `xstreamroll_errors_total`            | > 1 error/s over 5m                 |
+| QueueDepthHigh         | `xstreamroll_queue_depth`             | > 32 000 for 10m                    |
+| LockRenewalFailures    | `xstreamroll_lock_renewals_failed`    | > 0.1/s over 5m                     |
+| WebSocketConnectionDrop| `websocket_active_connections`        | < 50% of historical peak for 10m    |
+
+Every alert ships at `severity: warning` by default so operators can tune
+thresholds before upgrading to `critical`.
+
+#### Installing the alert rules
+
+```bash
+# Validate the rules with promtool (requires promtool in your PATH).
+promtool check rules monitoring/prometheus-rules.yaml
+
+# Apply to a Prometheus operator via a ConfigMap.
+kubectl -n monitoring create configmap xstreamroll-rules \
+  --from-file=monitoring/prometheus-rules.yaml \
+  --dry-run=client -o yaml | kubectl apply -f -
+
+# If using kube-prometheus-stack, create a PrometheusRule CR instead.
+kubectl apply -f monitoring/prometheus-rules.yaml
+```
+
+#### Scraping the processing worker from Prometheus
+
+The worker exposes metrics on port `:3002` at `/metrics`. The accompanying
+`k8s/70-network-policies.yaml` already allows any pod in the `xstreamroll`
+namespace to reach that port. If your Prometheus instance runs in a separate
+`monitoring` namespace, add a `namespaceSelector` to the NetworkPolicy:
+
+```yaml
+# In k8s/70-network-policies.yaml, add to the allow-processing-metrics rule:
+  ingress:
+    - from:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: monitoring
+```
+
+Alternatively, annotate the worker pod template so Prometheus discovers it:
+
+```yaml
+# In k8s/40-processing.yaml, under spec.template.metadata:
+  annotations:
+    prometheus.io/scrape: "true"
+    prometheus.io/port: "3002"
+    prometheus.io/path: "/metrics/prometheus"
+```
+
+### Grafana dashboard
+
+`monitoring/grafana-dashboard.json` is a ready-to-import dashboard with
+panels covering:
+
+- **API Overview** — request rate by HTTP method, latency histogram (p50/p95/p99)
+- **WebSocket Connections** — active connections, connection rate
+- **Processing Worker** — queue depth, active session locks, messages
+  processed rate, throughput vs errors, uptime
+- **Database Connection Pool** — active/idle/waiting connections
+
+#### Importing the dashboard
+
+1. Open Grafana → Dashboards → Import.
+2. Paste the contents of `monitoring/grafana-dashboard.json` or upload
+   the file directly.
+3. Select your Prometheus data source when prompted.
+4. Save — the dashboard auto-refreshes every 30s.
+
+#### Customising for your deployment
+
+The dashboard uses the `DS_PROMETHEUS` template variable — you can swap
+this in Grafana without editing the JSON. All panels query `sum()`
+aggregations so they work correctly whether you have one worker pod or ten.
+
 ## Release process
 
 Images are published by `.github/workflows/release.yml` and are restricted