export function startScheduleReminderWorker({ scheduleService, sendWechatTextToUser, notificationDispatcher, logger = console, intervalMs = Number(process.env.H5_REMINDER_SCAN_INTERVAL_MS ?? 30_000), maxAttempts = Number(process.env.H5_REMINDER_MAX_ATTEMPTS ?? 5), runOnStart = true, } = {}) { const sendScheduleNotification = notificationDispatcher?.sendScheduleNotification?.bind(notificationDispatcher) ?? (typeof sendWechatTextToUser === 'function' ? async ({ userId, text }) => sendWechatTextToUser(userId, text) : null); if (!scheduleService || typeof sendScheduleNotification !== 'function') { return { stop() {} }; } let stopped = false; let running = false; const runOnce = async () => { if (running || stopped) return; running = true; try { const dueReminders = await scheduleService.listDueReminders({ limit: 50 }); for (const candidate of dueReminders) { const reminder = await scheduleService.lockReminder(candidate.id); if (!reminder) continue; try { const text = await scheduleService.buildReminderText(reminder); if (!text) { await scheduleService.markReminderCancelled(reminder, '事项已失效或不存在'); continue; } await scheduleService.createUserNotification?.({ userId: reminder.userId, channel: 'web', notificationType: 'schedule_reminder', title: '待办提醒', body: text, data: { reminderId: reminder.id, itemId: reminder.itemId, }, }); if (reminder.channel === 'wechat') { await sendScheduleNotification({ userId: reminder.userId, text }); } await scheduleService.logDelivery({ reminderId: reminder.id, userId: reminder.userId, channel: reminder.channel, status: 'success', }); await scheduleService.markReminderSent(reminder); } catch (err) { logger.warn?.('Schedule reminder delivery failed:', err); await scheduleService.logDelivery({ reminderId: reminder.id, userId: reminder.userId, channel: reminder.channel, status: 'failed', errorMessage: err instanceof Error ? err.message : String(err), }).catch(() => {}); await scheduleService.markReminderFailed(reminder, err, { maxAttempts }); } } const due = await scheduleService.listDueDigestSubscriptions({ limit: 50 }); for (const candidate of due) { const subscription = await scheduleService.lockDigestSubscription(candidate.id); if (!subscription) continue; try { const text = await scheduleService.buildTodoDigestText({ userId: subscription.userId, timezone: subscription.timezone, }); await scheduleService.createUserNotification?.({ userId: subscription.userId, channel: 'web', notificationType: 'todo_digest', title: '今日待办摘要', body: text, data: { subscriptionId: subscription.id, timezone: subscription.timezone, }, }); await sendScheduleNotification({ userId: subscription.userId, text }); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, channel: subscription.channel, status: 'success', }); await scheduleService.markDigestSent(subscription); } catch (err) { logger.warn?.('Schedule digest delivery failed:', err); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, channel: subscription.channel, status: 'failed', errorMessage: err instanceof Error ? err.message : String(err), }).catch(() => {}); await scheduleService.markDigestFailed(subscription, err, { maxAttempts }); } } const dueBalance = await scheduleService.listDueBalanceAlerts({ limit: 50 }); for (const candidate of dueBalance) { const subscription = await scheduleService.lockBalanceAlert(candidate.id); if (!subscription) continue; try { const user = await scheduleService.getUserWalletSnapshot?.(subscription.userId); const balanceCents = Number(user?.balanceCents ?? user?.balance_cents ?? 0); if (balanceCents > subscription.thresholdCents) { await scheduleService.markBalanceAlertSent({ ...subscription, lastNotifiedBalanceCents: balanceCents, }); continue; } const yuan = (subscription.thresholdCents / 100).toFixed(2); const text = `余额不足提醒\n\n你的账户余额已低于 ${yuan} 元,请及时充值。`; await scheduleService.createUserNotification?.({ userId: subscription.userId, channel: 'web', notificationType: 'balance_low', title: '余额不足提醒', body: text, data: { thresholdCents: subscription.thresholdCents, balanceCents, }, }); await sendScheduleNotification({ userId: subscription.userId, text }); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, channel: subscription.channel, status: 'success', }); await scheduleService.markBalanceAlertSent({ ...subscription, lastNotifiedBalanceCents: balanceCents, }); } catch (err) { logger.warn?.('Balance alert delivery failed:', err); await scheduleService.logDelivery({ subscriptionId: subscription.id, userId: subscription.userId, channel: subscription.channel, status: 'failed', errorMessage: err instanceof Error ? err.message : String(err), }).catch(() => {}); await scheduleService.markBalanceAlertFailed(subscription, err, { maxAttempts }); } } } catch (err) { logger.warn?.('Schedule reminder worker failed:', err); } finally { running = false; } }; const timer = setInterval(runOnce, Math.max(1000, Number(intervalMs) || 30_000)); timer.unref?.(); if (runOnStart) void runOnce(); return { runOnce, stop() { stopped = true; clearInterval(timer); }, }; }