server-proxy.ts

#!/usr/bin/env bun
/**
 * PAI Voice Proxy Server
 *
 * Listens on port 8888 (where all PAI callers point) and forwards
 * notifications to a configurable TTS service.
 *
 * Zero upstream PAI files modified.
 */

import { readFileSync, writeFileSync, existsSync, mkdirSync } from "fs";
import { join, dirname } from "path";

// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------

interface PAINotifyPayload {
  message: string;
  title?: string;
  voice_enabled?: boolean;
  voice_id?: string;
  voice_settings?: {
    stability?: number;
    similarity_boost?: number;
    style?: number;
    speed?: number;
    use_speaker_boost?: boolean;
  };
  volume?: number;
  language?: string;
}

interface FieldMapping {
  from: string; // dot-notation source path
  to: string; // dot-notation target path
  transform?: "passthrough" | `static:${string}` | `template:${string}`;
}

interface ProxyConfig {
  version: number;
  target_url: string;
  target_method: string;
  target_headers: Record<string, string>;
  default_language: string;
  default_source: string;
  mode: "default" | "mapped" | "passthrough";
  field_mappings?: FieldMapping[];
  timeout_ms: number;
  local_logging: boolean;
  port: number;
}

interface LogEntry {
  ts: string;
  endpoint: string;
  incoming: Record<string, unknown>;
  outgoing: Record<string, unknown> | null;
  target_url: string;
  target_status: number | null;
  target_body: string | null;
  duration_ms: number;
  error: string | null;
  skipped: boolean;
}

interface Stats {
  started_at: string;
  requests: number;
  forwards: number;
  skipped: number;
  errors: number;
  last_forward: {
    ts: string;
    status: number | null;
    error: string | null;
  } | null;
}

// ---------------------------------------------------------------------------
// Globals
// ---------------------------------------------------------------------------

const CONFIG_PATH = join(dirname(new URL(import.meta.url).pathname), "proxy-config.json");
const LOG_DIR = join(dirname(new URL(import.meta.url).pathname), "logs");

let config: ProxyConfig = loadConfig();
const stats: Stats = {
  started_at: new Date().toISOString(),
  requests: 0,
  forwards: 0,
  skipped: 0,
  errors: 0,
  last_forward: null,
};

// Rate limiting: 10 requests per 60 seconds per IP
const rateLimitWindow = 60_000;
const rateLimitMax = 10;
const rateLimitMap = new Map<string, number[]>();

// ---------------------------------------------------------------------------
// Config
// ---------------------------------------------------------------------------

function loadConfig(): ProxyConfig {
  try {
    const raw = readFileSync(CONFIG_PATH, "utf-8");
    return JSON.parse(raw) as ProxyConfig;
  } catch (e) {
    console.error(`[VoiceProxy] Failed to load config from ${CONFIG_PATH}:`, e);
    process.exit(1);
  }
}

function redactConfig(cfg: ProxyConfig): Record<string, unknown> {
  const redacted = { ...cfg } as Record<string, unknown>;
  if (typeof cfg.target_url === "string" && cfg.target_url.length > 30) {
    redacted.target_url = cfg.target_url.slice(0, 30) + "...REDACTED";
  }
  const headers = { ...cfg.target_headers };
  for (const key of Object.keys(headers)) {
    if (/auth|token|key|secret/i.test(key)) {
      headers[key] = "***REDACTED***";
    }
  }
  redacted.target_headers = headers;
  return redacted;
}

// ---------------------------------------------------------------------------
// Logging
// ---------------------------------------------------------------------------

function ensureLogDir() {
  if (!existsSync(LOG_DIR)) {
    mkdirSync(LOG_DIR, { recursive: true });
  }
}

function logForward(entry: LogEntry) {
  if (!config.local_logging) return;
  ensureLogDir();
  const logFile = join(LOG_DIR, "forwards.jsonl");
  try {
    const line = JSON.stringify(entry) + "\n";
    const { appendFileSync } = require("fs");
    appendFileSync(logFile, line);
  } catch {
    // Logging should never break the proxy
  }
}

// ---------------------------------------------------------------------------
// Rate Limiting
// ---------------------------------------------------------------------------

function isRateLimited(ip: string): boolean {
  const now = Date.now();
  const timestamps = rateLimitMap.get(ip) ?? [];
  const recent = timestamps.filter((t) => now - t < rateLimitWindow);
  rateLimitMap.set(ip, recent);

  if (recent.length >= rateLimitMax) {
    return true;
  }
  recent.push(now);
  return false;
}

// ---------------------------------------------------------------------------
// Payload Translation
// ---------------------------------------------------------------------------

