274 lines
13 KiB
TypeScript
274 lines
13 KiB
TypeScript
// src/lib/monitor/worker.ts
|
||
import { getDb } from '@/lib/db'
|
||
import cron from 'node-cron'
|
||
import { MailMonitor } from './mail-monitor'
|
||
import { TicketProcessor } from './ticket-processor'
|
||
import { WeChatPusher } from './wechat-pusher'
|
||
import { AvailabilityEngine } from './availability-engine'
|
||
import { ReminderEngine } from './reminder-engine'
|
||
import { getMonitorConfig, getSetting, setSetting } from './settings-manager'
|
||
import type { MonitorConfig } from './types'
|
||
import { formatBeijingTime } from './types'
|
||
|
||
const logger = {
|
||
info: (msg: string) => console.log(`[Worker] ${formatBeijingTime()} INFO ${msg}`),
|
||
error: (msg: string) => console.error(`[Worker] ${formatBeijingTime()} ERROR ${msg}`),
|
||
}
|
||
|
||
export class BackgroundWorker {
|
||
private mailMonitor = new MailMonitor()
|
||
private ticketProcessor = new TicketProcessor()
|
||
private wechatPusher = new WeChatPusher()
|
||
private availabilityEngine = new AvailabilityEngine()
|
||
private reminderEngine = new ReminderEngine()
|
||
private processingPromise: Promise<void> | null = null
|
||
cronJob: any = null
|
||
private lastTickTime = 0
|
||
private db = getDb()
|
||
|
||
start(): void {
|
||
if (this.cronJob) return
|
||
this.cronJob = cron.schedule('* * * * *', () => { this.tick() })
|
||
logger.info('Cron job started (every 60 seconds)')
|
||
}
|
||
|
||
stop(): void {
|
||
if (this.cronJob) {
|
||
this.cronJob.stop()
|
||
this.cronJob = null
|
||
}
|
||
this.mailMonitor.disconnect()
|
||
logger.info('Worker stopped')
|
||
}
|
||
|
||
async tick(): Promise<void> {
|
||
if (this.processingPromise) return
|
||
this.processingPromise = this.doTick()
|
||
try { await this.processingPromise }
|
||
finally { this.processingPromise = null }
|
||
}
|
||
|
||
private async doTick(): Promise<void> {
|
||
const startTime = Date.now()
|
||
try {
|
||
const config = getMonitorConfig()
|
||
if (!config.monitor.enabled) return
|
||
|
||
// 检查是否被 API 手动触发
|
||
const triggerRequested = getSetting('monitor.trigger_requested') === 'true'
|
||
if (triggerRequested) {
|
||
setSetting('monitor.trigger_requested', 'false', 'monitor')
|
||
// 立即执行，忽略 interval_seconds
|
||
} else if (Date.now() - this.lastTickTime < config.monitor.interval_seconds * 1000) {
|
||
return
|
||
}
|
||
this.lastTickTime = Date.now()
|
||
|
||
// Step 3: 检查恢复
|
||
await this.checkRecoveryUpdates(config)
|
||
|
||
// Step 4: 7 点 flush
|
||
if (this.reminderEngine.isTimeToFlush()) {
|
||
await this.reminderEngine.flushPendingMessages(
|
||
(text) => this.wechatPusher.pushText(text, config.wechat.webhook_url),
|
||
(ip, sn) => this.ticketProcessor.getRackPosition(ip, sn)
|
||
)
|
||
this.reminderEngine.markFlushed()
|
||
}
|
||
|
||
// Step 5-6: 检查提醒
|
||
const pushText = (text: string) => this.wechatPusher.pushText(text, config.wechat.webhook_url)
|
||
const getRack = (ip: string | null, sn: string | null) => this.ticketProcessor.getRackPosition(ip, sn)
|
||
await this.reminderEngine.checkOemDiagReminders(pushText, config.monitor.push_delay_ms, getRack, config)
|
||
await this.reminderEngine.checkOemRepairReminders(pushText, config.monitor.push_delay_ms, getRack, config)
|
||
|
||
// Step 7: IMAP 连接
|
||
if (!this.mailMonitor.isConnected()) {
|
||
try { await this.mailMonitor.connect(config.mail) }
|
||
catch (e) { await this.mailMonitor.reconnect(config.mail) }
|
||
}
|
||
|
||
// Step 8-9: 搜索并处理邮件
|
||
const emails = await this.mailMonitor.fetchUnread(config)
|
||
let processed = 0, errors = 0
|
||
|
||
for (const email of emails) {
|
||
try {
|
||
// 去重检查
|
||
const existing = this.db.prepare('SELECT msg_id FROM processed_emails WHERE msg_id = ?').get(email.msg_id)
|
||
if (existing) continue
|
||
|
||
// 过滤
|
||
const { pass, type } = this.mailMonitor.filterEmail(email, config.filter)
|
||
if (!pass || !type) continue
|
||
|
||
// 提取表格和故障信息
|
||
const tableData = email.html ? this.mailMonitor.extractTable(email.html) : null
|
||
const faultInfo = email.html ? this.mailMonitor.extractFaultInfo(email.html, tableData) : null
|
||
if (!faultInfo?.order_number || !faultInfo?.fault_time) continue
|
||
|
||
// 创建工单
|
||
const content = faultInfo.fault_detail || ''
|
||
this.ticketProcessor.createTicketFromEmail(faultInfo, type, config, content)
|
||
|
||
// 获取机架位置
|
||
const rackPosition = await this.ticketProcessor.getRackPosition(faultInfo.server_ip, faultInfo.server_sn)
|
||
|
||
// 计算可用性
|
||
const monthKey = faultInfo.fault_time.slice(0, 7)
|
||
let deadlines: Record<string, { deadline: string; hours: number }> = {}
|
||
let oemDeadline: string | null = null
|
||
if (type === 'oem_repair') {
|
||
deadlines = this.availabilityEngine.calculateTierDeadlines(monthKey, faultInfo.server_sn!, faultInfo.fault_time)
|
||
} else {
|
||
oemDeadline = this.availabilityEngine.calculateOemDiagDeadline(monthKey, faultInfo.fault_time)
|
||
}
|
||
|
||
// 推送微信
|
||
const message = this.wechatPusher.formatAvailabilityMessage(deadlines, faultInfo, type === 'oem_diag', oemDeadline, faultInfo.order_number, rackPosition, faultInfo.fault_detail)
|
||
await this.wechatPusher.pushText(message, config.wechat.webhook_url)
|
||
|
||
// 记录故障
|
||
this.ticketProcessor.recordFault({
|
||
server_sn: faultInfo.server_sn!, server_ip: faultInfo.server_ip, order_number: faultInfo.order_number,
|
||
fault_type: type, fault_time: faultInfo.fault_time!, fault_detail: faultInfo.fault_detail, status: 'ongoing',
|
||
recovery_time: null, duration_seconds: null,
|
||
})
|
||
|
||
// 创建提醒
|
||
if (type === 'oem_diag' && oemDeadline) {
|
||
this.reminderEngine.createReminder({
|
||
server_sn: faultInfo.server_sn!, server_ip: faultInfo.server_ip, order_number: faultInfo.order_number,
|
||
reminder_type: 'oem_diag', deadline_time: oemDeadline, deadline_hours: null,
|
||
fault_time: faultInfo.fault_time, tier_name: null, notification_phase: 'none', notified_at: null,
|
||
})
|
||
} else if (type === 'oem_repair') {
|
||
for (const [tierName, tierData] of Object.entries(deadlines)) {
|
||
this.reminderEngine.createReminder({
|
||
server_sn: faultInfo.server_sn!, server_ip: faultInfo.server_ip, order_number: faultInfo.order_number,
|
||
reminder_type: 'oem_repair', deadline_time: tierData.deadline, deadline_hours: tierData.hours,
|
||
fault_time: faultInfo.fault_time, tier_name: tierName, notification_phase: 'none', notified_at: null,
|
||
})
|
||
}
|
||
}
|
||
|
||
// 标记已处理
|
||
this.db.prepare("INSERT OR IGNORE INTO processed_emails (msg_id, subject) VALUES (?, ?)").run(email.msg_id, email.subject)
|
||
processed++
|
||
} catch (e) {
|
||
errors++
|
||
logger.error(`Email error: ${e instanceof Error ? e.message : e}`)
|
||
}
|
||
}
|
||
|
||
// NOOP 保活
|
||
await this.mailMonitor.keepAlive()
|
||
|
||
// 日志清理（每天北京时间 3 点）
|
||
const nowBeijing = formatBeijingTime()
|
||
const hour = parseInt(nowBeijing.slice(11, 13))
|
||
const minute = parseInt(nowBeijing.slice(14, 16))
|
||
if (hour === 3 && minute === 0) {
|
||
this.db.prepare("DELETE FROM monitor_logs WHERE created_at < datetime('now', '+8 hours', '-90 days')").run()
|
||
this.db.prepare("DELETE FROM processed_emails WHERE processed_at < datetime('now', '+8 hours', '-90 days')").run()
|
||
this.db.prepare("DELETE FROM pending_messages WHERE created_at < datetime('now', '+8 hours', '-24 hours')").run()
|
||
}
|
||
|
||
// 记录 tick
|
||
const duration = Date.now() - startTime
|
||
this.db.prepare("INSERT INTO monitor_logs (action, details) VALUES (?, ?)").run(
|
||
'tick_complete', JSON.stringify({ processed, errors, duration, timestamp: formatBeijingTime() })
|
||
)
|
||
if (processed > 0) logger.info(`Tick complete: ${processed} processed, ${errors} errors, ${duration}ms`)
|
||
} catch (e) {
|
||
const errorMsg = e instanceof Error ? e.message : String(e)
|
||
logger.error(`Tick error: ${errorMsg}`)
|
||
// 将错误写入数据库，便于状态 API 检测
|
||
try {
|
||
this.db.prepare("INSERT INTO monitor_logs (action, details) VALUES (?, ?)").run(
|
||
'error', JSON.stringify({ error: errorMsg, timestamp: formatBeijingTime() })
|
||
)
|
||
} catch { /* 忽略日志写入失败 */ }
|
||
}
|
||
}
|
||
|
||
private async checkRecoveryUpdates(config: MonitorConfig): Promise<void> {
|
||
const ongoing = this.db.prepare("SELECT * FROM fault_records WHERE status = 'ongoing'").all() as {
|
||
id: number; order_number: string | null; server_sn: string; server_ip: string | null;
|
||
fault_time: string; fault_type: 'oem_diag' | 'oem_repair' | null
|
||
}[]
|
||
for (const fault of ongoing) {
|
||
if (!fault.order_number) continue
|
||
if (!fault.fault_time) continue
|
||
const ticket = this.db.prepare('SELECT current_status, close_time FROM tickets WHERE id = ?').get(parseInt(fault.order_number)) as { current_status: string; close_time: string | null } | undefined
|
||
if (ticket && ['resolved', 'closed'].includes(ticket.current_status)) {
|
||
const recoveryTime = ticket.close_time || formatBeijingTime(new Date())
|
||
const faultDate = new Date(fault.fault_time.replace(' ', 'T') + '+08:00')
|
||
const recoveryDate = new Date(recoveryTime.replace(' ', 'T') + '+08:00')
|
||
const duration = Math.floor((recoveryDate.getTime() - faultDate.getTime()) / 1000)
|
||
|
||
// 获取机架位置
|
||
const rackPosition = await this.ticketProcessor.getRackPosition(fault.server_ip, fault.server_sn)
|
||
|
||
// 格式化处理时长
|
||
const hours = Math.floor(duration / 3600)
|
||
const minutes = Math.floor((duration % 3600) / 60)
|
||
const durationStr = `${hours}小时${minutes}分钟`
|
||
|
||
// 按故障类型构建结单消息
|
||
let message: string
|
||
if (fault.fault_type === 'oem_diag') {
|
||
message = `✅ ${fault.order_number}工单已结单\n\n`
|
||
+ `服务器IP：${fault.server_ip || '未知'}\n`
|
||
+ `服务器SN：${fault.server_sn}\n`
|
||
+ (rackPosition ? `机架位置：${rackPosition}\n` : '')
|
||
+ `故障类型：OEM诊断\n`
|
||
+ `故障时间：${fault.fault_time}\n`
|
||
+ `结单时间：${recoveryTime}\n`
|
||
+ `本次处理时长：${durationStr}`
|
||
} else if (fault.fault_type === 'oem_repair') {
|
||
const monthKey = fault.fault_time.substring(0, 7)
|
||
const monthCount = (this.db.prepare(
|
||
"SELECT COUNT(*) as cnt FROM fault_records WHERE server_sn = ? AND status = 'resolved' AND strftime('%Y-%m', fault_time) = ?"
|
||
).get(fault.server_sn, monthKey) as { cnt: number } | undefined) || { cnt: 0 }
|
||
const monthTotal = (this.db.prepare(
|
||
"SELECT COALESCE(SUM(duration_seconds), 0) as total FROM fault_records WHERE server_sn = ? AND status = 'resolved' AND strftime('%Y-%m', fault_time) = ?"
|
||
).get(fault.server_sn, monthKey) as { total: number } | undefined) || { total: 0 }
|
||
// 先累加秒数再转时分（避免分钟进位问题）
|
||
const totalSec = monthTotal.total + duration
|
||
const totalHours = Math.floor(totalSec / 3600)
|
||
const totalMinutes = Math.floor((totalSec % 3600) / 60)
|
||
|
||
message = `✅ ${fault.order_number}工单已结单\n\n`
|
||
+ `服务器IP：${fault.server_ip || '未知'}\n`
|
||
+ `服务器SN：${fault.server_sn}\n`
|
||
+ (rackPosition ? `机架位置：${rackPosition}\n` : '')
|
||
+ `故障类型：OEM维修\n`
|
||
+ `故障时间：${fault.fault_time}\n`
|
||
+ `结单时间：${recoveryTime}\n`
|
||
+ `本次处理时长：${durationStr}\n`
|
||
+ `本月故障次数：${monthCount.cnt + 1}次\n`
|
||
+ `本月总处理时长：${totalHours}小时${totalMinutes}分钟`
|
||
} else {
|
||
message = `✅ ${fault.order_number}工单已结单\n\n`
|
||
+ `服务器IP：${fault.server_ip || '未知'}\n`
|
||
+ `服务器SN：${fault.server_sn}\n`
|
||
+ (rackPosition ? `机架位置：${rackPosition}\n` : '')
|
||
+ `结单时间：${recoveryTime}\n`
|
||
+ `本次处理时长：${durationStr}`
|
||
}
|
||
|
||
// 先推送消息，成功后再更新数据库（防止推送失败导致通知丢失）
|
||
const pushed = await this.wechatPusher.pushText(message, config.wechat.webhook_url)
|
||
if (pushed) {
|
||
this.db.prepare("UPDATE fault_records SET status = 'resolved', recovery_time = ?, duration_seconds = ? WHERE id = ?").run(recoveryTime, duration, fault.id)
|
||
this.reminderEngine.cleanupForTicket(fault.order_number)
|
||
logger.info(`Recovery detected: ${fault.order_number}`)
|
||
} else {
|
||
logger.error(`Recovery push failed, will retry next tick: ${fault.order_number}`)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|