import { createRssFetcher, createSharedSiteRateLimiter, createSiteAdapter, createSiteRateLimiter, createTmdbClient, getSiteDefinition, type RssFetcher, type SecretResolver, type SiteAdapter, type SiteDefinition, type SiteRateLimiter, type TmdbClient, type TmdbRequestKind, type TmdbSharedCache, } from "@umt/adapters"; import { DEFAULT_SETTINGS, type SitesSettings } from "@umt/contracts"; import { pgTokenBucketFor, type SiteRow, sql } from "@umt/db"; import { egressKey, externalFetch, siteEgress, siteFetchFor } from "../../egress.ts"; import { CACHE_REGIONS, metadataCache, type TmdbCachePolicy, tmdbCachePolicy, watchedPin } from "../../shared-cache.ts"; import type { ServerContext } from "../types.ts"; import { CloudflareGate, withClearance } from "./service/cloudflare.ts"; import { failureSampleSink } from "./service/failure-sample.ts"; import { readBrowserUserAgent, resolveUserAgent, type UserAgentDefaults } from "./service/user-agent.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[0]; /** 调度 tick 间隔(毫秒),默认 30s */ schedulerIntervalMs?: number; /** false = 站点限流只用进程内令牌桶(缺省走 PG 跨进程令牌桶 rate_limit_bucket) */ sharedRateLimit?: boolean; /** 豆瓣 / bgm.tv 请求用的 fetch;null = 不在线查(测试缺省不出外网) */ metaFetch?: typeof globalThis.fetch | null; } interface CachedAdapter { sig: string; limiter: SiteRateLimiter; adapter: SiteAdapter; } export class SourcesRuntime { readonly now: () => Date; readonly rss: RssFetcher; private readonly adapters = new Map(); private tmdbClient: TmdbClient | null | undefined; /** 站点默认(settings.sites)缓存:siteAdapter 是同步的,读设置的地方顺手刷新 */ private siteDefaultsCache: SitesSettings = DEFAULT_SETTINGS.sites; private siteDefaultsAt = 0; private tmdbSig: string | null = null; /** CF-2 Cloudflare 挑战处理(站点请求经 siteFetch 进来) */ readonly cloudflare: CloudflareGate; constructor( readonly ctx: ServerContext, readonly overrides: RuntimeOverrides = {}, ) { this.now = overrides.now ?? (() => new Date()); this.cloudflare = new CloudflareGate(this); 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。设置变化(引用/地址)时重建。 */ /** 豆瓣 / bgm.tv 请求:注入优先;测试进程(VITEST)缺省不出外网;否则按环境变量代理出外网 */ metaFetch(): typeof globalThis.fetch | null { if (this.overrides.metaFetch !== undefined) return this.overrides.metaFetch; return Deno.env.get("VITEST") ? null : externalFetch; } async tmdb(): Promise { if (this.overrides.tmdb) { if (this.tmdbClient === undefined) this.tmdbClient = this.overrides.tmdb(); return this.tmdbClient; } const { sig, cfg } = await this.tmdbConfig(); if (this.tmdbClient !== undefined && this.tmdbSig === sig) return this.tmdbClient; this.tmdbSig = sig; this.tmdbClient = cfg ? createTmdbClient(cfg) : null; return this.tmdbClient; } /** * 批量任务用的独立 TMDB 客户端(FIX-5 元数据回刷):串行、网络请求间隔 ≥ minIntervalMs(缓存命中不占额度), * 同样走 PG 请求缓存;requests() 返回实际发出的网络请求数。测试注入的客户端原样返回(requests 为 null)。 */ async tmdbForBatch(opts: { minIntervalMs: number; }): Promise<{ client: TmdbClient; requests: () => number | null } | null> { if (this.overrides.tmdb) { const client = await this.tmdb(); return client ? { client, requests: () => null } : null; } const { cfg } = await this.tmdbConfig(); if (!cfg) return null; let n = 0; const base = cfg.fetch ?? externalFetch; const counted: typeof fetch = (input, init) => { n++; return base(input, init); }; const client = createTmdbClient({ ...cfg, fetch: counted, maxConcurrent: 1, minIntervalMs: opts.minIntervalMs }); return { client, requests: () => n }; } /** TMDB 客户端配置(未配置凭证时 cfg 为 null)与签名(设置变化时重建客户端) */ private async tmdbConfig(): Promise<{ sig: string; cfg: Parameters[0] | null }> { 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")], ]); 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"); } const cfg = apiKey || accessToken ? { ...(apiKey ? { apiKey } : {}), ...(accessToken ? { accessToken } : {}), ...(baseUrl ? { baseUrl } : {}), // 外网请求:显式按环境变量走代理(CF-2 出口表;测试注入优先) fetch: this.overrides.fetch ?? externalFetch, // 请求响应进 PG 缓存(region tmdb.request),三进程共享;TTL / 条数上限 / 空结果短 TTL 取设置(对照 MP) cache: { ttlMs: policy.ttl * 1000, searchTtlMs: policy.searchTtl * 1000, emptyTtlMs: policy.emptyTtl * 1000, listTtlMs: policy.listTtl * 1000, imageTtlMs: policy.imageTtl * 1000, }, sharedCache: tmdbSharedCache(this.ctx, policy), } : null; return { sig, cfg }; } /** 站点默认(settings.sites),读库并刷新缓存;读失败用上次值 */ async siteDefaults(): Promise { try { this.siteDefaultsCache = { ...DEFAULT_SETTINGS.sites, ...(await this.ctx.settings.get("sites")) }; } catch { // 保留缓存 } return this.siteDefaultsCache; } /** * 站点请求用的 fetch:测试注入优先;站点开了"使用代理"且设置了 proxyUrl 时走代理,否则**明确直连** * (CF-2:不再隐式继承容器的 HTTP(S)_PROXY,见 egress.ts)。 * BR-4:每条请求都带 UA(请求自己设了就不动;否则 ua 参数,缺省按"跟随浏览器"),不再发 Deno 缺省的 `Deno/x.y.z`。 * ua 可以是函数:每次请求现场取(站点适配器按当前账号取,账号切换 UA 后立即生效)。 * CF-2:给了 site.id 的请求经 Cloudflare 处理(浏览器代理期直接走浏览器、带可复用的 cf_clearance、遇挑战过挑战 / 重试 / 浏览器代抓)。 */ siteFetch( site: Pick & Partial>, defaults: SitesSettings = this.siteDefaultsCache, ua?: string | (() => Promise), ): typeof globalThis.fetch { const base = this.transportFetch(site, defaults); const egress = this.egressKeyOf(site, defaults); const siteId = site.id; return (async (input: RequestInfo | URL, init?: RequestInit) => { const headers = new Headers(init?.headers ?? (input instanceof Request ? input.headers : undefined)); if (!headers.has("user-agent")) { const pick = typeof ua === "function" ? ua() : ua !== undefined ? Promise.resolve(ua) : this.defaultUserAgent(); headers.set("user-agent", await pick); } if (!siteId) return base(input, { ...init, headers }); const url = input instanceof Request ? input.url : String(input); const method = init?.method ?? (input instanceof Request ? input.method : "GET"); return this.cloudflare.handle( siteId, { url, method, accept: headers.get("accept"), userAgent: headers.get("user-agent"), egress }, (extra) => { const h = new Headers(headers); const cookie = withClearance(h.get("cookie"), extra); if (cookie) h.set("cookie", cookie); // 过挑战可能超过单次请求的超时:重试时原信号已中止就换一个新的超时(调用方取消照样生效在级联之外) const signal = init?.signal?.aborted ? AbortSignal.timeout(30_000) : init?.signal; return base(input, { ...init, headers: h, ...(signal ? { signal } : {}) }); }, ); }) as typeof globalThis.fetch; } /** 站点 HTTP 出口键(direct / proxy:<主机>),与浏览器会话的出口同一口径 */ egressKeyOf(site: Pick, defaults: SitesSettings = this.siteDefaultsCache): string { return egressKey(siteEgress(site.proxy, defaults.proxyUrl)); } /** 不加 UA 的底层 fetch(测试注入 / 按站点出口:代理或明确直连) */ private transportFetch(site: Pick, defaults: SitesSettings): typeof globalThis.fetch { if (this.overrides.fetch) return this.overrides.fetch; return siteFetchFor(siteEgress(site.proxy, defaults.proxyUrl)); } // ---------- BR-4 站点请求 UA ---------- private browserUa: { at: number; ua: string | null } = { at: 0, ua: null }; /** 浏览器固定身份的 UA(60 秒缓存);读不到为 null(调用方退回常量) */ async browserUserAgent(): Promise { if (Date.now() - this.browserUa.at < 60_000) return this.browserUa.ua; let ua: string | null = this.browserUa.ua; try { ua = (await readBrowserUserAgent(this.ctx)) ?? ua; } catch (err) { this.ctx.logger.debug({ err: { message: (err as Error).message } }, "读取浏览器身份 UA 失败,沿用上次的值"); } this.browserUa = { at: Date.now(), ua }; return ua; } /** UA 解析用的全局部分:设置里的全局 UA + 浏览器 UA(已退回常量) */ async uaDefaults(): Promise { const [s, browser] = await Promise.all([this.siteDefaults(), this.browserUserAgent()]); return { global: s.userAgent ?? "", browser: browser ?? "" }; } /** 不挂账号的请求(通用 RSS 等)用的 UA:全局 UA 或跟随浏览器 */ async defaultUserAgent(): Promise { return resolveUserAgent(null, await this.uaDefaults()).ua; } /** * 站点请求 UA:取与站点凭证相同的账号(第一个启用、未失效、非 Passkey 的账号;没有就第一个启用的),按账号解析。 * 不缓存:与 Cookie 同一时刻读库,账号切换 Cookie + UA 后下一条请求两者一起生效。 */ async siteUserAgent(siteId: string): Promise { const d = await this.uaDefaults(); const rows = await this.ctx.db .selectFrom("site_account") .select(["user_agent", "ua_pending", "status", "auth_kind"]) .where("site_id", "=", siteId) .where("enabled", "=", true) .orderBy("created_at") .execute(); const acc = rows.find((r) => r.auth_kind !== "passkey" && (r.status === "unknown" || r.status === "active")) ?? rows.find((r) => r.auth_kind !== "passkey") ?? rows[0] ?? null; return resolveUserAgent(acc, d).ua; } /** * 站点适配器:每站一个实例,列表/搜索/详情/下载共享同一个限流器。 * 站点限流参数、基址、代理或站点默认变化时重建(签名比较)。 */ siteAdapter( site: Pick & Partial>, ): 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; // Cloudflare 处理状态(config.cloudflare)经常变,不参与重建判断 const { cloudflare: _cf, ...configSig } = (site.config ?? {}) as Record; const sig = JSON.stringify([ site.base_url, site.rate_limit, configSig, 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:);进程内先串行排队 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, // 每条请求都经 siteFetch:代理 + 按账号现场取 UA(BR-4) fetch: this.siteFetch({ id: site.id, proxy }, defaults, () => this.siteUserAgent(site.id)), ...(baseUrl ? { baseUrl } : {}), // parse_error / challenge 的脱敏页面样本存进 site_failure_sample(每站最近 5 份) onFailureSample: failureSampleSink(this.ctx, site.id), }); 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 { 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 { 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 请求响应缓存:统一元数据缓存 region=tmdb.request(L1 进程内存 + L2 kv_cache,过期先用旧值后台刷新、同请求合并)。 * TTL 按请求类别取策略表(搜索 / 详情 / 列表 / 图片);关注中作品的条目写入 pinned、命中续期。 * PG 出错按未命中 / 不写处理,不让 TMDB 请求失败。get / set 保留给不走 fetch 的旧调用方。 */ export function tmdbSharedCache(ctx: ServerContext, policy: TmdbCachePolicy): TmdbSharedCache { const cache = metadataCache(ctx); const region = CACHE_REGIONS.tmdbRequest; const ttlOf = (kind: TmdbRequestKind) => kind === "search" ? { ttl: policy.searchTtl, emptyTtl: policy.emptyTtl } : kind === "list" ? { ttl: policy.listTtl, emptyTtl: policy.listEmptyTtl } : kind === "image" ? { ttl: policy.imageTtl, emptyTtl: policy.emptyTtl } : { ttl: policy.ttl, emptyTtl: policy.emptyTtl }; return { fetch: (req, load) => cache.fetch( region, req.key, { ...ttlOf(req.kind), isEmpty: req.isEmpty, maxsize: policy.maxsize, pin: watchedPin(ctx, req.key) }, load, ), get: (key) => cache.get(region, key), set: (key, value, ttlMs) => cache .set(region, key, value, { ttl: ttlMs / 1000, maxsize: policy.maxsize }) .catch((e) => ctx.logger.warn({ err: { message: (e as Error).message } }, "TMDB 缓存读写失败,按未命中处理")), }; } 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), }; }