UHD_Media_Tool/packages/adapters/src/rss/fetch.ts

/**
 * RSS 抓取。迁移自 MP `app/application/rss.py RssHelper.parse/_parse_impl`(GET + UA + 自定义头,非 200 视为失败,
 * 大小上限后再解析);条件请求、退避、Retry-After、熔断、脱敏为任务书新增要求(MP 无)。
 */
import { parseFeed } from "./parse.ts";
import { redactText, redactUrl } from "./redact.ts";
import { FeedError, type FeedFetchResult, type FeedFetchState } from "./types.ts";

export interface RssFetcherOptions {
  fetch?: typeof globalThis.fetch;
  now?: () => Date;
  /** 单次请求超时,默认 30s(MP `timeout or 30`) */
  timeoutMs?: number;
  /** 解压后响应体上限(字节),默认 50 MiB(MP MAX_RSS_SIZE);同时约束 gzip/br 解压量 */
  maxBytes?: number;
  /** 条目上限,默认 1000(MP MAX_RSS_ITEMS) */
  maxItems?: number;
  userAgent?: string;
  /** 指数退避:base * 2^(n-1),上限 max。默认 60s / 6h */
  backoff?: { baseMs: number; maxMs: number };
  /** 连续失败达到该次数进入熔断(open),默认 5 */
  circuitThreshold?: number;
  /** 熔断冷却后允许一次半开探测;null(默认)= 只能手动 resetFeedCircuit 恢复 */
  circuitCooldownMs?: number | null;
}

export interface FetchFeedOptions {
  signal?: AbortSignal;
  /** 手动刷新:忽略退避窗口(不忽略熔断) */
  ignoreBackoff?: boolean;
}

export function initialFeedState(): FeedFetchState {
  return { consecutiveFailures: 0, circuit: "closed" };
}

/** 手动恢复:关闭熔断并清空退避,保留 ETag/Last-Modified。 */
export function resetFeedCircuit(state: FeedFetchState): FeedFetchState {
  const { nextAttemptAt: _n, circuitOpenedAt: _c, ...rest } = state;
  return { ...rest, consecutiveFailures: 0, circuit: "closed" };
}

export function computeBackoffMs(failures: number, baseMs: number, maxMs: number): number {
  if (failures <= 0) return 0;
  return Math.min(maxMs, baseMs * 2 ** Math.min(failures - 1, 30));
}

/** Retry-After:秒数或 HTTP-date;返回毫秒,无效返回 undefined。 */
export function parseRetryAfter(value: string | null | undefined, now: Date): number | undefined {
  if (!value) return undefined;
  const v = value.trim();
  if (/^\d+$/.test(v)) return Number(v) * 1000;
  const t = Date.parse(v);
  if (Number.isNaN(t)) return undefined;
  return Math.max(0, t - now.getTime());
}

async function readCapped(res: Response, maxBytes: number): Promise<Uint8Array> {
  const len = Number(res.headers.get("content-length") ?? "");
  // content-length 是压缩后长度,只做早拒;真正的上限在流式读取里按解压后字节执行
  if (Number.isFinite(len) && len > maxBytes) {
    await res.body?.cancel().catch(() => {});
    throw new FeedError("too_large", `响应声明长度 ${len} 超过上限 ${maxBytes}`);
  }
  if (!res.body) return new Uint8Array();
  const reader = res.body.getReader();
  const chunks: Uint8Array[] = [];
  let total = 0;
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    total += value.byteLength;
    if (total > maxBytes) {
      await reader.cancel().catch(() => {});
      throw new FeedError("too_large", `响应体(解压后)超过上限 ${maxBytes} 字节`);
    }
    chunks.push(value);
  }
  const out = new Uint8Array(total);
  let off = 0;
  for (const c of chunks) {
    out.set(c, off);
    off += c.byteLength;
  }
  return out;
}

