wurenzhi/makemd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(services): 重构服务模块结构，按功能分类移动文件

Browse Source 
将服务文件按功能分类移动到对应子目录，包括财务、营销、订单等模块
更新相关路由和导入路径，修复文件引用错误
归档旧版任务文档，更新README和任务统计信息
This commit is contained in:
wurenzhi
2026-03-23 15:41:50 +08:00
parent 2b86715c09
commit e59d7c6620
156 changed files with 14658 additions and 7774 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
server/src/services/utils/ActionAuditService.ts Normal file
View File

@@ -0,0 +1,15 @@
import { logger } from '../utils/logger';
/**
* Action Audit Service
* @description 操作审计服务，负责审计用户操作
*/
export class ActionAuditService {
/**
* 初始化数据库表
*/
static async initTable() {
logger.info('🚀 ActionAuditService table initialized');
// 这里可以添加数据库表初始化逻辑
}
}

276
server/src/services/utils/AuditService.ts Normal file
View File

@@ -0,0 +1,276 @@
import { Queue } from 'bullmq';
import db from '../config/database';
import { logger } from '../utils/logger';
import { DomainEventBus, DomainEvent } from '../core/runtime/DomainEventBus';
export interface AuditLogEntry {
tenantId: string;
shopId?: string;
taskId?: string;
traceId: string;
userId: string;
roleCode?: string;
module: string;
action: string;
resourceType: string;
resourceId?: string;
beforeSnapshot?: any;
afterSnapshot?: any;
result: 'success' | 'failed';
errorCode?: string;
errorMessage?: string;
clientIp?: string;
userAgent?: string;
source: 'console' | 'extension' | 'node';
metadata?: Record<string, any>;
businessType?: 'TOC' | 'TOB';
timestamp: number;
}
/**
* @description 跨境风控审计内核 (CORE_DEV_04 / BIZ_GOV_20)
* 实现基于 BullMQ 的异步审计日志系统，并对接 DomainEventBus 实现全量业务事件审计。
*/
export class AuditService {
/**
* 初始化数据库表并启动事件监听 (BIZ_GOV_20)
*/
static async initTable() {
// 启动全量事件监听 (BIZ_GOV_20)
this.startEventListener();
// 1. Audit Log 表
const hasAuditLogTable = await db.schema.hasTable('cf_audit_log');
if (!hasAuditLogTable) {
console.log('📦 Creating cf_audit_log table...');
await db.schema.createTable('cf_audit_log', (table) => {
table.increments('id').primary();
table.string('user_id', 100);
table.string('action', 100).notNullable();
table.string('target_type', 100).notNullable();
table.string('target_id', 100).notNullable();
table.string('tenant_id', 100).index();
table.string('shop_id', 100).index();
table.string('trace_id', 100).index();
table.text('old_data');
table.text('new_data');
table.text('metadata');
table.timestamp('created_at').defaultTo(db.fn.now());
table.index(['action', 'target_type', 'target_id'], 'idx_audit_action_target');
table.index(['tenant_id', 'shop_id'], 'idx_audit_tenant_shop');
});
console.log('✅ Table cf_audit_log created');
}
// 2. Operation Log 表
const hasOpLogTable = await db.schema.hasTable('cf_operation_log');
if (!hasOpLogTable) {
console.log('📦 Creating cf_operation_log table...');
await db.schema.createTable('cf_operation_log', (table) => {
table.bigIncrements('id').primary();
table.string('tenant_id', 64).notNullable();
table.string('shop_id', 64);
table.string('task_id', 64);
table.string('trace_id', 128).notNullable();
table.string('user_id', 64).notNullable();
table.string('role_code', 32);
table.string('module', 64).notNullable();
table.string('action', 64).notNullable();
table.string('resource_type', 64).notNullable();
table.string('resource_id', 128);
table.json('before_snapshot');
table.json('after_snapshot');
table.string('result', 16).notNullable();
table.string('error_code', 64);
table.string('error_message', 512);
table.string('client_ip', 64);
table.string('user_agent', 512);
table.string('source', 16).notNullable();
table.timestamp('created_at').defaultTo(db.fn.now());
table.index(['tenant_id', 'created_at'], 'idx_oplog_tenant_time');
table.index(['trace_id'], 'idx_oplog_trace');
table.index(['user_id', 'created_at'], 'idx_oplog_user_time');
table.index(['task_id'], 'idx_oplog_task');
table.index(['module', 'action', 'created_at'], 'idx_oplog_module_action_time');
});
console.log('✅ Table cf_operation_log created');
}
}
/**
* 监听 DomainEventBus 上的所有事件并转化为审计日志
*/
private static startEventListener() {
logger.info('[AuditService] Starting DomainEventBus listener for full-pipeline auditing...');
DomainEventBus.getInstance().subscribeAll(async (event: DomainEvent) => {
try {
await this.log({
tenantId: event.tenantId,
traceId: event.traceId || `bus-${Date.now()}`,
userId: event.userId || 'EVENT_BUS',
module: event.module,
action: event.action,
resourceType: event.resourceType,
resourceId: event.resourceId,
afterSnapshot: event.data,
result: 'success',
source: 'node',
metadata: { eventTimestamp: event.timestamp }
});
} catch (err: any) {
logger.error(`[AuditService] Failed to process bus event: ${err.message}`);
}
});
}
private static auditQueue = new Queue('audit-log', {
connection: {
host: process.env.REDIS_HOST || '127.0.0.1',
port: parseInt(process.env.REDIS_PORT || '6379'),
}
});
/**
* [ERP_MST_02] 记录配置项变更审计 (Granular Config Audit)
* @description 针对 ERP 核心配置（如税码、财务科目、费率模板）的颗粒度变更进行审计溯源。
*/
static async logConfigChange(params: {
tenantId: string;
userId: string;
traceId: string;
configKey: string;
oldValue: any;
newValue: any;
changeReason?: string;
metadata?: any;
}) {
logger.info(`[AuditService] Recording config change for ${params.configKey} (Tenant: ${params.tenantId})`);
await this.log({
tenantId: params.tenantId,
userId: params.userId,
traceId: params.traceId,
module: 'ERP_CONFIG',
action: 'CONFIG_CHANGE',
resourceType: 'SYSTEM_CONFIG',
resourceId: params.configKey,
beforeSnapshot: params.oldValue,
afterSnapshot: params.newValue,
result: 'success',
source: 'node',
metadata: {
...params.metadata,
changeReason: params.changeReason
}
});
}
/**
* @description 异步记录审计日志
*/
static async log(entry: Omit<AuditLogEntry, 'timestamp'>) {
try {
const fullEntry: AuditLogEntry = {
...entry,
timestamp: Date.now(),
};
// 1. 自动敏感数据脱敏
this.maskSensitiveData(fullEntry);
// 2. 加入队列异步处理
await this.auditQueue.add('log', fullEntry, {
removeOnComplete: true,
removeOnFail: 1000,
});
logger.info(`[Audit] Job added to queue: ${entry.action} on ${entry.resourceType}:${entry.resourceId}`);
} catch (err: any) {
logger.error(`[Audit] Failed to queue audit log: ${err.message}`);
}
}
/**
* @description 系统后台任务审计日志 (无 Request 上下文)
*/
static async logSystem(params: {
action: string;
resourceType: string;
resourceId: string;
beforeSnapshot?: any;
afterSnapshot?: any;
metadata?: any;
}) {
await this.log({
tenantId: 'SYSTEM',
traceId: `sys-${Date.now()}`,
userId: 'SYSTEM_BOT',
module: 'SYSTEM',
source: 'node',
result: 'success',
...params
});
}
/**
* @description 对比两个对象并记录差异
*/
static async logDiff(
params: {
tenantId: string;
shopId?: string;
taskId?: string;
traceId: string;
userId: string;
module: string;
action: string;
resourceType: string;
resourceId: string;
beforeSnapshot: any;
afterSnapshot: any;
source: 'console' | 'extension' | 'node';
}
) {
const { beforeSnapshot, afterSnapshot, ...rest } = params;
const diff = this.getDiff(beforeSnapshot, afterSnapshot);
if (Object.keys(diff.changed).length === 0) return;
await this.log({
...rest,
beforeSnapshot: diff.old,
afterSnapshot: diff.new,
result: 'success',
metadata: { diffKeys: Object.keys(diff.changed) }
});
}
private static maskSensitiveData(entry: any) {
const sensitiveKeys = ['password', 'token', 'secret', 'costPrice', 'apiKey', 'encryptedData', 'iv', 'tag'];
const mask = (obj: any) => {
if (!obj || typeof obj !== 'object') return;
for (const key in obj) {
if (sensitiveKeys.some(sk => key.toLowerCase().includes(sk.toLowerCase()))) {
obj[key] = '***MASKED***';
} else if (typeof obj[key] === 'object') {
mask(obj[key]);
}
}
};
mask(entry.beforeSnapshot);
mask(entry.afterSnapshot);
}
private static getDiff(oldData: any, newData: any) {
const diff: { old: any; new: any; changed: any } = { old: {}, new: {}, changed: {} };
for (const key in newData) {
if (JSON.stringify(oldData[key]) !== JSON.stringify(newData[key])) {
diff.old[key] = oldData[key];
diff.new[key] = newData[key];
diff.changed[key] = true;
}
}
return diff;
}
}

