后端迁移低风险仓储模块

This commit is contained in:
yml
2026-05-21 13:46:10 +08:00
parent 1d74b63800
commit 6eeb1a49df
5 changed files with 216 additions and 87 deletions
@@ -1,7 +1,72 @@
import { query } from '../db/client.js' import { query } from '../db/client.js'
import type {
MessageDeliveryListQueryResult,
MessageDeliveryRow,
} from '../types/repository-rows.js'
export async function createMessageDelivery(input) { type MessageDeliveryCreateInput = {
const result = await query( provider: string
platform: string
shopId?: string
shopName?: string
channel: string
orderId?: number | string | null
taskId?: number | string | null
platformOrderId?: string
recipientKey?: string
messageContent?: string
claimUrl?: string
status?: string
requestUrl?: string
requestHeadersJson?: string | Record<string, unknown>
requestBodyJson?: string | Record<string, unknown>
responseStatus?: number | string
responseJson?: string | Record<string, unknown>
errorMessage?: string
sentAt?: string | null
createdAt: string
updatedAt: string
}
type MessageDeliveryPatch = Partial<Pick<
MessageDeliveryRow,
| 'status'
| 'request_url'
| 'request_headers_json'
| 'request_body_json'
| 'response_status'
| 'response_json'
| 'error_message'
| 'sent_at'
| 'updated_at'
>>
type LatestSuccessfulMessageDeliveryQuery = {
provider?: string
platform?: string
shopId?: string
platformOrderId?: string
channel?: string
claimUrl?: string
}
type MessageDeliveryListQuery = {
page?: number
pageSize?: number
provider?: string
platform?: string
status?: string
shopId?: string
platformOrderId?: string
taskNo?: string
dateFrom?: string
dateTo?: string
}
export async function createMessageDelivery(
input: MessageDeliveryCreateInput,
): Promise<MessageDeliveryRow | null> {
const result = await query<MessageDeliveryRow>(
` `
INSERT INTO message_deliveries ( INSERT INTO message_deliveries (
provider, provider,
@@ -56,19 +121,24 @@ export async function createMessageDelivery(input) {
return result.rows[0] || null return result.rows[0] || null
} }
export async function getMessageDeliveryById(deliveryId) { export async function getMessageDeliveryById(deliveryId: number | string): Promise<MessageDeliveryRow | null> {
const result = await query('SELECT * FROM message_deliveries WHERE id = $1 LIMIT 1', [Number(deliveryId)]) const result = await query<MessageDeliveryRow>('SELECT * FROM message_deliveries WHERE id = $1 LIMIT 1', [
Number(deliveryId),
])
return result.rows[0] || null return result.rows[0] || null
} }
export async function updateMessageDelivery(deliveryId, patch = {}) { export async function updateMessageDelivery(
deliveryId: number | string,
patch: MessageDeliveryPatch = {},
): Promise<MessageDeliveryRow | null> {
const current = await getMessageDeliveryById(deliveryId) const current = await getMessageDeliveryById(deliveryId)
if (!current) { if (!current) {
return null return null
} }
const next = { ...current, ...patch } const next = { ...current, ...patch }
const result = await query( const result = await query<MessageDeliveryRow>(
` `
UPDATE message_deliveries UPDATE message_deliveries
SET SET
@@ -101,8 +171,11 @@ export async function updateMessageDelivery(deliveryId, patch = {}) {
return result.rows[0] || null return result.rows[0] || null
} }
export async function findLatestSuccessfulMessageDeliveryByTask(taskId, channel) { export async function findLatestSuccessfulMessageDeliveryByTask(
const result = await query( taskId: number | string,
channel: string,
): Promise<MessageDeliveryRow | null> {
const result = await query<MessageDeliveryRow>(
` `
SELECT * SELECT *
FROM message_deliveries FROM message_deliveries
@@ -123,8 +196,8 @@ export async function findLatestSuccessfulMessageDelivery({
platformOrderId = '', platformOrderId = '',
channel = '', channel = '',
claimUrl = '', claimUrl = '',
} = {}) { }: LatestSuccessfulMessageDeliveryQuery = {}): Promise<MessageDeliveryRow | null> {
const result = await query( const result = await query<MessageDeliveryRow>(
` `
SELECT * SELECT *
FROM message_deliveries FROM message_deliveries
@@ -162,10 +235,10 @@ export async function listMessageDeliveries({
taskNo = '', taskNo = '',
dateFrom = '', dateFrom = '',
dateTo = '', dateTo = '',
} = {}) { }: MessageDeliveryListQuery = {}): Promise<MessageDeliveryListQueryResult> {
const offset = (page - 1) * pageSize const offset = (page - 1) * pageSize
const filters = [] const filters: string[] = []
const params = [] const params: unknown[] = []
if (provider) { if (provider) {
params.push(provider) params.push(provider)
@@ -213,14 +286,14 @@ export async function listMessageDeliveries({
LEFT JOIN fulfillment_tasks ft ON ft.id = md.task_id LEFT JOIN fulfillment_tasks ft ON ft.id = md.task_id
` `
const totalResult = await query( const totalResult = await query<{ [column: string]: unknown, total: number }>(
`SELECT COUNT(*)::int AS total ${fromClause} ${whereClause}`, `SELECT COUNT(*)::int AS total ${fromClause} ${whereClause}`,
params, params,
) )
params.push(pageSize) params.push(pageSize)
params.push(offset) params.push(offset)
const itemsResult = await query( const itemsResult = await query<MessageDeliveryRow>(
` `
SELECT SELECT
md.*, md.*,
@@ -1,20 +1,32 @@
// @ts-check import type { PoolClient, QueryResult } from 'pg'
import { query, withTransaction } from '../db/client.js' import { query, withTransaction } from '../db/client.js'
import type { OrderItemReplaceInput } from '../types/repository-inputs.js'
import type { OrderItemRow } from '../types/repository-rows.js'
/** @typedef {import('../types/repository-inputs.js').OrderItemReplaceInput} OrderItemReplaceInput */ type QueryExecutor = (text: string, params?: unknown[]) => Promise<QueryResult<any>>
/** @typedef {import('../types/repository-rows.js').OrderItemRow} OrderItemRow */
/** @returns {Promise<OrderItemRow[]>} */ type OrderItemSyncPlan = {
export async function listOrderItemsByOrderId(orderId) { updates: Array<{
orderItemId: number
item: OrderItemReplaceInput
}>
creates: OrderItemReplaceInput[]
deletes: number[]
}
type OrderItemIdentityInput = Partial<OrderItemReplaceInput> & Partial<OrderItemRow>
export async function listOrderItemsByOrderId(orderId: number | string): Promise<OrderItemRow[]> {
return listOrderItemsByOrderIdWithExecutor(query, orderId) return listOrderItemsByOrderIdWithExecutor(query, orderId)
} }
/** @returns {Promise<OrderItemRow[]>} */ export async function replaceOrderItems(
/** @param {OrderItemReplaceInput[]} items */ orderId: number | string,
export async function replaceOrderItems(orderId, items) { items: OrderItemReplaceInput[],
return withTransaction(async (client) => { ): Promise<OrderItemRow[]> {
const executor = client.query.bind(client) return withTransaction(async (client: PoolClient) => {
const executor: QueryExecutor = client.query.bind(client)
const existingItems = await listOrderItemsByOrderIdWithExecutor(executor, orderId) const existingItems = await listOrderItemsByOrderIdWithExecutor(executor, orderId)
const plan = resolveOrderItemSyncPlan(existingItems, items) const plan = resolveOrderItemSyncPlan(existingItems, items)
@@ -74,10 +86,7 @@ export async function replaceOrderItems(orderId, items) {
const deletableIds = await listDeletableOrderItemIdsWithExecutor(executor, plan.deletes) const deletableIds = await listDeletableOrderItemIdsWithExecutor(executor, plan.deletes)
if (deletableIds.length > 0) { if (deletableIds.length > 0) {
await executor( await executor('DELETE FROM order_items WHERE id = ANY($1::bigint[])', [deletableIds])
'DELETE FROM order_items WHERE id = ANY($1::bigint[])',
[deletableIds],
)
} }
} }
@@ -85,16 +94,15 @@ export async function replaceOrderItems(orderId, items) {
}) })
} }
/** export function resolveOrderItemSyncPlan(
* @param {OrderItemRow[]} existingItems existingItems: OrderItemRow[],
* @param {OrderItemReplaceInput[]} nextItems nextItems: OrderItemReplaceInput[],
*/ ): OrderItemSyncPlan {
export function resolveOrderItemSyncPlan(existingItems, nextItems) {
const normalizedExisting = Array.isArray(existingItems) ? existingItems : [] const normalizedExisting = Array.isArray(existingItems) ? existingItems : []
const normalizedNext = Array.isArray(nextItems) ? nextItems : [] const normalizedNext = Array.isArray(nextItems) ? nextItems : []
const unmatchedExisting = [...normalizedExisting] const unmatchedExisting = [...normalizedExisting]
const updates = [] const updates: OrderItemSyncPlan['updates'] = []
const creates = [] const creates: OrderItemReplaceInput[] = []
for (const item of normalizedNext) { for (const item of normalizedNext) {
const matchedIndex = findMatchingExistingOrderItemIndex(unmatchedExisting, item) const matchedIndex = findMatchingExistingOrderItemIndex(unmatchedExisting, item)
@@ -110,10 +118,12 @@ export function resolveOrderItemSyncPlan(existingItems, nextItems) {
if (unmatchedExisting.length > 0) { if (unmatchedExisting.length > 0) {
const matched = unmatchedExisting.shift() const matched = unmatchedExisting.shift()
if (matched) {
updates.push({ updates.push({
orderItemId: matched.id, orderItemId: matched.id,
item, item,
}) })
}
continue continue
} }
@@ -127,21 +137,27 @@ export function resolveOrderItemSyncPlan(existingItems, nextItems) {
} }
} }
function findMatchingExistingOrderItemIndex(existingItems, nextItem) { function findMatchingExistingOrderItemIndex(
existingItems: OrderItemRow[],
nextItem: OrderItemReplaceInput,
): number {
const nextIdentity = buildOrderItemIdentity(nextItem) const nextIdentity = buildOrderItemIdentity(nextItem)
return existingItems.findIndex((item) => buildOrderItemIdentity(item) === nextIdentity) return existingItems.findIndex((item) => buildOrderItemIdentity(item) === nextIdentity)
} }
function buildOrderItemIdentity(item) { function buildOrderItemIdentity(item: OrderItemIdentityInput): string {
return [ return [
String(item?.skuCode ?? item?.sku_code ?? '').trim(), String(item.skuCode ?? item.sku_code ?? '').trim(),
String(item?.skuName ?? item?.sku_name ?? '').trim(), String(item.skuName ?? item.sku_name ?? '').trim(),
String(Math.max(1, Number(item?.quantity || 1))), String(Math.max(1, Number(item.quantity || 1))),
].join('::') ].join('::')
} }
async function listDeletableOrderItemIdsWithExecutor(executor, orderItemIds) { async function listDeletableOrderItemIdsWithExecutor(
executor: QueryExecutor,
orderItemIds: number[],
): Promise<number[]> {
const normalizedIds = Array.isArray(orderItemIds) const normalizedIds = Array.isArray(orderItemIds)
? orderItemIds.map((value) => Number(value)).filter((value) => value > 0) ? orderItemIds.map((value) => Number(value)).filter((value) => value > 0)
: [] : []
@@ -166,7 +182,10 @@ async function listDeletableOrderItemIdsWithExecutor(executor, orderItemIds) {
return result.rows.map((row) => Number(row.id)).filter((value) => value > 0) return result.rows.map((row) => Number(row.id)).filter((value) => value > 0)
} }
async function listOrderItemsByOrderIdWithExecutor(executor, orderId) { async function listOrderItemsByOrderIdWithExecutor(
executor: QueryExecutor,
orderId: number | string,
): Promise<OrderItemRow[]> {
const result = await executor( const result = await executor(
` `
SELECT * SELECT *
@@ -177,11 +196,10 @@ async function listOrderItemsByOrderIdWithExecutor(executor, orderId) {
[Number(orderId)], [Number(orderId)],
) )
return /** @type {OrderItemRow[]} */ (result.rows) return result.rows
} }
/** @returns {Promise<OrderItemRow | null>} */ export async function getOrderItemById(orderItemId: number | string): Promise<OrderItemRow | null> {
export async function getOrderItemById(orderItemId) { const result = await query<OrderItemRow>('SELECT * FROM order_items WHERE id = $1 LIMIT 1', [Number(orderItemId)])
const result = await query('SELECT * FROM order_items WHERE id = $1 LIMIT 1', [Number(orderItemId)]) return result.rows[0] || null
return /** @type {OrderItemRow | null} */ (result.rows[0] || null)
} }
@@ -1,17 +1,16 @@
// @ts-check
import { query } from '../db/client.js' import { query } from '../db/client.js'
import type {
WebhookEventCreateInput,
WebhookEventListQueryInput,
WebhookEventUpdatePatch,
} from '../types/repository-inputs.js'
import type {
WebhookEventListQueryResult,
WebhookEventRow,
} from '../types/repository-rows.js'
/** @typedef {import('../types/repository-inputs.js').WebhookEventCreateInput} WebhookEventCreateInput */ export async function createWebhookEvent(input: WebhookEventCreateInput): Promise<WebhookEventRow | null> {
/** @typedef {import('../types/repository-inputs.js').WebhookEventListQueryInput} WebhookEventListQueryInput */ const result = await query<WebhookEventRow>(
/** @typedef {import('../types/repository-inputs.js').WebhookEventUpdatePatch} WebhookEventUpdatePatch */
/** @typedef {import('../types/repository-rows.js').WebhookEventListQueryResult} WebhookEventListQueryResult */
/** @typedef {import('../types/repository-rows.js').WebhookEventRow} WebhookEventRow */
/** @returns {Promise<WebhookEventRow | null>} */
/** @param {WebhookEventCreateInput} input */
export async function createWebhookEvent(input) {
const result = await query(
` `
INSERT INTO webhook_events ( INSERT INTO webhook_events (
provider, provider,
@@ -49,19 +48,20 @@ export async function createWebhookEvent(input) {
], ],
) )
return /** @type {WebhookEventRow | null} */ (result.rows[0] || null) return result.rows[0] || null
} }
/** @returns {Promise<WebhookEventRow | null>} */ export async function updateWebhookEvent(
/** @param {WebhookEventUpdatePatch} patch */ eventId: number | string,
export async function updateWebhookEvent(eventId, patch) { patch: WebhookEventUpdatePatch,
): Promise<WebhookEventRow | null> {
const current = await getWebhookEventById(eventId) const current = await getWebhookEventById(eventId)
if (!current) { if (!current) {
return null return null
} }
const next = { ...current, ...patch } const next = { ...current, ...patch }
const result = await query( const result = await query<WebhookEventRow>(
` `
UPDATE webhook_events UPDATE webhook_events
SET SET
@@ -74,17 +74,16 @@ export async function updateWebhookEvent(eventId, patch) {
[Boolean(next.processed), next.process_error || '', next.related_order_id || null, Number(eventId)], [Boolean(next.processed), next.process_error || '', next.related_order_id || null, Number(eventId)],
) )
return /** @type {WebhookEventRow | null} */ (result.rows[0] || null) return result.rows[0] || null
} }
/** @returns {Promise<WebhookEventRow | null>} */ export async function getWebhookEventById(eventId: number | string): Promise<WebhookEventRow | null> {
export async function getWebhookEventById(eventId) { const result = await query<WebhookEventRow>('SELECT * FROM webhook_events WHERE id = $1 LIMIT 1', [
const result = await query('SELECT * FROM webhook_events WHERE id = $1 LIMIT 1', [Number(eventId)]) Number(eventId),
return /** @type {WebhookEventRow | null} */ (result.rows[0] || null) ])
return result.rows[0] || null
} }
/** @returns {Promise<WebhookEventListQueryResult>} */
/** @param {WebhookEventListQueryInput} [queryInput] */
export async function listWebhookEvents({ export async function listWebhookEvents({
page = 1, page = 1,
pageSize = 20, pageSize = 20,
@@ -96,10 +95,10 @@ export async function listWebhookEvents({
relatedOrderId = '', relatedOrderId = '',
dateFrom = '', dateFrom = '',
dateTo = '', dateTo = '',
} = /** @type {WebhookEventListQueryInput} */ ({})) { }: WebhookEventListQueryInput = {}): Promise<WebhookEventListQueryResult> {
const offset = (page - 1) * pageSize const offset = (page - 1) * pageSize
const filters = [] const filters: string[] = []
const params = [] const params: unknown[] = []
if (provider) { if (provider) {
params.push(provider) params.push(provider)
@@ -143,11 +142,14 @@ export async function listWebhookEvents({
} }
const whereClause = filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : '' const whereClause = filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : ''
const totalResult = await query(`SELECT COUNT(*)::int AS total FROM webhook_events ${whereClause}`, params) const totalResult = await query<{ [column: string]: unknown, total: number }>(
`SELECT COUNT(*)::int AS total FROM webhook_events ${whereClause}`,
params,
)
params.push(pageSize) params.push(pageSize)
params.push(offset) params.push(offset)
const itemsResult = await query( const itemsResult = await query<WebhookEventRow>(
` `
SELECT * SELECT *
FROM webhook_events FROM webhook_events
@@ -159,14 +161,13 @@ export async function listWebhookEvents({
) )
return { return {
items: /** @type {WebhookEventRow[]} */ (itemsResult.rows), items: itemsResult.rows,
total: Number(totalResult.rows[0]?.total || 0), total: Number(totalResult.rows[0]?.total || 0),
} }
} }
/** @returns {Promise<WebhookEventRow[]>} */ export async function listWebhookEventsByOrderId(orderId: number | string): Promise<WebhookEventRow[]> {
export async function listWebhookEventsByOrderId(orderId) { const result = await query<WebhookEventRow>(
const result = await query(
` `
SELECT * SELECT *
FROM webhook_events FROM webhook_events
@@ -176,5 +177,5 @@ export async function listWebhookEventsByOrderId(orderId) {
[Number(orderId)], [Number(orderId)],
) )
return /** @type {WebhookEventRow[]} */ (result.rows) return result.rows
} }
+32
View File
@@ -119,6 +119,33 @@ export type WebhookEventRow = {
created_at: string created_at: string
} }
export type MessageDeliveryRow = {
id: number
provider: string
platform: string
shop_id: string
shop_name: string
channel: string
order_id: number | null
task_id: number | null
platform_order_id: string
recipient_key: string
message_content: string
claim_url: string
status: string
request_url: string
request_headers_json: string | Record<string, unknown>
request_body_json: string | Record<string, unknown>
response_status: number
response_json: string | Record<string, unknown>
error_message: string
sent_at: string | null
created_at: string
updated_at: string
task_no?: string
task_status?: string
}
export type TaskInventoryBindingRow = { export type TaskInventoryBindingRow = {
id: number id: number
task_id: number task_id: number
@@ -184,3 +211,8 @@ export type WebhookEventListQueryResult = {
items: WebhookEventRow[] items: WebhookEventRow[]
total: number total: number
} }
export type MessageDeliveryListQueryResult = {
items: MessageDeliveryRow[]
total: number
}
+6 -1
View File
@@ -219,12 +219,17 @@
- `npm run build` - `npm run build`
- `npm test` - `npm test`
11. `src/types/*.js` 已改写为真正的 `.ts` 类型导出文件,现有 JS JSDoc `import('...js').TypeName` 引用保持可用 11. `src/types/*.js` 已改写为真正的 `.ts` 类型导出文件,现有 JS JSDoc `import('...js').TypeName` 引用保持可用
12. 低风险 repository 第二批已迁移到 `.ts`
- `src/repositories/order-item-repo.ts`
- `src/repositories/message-delivery-repo.ts`
- `src/repositories/webhook-event-repo.ts`
13. `MessageDeliveryRow` 与列表查询结果已进入共享 repository row 类型
## 下一步建议 ## 下一步建议
第一批继续推进时,建议按这个顺序: 第一批继续推进时,建议按这个顺序:
1. 继续迁移低风险 repository`order-item-repo``message-delivery-repo``webhook-event-repo` 1. 继续迁移剩余 repository`admin-user-repo``inventory-repo``order-repo``task-repo`
2. 拆分并迁移 `runtime.js`,把 env 解析、默认配置加载、配置合并分开 2. 拆分并迁移 `runtime.js`,把 env 解析、默认配置加载、配置合并分开
3. 为 webhook、库存换码、自动发货补测试 3. 为 webhook、库存换码、自动发货补测试