UHD_Media_Tool/apps/server/src/modules/sources-identity/runtime.ts

import {
  createRssFetcher,
  createSharedSiteRateLimiter,
  createSiteAdapter,
  createSiteRateLimiter,
  createTmdbClient,
  getSiteDefinition,
  type RssFetcher,
  type SecretResolver,
  type SiteAdapter,
  type SiteDefinition,
  type SiteRateLimiter,
  type TmdbClient,
  type TmdbSharedCache,
} from "@umt/adapters";
import { DEFAULT_SETTINGS, type SitesSettings } from "@umt/contracts";
import { pgTokenBucketFor, type SiteRow, sql } from "@umt/db";
import { CACHE_REGIONS, sharedCache, type TmdbCachePolicy, tmdbCachePolicy } from "../../shared-cache.ts";
import type { ServerContext } from "../types.ts";

// 模块运行时依赖:站点适配器与限流器(每站一个实例,令牌在 PG 跨进程共享)、RSS 抓取器、TMDB 客户端、时钟。
// 测试通过 createSourcesIdentityModule(overrides) 注入 fetch / TMDB / 时钟。

export interface RuntimeOverrides {
  /** 站点与 RSS 请求使用的 fetch(测试指向 mock) */
  fetch?: typeof globalThis.fetch;
  /** TMDB 客户端工厂;返回 null 表示未配置(只做本地召回) */
  tmdb?: (() => TmdbClient | null) | undefined;
  now?: () => Date;
  /** RSS 抓取器选项覆盖(退避/熔断阈值) */
  rss?: Parameters<typeof createRssFetcher>[0];
  /** 调度 tick 间隔(毫秒),默认 30s */
  schedulerIntervalMs?: number;
  /** false = 站点限流只用进程内令牌桶(缺省走 PG 跨进程令牌桶 rate_limit_bucket) */
  sharedRateLimit?: boolean;
}

interface CachedAdapter {
  sig: string;
  limiter: SiteRateLimiter;
  adapter: SiteAdapter;
}

export class SourcesRuntime {
  readonly now: () => Date;
  readonly rss: RssFetcher;
  private readonly adapters = new Map<string, CachedAdapter>();
  private tmdbClient: TmdbClient | null | undefined;
  /** 站点默认(settings.sites)缓存:siteAdapter 是同步的,读设置的地方顺手刷新 */
  private siteDefaultsCache: SitesSettings = DEFAULT_SETTINGS.sites;
  private siteDefaultsAt = 0;
  private tmdbSig: string | null = null;

  constructor(
    readonly ctx: ServerContext,
    readonly overrides: RuntimeOverrides = {},
  ) {
    this.now = overrides.now ?? (() => new Date());
    this.rss = createRssFetcher({
      ...(overrides.fetch ? { fetch: overrides.fetch } : {}),
      now: this.now,
      ...overrides.rss,
    });
  }

  /**
   * TMDB 客户端:优先注入(测试);否则读设置 metadata.tmdbApiKeySecretRef(secret 现场解密)与 tmdbBaseUrl,
   * 未配置时回退环境变量 TMDB_API_KEY / TMDB_ACCESS_TOKEN / TMDB_BASE_URL。设置变化(引用/地址)时重建。
   */
  async tmdb(): Promise<TmdbClient | null> {
    if (this.overrides.tmdb) {
      if (this.tmdbClient === undefined) this.tmdbClient = this.overrides.tmdb();
      return this.tmdbClient;
    }
    const meta = await this.ctx.settings.get("metadata").catch(() => null);
    const policy = await tmdbCachePolicy(this.ctx);
    const env = (k: string) => Deno.env.get(k);
    const ref = meta?.tmdbApiKeySecretRef ?? null;
    const baseUrl = meta?.tmdbBaseUrl ?? env("TMDB_BASE_URL");
    const sig = JSON.stringify([
      ref,
      baseUrl ?? null,
      policy,
      ref ? null : [!!env("TMDB_API_KEY"), !!env("TMDB_ACCESS_TOKEN")],
    ]);
    if (this.tmdbClient !== undefined && this.tmdbSig === sig) return this.tmdbClient;
    let apiKey: string | undefined;
    let accessToken: string | undefined;
    if (ref) {
      try {
        const secret = (await this.ctx.secrets.resolve(ref)).trim();
        // v3 API Key 是 32 位十六进制;其余(v4 read access token,JWT)走 Bearer
        if (/^[0-9a-f]{32}$/i.test(secret)) apiKey = secret;
        else accessToken = secret;
      } catch (err) {
        this.ctx.logger.warn({ err: { message: (err as Error).message } }, "TMDB 密钥解密失败，按未配置处理");
      }
    } else {
      apiKey = env("TMDB_API_KEY");
      accessToken = env("TMDB_ACCESS_TOKEN");
    }
    this.tmdbSig = sig;
    this.tmdbClient =
      apiKey || accessToken
        ? createTmdbClient({
            ...(apiKey ? { apiKey } : {}),
            ...(accessToken ? { accessToken } : {}),
            ...(baseUrl ? { baseUrl } : {}),
            ...(this.overrides.fetch ? { fetch: this.overrides.fetch } : {}),
            // 请求响应进 PG 缓存(region tmdb.request),三进程共享;TTL / 条数上限 / 空结果短 TTL 取设置(对照 MP)
            cache: { ttlMs: policy.ttl * 1000, emptyTtlMs: policy.emptyTtl * 1000 },
            sharedCache: tmdbSharedCache(this.ctx, policy),
          })
        : null;
    return this.tmdbClient;
  }

