CRM-LeadRadar/apps/api/src/telegram/outbox.ts

import { and, eq, inArray, isNotNull, isNull, sql } from 'drizzle-orm';
import type { Db, Tx } from '../db/client.ts';
import { leadNotifications, leads, outbox, pipelineStatuses, users } from '../db/schema.ts';
import type { Config } from '../lib/config.ts';
import { TelegramError, type Keyboard, type TelegramApi } from './client.ts';
import { esc, leadMessage, texts } from './texts.ts';

// Очередь сообщений Telegram (ТЗ: «очередь для уведомлений», лимиты Bot API, повтор при ошибке).
// Сообщение сначала записывается в таблицу outbox в той же транзакции, что и изменение лида,
// поэтому не теряется при сбое; воркер отправляет его отдельно.

type User = typeof users.$inferSelect;

/** Сейчас у пользователя тихие часы (время Еревана, UTC+4). */
export function inQuietHours(user: Pick<User, 'quietFrom' | 'quietTo'>, now = new Date()): boolean {
  if (!user.quietFrom || !user.quietTo || user.quietFrom === user.quietTo) return false;
  const d = new Date(now.getTime() + 4 * 3_600_000);
  const hm = `${String(d.getUTCHours()).padStart(2, '0')}:${String(d.getUTCMinutes()).padStart(2, '0')}`;
  return user.quietFrom < user.quietTo
    ? hm >= user.quietFrom && hm < user.quietTo
    : hm >= user.quietFrom || hm < user.quietTo;
}

/** Уведомление о лиде с кнопкой «Беру». Пауза и тихие часы пользователя учитываются здесь. */
export async function enqueueLead(tx: Tx, leadId: number, recipients: User[], again = false, now = new Date()) {
  const rows = recipients
    .filter((u) => u.telegramId && !u.removedAt && !u.notificationsPaused && !inQuietHours(u, now))
    .map((u) => ({ kind: 'lead', userId: u.id, leadId, payload: { again } }));
  if (rows.length) await tx.insert(outbox).values(rows);
  return rows.length;
}

/** Обычное сообщение пользователю. `key` — ключ текста из texts(), `args` — его параметры. */
export async function enqueueText(tx: Tx, userIds: number[], key: string, args: unknown[] = [], leadId?: number) {
  if (!userIds.length) return;
  await tx.insert(outbox).values(
    [...new Set(userIds)].map((userId) => ({ kind: 'text', userId, leadId: leadId ?? null, payload: { key, args } })),
  );
}

/** Сообщение всем суперадминам с привязанным Telegram: новые регистрации и компании. */
export async function enqueueForSuperadmins(tx: Tx, key: string, args: unknown[] = []) {
  const admins = await tx.select({ id: users.id }).from(users).where(and(eq(users.role, 'superadmin'), isNotNull(users.telegramId), isNull(users.removedAt)));
  await enqueueText(tx, admins.map((u) => u.id), key, args);
}

/** Как связаться с человеком: email и Telegram. */
export const contactOf = (u: { email: string | null; telegramUsername: string | null; telegramId: number | null }) =>
  [u.email, u.telegramUsername ? `@${u.telegramUsername}` : u.telegramId ? `Telegram ID ${u.telegramId}` : null].filter(Boolean).join(' · ') || '—';

/** Произвольный текст (сообщения суперадмина). */
export async function enqueueRaw(tx: Tx, userIds: number[], text: string) {
  if (!userIds.length) return;
  await tx.insert(outbox).values(userIds.map((userId) => ({ kind: 'text', userId, payload: { raw: text } })));
}

/** Лид взяли: у всех получателей обновить сообщение и убрать «Беру»; неотправленные уведомления отменить. */
export async function enqueueTaken(tx: Tx, leadId: number, takenById: number) {
  await tx
    .update(outbox)
    .set({ doneAt: new Date(), error: 'canceled: lead taken' })
    .where(and(eq(outbox.leadId, leadId), eq(outbox.kind, 'lead'), isNull(outbox.doneAt)));
  const sent = await tx.select().from(leadNotifications).where(eq(leadNotifications.leadId, leadId));
  if (!sent.length) return;
  await tx.insert(outbox).values(
    sent.map((n) => ({ kind: 'edit', userId: n.userId, leadId, payload: { notificationId: n.id, takenById } })),
  );
}

const BATCH = 20;
/** Не больше ~25 сообщений в секунду на бота (лимит Bot API — 30). */
const GAP_MS = 40;
const MAX_ATTEMPTS = 6;

export interface OutboxDeps {
  db: Db;
  tg: TelegramApi;
  config: Config;
  sleep?: (ms: number) => Promise<void>;
}

const openButton = (config: Config, lang: string, leadId: number): Keyboard[number] => {
  // Telegram принимает в кнопках только https-адреса; на локальном стенде кнопки «Открыть» нет.
  if (!config.appUrl.startsWith('https://')) return [];
  return [{ text: texts(lang).open, webApp: `${config.appUrl}/leads/${leadId}` }];
};