55
server/src/services/utils/AuditWorker.ts Normal file
View File

@@ -0,0 +1,55 @@
import { Worker } from 'bullmq';
import { logger } from '../utils/logger';
import { AuditLogEntry } from './AuditService';
import db from '../config/database';
/**
* @description 审计日志异步处理 Worker (CORE_DEV_04)
*/
export const startAuditWorker = () => {
const worker = new Worker('audit-log', async (job) => {
const entry: AuditLogEntry = job.data;
try {
// 1. 持久化到数据库 (V22.1)
await db('cf_operation_log').insert({
tenant_id: entry.tenantId,
shop_id: entry.shopId,
task_id: entry.taskId,
trace_id: entry.traceId,
user_id: entry.userId,
role_code: entry.roleCode,
module: entry.module,
action: entry.action,
resource_type: entry.resourceType,
resource_id: entry.resourceId,
before_snapshot: entry.beforeSnapshot ? JSON.stringify(entry.beforeSnapshot) : null,
after_snapshot: entry.afterSnapshot ? JSON.stringify(entry.afterSnapshot) : null,
result: entry.result,
error_code: entry.errorCode,
error_message: entry.errorMessage,
client_ip: entry.clientIp,
user_agent: entry.userAgent,
source: entry.source,
created_at: new Date(entry.timestamp)
});
logger.info(`[AuditWorker] Successfully processed operation log: ${entry.action}`);
} catch (err: any) {
logger.error(`[AuditWorker] Failed to persist audit log: ${err.message}`);
throw err; // 触发 BullMQ 的重试机制
}
}, {
connection: {
host: process.env.REDIS_HOST || '127.0.0.1',
port: parseInt(process.env.REDIS_PORT || '6379'),
},
concurrency: 5 // 限制并发数
});
worker.on('failed', (job, err) => {
logger.error(`[AuditWorker] Job ${job?.id} failed: ${err.message}`);
});
return worker;
};

163
server/src/services/utils/CacheService.ts Normal file
View File

@@ -0,0 +1,163 @@
/**
* 统一缓存服务
* 提供标准化的缓存操作，包括缓存键生成、读写、过期时间管理等
*/
import RedisService from './RedisService';
import { logger } from '../utils/logger';
export interface CacheOptions {
ttl?: number; // 缓存过期时间（秒）
prefix?: string; // 缓存键前缀
namespace?: string; // 命名空间
}
export class CacheService {
/**
* 生成缓存键
* @param key 基础键名
* @param options 缓存选项
* @param params 额外参数（用于生成唯一键）
*/
static generateKey(key: string, options: CacheOptions = {}, ...params: (string | number | undefined)[]): string {
const prefix = options.prefix || 'cache';
const namespace = options.namespace ? `${options.namespace}:` : '';
const paramsString = params
.filter((p): p is string | number => p !== undefined)
.map(p => String(p))
.join(':');
return `${prefix}:${namespace}${key}${paramsString ? `:${paramsString}` : ''}`;
}
/**
* 获取缓存
* @param key 缓存键
* @param options 缓存选项
* @param params 额外参数（用于生成唯一键）
*/
static async get<T>(key: string, options: CacheOptions = {}, ...params: (string | number | undefined)[]): Promise<T | null> {
try {
const cacheKey = this.generateKey(key, options, ...params);
const value = await RedisService.get(cacheKey);
return value as T | null;
} catch (error) {
logger.warn(`[CacheService] Failed to get cache: ${(error as any).message}`);
return null;
}
}
/**
* 设置缓存
* @param key 缓存键
* @param value 缓存值
* @param options 缓存选项
* @param params 额外参数（用于生成唯一键）
*/
static async set<T>(key: string, value: T, options: CacheOptions = {}, ...params: (string | number | undefined)[]): Promise<boolean> {
try {
const cacheKey = this.generateKey(key, options, ...params);
const ttl = options.ttl || 3600; // 默认1小时
return await RedisService.set(cacheKey, value, ttl);
} catch (error) {
logger.warn(`[CacheService] Failed to set cache: ${(error as any).message}`);
return false;
}
}
/**
* 删除缓存
* @param key 缓存键
* @param options 缓存选项
* @param params 额外参数（用于生成唯一键）
*/
static async delete(key: string, options: CacheOptions = {}, ...params: (string | number | undefined)[]): Promise<boolean> {
try {
const cacheKey = this.generateKey(key, options, ...params);
return await RedisService.del(cacheKey);
} catch (error) {
logger.warn(`[CacheService] Failed to delete cache: ${(error as any).message}`);
return false;
}
}
/**
* 批量删除缓存
* @param keys 缓存键列表
* @param options 缓存选项
*/
static async deleteMultiple(keys: string[], options: CacheOptions = {}): Promise<boolean> {
try {
const cacheKeys = keys.map(key => this.generateKey(key, options));
return await RedisService.delMultiple(cacheKeys);
} catch (error) {
logger.warn(`[CacheService] Failed to delete multiple caches: ${(error as any).message}`);
return false;
}
}
/**
* 删除匹配模式的缓存
* @param pattern 键模式（支持通配符）
* @param options 缓存选项
*/
static async deletePattern(pattern: string, options: CacheOptions = {}): Promise<boolean> {
try {
const prefix = options.prefix || 'cache';
const namespace = options.namespace ? `${options.namespace}:` : '';
const fullPattern = `${prefix}:${namespace}${pattern}`;
return await RedisService.deletePattern(fullPattern);
} catch (error) {
logger.warn(`[CacheService] Failed to delete pattern: ${(error as any).message}`);
return false;
}
}
/**
* 缓存包装器
* @param key 缓存键
* @param fn 要执行的函数
* @param options 缓存选项
* @param params 额外参数（用于生成唯一键）
*/
static async wrap<T>(
key: string,
fn: () => Promise<T>,
options: CacheOptions = {},
...params: (string | number | undefined)[]
): Promise<T> {
// 尝试从缓存获取
const cached = await this.get<T>(key, options, ...params);
if (cached !== null) {
return cached;
}
// 执行函数获取结果
const result = await fn();
// 设置缓存
await this.set(key, result, options, ...params);
return result;
}
/**
* 清除命名空间下的所有缓存
* @param namespace 命名空间
* @param options 缓存选项
*/
static async clearNamespace(namespace: string, options: CacheOptions = {}): Promise<boolean> {
return this.deletePattern('*', { ...options, namespace });
}
/**
* 清除所有缓存
* @param options 缓存选项
*/
static async clearAll(options: CacheOptions = {}): Promise<boolean> {
return this.deletePattern('*', options);
}
}
export default CacheService;

315
server/src/services/utils/CacheStrategyService.ts Normal file
View File

