Build a Serverless Function Deployer

import { pool } from "../db";
import { Redis } from "ioredis";
import { randomBytes, createHash } from "node:crypto";
import { Worker } from "node:worker_threads";
import { writeFile, mkdir } from "node:fs/promises";
import { join } from "node:path";
const redis = new Redis(process.env.REDIS_URL!);

interface ServerlessFunction { id: string; name: string; code: string; runtime: "node" | "deno" | "bun"; version: number; envVars: Record<string, string>; customDomain: string | null; url: string; status: "deploying" | "active" | "error"; memory: number; timeout: number; createdAt: string; deployedAt: string; }
interface DeployResult { functionId: string; url: string; version: number; deployTime: number; }
interface FunctionLog { functionId: string; level: "info" | "error" | "warn"; message: string; timestamp: string; requestId: string; duration?: number; }

const FUNCTIONS_DIR = "/tmp/functions";
const workerPool = new Map<string, Worker>();

// Deploy function
export async function deploy(params: { name: string; code: string; runtime?: string; envVars?: Record<string, string>; memory?: number; timeout?: number }): Promise<DeployResult> {
  const start = Date.now();
  const id = createHash("md5").update(params.name).digest("hex").slice(0, 12);

  // Get current version
  const { rows: [existing] } = await pool.query("SELECT version FROM serverless_functions WHERE name = $1 ORDER BY version DESC LIMIT 1", [params.name]);
  const version = (existing?.version || 0) + 1;

  // Validate code
  try { new Function(params.code); } catch (e: any) { throw new Error(`Syntax error: ${e.message}`); }

  // Write function to disk
  const funcDir = join(FUNCTIONS_DIR, id);
  await mkdir(funcDir, { recursive: true });
  await writeFile(join(funcDir, "index.js"), wrapFunction(params.code, params.envVars || {}));

  // Create/update worker
  const worker = new Worker(join(funcDir, "index.js"), {
    eval: false,
    resourceLimits: { maxOldGenerationSizeMb: params.memory || 128 },
  });

  // Replace old worker
  const oldWorker = workerPool.get(id);
  if (oldWorker) { oldWorker.terminate(); }
  workerPool.set(id, worker);

  const url = `${process.env.FUNCTIONS_URL || 'https://fn.example.com'}/${params.name}`;

  const fn: ServerlessFunction = {
    id, name: params.name, code: params.code,
    runtime: (params.runtime || "node") as any, version,
    envVars: params.envVars || {}, customDomain: null, url,
    status: "active", memory: params.memory || 128,
    timeout: params.timeout || 30,
    createdAt: new Date().toISOString(), deployedAt: new Date().toISOString(),
  };

  await pool.query(
    `INSERT INTO serverless_functions (id, name, code, version, env_vars, url, status, memory_mb, timeout_s, deployed_at)
     VALUES ($1, $2, $3, $4, $5, $6, 'active', $7, $8, NOW())
     ON CONFLICT (name) DO UPDATE SET code = $3, version = $4, env_vars = $5, status = 'active', deployed_at = NOW()`,
    [id, params.name, params.code, version, JSON.stringify(fn.envVars), url, fn.memory, fn.timeout]
  );

  // Store routing info
  await redis.set(`fn:route:${params.name}`, JSON.stringify({ id, version }));

  return { functionId: id, url, version, deployTime: Date.now() - start };
}

// Invoke function
export async function invoke(functionName: string, request: { method: string; path: string; headers: Record<string, string>; body: any; query: Record<string, string> }): Promise<{ status: number; headers: Record<string, string>; body: any; duration: number; logs: string[] }> {
  const routeData = await redis.get(`fn:route:${functionName}`);
  if (!routeData) throw new Error(`Function '${functionName}' not found`);
  const { id } = JSON.parse(routeData);

  const worker = workerPool.get(id);
  if (!worker) throw new Error(`Function '${functionName}' not running`);

  const requestId = randomBytes(6).toString("hex");
  const start = Date.now();
  const logs: string[] = [];

  // Get timeout
  const { rows: [fn] } = await pool.query("SELECT timeout_s FROM serverless_functions WHERE id = $1", [id]);
  const timeout = (fn?.timeout_s || 30) * 1000;

  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => {
      reject(new Error(`Function timed out after ${timeout / 1000}s`));
    }, timeout);

    worker.once("message", (result) => {
      clearTimeout(timer);
      const duration = Date.now() - start;

      // Log invocation
      pool.query(
        "INSERT INTO function_logs (function_id, request_id, duration_ms, status_code, logged_at) VALUES ($1, $2, $3, $4, NOW())",
        [id, requestId, duration, result.status || 200]
      ).catch(() => {});

      // Track metrics
      redis.hincrby(`fn:metrics:${id}`, "invocations", 1).catch(() => {});
      redis.hincrby(`fn:metrics:${id}`, "totalDuration", duration).catch(() => {});

      resolve({ status: result.status || 200, headers: result.headers || {}, body: result.body, duration, logs });
    });

    worker.postMessage({ type: "invoke", request, requestId });
  });
}

function wrapFunction(code: string, envVars: Record<string, string>): string {
  return `
const { parentPort } = require('worker_threads');

// Set environment variables
${Object.entries(envVars).map(([k, v]) => `process.env['${k}'] = '${v}';`).join('\n')}

// User function
const handler = (function() {
  ${code}
  return typeof module !== 'undefined' && module.exports ? module.exports : (typeof handler === 'function' ? handler : typeof fetch === 'function' ? fetch : null);
})();

parentPort.on('message', async (msg) => {
  if (msg.type === 'invoke') {
    try {
      const result = await (handler.default || handler)(msg.request);
      parentPort.postMessage(result || { status: 200, body: 'OK' });
    } catch (error) {
      parentPort.postMessage({ status: 500, body: { error: error.message } });
    }
  }
});
`;
}

// Rollback to previous version
export async function rollback(functionName: string): Promise<DeployResult> {
  const { rows: [prev] } = await pool.query(
    "SELECT code, env_vars, memory_mb, timeout_s FROM serverless_functions WHERE name = $1 ORDER BY version DESC LIMIT 1 OFFSET 1",
    [functionName]
  );
  if (!prev) throw new Error("No previous version to rollback to");
  return deploy({ name: functionName, code: prev.code, envVars: JSON.parse(prev.env_vars), memory: prev.memory_mb, timeout: prev.timeout_s });
}

// Get function metrics
export async function getMetrics(functionId: string): Promise<{ invocations: number; avgDuration: number; errors: number }> {
  const stats = await redis.hgetall(`fn:metrics:${functionId}`);
  const invocations = parseInt(stats.invocations || "0");
  return { invocations, avgDuration: invocations > 0 ? parseInt(stats.totalDuration || "0") / invocations : 0, errors: parseInt(stats.errors || "0") };
}

// List all functions
export async function listFunctions(): Promise<Array<{ name: string; url: string; version: number; status: string; lastDeployed: string }>> {
  const { rows } = await pool.query("SELECT DISTINCT ON (name) name, url, version, status, deployed_at FROM serverless_functions ORDER BY name, version DESC");
  return rows.map((r: any) => ({ name: r.name, url: r.url, version: r.version, status: r.status, lastDeployed: r.deployed_at }));
}