彻底重构-1

apps/backend/src/repositories/task-repo.js

@@ -1,152 +1,227 @@
-import { getDb } from '../db/client.js'
+import { query, withTransaction } from '../db/client.js'
 
-export function listTasksByOrderId(orderId) {
-  return getDb().prepare(`
-    SELECT *
-    FROM delivery_tasks
-    WHERE order_id = ?
-    ORDER BY id ASC
-  `).all(orderId)
+const TASK_FIELDS = `
+  ft.*,
+  ct.id AS claim_token_id,
+  ct.token AS claim_token,
+  ct.status AS claim_token_status,
+  ct.expired_at AS claim_token_expired_at,
+  ctx.browser_session_id,
+  ctx.login_type,
+  ctx.nickname,
+  ctx.role_id,
+  ctx.role_name,
+  ctx.area,
+  ctx.partition_name,
+  ctx.screenshot_path,
+  ctx.artifacts_json,
+  ctx.state_json,
+  inv.inventory_item_id AS reserved_cdk_id,
+  inv.display_value AS cdk_code,
+  inv.credential_type AS cdk_credential_type,
+  inv.binding_status AS inventory_binding_status
+`
+
+const TASK_JOINS = `
+  FROM fulfillment_tasks ft
+  LEFT JOIN claim_tokens ct ON ct.task_id = ft.id
+  LEFT JOIN tencent_browser_contexts ctx ON ctx.task_id = ft.id
+  LEFT JOIN LATERAL (
+    SELECT
+      tib.inventory_item_id,
+      tib.binding_status,
+      ii.display_value,
+      ii.credential_type
+    FROM task_inventory_bindings tib
+    JOIN inventory_items ii ON ii.id = tib.inventory_item_id
+    WHERE tib.task_id = ft.id AND tib.binding_status IN ('reserved', 'consumed')
+    ORDER BY tib.id ASC
+    LIMIT 1
+  ) inv ON TRUE
+`
+
+function buildTaskSelect(extraFields = '') {
+  const normalizedExtra = String(extraFields || '').trim()
+  const fieldSql = normalizedExtra ? `${TASK_FIELDS}, ${normalizedExtra}` : TASK_FIELDS
+  return `SELECT ${fieldSql} ${TASK_JOINS}`
 }
 
-export function createTask(input) {
-  const result = getDb().prepare(`
-    INSERT INTO delivery_tasks (
-      order_id,
-      order_item_id,
-      platform_order_id,
-      task_no,
-      task_status,
-      login_type,
-      claim_token_id,
-      reserved_cdk_id,
-      browser_session_id,
-      nickname,
-      role_id,
-      role_name,
-      area,
-      partition_name,
-      result_code,
-      result_message,
-      screenshot_path,
-      artifacts_json,
-      last_error,
-      retry_count,
-      expires_at,
-      claimed_at,
-      role_confirmed_at,
-      redeemed_at,
-      created_at,
-      updated_at
-    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-  `).run(
-    input.orderId,
-    input.orderItemId,
-    input.platformOrderId,
-    input.taskNo,
-    input.taskStatus,
-    input.loginType,
-    input.claimTokenId,
-    input.reservedCdkId,
-    input.browserSessionId,
-    input.nickname,
-    input.roleId,
-    input.roleName,
-    input.area,
-    input.partitionName,
-    input.resultCode,
-    input.resultMessage,
-    input.screenshotPath,
-    input.artifactsJson,
-    input.lastError,
-    input.retryCount,
-    input.expiresAt,
-    input.claimedAt,
-    input.roleConfirmedAt,
-    input.redeemedAt,
-    input.createdAt,
-    input.updatedAt,
+export async function listTasksByOrderId(orderId) {
+  const result = await query(
+    `${buildTaskSelect()}
+     WHERE ft.order_id = $1
+     ORDER BY ft.id ASC`,
+    [Number(orderId)],
   )
-
-  return getTaskById(Number(result.lastInsertRowid))
+  return result.rows
 }
 
-export function updateTask(taskId, patch) {
-  const current = getTaskById(taskId)
+export async function createTask(input) {
+  return withTransaction(async (client) => {
+    const taskResult = await client.query(
+      `
+        INSERT INTO fulfillment_tasks (
+          order_id,
+          order_item_id,
+          unit_index,
+          task_no,
+          provider,
+          platform,
+          shop_id,
+          shop_name,
+          platform_order_id,
+          profile_id,
+          executor_key,
+          task_status,
+          inventory_status,
+          delivery_status,
+          result_code,
+          result_message,
+          claim_token,
+          claim_expires_at,
+          automation_mode,
+          requires_claim,
+          user_action_status,
+          attempt_count,
+          last_error,
+          context_json,
+          created_at,
+          updated_at
+        ) VALUES (
+          $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
+          $11, $12, $13, $14, $15, $16, $17, $18, $19,
+          $20, $21, $22, $23, $24::jsonb, $25, $26
+        )
+        RETURNING id
+      `,
+      [
+        input.orderId,
+        input.orderItemId,
+        input.unitIndex,
+        input.taskNo,
+        input.provider,
+        input.platform,
+        input.shopId || '',
+        input.shopName || '',
+        input.platformOrderId,
+        input.profileId,
+        input.executorKey,
+        input.taskStatus || 'pending_payment',
+        input.inventoryStatus || 'pending',
+        input.deliveryStatus || 'pending',
+        input.resultCode || '',
+        input.resultMessage || '',
+        input.claimToken || '',
+        input.claimExpiresAt || null,
+        input.automationMode || 'manual',
+        Boolean(input.requiresClaim),
+        input.userActionStatus || 'not_required',
+        input.attemptCount || 0,
+        input.lastError || '',
+        input.contextJson || '{}',
+        input.createdAt,
+        input.updatedAt,
+      ],
+    )
 
+    const taskId = Number(taskResult.rows[0]?.id || 0)
+    if (input.tencentContext) {
+      await upsertTencentBrowserContextWithClient(client, taskId, input.tencentContext, input.createdAt)
+    }
+
+    return getTaskById(taskId)
+  })
+}
+
+export async function updateTask(taskId, patch) {
+  const current = await getTaskById(taskId)
   if (!current) {
     return null
   }
 
-  const next = { ...current, ...patch }
+  return withTransaction(async (client) => {
+    const next = { ...current, ...patch }
 
-  getDb().prepare(`
-    UPDATE delivery_tasks
-    SET
-      task_status = ?,
-      login_type = ?,
-      claim_token_id = ?,
-      reserved_cdk_id = ?,
-      browser_session_id = ?,
-      nickname = ?,
-      role_id = ?,
-      role_name = ?,
-      area = ?,
-      partition_name = ?,
-      result_code = ?,
-      result_message = ?,
-      screenshot_path = ?,
-      artifacts_json = ?,
-      last_error = ?,
-      retry_count = ?,
-      expires_at = ?,
-      claimed_at = ?,
-      role_confirmed_at = ?,
-      redeemed_at = ?,
-      updated_at = ?
-    WHERE id = ?
-  `).run(
-    next.task_status,
-    next.login_type,
-    next.claim_token_id,
-    next.reserved_cdk_id,
-    next.browser_session_id,
-    next.nickname,
-    next.role_id,
-    next.role_name,
-    next.area,
-    next.partition_name,
-    next.result_code,
-    next.result_message,
-    next.screenshot_path,
-    next.artifacts_json,
-    next.last_error,
-    next.retry_count,
-    next.expires_at,
-    next.claimed_at,
-    next.role_confirmed_at,
-    next.redeemed_at,
-    next.updated_at,
-    taskId,
+    await client.query(
+      `
+        UPDATE fulfillment_tasks
+        SET
+          task_status = $1,
+          inventory_status = $2,
+          delivery_status = $3,
+          result_code = $4,
+          result_message = $5,
+          claim_token = $6,
+          claim_expires_at = $7,
+          automation_mode = $8,
+          requires_claim = $9,
+          user_action_status = $10,
+          attempt_count = $11,
+          last_error = $12,
+          context_json = $13::jsonb,
+          updated_at = $14
+        WHERE id = $15
+      `,
+      [
+        next.task_status,
+        next.inventory_status,
+        next.delivery_status,
+        next.result_code,
+        next.result_message,
+        next.claim_token || '',
+        next.claim_expires_at || null,
+        next.automation_mode || 'manual',
+        Boolean(next.requires_claim),
+        next.user_action_status || 'not_required',
+        next.attempt_count || 0,
+        next.last_error || '',
+        next.context_json || '{}',
+        next.updated_at,
+        Number(taskId),
+      ],
+    )
+
+    if (containsTencentPatch(patch)) {
+      await upsertTencentBrowserContextWithClient(client, Number(taskId), {
+        browserSessionId: patch.browser_session_id,
+        loginType: patch.login_type,
+        nickname: patch.nickname,
+        roleId: patch.role_id,
+        roleName: patch.role_name,
+        area: patch.area,
+        partitionName: patch.partition_name,
+        screenshotPath: patch.screenshot_path,
+        artifactsJson: patch.artifacts_json,
+        stateJson: patch.state_json,
+      }, patch.updated_at || current.updated_at)
+    }
+
+    return getTaskById(taskId)
+  })
+}
+
+export async function getTaskById(taskId) {
+  const result = await query(
+    `${buildTaskSelect()}
+     WHERE ft.id = $1
+     LIMIT 1`,
+    [Number(taskId)],
   )
-
-  return getTaskById(taskId)
+  return result.rows[0] || null
 }
 
-export function getTaskById(taskId) {
-  return getDb().prepare('SELECT * FROM delivery_tasks WHERE id = ? LIMIT 1').get(taskId) || null
+export async function findTaskByClaimTokenId(claimTokenId) {
+  const result = await query(
+    `${buildTaskSelect()}
+     JOIN claim_tokens ctf ON ctf.task_id = ft.id
+     WHERE ctf.id = $1
+     LIMIT 1`,
+    [Number(claimTokenId)],
+  )
+  return result.rows[0] || null
 }
 
-export function findTaskByClaimTokenId(claimTokenId) {
-  return getDb().prepare(`
-    SELECT *
-    FROM delivery_tasks
-    WHERE claim_token_id = ?
-    LIMIT 1
-  `).get(claimTokenId) || null
-}
-
-export function listTasks({
+export async function listTasks({
   page = 1,
   pageSize = 20,
   status = '',
@@ -162,67 +237,173 @@ export function listTasks({
   const params = []
 
   if (status) {
-    filters.push('dt.task_status = ?')
     params.push(status)
+    filters.push(`ft.task_status = $${params.length}`)
   }
 
   if (platformOrderId) {
-    filters.push('dt.platform_order_id LIKE ?')
     params.push(`%${platformOrderId}%`)
+    filters.push(`ft.platform_order_id ILIKE $${params.length}`)
   }
 
   if (taskNo) {
-    filters.push('dt.task_no LIKE ?')
     params.push(`%${taskNo}%`)
+    filters.push(`ft.task_no ILIKE $${params.length}`)
   }
 
   if (skuCode) {
-    filters.push('oi.sku_code = ?')
     params.push(skuCode)
+    filters.push(`oi.sku_code = $${params.length}`)
   }
 
   if (roleId) {
-    filters.push('dt.role_id LIKE ?')
     params.push(`%${roleId}%`)
+    filters.push(`ctx.role_id ILIKE $${params.length}`)
   }
 
   if (dateFrom) {
-    filters.push('dt.created_at >= ?')
     params.push(dateFrom)
+    filters.push(`ft.created_at >= $${params.length}`)
   }
 
   if (dateTo) {
-    filters.push('dt.created_at <= ?')
     params.push(dateTo)
+    filters.push(`ft.created_at <= $${params.length}`)
   }
 
   const whereClause = filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : ''
-  const db = getDb()
-  const totalRow = db.prepare(`
-    SELECT COUNT(*) AS total
-    FROM delivery_tasks dt
-    LEFT JOIN order_items oi ON oi.id = dt.order_item_id
-    ${whereClause}
-  `).get(...params)
+  const totalResult = await query(
+    `
+      SELECT COUNT(*)::int AS total
+      FROM fulfillment_tasks ft
+      LEFT JOIN order_items oi ON oi.id = ft.order_item_id
+      LEFT JOIN tencent_browser_contexts ctx ON ctx.task_id = ft.id
+      ${whereClause}
+    `,
+    params,
+  )
 
-  const items = db.prepare(`
-    SELECT
-      dt.*,
-      oi.sku_code,
-      oi.sku_name,
-      ci.cdk_code,
-      ct.token AS claim_token
-    FROM delivery_tasks dt
-    LEFT JOIN order_items oi ON oi.id = dt.order_item_id
-    LEFT JOIN cdk_inventory ci ON ci.id = dt.reserved_cdk_id
-    LEFT JOIN claim_tokens ct ON ct.id = dt.claim_token_id
-    ${whereClause}
-    ORDER BY dt.id DESC
-    LIMIT ? OFFSET ?
-  `).all(...params, pageSize, offset)
+  params.push(pageSize)
+  params.push(offset)
+  const itemsResult = await query(
+    `${buildTaskSelect('oi.sku_code, oi.sku_name, oi.quantity')}
+     LEFT JOIN order_items oi ON oi.id = ft.order_item_id
+     ${whereClause}
+     ORDER BY ft.id DESC
+     LIMIT $${params.length - 1} OFFSET $${params.length}`,
+    params,
+  )
 
   return {
-    items,
-    total: Number(totalRow?.total || 0),
+    items: itemsResult.rows,
+    total: Number(totalResult.rows[0]?.total || 0),
   }
 }
+
+async function upsertTencentBrowserContextWithClient(client, taskId, patch = {}, timestamp) {
+  const currentResult = await client.query(
+    'SELECT * FROM tencent_browser_contexts WHERE task_id = $1 LIMIT 1',
+    [Number(taskId)],
+  )
+  const current = currentResult.rows[0] || null
+
+  const next = {
+    browser_session_id: patch.browserSessionId ?? current?.browser_session_id ?? '',
+    login_type: patch.loginType ?? current?.login_type ?? '',
+    nickname: patch.nickname ?? current?.nickname ?? '',
+    role_id: patch.roleId ?? current?.role_id ?? '',
+    role_name: patch.roleName ?? current?.role_name ?? '',
+    area: patch.area ?? current?.area ?? '',
+    partition_name: patch.partitionName ?? current?.partition_name ?? '',
+    screenshot_path: patch.screenshotPath ?? current?.screenshot_path ?? '',
+    artifacts_json: patch.artifactsJson ?? current?.artifacts_json ?? '{}',
+    state_json: patch.stateJson ?? current?.state_json ?? '{}',
+    updated_at: timestamp,
+  }
+
+  if (!current) {
+    await client.query(
+      `
+        INSERT INTO tencent_browser_contexts (
+          task_id,
+          browser_session_id,
+          login_type,
+          nickname,
+          role_id,
+          role_name,
+          area,
+          partition_name,
+          screenshot_path,
+          artifacts_json,
+          state_json,
+          created_at,
+          updated_at
+        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11::jsonb, $12, $13)
+      `,
+      [
+        Number(taskId),
+        next.browser_session_id,
+        next.login_type,
+        next.nickname,
+        next.role_id,
+        next.role_name,
+        next.area,
+        next.partition_name,
+        next.screenshot_path,
+        typeof next.artifacts_json === 'string' ? next.artifacts_json : JSON.stringify(next.artifacts_json || {}),
+        typeof next.state_json === 'string' ? next.state_json : JSON.stringify(next.state_json || {}),
+        timestamp,
+        timestamp,
+      ],
+    )
+    return
+  }
+
+  await client.query(
+    `
+      UPDATE tencent_browser_contexts
+      SET
+        browser_session_id = $1,
+        login_type = $2,
+        nickname = $3,
+        role_id = $4,
+        role_name = $5,
+        area = $6,
+        partition_name = $7,
+        screenshot_path = $8,
+        artifacts_json = $9::jsonb,
+        state_json = $10::jsonb,
+        updated_at = $11
+      WHERE task_id = $12
+    `,
+    [
+      next.browser_session_id,
+      next.login_type,
+      next.nickname,
+      next.role_id,
+      next.role_name,
+      next.area,
+      next.partition_name,
+      next.screenshot_path,
+      typeof next.artifacts_json === 'string' ? next.artifacts_json : JSON.stringify(next.artifacts_json || {}),
+      typeof next.state_json === 'string' ? next.state_json : JSON.stringify(next.state_json || {}),
+      timestamp,
+      Number(taskId),
+    ],
+  )
+}
+
+function containsTencentPatch(patch = {}) {
+  return [
+    'browser_session_id',
+    'login_type',
+    'nickname',
+    'role_id',
+    'role_name',
+    'area',
+    'partition_name',
+    'screenshot_path',
+    'artifacts_json',
+    'state_json',
+  ].some((key) => Object.prototype.hasOwnProperty.call(patch, key))
+}