feat: add distribution release and recovery controls

2026-08-12 11:12:23 +08:00
parent ddad4b7659
commit 7ef150e08b
16 changed files with 717 additions and 33 deletions
--- a/lib/collection-service.ts
+++ b/lib/collection-service.ts
@@ -60,13 +60,23 @@ function shanghaiHourFromTimestamp(timestamp: number) {
  );
 }

+export function isCollectionScheduleDue(
+  scheduledDate: string,
+  timestamp: number,
+) {
+  const currentDate = shanghaiDateFromTimestamp(timestamp);
+  const currentHour = shanghaiHourFromTimestamp(timestamp);
+  return (
+    scheduledDate < currentDate ||
+    (scheduledDate === currentDate && currentHour >= 9)
+  );
+}
+
 function dueSchedules(
  startDate: string,
  days: number[],
  timestamp: number,
 ) {
-  const currentDate = shanghaiDateFromTimestamp(timestamp);
-  const currentHour = shanghaiHourFromTimestamp(timestamp);
  return days
    .map((scheduleDay) => ({
      scheduleDay,
@@ -78,8 +88,7 @@ function dueSchedules(
      ): value is { scheduleDay: number; scheduledDate: string } =>
        Boolean(
          value.scheduledDate &&
-            (value.scheduledDate < currentDate ||
-              (value.scheduledDate === currentDate && currentHour >= 10)),
+            isCollectionScheduleDue(value.scheduledDate, timestamp),
        ),
    );
 }
@@ -131,8 +140,8 @@ export async function createCollectionRunTasks(
            distribution.id,
            scheduledDate,
            scheduleDay,
-            `${scheduledDate}T10:00:00+08:00`,
-            `等待第${scheduleDay}天 10:00自动采集`,
+            `${scheduledDate}T09:00:00+08:00`,
+            `等待第${scheduleDay}天 09:00自动采集`,
          ),
      );
    }
