5cb056507d
- 临期/超期/自动释放查询按 task_type 区分完成口径：内容任务看 publish_url，截图任务看 result_submitted_at，截图任务不再被永久误判为未发布 - portal 领取改为事务内 SELECT ... FOR UPDATE 锁定可用内容，消除并发领取超发窗口
403 lines
12 KiB
TypeScript
403 lines
12 KiB
TypeScript
import type { DatabaseClient } from "./database";
|
||
import { shanghaiDateFromTimestamp } from "./collection-service";
|
||
import {
|
||
sendWecomAppMessage,
|
||
sendWecomRobotMessage,
|
||
sendWecomTemplateCard,
|
||
type WecomConfig,
|
||
} from "./wecom-client";
|
||
import { getRuntimeEnv } from "./runtime-env";
|
||
import { releaseUnfinishedDistribution } from "./distribution-release-service";
|
||
|
||
type FetchLike = typeof fetch;
|
||
|
||
export type WecomNotifySummary = {
|
||
dueSoonAttempted: number;
|
||
dueSoonSent: number;
|
||
dueSoonFailed: number;
|
||
dueSoonSkipped: number;
|
||
digestSent: boolean;
|
||
};
|
||
|
||
type DueSoonRow = {
|
||
distribution_id: string;
|
||
partner_id: string;
|
||
partner_name: string;
|
||
wecom_external_user_id: string | null;
|
||
task_name: string;
|
||
due_at: string;
|
||
content_title: string;
|
||
};
|
||
|
||
export function computeDueCutoff(today: string, dueDays: number) {
|
||
const match = /^(\d{4})-(\d{2})-(\d{2})$/.exec(today);
|
||
if (!match) return today;
|
||
const [, y, m, d] = match;
|
||
const date = new Date(
|
||
Date.UTC(Number(y), Number(m) - 1, Number(d)) + dueDays * 24 * 60 * 60 * 1_000,
|
||
);
|
||
return date.toISOString().slice(0, 10);
|
||
}
|
||
|
||
export async function runDueSoonWecomNotifications(
|
||
db: DatabaseClient,
|
||
config: WecomConfig,
|
||
fetchImpl: FetchLike = fetch,
|
||
now: number = Date.now(),
|
||
): Promise<WecomNotifySummary> {
|
||
const today = shanghaiDateFromTimestamp(now);
|
||
const cutoff = computeDueCutoff(today, config.dueDays);
|
||
const portalUrl = bindingValue(getRuntimeEnv().KOC_PORTAL_URL);
|
||
|
||
const result = await db
|
||
.prepare(
|
||
`SELECT
|
||
d.id AS distribution_id,
|
||
d.partner_id,
|
||
p.name AS partner_name,
|
||
p.wecom_external_user_id,
|
||
t.name AS task_name,
|
||
t.due_at,
|
||
c.title AS content_title
|
||
FROM distributions d
|
||
JOIN tasks t ON t.id = d.task_id
|
||
JOIN partners p ON p.id = d.partner_id
|
||
JOIN contents c ON c.id = d.content_id
|
||
WHERE (
|
||
(t.task_type = 'screenshot_collect' AND d.result_submitted_at IS NULL)
|
||
OR (t.task_type != 'screenshot_collect'
|
||
AND (d.publish_url IS NULL OR d.publish_url = ''))
|
||
)
|
||
AND t.due_at IS NOT NULL AND t.due_at != ''
|
||
AND t.due_at <= ?
|
||
ORDER BY t.due_at ASC, p.name ASC`,
|
||
)
|
||
.bind(cutoff)
|
||
.all<DueSoonRow>();
|
||
|
||
const grouped = new Map<
|
||
string,
|
||
{
|
||
partnerName: string;
|
||
externalUserId: string | null;
|
||
rows: DueSoonRow[];
|
||
}
|
||
>();
|
||
for (const row of result.results) {
|
||
const entry = grouped.get(row.partner_id) ?? {
|
||
partnerName: row.partner_name,
|
||
externalUserId: row.wecom_external_user_id,
|
||
rows: [],
|
||
};
|
||
entry.rows.push(row);
|
||
if (!entry.externalUserId && row.wecom_external_user_id) {
|
||
entry.externalUserId = row.wecom_external_user_id;
|
||
}
|
||
grouped.set(row.partner_id, entry);
|
||
}
|
||
|
||
let dueSoonAttempted = 0;
|
||
let dueSoonSent = 0;
|
||
let dueSoonFailed = 0;
|
||
let dueSoonSkipped = 0;
|
||
const digestTasks: string[] = [];
|
||
|
||
for (const [, entry] of grouped) {
|
||
dueSoonAttempted += 1;
|
||
const external = entry.externalUserId
|
||
? [entry.externalUserId]
|
||
: [];
|
||
const lines = entry.rows.slice(0, 5).map((row) => {
|
||
return `· 《${truncate(row.task_name, 24)}》— ${truncate(row.content_title, 24)}（截止 ${row.due_at}）`;
|
||
});
|
||
const overflow =
|
||
entry.rows.length > 5 ? `\n…还有 ${entry.rows.length - 5} 条` : "";
|
||
const link = portalUrl ? `\n领取链接：${portalUrl}` : "";
|
||
const content =
|
||
`${entry.partnerName}，你有 ${entry.rows.length} 条内容待发布：\n${lines.join("\n")}${overflow}${link}`;
|
||
const appResult = await sendWecomAppMessage(
|
||
external,
|
||
content,
|
||
config,
|
||
fetchImpl,
|
||
).catch((error: unknown) => {
|
||
console.warn(
|
||
"[KOC LOOP] wecom app message failed",
|
||
{ partner: entry.partnerName, error: safeError(error) },
|
||
);
|
||
return null;
|
||
});
|
||
if (appResult === null) {
|
||
dueSoonFailed += 1;
|
||
} else if (appResult.skipped) {
|
||
dueSoonSkipped += 1;
|
||
} else {
|
||
dueSoonSent += 1;
|
||
}
|
||
const earliestDue = entry.rows[0]?.due_at ?? "";
|
||
digestTasks.push(
|
||
`· ${entry.partnerName}（${entry.rows.length} 条，最近截止 ${earliestDue}）`,
|
||
);
|
||
}
|
||
|
||
let digestSent = false;
|
||
if (config.robotWebhook && digestTasks.length > 0) {
|
||
const digest =
|
||
`今日待发布催办（${today}，截止 ≤ ${cutoff}）：\n${digestTasks.join("\n")}`;
|
||
try {
|
||
await sendWecomRobotMessage(digest, config, fetchImpl);
|
||
digestSent = true;
|
||
} catch (error) {
|
||
console.warn(
|
||
"[KOC LOOP] wecom robot digest failed",
|
||
{ error: safeError(error) },
|
||
);
|
||
}
|
||
}
|
||
|
||
return {
|
||
dueSoonAttempted,
|
||
dueSoonSent,
|
||
dueSoonFailed,
|
||
dueSoonSkipped,
|
||
digestSent,
|
||
};
|
||
}
|
||
|
||
function bindingValue(value: unknown) {
|
||
return String(value ?? "").trim();
|
||
}
|
||
|
||
export function resolveAutoReleaseDays(value: string | undefined) {
|
||
const parsed = Number(String(value ?? "").trim());
|
||
if (!Number.isFinite(parsed) || parsed < 1) return 0;
|
||
return Math.min(90, Math.floor(parsed));
|
||
}
|
||
|
||
type OverdueRow = {
|
||
id: string;
|
||
task_id: string;
|
||
task_name: string;
|
||
due_at: string;
|
||
project_id: string;
|
||
partner_id: string;
|
||
partner_name: string;
|
||
};
|
||
|
||
export type OverdueEscalationSummary = {
|
||
overdueTasks: number;
|
||
overdueDistributions: number;
|
||
robotSent: boolean;
|
||
cardSent: number;
|
||
cardFailed: number;
|
||
};
|
||
|
||
async function queryOverdueUnpublished(
|
||
db: DatabaseClient,
|
||
cutoff: string,
|
||
): Promise<OverdueRow[]> {
|
||
const result = await db
|
||
.prepare(
|
||
`SELECT d.id, t.id AS task_id, t.name AS task_name, t.due_at,
|
||
t.project_id, p.id AS partner_id, p.name AS partner_name
|
||
FROM distributions d
|
||
JOIN tasks t ON t.id = d.task_id
|
||
JOIN partners p ON p.id = d.partner_id
|
||
WHERE (
|
||
(t.task_type = 'screenshot_collect' AND d.result_submitted_at IS NULL)
|
||
OR (t.task_type != 'screenshot_collect'
|
||
AND (d.publish_url IS NULL OR d.publish_url = ''))
|
||
)
|
||
AND t.due_at IS NOT NULL AND t.due_at != ''
|
||
AND t.due_at < ?
|
||
ORDER BY t.due_at ASC, t.name ASC`,
|
||
)
|
||
.bind(cutoff)
|
||
.all<OverdueRow>();
|
||
return result.results;
|
||
}
|
||
|
||
export async function runOverdueEscalation(
|
||
db: DatabaseClient,
|
||
config: WecomConfig,
|
||
fetchImpl: FetchLike = fetch,
|
||
now: number = Date.now(),
|
||
): Promise<OverdueEscalationSummary> {
|
||
const today = shanghaiDateFromTimestamp(now);
|
||
const rows = await queryOverdueUnpublished(db, today);
|
||
|
||
const byTask = new Map<
|
||
string,
|
||
{ taskName: string; dueAt: string; projectId: string; count: number; partners: Set<string> }
|
||
>();
|
||
for (const row of rows) {
|
||
const entry = byTask.get(row.task_id) ?? {
|
||
taskName: row.task_name,
|
||
dueAt: row.due_at,
|
||
projectId: bindingValue(row.project_id),
|
||
count: 0,
|
||
partners: new Set<string>(),
|
||
};
|
||
entry.count += 1;
|
||
entry.partners.add(row.partner_name || row.partner_id);
|
||
byTask.set(row.task_id, entry);
|
||
}
|
||
|
||
const summary: OverdueEscalationSummary = {
|
||
overdueTasks: byTask.size,
|
||
overdueDistributions: rows.length,
|
||
robotSent: false,
|
||
cardSent: 0,
|
||
cardFailed: 0,
|
||
};
|
||
|
||
if (rows.length === 0) return summary;
|
||
|
||
if (config.robotWebhook) {
|
||
const lines = [...byTask.values()].map((entry) => {
|
||
const partners = [...entry.partners].slice(0, 5).join("、");
|
||
const overflow = entry.partners.size > 5 ? ` 等 ${entry.partners.size} 位` : "";
|
||
return `· ${truncate(entry.taskName, 24)}（截止 ${entry.dueAt}）：${entry.count} 条超期未发布 — ${partners}${overflow}`;
|
||
});
|
||
try {
|
||
await sendWecomRobotMessage(
|
||
`【超期未发布升级】${today}\n${lines.join("\n")}\n请运营跟进催办或释放。`,
|
||
config,
|
||
fetchImpl,
|
||
);
|
||
summary.robotSent = true;
|
||
} catch (error) {
|
||
console.warn("[KOC LOOP] overdue escalation robot failed", {
|
||
error: safeError(error),
|
||
});
|
||
}
|
||
}
|
||
|
||
const env = getRuntimeEnv();
|
||
const cardUrl = bindingValue(env.APP_ORIGIN) || bindingValue(env.KOC_PORTAL_URL);
|
||
for (const [taskId, entry] of byTask) {
|
||
if (!entry.projectId) continue;
|
||
try {
|
||
const project = await db
|
||
.prepare("SELECT main_group_member_ids FROM projects WHERE id = ?")
|
||
.bind(entry.projectId)
|
||
.first<{ main_group_member_ids: string }>();
|
||
const members = safeParseIdList(project?.main_group_member_ids);
|
||
if (members.length === 0) continue;
|
||
const sent = await sendWecomTemplateCard(
|
||
config,
|
||
{
|
||
touser: members,
|
||
taskId: `task:${taskId}:due`,
|
||
title: `超期催办：${truncate(entry.taskName, 20)}`,
|
||
description: `${entry.count} 条内容超期未发布（截止 ${entry.dueAt}），请跟进催办或释放`,
|
||
url: cardUrl,
|
||
buttons: [
|
||
{ key: "view_task", name: "查看任务", url: cardUrl },
|
||
{ key: "mark_handled", name: "已处理" },
|
||
],
|
||
},
|
||
fetchImpl,
|
||
);
|
||
if (!sent.skipped) {
|
||
summary.cardSent += sent.sent;
|
||
summary.cardFailed += sent.failed;
|
||
}
|
||
} catch (error) {
|
||
summary.cardFailed += 1;
|
||
console.warn("[KOC LOOP] overdue escalation card failed", {
|
||
taskId,
|
||
error: safeError(error),
|
||
});
|
||
}
|
||
}
|
||
|
||
return summary;
|
||
}
|
||
|
||
export type AutoReleaseSummary = {
|
||
enabled: boolean;
|
||
days: number;
|
||
cutoff: string;
|
||
released: number;
|
||
blocked: number;
|
||
failed: number;
|
||
robotSent: boolean;
|
||
};
|
||
|
||
export async function runOverdueAutoRelease(
|
||
db: DatabaseClient,
|
||
config: WecomConfig,
|
||
fetchImpl: FetchLike = fetch,
|
||
now: number = Date.now(),
|
||
): Promise<AutoReleaseSummary> {
|
||
const days = resolveAutoReleaseDays(getRuntimeEnv().WECOM_AUTO_RELEASE_DAYS);
|
||
const summary: AutoReleaseSummary = {
|
||
enabled: days > 0,
|
||
days,
|
||
cutoff: "",
|
||
released: 0,
|
||
blocked: 0,
|
||
failed: 0,
|
||
robotSent: false,
|
||
};
|
||
if (days <= 0) return summary;
|
||
|
||
const today = shanghaiDateFromTimestamp(now);
|
||
summary.cutoff = computeDueCutoff(today, -days);
|
||
const rows = await queryOverdueUnpublished(db, summary.cutoff);
|
||
|
||
for (const row of rows) {
|
||
try {
|
||
await releaseUnfinishedDistribution(db, row.id);
|
||
summary.released += 1;
|
||
} catch (error) {
|
||
const status = (error as { status?: number }).status;
|
||
if (status === 409) {
|
||
summary.blocked += 1;
|
||
} else {
|
||
summary.failed += 1;
|
||
console.warn("[KOC LOOP] auto release failed", {
|
||
distributionId: row.id,
|
||
error: safeError(error),
|
||
});
|
||
}
|
||
}
|
||
}
|
||
|
||
if (summary.released > 0 && config.robotWebhook) {
|
||
try {
|
||
await sendWecomRobotMessage(
|
||
`【超期自动释放】截止 ${summary.cutoff} 前仍未发布的内容已释放 ${summary.released} 条回库存（跳过 ${summary.blocked} 条已提交、失败 ${summary.failed} 条），可重新领取。`,
|
||
config,
|
||
fetchImpl,
|
||
);
|
||
summary.robotSent = true;
|
||
} catch (error) {
|
||
console.warn("[KOC LOOP] auto release robot failed", {
|
||
error: safeError(error),
|
||
});
|
||
}
|
||
}
|
||
|
||
return summary;
|
||
}
|
||
|
||
function safeParseIdList(value: string | null | undefined) {
|
||
try {
|
||
const parsed = JSON.parse(String(value ?? "[]"));
|
||
if (!Array.isArray(parsed)) return [];
|
||
return parsed.map((item) => String(item)).filter(Boolean);
|
||
} catch {
|
||
return [];
|
||
}
|
||
}
|
||
|
||
function truncate(value: string, max: number) {
|
||
return value.length > max ? `${value.slice(0, max)}…` : value;
|
||
}
|
||
|
||
function safeError(error: unknown) {
|
||
return error instanceof Error ? error.message : String(error);
|
||
}
|