@@ -0,0 +1,315 @@
import Redis from 'ioredis';
import { config } from '../config/index';
export class CacheStrategyService {
private static redisClient: Redis;
/**
* 初始化Redis客户端
*/
private static initializeRedis() {
if (!this.redisClient) {
this.redisClient = new Redis({
host: config.redis.host,
port: config.redis.port,
password: config.redis.password,
});
}
}
/**
* 分析缓存使用情况
* @param params 分析参数
* @param traceInfo 追踪信息
* @returns 缓存使用分析结果
*/
public static async analyzeCacheUsage(
params: {
timeRange: { start: string; end: string };
cacheKeys?: string[];
},
traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}
): Promise<{
timestamp: string;
timeRange: { start: string; end: string };
totalCacheKeys: number;
cacheHitRate: number;
cacheMissRate: number;
topAccessedKeys: Array<{
key: string;
accessCount: number;
hitCount: number;
missCount: number;
hitRate: number;
}>;
recommendations: string[];
}> {
try {
const timestamp = new Date().toISOString();
// 模拟缓存使用分析数据
const topAccessedKeys = Array.from({ length: 10 }, (_, index) => ({
key: `product:${index + 1}`,
accessCount: Math.floor(Math.random() * 10000) + 1000,
hitCount: Math.floor(Math.random() * 9000) + 900,
missCount: Math.floor(Math.random() * 1000) + 100,
hitRate: (Math.random() * 30 + 70).toFixed(2),
}));
const totalCacheKeys = 1000;
const cacheHitRate = (Math.random() * 30 + 70).toFixed(2);
const cacheMissRate = (100 - parseFloat(cacheHitRate)).toFixed(2);
// 生成优化建议
const recommendations = [
'增加热门商品的缓存过期时间',
'实现缓存预热机制，提前加载热门数据',
'优化缓存键设计，使用更合理的命名规则',
'考虑使用二级缓存，提高缓存命中率',
'监控缓存使用情况，及时调整缓存策略',
];
return {
timestamp,
timeRange: params.timeRange,
totalCacheKeys,
cacheHitRate: parseFloat(cacheHitRate),
cacheMissRate: parseFloat(cacheMissRate),
topAccessedKeys: topAccessedKeys.map(key => ({
...key,
hitRate: parseFloat(key.hitRate),
})),
recommendations,
};
} catch (error) {
console.error('Error analyzing cache usage:', error);
throw new Error('Failed to analyze cache usage');
}
}
/**
* 优化缓存策略
* @param params 缓存策略参数
* @param traceInfo 追踪信息
* @returns 缓存策略优化结果
*/
public static async optimizeCacheStrategy(
params: {
cacheKeys: Array<{
key: string;
ttl: number;
priority: 'high' | 'medium' | 'low';
}>;
},
traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}
): Promise<{
timestamp: string;
optimizedKeys: number;
results: Array<{
key: string;
oldTtl: number;
newTtl: number;
status: string;
}>;
}> {
try {
const timestamp = new Date().toISOString();
const results = [];
for (const keyInfo of params.cacheKeys) {
// 根据优先级调整TTL
let newTtl = keyInfo.ttl;
if (keyInfo.priority === 'high') {
newTtl = 3600; // 1 hour
} else if (keyInfo.priority === 'medium') {
newTtl = 1800; // 30 minutes
} else if (keyInfo.priority === 'low') {
newTtl = 600; // 10 minutes
}
results.push({
key: keyInfo.key,
oldTtl: keyInfo.ttl,
newTtl,
status: 'success',
});
}
return {
timestamp,
optimizedKeys: params.cacheKeys.length,
results,
};
} catch (error) {
console.error('Error optimizing cache strategy:', error);
throw new Error('Failed to optimize cache strategy');
}
}
/**
* 实现缓存预热
* @param params 预热参数
* @param traceInfo 追踪信息
* @returns 预热结果
*/
public static async prewarmCache(
params: {
cacheKeys: Array<{
key: string;
data: any;
ttl: number;
}>;
},
traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}
): Promise<{
timestamp: string;
prewarmedKeys: number;
results: Array<{
key: string;
status: 'success' | 'failure';
message: string;
}>;
}> {
try {
const timestamp = new Date().toISOString();
const results = [];
let prewarmedKeys = 0;
for (const keyInfo of params.cacheKeys) {
try {
// 模拟缓存预热
results.push({
key: keyInfo.key,
status: 'success' as 'success',
message: 'Cache prewarmed successfully',
});
prewarmedKeys++;
} catch (error) {
results.push({
key: keyInfo.key,
status: 'failure' as 'failure',
message: (error as Error).message,
});
}
}
return {
timestamp,
prewarmedKeys,
results,
};
} catch (error) {
console.error('Error prewarming cache:', error);
throw new Error('Failed to prewarm cache');
}
}
/**
* 清理过期缓存
* @param params 清理参数
* @param traceInfo 追踪信息
* @returns 清理结果
*/
public static async cleanExpiredCache(
params: {
pattern?: string;
},
traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}
): Promise<{
timestamp: string;
cleanedKeys: number;
pattern: string;
}> {
try {
const timestamp = new Date().toISOString();
const pattern = params.pattern || '*';
// 模拟清理过期缓存
const cleanedKeys = Math.floor(Math.random() * 1000) + 100;
return {
timestamp,
cleanedKeys,
pattern,
};
} catch (error) {
console.error('Error cleaning expired cache:', error);
throw new Error('Failed to clean expired cache');
}
}
/**
* 获取缓存性能指标
* @param params 指标参数
* @param traceInfo 追踪信息
* @returns 性能指标
*/
public static async getCacheMetrics(
params: {
metrics: string[];
},
traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}
): Promise<{
timestamp: string;
metrics: Record<string, any>;
}> {
try {
const timestamp = new Date().toISOString();
const metrics: Record<string, any> = {};
if (params.metrics.includes('cacheHitRate')) {
metrics.cacheHitRate = (Math.random() * 30 + 70).toFixed(2);
}
if (params.metrics.includes('cacheMissRate')) {
metrics.cacheMissRate = (Math.random() * 30 + 5).toFixed(2);
}
if (params.metrics.includes('averageCacheTime')) {
metrics.averageCacheTime = (Math.random() * 10 + 1).toFixed(2);
}
if (params.metrics.includes('totalCacheSize')) {
metrics.totalCacheSize = Math.floor(Math.random() * 1000000000) + 100000000;
}
if (params.metrics.includes('cacheKeysCount')) {
metrics.cacheKeysCount = Math.floor(Math.random() * 10000) + 1000;
}
return {
timestamp,
metrics,
};
} catch (error) {
console.error('Error getting cache metrics:', error);
throw new Error('Failed to get cache metrics');
}
}
}

123
server/src/services/utils/ConfigService.ts Normal file
View File

@@ -0,0 +1,123 @@
import db from '../config/database';
import { RedisService } from '../utils/RedisService';
export interface AppConfig {
key: string;
value: any;
description?: string;
isEnabled: boolean;
version?: number;
tenantId?: string;
created_at?: Date;
updated_at?: Date;
}
export class ConfigService {
private static TABLE_NAME = 'cf_config';
private static HISTORY_TABLE = 'cf_config_history';
/**
* [CORE_DEV_06/16] 初始化配置表与历史表
*/
static async initTable() {
const exists = await db.schema.hasTable(this.TABLE_NAME);
if (!exists) {
await db.schema.createTable(this.TABLE_NAME, (table) => {
table.string('key').notNullable();
table.string('tenantId').defaultTo('SYSTEM').index();
table.json('value');
table.string('description');
table.boolean('isEnabled').defaultTo(true);
table.integer('version').defaultTo(1); // [CORE_DEV_16] 版本控制
table.timestamps(true, true);
table.primary(['key', 'tenantId']);
});
// ... (seed data remains same)
}
const historyExists = await db.schema.hasTable(this.HISTORY_TABLE);
if (!historyExists) {
await db.schema.createTable(this.HISTORY_TABLE, (table) => {
table.increments('id').primary();
table.string('key').notNullable();
table.string('tenantId').notNullable();
table.json('value');
table.integer('version').notNullable();
table.string('changeLog');
table.timestamp('created_at').defaultTo(db.fn.now());
table.index(['key', 'tenantId']);
});
}
}
/**
* [CORE_DEV_16] 更新配置并保存历史版本
*/
static async updateConfig(key: string, data: Partial<AppConfig> & { changeLog?: string }, tenantId: string = 'SYSTEM') {
const current = await this.getConfig(key, tenantId);
const newVersion = (current?.version || 0) + 1;
const updateData: any = { ...data, version: newVersion };
delete updateData.changeLog;
if (data.value) updateData.value = JSON.stringify(data.value);
await db.transaction(async (trx) => {
// 1. 更新主表
await trx(this.TABLE_NAME).where({ key, tenantId }).update(updateData);
// 2. 插入历史记录
await trx(this.HISTORY_TABLE).insert({
key,
tenantId,
value: updateData.value || JSON.stringify(current?.value),
version: newVersion,
changeLog: data.changeLog || 'Automated update'
});
});
// [CORE_DEV_12] 分布式配置热更新通知
await RedisService.publishConfigChange(tenantId, key, data.value || data.isEnabled);
}
/**
* [CORE_DEV_16] 配置版本回滚
*/
static async rollbackConfig(key: string, version: number, tenantId: string = 'SYSTEM') {
const history = await db(this.HISTORY_TABLE).where({ key, tenantId, version }).first();
if (!history) throw new Error(`Version ${version} not found for config ${key}`);
await this.updateConfig(key, {
value: typeof history.value === 'string' ? JSON.parse(history.value) : history.value,
changeLog: `Rollback to version ${version}`
}, tenantId);
}
static async isFeatureEnabled(key: string, tenantId: string = 'SYSTEM'): Promise<boolean> {
const config = await this.getConfig(key, tenantId);
return config ? config.isEnabled : false;
}
/**
* 获取所有配置
*/
static async getAllConfigs(tenantId: string = 'SYSTEM'): Promise<AppConfig[]> {
const configs = await db(this.TABLE_NAME).where({ tenantId });
return configs.map(config => ({
...config,
value: typeof config.value === 'string' ? JSON.parse(config.value) : config.value
}));
}
/**
* 获取单个配置
*/
static async getConfig(key: string, tenantId: string = 'SYSTEM'): Promise<AppConfig | null> {
const config = await db(this.TABLE_NAME).where({ key, tenantId }).first();
if (!config) return null;
return {
...config,
value: typeof config.value === 'string' ? JSON.parse(config.value) : config.value
};
}
}

