memind/schedule-reminder-worker.mjs

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);
    },
  };
}