feat(api): complete phase 2.4 control plane

2026-07-17 00:37:11 +08:00
parent c2702b5fc6
commit 7b91dbbad1
40 changed files with 4526 additions and 74 deletions
@@ -0,0 +1,110 @@
+import * as http from 'node:http';
+import * as https from 'node:https';
+import type { LookupFunction } from 'node:net';
+import type {
+  PinnedRequest,
+  ResolvedAddress,
+  TransportResponse,
+} from './safe-instance-transport.js';
+import { asUpstreamError, UpstreamError } from './upstream-error.js';
+
+export interface PinnedDispatchOptions {
+  readonly url: URL;
+  readonly method: 'GET' | 'POST';
+  readonly headers: Readonly<Record<string, string>>;
+  readonly body?: string;
+  readonly lookup: LookupFunction;
+  readonly followRedirects: false;
+  readonly timeoutMs: number;
+  readonly maxBodyBytes: number;
+}
+export interface PinnedHttpRequesterOptions {
+  readonly dispatch?: (options: PinnedDispatchOptions) => Promise<TransportResponse>;
+  readonly timeoutMs?: number;
+  readonly maxBodyBytes?: number;
+}
+const DEFAULT_TIMEOUT_MS = 5_000;
+const DEFAULT_MAX_BODY_BYTES = 65_536;
+const lookupFor = (addresses: readonly ResolvedAddress[]): LookupFunction => {
+  const first = addresses[0];
+  if (!first) throw new Error('No validated address');
+  return (_hostname, _options, callback) => callback(null, first.address, first.family);
+};
+export class PinnedHttpRequester {
+  private readonly dispatch: (options: PinnedDispatchOptions) => Promise<TransportResponse>;
+  private readonly timeoutMs: number;
+  private readonly maxBodyBytes: number;
+  constructor(options: PinnedHttpRequesterOptions = {}) {
+    this.dispatch = options.dispatch ?? dispatchNative;
+    this.timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+    this.maxBodyBytes = options.maxBodyBytes ?? DEFAULT_MAX_BODY_BYTES;
+  }
+  async request(
+    request: PinnedRequest,
+    addresses: readonly ResolvedAddress[],
+  ): Promise<TransportResponse> {
+    try {
+      return await this.dispatch({
+        url: new URL(request.url),
+        method: request.method,
+        headers: request.headers,
+        ...(request.body === undefined ? {} : { body: request.body }),
+        lookup: lookupFor(addresses),
+        followRedirects: false,
+        timeoutMs: this.timeoutMs,
+        maxBodyBytes: this.maxBodyBytes,
+      });
+    } catch (error) {
+      throw asUpstreamError(error);
+    }
+  }
+}
+const dispatchNative = (options: PinnedDispatchOptions): Promise<TransportResponse> =>
+  new Promise((resolve, reject) => {
+    const client = options.url.protocol === 'https:' ? https : http;
+    const request = client.request({
+      protocol: options.url.protocol,
+      hostname: options.url.hostname,
+      port: options.url.port || undefined,
+      path: `${options.url.pathname}${options.url.search}`,
+      method: options.method,
+      headers: options.headers,
+      lookup: options.lookup,
+      agent: false,
+      timeout: options.timeoutMs,
+      maxHeaderSize: 16_384,
+      ...(options.url.protocol === 'https:' ? { servername: options.url.hostname } : {}),
+    });
+    let settled = false;
+    const fail = (error: Error): void => {
+      if (settled) return;
+      settled = true;
+      request.destroy(error);
+      reject(error);
+    };
+    request.once('timeout', () => fail(new UpstreamError('UPSTREAM_TIMEOUT')));
+    request.once('error', fail);
+    request.once('response', (response) => {
+      const chunks: Buffer[] = [];
+      let bytes = 0;
+      response.on('data', (chunk: Buffer) => {
+        bytes += chunk.length;
+        if (bytes > options.maxBodyBytes) fail(new UpstreamError('UPSTREAM_RESPONSE_TOO_LARGE'));
+        else chunks.push(chunk);
+      });
+      response.once('error', fail);
+      response.once('end', () => {
+        if (settled) return;
+        settled = true;
+        const headers: Record<string, string | undefined> = {};
+        for (const [key, value] of Object.entries(response.headers))
+          headers[key] = Array.isArray(value) ? value.join(', ') : value;
+        resolve({
+          status: response.statusCode ?? 0,
+          headers,
+          body: Buffer.concat(chunks).toString('utf8'),
+        });
+      });
+    });
+    request.end(options.body);
+  });