451
server/src/services/utils/EventBusService.ts Normal file
View File

@@ -0,0 +1,451 @@
import db from '../config/database';
import { logger } from '../utils/logger';
import RedisService from './RedisService';
import { BullMQService } from './BullMQService';
// 事件接口
export interface Event {
id: string;
type: string;
data: any;
metadata: EventMetadata;
createdAt: Date;
}
export interface EventMetadata {
source: string;
correlationId: string;
timestamp: number;
version: string;
}
// 事件处理器类型
export type EventHandler = (event: Event) => Promise<void>;
// 订阅接口
export interface Subscription {
id: string;
topic: string;
handler: EventHandler;
filter?: EventFilter;
createdAt: Date;
}
// 事件过滤接口
export interface EventFilter {
[key: string]: any;
}
// 路由规则接口
export interface RouteRule {
id: string;
topic: string;
condition: EventFilter;
destinations: string[];
createdAt: Date;
}
// 健康状态接口
export interface HealthStatus {
status: 'healthy' | 'unhealthy';
message?: string;
metrics: {
eventCount: number;
subscriptionCount: number;
queueSize: number;
};
}
/**
* [CORE_EVT_01] 事件总线服务
* @description 分布式事件驱动架构，支持事件发布、订阅、路由
*/
export class EventBusService {
private static readonly EVENT_TABLE = 'cf_events';
private static readonly SUBSCRIPTION_TABLE = 'cf_event_subscriptions';
private static readonly ROUTE_RULE_TABLE = 'cf_event_route_rules';
private static readonly DEAD_LETTER_TABLE = 'cf_event_dead_letters';
private static subscriptions: Map<string, Subscription> = new Map();
private static routeRules: RouteRule[] = [];
/**
* 初始化数据库表
*/
static async initTable() {
// 创建事件表
const hasEventTable = await db.schema.hasTable(this.EVENT_TABLE);
if (!hasEventTable) {
console.log(`📦 Creating ${this.EVENT_TABLE} table...`);
await db.schema.createTable(this.EVENT_TABLE, (table) => {
table.string('id', 64).primary();
table.string('type', 128).index();
table.json('data').notNullable();
table.json('metadata').notNullable();
table.timestamps(true, true);
});
}
// 创建订阅表
const hasSubscriptionTable = await db.schema.hasTable(this.SUBSCRIPTION_TABLE);
if (!hasSubscriptionTable) {
console.log(`📦 Creating ${this.SUBSCRIPTION_TABLE} table...`);
await db.schema.createTable(this.SUBSCRIPTION_TABLE, (table) => {
table.string('id', 64).primary();
table.string('topic', 128).index();
table.string('endpoint', 512).nullable();
table.json('filter').nullable();
table.timestamps(true, true);
});
}
// 创建路由规则表
const hasRouteRuleTable = await db.schema.hasTable(this.ROUTE_RULE_TABLE);
if (!hasRouteRuleTable) {
console.log(`📦 Creating ${this.ROUTE_RULE_TABLE} table...`);
await db.schema.createTable(this.ROUTE_RULE_TABLE, (table) => {
table.string('id', 64).primary();
table.string('topic', 128).index();
table.json('condition').notNullable();
table.json('destinations').notNullable();
table.timestamps(true, true);
});
}
// 创建死信表
const hasDeadLetterTable = await db.schema.hasTable(this.DEAD_LETTER_TABLE);
if (!hasDeadLetterTable) {
console.log(`📦 Creating ${this.DEAD_LETTER_TABLE} table...`);
await db.schema.createTable(this.DEAD_LETTER_TABLE, (table) => {
table.string('id', 64).primary();
table.string('event_id', 64).index();
table.string('topic', 128).index();
table.json('event_data').notNullable();
table.string('error', 1024).notNullable();
table.integer('retry_count').defaultTo(0);
table.timestamps(true, true);
});
}
// 加载路由规则
await this.loadRouteRules();
}
/**
* 加载路由规则
*/
private static async loadRouteRules() {
try {
const rules = await db(this.ROUTE_RULE_TABLE).select('*');
this.routeRules = rules.map(rule => ({
id: rule.id,
topic: rule.topic,
condition: rule.condition,
destinations: rule.destinations,
createdAt: rule.created_at
}));
logger.info(`[EventBus] Loaded ${this.routeRules.length} route rules`);
} catch (error) {
logger.error(`[EventBus] Failed to load route rules: ${error}`);
}
}
/**
* 发布事件
*/
static async publish(event: Partial<Event>): Promise<Event> {
const eventId = `EVT-${Date.now()}-${Math.floor(Math.random() * 10000)}`;
const now = new Date();
const fullEvent: Event = {
id: eventId,
type: event.type!,
data: event.data!,
metadata: {
source: event.metadata?.source || 'unknown',
correlationId: event.metadata?.correlationId || eventId,
timestamp: event.metadata?.timestamp || now.getTime(),
version: event.metadata?.version || '1.0'
},
createdAt: now
};
logger.info(`[EventBus] Publishing event: ${fullEvent.type} (${fullEvent.id})`);
try {
// 存储事件到数据库
await db(this.EVENT_TABLE).insert({
id: fullEvent.id,
type: fullEvent.type,
data: fullEvent.data,
metadata: fullEvent.metadata,
created_at: fullEvent.createdAt,
updated_at: fullEvent.createdAt
});
// 路由事件
const destinations = this.routeEvent(fullEvent);
logger.info(`[EventBus] Event ${fullEvent.id} routed to ${destinations.length} destinations`);
// 发送到消息队列
await BullMQService.addJob('event.published', {
event: fullEvent,
destinations
});
// 通知本地订阅者
await this.notifyLocalSubscribers(fullEvent);
return fullEvent;
} catch (error) {
logger.error(`[EventBus] Failed to publish event: ${error}`);
throw error;
}
}
/**
* 订阅事件
*/
static async subscribe(topic: string, handler: EventHandler, filter?: EventFilter): Promise<Subscription> {
const subscriptionId = `SUB-${Date.now()}-${Math.floor(Math.random() * 10000)}`;
const subscription: Subscription = {
id: subscriptionId,
topic,
handler,
filter,
createdAt: new Date()
};
this.subscriptions.set(subscriptionId, subscription);
logger.info(`[EventBus] New subscription: ${topic} (${subscriptionId})`);
return subscription;
}
/**
* 取消订阅
*/
static async unsubscribe(subscription: Subscription): Promise<void> {
this.subscriptions.delete(subscription.id);
logger.info(`[EventBus] Unsubscribed: ${subscription.topic} (${subscription.id})`);
}
/**
* 路由事件
*/
private static routeEvent(event: Event): string[] {
const destinations: string[] = [];
this.routeRules.forEach(rule => {
if (rule.topic === event.type && this.matchesCondition(event, rule.condition)) {
destinations.push(...rule.destinations);
}
});
return destinations;
}
/**
* 发射事件（publish别名）
*/
static async emit(eventType: string, data: any): Promise<Event> {
return this.publish({
type: eventType,
data
});
}
/**
* 检查事件是否匹配条件
*/
private static matchesCondition(event: Event, condition: EventFilter): boolean {
for (const [key, value] of Object.entries(condition)) {
const eventValue = this.getNestedValue(event, key);
if (eventValue !== value) {
return false;
}
}
return true;
}
/**
* 获取嵌套值
*/
private static getNestedValue(obj: any, path: string): any {
return path.split('.').reduce((acc, key) => acc?.[key], obj);
}
/**
* 通知本地订阅者
*/
private static async notifyLocalSubscribers(event: Event): Promise<void> {
Array.from(this.subscriptions.values()).forEach(async (subscription) => {
if (subscription.topic === event.type && (!subscription.filter || this.matchesCondition(event, subscription.filter))) {
try {
await subscription.handler(event);
} catch (error) {
logger.error(`[EventBus] Failed to notify subscriber ${subscription.id}: ${error}`);
await this.handleDeadLetter(event, error as Error);
}
}
});
}
/**
* 处理死信
*/
private static async handleDeadLetter(event: Event, error: Error): Promise<void> {
const deadLetterId = `DLQ-${Date.now()}-${Math.floor(Math.random() * 10000)}`;
await db(this.DEAD_LETTER_TABLE).insert({
id: deadLetterId,
event_id: event.id,
topic: event.type,
event_data: event,
error: error.message,
retry_count: 0,
created_at: new Date(),
updated_at: new Date()
});
logger.warn(`[EventBus] Event ${event.id} moved to dead letter queue: ${error.message}`);
}
/**
* 注册路由规则
*/
static async registerRouteRule(rule: Omit<RouteRule, 'id' | 'createdAt'>): Promise<RouteRule> {
const ruleId = `RULE-${Date.now()}-${Math.floor(Math.random() * 10000)}`;
const newRule: RouteRule = {
id: ruleId,
...rule,
createdAt: new Date()
};
await db(this.ROUTE_RULE_TABLE).insert({
id: newRule.id,
topic: newRule.topic,
condition: newRule.condition,
destinations: newRule.destinations,
created_at: newRule.createdAt,
updated_at: newRule.createdAt
});
this.routeRules.push(newRule);
logger.info(`[EventBus] Registered route rule: ${newRule.topic} (${newRule.id})`);
return newRule;
}
/**
* 获取事件历史
*/
static async getEventHistory(filter: { type?: string; startDate?: Date; endDate?: Date; source?: string; correlationId?: string }): Promise<Event[]> {
let query = db(this.EVENT_TABLE).select('*');
if (filter.type) {
query = query.where('type', filter.type);
}
if (filter.startDate) {
query = query.where('created_at', '>=', filter.startDate);
}
if (filter.endDate) {
query = query.where('created_at', '<=', filter.endDate);
}
if (filter.source) {
query = query.where('metadata', '@>', { source: filter.source });
}
if (filter.correlationId) {
query = query.where('metadata', '@>', { correlationId: filter.correlationId });
}
const events = await query.orderBy('created_at', 'desc');
return events.map(event => ({
id: event.id,
type: event.type,
data: event.data,
metadata: event.metadata,
createdAt: event.created_at
}));
}
/**
* 健康检查
*/
static async healthCheck(): Promise<HealthStatus> {
try {
// 检查数据库连接
await db.raw('SELECT 1');
// 获取统计信息
const eventCount = await db(this.EVENT_TABLE).count('id as count').first();
const subscriptionCount = await db(this.SUBSCRIPTION_TABLE).count('id as count').first();
const queueSize = await BullMQService.getQueueSize('event.published');
return {
status: 'healthy',
metrics: {
eventCount: parseInt((eventCount as any).count || '0'),
subscriptionCount: parseInt((subscriptionCount as any).count || '0'),
queueSize: queueSize || 0
}
};
} catch (error) {
return {
status: 'unhealthy',
message: error instanceof Error ? error.message : 'Unknown error',
metrics: {
eventCount: 0,
subscriptionCount: 0,
queueSize: 0
}
};
}
}
/**
* 重试死信事件
*/
static async retryDeadLetter(deadLetterId: string): Promise<boolean> {
try {
const deadLetter = await db(this.DEAD_LETTER_TABLE).where('id', deadLetterId).first();
if (!deadLetter) {
return false;
}
const event = deadLetter.event_data;
await this.publish(event);
// 更新重试计数
await db(this.DEAD_LETTER_TABLE).where('id', deadLetterId).update({
retry_count: deadLetter.retry_count + 1,
updated_at: new Date()
});
logger.info(`[EventBus] Retried dead letter event: ${event.id}`);
return true;
} catch (error) {
logger.error(`[EventBus] Failed to retry dead letter: ${error}`);
return false;
}
}
/**
* 清理过期事件
*/
static async cleanExpiredEvents(days: number = 30): Promise<number> {
const cutoffDate = new Date();
cutoffDate.setDate(cutoffDate.getDate() - days);
const deleted = await db(this.EVENT_TABLE)
.where('created_at', '<', cutoffDate)
.del();
logger.info(`[EventBus] Cleaned ${deleted} expired events`);
return deleted;
}
}

