feat(server): per-user AI usage tracking, cost alerting, and endpoint hardening

Usage tracking:
- new UsageEvent collection records every STT/TTS/LLM call per user with
  provider, model, volume (audio seconds, characters, tokens) and an
  estimated USD cost; self-hosted providers (Speaches/Piper) and cache
  hits record at $0
- pricing table for whisper-1, tts-1, gpt-5-nano & co. in server/utils/usage.ts
- weekly KPI mail gains an "AI-Nutzung & Kosten" section: weekly and
  rolling 30-day cost, per-kind breakdown, top 5 users by cost
- quota alert mail when rolling 30-day cost exceeds USAGE_ALERT_USD
  (default $5), at most once per calendar month (UsageAlertDelivery)

Hardening:
- /api/atc/say now requires an authenticated session (middleware
  exemption removed); useFlightLabAudio sends the bearer token
- /api/service/tools/latency requires auth (was a public LLM endpoint)
- per-user rate limits: PTT 20/min, say 60/min, latency 5/min
- cron endpoints (waitlist-drip, weekly-kpi-report) require a shared
  secret via ?secret= or x-cron-secret (CRON_SECRET, falls back to
  KPI_CRON_SECRET); allowed with a warning while unset so existing
  deployments keep working
- PTT records the actual transcribed audio duration for billing accuracy

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

server/api/atc/ptt.post.ts

@@ -1,6 +1,6 @@
 // server/api/atc/ptt.post.ts
 import { createError, readBody } from "h3";
-import { writeFile, rm } from "node:fs/promises";
+import { writeFile, rm, readFile } from "node:fs/promises";
 import { join } from "node:path";
 import { tmpdir } from "node:os";
 import { randomUUID } from "node:crypto";
@@ -9,6 +9,8 @@ import { getOpenAIClient } from "../../utils/openai";
 import { createReadStream } from "node:fs";
 import { TransmissionLog } from "../../models/TransmissionLog";
 import { getUserFromEvent } from "../../utils/auth";
+import { enforceRateLimit, getClientIp } from "../../utils/rateLimit";
+import { recordUsage } from "../../utils/usage";
 
 type AudioFormat = 'wav' | 'mp3' | 'ogg' | 'webm'
 
@@ -66,6 +68,13 @@ function decodeAudioPayload(encoded: string): Buffer {
     return buffer;
 }
 
