pikasTech-HWLAB/cmd/hwlab-deepseek-responses-bridge/main.ts

#!/usr/bin/env node
import { createServer, request as httpRequest } from "node:http";
import { request as httpsRequest } from "node:https";

import { decompress } from "fzstd";

const serviceId = "hwlab-deepseek-responses-bridge";
const port = parsePositiveInteger(process.env.PORT, 4000);
const upstreamBaseUrl = new URL(process.env.HWLAB_DEEPSEEK_BRIDGE_UPSTREAM || "http://127.0.0.1:4001");
const defaultModel = String(process.env.HWLAB_DEEPSEEK_BRIDGE_MODEL || process.env.HWLAB_CODE_AGENT_DEEPSEEK_MODEL || "deepseek-chat").trim() || "deepseek-chat";
const maxBodyBytes = parsePositiveInteger(process.env.HWLAB_DEEPSEEK_BRIDGE_MAX_BODY_BYTES, 64 * 1024 * 1024);

const hopByHopHeaders = new Set([
  "connection",
  "keep-alive",
  "proxy-authenticate",
  "proxy-authorization",
  "te",
  "trailer",
  "transfer-encoding",
  "upgrade"
]);

const server = createServer(async (clientReq, clientRes) => {
  try {
    const url = new URL(clientReq.url || "/", "http://hwlab-deepseek-bridge.local");
    if (url.pathname === "/health" || url.pathname === "/health/live" || url.pathname === "/health/liveliness") {
      sendJson(clientRes, 200, healthPayload());
      return;
    }
    if (url.pathname === "/health/readiness") {
      await sendReadiness(clientRes);
      return;
    }
    if (clientReq.method === "GET" && (url.pathname === "/v1/models" || url.pathname === "/models")) {
      await sendCodexModels(clientReq, clientRes, url);
      return;
    }
    if (clientReq.method === "POST" && url.pathname === "/v1/responses") {
      await proxyResponsesRequest(clientReq, clientRes, url);
      return;
    }
    await proxyRequest(clientReq, clientRes, url);
  } catch (error) {
    sendJson(clientRes, 502, {
      error: {
        code: "deepseek_bridge_proxy_failed",
        message: "DeepSeek Responses bridge failed to proxy request",
        reason: error.message
      }
    });
  }
});

server.listen(port, "0.0.0.0", () => {
  process.stdout.write(JSON.stringify({ serviceId, status: "listening", port, upstream: upstreamBaseUrl.origin }) + "\n");
});

async function sendReadiness(res) {
  const upstream = await fetchUpstreamReadiness();
  sendJson(res, upstream.ok ? 200 : 503, {
    ...healthPayload(),
    status: upstream.ok ? "ready" : "degraded",
    upstream: {
      ok: upstream.ok,
      status: upstream.status,
      source: upstream.source,
      payload: upstream.payload
    }
  });
}

async function fetchUpstreamReadiness() {
  const health = await fetchUpstreamJson("/health/readiness")
    .then((result) => ({ ...result, source: "/health/readiness" }))
    .catch((error) => ({ ok: false, status: 502, source: "/health/readiness", payload: { error: error.message } }));
  if (health.ok) return health;

  const models = await fetchUpstreamJson("/v1/models")
    .then((result) => ({ ...result, source: "/v1/models" }))
    .catch((error) => ({ ok: false, status: 502, source: "/v1/models", payload: { error: error.message } }));
  const modelCount = codexModelsFromUpstream(models.payload).length;
  if (models.ok && modelCount > 0) {
    return {
      ...models,
      payload: {
        modelCount,
        fallbackFrom: {
          source: health.source,
          status: health.status
        }
      }
    };
  }

  return {
    ...health,
    payload: {
      ...health.payload,
      modelProbe: {
        ok: models.ok,
        status: models.status,
        modelCount
      }
    }
  };
}

async function sendCodexModels(clientReq, clientRes, url) {
  const upstream = await fetchUpstreamJson(`${url.pathname}${url.search}`, clientReq.headers).catch((error) => ({ ok: false, status: 502, payload: { error: error.message } }));
  const models = codexModelsFromUpstream(upstream.payload);
  sendJson(clientRes, 200, {
    models: models.length > 0 ? models : [codexModel(defaultModel)],
    upstream: {
      ok: upstream.ok,
      status: upstream.status,
      source: upstream.ok ? "upstream" : "fallback"
    }
  });
}

