fix: 催办按任务类型分流 + 领取事务加锁防超发

- 临期/超期/自动释放查询按 task_type 区分完成口径:内容任务看 publish_url,截图任务看 result_submitted_at,截图任务不再被永久误判为未发布
- portal 领取改为事务内 SELECT ... FOR UPDATE 锁定可用内容,消除并发领取超发窗口
This commit is contained in:
ABAPPLO
2026-09-02 13:56:07 +08:00
parent e1005081fe
commit 5cb056507d
3 changed files with 192 additions and 87 deletions

View File

@@ -563,18 +563,6 @@ async function handlePost(request: Request) {
{ status: 400 },
);
}
const available = await db
.prepare(
`SELECT id FROM contents
WHERE task_id = ? AND status = 'available'
ORDER BY COALESCE(source_row, 999999), created_at, id
LIMIT ${quantity}`,
)
.bind(task.id)
.all<{ id: string }>();
if (available.results.length === 0) {
return Response.json({ error: "当前任务已领完" }, { status: 409 });
}
const partnerId = wecomIdentity
? (await resolveWecomClaimant(db, wecomIdentity.externalUserId)).partnerId
: `partner-ext-${hashText(
@@ -585,6 +573,18 @@ async function handlePost(request: Request) {
: (claimantIdentifier?.display ?? "");
const claimId = uid("claim");
const claimToken = crypto.randomUUID().replaceAll("-", "");
const claimResult = await db.transaction(async (tx) => {
const available = await tx
.prepare(
`SELECT id FROM contents
WHERE task_id = ? AND status = 'available'
ORDER BY COALESCE(source_row, 999999), created_at, id
LIMIT ${quantity}
FOR UPDATE`,
)
.bind(task.id)
.all<{ id: string }>();
if (available.results.length === 0) return null;
const statements: DatabaseStatement[] = [
wecomIdentity
? db
@@ -660,11 +660,15 @@ async function handlePost(request: Request) {
)
.bind(available.results.length, partnerId),
);
await db.batch(statements);
return Response.json({
claimToken,
claimedCount: available.results.length,
for (const statement of statements) {
await tx.execute(statement.sql, statement.values());
}
return { claimToken, claimedCount: available.results.length };
});
if (!claimResult) {
return Response.json({ error: "当前任务已领完" }, { status: 409 });
}
return Response.json(claimResult);
}
if (body.action === "create_delegation") {

View File

@@ -63,7 +63,11 @@ export async function runDueSoonWecomNotifications(
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 (d.publish_url IS NULL OR d.publish_url = '')
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`,
@@ -199,7 +203,11 @@ async function queryOverdueUnpublished(
FROM distributions d
JOIN tasks t ON t.id = d.task_id
JOIN partners p ON p.id = d.partner_id
WHERE (d.publish_url IS NULL OR d.publish_url = '')
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`,

View File

@@ -26,6 +26,9 @@ const overdueRows = [
task_id: "task-A",
task_name: "超期任务A",
due_at: "2026-08-20",
task_type: "content_publish",
publish_url: null,
result_submitted_at: null,
project_id: "proj-1",
partner_id: "p-1",
partner_name: "合作方甲",
@@ -35,6 +38,9 @@ const overdueRows = [
task_id: "task-A",
task_name: "超期任务A",
due_at: "2026-08-20",
task_type: "content_publish",
publish_url: null,
result_submitted_at: null,
project_id: "proj-1",
partner_id: "p-2",
partner_name: "合作方乙",
@@ -44,6 +50,9 @@ const overdueRows = [
task_id: "task-B",
task_name: "无项目任务B",
due_at: "2026-08-25",
task_type: "content_publish",
publish_url: null,
result_submitted_at: null,
project_id: "",
partner_id: "p-3",
partner_name: "合作方丙",
@@ -64,6 +73,15 @@ const releasableDistribution = {
result_submitted_at: null,
};
function filterOverdue(rows, cutoff) {
return rows.filter((row) => {
if (!(row.due_at && row.due_at < cutoff)) return false;
return row.task_type === "screenshot_collect"
? !row.result_submitted_at
: !row.publish_url;
});
}
function fakeEscalationDb({ rows = overdueRows } = {}) {
return {
prepare(sql) {
@@ -77,8 +95,11 @@ function fakeEscalationDb({ rows = overdueRows } = {}) {
return null;
},
async all() {
if (/FROM distributions d/.test(sql) && /publish_url IS NULL/.test(sql)) {
return { results: rows.filter((row) => row.due_at < params[0]) };
if (
/FROM distributions d/.test(sql) &&
/result_submitted_at IS NULL/.test(sql)
) {
return { results: filterOverdue(rows, params[0]) };
}
return { results: [] };
},
@@ -117,8 +138,11 @@ function fakeReleaseDb({ rows = overdueRows, blockedIds = [] } = {}) {
return null;
},
async all() {
if (/FROM distributions d/.test(sql) && /publish_url IS NULL/.test(sql)) {
return { results: rows.filter((row) => row.due_at < params[0]) };
if (
/FROM distributions d/.test(sql) &&
/result_submitted_at IS NULL/.test(sql)
) {
return { results: filterOverdue(rows, params[0]) };
}
return { results: [] };
},
@@ -226,6 +250,75 @@ test("runOverdueEscalation no-ops without overdue rows", async () => {
assert.equal(calls.length, 0);
});
test("runOverdueEscalation distinguishes screenshot tasks from content tasks", async () => {
const rows = [
{
id: "d-s1",
task_id: "task-S",
task_name: "截图任务S",
due_at: "2026-08-20",
task_type: "screenshot_collect",
publish_url: null,
result_submitted_at: null,
project_id: "",
partner_id: "p-1",
partner_name: "合作方甲",
},
{
id: "d-s2",
task_id: "task-S",
task_name: "截图任务S",
due_at: "2026-08-20",
task_type: "screenshot_collect",
publish_url: null,
result_submitted_at: "2026-08-21 09:00:00",
project_id: "",
partner_id: "p-2",
partner_name: "合作方乙",
},
{
id: "d-c1",
task_id: "task-A",
task_name: "内容任务A",
due_at: "2026-08-20",
task_type: "content_publish",
publish_url: "https://www.xiaohongshu.com/note/1",
result_submitted_at: null,
project_id: "",
partner_id: "p-3",
partner_name: "合作方丙",
},
{
id: "d-c2",
task_id: "task-A",
task_name: "内容任务A",
due_at: "2026-08-20",
task_type: "content_publish",
publish_url: null,
result_submitted_at: null,
project_id: "",
partner_id: "p-4",
partner_name: "合作方丁",
},
];
const calls = [];
const summary = await runOverdueEscalation(
fakeEscalationDb({ rows }),
config,
wecomFetch(calls),
NOW,
);
// 已提交截图的 d-s2 与已回填链接的 d-c1 不算超期
assert.equal(summary.overdueTasks, 2);
assert.equal(summary.overdueDistributions, 2);
const robotBodies = calls
.filter((call) => call.url.host === "qyapi.example.com")
.map((call) => JSON.parse(String(call.init.body)));
assert.equal(robotBodies.length, 1);
assert.match(robotBodies[0].text.content, /截图任务S(截止 2026-08-20):1 条/);
assert.match(robotBodies[0].text.content, /内容任务A(截止 2026-08-20):1 条/);
});
test("runOverdueAutoRelease is disabled without env switch", async () => {
const calls = [];
const db = fakeReleaseDb();