+function wavDurationSeconds(buffer: Buffer): number | undefined {
+    if (buffer.length < 44 || buffer.toString('ascii', 0, 4) !== 'RIFF') return undefined;
+    const byteRate = buffer.readUInt32LE(28);
+    if (!byteRate) return undefined;
+    return Math.round(((buffer.length - 44) / byteRate) * 100) / 100;
+}
+
 async function convertToWav(inputPath: string, outputPath: string) {
     await sh("ffmpeg", [
         "-y", "-i", inputPath,
@@ -76,7 +85,12 @@ async function convertToWav(inputPath: string, outputPath: string) {
     ]);
 }
 
+const PTT_RATE_LIMIT_PER_MINUTE = 20;
+
 export default defineEventHandler(async (event) => {
+    const user = await getUserFromEvent(event);
+    enforceRateLimit(event, 'atc-ptt', user ? String(user._id) : getClientIp(event), PTT_RATE_LIMIT_PER_MINUTE);
+
     const body = await readBody<PTTRequest>(event);
 
     if (!body.audio || !body.moduleId || !body.lessonId) {
@@ -115,6 +129,34 @@ export default defineEventHandler(async (event) => {
 
         const transcribedText = transcription.text.trim();
 
+        // Audio length for usage accounting. The Whisper input is 16kHz mono WAV
+        // in the normal path; fall back to a byte-rate estimate if header parsing fails.
+        let audioSeconds: number | undefined;
+        try {
+            const wavBuffer = audioFileForWhisper === tmpAudioInput && format === 'wav'
+                ? audioBuffer
+                : await readFile(audioFileForWhisper);
+            audioSeconds = wavDurationSeconds(wavBuffer) ?? Math.round((wavBuffer.length / 32000) * 100) / 100;
+        } catch {
+            audioSeconds = Math.round((audioBuffer.length / 32000) * 100) / 100;
+        }
+
+        // Prefer the explicit top-level sessionId (Python backend session).
+        // Fall back to the legacy context.flags.session_id for older clients.
+        const sessionId = body.sessionId
+            ?? (typeof body.context?.flags?.session_id === 'string' ? body.context.flags.session_id : undefined);
+
+        // Whisper bills the audio even when nothing was recognized.
+        await recordUsage({
+            user: user?._id ? String(user._id) : undefined,
+            sessionId,
+            kind: 'stt',
+            provider: 'openai',
+            model: 'whisper-1',
+            endpoint: '/api/atc/ptt',
+            audioSeconds,
+        });
+
         if (!transcribedText) {
             throw createError({ statusCode: 400, statusMessage: "No speech detected in audio" });
         }
@@ -125,12 +167,6 @@ export default defineEventHandler(async (event) => {
         }
 
         try {
-            const user = await getUserFromEvent(event);
-            // Prefer the explicit top-level sessionId (Python backend session).
-            // Fall back to the legacy context.flags.session_id for older clients.
-            const sessionId = body.sessionId
-                ?? (typeof body.context?.flags?.session_id === 'string' ? body.context.flags.session_id : undefined);
-
             await TransmissionLog.create({
                 user: user?._id,
                 role: "pilot",

server/api/atc/say.post.ts

@@ -7,6 +7,9 @@ import {normalize, TTS_MODEL, normalizeATC} from "../../utils/normalize";
 import { getServerRuntimeConfig } from "../../utils/runtimeConfig";
 import {request} from "node:http";
 import { TransmissionLog } from "../../models/TransmissionLog";
+import { requireUserSession } from "../../utils/auth";
+import { enforceRateLimit } from "../../utils/rateLimit";
+import { recordUsage } from "../../utils/usage";
 
 
 function outDir() {
@@ -157,7 +160,12 @@ async function speachesTTS(
     return Buffer.from(arr);
 }
 
+const SAY_RATE_LIMIT_PER_MINUTE = 60;
+
 export default defineEventHandler(async (event) => {
+    const user = await requireUserSession(event);
+    enforceRateLimit(event, 'atc-say', String(user._id), SAY_RATE_LIMIT_PER_MINUTE);
+
     const runtimeConfig = getServerRuntimeConfig();
     const body = await readBody<{
         text?: string;
@@ -177,8 +185,6 @@ export default defineEventHandler(async (event) => {
         preNormalized?: boolean;
     }>(event);
 
-    // const user = await requireUserSession(event);
-
     const rawSessionId = typeof body?.sessionId === "string"
         ? body.sessionId.trim()
         : "";
@@ -348,9 +354,20 @@ export default defineEventHandler(async (event) => {
             }
         };
 
+        await recordUsage({
+            user: String(user._id),
+            sessionId,
+            kind: 'tts',
+            // Cache hits cost nothing regardless of which provider filled the cache.
+            provider: cacheHit ? 'cache' : ttsProvider,
+            model: modelUsed,
+            endpoint: '/api/atc/say',
+            characters: normalized.length,
+        });
+
         try {
             await TransmissionLog.create({
-                // user: user._id,
+                user: user._id,
                 role: "atc",
                 channel: "say",
                 direction: "outgoing",

server/api/service/tools/latency.get.ts

@@ -2,6 +2,9 @@
 import { createError } from 'h3'
 import { getOpenAIClient } from '../../../utils/openai'
 import { getServerRuntimeConfig } from '../../../utils/runtimeConfig'
+import { requireUserSession } from '../../../utils/auth'
+import { enforceRateLimit } from '../../../utils/rateLimit'
+import { recordUsage } from '../../../utils/usage'
 
 const SYSTEM_PROMPT =
     'Check if the pilot readback contains ALL of: Frankfurt or EDDF, FL320, and 120.8 MHz. ' +
@@ -10,7 +13,10 @@ const SYSTEM_PROMPT =
 const READBACK =
     'Lufthanser four seven eight cleared fra via NORDA1A, climb 5000 feet, expect flight level tree too zero, dep 120 decimal 8, squawk 4213.';
 
-export default defineEventHandler(async () => {
+export default defineEventHandler(async (event) => {
+    const user = await requireUserSession(event)
+    enforceRateLimit(event, 'tools-latency', String(user._id), 5)
+
     const client = getOpenAIClient()
     const { llmModel } = getServerRuntimeConfig()
     const model = llmModel || 'chatgpt-5-nano'
@@ -32,6 +38,16 @@ export default defineEventHandler(async () => {
         const validResult = Number.isInteger(parsed) && parsed >= 0 && parsed <= 2 ? parsed : null
         const latencyMs = Date.now() - started
 
+        await recordUsage({
+            user: String(user._id),
+            kind: 'llm',
+            provider: 'openai',
+            model,
+            endpoint: '/api/service/tools/latency',
+            inputTokens: response.usage?.prompt_tokens,
+            outputTokens: response.usage?.completion_tokens,
+        })
+
         return {
             result: validResult,
             raw,

server/middleware/auth.global.ts

@@ -6,9 +6,6 @@ export default defineEventHandler(async (event) => {
     if (!url.pathname.startsWith('/api/')) {
         return
     }
-    if (url.pathname.startsWith('/api/atc/say')) {
-        return
-    }
     if (url.pathname.startsWith('/api/service/')) {
         return
     }

server/models/UsageEvent.ts

@@ -0,0 +1,45 @@
+import mongoose from 'mongoose'
+
+const { Schema } = mongoose
+
+export type UsageKind = 'stt' | 'tts' | 'llm'
+export type UsageProvider = 'openai' | 'speaches' | 'piper' | 'cache'
+
+export interface UsageEventDocument extends mongoose.Document {
+  user?: mongoose.Types.ObjectId
+  sessionId?: string
+  kind: UsageKind
+  provider: UsageProvider
+  model: string
+  endpoint: string
+  /** Audio length for STT in seconds */
+  audioSeconds?: number
+  /** Synthesized text length for TTS */
+  characters?: number
+  inputTokens?: number
+  outputTokens?: number
+  /** Estimated cost in USD (0 for self-hosted providers and cache hits) */
+  costUsd: number
+  createdAt: Date
+}
+
+const usageEventSchema = new mongoose.Schema<UsageEventDocument>({
+  user: { type: Schema.Types.ObjectId, ref: 'User', index: true },
+  sessionId: { type: String },
+  kind: { type: String, enum: ['stt', 'tts', 'llm'], required: true },
+  provider: { type: String, enum: ['openai', 'speaches', 'piper', 'cache'], required: true },
+  model: { type: String, required: true },
+  endpoint: { type: String, required: true },
+  audioSeconds: { type: Number },
+  characters: { type: Number },
+  inputTokens: { type: Number },
+  outputTokens: { type: Number },
+  costUsd: { type: Number, required: true, default: 0 },
+  createdAt: { type: Date, default: () => new Date(), index: true },
+})
+
+usageEventSchema.index({ user: 1, createdAt: -1 })
+
+export const UsageEvent =
+  (mongoose.models.UsageEvent as mongoose.Model<UsageEventDocument> | undefined) ||
+  mongoose.model<UsageEventDocument>('UsageEvent', usageEventSchema)

server/utils/rateLimit.ts

@@ -0,0 +1,65 @@
+import type { H3Event } from 'h3'
+import { createError } from 'h3'
+
+interface WindowEntry {
+  windowStart: number
+  count: number
+}
+
+// In-memory fixed-window limiter. Per process — good enough for a single
+// Coolify instance; swap for a shared store if the app ever scales out.
+const windows = new Map<string, WindowEntry>()
+
+const CLEANUP_INTERVAL_MS = 10 * 60 * 1000
+let lastCleanup = Date.now()
+
+function cleanup(now: number, windowMs: number) {
+  if (now - lastCleanup < CLEANUP_INTERVAL_MS) return
+  lastCleanup = now
+  for (const [key, entry] of windows) {
+    if (now - entry.windowStart > windowMs * 2) {
+      windows.delete(key)
+    }
+  }
+}
+
+export function getClientIp(event: H3Event): string {
+  const forwarded = event.node.req.headers['x-forwarded-for']
+  if (typeof forwarded === 'string' && forwarded.trim()) {
+    return forwarded.split(',')[0]!.trim()
+  }
+  return event.node.req.socket?.remoteAddress || 'unknown'
+}
+
+/**
+ * Throws 429 when `key` exceeds `limit` requests per `windowMs`.
+ * Use a user id as key where available, client IP otherwise.
+ */
+export function enforceRateLimit(
+  event: H3Event,
+  bucket: string,
+  key: string,
+  limit: number,
+  windowMs = 60_000,
+) {
+  const now = Date.now()
+  cleanup(now, windowMs)
+
+  const mapKey = `${bucket}:${key}`
+  const entry = windows.get(mapKey)
+
+  if (!entry || now - entry.windowStart >= windowMs) {
+    windows.set(mapKey, { windowStart: now, count: 1 })
+    return
+  }
+
+  entry.count += 1
+  if (entry.count > limit) {
+    const retryAfter = Math.ceil((entry.windowStart + windowMs - now) / 1000)
+    event.node.res.setHeader('Retry-After', String(Math.max(retryAfter, 1)))
+    throw createError({
+      statusCode: 429,
+      statusMessage: 'Too many requests — slow down and try again shortly.',
+    })
+  }
+}

server/utils/usage.ts

@@ -0,0 +1,165 @@
+import mongoose from 'mongoose'
+import { UsageEvent, type UsageKind, type UsageProvider } from '../models/UsageEvent'
+
+// OpenAI list prices (USD). Self-hosted providers (speaches/piper) cost 0.
+// Update here when models or prices change.
+const LLM_PRICES_PER_1M_TOKENS: Record<string, { input: number; output: number }> = {
+  'gpt-5-nano': { input: 0.05, output: 0.4 },
+  'gpt-5-mini': { input: 0.25, output: 2.0 },
+  'gpt-5': { input: 1.25, output: 10.0 },
+  'gpt-4o-mini': { input: 0.15, output: 0.6 },
+  'gpt-4o': { input: 2.5, output: 10.0 },
+}
+
+const STT_PRICES_PER_MINUTE: Record<string, number> = {
+  'whisper-1': 0.006,
+  'gpt-4o-transcribe': 0.006,
+  'gpt-4o-mini-transcribe': 0.003,
+}
+
+const TTS_PRICES_PER_1M_CHARS: Record<string, number> = {
+  'tts-1': 15,
+  'tts-1-hd': 30,
+  'gpt-4o-mini-tts': 12,
+}
+
+export interface UsageInput {
+  user?: mongoose.Types.ObjectId | string | null
+  sessionId?: string
+  kind: UsageKind
+  provider: UsageProvider
+  model: string
+  endpoint: string
+  audioSeconds?: number
+  characters?: number
+  inputTokens?: number
+  outputTokens?: number
+}
+
+export function estimateCostUsd(input: UsageInput): number {
+  if (input.provider !== 'openai') {
+    return 0
+  }
+
+  if (input.kind === 'llm') {
+    const price = LLM_PRICES_PER_1M_TOKENS[input.model]
+    if (!price) return 0
+    const inputCost = ((input.inputTokens || 0) / 1_000_000) * price.input
+    const outputCost = ((input.outputTokens || 0) / 1_000_000) * price.output
+    return inputCost + outputCost
+  }
+
+  if (input.kind === 'stt') {
+    const perMinute = STT_PRICES_PER_MINUTE[input.model]
+    if (!perMinute) return 0
+    return ((input.audioSeconds || 0) / 60) * perMinute
+  }
+
+  if (input.kind === 'tts') {
+    const per1M = TTS_PRICES_PER_1M_CHARS[input.model]
+    if (!per1M) return 0
+    return ((input.characters || 0) / 1_000_000) * per1M
+  }
+
+  return 0
+}
+
+/** Fire-and-forget usage recording — must never break the calling endpoint. */
+export async function recordUsage(input: UsageInput): Promise<void> {
+  try {
+    await UsageEvent.create({
+      user: input.user || undefined,
+      sessionId: input.sessionId,
+      kind: input.kind,
+      provider: input.provider,
+      model: input.model,
+      endpoint: input.endpoint,
+      audioSeconds: input.audioSeconds,
+      characters: input.characters,
+      inputTokens: input.inputTokens,
+      outputTokens: input.outputTokens,
+      costUsd: estimateCostUsd(input),
+    })
+  } catch (error) {
+    console.warn('[usage] Recording usage event failed', error)
+  }
+}
+
+export interface UsageSummary {
+  events: number
+  costUsd: number
+  sttSeconds: number
+  ttsCharacters: number
+  llmInputTokens: number
+  llmOutputTokens: number
+  byKind: Record<UsageKind, { events: number; costUsd: number }>
+  topUsers: Array<{ email: string; events: number; costUsd: number }>
+}
+
+export async function summarizeUsage(periodStart: Date, periodEnd: Date): Promise<UsageSummary> {
+  const match = { createdAt: { $gte: periodStart, $lt: periodEnd } }
+
+  const [totals, kinds, topUsers] = await Promise.all([
+    UsageEvent.aggregate([
+      { $match: match },
+      {
+        $group: {
+          _id: null,
+          events: { $sum: 1 },
+          costUsd: { $sum: '$costUsd' },
+          sttSeconds: { $sum: { $ifNull: ['$audioSeconds', 0] } },
+          ttsCharacters: { $sum: { $ifNull: ['$characters', 0] } },
+          llmInputTokens: { $sum: { $ifNull: ['$inputTokens', 0] } },
+          llmOutputTokens: { $sum: { $ifNull: ['$outputTokens', 0] } },
+        },
+      },
+    ]),
+    UsageEvent.aggregate([
+      { $match: match },
+      { $group: { _id: '$kind', events: { $sum: 1 }, costUsd: { $sum: '$costUsd' } } },
+    ]),
+    UsageEvent.aggregate([
+      { $match: { ...match, user: { $ne: null } } },
+      { $group: { _id: '$user', events: { $sum: 1 }, costUsd: { $sum: '$costUsd' } } },
+      { $sort: { costUsd: -1, events: -1 } },
+      { $limit: 5 },
+      { $lookup: { from: 'users', localField: '_id', foreignField: '_id', as: 'userDoc' } },
+    ]),
+  ])
+
+  const t = totals[0] || {}
+  const byKind: UsageSummary['byKind'] = {
+    stt: { events: 0, costUsd: 0 },
+    tts: { events: 0, costUsd: 0 },
+    llm: { events: 0, costUsd: 0 },
+  }
+  for (const row of kinds) {
+    if (row._id === 'stt' || row._id === 'tts' || row._id === 'llm') {
+      byKind[row._id as UsageKind] = { events: row.events || 0, costUsd: row.costUsd || 0 }
+    }
+  }
+
+  return {
+    events: t.events || 0,
+    costUsd: t.costUsd || 0,
+    sttSeconds: t.sttSeconds || 0,
+    ttsCharacters: t.ttsCharacters || 0,
+    llmInputTokens: t.llmInputTokens || 0,
+    llmOutputTokens: t.llmOutputTokens || 0,
+    byKind,
+    topUsers: topUsers.map((row: any) => ({
+      email: row.userDoc?.[0]?.email || String(row._id),
+      events: row.events || 0,
+      costUsd: row.costUsd || 0,
+    })),
+  }
+}
+
+export async function getRollingCostUsd(days: number, now = new Date()): Promise<number> {
+  const start = new Date(now.getTime() - days * 24 * 60 * 60 * 1000)
+  const rows = await UsageEvent.aggregate([
+    { $match: { createdAt: { $gte: start, $lt: now } } },
+    { $group: { _id: null, costUsd: { $sum: '$costUsd' } } },
+  ])
+  return rows[0]?.costUsd || 0
+}