import db from '../config/database'; import { logger } from '../utils/logger'; /** * [BE-I002] 库存同步服务 * @description 实现多商户库存的实时同步和管理 * @taskId BE-I002 * @version 1.0 */ export class InventorySyncService { private static generateId(prefix: string): string { return `${prefix}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`; } private static generateTraceId(): string { return `trace_${Date.now()}_${Math.random().toString(36).substr(2, 12)}`; } /** * 同步商户库存 * @param merchantId 商户ID * @param inventoryData 库存数据 */ static async syncMerchantInventory( merchantId: string, inventoryData: Array<{ skuId: string; quantity: number; warehouseId?: string; platform?: string; }> ): Promise<{ success: boolean; syncedCount: number; failedCount: number }> { const traceId = this.generateTraceId(); logger.info('[InventorySyncService] Syncing merchant inventory', { merchantId, itemCount: inventoryData.length, traceId, }); try { const merchant = await db('cf_merchant').where({ id: merchantId }).first(); if (!merchant || merchant.status !== 'ACTIVE') { throw new Error('Merchant not found or not active'); } let syncedCount = 0; let failedCount = 0; for (const item of inventoryData) { try { const existingInventory = await db('cf_inventory') .where({ tenant_id: merchant.tenant_id, sku_id: item.skuId, warehouse_id: item.warehouseId || 'default', }) .first(); if (existingInventory) { await db('cf_inventory') .where({ id: existingInventory.id }) .update({ total_qty: item.quantity, available_qty: item.quantity - (existingInventory.reserved_qty || 0), updated_at: new Date(), }); } else { await db('cf_inventory').insert({ id: this.generateId('inv'), tenant_id: merchant.tenant_id, sku_id: item.skuId, warehouse_id: item.warehouseId || 'default', total_qty: item.quantity, available_qty: item.quantity, reserved_qty: 0, trace_id: traceId, business_type: 'TOC', created_at: new Date(), updated_at: new Date(), }); } // 记录同步日志 await db('cf_inventory_sync_log').insert({ id: this.generateId('sync_log'), merchant_id: merchantId, sku_id: item.skuId, quantity: item.quantity, platform: item.platform || 'unknown', status: 'SUCCESS', trace_id: traceId, created_at: new Date(), }); syncedCount++; } catch (error: any) { logger.error('[InventorySyncService] Failed to sync inventory item', { merchantId, skuId: item.skuId, error: error.message, traceId, }); // 记录失败日志 await db('cf_inventory_sync_log').insert({ id: this.generateId('sync_log'), merchant_id: merchantId, sku_id: item.skuId, quantity: item.quantity, platform: item.platform || 'unknown', status: 'FAILED', error_message: error.message, trace_id: traceId, created_at: new Date(), }); failedCount++; } } logger.info('[InventorySyncService] Inventory sync completed', { merchantId, syncedCount, failedCount, traceId, }); return { success: failedCount === 0, syncedCount, failedCount }; } catch (error: any) { logger.error('[InventorySyncService] Inventory sync failed', { merchantId, traceId, error: error.message, }); throw error; } } /** * 获取库存同步历史 * @param merchantId 商户ID * @param limit 限制数量 */ static async getSyncHistory( merchantId: string, limit: number = 50 ): Promise { const traceId = this.generateTraceId(); logger.info('[InventorySyncService] Getting sync history', { merchantId, limit, traceId, }); try { const history = await db('cf_inventory_sync_log') .where({ merchant_id: merchantId }) .orderBy('created_at', 'desc') .limit(limit) .select('*'); logger.info('[InventorySyncService] Sync history retrieved', { merchantId, recordCount: history.length, traceId, }); return history; } catch (error: any) { logger.error('[InventorySyncService] Failed to get sync history', { merchantId, traceId, error: error.message, }); throw error; } } /** * 初始化库存同步相关数据库表 */ static async initTables(): Promise { const syncLogTableExists = await db.schema.hasTable('cf_inventory_sync_log'); if (!syncLogTableExists) { logger.info('📦 Creating cf_inventory_sync_log table...'); await db.schema.createTable('cf_inventory_sync_log', (table) => { table.string('id', 64).primary(); table.string('merchant_id', 64).notNullable(); table.string('sku_id', 64).notNullable(); table.integer('quantity').notNullable(); table.string('platform', 50); table.enum('status', ['SUCCESS', 'FAILED']).notNullable(); table.text('error_message'); table.string('trace_id', 64); table.timestamp('created_at').defaultTo(db.fn.now()); table.index(['merchant_id']); table.index(['sku_id']); table.index(['status']); table.index(['created_at']); }); logger.info('✅ Table cf_inventory_sync_log created'); } } }