async function proxyRequest(clientReq, clientRes, url) {
  const startedAt = Date.now();
  const originalBody = await readRequestBody(clientReq);
  const { body, headers, decompressed, sanitizedToolTypes } = decodeRequestBody(originalBody, clientReq.headers, url.pathname);
  const targetUrl = new URL(`${url.pathname}${url.search}`, upstreamBaseUrl);
  const requestImpl = targetUrl.protocol === "https:" ? httpsRequest : httpRequest;

  const upstreamReq = requestImpl({
    protocol: targetUrl.protocol,
    hostname: targetUrl.hostname,
    port: targetUrl.port || (targetUrl.protocol === "https:" ? 443 : 80),
    method: clientReq.method,
    path: `${targetUrl.pathname}${targetUrl.search}`,
    headers
  }, (upstreamRes) => {
    const responseHeaders = filterResponseHeaders(upstreamRes.headers);
    clientRes.writeHead(upstreamRes.statusCode || 502, responseHeaders);
    upstreamRes.pipe(clientRes);
    upstreamRes.on("end", () => {
      process.stderr.write(JSON.stringify({
        serviceId,
        method: clientReq.method,
        path: url.pathname,
        status: upstreamRes.statusCode || null,
        decompressed,
        sanitizedToolTypes,
        elapsedMs: Date.now() - startedAt
      }) + "\n");
    });
  });

  upstreamReq.on("error", (error) => {
    if (!clientRes.headersSent) {
      sendJson(clientRes, 502, {
        error: {
          code: "deepseek_bridge_upstream_error",
          message: "DeepSeek bridge upstream request failed",
          reason: error.message
        }
      });
    } else {
      clientRes.end();
    }
  });

  upstreamReq.end(body);
}

async function proxyResponsesRequest(clientReq, clientRes, url) {
  const startedAt = Date.now();
  const originalBody = await readRequestBody(clientReq);
  const { body, headers, decompressed, sanitizedToolTypes } = decodeRequestBody(originalBody, clientReq.headers, url.pathname);
  await proxyPreparedBody({
    clientReq,
    clientRes,
    url,
    body,
    headers,
    startedAt,
    logExtra: { decompressed, sanitizedToolTypes }
  });
}

async function proxyPreparedBody({ clientReq, clientRes, url, body, headers, startedAt, logExtra = {} }) {
  const targetUrl = new URL(`${url.pathname}${url.search}`, upstreamBaseUrl);
  const requestImpl = targetUrl.protocol === "https:" ? httpsRequest : httpRequest;

  const upstreamReq = requestImpl({
    protocol: targetUrl.protocol,
    hostname: targetUrl.hostname,
    port: targetUrl.port || (targetUrl.protocol === "https:" ? 443 : 80),
    method: clientReq.method,
    path: `${targetUrl.pathname}${targetUrl.search}`,
    headers
  }, (upstreamRes) => {
    const responseHeaders = filterResponseHeaders(upstreamRes.headers);
    clientRes.writeHead(upstreamRes.statusCode || 502, responseHeaders);
    upstreamRes.pipe(clientRes);
    upstreamRes.on("end", () => {
      process.stderr.write(JSON.stringify({
        serviceId,
        method: clientReq.method,
        path: url.pathname,
        status: upstreamRes.statusCode || null,
        elapsedMs: Date.now() - startedAt,
        ...logExtra
      }) + "\n");
    });
  });

  upstreamReq.on("error", (error) => {
    if (!clientRes.headersSent) {
      sendJson(clientRes, 502, {
        error: {
          code: "deepseek_bridge_upstream_error",
          message: "DeepSeek bridge upstream request failed",
          reason: error.message
        }
      });
    } else {
      clientRes.end();
    }
  });

  upstreamReq.end(body);
}

function decodeRequestBody(body, headers, pathname = "") {
  const encoding = String(headers["content-encoding"] || "").trim().toLowerCase();
  const decoded = encoding === "zstd" ? Buffer.from(decompress(new Uint8Array(body))) : body;
  const sanitized = sanitizeResponsesBody(decoded, headers, pathname);
  const forwarded = filterRequestHeaders(headers);
  if (encoding === "zstd") delete forwarded["content-encoding"];
  forwarded["content-length"] = String(sanitized.body.length);
  forwarded.host = upstreamBaseUrl.host;
  return { body: sanitized.body, headers: forwarded, decompressed: encoding === "zstd", sanitizedToolTypes: sanitized.droppedToolTypes };
}