64
server/src/services/utils/LogAnalyticsService.ts Normal file
View File

@@ -0,0 +1,64 @@
import { logger } from '../utils/logger';
export interface LogCluster {
pattern: string;
count: number;
lastOccurrence: Date;
isAnomaly: boolean;
}
/**
* [CORE_DEV_10] 分布式日志聚合与异常聚类分析 (Log Clustering)
* @description 基于模式识别实现日志的自动分类与异常聚类，支持在大规模节点环境下快速定位系统异常
*/
export class LogAnalyticsService {
private static clusters: Map<string, LogCluster> = new Map();
/**
* 分析日志并归类
* @param message 日志消息
*/
static analyzeLog(message: string): void {
// 1. 提取模式：通过正则屏蔽动态变量（如 UUID, IP, Timestamp）
const pattern = message
.replace(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi, '<UUID>')
.replace(/\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}/g, '<IP>')
.replace(/\d+/g, '<NUM>')
.trim();
// 2. 更新或创建聚类
const cluster = this.clusters.get(pattern) || {
pattern,
count: 0,
lastOccurrence: new Date(),
isAnomaly: false
};
cluster.count++;
cluster.lastOccurrence = new Date();
// 3. 简单的异常检测逻辑 (模拟)
// 逻辑：如果错误类日志在短时间内激增，则标记为异常
if (message.toLowerCase().includes('error') && cluster.count > 100) {
cluster.isAnomaly = true;
logger.error(`[LogAnalytics] Anomaly detected for pattern: ${pattern}`);
}
this.clusters.set(pattern, cluster);
}
/**
* 获取当前活跃的日志聚类报告
*/
static getClusterReport(): LogCluster[] {
return Array.from(this.clusters.values()).sort((a, b) => b.count - a.count);
}
/**
* 模拟从多节点聚合日志
*/
static async ingestRemoteLogs(nodeId: string, logs: string[]): Promise<void> {
logger.info(`[LogAnalytics] Ingesting ${logs.length} logs from node: ${nodeId}`);
logs.forEach(log => this.analyzeLog(log));
}
}

163
server/src/services/utils/LogService.ts Normal file
View File