function getNestedValue(obj: Record<string, unknown>, path: string): unknown {
  return path.split(".").reduce((acc: unknown, key) => {
    if (acc && typeof acc === "object") return (acc as Record<string, unknown>)[key];
    return undefined;
  }, obj);
}

function setNestedValue(obj: Record<string, unknown>, path: string, value: unknown) {
  const keys = path.split(".");
  let current = obj;
  for (let i = 0; i < keys.length - 1; i++) {
    if (!current[keys[i]] || typeof current[keys[i]] !== "object") {
      current[keys[i]] = {};
    }
    current = current[keys[i]] as Record<string, unknown>;
  }
  current[keys[keys.length - 1]] = value;
}

function translatePayload(
  payload: PAINotifyPayload,
  languageOverride?: string
): Record<string, unknown> | null {
  const language = languageOverride ?? payload.language ?? config.default_language;

  switch (config.mode) {
    case "default": {
      return {
        message: payload.message,
        language,
        source: config.default_source,
      };
    }

    case "mapped": {
      if (!config.field_mappings?.length) {
        console.warn("[VoiceProxy] mode=mapped but no field_mappings defined, using passthrough");
        return payload as unknown as Record<string, unknown>;
      }
      const result: Record<string, unknown> = {};
      for (const mapping of config.field_mappings) {
        let value: unknown;
        if (mapping.transform) {
          if (mapping.transform === "passthrough") {
            value = getNestedValue(payload as unknown as Record<string, unknown>, mapping.from);
          } else if (mapping.transform.startsWith("static:")) {
            value = mapping.transform.slice(7);
          } else if (mapping.transform.startsWith("template:")) {
            const tmpl = mapping.transform.slice(9);
            value = tmpl.replace(/\{(\w+(?:\.\w+)*)\}/g, (_match, field) => {
              const v = getNestedValue(payload as unknown as Record<string, unknown>, field);
              return v !== undefined ? String(v) : "";
            });
          }
        } else {
          value = getNestedValue(payload as unknown as Record<string, unknown>, mapping.from);
        }
        if (value !== undefined) {
          setNestedValue(result, mapping.to, value);
        }
      }
      return result;
    }

    case "passthrough": {
      return payload as unknown as Record<string, unknown>;
    }

    default:
      console.warn(`[VoiceProxy] Unknown mode: ${config.mode}, using passthrough`);
      return payload as unknown as Record<string, unknown>;
  }
}

// ---------------------------------------------------------------------------
// Forwarding
// ---------------------------------------------------------------------------

async function forwardToTarget(
  translated: Record<string, unknown>
): Promise<{ status: number; body: string }> {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), config.timeout_ms);

  try {
    const resp = await fetch(config.target_url, {
      method: config.target_method,
      headers: {
        "Content-Type": "application/json",
        ...config.target_headers,
      },
      body: JSON.stringify(translated),
      signal: controller.signal,
    });
    const body = await resp.text();
    return { status: resp.status, body };
  } finally {
    clearTimeout(timeout);
  }
}

// ---------------------------------------------------------------------------
// CORS Headers
// ---------------------------------------------------------------------------

function corsHeaders(): Record<string, string> {
  return {
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type",
  };
}

// ---------------------------------------------------------------------------
// Request Handler
// ---------------------------------------------------------------------------