  /** 站点默认(settings.sites),读库并刷新缓存;读失败用上次值 */
  async siteDefaults(): Promise<SitesSettings> {
    try {
      this.siteDefaultsCache = { ...DEFAULT_SETTINGS.sites, ...(await this.ctx.settings.get("sites")) };
    } catch {
      // 保留缓存
    }
    return this.siteDefaultsCache;
  }

  /**
   * 站点请求用的 fetch:测试注入优先;站点开了"使用代理"且设置了 proxyUrl 时走代理(Deno.createHttpClient)。
   */
  siteFetch(site: Pick<SiteRow, "proxy">, defaults: SitesSettings = this.siteDefaultsCache): typeof globalThis.fetch {
    if (this.overrides.fetch) return this.overrides.fetch;
    if (!site.proxy || !defaults.proxyUrl) return globalThis.fetch;
    const key = defaults.proxyUrl;
    let client = this.proxyClients.get(key);
    if (!client) {
      client = Deno.createHttpClient({ proxy: { url: key } });
      this.proxyClients.set(key, client);
    }
    const c = client;
    return ((input: RequestInfo | URL, init?: RequestInit) =>
      globalThis.fetch(input, { ...init, client: c } as RequestInit)) as typeof globalThis.fetch;
  }
  private readonly proxyClients = new Map<string, Deno.HttpClient>();

  /**
   * 站点适配器:每站一个实例,列表/搜索/详情/下载共享同一个限流器。
   * 站点限流参数、基址、代理或站点默认变化时重建(签名比较)。
   */
  siteAdapter(
    site: Pick<SiteRow, "id" | "key" | "base_url" | "rate_limit" | "config"> & Partial<Pick<SiteRow, "proxy">>,
  ): SiteAdapter {
    const def = getSiteDefinition(site.key);
    if (!def) throw new Error(`站点 ${site.key} 没有随产品发布的适配器定义`);
    if (Date.now() - this.siteDefaultsAt > 30_000) {
      this.siteDefaultsAt = Date.now();
      void this.siteDefaults();
    }
    const defaults = this.siteDefaultsCache;
    const proxy = site.proxy ?? false;
    const sig = JSON.stringify([
      site.base_url,
      site.rate_limit,
      site.config,
      proxy,
      proxy ? defaults.proxyUrl : null,
      defaults.defaultRateLimit,
    ]);
    const hit = this.adapters.get(site.id);
    if (hit && hit.sig === sig) return hit.adapter;
    const lo = limiterOptions(def, site.rate_limit, defaults.defaultRateLimit);
    // 跨进程:api / worker / worker:agent 对同一站点共用 PG 令牌桶(key = site:<id>);进程内先串行排队
    const inner =
      this.overrides.sharedRateLimit === false
        ? createSiteRateLimiter(lo)
        : createSharedSiteRateLimiter({
            ...lo,
            key: `site:${site.id}`,
            bucket: pgTokenBucketFor(this.ctx.db),
            onError: (err) =>
              this.ctx.logger.warn(
                { siteId: site.id, err: { message: (err as Error).message } },
                "站点共享限流不可用，本次退回进程内限流",
              ),
          });
    const limiter = countingLimiter(inner, () => this.countSiteRequest(site.id));
    const cfg = (site.config ?? {}) as { apiBaseUrl?: string };
    const baseUrl =
      def.umt.kind === "mteam" ? cfg.apiBaseUrl : sameUrl(site.base_url, def.domain) ? undefined : site.base_url;
    const adapter = createSiteAdapter(def, {
      secrets: this.siteSecrets(site.id),
      limiter,
      now: this.now,
      ...(this.overrides.fetch || proxy ? { fetch: this.siteFetch({ proxy }, defaults) } : {}),
      ...(baseUrl ? { baseUrl } : {}),
    });
    this.adapters.set(site.id, { sig, limiter, adapter });
    return adapter;
  }