@@ -0,0 +1,163 @@
/**
* 统一日志服务
* 提供标准化的日志格式和方法
*/
interface LogContext {
service?: string;
method?: string;
tenantId?: string;
shopId?: string;
taskId?: string;
traceId?: string;
businessType?: 'TOC' | 'TOB';
[key: string]: any;
}
interface LogOptions {
context?: LogContext;
error?: Error;
data?: any;
}
export class LogService {
/**
* 生成标准日志格式
* @param level 日志级别
* @param message 日志消息
* @param options 日志选项
*/
private static formatLog(level: string, message: string, options: LogOptions = {}): string {
const timestamp = new Date().toISOString();
const context = options.context || {};
const data = options.data;
const error = options.error;
// 构建日志对象
const logObj = {
timestamp,
level,
message,
context,
...(data && { data }),
...(error && {
error: {
message: error.message,
stack: error.stack
}
})
};
return JSON.stringify(logObj);
}
/**
* 信息日志
* @param message 日志消息
* @param options 日志选项
*/
static info(message: string, options: LogOptions = {}): void {
const formattedLog = this.formatLog('INFO', message, options);
console.log(formattedLog);
}
/**
* 错误日志
* @param message 日志消息
* @param options 日志选项
*/
static error(message: string, options: LogOptions = {}): void {
const formattedLog = this.formatLog('ERROR', message, options);
console.error(formattedLog);
}
/**
* 警告日志
* @param message 日志消息
* @param options 日志选项
*/
static warn(message: string, options: LogOptions = {}): void {
const formattedLog = this.formatLog('WARN', message, options);
console.warn(formattedLog);
}
/**
* 调试日志
* @param message 日志消息
* @param options 日志选项
*/
static debug(message: string, options: LogOptions = {}): void {
if (process.env.NODE_ENV === 'development') {
const formattedLog = this.formatLog('DEBUG', message, options);
console.debug(formattedLog);
}
}
/**
* 追踪日志
* @param message 日志消息
* @param options 日志选项
*/
static trace(message: string, options: LogOptions = {}): void {
if (process.env.NODE_ENV === 'development') {
const formattedLog = this.formatLog('TRACE', message, options);
console.trace(formattedLog);
}
}
/**
* 创建服务特定的日志实例
* @param serviceName 服务名称
*/
static createLogger(serviceName: string) {
return {
info: (message: string, options: Omit<LogOptions, 'context'> & { context?: Omit<LogContext, 'service'> } = {}) => {
this.info(message, {
...options,
context: {
service: serviceName,
...options.context
}
});
},
error: (message: string, options: Omit<LogOptions, 'context'> & { context?: Omit<LogContext, 'service'> } = {}) => {
this.error(message, {
...options,
context: {
service: serviceName,
...options.context
}
});
},
warn: (message: string, options: Omit<LogOptions, 'context'> & { context?: Omit<LogContext, 'service'> } = {}) => {
this.warn(message, {
...options,
context: {
service: serviceName,
...options.context
}
});
},
debug: (message: string, options: Omit<LogOptions, 'context'> & { context?: Omit<LogContext, 'service'> } = {}) => {
this.debug(message, {
...options,
context: {
service: serviceName,
...options.context
}
});
},
trace: (message: string, options: Omit<LogOptions, 'context'> & { context?: Omit<LogContext, 'service'> } = {}) => {
this.trace(message, {
...options,
context: {
service: serviceName,
...options.context
}
});
}
};
}
}
export default LogService;

436
server/src/services/utils/RedisService.ts Normal file
View File

@@ -0,0 +1,436 @@
import { createClient, RedisClientType } from 'redis';
import { logger } from '../utils/logger';
class RedisService {
private client: RedisClientType;
private isConnected: boolean = false;
constructor() {
this.client = createClient({
url: process.env.REDIS_URL || 'redis://localhost:6379'
});
this.client.on('error', (err) => {
logger.error(`[RedisService] Error: ${err.message}`);
this.isConnected = false;
});
this.client.on('connect', () => {
logger.info('[RedisService] Connected to Redis');
this.isConnected = true;
});
this.client.on('end', () => {
logger.info('[RedisService] Disconnected from Redis');
this.isConnected = false;
});
this.connect();
}
private async connect() {
try {
await this.client.connect();
this.isConnected = true;
} catch (error) {
logger.error(`[RedisService] Failed to connect: ${(error as any).message}`);
this.isConnected = false;
}
}
/**
* 检查Redis连接状态
*/
public async checkConnection(): Promise<boolean> {
if (!this.isConnected) {
await this.connect();
}
return this.isConnected;
}
/**
* 设置缓存
*/
public async set(key: string, value: any, expireSeconds?: number): Promise<boolean> {
try {
if (!await this.checkConnection()) {
return false;
}
const stringValue = typeof value === 'string' ? value : JSON.stringify(value);
if (expireSeconds) {
await this.client.set(key, stringValue, { EX: expireSeconds });
} else {
await this.client.set(key, stringValue);
}
return true;
} catch (error) {
logger.warn(`[RedisService] Failed to set key ${key}: ${(error as any).message}`);
return false;
}
}
/**
* 获取缓存
*/
public async get(key: string): Promise<any> {
try {
if (!await this.checkConnection()) {
return null;
}
const value = await this.client.get(key);
if (!value) {
return null;
}
try {
return JSON.parse(value);
} catch {
return value;
}
} catch (error) {
logger.warn(`[RedisService] Failed to get key ${key}: ${(error as any).message}`);
return null;
}
}
/**
* 删除缓存
*/
public async del(key: string): Promise<boolean> {
try {
if (!await this.checkConnection()) {
return false;
}
await this.client.del(key);
return true;
} catch (error) {
logger.warn(`[RedisService] Failed to delete key ${key}: ${(error as any).message}`);
return false;
}
}
/**
* 批量删除缓存
*/
public async delMultiple(keys: string[]): Promise<boolean> {
try {
if (!await this.checkConnection() || keys.length === 0) {
return false;
}
await this.client.del(keys);
return true;
} catch (error) {
logger.warn(`[RedisService] Failed to delete multiple keys: ${(error as any).message}`);
return false;
}
}
/**
* 设置哈希表字段
*/
public async hset(key: string, field: string, value: any): Promise<boolean> {
try {
if (!await this.checkConnection()) {
return false;
}
const stringValue = typeof value === 'string' ? value : JSON.stringify(value);
await this.client.hSet(key, field, stringValue);
return true;
} catch (error) {
logger.warn(`[RedisService] Failed to hset key ${key}: ${(error as any).message}`);
return false;
}
}
/**
* 获取哈希表字段
*/
public async hget(key: string, field: string): Promise<any> {
try {
if (!await this.checkConnection()) {
return null;
}
const value = await this.client.hGet(key, field);
if (!value) {
return null;
}
try {
return JSON.parse(value);
} catch {
return value;
}
} catch (error) {
logger.warn(`[RedisService] Failed to hget key ${key}: ${(error as any).message}`);
return null;
}
}
/**
* 获取哈希表所有字段
*/
public async hgetall(key: string): Promise<Record<string, any>> {
try {
if (!await this.checkConnection()) {
return {};
}
const values = await this.client.hGetAll(key);
const result: Record<string, any> = {};
for (const [field, value] of Object.entries(values)) {
try {
result[field] = JSON.parse(value);
} catch {
result[field] = value;
}
}
return result;
} catch (error) {
logger.warn(`[RedisService] Failed to hgetall key ${key}: ${(error as any).message}`);
return {};
}
}
/**
* 增加计数器
*/
public async incr(key: string): Promise<number | null> {
try {
if (!await this.checkConnection()) {
return null;
}
return await this.client.incr(key);
} catch (error) {
logger.warn(`[RedisService] Failed to incr key ${key}: ${(error as any).message}`);
return null;
}
}
/**
* 根据模式删除缓存
*/
public async deletePattern(pattern: string): Promise<boolean> {
try {
if (!await this.checkConnection()) {
return false;
}
const keys = await this.client.keys(pattern);
if (keys.length > 0) {
await this.client.del(keys);
}
return true;
} catch (error) {
logger.warn(`[RedisService] Failed to delete pattern ${pattern}: ${(error as any).message}`);
return false;
}
}
/**
* 关闭连接
*/
public async close(): Promise<void> {
try {
await this.client.disconnect();
this.isConnected = false;
} catch (error) {
logger.warn(`[RedisService] Failed to close connection: ${(error as any).message}`);
}
}
/**
* Ping Redis服务器
*/
public async ping(): Promise<string> {
try {
if (!await this.checkConnection()) {
throw new Error('Redis not connected');
}
return await this.client.ping();
} catch (error) {
logger.warn(`[RedisService] Ping failed: ${(error as any).message}`);
throw error;
}
}
/**
* 获取匹配模式的所有键
*/
public async keys(pattern: string): Promise<string[]> {
try {
if (!await this.checkConnection()) {
return [];
}
return await this.client.keys(pattern);
} catch (error) {
logger.warn(`[RedisService] Failed to get keys for pattern ${pattern}: ${(error as any).message}`);
return [];
}
}
/**
* 关闭连接（quit别名）
*/
public async quit(): Promise<void> {
return this.close();
}
/**
* 列表左侧推入
*/
public async lpush(key: string, value: any): Promise<number | null> {
try {
if (!await this.checkConnection()) {
return null;
}
const stringValue = typeof value === 'string' ? value : JSON.stringify(value);
return await this.client.lPush(key, stringValue);
} catch (error) {
logger.warn(`[RedisService] Failed to lpush key ${key}: ${(error as any).message}`);
return null;
}
}
/**
* 列表裁剪
*/
public async ltrim(key: string, start: number, stop: number): Promise<boolean> {
try {
if (!await this.checkConnection()) {
return false;
}
await this.client.lTrim(key, start, stop);
return true;
} catch (error) {
logger.warn(`[RedisService] Failed to ltrim key ${key}: ${(error as any).message}`);
return false;
}
}
/**
* 获取键的TTL
*/
public async ttl(key: string): Promise<number> {
try {
if (!await this.checkConnection()) {
return -1;
}
return await this.client.ttl(key);
} catch (error) {
logger.warn(`[RedisService] Failed to get ttl for key ${key}: ${(error as any).message}`);
return -1;
}
}
/**
* 设置键的过期时间
*/
public async expire(key: string, seconds: number): Promise<boolean> {
try {
if (!await this.checkConnection()) {
return false;
}
const result = await this.client.expire(key, seconds);
return result === 1;
} catch (error) {
logger.warn(`[RedisService] Failed to set expire for key ${key}: ${(error as any).message}`);
return false;
}
}
/**
* 执行Lua脚本
*/
public async eval(script: string, keys: string[], args: (string | number)[]): Promise<any> {
try {
if (!await this.checkConnection()) {
return null;
}
return await this.client.eval(script, { keys, arguments: args.map(String) });
} catch (error) {
logger.warn(`[RedisService] Failed to eval script: ${(error as any).message}`);
return null;
}
}
/**
* 发布消息
*/
public async publish(channel: string, message: string): Promise<number> {
try {
if (!await this.checkConnection()) {
return 0;
}
return await this.client.publish(channel, message);
} catch (error) {
logger.warn(`[RedisService] Failed to publish to channel ${channel}: ${(error as any).message}`);
return 0;
}
}
/**
* 订阅频道
*/
public async subscribe(channel: string, callback: (message: string) => void): Promise<void> {
try {
if (!await this.checkConnection()) {
return;
}
const subscriber = this.client.duplicate();
await subscriber.connect();
await subscriber.subscribe(channel, callback);
} catch (error) {
logger.warn(`[RedisService] Failed to subscribe to channel ${channel}: ${(error as any).message}`);
}
}
/**
* 获取列表范围内的元素
*/
public async lrange(key: string, start: number, stop: number): Promise<string[]> {
try {
if (!await this.checkConnection()) {
return [];
}
return await this.client.lRange(key, start, stop);
} catch (error) {
logger.warn(`[RedisService] Failed to lrange key ${key}: ${(error as any).message}`);
return [];
}
}
/**
* 设置缓存并指定过期时间 (setex 别名)
*/
public async setex(key: string, expireSeconds: number, value: any): Promise<boolean> {
return this.set(key, value, expireSeconds);
}
/**
* 删除缓存 (del 别名)
*/
public async delete(key: string): Promise<boolean> {
return this.del(key);
}
}
export default new RedisService();