@@ -162,7 +171,7 @@ export async function collectDistributionMetrics(
  if (!current) throw new Error("分发记录不存在");
  if (!current.publish_url) throw new Error("笔记尚未回填发布链接");

-  const scheduledAt = `${scheduledDate}T10:00:00+08:00`;
+  const scheduledAt = `${scheduledDate}T09:00:00+08:00`;
  const runId = uid("run");
  await db
    .prepare(
@@ -231,7 +240,7 @@ export async function collectDistributionMetrics(
    const dayWeight = scheduleDay ?? 1;
    const successDescription =
      source === "automatic"
-        ? `成功 · 第${dayWeight}天 10:00自动采集`
+        ? `成功 · 第${dayWeight}天 09:00自动采集`
        : source === "catchup"
          ? `成功 · 第${dayWeight}天自动追采`
          : "成功 · 手动采集";
--- a/lib/database.ts
+++ b/lib/database.ts
@@ -155,6 +155,23 @@ export class DatabaseClient {
      connection.release();
    }
  }
+
+  async transaction<T>(operation: (database: DatabaseClient) => Promise<T>) {
+    const pool = getPool();
+    const connection = await pool.getConnection();
+    try {
+      await connection.beginTransaction();
+      const transaction = new DatabaseClient(connection);
+      const result = await operation(transaction);
+      await connection.commit();
+      return result;
+    } catch (error) {
+      await connection.rollback();
+      throw error;
+    } finally {
+      connection.release();
+    }
+  }
 }

 declare global {
--- a/lib/distribution-release-service.ts
+++ b/lib/distribution-release-service.ts
@@ -0,0 +1,153 @@
+import type { DatabaseClient } from "./database";
+
+type ReleasableDistribution = {
+  id: string;
+  task_id: string;
+  task_type: string;
+  content_id: string;
+  content_title: string;
+  partner_id: string;
+  partner_name: string;
+  claim_id: string | null;
+  delegation_bundle_id: string | null;
+  publish_url: string | null;
+  result_submitted_at: string | null;
+};
+
+export class DistributionReleaseError extends Error {
+  readonly status: number;
+
+  constructor(message: string, status: number) {
+    super(message);
+    this.name = "DistributionReleaseError";
+    this.status = status;
+  }
+}
+
+export function distributionReleaseBlockReason(input: {
+  publishUrl?: string | null;
+  resultSubmittedAt?: string | null;
+  taskType?: string | null;
+}) {
+  if (input.publishUrl) return "已回填发布链接的笔记不能释放";
+  if (input.resultSubmittedAt) return "已提交结果截图的任务不能释放";
+  return null;
+}
+
+export async function releaseUnfinishedDistribution(
+  database: DatabaseClient,
+  distributionId: string,
+) {
+  if (!distributionId) {
+    throw new DistributionReleaseError("请选择需要释放的领取记录", 400);
+  }
+
+  return database.transaction(async (db) => {
+    const distribution = await db
+      .prepare(
+        `SELECT
+          d.id,
+          d.task_id,
+          t.task_type,
+          d.content_id,
+          c.title AS content_title,
+          d.partner_id,
+          p.name AS partner_name,
+          d.claim_id,
+          d.delegation_bundle_id,
+          d.publish_url,
+          d.result_submitted_at
+         FROM distributions d
+         JOIN tasks t ON t.id = d.task_id
+         JOIN contents c ON c.id = d.content_id
+         JOIN partners p ON p.id = d.partner_id
+         WHERE d.id = ?
+         FOR UPDATE`,
+      )
+      .bind(distributionId)
+      .first<ReleasableDistribution>();
+
+    if (!distribution) {
+      throw new DistributionReleaseError("领取记录不存在或已被释放", 404);
+    }
+    const blocked = distributionReleaseBlockReason({
+      publishUrl: distribution.publish_url,
+      resultSubmittedAt: distribution.result_submitted_at,
+      taskType: distribution.task_type,
+    });
+    if (blocked) throw new DistributionReleaseError(blocked, 409);
+
+    await db
+      .prepare("DELETE FROM collection_runs WHERE distribution_id = ?")
+      .bind(distribution.id)
+      .run();
+    await db
+      .prepare("DELETE FROM distributions WHERE id = ?")
+      .bind(distribution.id)
+      .run();
+    await db
+      .prepare("UPDATE contents SET status = 'available' WHERE id = ?")
+      .bind(distribution.content_id)
+      .run();
+    await db
+      .prepare(
+        `UPDATE tasks
+         SET claimed_quantity = GREATEST(claimed_quantity - 1, 0)
+         WHERE id = ?`,
+      )
+      .bind(distribution.task_id)
+      .run();
+    await db
+      .prepare(
+        `UPDATE partners
+         SET claimed_total = GREATEST(claimed_total - 1, 0)
+         WHERE id = ?`,
+      )
+      .bind(distribution.partner_id)
+      .run();
+
+    if (distribution.delegation_bundle_id) {
+      await db
+        .prepare(
+          `UPDATE delegation_bundles
+           SET quantity = GREATEST(quantity - 1, 0),
+               status = CASE WHEN quantity <= 1 THEN 'revoked' ELSE status END,
+               revoked_at = CASE WHEN quantity <= 1 THEN CURRENT_TIMESTAMP ELSE revoked_at END,
+               updated_at = CURRENT_TIMESTAMP
+           WHERE id = ?`,
+        )
+        .bind(distribution.delegation_bundle_id)
+        .run();
+    }
+
+    if (distribution.claim_id) {
+      await db
+        .prepare(
+          `UPDATE claims
+           SET quantity = GREATEST(quantity - 1, 0)
+           WHERE id = ?`,
+        )
+        .bind(distribution.claim_id)
+        .run();
+      await db
+        .prepare(
+          `DELETE FROM claims
+           WHERE id = ?
+             AND quantity <= 0
+             AND NOT EXISTS (
+               SELECT 1 FROM distributions WHERE claim_id = ?
+             )`,
+        )
+        .bind(distribution.claim_id, distribution.claim_id)
+        .run();
+    }
+
+    return {
+      distributionId: distribution.id,
+      taskId: distribution.task_id,
+      contentId: distribution.content_id,
+      contentTitle: distribution.content_title,
+      partnerName: distribution.partner_name,
+    };
+  });
+}
--- a/lib/mcp-operations.ts
+++ b/lib/mcp-operations.ts
@@ -284,7 +284,7 @@ export async function setCollectionPlan(
       collection_status_description = CASE WHEN latest_likes IS NULL THEN ? ELSE collection_status_description END,
       updated_at = CURRENT_TIMESTAMP
       WHERE task_id = ? AND COALESCE(publish_url, '') != ''`,
-    ).bind(`已安排${normalizedDays.length}个采集日，每日10:00执行`, taskId),
+    ).bind(`已安排${normalizedDays.length}个采集日，每日09:00执行`, taskId),
  ]);
  const created = await createCollectionRunTasks(db, taskId, startDate, normalizedDays);
  const catchup = await runDueScheduledCollections(
--- a/lib/mcp-tools.ts
+++ b/lib/mcp-tools.ts
@@ -127,7 +127,7 @@ export function registerMcpOperationTools(server: McpServer, options: Options) {
    "collection_plan_set",
    {
      title: "设置自动采集计划",
-      description: "为任务设置开始日期及第1至第7天的自动采集日，系统在北京时间10:00执行。",
+      description: "为任务设置开始日期及第1至第7天的自动采集日，系统在北京时间09:00执行。",
      inputSchema: z.object({
        task_id: z.string().min(1),
        start_date: z.string().regex(/^\d{4}-\d{2}-\d{2}$/).describe("北京时间开始日期 YYYY-MM-DD"),
--- a/lib/scheduler.ts
+++ b/lib/scheduler.ts
@@ -33,11 +33,11 @@ export function startScheduler() {
  if (!isEnabled(getRuntimeEnv().ENABLE_SCHEDULER, true)) return;
  if (globalThis.__kocLoopScheduler) return;
  globalThis.__kocLoopScheduler = cron.schedule(
-    "0 10 * * *",
+    "0 9 * * *",
    () => void runDailyJob().catch((error) => {
      console.error("[KOC LOOP] daily scheduler failed", error);
    }),
    { timezone: "Asia/Shanghai", noOverlap: true },
  );
-  console.info("[KOC LOOP] scheduler enabled at 10:00 Asia/Shanghai");
+  console.info("[KOC LOOP] scheduler enabled at 09:00 Asia/Shanghai");
 }
--- a/lib/sort-utils.ts
+++ b/lib/sort-utils.ts
@@ -0,0 +1,26 @@
+export type SortDirection = "asc" | "desc";
+
+export function sortWithNullsLast<T>(
+  items: T[],
+  valueFor: (item: T) => number | null,
+  direction: SortDirection,
+) {
+  return items
+    .map((item, index) => ({ item, index }))
+    .sort((left, right) => {
+      const leftValue = valueFor(left.item);
+      const rightValue = valueFor(right.item);
+      if (leftValue === null && rightValue === null) {
+        return left.index - right.index;
+      }
+      if (leftValue === null) return 1;
+      if (rightValue === null) return -1;
+      const compared = leftValue - rightValue;
+      return compared === 0
+        ? left.index - right.index
+        : direction === "asc"
+          ? compared
+          : -compared;
+    })
+    .map(({ item }) => item);
+}