function sanitizeResponsesBody(body, headers, pathname) {
  const contentType = String(headers["content-type"] || "").toLowerCase();
  if (pathname !== "/v1/responses" || !contentType.includes("application/json")) {
    return { body, droppedToolTypes: [] };
  }
  let payload = null;
  try {
    payload = JSON.parse(body.toString("utf8"));
  } catch {
    return { body, droppedToolTypes: [] };
  }
  if (!payload || typeof payload !== "object" || !Array.isArray(payload.tools)) {
    return { body, droppedToolTypes: [] };
  }
  const keptTools = [];
  const droppedToolTypes = [];
  for (const tool of payload.tools) {
    const type = String(tool?.type || "").trim();
    if (type === "function") {
      keptTools.push(tool);
    } else {
      droppedToolTypes.push(type || "unknown");
    }
  }
  if (droppedToolTypes.length === 0) return { body, droppedToolTypes: [] };
  const sanitized = { ...payload, tools: keptTools };
  return { body: Buffer.from(JSON.stringify(sanitized), "utf8"), droppedToolTypes };
}

async function readRequestBody(req) {
  const chunks = [];
  let size = 0;
  for await (const chunk of req) {
    size += chunk.length;
    if (size > maxBodyBytes) throw new Error(`request body exceeds ${maxBodyBytes} bytes`);
    chunks.push(chunk);
  }
  return Buffer.concat(chunks);
}

async function fetchUpstreamJson(pathAndSearch, headers = {}) {
  const targetUrl = new URL(pathAndSearch, upstreamBaseUrl);
  const response = await fetch(targetUrl, {
    headers: {
      accept: "application/json",
      ...filterRequestHeaders(headers, { keepContentHeaders: false }),
      host: upstreamBaseUrl.host
    }
  });
  const text = await response.text();
  let payload = null;
  try {
    payload = text ? JSON.parse(text) : null;
  } catch {
    payload = { text };
  }
  return { ok: response.ok, status: response.status, payload };
}

function codexModelsFromUpstream(payload) {
  if (payload && typeof payload === "object" && Array.isArray(payload.models)) {
    return payload.models.map(normalizeCodexModel).filter(Boolean);
  }
  if (payload && typeof payload === "object" && Array.isArray(payload.data)) {
    return payload.data
      .map((item) => String(item?.id || item?.model_name || "").trim())
      .filter(Boolean)
      .map(codexModel);
  }
  return [];
}

function normalizeCodexModel(model) {
  const slug = String(model?.slug || model?.id || "").trim();
  return slug ? { ...codexModel(slug), ...model, slug } : null;
}

function codexModel(slug) {
  return {
    slug,
    display_name: slug,
    description: `${slug} via HWLAB DeepSeek bridge`,
    default_reasoning_level: null,
    supported_reasoning_levels: [],
    shell_type: "shell_command",
    visibility: "list",
    supported_in_api: true,
    priority: 0
  };
}

function filterRequestHeaders(headers, options = {}) {
  const keepContentHeaders = options.keepContentHeaders !== false;
  const result = {};
  for (const [name, value] of Object.entries(headers)) {
    const lower = name.toLowerCase();
    if (hopByHopHeaders.has(lower)) continue;
    if (lower === "host") continue;
    if (!keepContentHeaders && (lower === "content-length" || lower === "content-type" || lower === "content-encoding")) continue;
    result[lower] = value;
  }
  return result;
}

function filterResponseHeaders(headers) {
  const result = {};
  for (const [name, value] of Object.entries(headers)) {
    if (!hopByHopHeaders.has(name.toLowerCase())) result[name] = value;
  }
  return result;
}

function healthPayload() {
  return {
    serviceId,
    status: "ok",
    upstream: upstreamBaseUrl.origin,
    model: defaultModel,
    zstdRequestBody: "decode",
    responsesTools: "drop-non-function"
  };
}

function sendJson(res, statusCode, payload) {
  const body = JSON.stringify(payload, null, 2);
  res.writeHead(statusCode, {
    "content-type": "application/json; charset=utf-8",
    "content-length": Buffer.byteLength(body)
  });
  res.end(body);
}

function parsePositiveInteger(value, fallback) {
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : fallback;
}