async function handleNotify(
  req: Request,
  endpoint: string
): Promise<Response> {
  const start = Date.now();
  let incoming: Record<string, unknown> = {};

  try {
    incoming = (await req.json()) as Record<string, unknown>;
  } catch {
    return new Response(JSON.stringify({ status: "error", error: "Invalid JSON" }), {
      status: 400,
      headers: { "Content-Type": "application/json", ...corsHeaders() },
    });
  }

  const payload = incoming as unknown as PAINotifyPayload;
  stats.requests++;

  // Check voice_enabled flag
  if (payload.voice_enabled === false) {
    stats.skipped++;
    logForward({
      ts: new Date().toISOString(),
      endpoint,
      incoming,
      outgoing: null,
      target_url: config.target_url,
      target_status: null,
      target_body: null,
      duration_ms: Date.now() - start,
      error: null,
      skipped: true,
    });
    return new Response(
      JSON.stringify({ status: "ok", action: "skipped", reason: "voice_enabled=false" }),
      { status: 200, headers: { "Content-Type": "application/json", ...corsHeaders() } }
    );
  }

  // Language override from query param
  const url = new URL(req.url);
  const langParam = url.searchParams.get("language") ?? undefined;

  const translated = translatePayload(payload, langParam);
  if (!translated) {
    stats.errors++;
    return new Response(
      JSON.stringify({ status: "ok", action: "skipped", reason: "translation returned null" }),
      { status: 200, headers: { "Content-Type": "application/json", ...corsHeaders() } }
    );
  }

  // Forward asynchronously — always return 200 to PAI callers
  let targetStatus: number | null = null;
  let targetBody: string | null = null;
  let forwardError: string | null = null;

  try {
    const result = await forwardToTarget(translated);
    targetStatus = result.status;
    targetBody = result.body;
    stats.forwards++;
    stats.last_forward = {
      ts: new Date().toISOString(),
      status: targetStatus,
      error: null,
    };
  } catch (e) {
    forwardError = e instanceof Error ? e.message : String(e);
    stats.errors++;
    stats.last_forward = {
      ts: new Date().toISOString(),
      status: null,
      error: forwardError,
    };
    console.error(`[VoiceProxy] Forward failed:`, forwardError);
  }

  logForward({
    ts: new Date().toISOString(),
    endpoint,
    incoming,
    outgoing: translated,
    target_url: config.target_url,
    target_status: targetStatus,
    target_body: targetBody,
    duration_ms: Date.now() - start,
    error: forwardError,
    skipped: false,
  });

  // Always 200 to PAI
  return new Response(
    JSON.stringify({
      status: "ok",
      action: "forwarded",
      target_status: targetStatus,
      error: forwardError,
    }),
    { status: 200, headers: { "Content-Type": "application/json", ...corsHeaders() } }
  );
}

// ---------------------------------------------------------------------------
// Server
// ---------------------------------------------------------------------------

const server = Bun.serve({
  port: config.port,
  async fetch(req) {
    const url = new URL(req.url);
    const method = req.method;

    // CORS preflight
    if (method === "OPTIONS") {
      return new Response(null, { status: 204, headers: corsHeaders() });
    }

    // Rate limiting
    const ip =
      req.headers.get("x-forwarded-for")?.split(",")[0]?.trim() ??
      server.requestIP(req)?.address ??
      "unknown";
    if (method === "POST" && isRateLimited(ip)) {
      return new Response(
        JSON.stringify({ status: "error", error: "Rate limited" }),
        { status: 429, headers: { "Content-Type": "application/json", ...corsHeaders() } }
      );
    }

    // Routes
    const path = url.pathname;

    // POST /notify, /pai, /notify/personality — all forward to TTS
    if (
      method === "POST" &&
      (path === "/notify" || path === "/pai" || path === "/notify/personality")
    ) {
      return handleNotify(req, path);
    }

    // GET /health
    if (method === "GET" && path === "/health") {
      return new Response(
        JSON.stringify({
          voice_system: "proxy",
          status: "running",
          version: "1.0.0",
          mode: config.mode,
          target_url_prefix: config.target_url.slice(0, 30) + "...",
          stats,
          uptime_s: Math.floor((Date.now() - new Date(stats.started_at).getTime()) / 1000),
        }),
        { status: 200, headers: { "Content-Type": "application/json", ...corsHeaders() } }
      );
    }

    // GET /config
    if (method === "GET" && path === "/config") {
      return new Response(JSON.stringify(redactConfig(config), null, 2), {
        status: 200,
        headers: { "Content-Type": "application/json", ...corsHeaders() },
      });
    }

    // POST /config/reload
    if (method === "POST" && path === "/config/reload") {
      try {
        config = loadConfig();
        console.log("[VoiceProxy] Config reloaded successfully");
        return new Response(
          JSON.stringify({ status: "ok", config: redactConfig(config) }),
          { status: 200, headers: { "Content-Type": "application/json", ...corsHeaders() } }
        );
      } catch (e) {
        return new Response(
          JSON.stringify({
            status: "error",
            error: e instanceof Error ? e.message : String(e),
          }),
          { status: 500, headers: { "Content-Type": "application/json", ...corsHeaders() } }
        );
      }
    }

    // 404
    return new Response(
      JSON.stringify({ status: "error", error: "Not found", endpoints: ["/notify", "/pai", "/notify/personality", "/health", "/config", "/config/reload"] }),
      { status: 404, headers: { "Content-Type": "application/json", ...corsHeaders() } }
    );
  },
});

console.log(`[VoiceProxy] Running on http://localhost:${config.port}`);
console.log(`[VoiceProxy] Mode: ${config.mode}`);
console.log(`[VoiceProxy] Target: ${config.target_url.slice(0, 40)}...`);
console.log(`[VoiceProxy] Logging: ${config.local_logging ? "enabled" : "disabled"}`);