Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions cloudflare/automation/src/github.js
Original file line number Diff line number Diff line change
Expand Up @@ -120,12 +120,18 @@ export function isMergedPullRequest(pullRequest) {
return pullRequest?.merged === true || Boolean(pullRequest?.merged_at);
}

export async function listRecentlyMergedPullRequests(env, repository, since) {
export async function listMergedPullRequests(env, repository, since) {
const normalizedRepository = repositoryName(repository);
const [owner, repo] = normalizedRepository.split("/");
const value = await githubApi(env, "GET", `/repos/${encodeSegment(owner)}/${encodeSegment(repo)}/pulls?state=closed&per_page=100&sort=updated&direction=desc`);
const pullRequests = [];
for (let page = 1; ; page += 1) {
const value = await githubApi(env, "GET", `/repos/${encodeSegment(owner)}/${encodeSegment(repo)}/pulls?state=closed&per_page=100&sort=updated&direction=desc&page=${page}`);
if (!Array.isArray(value)) throw new Error("GitHub 已关闭 PR 列表格式无效");
pullRequests.push(...value);
if (value.length < 100) break;
}
const sinceTime = Date.parse(String(since || ""));
return (Array.isArray(value) ? value : []).filter((pullRequest) => isMergedPullRequest(pullRequest)
return pullRequests.filter((pullRequest) => isMergedPullRequest(pullRequest)
&& (!Number.isFinite(sinceTime) || Date.parse(String(pullRequest.merged_at || "")) >= sinceTime));
}

Expand Down
66 changes: 58 additions & 8 deletions cloudflare/automation/src/online-runtime.js
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
import { listChatMembers, sendFeishuText, formatMemberOnboarding } from "./feishu.js";
import { listOpenPullRequests, listRecentlyMergedPullRequests } from "./github.js";
import { listMergedPullRequests, listOpenPullRequests } from "./github.js";
import {
ensureNextStageTask,
ensureStageOneTask,
listProjectRecords,
projectForRepository,
shouldReconcileMergedPullRequest,
taskRecordSnapshot,
} from "./task-board.js";
import { formatDailyReport } from "./online.js";
Expand Down Expand Up @@ -36,6 +37,23 @@ function reconciliationBucket() {
return Math.floor(Date.now() / (10 * 60 * 1000));
}

function shanghaiDateKey(date) {
const parts = Object.fromEntries(new Intl.DateTimeFormat("en-CA", {
timeZone: "Asia/Shanghai",
year: "numeric",
month: "2-digit",
day: "2-digit",
}).formatToParts(date).map(({ type, value }) => [type, value]));
return `${parts.year}-${parts.month}-${parts.day}`;
}

export function mergedPullRequestSyncKey(repository, pullRequest, date = new Date()) {
const number = Number(pullRequest?.number);
const mergeId = String(pullRequest?.merge_commit_sha || pullRequest?.head?.sha || pullRequest?.merged_at || "").trim();
if (!Number.isInteger(number) || number < 1 || !mergeId) throw new Error("合并 PR 的编号或提交标识缺失");
return `merged-pr-task-sync:${repository}#${number}:${mergeId}:${shanghaiDateKey(date)}`;
}

async function enqueueReviewForOpenPullRequest({ env, project, pullRequest, enqueueEvent, runAction, eventId }) {
const number = Number(pullRequest.number);
const headSha = String(pullRequest.head?.sha || "");
Expand All @@ -56,6 +74,24 @@ async function enqueueReviewForOpenPullRequest({ env, project, pullRequest, enqu
);
}

async function enqueueMergedPullRequestTaskSync({ env, project, pullRequest, enqueueEvent, runAction, eventId, date }) {
const number = Number(pullRequest.number);
const actionKey = mergedPullRequestSyncKey(project.repository, pullRequest, date);
return runAction(
env,
eventForCron(eventId),
"merged_pr_task_sync_enqueue",
actionKey,
() => enqueueEvent({
event_id: actionKey,
source: "internal",
event_type: "reconcile",
delivery_id: `merged-pr:${project.repository}#${number}:${pullRequest.merge_commit_sha || pullRequest.head?.sha || pullRequest.merged_at}`,
payload: { repository: project.repository, number },
}).then(() => ({ queued: true, repository: project.repository, number })),
);
}

async function onboardMembers({ env, project, members, records, runAction, eventId }) {
let created = 0;
let messages = 0;
Expand Down Expand Up @@ -127,7 +163,7 @@ async function continueCompletedTasks({ env, project, records, runAction, eventI
export async function runOnlineReconciliation({ env, enqueueEvent, runAction } = {}) {
if (typeof enqueueEvent !== "function" || typeof runAction !== "function") throw new Error("online reconciliation dependencies are missing");
const eventId = `cron-reconcile:${reconciliationBucket()}`;
const summary = { projects: 0, members: 0, stageOneTasks: 0, nextStageTasks: 0, nextStageSkipped: {}, messages: 0, reviewsQueued: 0, openPullRequests: 0, mergedPullRequests: 0 };
const summary = { projects: 0, members: 0, stageOneTasks: 0, nextStageTasks: 0, nextStageSkipped: {}, messages: 0, reviewsQueued: 0, openPullRequests: 0, mergedPullRequests: 0, mergedTaskSyncQueued: 0 };
for (const repository of PROJECT_REPOSITORIES) {
const project = projectForRepository(env, repository);
if (!project) throw new Error(`project is not configured: ${repository}`);
Expand All @@ -153,12 +189,26 @@ export async function runOnlineReconciliation({ env, enqueueEvent, runAction } =
const result = await enqueueReviewForOpenPullRequest({ env, project, pullRequest, enqueueEvent, runAction, eventId });
if (result.queued) summary.reviewsQueued += 1;
}
const recentMerges = await listRecentlyMergedPullRequests(env, project.repository, new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString());
summary.mergedPullRequests += recentMerges.length;
for (const pullRequest of recentMerges) {
const result = await enqueueReviewForOpenPullRequest({ env, project, pullRequest: { ...pullRequest, head: { sha: pullRequest.head?.sha || pullRequest.merge_commit_sha || pullRequest.merged_at } }, enqueueEvent, runAction, eventId });
if (result.queued) summary.reviewsQueued += 1;
}
const syncDate = new Date();
const syncDay = shanghaiDateKey(syncDate);
const mergedScan = await runAction(
env,
eventForCron(eventId),
"online_merged_pr_scan",
`online-merged-pr-scan:${project.key}:${syncDay}`,
async () => {
const mergedPullRequests = await listMergedPullRequests(env, project.repository);
let queued = 0;
for (const pullRequest of mergedPullRequests) {
if (!await shouldReconcileMergedPullRequest(env, project, records, pullRequest)) continue;
const result = await enqueueMergedPullRequestTaskSync({ env, project, pullRequest, enqueueEvent, runAction, eventId, date: syncDate });
if (result.queued) queued += 1;
}
return { mergedPullRequests: mergedPullRequests.length, queued };
},
);
summary.mergedPullRequests += Number(mergedScan.mergedPullRequests || 0);
summary.mergedTaskSyncQueued += Number(mergedScan.queued || 0);
summary.projects += 1;
}
return summary;
Expand Down
69 changes: 58 additions & 11 deletions cloudflare/automation/src/task-board.js
Original file line number Diff line number Diff line change
Expand Up @@ -517,6 +517,62 @@ function selectLinkedTaskRecord(records, prUrl, number) {
})[0] || null;
}

export function selectPreviouslyLinkedTaskRecord(records, responses) {
const recordIds = [...new Set((Array.isArray(responses) ? responses : [])
.filter((response) => response?.matched === true && response.record_id)
.map((response) => String(response.record_id)))];
if (recordIds.length !== 1) return null;
return (Array.isArray(records) ? records : [])
.find((record) => record?.record_id === recordIds[0]) || null;
}

async function previouslyLinkedTaskRecord(env, project, records, outcome) {
const number = Number(outcome?.number);
if (!env.STATE_DB || !Number.isInteger(number) || number < 1) return null;
const prefix = `task-pr:${project.repository}#${number}:%`;
const result = await env.STATE_DB.prepare(
"SELECT response_json FROM action_log WHERE action_type = 'task_update' AND status = 'completed' AND action_key LIKE ? ORDER BY updated_at DESC LIMIT 100",
).bind(prefix).all();
const responses = (result.results || []).map((row) => JSON.parse(row.response_json || "{}"));
return selectPreviouslyLinkedTaskRecord(records, responses);
}

async function findTaskRecordForOutcome(env, project, records, outcome, prUrl) {
let record = selectLinkedTaskRecord(records, prUrl, outcome.number);
let linkedAutomatically = false;
if (!record) {
record = await previouslyLinkedTaskRecord(env, project, records, outcome);
linkedAutomatically = Boolean(record);
}
const githubLogin = String(outcome.author_login || "").trim();
if (!record && githubLogin) {
const identity = await lookupGithubMember(env, project.repository, githubLogin);
const candidate = identity ? selectTaskForGithubIdentity(records, identity.assignee_open_id) : null;
if (candidate && pullRequestStageMatchesTask(valueText(candidate.fields[FIELD_NAMES.title]), outcome)) {
record = candidate;
linkedAutomatically = true;
}
}
return { record, linkedAutomatically };
}

export async function shouldReconcileMergedPullRequest(env, project, records, pullRequest) {
const number = Number(pullRequest?.number);
if (!Number.isInteger(number) || number < 1) return false;
const prUrl = `https://github.com/${project.repository}/pull/${number}`;
const outcome = {
number,
author_login: pullRequest.user?.login || pullRequest.author_login || "",
headRefName: pullRequest.head?.ref || "",
title: pullRequest.title || "",
};
const { record } = await findTaskRecordForOutcome(env, project, records, outcome, prUrl);
if (!record) return false;
const snapshot = taskRecordSnapshot(record);
if (snapshot.status !== "已完成") return true;
return nextStageEligibility(records, snapshot).eligible;
}

function outcomeText(outcome) {
const review = outcome.review || {};
const merged = outcome.merged === true || outcome.merge?.merged === true;
Expand All @@ -535,24 +591,15 @@ export async function applyGithubOutcome(env, outcome) {
const records = await listRecords(env, project);
const prUrl = `https://github.com/${project.repository}/pull/${Number(outcome.number)}`;
const githubLogin = String(outcome.author_login || "").trim();
let record = selectLinkedTaskRecord(records, prUrl, outcome.number);
let linkedAutomatically = false;
if (!record && githubLogin) {
const identity = await lookupGithubMember(env, project.repository, githubLogin);
const candidate = identity ? selectTaskForGithubIdentity(records, identity.assignee_open_id) : null;
if (candidate && pullRequestStageMatchesTask(valueText(candidate.fields[FIELD_NAMES.title]), outcome)) {
await updateRecord(env, project, candidate.record_id, { [FIELD_NAMES.pr]: prUrl });
record = { ...candidate, fields: { ...candidate.fields, [FIELD_NAMES.pr]: prUrl } };
linkedAutomatically = true;
}
}
const { record, linkedAutomatically } = await findTaskRecordForOutcome(env, project, records, outcome, prUrl);
if (!record) return { matched: false, reason: "task_not_linked", project: project.key };
const assignee = assigneeId(record);
const assigneeName = valueText(record.fields[FIELD_NAMES.assignee]) || "成员";
if (githubLogin && assignee) await rememberGithubMember(env, project.repository, githubLogin, assignee, assigneeName);
const merged = outcome.merged === true || outcome.merge?.merged === true;
const completed = merged || outcome.review?.decision === "approve";
const updateFields = { [FIELD_NAMES.status]: completed ? ["已完成"] : ["进行中"], [FIELD_NAMES.verification]: outcomeText(outcome), [FIELD_NAMES.blocked]: null };
if (linkedAutomatically) updateFields[FIELD_NAMES.pr] = prUrl;
if (env.FEISHU_BOT_OPEN_ID) updateFields[FIELD_NAMES.reviewer] = [{ id: env.FEISHU_BOT_OPEN_ID }];
await updateRecord(env, project, record.record_id, updateFields);
let nextTask = null;
Expand Down
17 changes: 17 additions & 0 deletions cloudflare/automation/test/online-routing.test.js
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import test from "node:test";
import assert from "node:assert/strict";
import { mergedPullRequestSyncKey } from "../src/online-runtime.js";
import {
extractPullRequestReference,
actionCanBeReclaimed,
Expand All @@ -10,6 +11,22 @@ import {
scheduledMode,
} from "../src/online.js";

test("合并 PR 的任务同步标识在同一个中国日期内保持稳定", () => {
const pullRequest = { number: 28, merge_commit_sha: "merge-sha" };
assert.equal(
mergedPullRequestSyncKey("XiaoCow666/CodeSense", pullRequest, new Date("2026-09-24T00:00:00Z")),
"merged-pr-task-sync:XiaoCow666/CodeSense#28:merge-sha:2026-09-24",
);
assert.equal(
mergedPullRequestSyncKey("XiaoCow666/CodeSense", pullRequest, new Date("2026-09-24T15:59:59Z")),
"merged-pr-task-sync:XiaoCow666/CodeSense#28:merge-sha:2026-09-24",
);
assert.equal(
mergedPullRequestSyncKey("XiaoCow666/CodeSense", pullRequest, new Date("2026-09-24T16:00:00Z")),
"merged-pr-task-sync:XiaoCow666/CodeSense#28:merge-sha:2026-09-25",
);
});

test("Cloudflare cron keeps reconciliation and the daily report separate", () => {
assert.equal(scheduledMode("*/10 * * * *"), "reconcile");
assert.equal(scheduledMode("0 10 * * *"), "daily-report");
Expand Down
47 changes: 47 additions & 0 deletions cloudflare/automation/test/task-routing.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,56 @@ import {
stageTaskContent,
selectTaskForGithubIdentity,
ensureNextStageTask,
selectPreviouslyLinkedTaskRecord,
shouldReconcileMergedPullRequest,
taskRecordSnapshot,
} from "../src/task-board.js";

test("PR 链接被清空后仍可按已保存的任务关联找回记录", () => {
const record = { record_id: "rec_stage_2", fields: { "GitHub PR / Issue": null } };
assert.equal(
selectPreviouslyLinkedTaskRecord([record], [{ matched: true, record_id: "rec_stage_2" }]),
record,
);
assert.equal(
selectPreviouslyLinkedTaskRecord([record], [
{ matched: true, record_id: "rec_stage_2" },
{ matched: true, record_id: "rec_other" },
]),
null,
);
});

test("合并 PR 只为未完成任务或缺少下一阶段的任务重新排队", async () => {
const stageTwo = {
record_id: "rec_stage_2",
fields: {
"任务名称": "阶段二:修复问题(张诗若)",
"状态": ["进行中"],
"负责人": [{ id: "ou_member", name: "张诗若" }],
"任务模式": ["阶段任务"],
"GitHub PR / Issue": "https://github.com/XiaoCow666/CodeSense/pull/28",
},
};
const project = { repository: "XiaoCow666/CodeSense" };
const pullRequest = { number: 28, merged: true, user: { login: "zhang" } };
assert.equal(await shouldReconcileMergedPullRequest({}, project, [stageTwo], pullRequest), true);

const completedStageTwo = { ...stageTwo, fields: { ...stageTwo.fields, "状态": ["已完成"] } };
assert.equal(await shouldReconcileMergedPullRequest({}, project, [completedStageTwo], pullRequest), true);
const stageThree = {
record_id: "rec_stage_3",
fields: {
"任务名称": "阶段三:追到问题根因(张诗若)",
"状态": ["待开始"],
"负责人": [{ id: "ou_member", name: "张诗若" }],
"任务模式": ["阶段任务"],
"GitHub PR / Issue": null,
},
};
assert.equal(await shouldReconcileMergedPullRequest({}, project, [completedStageTwo, stageThree], pullRequest), false);
});

test("stage parsing supports Arabic and Chinese stage names", () => {
assert.equal(parseStageNumber("阶段 8:真实问题"), 8);
assert.equal(parseStageNumber("阶段八:真实问题"), 8);
Expand Down
Loading