  /**
   * 站点请求计数(site_request_count,按小时聚合):每拿到一个令牌记一次。写失败只记日志,不影响请求;
   * 顺手删 7 天前的行(每进程每小时最多一次)。
   */
  private lastCountPrune = 0;
  countSiteRequest(siteId: string): void {
    const hour = new Date(Math.floor(this.now().getTime() / 3_600_000) * 3_600_000);
    const db = this.ctx.db;
    void sql`insert into site_request_count (site_id, hour, count) values (${siteId}, ${hour}, 1)
      on conflict (site_id, hour) do update set count = site_request_count.count + 1`
      .execute(db)
      .then(async () => {
        if (Date.now() - this.lastCountPrune < 3_600_000) return;
        this.lastCountPrune = Date.now();
        await sql`delete from site_request_count where hour < ${new Date(hour.getTime() - 7 * 86_400_000)}`.execute(db);
      })
      .catch((err) =>
        this.ctx.logger.warn({ siteId, err: { message: (err as Error).message } }, "站点请求计数写入失败"),
      );
  }

  /** 最近 hours 小时(含当前小时)该站点的请求次数 */
  async siteRequestCount(siteId: string, hours = 24): Promise<number> {
    const hour = Math.floor(this.now().getTime() / 3_600_000) * 3_600_000;
    const since = new Date(hour - (hours - 1) * 3_600_000);
    const r = await sql<{ n: number }>`select coalesce(sum(count), 0)::int as n from site_request_count
      where site_id = ${siteId} and hour >= ${since}`.execute(this.ctx.db);
    return Number(r.rows[0]?.n ?? 0);
  }

  /** 站点访问默认:站点定义 umt.defaults 与 settings.sites 取更保守的一方 */
  async siteAccess(siteKey: string): Promise<SiteAccess> {
    return siteAccessFor(getSiteDefinition(siteKey) ?? null, await this.siteDefaults());
  }

  /** 已建立的站点限流器(测试与诊断用) */
  limiterOf(siteId: string): SiteRateLimiter | null {
    return this.adapters.get(siteId)?.limiter ?? null;
  }

  /**
   * 站点凭证:取该站第一个启用且未失效的账号;明文不缓存。
   * cookie:Cookie 型账号的 secret,或账密账号登录得到的会话 Cookie;apiKey:API Key 型账号;
   * passkey:账号附挂的 passkey 或 passkey 型账号(RSS / 下载用,不看登录状态)。
   */
  private siteSecrets(siteId: string): SecretResolver {
    const { db, secrets } = this.ctx;
    return {
      async resolve({ kind }) {
        let q = db
          .selectFrom("site_account")
          .select(["auth_kind", "secret_ref", "session_secret_ref", "passkey_secret_ref"])
          .where("site_id", "=", siteId)
          .where("enabled", "=", true)
          .orderBy("created_at");
        if (kind !== "passkey") q = q.where("status", "in", ["unknown", "active"]);
        const rows = await q.execute();
        for (const r of rows) {
          const ref =
            kind === "cookie"
              ? r.auth_kind === "cookie"
                ? r.secret_ref
                : r.auth_kind === "password"
                  ? r.session_secret_ref
                  : null
              : kind === "apiKey"
                ? r.auth_kind === "api_key"
                  ? r.secret_ref
                  : null
                : (r.passkey_secret_ref ?? (r.auth_kind === "passkey" ? r.secret_ref : null));
          if (ref) return secrets.resolve(ref);
        }
        return null;
      },
    };
  }
}

/** TMDB 请求响应缓存:PG kv_cache region=tmdb.request;PG 出错按未命中 / 不写处理,不让 TMDB 请求失败 */
function tmdbSharedCache(ctx: ServerContext, policy: TmdbCachePolicy): TmdbSharedCache {
  const cache = sharedCache(ctx);
  const warn = (err: unknown) =>
    ctx.logger.warn({ err: { message: (err as Error).message } }, "TMDB 缓存读写失败，按未命中处理");
  return {
    get: (key) =>
      cache.get(CACHE_REGIONS.tmdbRequest, key).catch((e) => {
        warn(e);
        return null;
      }),
    set: (key, value, ttlMs) =>
      cache.set(CACHE_REGIONS.tmdbRequest, key, value, { ttl: ttlMs / 1000, maxsize: policy.maxsize }).catch(warn),
  };
}

