v1.3.1: improve TypeScript client reliability

2026-08-21 15:06:11 +08:00
parent 503d0904d5
commit eb290cc88c
9 changed files with 1024 additions and 65 deletions
@@ -12,7 +12,15 @@ func (ctx templateTs) genClientTemplate() string {
 };

 export type resultStatus = 0 | 1 | 2 | 4 | 5;
-// 0 成功；1 框架错误；2 业务错误；4 网络错误；5 超时
+// 0 success; 1 framework/client protocol error; 2 business error; 4 external request error; 5 external timeout
+
+export type RequestOptions = {
+  signal?: AbortSignal;
+};
+
+export type StreamOptions = {
+  signal?: AbortSignal;
+};

 export type RequestInterceptor = (
  serviceName: string,
@@ -21,13 +29,110 @@ export type RequestInterceptor = (
  dto?: any
 ) => Promise<void> | void;

-// 返回新 result 将替换原结果继续向下传递（可用于集中换 token / 错误处理）
 export type ResponseInterceptor = (
  serviceName: string,
  methodName: string,
  result: result<any>
 ) => Promise<result<any> | void> | result<any> | void;

+function messageOf(error: unknown): string {
+  if (error instanceof Error && error.message) return error.message;
+  if (typeof error === "string" && error) return error;
+  return "unknown error";
+}
+
+function failure(status: resultStatus, msg: string): result<any> {
+  return { status, msg };
+}
+
+function isTimeout(error: unknown, signal?: AbortSignal): boolean {
+  const errorName = error !== null && typeof error === "object"
+    ? (error as { name?: unknown }).name
+    : undefined;
+  const reason = signal?.reason;
+  const reasonName = reason !== null && typeof reason === "object"
+    ? (reason as { name?: unknown }).name
+    : undefined;
+  return errorName === "TimeoutError" || reasonName === "TimeoutError";
+}
+
+function requestFailure(error: unknown, signal: AbortSignal | undefined, stream: boolean): result<any> {
+  if (isTimeout(error, signal)) {
+    return failure(5, stream ? "Stream timed out" : "Request timed out");
+  }
+  if (signal?.aborted === true) {
+    return failure(4, stream ? "Stream aborted" : "Request aborted");
+  }
+  const kind = stream ? "External stream request" : "External request";
+  return failure(4, ` + "`${kind} failed: ${messageOf(error)}`" + `);
+}
+
+function isResult(value: unknown): value is result<any> {
+  return value !== null && typeof value === "object" &&
+    typeof (value as { status?: unknown }).status === "number";
+}
+
+function mediaType(response: Response): string {
+  return (response.headers.get("content-type") || "").split(";", 1)[0].trim().toLowerCase();
+}
+
+function excerpt(text: string, limit = 180): string {
+  const value = text.replace(/\s+/g, " ").trim();
+  return value.length <= limit ? value : ` + "`${value.slice(0, limit)}...`" + `;
+}
+
+function externalFailure(response: Response, detail?: string): result<any> {
+  const timeout = response.status === 408 || response.status === 504;
+  const statusText = response.statusText || (timeout ? "timeout" : "request failed");
+  const suffix = detail ? ` + "`: ${detail}`" + ` : "";
+  return failure(timeout ? 5 : 4, ` + "`HTTP ${response.status} ${statusText}${suffix}`" + `);
+}
+
+function responseReadFailure(
+  error: unknown,
+  response: Response,
+  signal: AbortSignal | undefined,
+  stream: boolean
+): result<any> {
+  if (isTimeout(error, signal)) {
+    return failure(5, stream ? "Stream timed out" : "Request timed out");
+  }
+  if (signal?.aborted === true) {
+    return failure(4, stream ? "Stream aborted" : "Request aborted");
+  }
+  const kind = stream ? "Stream" : "Response body";
+  return response.ok
+    ? failure(4, ` + "`${kind} failed: ${messageOf(error)}`" + `)
+    : externalFailure(response, ` + "`response body failed: ${messageOf(error)}`" + `);
+}
+
+function parseResult(response: Response, text: string): result<any> {
+  const body = text.trim();
+  if (!body) {
+    return response.ok
+      ? failure(1, "Empty response body")
+      : externalFailure(response);
+  }
+
+  let value: unknown;
+  try {
+    value = JSON.parse(body);
+  } catch {
+    if (!response.ok) return externalFailure(response, excerpt(body));
+    const type = mediaType(response);
+    if (type === "text/html" || /^\s*(?:<!doctype\s+html|<html\b)/i.test(body)) {
+      return failure(1, ` + "`Unexpected HTML response: ${excerpt(body)}`" + `);
+    }
+    return failure(1, ` + "`Invalid JSON response: ${excerpt(body)}`" + `);
+  }
+  if (!isResult(value)) {
+    return response.ok
+      ? failure(1, "Invalid fun response")
+      : externalFailure(response, "invalid fun response");
+  }
+  return value;
+}
+
 export class Client {
  private url: string;
  private state: Record<string, string> = {};
@@ -42,81 +147,293 @@ export class Client {
    this.state = state;
  }

-  addRequestInterceptor(i: RequestInterceptor) {
-    this.requestInterceptors.push(i);
+  addRequestInterceptor(interceptor: RequestInterceptor) {
+    this.requestInterceptors.push(interceptor);
  }

-  addResponseInterceptor(i: ResponseInterceptor) {
-    this.responseInterceptors.push(i);
+  addResponseInterceptor(interceptor: ResponseInterceptor) {
+    this.responseInterceptors.push(interceptor);
  }

-  async request<T>(serviceName: string, methodName: string, dto?: any): Promise<result<T>> {
-    const state: Record<string, string> = { ...this.state };
-    for (const i of this.requestInterceptors) {
-      await i(serviceName, methodName, state, dto);
+  private async interceptResponse(
+    serviceName: string,
+    methodName: string,
+    initial: result<any>
+  ): Promise<result<any>> {
+    let current = initial;
+    for (const interceptor of this.responseInterceptors) {
+      try {
+        const replaced = await interceptor(serviceName, methodName, current);
+        if (replaced) current = replaced;
+      } catch (error) {
+        current = failure(1, ` + "`Response interceptor failed: ${messageOf(error)}`" + `);
+      }
    }
-    let out: result<T>;
+    return current;
+  }
+
+  private async requestState(serviceName: string, methodName: string, dto: any): Promise<Record<string, string>> {
+    const state: Record<string, string> = { ...this.state };
+    for (const interceptor of this.requestInterceptors) {
+      await interceptor(serviceName, methodName, state, dto);
+    }
+    return state;
+  }
+
+  async request<T>(
+    serviceName: string,
+    methodName: string,
+    dto?: any,
+    options?: RequestOptions
+  ): Promise<result<T>> {
+    let state: Record<string, string>;
    try {
-      const res = await fetch(this.url + "/cell", {
+      state = await this.requestState(serviceName, methodName, dto);
+    } catch (error) {
+      return await this.interceptResponse(
+        serviceName,
+        methodName,
+        failure(1, ` + "`Request interceptor failed: ${messageOf(error)}`" + `)
+      ) as result<T>;
+    }
+
+    let body: string;
+    try {
+      const serialized = JSON.stringify({
+        serviceName,
+        methodName,
+        data: dto,
+        ...(Object.keys(state).length ? { state } : {}),
+      });
+      if (serialized === undefined) throw new Error("serialization produced no output");
+      body = serialized;
+    } catch (error) {
+      return await this.interceptResponse(
+        serviceName,
+        methodName,
+        failure(1, ` + "`Could not serialize request: ${messageOf(error)}`" + `)
+      ) as result<T>;
+    }
+
+    let output: result<any>;
+    try {
+      const response = await fetch(` + "`${this.url}/cell`" + `, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({ serviceName, methodName, data: dto, ...(Object.keys(state).length ? { state } : {}) }),
+        body,
+        signal: options?.signal,
      });
-      out = (await res.json()) as result<T>;
-    } catch (e: any) {
-      out = { status: 4, msg: (e && e.message) || "网络错误" } as result<T>;
+      try {
+        output = parseResult(response, await response.text());
+      } catch (error) {
+        output = responseReadFailure(error, response, options?.signal, false);
+      }
+    } catch (error) {
+      output = requestFailure(error, options?.signal, false);
    }
-    let cur: result<any> = out as result<any>;
-    for (const i of this.responseInterceptors) {
-      const replaced = await i(serviceName, methodName, cur);
-      if (replaced) cur = replaced;
-    }
-    return cur as result<T>;
+    return await this.interceptResponse(serviceName, methodName, output) as result<T>;
  }

  async stream<T>(
    serviceName: string,
    methodName: string,
    dto: any | undefined,
-    onMessage: (data: T) => void
-  ): Promise<void> {
-    const state: Record<string, string> = { ...this.state };
-    for (const i of this.requestInterceptors) {
-      await i(serviceName, methodName, state, dto);
+    onMessage: (data: T) => unknown,
+    options?: StreamOptions
+  ): Promise<result<void>> {
+    let state: Record<string, string>;
+    try {
+      state = await this.requestState(serviceName, methodName, dto);
+    } catch (error) {
+      return await this.interceptResponse(
+        serviceName,
+        methodName,
+        failure(1, ` + "`Request interceptor failed: ${messageOf(error)}`" + `)
+      );
    }
-    const res = await fetch(this.url + "/cell", {
-      method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({ serviceName, methodName, data: dto, ...(Object.keys(state).length ? { state } : {}) }),
-    });
-    if (!res.ok) return;
-    const anyResult: result<any> = { status: 0 };
-    for (const i of this.responseInterceptors) {
-      await i(serviceName, methodName, anyResult);
+
+    let body: string;
+    try {
+      const serialized = JSON.stringify({
+        serviceName,
+        methodName,
+        data: dto,
+        ...(Object.keys(state).length ? { state } : {}),
+      });
+      if (serialized === undefined) throw new Error("serialization produced no output");
+      body = serialized;
+    } catch (error) {
+      return await this.interceptResponse(
+        serviceName,
+        methodName,
+        failure(1, ` + "`Could not serialize request: ${messageOf(error)}`" + `)
+      );
    }
-    if (!res.body) return;
-    const reader = res.body.getReader();
-    const decoder = new TextDecoder();
+
+    let response: Response;
+    try {
+      response = await fetch(` + "`${this.url}/cell`" + `, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body,
+        signal: options?.signal,
+      });
+    } catch (error) {
+      return await this.interceptResponse(
+        serviceName,
+        methodName,
+        requestFailure(error, options?.signal, true)
+      );
+    }
+
+    if (!response.ok) {
+      let text: string;
+      try {
+        text = await response.text();
+      } catch (error) {
+        return await this.interceptResponse(
+          serviceName,
+          methodName,
+          responseReadFailure(error, response, options?.signal, true)
+        );
+      }
+      return await this.interceptResponse(serviceName, methodName, parseResult(response, text));
+    }
+
+    if (mediaType(response) !== "application/x-ndjson") {
+      let text: string;
+      try {
+        text = await response.text();
+      } catch (error) {
+        return await this.interceptResponse(
+          serviceName,
+          methodName,
+          responseReadFailure(error, response, options?.signal, true)
+        );
+      }
+      const rpcResult = parseResult(response, text);
+      return await this.interceptResponse(
+        serviceName,
+        methodName,
+        rpcResult.status === 0
+          ? failure(1, "Expected application/x-ndjson response")
+          : rpcResult
+      );
+    }
+
+    if (!response.body) {
+      return await this.interceptResponse(serviceName, methodName, { status: 0 });
+    }
+
+    let reader: ReadableStreamDefaultReader<Uint8Array>;
+    try {
+      reader = response.body.getReader();
+    } catch (error) {
+      return await this.interceptResponse(
+        serviceName,
+        methodName,
+        responseReadFailure(error, response, options?.signal, true)
+      );
+    }
+    const decoder = new TextDecoder("utf-8", { fatal: true });
    let buffer = "";
-    for (;;) {
-      const { done, value } = await reader.read();
-      if (done) break;
-      buffer += decoder.decode(value, { stream: true });
-      const lines = buffer.split("\n");
-      buffer = lines.pop() ?? "";
-      for (const line of lines) {
-        const payload = line.trim();
-        if (!payload) continue;
-        const data = JSON.parse(payload) as T;
-        onMessage(data);
+    let lineNumber = 0;
+    let failed: result<any> | undefined;
+    let cause: unknown;
+
+    const emitLine = async (line: string) => {
+      const payload = line.replace(/\r$/, "").trim();
+      if (!payload) return;
+      let data: T;
+      try {
+        data = JSON.parse(payload) as T;
+      } catch (error) {
+        failed = failure(1, ` + "`Invalid NDJSON at line ${lineNumber}: ${excerpt(payload)}`" + `);
+        cause = error;
+        return;
+      }
+      try {
+        await onMessage(data);
+      } catch (error) {
+        failed = failure(1, ` + "`Stream callback failed: ${messageOf(error)}`" + `);
+        cause = error;
+      }
+    };
+
+    try {
+      for (;;) {
+        let part: ReadableStreamReadResult<Uint8Array>;
+        try {
+          part = await reader.read();
+        } catch (error) {
+          cause = error;
+          if (isTimeout(error, options?.signal)) {
+            failed = failure(5, "Stream timed out");
+          } else if (options?.signal?.aborted === true) {
+            failed = failure(4, "Stream aborted");
+          } else {
+            failed = failure(4, ` + "`Stream read failed: ${messageOf(error)}`" + `);
+          }
+          break;
+        }
+        if (part.done) break;
+        try {
+          buffer += decoder.decode(part.value, { stream: true });
+        } catch (error) {
+          failed = failure(1, ` + "`Invalid UTF-8 stream data: ${messageOf(error)}`" + `);
+          cause = error;
+          break;
+        }
+        for (;;) {
+          const newline = buffer.indexOf("\n");
+          if (newline < 0) break;
+          const line = buffer.slice(0, newline);
+          buffer = buffer.slice(newline + 1);
+          lineNumber++;
+          await emitLine(line);
+          if (failed) break;
+        }
+        if (failed) break;
+      }
+
+      if (!failed) {
+        try {
+          buffer += decoder.decode();
+        } catch (error) {
+          failed = failure(1, ` + "`Invalid UTF-8 stream data: ${messageOf(error)}`" + `);
+          cause = error;
+        }
+      }
+      if (!failed && buffer.length > 0) {
+        lineNumber++;
+        await emitLine(buffer);
+      }
+
+    } finally {
+      if (failed) {
+        try {
+          await reader.cancel(cause);
+        } catch {
+          // The reader may already be closed by the runtime.
+        }
+      }
+      try {
+        reader.releaseLock();
+      } catch {
+        // The reader may already be errored or released.
      }
    }
+    return await this.interceptResponse(
+      serviceName,
+      methodName,
+      failed || { status: 0 }
+    );
  }
 }`
 }

 func (ctx templateTs) genDefaultServiceTemplate() string {
-	return `import { Client, type result } from "./client";
+	return `import { Client } from "./client";
 {{- range .GenServiceList}}
 import {{.ServiceName}} from "./{{.ServiceName}}";
 {{- end}}
@@ -138,7 +455,7 @@ export default class api {
 }

 func (ctx templateTs) genServiceTemplate() string {
-	return `import { Client, type result } from "./client"
+	return `import { Client{{if .IsIncludeRequest}}, type result, type RequestOptions{{end}}{{if .IsIncludeStream}}{{if not .IsIncludeRequest}}, type result{{end}}, type StreamOptions{{end}} } from "./client";
 {{- range .GenImport}}
 import type {{.Name}} from "./{{.Name}}";
 {{- end}}
@@ -150,10 +467,10 @@ export default class {{.ServiceName}} {
  }
  {{- $serviceName := .ServiceName }}
  {{- range .GenMethodTypeList}}
-  {{if .IsStream }}async {{.MethodName}}({{.DtoText}}{{if .DtoText}},{{end}}onMessage: (data: {{.GenericTypeText}}) => void): Promise<void> {
-    return await this.client.stream<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, onMessage)
-  }{{else}}async {{.MethodName}}({{.DtoText}}): Promise<{{.ReturnValueText}}> {
-    return await this.client.request<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}"{{.ArgsText}})
+  {{if .IsStream }}async {{.MethodName}}({{if .DtoText}}{{.DtoText}}, {{end}}onMessage: (data: {{.GenericTypeText}}) => unknown, options?: StreamOptions): Promise<result<void>> {
+    return await this.client.stream<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, onMessage, options)
+  }{{else}}async {{.MethodName}}({{if .DtoText}}{{.DtoText}}, {{end}}options?: RequestOptions): Promise<{{.ReturnValueText}}> {
+    return await this.client.request<{{.GenericTypeText}}>("{{$serviceName}}", "{{.MethodName}}", {{if .DtoText}}dto{{else}}undefined{{end}}, options)
  }{{end}}
  {{- end}}
 }`