优化任务状态投影和部署文档

2026-05-28 11:12:40 +08:00
parent 9709643583
commit ccc350cfea
15 changed files with 759 additions and 55 deletions
@@ -0,0 +1,111 @@
+import { query } from '../db/client.js'
+import { maskCode } from '../utils/masking.js'
+import { parseTaskContext } from '../utils/task-json.js'
+
+type QueryExecutor = (text: string, params?: unknown[]) => Promise<{ rows: unknown[] }>
+type JsonRecord = Record<string, any>
+
+export type KuaishouCloudTaskStateSyncInput = {
+  taskId: number | string
+  executorKey?: unknown
+  contextJson?: unknown
+  updatedAt?: unknown
+}
+
+export async function syncKuaishouCloudTaskStateForTask(
+  input: KuaishouCloudTaskStateSyncInput,
+  executor: QueryExecutor = query,
+): Promise<void> {
+  const taskId = Number(input.taskId || 0)
+  if (!Number.isFinite(taskId) || taskId <= 0) {
+    return
+  }
+
+  if (String(input.executorKey || '').trim() !== 'kuaishou_ct_assisted') {
+    await executor('DELETE FROM kuaishou_cloud_task_states WHERE task_id = $1', [taskId])
+    return
+  }
+
+  const flow = resolveKuaishouCloudFlow(input.contextJson)
+  if (!flow) {
+    await executor('DELETE FROM kuaishou_cloud_task_states WHERE task_id = $1', [taskId])
+    return
+  }
+
+  const ticket = asRecord(flow.ticket)
+  const binding = asRecord(flow.binding)
+  const role = asRecord(flow.role)
+  const dispatch = asRecord(flow.dispatch)
+  const consume = asRecord(flow.consume)
+  const roleName = pickFirstNonEmpty([role.name, binding.roleName])
+  const roleId = pickFirstNonEmpty([role.rid, binding.roleId])
+  const updatedAt = String(input.updatedAt || '').trim() || new Date().toISOString()
+
+  await executor(
+    `
+      INSERT INTO kuaishou_cloud_task_states (
+        task_id,
+        ticket_status,
+        ticket_code_masked,
+        bind_url,
+        vn_phone,
+        role_name,
+        role_id,
+        dispatch_status,
+        consume_status,
+        source_key,
+        updated_at
+      ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
+      ON CONFLICT (task_id) DO UPDATE
+      SET
+        ticket_status = EXCLUDED.ticket_status,
+        ticket_code_masked = EXCLUDED.ticket_code_masked,
+        bind_url = EXCLUDED.bind_url,
+        vn_phone = EXCLUDED.vn_phone,
+        role_name = EXCLUDED.role_name,
+        role_id = EXCLUDED.role_id,
+        dispatch_status = EXCLUDED.dispatch_status,
+        consume_status = EXCLUDED.consume_status,
+        source_key = EXCLUDED.source_key,
+        updated_at = EXCLUDED.updated_at
+    `,
+    [
+      taskId,
+      normalizeStatus(ticket.status),
+      maskCode(ticket.code),
+      String(binding.bindUrl || '').trim(),
+      String(binding.vnPhone || '').trim(),
+      roleName,
+      roleId,
+      normalizeStatus(dispatch.status),
+      normalizeStatus(consume.status),
+      String(binding.resolvedSourceKey || '').trim(),
+      updatedAt,
+    ],
+  )
+}
+
+function resolveKuaishouCloudFlow(contextJson: unknown): JsonRecord | null {
+  const context = parseTaskContext({ context_json: contextJson })
+  const flow = asRecord(context.kuaishouCloudFulfillment)
+  return Object.keys(flow).length > 0 ? flow : null
+}
+
+function asRecord(value: unknown): JsonRecord {
+  return value && typeof value === 'object' && !Array.isArray(value) ? value as JsonRecord : {}
+}
+
+function normalizeStatus(value: unknown): string {
+  return String(value || 'pending').trim() || 'pending'
+}
+
+function pickFirstNonEmpty(values: unknown[]): string {
+  for (const value of values) {
+    const normalized = String(value || '').trim()
+    if (normalized) {
+      return normalized
+    }
+  }
+
+  return ''
+}
@@ -1,6 +1,7 @@
 import { query, withTransaction } from '../db/client.js'
 import type { PoolClient } from 'pg'

+import { syncKuaishouCloudTaskStateForTask } from './kuaishou-cloud-task-state-repo.js'
 import type {
  TaskCreateInput,
  TaskListQueryInput,
@@ -42,13 +43,24 @@ const TASK_FIELDS = `
  ctx.partition_name,
  ctx.screenshot_path,
  ctx.artifacts_json,
-  ctx.state_json
+  ctx.state_json,
+  kcts.ticket_status AS kuaishou_ticket_status,
+  kcts.ticket_code_masked AS kuaishou_ticket_code_masked,
+  kcts.bind_url AS kuaishou_bind_url,
+  kcts.vn_phone AS kuaishou_vn_phone,
+  kcts.role_name AS kuaishou_role_name,
+  kcts.role_id AS kuaishou_role_id,
+  kcts.dispatch_status AS kuaishou_dispatch_status,
+  kcts.consume_status AS kuaishou_consume_status,
+  kcts.source_key AS kuaishou_source_key,
+  kcts.updated_at AS kuaishou_state_updated_at
 `

 const TASK_JOINS = `
  FROM fulfillment_tasks ft
  LEFT JOIN claim_tokens ct ON ct.task_id = ft.id
  LEFT JOIN task_runtime_contexts ctx ON ctx.task_id = ft.id
+  LEFT JOIN kuaishou_cloud_task_states kcts ON kcts.task_id = ft.id
 `

 function buildTaskSelect(extraFields = ''): string {
@@ -138,6 +150,13 @@ export async function createTask(input: TaskCreateInput): Promise<TaskRow | null
      await upsertTaskRuntimeContextWithClient(client, taskId, input.runtimeContext, input.createdAt)
    }

+    await syncKuaishouCloudTaskStateForTask({
+      taskId,
+      executorKey: input.executorKey,
+      contextJson: input.contextJson || '{}',
+      updatedAt: input.updatedAt,
+    }, client.query.bind(client) as TaskQueryExecutor)
+
    return getTaskByIdWithExecutor(client.query.bind(client) as TaskQueryExecutor, taskId)
  })
 }