function sameUrl(a: string, b: string): boolean {
  return a.replace(/\/+$/, "") === b.replace(/\/+$/, "");
}

/** 站点访问默认(已合并定义与设置) */
export interface SiteAccess {
  /** 站点行没设快照周期时用 */
  snapshotIntervalSec: number;
  snapshotMaxPages: number;
  healthCheckIntervalSec: number;
  detailCacheSec: number;
  searchConcurrency: number;
}

type DefinitionDefaults = {
  snapshotIntervalSec?: number;
  snapshotMaxPages?: number;
  healthCheckIntervalSec?: number;
  detailCacheSec?: number;
  searchConcurrency?: number;
};

/** 周期 / 缓存取较长,页数 / 并发取较小(定义值是按站点实测给的保守下限) */
export function siteAccessFor(def: SiteDefinition | null, s: SitesSettings): SiteAccess {
  const d = ((def?.umt as { defaults?: DefinitionDefaults } | undefined)?.defaults ?? {}) as DefinitionDefaults;
  const max = (a: number | undefined, b: number) => (a === undefined ? b : Math.max(a, b));
  const min = (a: number | undefined, b: number) => (a === undefined ? b : Math.min(a, b));
  return {
    snapshotIntervalSec: max(d.snapshotIntervalSec, s.defaultSnapshotIntervalSec),
    snapshotMaxPages: min(d.snapshotMaxPages, s.defaultSnapshotMaxPages),
    healthCheckIntervalSec: max(d.healthCheckIntervalSec, s.defaultHealthCheckIntervalSec),
    detailCacheSec: max(d.detailCacheSec, s.defaultDetailCacheSec),
    searchConcurrency: d.searchConcurrency ?? 1,
  };
}

/** 同站实时搜索并发(只看站点定义,同步取值;缺省 1) */
export function siteSearchConcurrency(siteKey: string): number {
  const def = getSiteDefinition(siteKey);
  return ((def?.umt as { defaults?: DefinitionDefaults } | undefined)?.defaults?.searchConcurrency ?? 1) as number;
}

/** 包一层限流器:每发放一个令牌回调一次(计数) */
function countingLimiter(inner: SiteRateLimiter, onGrant: () => void): SiteRateLimiter {
  return {
    get stats() {
      return inner.stats;
    },
    async acquire(signal) {
      await inner.acquire(signal);
      onGrant();
    },
    tryAcquire() {
      const ok = inner.tryAcquire();
      if (ok) onGrant();
      return ok;
    },
    penalize: (ms) => inner.penalize(ms),
  };
}

type RateLimitLike = { requestsPerMinute?: number; minIntervalMs?: number; concurrency?: number };

/**
 * 站点行 rate_limit 优先;为空时用全局默认(settings.sites.defaultRateLimit),但不比定义里的 umt.rateLimit 更激进
 * (定义值是按站点实测调的下限);都没有时用定义值。
 */
export function limiterOptions(
  def: SiteDefinition,
  rateLimit: unknown,
  globalDefault?: RateLimitLike | null,
): { minIntervalMs: number; burst: number } {
  const own = (rateLimit ?? {}) as RateLimitLike;
  const hasOwn =
    own.minIntervalMs !== undefined || own.requestsPerMinute !== undefined || own.concurrency !== undefined;
  const interval = (rl: RateLimitLike) =>
    rl.minIntervalMs ?? (rl.requestsPerMinute ? Math.ceil(60_000 / rl.requestsPerMinute) : undefined);
  if (hasOwn) {
    return {
      minIntervalMs: interval(own) ?? def.umt.rateLimit.minIntervalMs,
      burst: own.concurrency ?? def.umt.rateLimit.burst,
    };
  }
  const g = globalDefault ?? {};
  const gi = interval(g);
  return {
    minIntervalMs: gi === undefined ? def.umt.rateLimit.minIntervalMs : Math.max(gi, def.umt.rateLimit.minIntervalMs),
    burst: g.concurrency === undefined ? def.umt.rateLimit.burst : Math.min(g.concurrency, def.umt.rateLimit.burst),
  };
}