队列租约与调度分流 深度解析

146	private async enqueue(task: {
147		contactId?: string;
148		companyId?: string;
149		kind: string;
150		reason: string;
151		priority: number;
152		budget: number;
153	}): Promise<void> {
154		try {
155			const pending = await this.db.agentTask.findFirst({
156				where: {
157					kind: task.kind,
158					finishedAt: null,
159					...(task.contactId ? { contactId: task.contactId } : {}),
160					...(task.companyId ? { companyId: task.companyId } : {}),
161				},
162				select: { id: true },
163			});
164
165			if (pending) return;
166
167			await this.db.agentTask.create({
168				data: {
169					contactId: task.contactId ?? null,
170					companyId: task.companyId ?? null,
171					kind: task.kind,
172					reason: task.reason,
173					priority: task.priority,
174					budget: task.budget,
175					dueAt: new Date(),
176				},
177			});
178
179			this.logger.log({
180				message: "Agent task queued",
181				kind: task.kind,
182				contactId: task.contactId,
183				companyId: task.companyId,
184			});
185
186			this.poke();
187		} catch (error) {
188			this.logger.error(
189				{ message: "Could not queue agent task", kind: task.kind },
190				error instanceof Error ? error.stack : String(error),
191			);
192		}
193	}
194
195	private poke(): void {
196		const agent = bridge();
197		if (!agent) return;
198
199		const missed = (error: unknown) => {
200			this.logger.debug({
201				message: "Agent poke did not land; the cron will pick this up",
202				reason: error instanceof Error ? error.message : String(error),
203			});
204		};
205
206		try {
207			void fetch(agent.url("/internal/crm/dispatch"), {
208				method: "POST",
209				headers: { authorization: `Bearer ${agent.secret}` },
210				signal: AbortSignal.timeout(POKE_TIMEOUT_MS),
211			}).catch(missed);
212		} catch (error) {
213			missed(error);
214		}
215	}
216}

28export async function claimDue(
29	limit: number,
30	kinds: { only: readonly string[] } | { except: readonly string[] },
31	leaseMs = LEASE_MS,
32): Promise<LeasedTask[]> {
33	const now = new Date();
34	const until = new Date(now.getTime() + leaseMs);
35
36	const list = "only" in kinds ? kinds.only : kinds.except;
37	if ("only" in kinds && list.length === 0) return [];
38
39	const match = Prisma.sql`t2.kind ${"only" in kinds ? Prisma.sql`IN` : Prisma.sql`NOT IN`} (${Prisma.join(list)})`;
40
41	const claimed = await db.$queryRaw<LeasedTask[]>`
42		UPDATE "agentTask" AS t
43		SET "leasedUntil" = ${until},
44			"startedAt" = COALESCE(t."startedAt", ${now}),
45			"attempts" = t."attempts" + 1
46		FROM (
47			SELECT t2.id FROM "agentTask" AS t2
48			WHERE t2."finishedAt" IS NULL
49				AND t2."dueAt" <= ${now}
50				AND (t2."leasedUntil" IS NULL OR t2."leasedUntil" < ${now})
51				AND t2."attempts" < ${MAX_ATTEMPTS}
52				AND ${match}
53			ORDER BY t2."priority" DESC, t2."dueAt" ASC
54			LIMIT ${limit}
55			FOR UPDATE SKIP LOCKED
56		) AS due
57		WHERE t.id = due.id
58		RETURNING t.id, t."contactId", t."companyId", t.kind, t.reason,
59			t.budget, t.attempts, t.priority, t."dueAt";
60	`;
61
62	return claimed.sort(
63		(a, b) => b.priority - a.priority || a.dueAt.getTime() - b.dueAt.getTime(),
64	);
65}