275
server/src/services/utils/ServiceHealthCheck.ts Normal file
View File

@@ -0,0 +1,275 @@
/**
* [OP-MV002] 商户服务健康检查服务
* @description 检查商户服务的健康状态，生成健康报告
* @version 1.0
*/
export class ServiceHealthCheck {
/**
* 检查商户服务健康状态
* @param params 检查参数
* @param traceInfo 追踪信息
* @returns 健康检查结果
*/
public static async checkMerchantServiceHealth(params: {
merchantId?: string;
services?: string[];
}, traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}): Promise<{
overallStatus: 'healthy' | 'degraded' | 'unhealthy';
services: Array<{
serviceName: string;
status: 'healthy' | 'degraded' | 'unhealthy';
responseTime: number;
errorCount: number;
lastChecked: string;
}>;
recommendations: string[];
}> {
// 模拟服务列表
const servicesToCheck = params.services || [
'inventory',
'order',
'payment',
'shipping',
'notification'
];
// 模拟健康检查结果
const services = servicesToCheck.map(service => {
// 生成随机响应时间和错误数
const responseTime = Math.floor(Math.random() * 500) + 50;
const errorCount = Math.floor(Math.random() * 5);
// 计算服务状态
let status: 'healthy' | 'degraded' | 'unhealthy' = 'healthy';
if (errorCount > 3 || responseTime > 400) {
status = 'unhealthy';
} else if (errorCount > 1 || responseTime > 200) {
status = 'degraded';
}
return {
serviceName: service,
status,
responseTime,
errorCount,
lastChecked: new Date().toISOString()
};
});
// 计算整体状态
const unhealthyCount = services.filter(s => s.status === 'unhealthy').length;
const degradedCount = services.filter(s => s.status === 'degraded').length;
let overallStatus: 'healthy' | 'degraded' | 'unhealthy' = 'healthy';
if (unhealthyCount > 0) {
overallStatus = 'unhealthy';
} else if (degradedCount > 0) {
overallStatus = 'degraded';
}
// 生成建议
const recommendations: string[] = [];
if (overallStatus === 'unhealthy') {
recommendations.push('立即检查不健康的服务');
recommendations.push('考虑重启相关服务');
} else if (overallStatus === 'degraded') {
recommendations.push('检查性能瓶颈');
recommendations.push('优化响应时间');
} else {
recommendations.push('保持当前状态');
recommendations.push('定期进行健康检查');
}
// 记录健康检查日志
console.log(`[ServiceHealthCheck] 健康检查完成 - 整体状态: ${overallStatus}`, {
...traceInfo,
servicesCount: services.length,
unhealthyCount,
degradedCount
});
return {
overallStatus,
services,
recommendations
};
}
/**
* 生成健康报告
* @param merchantId 商户ID
* @param traceInfo 追踪信息
* @returns 健康报告
*/
public static async generateHealthReport(merchantId: string, traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}): Promise<{
reportId: string;
merchantId: string;
timestamp: string;
overallStatus: string;
serviceDetails: any;
performanceMetrics: any;
recommendations: string[];
}> {
// 生成报告ID
const reportId = `HR-${Date.now()}`;
const timestamp = new Date().toISOString();
// 模拟服务健康数据
const serviceDetails = {
inventory: {
status: 'healthy',
responseTime: 120,
errorRate: 0.01
},
order: {
status: 'degraded',
responseTime: 250,
errorRate: 0.05
},
payment: {
status: 'healthy',
responseTime: 90,
errorRate: 0.005
},
shipping: {
status: 'healthy',
responseTime: 150,
errorRate: 0.02
},
notification: {
status: 'healthy',
responseTime: 80,
errorRate: 0.01
}
};
// 计算整体状态
const overallStatus = Object.values(serviceDetails).some(s => s.status === 'unhealthy')
? 'unhealthy'
: Object.values(serviceDetails).some(s => s.status === 'degraded')
? 'degraded'
: 'healthy';
// 计算性能指标
const performanceMetrics = {
averageResponseTime: Object.values(serviceDetails).reduce((sum, s) => sum + s.responseTime, 0) / Object.values(serviceDetails).length,
averageErrorRate: Object.values(serviceDetails).reduce((sum, s) => sum + s.errorRate, 0) / Object.values(serviceDetails).length,
healthyServices: Object.values(serviceDetails).filter(s => s.status === 'healthy').length,
totalServices: Object.values(serviceDetails).length
};
// 生成建议
const recommendations = [
overallStatus === 'unhealthy' ? '立即修复不健康的服务' : '保持服务健康状态',
performanceMetrics.averageResponseTime > 200 ? '优化服务响应时间' : '响应时间正常',
performanceMetrics.averageErrorRate > 0.05 ? '减少服务错误率' : '错误率在可接受范围内'
];
// 记录报告生成日志
console.log(`[ServiceHealthCheck] 生成健康报告 - ID: ${reportId}, 商户: ${merchantId}`, {
...traceInfo,
reportId
});
return {
reportId,
merchantId,
timestamp,
overallStatus,
serviceDetails,
performanceMetrics,
recommendations
};
}
/**
* 执行服务可用性测试
* @param params 测试参数
* @param traceInfo 追踪信息
* @returns 测试结果
*/
public static async testServiceAvailability(params: {
merchantId: string;
services: string[];
testDuration: number; // 测试持续时间（秒）
}, traceInfo: {
tenantId: string;
shopId: string;
taskId: string;
traceId: string;
businessType: 'TOC' | 'TOB';
}): Promise<{
testId: string;
merchantId: string;
startTime: string;
endTime: string;
results: Array<{
serviceName: string;
availability: number; // 可用性百分比
responseTimes: number[];
errorCount: number;
}>;
}> {
// 生成测试ID
const testId = `AT-${Date.now()}`;
const startTime = new Date().toISOString();
// 模拟测试结果
const results = params.services.map(service => {
// 生成模拟响应时间和错误数
const responseTimes: number[] = [];
let errorCount = 0;
// 模拟测试过程
for (let i = 0; i < 10; i++) {
const responseTime = Math.floor(Math.random() * 300) + 50;
responseTimes.push(responseTime);
if (Math.random() < 0.05) {
errorCount++;
}
}
// 计算可用性
const availability = ((10 - errorCount) / 10) * 100;
return {
serviceName: service,
availability,
responseTimes,
errorCount
};
});
// 模拟测试持续时间
await new Promise(resolve => setTimeout(resolve, params.testDuration * 1000));
const endTime = new Date().toISOString();
// 记录测试日志
console.log(`[ServiceHealthCheck] 服务可用性测试完成 - ID: ${testId}, 商户: ${params.merchantId}`, {
...traceInfo,
testId,
duration: params.testDuration
});
return {
testId,
merchantId: params.merchantId,
startTime,
endTime,
results
};
}
}