/** Отправляет одну пачку сообщений из очереди. Возвращает число обработанных. */
export async function processOutbox({ db, tg, config, sleep }: OutboxDeps): Promise<number> {
  const wait = sleep ?? ((ms: number) => new Promise<void>((r) => setTimeout(r, ms)));
  // Забираем пачку и сразу отодвигаем run_at: если воркер упадёт, сообщения вернутся в очередь сами.
  const jobs = await db
    .update(outbox)
    .set({ runAt: sql`now() + interval '2 minutes'`, attempts: sql`${outbox.attempts} + 1` })
    .where(
      inArray(
        outbox.id,
        sql`(select id from outbox where done_at is null and run_at <= now() order by id limit ${BATCH} for update skip locked)`,
      ),
    )
    .returning();
  jobs.sort((a, b) => a.id - b.id);

  for (const job of jobs) {
    const finish = (error: string | null = null) =>
      db.update(outbox).set({ doneAt: new Date(), error }).where(eq(outbox.id, job.id));
    try {
      const [found] = job.userId ? await db.select().from(users).where(eq(users.id, job.userId)) : [];
      // Сообщения поддержки идут прямо в чат: у написавшего может не быть аккаунта.
      const direct = job.payload as { chatId?: number; lang?: string };
      if (!direct.chatId && (!found?.telegramId || found.removedAt)) {
        await finish('skipped: no telegram');
        continue;
      }
      const user = found ?? { id: 0, lang: direct.lang ?? 'ru' };
      const chatId = direct.chatId ?? found!.telegramId!;
      const t = texts(user.lang);

      if (job.kind === 'text') {
        const p = job.payload as { raw?: string; key?: string; args?: unknown[]; buttons?: { key: string; callback: string }[] };
        // Кнопки под сообщением: подпись берётся из текстов бота на языке получателя.
        const buttons: Keyboard | undefined = p.buttons?.length
          ? [p.buttons.map((b) => ({ text: String((t as unknown as Record<string, unknown>)[b.key] ?? b.key), callback: b.callback }))]
          : undefined;
        const fn = p.key ? (t as unknown as Record<string, unknown>)[p.key] : undefined;
        const text =
          p.raw !== undefined
            ? esc(p.raw)
            : typeof fn === 'function'
              ? (fn as (...a: unknown[]) => string)(...(p.args ?? []).map((a) => (typeof a === 'string' ? esc(a) : a)))
              : String(fn ?? '');
        if (text) await tg.sendMessage(chatId, text, buttons ?? (job.leadId ? [openButton(config, user.lang, job.leadId)].filter((r) => r.length) : undefined));
        await finish();
      } else if (job.kind === 'lead') {
        const [row] = await db
          .select({ lead: leads, code: pipelineStatuses.code })
          .from(leads)
          .innerJoin(pipelineStatuses, eq(pipelineStatuses.id, leads.statusId))
          .where(eq(leads.id, job.leadId!));
        // Пока сообщение ждало в очереди, лид могли взять.
        if (!row || row.lead.assigneeId !== null || row.code !== 'new') {
          await finish('skipped: lead not in pool');
          continue;
        }
        const { lead } = row;
        const again = Boolean((job.payload as { again?: boolean }).again);
        const text = leadMessage(lead, user.lang, again ? t.backInPool : t.newLead);
        const keyboard: Keyboard = [
          [{ text: t.take, callback: `take:${lead.id}` }, ...openButton(config, user.lang, lead.id)],
        ];
        const photo = lead.photos[0];
        let sent;
        let hasPhoto = false;
        if (photo) {
          try {
            sent = await tg.sendPhoto(chatId, photo, text, keyboard);
            hasPhoto = true;
          } catch (e) {
            // Telegram не смог скачать фото — отправляем без него. Остальные ошибки обрабатываются ниже.
            if (!(e instanceof TelegramError) || e.code !== 400) throw e;
          }
        }
        sent ??= await tg.sendMessage(chatId, text, keyboard);
        await db.insert(leadNotifications).values({
          leadId: lead.id,
          userId: user.id,
          chatId: sent.chatId,
          messageId: sent.messageId,
          hasPhoto,
          sentAt: new Date(),
        });
        await finish();
      } else if (job.kind === 'edit') {
        const p = job.payload as { notificationId: number; takenById: number };
        const [n] = await db.select().from(leadNotifications).where(eq(leadNotifications.id, p.notificationId));
        const [lead] = await db.select().from(leads).where(eq(leads.id, job.leadId!));
        if (n?.chatId && n.messageId && lead) {
          const [taker] = await db.select({ name: users.name }).from(users).where(eq(users.id, p.takenById));
          const footer = p.takenById === user.id ? t.youTook : `<i>${t.takenBy(esc(taker?.name ?? '—'))}</i>`;
          const mine = p.takenById === user.id ? [openButton(config, user.lang, lead.id)].filter((r) => r.length) : [];
          await tg.editMessage({ chatId: n.chatId, messageId: n.messageId }, n.hasPhoto, leadMessage(lead, user.lang, t.newLead, footer), mine);
        }
        await finish();
      } else {
        await finish(`unknown kind ${job.kind}`);
      }
    } catch (e) {
      const message = e instanceof Error ? e.message : String(e);
      if (e instanceof TelegramError && e.code === 429) {
        // Telegram просит подождать: попытку не считаем, всю пачку откладываем.
        const until = sql`now() + make_interval(secs => ${(e.retryAfter ?? 5) + 1})`;
        await db.update(outbox).set({ runAt: until, attempts: sql`${outbox.attempts} - 1`, error: message }).where(eq(outbox.id, job.id));
        await wait(((e.retryAfter ?? 5) + 1) * 1000);
      } else if ((e instanceof TelegramError && e.permanent) || job.attempts >= MAX_ATTEMPTS) {
        await finish(`failed: ${message}`);
      } else {
        // Повтор с нарастающей паузой: 10 с, 40 с, 90 с…
        await db
          .update(outbox)
          .set({ runAt: sql`now() + make_interval(secs => ${job.attempts * job.attempts * 10})`, error: message })
          .where(eq(outbox.id, job.id));
      }
    }
    await wait(GAP_MS);
  }
  return jobs.length;
}