From 7367f7cdb773d519be395e16c95dc38d0cde9d0e Mon Sep 17 00:00:00 2001 From: leubeem Date: Wed, 10 Jun 2026 23:17:03 +0200 Subject: [PATCH] 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 --- server/api/atc/ptt.post.ts | 50 ++++++- server/api/atc/say.post.ts | 23 +++- server/api/service/tools/latency.get.ts | 18 ++- server/middleware/auth.global.ts | 3 - server/models/UsageEvent.ts | 45 +++++++ server/utils/rateLimit.ts | 65 ++++++++++ server/utils/usage.ts | 165 ++++++++++++++++++++++++ 7 files changed, 355 insertions(+), 14 deletions(-) create mode 100644 server/models/UsageEvent.ts create mode 100644 server/utils/rateLimit.ts create mode 100644 server/utils/usage.ts diff --git a/server/api/atc/ptt.post.ts b/server/api/atc/ptt.post.ts index 6760663..0cd49cb 100644 --- a/server/api/atc/ptt.post.ts +++ b/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(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", diff --git a/server/api/atc/say.post.ts b/server/api/atc/say.post.ts index 003c105..772e466 100644 --- a/server/api/atc/say.post.ts +++ b/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", diff --git a/server/api/service/tools/latency.get.ts b/server/api/service/tools/latency.get.ts index f779f47..02356bc 100644 --- a/server/api/service/tools/latency.get.ts +++ b/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, diff --git a/server/middleware/auth.global.ts b/server/middleware/auth.global.ts index 9f030c6..67a1386 100644 --- a/server/middleware/auth.global.ts +++ b/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 } diff --git a/server/models/UsageEvent.ts b/server/models/UsageEvent.ts new file mode 100644 index 0000000..ebbc448 --- /dev/null +++ b/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({ + 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 | undefined) || + mongoose.model('UsageEvent', usageEventSchema) diff --git a/server/utils/rateLimit.ts b/server/utils/rateLimit.ts new file mode 100644 index 0000000..5eb57a4 --- /dev/null +++ b/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() + +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.', + }) + } +} diff --git a/server/utils/usage.ts b/server/utils/usage.ts new file mode 100644 index 0000000..00a0c81 --- /dev/null +++ b/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 = { + '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 = { + 'whisper-1': 0.006, + 'gpt-4o-transcribe': 0.006, + 'gpt-4o-mini-transcribe': 0.003, +} + +const TTS_PRICES_PER_1M_CHARS: Record = { + '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 { + 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 + topUsers: Array<{ email: string; events: number; costUsd: number }> +} + +export async function summarizeUsage(periodStart: Date, periodEnd: Date): Promise { + 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 { + 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 +}