import { query, withTransaction } from '../db/client.js' import type { PoolClient } from 'pg' import { syncKuaishouCloudTaskStateForTask } from './kuaishou-cloud-task-state-repo.js' import { normalizeTimestampIso } from '../utils/time.js' import type { TaskCreateInput, TaskListQueryInput, TaskRuntimeContextPatch, TaskUpdatePatch, } from '../types/repository/inputs.js' import type { TaskListQueryResult, TaskRow } from '../types/repository/rows.js' type TaskQueryExecutor = (text: string, params?: unknown[]) => Promise<{ rows: unknown[] }> const KUAISHOU_FEIFEI_EXECUTOR_KEY = 'kuaishou_feifei' const AFFILIATE_DASH_EXECUTOR_KEY = 'affiliate_dash' type TaskRuntimeContextRow = { runtime_session_id: string login_type: string nickname: string role_id: string role_name: string area: string partition_name: string screenshot_path: string artifacts_json: string | Record state_json: string | Record } const TASK_FIELDS = ` ft.*, ct.id AS primary_claim_token_id, ct.token AS primary_claim_token, ct.status AS primary_claim_token_status, ct.expired_at AS primary_claim_expires_at, ctx.runtime_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, 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 { const normalizedExtra = String(extraFields || '').trim() const fieldSql = normalizedExtra ? `${TASK_FIELDS}, ${normalizedExtra}` : TASK_FIELDS return `SELECT ${fieldSql} ${TASK_JOINS}` } export async function listTasksByOrderId(orderId: number | string): Promise { const result = await query( `${buildTaskSelect()} WHERE ft.order_id = $1 ORDER BY ft.id ASC`, [Number(orderId)], ) return result.rows } export async function listTasksByOrderIds(orderIds: number[]): Promise { const normalizedIds = orderIds.map(Number).filter((value) => value > 0) if (normalizedIds.length === 0) return [] const result = await query( `${buildTaskSelect()} WHERE ft.order_id = ANY($1::bigint[]) ORDER BY ft.order_id ASC, ft.id ASC`, [normalizedIds], ) return result.rows } export async function createTask(input: TaskCreateInput): Promise { return withTransaction(async (client) => { const taskResult = await client.query<{ id: number }>( ` 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, 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::jsonb, $24, $25 ) 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.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.runtimeContext) { 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) }) } export async function updateTask( taskId: number | string, patch: TaskUpdatePatch, ): Promise { const current = await getTaskById(taskId) if (!current) { return null } return withTransaction(async (client) => { const next = { ...current, ...patch } await client.query( ` UPDATE fulfillment_tasks SET task_status = $1, delivery_status = $2, result_code = $3, result_message = $4, claim_token = $5, claim_expires_at = $6, automation_mode = $7, requires_claim = $8, user_action_status = $9, attempt_count = $10, last_error = $11, context_json = $12::jsonb, claimed_at = $13, role_confirmed_at = $14, redeemed_at = $15, updated_at = $16 WHERE id = $17 `, [ next.task_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.claimed_at || null, next.role_confirmed_at || null, next.redeemed_at || null, next.updated_at, Number(taskId), ], ) if (containsRuntimeContextPatch(patch)) { const runtimeContextPatch: TaskRuntimeContextPatch = {} if (patch.runtime_session_id !== undefined) runtimeContextPatch.runtimeSessionId = patch.runtime_session_id if (patch.login_type !== undefined) runtimeContextPatch.loginType = patch.login_type if (patch.nickname !== undefined) runtimeContextPatch.nickname = patch.nickname if (patch.role_id !== undefined) runtimeContextPatch.roleId = patch.role_id if (patch.role_name !== undefined) runtimeContextPatch.roleName = patch.role_name if (patch.area !== undefined) runtimeContextPatch.area = patch.area if (patch.partition_name !== undefined) runtimeContextPatch.partitionName = patch.partition_name if (patch.screenshot_path !== undefined) runtimeContextPatch.screenshotPath = patch.screenshot_path if (patch.artifacts_json !== undefined) runtimeContextPatch.artifactsJson = patch.artifacts_json if (patch.state_json !== undefined) runtimeContextPatch.stateJson = patch.state_json 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) }) } export async function updateTaskStatusIfCurrent( taskId: number | string, currentStatus: string, patch: TaskUpdatePatch, ): Promise { const now = normalizeTimestampIso(patch.updated_at, '') return withTransaction(async (client) => { const result = await client.query<{ id: number executor_key: string context_json: string | Record 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, ], ) const updated = result.rows[0] if (!updated) { return null } 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 { return getTaskByIdWithExecutor(query, taskId) } export async function findTaskByClaimTokenId( claimTokenId: number | string, ): Promise { 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 async function findKuaishouFeifeiTaskByOrder(input: { platformOrderNo?: unknown orderNo?: unknown }): Promise { const platformOrderNo = String(input.platformOrderNo || '').trim() const orderNo = String(input.orderNo || '').trim() const params: unknown[] = [KUAISHOU_FEIFEI_EXECUTOR_KEY] const orderFilters: string[] = [] if (platformOrderNo) { params.push(platformOrderNo) orderFilters.push(`ft.context_json #>> '{kuaishouFeifei,platformOrderNo}' = $${params.length}`) } if (orderNo) { params.push(orderNo) orderFilters.push(`ft.context_json #>> '{kuaishouFeifei,orderNo}' = $${params.length}`) } if (orderFilters.length === 0) { return null } const result = await query( `${buildTaskSelect()} WHERE ft.executor_key = $1 AND (${orderFilters.join(' OR ')}) ORDER BY ft.id DESC LIMIT 1`, params, ) return result.rows[0] || null } export async function findAffiliateDashTaskByOrder(input: { orderNo?: unknown clientOrderNo?: unknown }): Promise { const orderNo = String(input.orderNo || '').trim() const clientOrderNo = String(input.clientOrderNo || '').trim() const params: unknown[] = [AFFILIATE_DASH_EXECUTOR_KEY] const orderFilters: string[] = [] if (orderNo) { params.push(orderNo) orderFilters.push(`ft.context_json #>> '{affiliateDash,orderNo}' = $${params.length}`) } if (clientOrderNo) { params.push(clientOrderNo) orderFilters.push(`ft.context_json #>> '{affiliateDash,clientOrderNo}' = $${params.length}`) } if (orderFilters.length === 0) { return null } const result = await query( `${buildTaskSelect()} WHERE ft.executor_key = $1 AND (${orderFilters.join(' OR ')}) ORDER BY ft.id DESC LIMIT 1`, params, ) return result.rows[0] || null } export async function listTasks({ page = 1, pageSize = 20, ...filters }: TaskListQueryInput = {}): Promise { const offset = (page - 1) * pageSize const { whereClause, params } = buildTaskListFilters(filters) const totalResult = await query<{ [column: string]: unknown; total: number }>( ` SELECT COUNT(*)::int AS total 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, ) 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: itemsResult.rows, total: Number(totalResult.rows[0]?.total || 0), } } export function buildTaskListFilters({ keyword = '', executorKey = '', status = '', platformOrderId = '', taskNo = '', skuCode = '', roleId = '', dateFrom = '', dateTo = '', }: TaskListQueryInput = {}): { whereClause: string; params: unknown[] } { const filters: string[] = [] const params: unknown[] = [] if (keyword) { params.push(`%${keyword}%`) const keywordParam = `$${params.length}` filters.push(`( (ft.task_no || ' ' || ft.platform_order_id) ILIKE ${keywordParam} OR oi.sku_code ILIKE ${keywordParam} OR oi.sku_name ILIKE ${keywordParam} OR COALESCE(oi.item_snapshot_json->'kuaishouSendCode'->>'itemTitle', '') ILIKE ${keywordParam} OR COALESCE(oi.item_snapshot_json->'kuaishouSendCode'->>'skuNick', '') ILIKE ${keywordParam} OR COALESCE(NULLIF(kcts.role_id, ''), ctx.role_id, '') ILIKE ${keywordParam} OR COALESCE(NULLIF(kcts.role_name, ''), NULLIF(ctx.role_name, ''), ctx.nickname, '') ILIKE ${keywordParam} )`) } if (executorKey) { params.push(executorKey) filters.push(`ft.executor_key = $${params.length}`) } if (status) { params.push(status) filters.push(`ft.task_status = $${params.length}`) } if (platformOrderId) { params.push(`%${platformOrderId}%`) filters.push(`ft.platform_order_id ILIKE $${params.length}`) } if (taskNo) { params.push(`%${taskNo}%`) filters.push(`ft.task_no ILIKE $${params.length}`) } if (skuCode) { params.push(skuCode) filters.push(`oi.sku_code = $${params.length}`) } if (roleId) { params.push(`%${roleId}%`) filters.push(`COALESCE(NULLIF(kcts.role_id, ''), ctx.role_id) ILIKE $${params.length}`) } if (dateFrom) { params.push(dateFrom) filters.push(`ft.created_at >= $${params.length}`) } if (dateTo) { params.push(dateTo) filters.push(`ft.created_at <= $${params.length}`) } return { whereClause: filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : '', params, } } async function upsertTaskRuntimeContextWithClient( client: PoolClient, taskId: number | string, patch: TaskRuntimeContextPatch = {}, timestamp: string | undefined, ): Promise { const currentResult = await client.query( 'SELECT * FROM task_runtime_contexts WHERE task_id = $1 LIMIT 1', [Number(taskId)], ) const current = currentResult.rows[0] || null const next = { runtime_session_id: patch.runtimeSessionId ?? current?.runtime_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 task_runtime_contexts ( task_id, runtime_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.runtime_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 task_runtime_contexts SET runtime_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.runtime_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), ], ) } async function getTaskByIdWithExecutor( executor: TaskQueryExecutor, taskId: number | string, ): Promise { const result = await executor( `${buildTaskSelect()} WHERE ft.id = $1 LIMIT 1`, [Number(taskId)], ) return (result.rows[0] as TaskRow | undefined) || null } function containsRuntimeContextPatch(patch: TaskUpdatePatch = {}): boolean { return [ 'runtime_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)) }