'use strict'; // 每日进销存 · 定时截图通知 // // 执行流程: // ① 单实例锁(MySQL GET_LOCK),避免多实例重复截图推送 // ② 校验:功能开启 / 有收件人 / 通知中心已配置 / 对象存储已配置,任一不满足就记录原因并退出 // ③ 按配置时间(默认 01:00 Asia/Shanghai)触发;业务日期 = 北京时间昨日(与页面 T-1 口径一致) // ④ 截图 → 上传对象存储 → 拿到公开 HTTPS 地址后才发通知 // ⑤ 逐用户发送,用「业务日期 + 收件人」幂等键去重;单人失败不影响其他人 // ⑥ 截图/上传/通知失败均做有限次数重试与指数退避 // ⑦ 每次执行都落一条记录(时间 / 截图地址 / 通知对象 / 成功状态 / 失败原因) // // 立即测试(mode='test'):走同一条链路,但幂等键带随机数,**不写入正式幂等记录**, // 因此不会影响下一次正式定时推送。 const crypto = require('crypto'); const cron = require('node-cron'); const database = require('../config/database'); const { appendRunRecord, buildIdempotencyKey, getDailyInventoryNotificationRuntimeConfig, getObjectStorageStatus, hasNotified, loadDailyInventoryNotificationState, markNotified, normalizeRecipients, } = require('../services/dailyInventoryNotificationSettings'); const { captureAndUploadDailyInventory, chinaTimeParts, resolvePageUrl, } = require('../services/dailyInventoryScreenshot'); const { getYydNotificationCenterRuntimeConfig, isYydNotificationCenterConfigured, sendYydNotification, } = require('../services/yydNotificationCenterClient'); const TIMEZONE = 'Asia/Shanghai'; const LOCK_KEY = 'daily_inventory_notification_job'; const LOCK_WAIT_SECONDS = 2; const NOTIFICATION_TITLE = '每日进销存数据'; const ACTION_TITLE = '查看每日进销存'; // 重试策略:最多 3 次,退避 2s / 4s const MAX_ATTEMPTS = 3; const RETRY_BASE_DELAY_MS = 2000; // 命中时间后允许补跑的窗口与次数上限(防止全天每分钟重试刷屏) const RETRY_WINDOW_MINUTES = 30; const MAX_SCHEDULED_ATTEMPTS_PER_DAY = 3; let running = false; let timer = null; const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); /** 有限次数重试 + 指数退避 */ async function withRetry(task, { label, attempts = MAX_ATTEMPTS, baseDelayMs = RETRY_BASE_DELAY_MS } = {}) { let lastError = null; for (let index = 0; index < attempts; index += 1) { try { return await task(); } catch (error) { lastError = error; const delay = baseDelayMs * (2 ** index); console.error( `[DailyInventoryNotify] ${label} 第 ${index + 1}/${attempts} 次失败:${error.message}` + `${index < attempts - 1 ? `,${delay}ms 后重试` : ''}` ); if (index < attempts - 1) await sleep(delay); } } throw lastError; } /** 业务日期 = 北京时间昨日(页面标注 T-1,凌晨跑批产出昨日快照) */ function resolveBizDate(now = new Date()) { const target = now instanceof Date ? now : new Date(now); const shifted = new Date(target.getTime() - 24 * 60 * 60 * 1000); return chinaTimeParts(shifted).bizDate; } function beijingTimeText(date = new Date()) { const target = date instanceof Date ? date : new Date(date); return new Intl.DateTimeFormat('zh-CN', { timeZone: TIMEZONE, year: 'numeric', month: '2-digit', day: '2-digit', hour: '2-digit', minute: '2-digit', second: '2-digit', hour12: false, }).format(target); } function currentBeijingHhMm(now = new Date()) { const text = new Intl.DateTimeFormat('en-GB', { timeZone: TIMEZONE, hour: '2-digit', minute: '2-digit', hour12: false, }).format(now instanceof Date ? now : new Date(now)); return text; } /** 全实例锁:拿到锁才继续执行,拿不到说明其它实例正在跑 */ async function withJobLock(runner) { const pool = database.local.pool; let connection = null; try { connection = await pool.getConnection(); const [rows] = await connection.query('SELECT GET_LOCK(?, ?) AS lockStatus', [LOCK_KEY, LOCK_WAIT_SECONDS]); const acquired = Number(rows?.[0]?.lockStatus || 0) === 1; if (!acquired) { return { success: false, skipped: true, reason: 'lock_not_acquired' }; } return await runner(); } catch (error) { return { success: false, skipped: true, reason: 'lock_error', error: error.message }; } finally { if (connection) { try { await connection.query('DO RELEASE_LOCK(?)', [LOCK_KEY]); } catch (_error) { // ignore } connection.release(); } } } function buildNotificationContent({ bizDate, takenAt, imageUrl, pageUrl }) { return [ `**数据日期:** ${bizDate}`, `**截图生成时间:** ${takenAt}(${TIMEZONE})`, `**图片地址:** ${imageUrl}`, `**页面地址:** ${pageUrl}`, '', `点击右下角「${ACTION_TITLE}」可直达每日进销存页面。`, ].join('\n'); } /** * 执行一次(正式 / 测试) * @param {Object} options * mode 'scheduled' | 'manual' | 'test' * force true 时忽略「今天已调度过」的节流判断(手动触发用) * now 指定执行时刻,便于测试 */ async function runDailyInventoryNotificationJob(options = {}) { const mode = options.mode || 'scheduled'; const now = options.now ? new Date(options.now) : new Date(); const startedAt = now.toISOString(); if (running) { return { success: false, skipped: true, reason: 'already_running' }; } running = true; try { return await withJobLock(async () => { const config = await getDailyInventoryNotificationRuntimeConfig({ forceRefresh: true }); const recipients = normalizeRecipients(config.recipients); const bizDate = resolveBizDate(now); const finish = async (payload) => { await appendRunRecord({ ...payload, runAt: startedAt, bizDate, mode, notifyTime: config.notifyTime }); return payload; }; if (!config.enabled) { return finish({ success: false, reason: 'disabled', error: '功能未启用' }); } if (!recipients.length) { return finish({ success: false, reason: 'no_recipient', error: '未配置通知用户' }); } let centerConfig = null; try { centerConfig = await getYydNotificationCenterRuntimeConfig(); } catch (_error) { centerConfig = null; } if (!isYydNotificationCenterConfigured(centerConfig)) { return finish({ success: false, reason: 'notification_center_not_configured', error: '统一通知中心未配置' }); } const storage = await getObjectStorageStatus(); if (!storage.configured) { return finish({ success: false, reason: 'object_storage_not_configured', error: '对象存储未配置' }); } // 截图 → 上传(失败直接结束,绝不发通知) let shot; try { shot = await withRetry( // 对象键按业务日期归档:daily-inventory/2026-09-25/daily-inventory-20260925-010000.png () => captureAndUploadDailyInventory({ now, bizDate }), { label: '截图并上传' } ); } catch (error) { return finish({ success: false, reason: 'screenshot_failed', error: error.message }); } const pageUrl = resolvePageUrl(); const takenAt = beijingTimeText(now); const nonce = crypto.randomBytes(8).toString('hex'); const deliveries = []; for (const recipient of recipients) { const idempotencyKey = buildIdempotencyKey({ bizDate, recipient, mode, nonce }); // 正式推送先查幂等:同一业务日期 + 同一用户只成功通知一次 if (mode !== 'test' && await hasNotified(bizDate, recipient)) { deliveries.push({ recipient, status: 'skipped', reason: 'already_notified' }); continue; } try { const response = await withRetry( () => sendYydNotification({ idempotencyKey, recipientAlias: recipient, title: NOTIFICATION_TITLE, content: buildNotificationContent({ bizDate, takenAt, imageUrl: shot.url, pageUrl, }), actionTitle: ACTION_TITLE, actionUrl: pageUrl, }), { label: `通知 ${recipient}` } ); const deliveryId = response?.id || response?.data?.id || null; if (mode !== 'test') await markNotified(bizDate, recipient, { deliveryId, idempotencyKey }); deliveries.push({ recipient, status: 'sent', deliveryId }); } catch (error) { // 单人失败不阻断其他人 console.error(`[DailyInventoryNotify] 通知 ${recipient} 失败:${error.message}`); deliveries.push({ recipient, status: 'failed', error: error.message }); } } const sentCount = deliveries.filter((d) => d.status === 'sent').length; const failedCount = deliveries.filter((d) => d.status === 'failed').length; const success = sentCount > 0; console.log( `[DailyInventoryNotify] ${mode} 完成:业务日 ${bizDate},截图 ${shot.url},` + `通知成功 ${sentCount} / 失败 ${failedCount} / 跳过 ${deliveries.length - sentCount - failedCount}` ); return finish({ success, reason: success ? null : 'notify_failed', error: success ? null : '所有收件人均通知失败', screenshotUrl: shot.url, screenshotKey: shot.key, rowCount: shot.rowCount, recipients: deliveries, }); }); } finally { running = false; } } /** 同一业务日期已经发起过的正式调度次数与最后一次结果(用于失败重试节流) */ async function scheduledAttemptStatus(bizDate) { const state = await loadDailyInventoryNotificationState(); const runs = (Array.isArray(state.runs) ? state.runs : []) .filter((r) => r.bizDate === bizDate && r.mode === 'scheduled'); return { count: runs.length, lastSuccess: runs.length ? runs[runs.length - 1].success === true : null, }; } function minutesOfDay(hhmm) { const match = /^(\d{1,2}):(\d{2})$/.exec(String(hhmm || '')); if (!match) return null; const hour = Number(match[1]); const minute = Number(match[2]); if (hour > 23 || minute > 59) return null; return hour * 60 + minute; } /** * 是否应当发起一次正式调度: * · 恰好命中配置时间 → 执行 * · 命中时间之后的 RETRY_WINDOW_MINUTES 分钟内 → 仅在上一次失败且未超出重试上限时补跑 * · 其余时间 → 不执行 */ async function shouldRunScheduled(config, now) { const target = minutesOfDay(config.notifyTime); const current = minutesOfDay(currentBeijingHhMm(now)); if (target === null || current === null) return false; const diff = current - target; if (diff < 0 || diff > RETRY_WINDOW_MINUTES) return false; if (diff === 0) return true; const status = await scheduledAttemptStatus(resolveBizDate(now)); return status.lastSuccess === false && status.count < MAX_SCHEDULED_ATTEMPTS_PER_DAY; } function startDailyInventoryNotificationScheduler() { const disabled = String(process.env.DAILY_INVENTORY_NOTIFICATION_ENABLED || 'true').trim().toLowerCase() === 'false'; if (disabled) { console.log('[DailyInventoryNotify] 已通过环境变量关闭'); return null; } if (timer) return timer; // 每分钟检查一次:命中配置的通知时间才真正执行,配置改动无需重启 timer = cron.schedule('* * * * *', async () => { try { const config = await getDailyInventoryNotificationRuntimeConfig({ forceRefresh: true }); if (!config.enabled) return; const now = new Date(); if (!(await shouldRunScheduled(config, now))) return; await runDailyInventoryNotificationJob({ mode: 'scheduled', now }); } catch (error) { console.error('[DailyInventoryNotify] 定时检查失败:', error.message); } }, { timezone: TIMEZONE }); console.log( `[Scheduler] 每日进销存截图通知已注册(按配置时间触发,${TIMEZONE},` + `失败后 ${RETRY_WINDOW_MINUTES} 分钟内最多补跑 ${MAX_SCHEDULED_ATTEMPTS_PER_DAY} 次)` ); return timer; } function stopDailyInventoryNotificationScheduler() { if (timer) { timer.stop(); timer = null; } } module.exports = { ACTION_TITLE, LOCK_KEY, MAX_ATTEMPTS, MAX_SCHEDULED_ATTEMPTS_PER_DAY, NOTIFICATION_TITLE, RETRY_BASE_DELAY_MS, TIMEZONE, RETRY_WINDOW_MINUTES, beijingTimeText, buildNotificationContent, currentBeijingHhMm, minutesOfDay, resolveBizDate, runDailyInventoryNotificationJob, startDailyInventoryNotificationScheduler, stopDailyInventoryNotificationScheduler, withRetry, };