后端迁移任务仓储模块
This commit is contained in:
yml
@@ -0,0 +1,460 @@
|
||||
import { query, withTransaction } from '../db/client.js'
|
||||
import type { PoolClient } from 'pg'
|
||||
|
||||
import type {
|
||||
TaskCreateInput,
|
||||
TaskListQueryInput,
|
||||
TaskTencentContextPatch,
|
||||
TaskUpdatePatch,
|
||||
} from '../types/repository-inputs.js'
|
||||
import type {
|
||||
TaskListQueryResult,
|
||||
TaskRow,
|
||||
} from '../types/repository-rows.js'
|
||||
|
||||
type TaskQueryExecutor = (text: string, params?: unknown[]) => Promise<{ rows: unknown[] }>
|
||||
|
||||
type TencentBrowserContextRow = {
|
||||
browser_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<string, unknown>
|
||||
state_json: string | Record<string, unknown>
|
||||
}
|
||||
|
||||
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.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 primary_inventory_item_id,
|
||||
inv.display_value AS primary_inventory_display_value,
|
||||
inv.credential_type AS primary_inventory_credential_type,
|
||||
inv.binding_status AS primary_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
|
||||
CASE tib.binding_status
|
||||
WHEN 'reserved' THEN 0
|
||||
WHEN 'consumed' THEN 1
|
||||
ELSE 2
|
||||
END ASC,
|
||||
tib.id DESC
|
||||
LIMIT 1
|
||||
) inv ON TRUE
|
||||
`
|
||||
|
||||
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<TaskRow[]> {
|
||||
const result = await query<TaskRow>(
|
||||
`${buildTaskSelect()}
|
||||
WHERE ft.order_id = $1
|
||||
ORDER BY ft.id ASC`,
|
||||
[Number(orderId)],
|
||||
)
|
||||
return result.rows
|
||||
}
|
||||
|
||||
export async function createTask(input: TaskCreateInput): Promise<TaskRow | null> {
|
||||
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,
|
||||
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 getTaskByIdWithExecutor(client.query.bind(client) as TaskQueryExecutor, taskId)
|
||||
})
|
||||
}
|
||||
|
||||
export async function updateTask(taskId: number | string, patch: TaskUpdatePatch): Promise<TaskRow | null> {
|
||||
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,
|
||||
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,
|
||||
claimed_at = $14,
|
||||
role_confirmed_at = $15,
|
||||
redeemed_at = $16,
|
||||
updated_at = $17
|
||||
WHERE id = $18
|
||||
`,
|
||||
[
|
||||
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.claimed_at || null,
|
||||
next.role_confirmed_at || null,
|
||||
next.redeemed_at || null,
|
||||
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 getTaskByIdWithExecutor(client.query.bind(client) as TaskQueryExecutor, taskId)
|
||||
})
|
||||
}
|
||||
|
||||
export async function getTaskById(taskId: number | string): Promise<TaskRow | null> {
|
||||
return getTaskByIdWithExecutor(query, taskId)
|
||||
}
|
||||
|
||||
export async function findTaskByClaimTokenId(claimTokenId: number | string): Promise<TaskRow | null> {
|
||||
const result = await query<TaskRow>(
|
||||
`${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 listTasks({
|
||||
page = 1,
|
||||
pageSize = 20,
|
||||
status = '',
|
||||
platformOrderId = '',
|
||||
taskNo = '',
|
||||
skuCode = '',
|
||||
roleId = '',
|
||||
dateFrom = '',
|
||||
dateTo = '',
|
||||
}: TaskListQueryInput = {}): Promise<TaskListQueryResult> {
|
||||
const offset = (page - 1) * pageSize
|
||||
const filters: string[] = []
|
||||
const params: unknown[] = []
|
||||
|
||||
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(`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}`)
|
||||
}
|
||||
|
||||
const whereClause = filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : ''
|
||||
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 tencent_browser_contexts ctx ON ctx.task_id = ft.id
|
||||
${whereClause}
|
||||
`,
|
||||
params,
|
||||
)
|
||||
|
||||
params.push(pageSize)
|
||||
params.push(offset)
|
||||
const itemsResult = await query<TaskRow>(
|
||||
`${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),
|
||||
}
|
||||
}
|
||||
|
||||
async function upsertTencentBrowserContextWithClient(
|
||||
client: PoolClient,
|
||||
taskId: number | string,
|
||||
patch: TaskTencentContextPatch = {},
|
||||
timestamp: string | undefined,
|
||||
): Promise<void> {
|
||||
const currentResult = await client.query<TencentBrowserContextRow>(
|
||||
'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),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
async function getTaskByIdWithExecutor(
|
||||
executor: TaskQueryExecutor,
|
||||
taskId: number | string,
|
||||
): Promise<TaskRow | null> {
|
||||
const result = await executor(
|
||||
`${buildTaskSelect()}
|
||||
WHERE ft.id = $1
|
||||
LIMIT 1`,
|
||||
[Number(taskId)],
|
||||
)
|
||||
return (result.rows[0] as TaskRow | undefined) || null
|
||||
}
|
||||
|
||||
function containsTencentPatch(patch: TaskUpdatePatch = {}): boolean {
|
||||
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))
|
||||
}
|
||||
Reference in New Issue
Block a user