feat(browser): unify site credentials with browser profiles (BR-2)

- Inject the site account Cookie (manual + auto-login session, session wins)
  into the camofox profile before the first tab opens; password accounts
  without a session run refresh-login first; API Key sites are skipped.
- Cookie expired after injection -> credential.status=expired; cookie
  accounts go to waiting/login_required ("log in once, applies site-wide").
- Write back site-domain cookies from the browser on handover, session close,
  every 10 min of activity, and from refresh-login when the stored cookie is
  invalid (verified before saving). user_info.cookieOrigin records
  source=browser; source shown as manual/login/browser in the origin view.
- New camofox plugin umt-cookies (GET/POST /umt/sessions/:userId/cookies
  under the access key); profile dir 0700/files 0600 in the desktop image.
- Single entry: sources-identity/service/credentials.ts.
- Real check: HHClub cookie injected into a fresh profile -> home page
  logged in (logout link present), 5 cookies read back.

Requires rebuilding umt-browser-desktop (docker compose build camofox &&
docker compose up -d camofox) to take effect.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed 2026-09-29 15:25:07 +08:00
1 parent 901b01d1a1
commit 28974d1e1a
58 files changed
+2843 -1112

No files matched your search

@@ -0,0 +1,203 @@
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { readAccountCookie } from "../../sources-identity/service/credentials.ts";
import { type BrowserHarness, createBrowserHarness, type Loose, ORIGIN, pgAvailable } from "./harness.ts";
// BR-2 凭据统一(对 testkit mock camofox + mock NexusPHP 登录态判定):
// 打开会话注入站点账号 Cookie → 首页 / 受保护页是登录态;Cookie 过期 → 需要人工登录;
// 人工在画面里登录后交还 → 读回站点域名 Cookie 写进账号(来源 browser);关闭时也回写;没有插件时降级说明。
const up = await pgAvailable();
const GOOD = "good-session-value-7f3a";
const STALE = "stale-session-value-11";
const FRESH = "fresh-browser-login-9c2";
describe.skipIf(!up)("BR-2 站点账号 Cookie ⇄ 浏览器 profile", () => {
let h: BrowserHarness;
let cookieAcc: string;
const setCookieAccount = async (cookie: string) => {
const ref = await h.secrets.create({ purpose: "site_cookie", plaintext: cookie });
await h.db
.updateTable("site_account")
.set({ auth_kind: "cookie", secret_ref: ref, session_secret_ref: null, status: "active", user_info: "{}" })
.where("id", "=", cookieAcc)
.execute();
};
const open = async (path = "/torrents.php") =>
(await h.req("/browser/sessions", { body: { siteId: h.siteId, accountId: cookieAcc, url: `${ORIGIN}${path}` } }))
.body.data as Loose;
const closeAll = async () => {
for (const s of (await h.req("/browser/sessions")).body.data.items as Loose[])
await h.req(`/browser/sessions/${s.id}`, { method: "DELETE" });
};
beforeAll(async () => {
h = await createBrowserHarness("cred", {
sessionCookie: { name: "c_secure_pass", valid: [GOOD], fresh: FRESH },
});
cookieAcc = h.accountId;
}, 120_000);
afterAll(async () => {
await h?.close();
});
it("注入:Cookie 账号打开会话前写进该 profile,受保护页直接是登录态;响应里没有 Cookie 值", async () => {
await setCookieAccount(`c_secure_uid=1001; c_secure_pass=${GOOD}`);
const s = await open();
expect(s.controller).toBe("automation");
expect(s.currentUrl).toBe(`${ORIGIN}/torrents.php`);
expect(s.credential).toMatchObject({ status: "injected", source: "manual", count: 2 });
expect(JSON.stringify(s)).not.toContain(GOOD);
// 注入走 umt-cookies 插件,早于打开标签页;Cookie 按站点主机 host-only、https → secure
const calls = h.mock.calls.map((c) => `${c.method} ${c.path}`);
const inj = calls.indexOf(`POST /umt/sessions/${encodeURIComponent(s.profileKey)}/cookies`);
expect(inj).toBeGreaterThanOrEqual(0);
expect(inj).toBeLessThan(calls.lastIndexOf("POST /tabs"));
const jar = h.mock.cookieJars.get(s.profileKey)!;
expect(jar.find((c) => c.name === "c_secure_pass")).toMatchObject({
value: GOOD,
domain: "audiences.example",
secure: true,
});
await closeAll();
});
it("过期:注入后仍落在登录页 → 需要人工登录;人工登录后交还 → 回写账号 Cookie(来源浏览器),下次抓取用新值", async () => {
h.mock.loggedIn.clear();
h.mock.cookieJars.clear();
await setCookieAccount(`c_secure_uid=1001; c_secure_pass=${STALE}`);
const s = await open();
expect(s.controller).toBe("waiting");
expect(s.waitingReason).toBe("login_required");
expect(s.credential).toMatchObject({ status: "expired" });
expect(s.credential.message).toContain("登录一次");
const d = await h.db.selectFrom("decision").selectAll().where("subject_id", "=", s.id).executeTakeFirstOrThrow();
expect(d.summary).toContain("交还后全站生效");
// 人工接管,在画面里登录(桌面模式里的导航不经 API:直接改 mock 的登录态与当前地址)
expect((await h.req(`/browser/sessions/${s.id}/control/acquire`, { body: { reason: "登录" } })).status).toBe(200);
const tab = [...h.mock.tabs.values()].find((t) => t.userId === s.profileKey)!;
h.mock.loggedIn.add(s.profileKey);
h.mock.cookieJars.get(s.profileKey)!.push(
{
name: "c_secure_pass",
value: FRESH,
domain: "audiences.example",
path: "/",
expires: -1,
httpOnly: true,
secure: true,
},
// 别的域名的 Cookie 不回写
{
name: "_ga",
value: "tracker",
domain: ".analytics.example",
path: "/",
expires: -1,
httpOnly: false,
secure: true,
},
);
tab.url = `${ORIGIN}/torrents.php`;
const r = await h.req(`/browser/sessions/${s.id}/control/release`, { body: {} });
expect(r.status).toBe(200);
const acc = await readAccountCookie(h.ctx, cookieAcc);
expect(acc).toMatchObject({ source: "browser", applicable: true });
expect(acc!.header).toContain(`c_secure_pass=${FRESH}`);
expect(acc!.header).toContain("c_secure_uid=1001");
expect(acc!.header).not.toContain("_ga");
const row = await h.db.selectFrom("site_account").selectAll().where("id", "=", cookieAcc).executeTakeFirstOrThrow();
expect((row.user_info as Loose).cookieOrigin).toMatchObject({ source: "browser", by: "browser" });
const view = (await h.req(`/browser/sessions/${s.id}`)).body.data;
expect(view.credential).toMatchObject({ status: "injected", source: "browser" });
expect(view.credential.syncedAt).toBeTruthy();
expect(JSON.stringify(view)).not.toContain(FRESH);
await closeAll();
});
it("关闭会话时回写;停在登录页不回写(未登录的 Cookie 不覆盖存储)", async () => {
await setCookieAccount(`c_secure_uid=1001; c_secure_pass=${GOOD}`);
h.mock.cookieJars.clear();
const s = await open();
expect(s.credential.status).toBe("injected");
// 站点在浏览器里轮换了会话 Cookie
const jar = h.mock.cookieJars.get(s.profileKey)!;
jar.find((c) => c.name === "c_secure_pass")!.value = "rotated-by-site-44";
await h.req(`/browser/sessions/${s.id}`, { method: "DELETE" });
expect((await readAccountCookie(h.ctx, cookieAcc))!.header).toContain("c_secure_pass=rotated-by-site-44");
// 登录页:不回写
await setCookieAccount(`c_secure_uid=1001; c_secure_pass=${GOOD}`);
const s2 = await open("/login.php");
h.mock.cookieJars.get(s2.profileKey)!.find((c) => c.name === "c_secure_pass")!.value = "anonymous-junk";
await h.req(`/browser/sessions/${s2.id}`, { method: "DELETE" });
expect((await readAccountCookie(h.ctx, cookieAcc))!.header).toContain(`c_secure_pass=${GOOD}`);
});
it("有活动时每 10 分钟读回一次", async () => {
await setCookieAccount(`c_secure_uid=1001; c_secure_pass=${GOOD}`);
h.mock.cookieJars.clear();
const s = await open();
h.mock.cookieJars.get(s.profileKey)!.find((c) => c.name === "c_secure_pass")!.value = "periodic-rotated-5";
// 刚打开:不读回
await h.req(`/browser/sessions/${s.id}/snapshot`);
expect((await readAccountCookie(h.ctx, cookieAcc))!.header).toContain(`c_secure_pass=${GOOD}`);
// 会话创建时间拨到 11 分钟前:下一次活动触发后台读回
await h.db
.updateTable("browser_session")
.set({ created_at: new Date(Date.now() - 11 * 60_000) })
.where("id", "=", s.id)
.execute();
await h.req(`/browser/sessions/${s.id}/snapshot`);
await expect
.poll(async () => (await readAccountCookie(h.ctx, cookieAcc))!.header, { timeout: 3000 })
.toContain("c_secure_pass=periodic-rotated-5");
await closeAll();
});
it("匿名会话与 API Key 账号不注入", async () => {
const r = await h.req("/browser/sessions", { body: { siteId: h.siteId, accountId: null, url: `${ORIGIN}/` } });
expect(r.body.data.credential).toMatchObject({ status: "not_applicable" });
await closeAll();
const ref = await h.secrets.create({ purpose: "site_api_key", plaintext: "api-key-xyz" });
await h.db
.updateTable("site_account")
.set({ auth_kind: "api_key", secret_ref: ref })
.where("id", "=", cookieAcc)
.execute();
const before = h.mock.callsTo(/\/cookies$/).length;
const s = await open("/");
expect(s.credential).toMatchObject({ status: "not_applicable" });
expect(h.mock.callsTo(/\/cookies$/).length).toBe(before);
await closeAll();
});
});
describe.skipIf(!up)("BR-2 浏览器服务没有 umt-cookies 插件", () => {
let h: BrowserHarness;
beforeAll(async () => {
h = await createBrowserHarness("cred_noplugin", {
sessionCookie: { name: "c_secure_pass", valid: [GOOD], fresh: FRESH },
cookieApi: "none",
});
}, 120_000);
afterAll(async () => {
await h?.close();
});
it("注入报不支持,会话照常打开并说明原因", async () => {
const ref = await h.secrets.create({ purpose: "site_cookie", plaintext: `c_secure_pass=${GOOD}` });
await h.db
.updateTable("site_account")
.set({ auth_kind: "cookie", secret_ref: ref })
.where("id", "=", h.accountId)
.execute();
const r = await h.req("/browser/sessions", { body: { siteId: h.siteId, url: `${ORIGIN}/` } });
expect(r.status).toBe(200);
expect(r.body.data.credential).toMatchObject({ status: "unsupported", count: 0 });
expect(r.body.data.credential.message).toContain("umt-cookies");
});
});
@@ -53,11 +53,23 @@ export interface BrowserHarness {
close(): Promise<void>;
}
export async function createBrowserHarness(name: string, opts: { captcha?: boolean } = {}): Promise<BrowserHarness> {
export async function createBrowserHarness(
name: string,
opts: {
captcha?: boolean;
/** BR-2:mock 站点按这个 Cookie 判定登录态 */
sessionCookie?: { name: string; valid: string[]; fresh: string };
cookieApi?: "plugin" | "upstream" | "none";
cookieImportKey?: string;
} = {},
): Promise<BrowserHarness> {
const torrent = buildFakeTorrent({ name: "Oppenheimer.2023.2160p.UHD.BluRay.x265-GRP" });
const mock = await createMockCamofox({
accessKey: ACCESS_KEY,
...(opts.cookieApi ? { cookieApi: opts.cookieApi } : {}),
...(opts.cookieImportKey ? { cookieImportKey: opts.cookieImportKey } : {}),
site: {
...(opts.sessionCookie ? { sessionCookie: opts.sessionCookie } : {}),
origin: ORIGIN,
username: "fixture-user",
password: PASSWORD,
+4 -1
View File
@@ -40,7 +40,10 @@ export function createBrowserModule(opts: BrowserModuleOptions = {}): ServerModu
registerRoutes(app, ctx) {
registerBrowserRoutes(app, () => service(ctx));
},
registerTasks() {},
registerTasks(_registry, ctx) {
// worker 里的 refresh-login 也要能从浏览器 profile 读回 Cookie(BR-2):建服务即注册读回钩子
service(ctx);
},
registerCapabilities(registry, ctx) {
service(ctx);
registerBrowserCapabilities(registry, () => service(ctx));
+203 -2
View File
@@ -5,9 +5,12 @@ import {
type BrowserProvider,
type BrowserSecretResolver,
CAMOFOX_VERIFIED_RELEASES,
cookiesFromHeader,
createCamofoxBrowserProvider,
type DownloadSink,
type ElementTarget,
getSiteDefinition,
headerFromCookies,
hostAllowed,
imageDimensions,
type PageMetrics,
@@ -22,7 +25,9 @@ import type {
BrowserCapabilities,
BrowserControlLease,
BrowserController,
BrowserCookie,
BrowserCoordinateSpace,
BrowserCredentialState,
BrowserDownloadVerification,
BrowserEvidenceRef,
BrowserHealth,
@@ -42,6 +47,12 @@ import {
sql,
} from "@umt/db";
import type { SourcesRuntime } from "../sources-identity/runtime.ts";
import { refreshSiteAccount } from "../sources-identity/service/account.ts";
import {
readAccountCookie,
registerBrowserCookieReader,
writeBrowserCookie,
} from "../sources-identity/service/credentials.ts";
import type { ServerContext } from "../types.ts";
// C1-E 浏览器服务层:BrowserProvider(camofox)之上的会话/标签/控制权持久化(数据库为准)、
@@ -134,8 +145,13 @@ export interface RecoveryState {
waitingDecisionId?: string | null;
releasedAt?: string;
downloads?: { sha256: string; artifactId: string; sourceReleaseId: string | null; planId: string | null }[];
/** BR-2:打开时的凭据注入结果与最近一次回写检查 */
credential?: BrowserCredentialState;
}
/** 有活动的会话每隔多久从浏览器读回一次 Cookie */
export const COOKIE_SYNC_INTERVAL_MS = 10 * 60_000;
/** 环境变量(BROWSER_PROVIDER_URL)登记的浏览器服务 */
export const ENV_PROVIDER_NAME = "camofox-env";
const ENV_MANAGED = "env";
@@ -150,6 +166,8 @@ export class BrowserService {
readonly opts: BrowserModuleOptions = {},
) {
this.now = opts.now ?? (() => new Date());
// refresh-login 发现 Cookie 失效时从浏览器 profile 读回一次(sources-identity 经钩子调用,不直接依赖本模块)
registerBrowserCookieReader(ctx, (siteId, accountId) => this.readProfileCookie(siteId, accountId));
}
// ---------- 浏览器服务 ----------
@@ -622,6 +640,7 @@ export class BrowserService {
lease: lease ? toLease(lease) : null,
needsSnapshot: rs.needsSnapshot === true,
waitingReason: row.controller === "waiting" || row.controller === "human" ? (rs.waitingReason ?? null) : null,
credential: rs.credential ?? null,
};
}
@@ -680,6 +699,7 @@ export class BrowserService {
// 数据库里已没有未关闭的会话:本进程内存里同 profile 还开着的是别的进程关掉的,标关闭,否则适配器拒绝新开
for (const m of await provider.listSessions())
if (m.profileKey === profileKey && m.controller !== "closed") provider.setController(m.id, "closed");
const prep = await this.prepareCookies(site, accountId);
await this.rateLimit(site);
let res: Awaited<ReturnType<BrowserProvider["openSession"]>>;
try {
@@ -690,11 +710,28 @@ export class BrowserService {
profileKey,
initialUrl: req.url ?? site.base_url,
budget,
...(prep.cookies.length ? { cookies: prep.cookies } : {}),
});
} catch (e) {
throw fromBrowserError(e);
}
const s = res.session;
const credential: BrowserCredentialState = { ...prep.state };
if (res.cookies) {
credential.status = res.cookies.status;
credential.count = res.cookies.count;
if (res.cookies.status === "unsupported")
credential.message = "浏览器服务没有 Cookie 接口(镜像未带 umt-cookies 插件),没有注入账号登录态";
else if (res.cookies.status === "failed") credential.message = "注入账号 Cookie 失败";
}
// 注入了 Cookie 仍落在登录页:Cookie 过期。Cookie 账号(没有密码可自动登录)转"需要人工登录":在画面里登录一次,
// 交还 / 关闭时回写,全站生效;账密账号留给 Agent 用凭证注入登录(browser.interact credential),只记状态
const loginPage = provider.sessionInfo(s.id)?.loginPage === true;
const needHuman = credential.status === "injected" && loginPage && prep.authKind === "cookie";
if (credential.status === "injected" && loginPage) {
credential.status = "expired";
credential.message = "账号登录态已失效:接管后在画面里登录一次,交还后 HTTP 抓取、RSS、快照都用新的登录态";
}
await this.ctx.db.transaction().execute(async (trx) => {
await trx
.insertInto("browser_session")
@@ -708,7 +745,7 @@ export class BrowserService {
current_url: s.currentUrl,
last_activity_at: new Date(s.lastActivityAt),
upstream_ref: JSON.stringify(s.upstreamRef),
recovery_state: JSON.stringify({ currentTabId: res.tab.upstreamTabId, needsSnapshot: true }),
recovery_state: JSON.stringify({ currentTabId: res.tab.upstreamTabId, needsSnapshot: true, credential }),
})
.execute();
await trx
@@ -722,11 +759,160 @@ export class BrowserService {
})
.execute();
});
if (needHuman) await this.enterWaiting(await this.sessionRow(s.id), "login_required");
return s.id;
}
// ---------- 凭据统一(BR-2) ----------
/**
* 打开会话前准备要注入的 Cookie:匿名会话与 API Key 类站点(馒头:网页登录态与 API Key 无关)不注入;
* 账密账号还没有会话 Cookie 时先走一次 refresh-login(与站点中心"刷新登录"同一流程)再取。
*/
private async prepareCookies(
site: SiteRow,
accountId: string | null,
): Promise<{ cookies: BrowserCookie[]; state: BrowserCredentialState; authKind: string | null }> {
const state = (
status: BrowserCredentialState["status"],
message: string | null,
extra: Partial<BrowserCredentialState> = {},
): BrowserCredentialState => ({
status,
source: null,
sourceAt: null,
count: 0,
message,
syncedAt: null,
...extra,
});
if (!accountId)
return { cookies: [], authKind: null, state: state("not_applicable", "匿名会话,不注入账号登录态") };
const kind = (getSiteDefinition(site.key)?.umt as { kind?: string } | undefined)?.kind;
let acc = await readAccountCookie(this.ctx, accountId);
if (!acc?.applicable || kind === "mteam")
return {
cookies: [],
authKind: acc?.authKind ?? null,
state: state("not_applicable", "API Key 类站点 / 账号:网页登录态与 API Key 无关,不注入;需要时在桌面里登录"),
};
const rt = this.opts.runtime?.(this.ctx);
if (!acc.header && acc.authKind === "password" && rt) {
await refreshSiteAccount(rt, accountId, { mode: "login" }).catch(() => null);
acc = (await readAccountCookie(this.ctx, accountId)) ?? acc;
}
if (!acc.header)
return { cookies: [], authKind: acc.authKind, state: state("none", "账号还没有可用的登录态(Cookie)") };
const cookies = cookiesFromHeader(acc.header, site.base_url, Math.floor(this.now().getTime() / 1000));
return {
cookies,
authKind: acc.authKind,
state: state("injected", null, {
source: acc.source,
sourceAt: acc.sourceAt ? acc.sourceAt.toISOString() : null,
count: cookies.length,
}),
};
}
/** 从该站点 / 账号的浏览器 profile 读回站点域名的 Cookie 头串(没有启用的浏览器服务或插件时 null) */
async readProfileCookie(siteId: string, accountId: string): Promise<string | null> {
const prow = await this.providerRow().catch(() => null);
if (!prow) return null;
const provider = this.providerFor(prow);
if (!provider.getCookies) return null;
const site = await this.liveSite(siteId);
const list = await provider.getCookies(profileKeyFor(siteId, accountId), [site.base_url]);
return headerFromCookies(list, site.base_url, Math.floor(this.now().getTime() / 1000));
}
/**
* 回写:从会话的浏览器 profile 读回站点域名的 Cookie,与凭据存储比较,有变化则更新(来源 browser)。
* 当前页面是登录页时不回写(未登录的 Cookie 不能覆盖可能仍有效的登录态)。失败只记在会话状态里,不抛出。
*/
async syncCookies(id: string, why: "release" | "close" | "periodic"): Promise<{ changed: boolean } | null> {
try {
const row = await this.sessionRow(id);
if (!row.account_id) return null;
const provider = this.providers.get(row.provider_id)?.provider;
if (!provider?.getCookies) return null;
const rs = obj(row.recovery_state) as RecoveryState;
if (rs.credential?.status === "not_applicable") return null;
const at = this.now();
const patch = async (c: Partial<BrowserCredentialState>) => {
const cur = obj((await this.sessionRow(id)).recovery_state) as RecoveryState;
const base: BrowserCredentialState = cur.credential ?? {
status: "none",
source: null,
sourceAt: null,
count: 0,
message: null,
syncedAt: null,
};
await this.ctx.db
.updateTable("browser_session")
.set({
recovery_state: JSON.stringify({ ...cur, credential: { ...base, syncedAt: at.toISOString(), ...c } }),
})
.where("id", "=", id)
.execute();
};
// 先按上游同步当前地址(桌面里的导航不经 API),再判断是否停在登录页
if (provider.sessionInfo(id)) {
await provider.reconcileTabs(id).catch(() => null);
await this.persist(id, provider).catch(() => {});
}
if (provider.sessionInfo(id)?.loginPage === true) {
await patch({});
return { changed: false };
}
const site = await this.liveSite(row.site_id);
let header: string | null;
try {
header = headerFromCookies(
await provider.getCookies(row.profile_key, [site.base_url]),
site.base_url,
Math.floor(at.getTime() / 1000),
);
} catch (e) {
if (e instanceof BrowserError && e.kind === "not_implemented")
await patch({ message: "浏览器服务不支持读回 Cookie(镜像未带 umt-cookies 插件),浏览器里的登录不会回写" });
return null;
}
if (!header) {
await patch({});
return { changed: false };
}
const r = await writeBrowserCookie(this.ctx, row.account_id, header, at);
await patch(
r.changed
? {
source: "browser",
sourceAt: at.toISOString(),
...(rs.credential?.status === "expired" ? { status: "injected" as const, message: null } : {}),
}
: {},
);
if (r.changed)
this.ctx.logger.info(
{ component: "browser", sessionId: id, why, count: r.count },
"浏览器登录态已回写站点账号",
);
return { changed: r.changed };
} catch (e) {
this.ctx.logger.warn(
{ component: "browser", sessionId: id, why, err: String((e as Error)?.message ?? e) },
"浏览器 Cookie 回写失败",
);
return null;
}
}
private async closeRow(row: BrowserSessionRow, reason: string, prow?: BrowserProviderRow): Promise<void> {
const provider = prow ? this.providerFor(prow) : this.providers.get(row.provider_id)?.provider;
// 关闭前趁上下文还在读回一次登录态(站点 / 浏览器服务停用时不再访问)
if (provider?.sessionInfo(row.id) && reason !== "site_disabled" && reason !== "provider_disabled")
await this.syncCookies(row.id, "close");
if (provider?.sessionInfo(row.id)) await provider.closeSession(row.id, reason).catch(() => {});
const at = this.now();
await this.ctx.db.transaction().execute(async (trx) => {
@@ -949,6 +1135,8 @@ export class BrowserService {
});
if (opts.to === "automation") await this.resolveWaiting(row, lease.holder);
await this.persist(id, provider);
// 人工在桌面 / 截图模式里登录过:交还时读回登录态写进站点账号,全站(HTTP 抓取 / RSS / 快照)生效
await this.syncCookies(id, "release");
return toLease(released);
}
@@ -1006,9 +1194,22 @@ export class BrowserService {
throw fromBrowserError(e);
}
if (opts.persist !== false) await this.persist(id, loaded.provider, opts.patch?.(out) ?? {});
this.maybeSyncCookies(row);
return out;
}
/** 有活动时每 COOKIE_SYNC_INTERVAL_MS 读回一次登录态(后台,不拖慢动作;同一会话同时只跑一个) */
private readonly syncing = new Set<string>();
private maybeSyncCookies(row: BrowserSessionRow): void {
if (!row.account_id || this.syncing.has(row.id)) return;
const c = (obj(row.recovery_state) as RecoveryState).credential;
if (c?.status === "not_applicable") return;
const last = c?.syncedAt ? new Date(c.syncedAt).getTime() : new Date(row.created_at).getTime();
if (this.now().getTime() - last < COOKIE_SYNC_INTERVAL_MS) return;
this.syncing.add(row.id);
void this.syncCookies(row.id, "periodic").finally(() => this.syncing.delete(row.id));
}
/** 导航前按站点限流取令牌 */
async limited(site: SiteRow): Promise<void> {
await this.rateLimit(site);
@@ -1350,7 +1551,7 @@ const REASON_TEXT: Record<string, string> = {
challenge: "页面需要人工验证",
two_factor: "需要二次验证或本人设备确认",
challenge_unsolved: "自动处理页面挑战未成功",
login_required: "登录已失效",
login_required: "登录已失效:接管后在画面里登录一次,交还后全站生效",
lease_expired: "人工接管超时未交还",
agent_request: "Agent 请求人工处理",
human_released_waiting: "人工交还时选择继续等待",
@@ -14,7 +14,7 @@ import {
type TmdbSharedCache,
} from "@umt/adapters";
import { DEFAULT_SETTINGS, type SitesSettings } from "@umt/contracts";
import { pgTokenBucketFor, type SiteRow, sql } from "@umt/db";
import { pgTokenBucketFor, type SiteRow } from "@umt/db";
import { CACHE_REGIONS, sharedCache, type TmdbCachePolicy, tmdbCachePolicy } from "../../shared-cache.ts";
import type { ServerContext } from "../types.ts";
@@ -170,7 +170,7 @@ export class SourcesRuntime {
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 =
const limiter =
this.overrides.sharedRateLimit === false
? createSiteRateLimiter(lo)
: createSharedSiteRateLimiter({
@@ -183,7 +183,6 @@ export class SourcesRuntime {
"站点共享限流不可用,本次退回进程内限流",
),
});
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;
@@ -198,41 +197,6 @@ export class SourcesRuntime {
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;
@@ -296,63 +260,6 @@ 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 };
/**
@@ -1,7 +1,6 @@
import { randomUUID } from "node:crypto";
import { getSiteDefinition } from "@umt/adapters";
import { sql } from "@umt/db";
import { type SourcesRuntime, siteAccessFor } from "./runtime.ts";
import type { SourcesRuntime } from "./runtime.ts";
import { archiveStalePool } from "./service/identity-decisions.ts";
// worker 后台调度:按 Feed 轮询策略发 feed.due(outbox → ingestFeed)、按站点快照周期入队 snapshotSite、
@@ -68,32 +67,24 @@ export async function tickFeeds(rt: SourcesRuntime): Promise<number> {
export async function tickSnapshots(rt: SourcesRuntime): Promise<number> {
const { ctx } = rt;
const now = rt.now();
// 站点未设周期时用站点定义与全局默认里更长的一个(settings.sites.defaultSnapshotIntervalSec / umt.defaults)
const defaults = await rt.siteDefaults();
const rows = await ctx.db
.selectFrom("site")
.select(["id", "key", "snapshot_interval_sec", "last_snapshot_at"])
.where("enabled", "=", true)
.where("snapshot_enabled", "=", true)
.where("deleted_at", "is", null)
.execute();
const due: Array<{ id: string; interval: number }> = [];
for (const r of rows) {
const interval =
r.snapshot_interval_sec ?? siteAccessFor(getSiteDefinition(r.key) ?? null, defaults).snapshotIntervalSec;
if (!r.last_snapshot_at || r.last_snapshot_at.getTime() + interval * 1000 <= now.getTime()) {
due.push({ id: r.id, interval });
}
}
for (const s of due) {
const slot = Math.floor(now.getTime() / 1000 / s.interval);
// 站点未设周期时用全局默认(settings.sites.defaultSnapshotIntervalSec)
const fallback = (await rt.siteDefaults()).defaultSnapshotIntervalSec;
const sites = await sql<{ id: string; snapshot_interval_sec: number }>`
select id, coalesce(snapshot_interval_sec, ${fallback}::int) as snapshot_interval_sec from site
where enabled and snapshot_enabled and deleted_at is null
and (last_snapshot_at is null
or last_snapshot_at + make_interval(secs => coalesce(snapshot_interval_sec, ${fallback}::int)) <= ${now})`.execute(
ctx.db,
);
for (const s of sites.rows) {
const slot = Math.floor(now.getTime() / 1000 / s.snapshot_interval_sec);
await ctx.engine.enqueue(
"snapshotSite",
{ siteId: s.id, kind: "periodic" },
{ idempotencyKey: `snapshotSite:${s.id}:${slot}`, traceId: randomUUID() },
);
}
return due.length;
return sites.rows.length;
}
/**
@@ -25,6 +25,7 @@ import { type SiteAccountRow, type SiteRow, sql } from "@umt/db";
import { toApi } from "../rows.ts";
import type { SourcesRuntime } from "../runtime.ts";
import { type Actor, audit, closeDecisions, type Exec, openDecision, SYSTEM } from "./common.ts";
import { cookieFromBrowser, writeBrowserCookie } from "./credentials.ts";
// W9-A 站点中心(docs/design/site-center.md):凭证写入、刷新登录(账密 takelogin / Cookie / API Key 校验)、
// 用户数据与按天历史、健康检测、删除。凭证明文只在内存里用,不进日志、审计与错误信息。
@@ -440,6 +441,17 @@ export async function refreshSiteAccount(
const plain = await secrets.resolve(acc.secret_ref);
let r = await fetchUserData(acc.auth_kind === "cookie" ? { cookie: plain } : { apikey: plain });
if (r.kind === "no_parser") r = await viaHealth();
if (r.kind === "auth" && acc.auth_kind === "cookie") {
// BR-2:存的 Cookie 失效,但浏览器里可能已人工登录过 → 读回浏览器 profile 的 Cookie 试一次,有效才回写
const alt = await cookieFromBrowser(rt.ctx, acc, plain);
if (alt) {
const r2 = await fetchUserData({ cookie: alt });
if (r2.kind === "ok") {
await writeBrowserCookie(rt.ctx, acc.id, alt, rt.now());
return await succeed(r2.info, "存的 Cookie 已失效;浏览器里的登录态有效,已回写");
}
}
}
if (r.kind === "ok") return await succeed(r.info);
if (r.kind === "no_parser") return await succeed(null, "站点没有用户数据解析器,只校验了登录态");
if (r.kind === "network") return await fail("network", r.message);
@@ -889,7 +901,7 @@ export async function tickSiteUserData(rt: SourcesRuntime, slot: string): Promis
return rows.length;
}
/** sites.healthCheck:启用站点逐个检测(同步,走站点限流);按站点访问默认的间隔跳过刚检测过的站点 */
/** sites.healthCheck:启用站点逐个检测(同步,走站点限流) */
export async function tickSiteHealth(rt: SourcesRuntime): Promise<number> {
const sites = await rt.ctx.db
.selectFrom("site")
@@ -899,11 +911,7 @@ export async function tickSiteHealth(rt: SourcesRuntime): Promise<number> {
.where("adapter", "!=", "generic_rss")
.execute();
let n = 0;
const now = rt.now().getTime();
for (const s of sites) {
// 站点访问默认的最短间隔内检测过(手动或定时)就跳过;留 2 分钟余量,免得和定时任务周期错开一轮
const { healthCheckIntervalSec } = await rt.siteAccess(s.key);
if (s.last_health_at && now - s.last_health_at.getTime() < healthCheckIntervalSec * 1000 - 120_000) continue;
try {
await checkSiteHealth(rt, s);
n++;
@@ -0,0 +1,154 @@
import { mergeCookieHeaders, parseCookieHeader, sameCookieHeader } from "@umt/adapters";
import type { SiteCookieSource } from "@umt/contracts";
import type { SiteAccountRow } from "@umt/db";
import type { ServerContext } from "../../types.ts";
// BR-2 凭据统一:站点账号登录态(Cookie)的单一读写入口。
// - 读:用户填的 Cookie(auth_kind = cookie 的 secret_ref)+ 账密自动登录得到的会话 Cookie(session_secret_ref),同名以会话 Cookie 为准;
// - 写(浏览器回写):与当前值比较,有变化才替换对应 secret,并在 user_info.cookieOrigin 记来源 browser 与时间;
// - 来源判定不改其它写入路径:手填 / 自动登录照旧替换 secret(rotated_at 变新),比 cookieOrigin.at 新就说明已被覆盖。
// Cookie 明文只在内存里用,不进日志、审计、错误信息与模型上下文。
type Ctx = Pick<ServerContext, "db" | "secrets" | "logger">;
export interface AccountCookie {
accountId: string;
siteId: string;
authKind: SiteAccountRow["auth_kind"];
/** API Key / passkey 账号为 false:网页登录态与这些凭据无关,不注入浏览器 */
applicable: boolean;
/** 合并后的 Cookie 头串;没有为 null */
header: string | null;
source: SiteCookieSource | null;
sourceAt: Date | null;
}
/** user_info.cookieOrigin:最近一次由浏览器回写的记录(手填 / 自动登录不写,按 secret 时间推断) */
interface CookieOrigin {
source: "browser";
by: "browser";
at: string;
}
const obj = (v: unknown): Record<string, unknown> =>
v && typeof v === "object" && !Array.isArray(v) ? (v as Record<string, unknown>) : {};
/** 主 Cookie secret:账密账号 = 会话 Cookie;Cookie 账号 = 用户填的 */
function primaryRef(acc: SiteAccountRow): string | null {
if (acc.auth_kind === "password") return acc.session_secret_ref;
if (acc.auth_kind === "cookie") return acc.secret_ref;
return null;
}
async function secretTime(ctx: Ctx, ref: string | null): Promise<Date | null> {
if (!ref) return null;
const r = await ctx.db
.selectFrom("secret")
.select(["created_at", "rotated_at"])
.where("id", "=", ref)
.executeTakeFirst();
if (!r) return null;
return new Date(r.rotated_at ?? r.created_at);
}
/** 来源判定:cookieOrigin 记的时间不早于 secret 最近写入(留 5 秒余量)→ 浏览器回写;否则手填 / 自动登录 */
async function originOf(ctx: Ctx, acc: SiteAccountRow): Promise<{ source: SiteCookieSource | null; at: Date | null }> {
const ref = primaryRef(acc);
if (!ref) return { source: null, at: null };
const written = await secretTime(ctx, ref);
const o = obj(obj(acc.user_info).cookieOrigin);
if (o.source === "browser" && typeof o.at === "string") {
const at = new Date(o.at);
if (!written || at.getTime() + 5_000 >= written.getTime()) return { source: "browser", at };
}
return { source: acc.auth_kind === "cookie" ? "manual" : "login", at: written };
}
/** 读站点账号当前登录态(Cookie)。账号不存在返回 null */
export async function readAccountCookie(ctx: Ctx, accountId: string): Promise<AccountCookie | null> {
const acc = await ctx.db.selectFrom("site_account").selectAll().where("id", "=", accountId).executeTakeFirst();
if (!acc) return null;
const base = { accountId: acc.id, siteId: acc.site_id, authKind: acc.auth_kind };
if (acc.auth_kind !== "cookie" && acc.auth_kind !== "password")
return { ...base, applicable: false, header: null, source: null, sourceAt: null };
const user = acc.auth_kind === "cookie" && acc.secret_ref ? await ctx.secrets.resolve(acc.secret_ref) : null;
const session = acc.session_secret_ref ? await ctx.secrets.resolve(acc.session_secret_ref) : null;
const header = mergeCookieHeaders(user, session);
const o = header ? await originOf(ctx, acc) : { source: null, at: null };
return { ...base, applicable: true, header, source: o.source, sourceAt: o.at };
}
/**
* 浏览器回写:与当前 Cookie 比较(忽略顺序),有变化才写。
* Cookie 账号替换 secret_ref;账密账号替换(或新建)会话 Cookie。失效状态的账号回到 unknown(下次抓取 / 健康检测再验证)。
* 返回是否写入。调用方负责只传该站域名的 Cookie,并确认浏览器当前不在登录页。
*/
export async function writeBrowserCookie(
ctx: Ctx,
accountId: string,
header: string,
now: Date = new Date(),
): Promise<{ changed: boolean; count: number }> {
const count = parseCookieHeader(header).size;
if (!count) return { changed: false, count: 0 };
const acc = await ctx.db.selectFrom("site_account").selectAll().where("id", "=", accountId).executeTakeFirst();
if (!acc || (acc.auth_kind !== "cookie" && acc.auth_kind !== "password")) return { changed: false, count };
const cur = await readAccountCookie(ctx, accountId);
if (cur && sameCookieHeader(cur.header, header)) return { changed: false, count };
let sessionRef = acc.session_secret_ref;
if (acc.auth_kind === "cookie") {
if (!acc.secret_ref) return { changed: false, count };
await ctx.secrets.replace(acc.secret_ref, header);
// 旧会话 Cookie 会在合并时覆盖同名项,回写后以浏览器为准
if (sessionRef) await ctx.secrets.replace(sessionRef, header);
} else if (sessionRef) {
await ctx.secrets.replace(sessionRef, header);
} else {
sessionRef = await ctx.secrets.create({
purpose: "site_cookie",
plaintext: header,
label: `site_account:${acc.site_id}:${acc.label}:session:${crypto.randomUUID().slice(0, 8)}`,
});
}
const origin: CookieOrigin = { source: "browser", by: "browser", at: now.toISOString() };
const stale = acc.status === "expired" || acc.status === "invalid";
await ctx.db
.updateTable("site_account")
.set({
user_info: JSON.stringify({ ...obj(acc.user_info), cookieOrigin: origin }),
session_secret_ref: sessionRef,
...(stale ? { status: "unknown" as const, last_error: null } : {}),
updated_at: now,
})
.where("id", "=", acc.id)
.execute();
ctx.logger.info(
{ component: "credentials", accountId, count, updatedBy: "browser" },
"站点账号 Cookie 已从浏览器回写",
);
return { changed: true, count };
}
// ---------- 浏览器读回钩子(浏览器模块注册;sources-identity 不直接依赖浏览器模块) ----------
/** 从该账号的浏览器 profile 读回站点域名的 Cookie 头串;没有浏览器服务 / 插件时返回 null */
export type BrowserCookieReader = (siteId: string, accountId: string) => Promise<string | null>;
const readers = new WeakMap<object, BrowserCookieReader>();
export function registerBrowserCookieReader(ctx: object, reader: BrowserCookieReader): void {
readers.set(ctx, reader);
}
/** refresh-login 发现 Cookie 失效时用:浏览器 profile 里的 Cookie(与存储的相同或读不到返回 null) */
export async function cookieFromBrowser(
ctx: Ctx,
acc: Pick<SiteAccountRow, "id" | "site_id">,
current: string | null,
): Promise<string | null> {
const reader = readers.get(ctx);
if (!reader) return null;
const header = await reader(acc.site_id, acc.id).catch(() => null);
if (!header || sameCookieHeader(header, current)) return null;
return header;
}
@@ -119,11 +119,7 @@ export async function snapshotSite(rt: SourcesRuntime, input: SnapshotInput): Pr
return { status: "skipped", reason: (e as Error).message, ...none };
}
const cfgPages = Number((site.config as { snapshotPages?: number })?.snapshotPages) || 1;
let maxPages = Math.min(20, Math.max(1, input.maxPages ?? cfgPages));
// 定时 / 手动快照受站点访问默认的页数上限约束(补漏有自己的页数预算,实时补查只抓 1 页)
if (input.kind === "periodic" || input.kind === "manual") {
maxPages = Math.min(maxPages, (await rt.siteAccess(site.key)).snapshotMaxPages);
}
const maxPages = Math.min(20, Math.max(1, input.maxPages ?? cfgPages));
const snap = await db
.insertInto("site_snapshot")
.values({
@@ -215,14 +211,9 @@ export async function refreshReleaseDetail(rt: SourcesRuntime, srId: string): Pr
if (!rel) return "not_found";
const site = await db.selectFrom("site").selectAll().where("id", "=", rel.site_id).executeTakeFirst();
if (!site?.enabled) return "site_disabled";
// 详情缓存:站点访问默认的缓存期内抓过就不再请求站点(列表快照会持续更新做种 / 促销)
const { detailCacheSec } = await rt.siteAccess(site.key);
if (rel.detail_fetched_at && rt.now().getTime() - rel.detail_fetched_at.getTime() < detailCacheSec * 1000) {
return null;
}
let detail: SiteRelease;
try {
detail = mergeDetailOverList(await rt.siteAdapter(site).detail(rel.site_release_id), rel);
detail = await rt.siteAdapter(site).detail(rel.site_release_id);
} catch (e) {
if (e instanceof SiteError) {
if (e.isAuthError) await db.transaction().execute((trx) => onAuthError(rt, trx, site, e));
@@ -235,47 +226,7 @@ export async function refreshReleaseDetail(rt: SourcesRuntime, srId: string): Pr
}
await db.transaction().execute(async (trx) => {
await upsertSiteRelease(rt, trx, site, detail);
const keepPromotion = detail.promotion.type === "unknown";
await trx
.updateTable("source_release")
.set({
detail_fetched_at: rt.now(),
// 详情页没解析出促销时保留列表快照的促销,不用"未知"覆盖
...(keepPromotion
? { promotion: jsonb(rel.promotion ?? {}), promotion_expires_at: rel.promotion_expires_at }
: {}),
})
.where("id", "=", srId)
.execute();
await trx.updateTable("source_release").set({ detail_fetched_at: rt.now() }).where("id", "=", srId).execute();
});
return null;
}
/**
* 详情页缺的字段沿用列表快照已有的值:有些站点(如 HHClub 的 div 详情页)详情里没有做种 / 完成数 / 分类名,
* upsert 时 seeders / leechers 与 raw._site 会被整体覆盖,不能用 null 冲掉列表数据。
*/
export function mergeDetailOverList(
detail: SiteRelease,
rel: {
seeders: number | null;
leechers: number | null;
category_raw: string | null;
tags_raw: string[] | null;
raw: unknown;
},
): SiteRelease {
const prev = ((rel.raw ?? {}) as { _site?: Record<string, unknown> })._site ?? {};
return {
...detail,
seeders: detail.seeders ?? rel.seeders,
leechers: detail.leechers ?? rel.leechers,
snatched: detail.snatched ?? (typeof prev.snatched === "number" ? prev.snatched : null),
category: {
id: detail.category.id ?? rel.category_raw,
name: detail.category.name ?? (typeof prev.categoryName === "string" ? prev.categoryName : null),
},
hr: detail.hr === "unknown" && typeof prev.hr === "boolean" ? prev.hr : detail.hr,
tags: detail.tags.length ? detail.tags : (rel.tags_raw ?? []),
};
}
@@ -9,7 +9,7 @@ import { type SiteRow, sql } from "@umt/db";
import { normalizeTitle } from "@umt/domain";
import { ApiException } from "../../../http/errors.ts";
import { decodeCursor, page, toApi } from "../rows.ts";
import { type SourcesRuntime, siteSearchConcurrency } from "../runtime.ts";
import type { SourcesRuntime } from "../runtime.ts";
import { refreshReleaseDetail, snapshotSite } from "./snapshot.ts";
// 站点工作台(整理视图):本地快照优先;本地无结果 / 快照过期 / 用户要求时对已开启站点实时补查,
@@ -20,15 +20,13 @@ const LIVE_FAIL_BACKOFF_MS = 60_000;
interface LiveState {
inflight: Map<string, Promise<{ error: string | null }>>;
failedUntil: Map<string, { until: number; error: string }>;
/** 按站点串行的实时搜索(站点访问默认 searchConcurrency = 1 时) */
siteChain: Map<string, Promise<unknown>>;
}
const liveStates = new WeakMap<SourcesRuntime, LiveState>();
function liveState(rt: SourcesRuntime): LiveState {
let s = liveStates.get(rt);
if (!s) {
s = { inflight: new Map(), failedUntil: new Map(), siteChain: new Map() };
s = { inflight: new Map(), failedUntil: new Map() };
liveStates.set(rt, s);
}
return s;
@@ -46,22 +44,14 @@ export async function liveSearch(
if (failed && failed.until > rt.now().getTime()) return { error: `${failed.error}(退避中)`, coalesced: false };
const running = st.inflight.get(key);
if (running) return { ...(await running), coalesced: true };
const searchConcurrency = siteSearchConcurrency(site.key);
const prev = searchConcurrency <= 1 ? st.siteChain.get(site.id) : undefined;
const p = (async () => {
// 同站不同关键词的实时搜索排队执行(不并发打站点)
if (prev) await prev.catch(() => {});
const r = await snapshotSite(rt, { siteId: site.id, kind: "live_search", query: q, maxPages: 1 });
const error = r.status === "failed" || r.status === "skipped" ? (r.error ?? r.reason ?? "failed") : null;
if (error) st.failedUntil.set(key, { until: rt.now().getTime() + LIVE_FAIL_BACKOFF_MS, error });
else st.failedUntil.delete(key);
return { error };
})().finally(() => {
st.inflight.delete(key);
if (st.siteChain.get(site.id) === p) st.siteChain.delete(site.id);
});
})().finally(() => st.inflight.delete(key));
st.inflight.set(key, p);
if (searchConcurrency <= 1) st.siteChain.set(site.id, p);
return { ...(await p), coalesced: false };
}