BPOS-TMP-REPO/apps/api/src/app/workers/mail-poller.ts

/**
 * 邮件轮询工作者 —— 每 tick 挑到期账号各拉一轮(设计稿 §4);跨进程租约在 store 里,
 * 两个后端连同一个库时一个信箱同一时刻只有一个在拉。
 * 与 startDispatcher / startTaskAdvancer 同形;上一 tick 没跑完(账号多、信多)就跳过这一拍,不叠着跑。
 */
import type { MailPollOutcome } from '@modules/mail';
import type { Container } from '../container.ts';

export type MailPollerDeps = { readonly container: Pick<Container, 'mail'>; readonly now?: () => Date };

export async function pollMailOnce(deps: MailPollerDeps): Promise<MailPollOutcome[]> {
  const outcomes = await deps.container.mail.pollDue((deps.now ?? (() => new Date()))());
  for (const outcome of outcomes) {
    if (outcome.error !== undefined) process.stderr.write(`[mail-poller] ${outcome.accountId}:${outcome.error}\n`);
  }
  return outcomes;
}

export function startMailPoller(deps: MailPollerDeps, tickMs: number): { stop(): void } {
  let stopped = false;
  let running = false;
  const tick = async (): Promise<void> => {
    if (stopped || running) return;
    running = true;
    try {
      await pollMailOnce(deps);
    } catch (error) {
      process.stderr.write(`[mail-poller] ${error instanceof Error ? error.message : String(error)}\n`);
    } finally {
      running = false;
    }
  };
  const timer = setInterval(() => void tick(), tickMs);
  timer.unref();
  return {
    stop() {
      stopped = true;
      clearInterval(timer);
    },
  };
}