function decodeBody(bytes: Uint8Array, contentType: string | null): string {
  const head = new TextDecoder("latin1").decode(bytes.subarray(0, 256));
  const label =
    /charset=([\w-]+)/i.exec(contentType ?? "")?.[1] ??
    /<\?xml[^>]*encoding=["']([\w-]+)["']/i.exec(head)?.[1] ??
    "utf-8";
  try {
    return new TextDecoder(label).decode(bytes);
  } catch {
    return new TextDecoder("utf-8").decode(bytes);
  }
}

export interface RssFetcher {
  fetchFeed(url: string, state?: FeedFetchState, opts?: FetchFeedOptions): Promise<FeedFetchResult>;
}

export function createRssFetcher(options: RssFetcherOptions = {}): RssFetcher {
  const doFetch = options.fetch ?? globalThis.fetch;
  const now = options.now ?? (() => new Date());
  const timeoutMs = options.timeoutMs ?? 30_000;
  const maxBytes = options.maxBytes ?? 50 * 1024 * 1024;
  const backoff = options.backoff ?? { baseMs: 60_000, maxMs: 6 * 3600_000 };
  const threshold = options.circuitThreshold ?? 5;
  const cooldown = options.circuitCooldownMs ?? null;

  function fail(state: FeedFetchState, error: FeedError): FeedFetchResult {
    const at = now();
    const failures = state.consecutiveFailures + 1;
    const wait = Math.max(computeBackoffMs(failures, backoff.baseMs, backoff.maxMs), error.retryAfterMs ?? 0);
    const next: FeedFetchState = {
      ...state,
      consecutiveFailures: failures,
      nextAttemptAt: new Date(at.getTime() + wait).toISOString(),
      lastError: {
        kind: error.kind,
        message: error.message,
        at: at.toISOString(),
        ...(error.httpStatus !== undefined ? { httpStatus: error.httpStatus } : {}),
      },
    };
    if (failures >= threshold) {
      next.circuit = "open";
      next.circuitOpenedAt = at.toISOString();
    }
    return { kind: "error", error, state: next };
  }

  async function fetchFeed(
    url: string,
    state: FeedFetchState = initialFeedState(),
    opts: FetchFeedOptions = {},
  ): Promise<FeedFetchResult> {
    const at = now();
    if (state.circuit === "open") {
      const openedAt = state.circuitOpenedAt ? Date.parse(state.circuitOpenedAt) : 0;
      const halfOpen = cooldown !== null && at.getTime() - openedAt >= cooldown;
      if (!halfOpen) {
        return {
          kind: "skipped",
          reason: "circuit_open",
          ...(cooldown !== null ? { retryAt: new Date(openedAt + cooldown).toISOString() } : {}),
          state,
        };
      }
    } else if (!opts.ignoreBackoff && state.nextAttemptAt && Date.parse(state.nextAttemptAt) > at.getTime()) {
      return { kind: "skipped", reason: "backoff", retryAt: state.nextAttemptAt, state };
    }

    const safeUrl = redactUrl(url);
    const headers: Record<string, string> = {
      Accept: "application/rss+xml, application/atom+xml, application/xml;q=0.9, text/xml;q=0.8, */*;q=0.1",
    };
    if (options.userAgent) headers["User-Agent"] = options.userAgent;
    if (state.etag) headers["If-None-Match"] = state.etag;
    if (state.lastModified) headers["If-Modified-Since"] = state.lastModified;

    const timeout = AbortSignal.timeout(timeoutMs);
    const signal = opts.signal ? AbortSignal.any([opts.signal, timeout]) : timeout;
    let res: Response;
    try {
      res = await doFetch(url, { headers, signal, redirect: "follow" });
    } catch (e) {
      if (opts.signal?.aborted) throw e;
      const err = e as Error;
      const kind = timeout.aborted ? "timeout" : "network";
      return fail(state, new FeedError(kind, redactText(`请求 ${safeUrl} 失败:${err.message}`)));
    }

    if (res.status === 304) {
      await res.body?.cancel().catch(() => {});
      const next: FeedFetchState = {
        ...state,
        consecutiveFailures: 0,
        circuit: "closed",
        lastSuccessAt: at.toISOString(),
      };
      delete next.nextAttemptAt;
      delete next.circuitOpenedAt;
      return { kind: "not_modified", state: next };
    }
    if (!res.ok) {
      await res.body?.cancel().catch(() => {});
      const retryAfterMs = parseRetryAfter(res.headers.get("retry-after"), at);
      const kind =
        res.status === 429 || (res.status === 503 && retryAfterMs !== undefined) ? "rate_limited" : "http_error";
      return fail(
        state,
        new FeedError(kind, `请求 ${safeUrl} 返回 HTTP ${res.status}`, {
          httpStatus: res.status,
          ...(retryAfterMs !== undefined ? { retryAfterMs } : {}),
        }),
      );
    }

    try {
      const bytes = await readCapped(res, maxBytes);
      const text = decodeBody(bytes, res.headers.get("content-type"));
      const feed = parseFeed(text, { maxItems: options.maxItems ?? 1000, maxChars: maxBytes, fetchedAt: at });
      const next: FeedFetchState = {
        ...state,
        consecutiveFailures: 0,
        circuit: "closed",
        lastSuccessAt: at.toISOString(),
      };
      delete next.nextAttemptAt;
      delete next.circuitOpenedAt;
      const etag = res.headers.get("etag");
      const lastModified = res.headers.get("last-modified");
      if (etag) next.etag = etag;
      if (lastModified) next.lastModified = lastModified;
      return { kind: "ok", feed, state: next, httpStatus: res.status };
    } catch (e) {
      if (opts.signal?.aborted) throw e;
      if (e instanceof FeedError) return fail(state, new FeedError(e.kind, redactText(e.message)));
      const kind = timeout.aborted ? "timeout" : "network";
      return fail(state, new FeedError(kind, redactText(`读取 ${safeUrl} 失败:${(e as Error).message}`)));
    }
  }

  return { fetchFeed };
}