239
server/src/services/utils/ServiceManagementService.ts Normal file
View File

@@ -0,0 +1,239 @@
import { ServiceConfig, ServiceStatus, ServiceHealthCheck, ServiceStatusType, ServiceHealthStatus } from '../types/service';
import { logger } from '../utils/logger';
/**
* 服务管理服务
* 负责管理系统中的所有服务，包括服务的注册、启动、停止、监控等
*/
export class ServiceManagementService {
private static services: Map<string, ServiceConfig> = new Map();
private static serviceHealth: Map<string, ServiceStatus> = new Map();
/**
* 注册服务
* @param service 服务配置
* @returns 注册结果
*/
static async registerService(service: ServiceConfig): Promise<{ success: boolean; message: string }> {
try {
if (this.services.has(service.id)) {
return { success: false, message: 'Service already registered' };
}
this.services.set(service.id, service);
this.serviceHealth.set(service.id, { status: ServiceStatusType.STOPPED, lastCheck: new Date() });
logger.info(`Service ${service.id} registered successfully`);
return { success: true, message: 'Service registered successfully' };
} catch (error) {
logger.error(`Error registering service: ${error}`);
return { success: false, message: `Error registering service: ${error}` };
}
}
/**
* 启动服务
* @param serviceId 服务ID
* @returns 启动结果
*/
static async startService(serviceId: string): Promise<{ success: boolean; message: string }> {
try {
const service = this.services.get(serviceId);
if (!service) {
return { success: false, message: 'Service not found' };
}
// 模拟服务启动
await new Promise(resolve => setTimeout(resolve, 1000));
this.serviceHealth.set(serviceId, { status: ServiceStatusType.RUNNING, lastCheck: new Date() });
logger.info(`Service ${serviceId} started successfully`);
return { success: true, message: 'Service started successfully' };
} catch (error) {
logger.error(`Error starting service: ${error}`);
return { success: false, message: `Error starting service: ${error}` };
}
}
/**
* 停止服务
* @param serviceId 服务ID
* @returns 停止结果
*/
static async stopService(serviceId: string): Promise<{ success: boolean; message: string }> {
try {
const service = this.services.get(serviceId);
if (!service) {
return { success: false, message: 'Service not found' };
}
// 模拟服务停止
await new Promise(resolve => setTimeout(resolve, 500));
this.serviceHealth.set(serviceId, { status: ServiceStatusType.STOPPED, lastCheck: new Date() });
logger.info(`Service ${serviceId} stopped successfully`);
return { success: true, message: 'Service stopped successfully' };
} catch (error) {
logger.error(`Error stopping service: ${error}`);
return { success: false, message: `Error stopping service: ${error}` };
}
}
/**
* 获取服务状态
* @param serviceId 服务ID
* @returns 服务状态
*/
static async getServiceStatus(serviceId: string): Promise<ServiceStatus | null> {
try {
const status = this.serviceHealth.get(serviceId);
if (!status) {
return null;
}
// 更新最后检查时间
status.lastCheck = new Date();
this.serviceHealth.set(serviceId, status);
return status;
} catch (error) {
logger.error(`Error getting service status: ${error}`);
return null;
}
}
/**
* 获取所有服务列表
* @returns 服务列表
*/
static async listServices(): Promise<Array<{ id: string; config: ServiceConfig; status: ServiceStatus }>> {
try {
const services: Array<{ id: string; config: ServiceConfig; status: ServiceStatus }> = [];
for (const [id, config] of this.services.entries()) {
const status = this.serviceHealth.get(id) || { status: ServiceStatusType.UNKNOWN, lastCheck: new Date() };
services.push({ id, config, status });
}
return services;
} catch (error) {
logger.error(`Error listing services: ${error}`);
return [];
}
}
/**
* 检查服务健康状态
* @param serviceId 服务ID
* @returns 健康检查结果
*/
static async checkServiceHealth(serviceId: string): Promise<ServiceHealthCheck> {
try {
const service = this.services.get(serviceId);
if (!service) {
return {
serviceId,
status: ServiceHealthStatus.ERROR,
message: 'Service not found',
timestamp: new Date()
};
}
const status = this.serviceHealth.get(serviceId);
if (!status) {
return {
serviceId,
status: ServiceHealthStatus.ERROR,
message: 'Service status not found',
timestamp: new Date()
};
}
// 模拟健康检查
const isHealthy = status.status === ServiceStatusType.RUNNING;
return {
serviceId,
status: isHealthy ? ServiceHealthStatus.HEALTHY : ServiceHealthStatus.UNHEALTHY,
message: isHealthy ? 'Service is running normally' : 'Service is not running',
timestamp: new Date()
};
} catch (error) {
logger.error(`Error checking service health: ${error}`);
return {
serviceId,
status: ServiceHealthStatus.ERROR,
message: `Error checking health: ${error}`,
timestamp: new Date()
};
}
}
/**
* 检查所有服务健康状态
* @returns 健康检查结果列表
*/
static async checkAllServicesHealth(): Promise<ServiceHealthCheck[]> {
try {
const healthChecks: ServiceHealthCheck[] = [];
for (const serviceId of this.services.keys()) {
const healthCheck = await this.checkServiceHealth(serviceId);
healthChecks.push(healthCheck);
}
return healthChecks;
} catch (error) {
logger.error(`Error checking all services health: ${error}`);
return [];
}
}
/**
* 重启服务
* @param serviceId 服务ID
* @returns 重启结果
*/
static async restartService(serviceId: string): Promise<{ success: boolean; message: string }> {
try {
// 先停止服务
const stopResult = await this.stopService(serviceId);
if (!stopResult.success) {
return stopResult;
}
// 再启动服务
return await this.startService(serviceId);
} catch (error) {
logger.error(`Error restarting service: ${error}`);
return { success: false, message: `Error restarting service: ${error}` };
}
}
/**
* 卸载服务
* @param serviceId 服务ID
* @returns 卸载结果
*/
static async unregisterService(serviceId: string): Promise<{ success: boolean; message: string }> {
try {
const service = this.services.get(serviceId);
if (!service) {
return { success: false, message: 'Service not found' };
}
// 先停止服务
await this.stopService(serviceId);
// 移除服务
this.services.delete(serviceId);
this.serviceHealth.delete(serviceId);
logger.info(`Service ${serviceId} unregistered successfully`);
return { success: true, message: 'Service unregistered successfully' };
} catch (error) {
logger.error(`Error unregistering service: ${error}`);
return { success: false, message: `Error unregistering service: ${error}` };
}
}
}