17export const VISIBLE_BATCH = 60;
18export const VISIBLE_CONCURRENCY = 6;
19export const VISIBLE_LEASE_MS = 2 * 60_000;
20
21export const RESEARCH_BATCH = 12;
22export const RESEARCH_LEASE_MS = 30 * 60_000;
23
24export async function retireAbandoned(): Promise<void> {
25	let abandoned: TaskSubject[] = [];
26
27	try {
28		abandoned = await retireExhausted();
29	} catch {
30		return;
31	}
32
33	for (const task of abandoned) {
34		await settle(
35			task,
36			EnrichmentStatus.FAILED,
37			"Research was attempted several times and never completed.",
38		).catch(() => {});
39	}
40}
41
42export async function runVisibleLane(): Promise<number> {
43	let handled = 0;
44
45	while (handled < VISIBLE_BATCH) {
46		const tasks = await claimDue(
47			Math.min(VISIBLE_CONCURRENCY, VISIBLE_BATCH - handled),
48			{ only: DIRECT_KINDS },
49			VISIBLE_LEASE_MS,
50		);
51
52		if (tasks.length === 0) break;
53
54		await runLimited(VISIBLE_CONCURRENCY, tasks, runDirect);
55		handled += tasks.length;
56	}
57
58	return handled;
59}
60
61async function runDirect(task: LeasedTask): Promise<void> {
62	try {
63		if (task.kind === "brand" && task.companyId) {
64			const result = await runBrand({ companyId: task.companyId });
65			if (result.retryable) return;
66
67			await completeTask(task.id, brandOutcome(result));
68			return;
69		}
70
71		if (task.kind === "portrait" && task.contactId) {
72			const portrait = await runPortrait({
73				contactId: task.contactId,
74				spend: () => ({ ok: true }),
75			});
76
77			await completeTask(
78				task.id,
79				portrait.stored
80					? `Picture stored from ${portrait.source}.`
81					: (portrait.reason ?? "No picture found."),
82			);
83			return;
84		}
85
86		await completeTask(task.id, "The record this names is gone.");
87	} catch (error) {
88		const reason = error instanceof Error ? error.message : String(error);
89		await settle(task, EnrichmentStatus.FAILED, reason).catch(() => {});
90	}
91}
92
93export async function runResearchLane(
94	start: (task: LeasedTask) => Promise<{ id: string }>,
95): Promise<number> {
96	const tasks = await claimDue(
97		RESEARCH_BATCH,
98		{ except: DIRECT_KINDS },
99		RESEARCH_LEASE_MS,
100	);
101	if (tasks.length === 0) return 0;
102
103	await Promise.all(
104		tasks.map(async (task) => {
105			try {
106				await markRunning(task);
107				const session = await start(task);
108				await noteSession(task.id, session.id);
109			} catch (error) {
110				const reason = error instanceof Error ? error.message : String(error);
111				await settle(task, EnrichmentStatus.FAILED, reason).catch(() => {});
112			}
113		}),
114	);
115
116	return tasks.length;
117}
118
119export function taskAuth(task: LeasedTask, base: AppAuth = APP_AUTH): AppAuth {
120	return {
121		...base,
122		attributes: {
123			taskKind: task.kind,
124			reason: task.reason,
125			budget: String(task.budget),
126			...(task.contactId ? { contactId: task.contactId } : {}),
127			...(task.companyId ? { companyId: task.companyId } : {}),
128		},
129	};
130}
131
132export const drainAll = collapsing(
133	async (start: (task: LeasedTask) => Promise<{ id: string }>) => {
134		await retireAbandoned();
135		await Promise.all([runVisibleLane(), runResearchLane(start)]);
136	},
137);

17export function taskToken(taskId: string): string {
18	return `${TASK_MARKER}${taskId}`;
19}
20
21export function taskFromToken(token: string | undefined): string | null {
22	if (!token) return null;
23
24	const marker = token.lastIndexOf(TASK_MARKER);
25	if (marker === -1) return null;
26
27	const id = token.slice(marker + TASK_MARKER.length);
28	return id.length > 0 ? id : null;
29}