@@ -210,6 +229,13 @@ export async function updateTask(taskId: number | string, patch: TaskUpdatePatch
      await upsertTaskRuntimeContextWithClient(client, Number(taskId), runtimeContextPatch, patch.updated_at || current.updated_at)
    }

+    await syncKuaishouCloudTaskStateForTask({
+      taskId,
+      executorKey: next.executor_key,
+      contextJson: next.context_json,
+      updatedAt: next.updated_at,
+    }, client.query.bind(client) as TaskQueryExecutor)
+
    return getTaskByIdWithExecutor(client.query.bind(client) as TaskQueryExecutor, taskId)
  })
 }
@@ -220,47 +246,62 @@ export async function updateTaskStatusIfCurrent(
  patch: TaskUpdatePatch,
 ): Promise<TaskRow | null> {
  const now = String(patch.updated_at || '').trim()
-  const result = await query<{ id: number }>(
-    `
-      UPDATE fulfillment_tasks
-      SET
-        task_status = $1,
-        delivery_status = COALESCE($2, delivery_status),
-        result_code = COALESCE($3, result_code),
-        result_message = COALESCE($4, result_message),
-        user_action_status = COALESCE($5, user_action_status),
-        last_error = COALESCE($6, last_error),
-        context_json = COALESCE($7::jsonb, context_json),
-        redeemed_at = COALESCE($8, redeemed_at),
-        updated_at = COALESCE($9, updated_at)
-      WHERE id = $10
-        AND task_status = $11
-      RETURNING id
-    `,
-    [
-      patch.task_status,
-      patch.delivery_status ?? null,
-      patch.result_code ?? null,
-      patch.result_message ?? null,
-      patch.user_action_status ?? null,
-      patch.last_error ?? null,
-      typeof patch.context_json === 'undefined'
-        ? null
-        : typeof patch.context_json === 'string'
-          ? patch.context_json
-          : JSON.stringify(patch.context_json || {}),
-      patch.redeemed_at ?? null,
-      now || null,
-      Number(taskId),
-      currentStatus,
-    ],
-  )
+  return withTransaction(async (client) => {
+    const result = await client.query<{
+      id: number
+      executor_key: string
+      context_json: string | Record<string, unknown>
+      updated_at: string
+    }>(
+      `
+        UPDATE fulfillment_tasks
+        SET
+          task_status = $1,
+          delivery_status = COALESCE($2, delivery_status),
+          result_code = COALESCE($3, result_code),
+          result_message = COALESCE($4, result_message),
+          user_action_status = COALESCE($5, user_action_status),
+          last_error = COALESCE($6, last_error),
+          context_json = COALESCE($7::jsonb, context_json),
+          redeemed_at = COALESCE($8, redeemed_at),
+          updated_at = COALESCE($9, updated_at)
+        WHERE id = $10
+          AND task_status = $11
+        RETURNING id, executor_key, context_json, updated_at
+      `,
+      [
+        patch.task_status,
+        patch.delivery_status ?? null,
+        patch.result_code ?? null,
+        patch.result_message ?? null,
+        patch.user_action_status ?? null,
+        patch.last_error ?? null,
+        typeof patch.context_json === 'undefined'
+          ? null
+          : typeof patch.context_json === 'string'
+            ? patch.context_json
+            : JSON.stringify(patch.context_json || {}),
+        patch.redeemed_at ?? null,
+        now || null,
+        Number(taskId),
+        currentStatus,
+      ],
+    )

-  if (!result.rows[0]) {
-    return null
-  }
+    const updated = result.rows[0]
+    if (!updated) {
+      return null
+    }

-  return getTaskById(taskId)
+    await syncKuaishouCloudTaskStateForTask({
+      taskId: updated.id,
+      executorKey: updated.executor_key,
+      contextJson: updated.context_json,
+      updatedAt: updated.updated_at,
+    }, client.query.bind(client) as TaskQueryExecutor)
+
+    return getTaskByIdWithExecutor(client.query.bind(client) as TaskQueryExecutor, taskId)
+  })
 }

 export async function getTaskById(taskId: number | string): Promise<TaskRow | null> {
@@ -315,7 +356,7 @@ export async function listTasks({

  if (roleId) {
    params.push(`%${roleId}%`)
-    filters.push(`ctx.role_id ILIKE $${params.length}`)
+    filters.push(`COALESCE(NULLIF(kcts.role_id, ''), ctx.role_id) ILIKE $${params.length}`)
  }

  if (dateFrom) {
@@ -335,6 +376,7 @@ export async function listTasks({
      FROM fulfillment_tasks ft
      LEFT JOIN order_items oi ON oi.id = ft.order_item_id
      LEFT JOIN task_runtime_contexts ctx ON ctx.task_id = ft.id
+      LEFT JOIN kuaishou_cloud_task_states kcts ON kcts.task_id = ft.id
      ${whereClause}
    `,
    params,