406 lines
16 KiB
TypeScript
406 lines
16 KiB
TypeScript
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,
|
|
renewTrackedTmdbDetail,
|
|
sharedCache,
|
|
type TmdbCachePolicy,
|
|
tmdbCachePolicy,
|
|
tmdbDetailRef,
|
|
} 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,
|
|
searchTtlMs: policy.searchTtl * 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 请求失败。
|
|
* 详情命中且属于追踪中的作品时续期(CACHE-2),续期失败只记 debug。
|
|
*/
|
|
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: async (key) => {
|
|
const v = await cache.get(CACHE_REGIONS.tmdbRequest, key).catch((e) => {
|
|
warn(e);
|
|
return null;
|
|
});
|
|
if (v !== null && tmdbDetailRef(key)) {
|
|
await renewTrackedTmdbDetail(ctx, key, policy.ttl).catch((e) =>
|
|
ctx.logger.debug({ err: { message: (e as Error).message } }, "TMDB 详情缓存续期失败"),
|
|
);
|
|
}
|
|
return v;
|
|
},
|
|
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),
|
|
};
|
|
}
|