ApiCore/ApiCore-Backend/src/telegram/telegram-sender.ts

import { Injectable, Logger } from '@nestjs/common';
import { Prisma } from '../../generated/prisma';
import { PermissionResolverService } from '../auth/permissions/permission-resolver.service';
import { berlinToday } from '../common/dates';
import { loggableError } from '../common/loggable-error';
import {
  claimBatch,
  deferClaim,
  releaseStaleClaimRows,
  settleFailure,
  settleSkipped,
  settleSuccess,
} from '../common/outbox/outbox-claims';
import { DRAIN_MAX_ROUNDS } from '../domain-events/domain-event-dispatcher';
import { NotificationsService } from '../notifications/notifications.service';
import { PrismaService } from '../prisma/prisma.service';
import { lockRowForUpdate } from '../prisma/row-lock';
import { removeLinkIn } from './telegram-link-removal';
import { TelegramApiError } from './telegram-api-error';
import { TelegramClient } from './telegram-client';
import {
  renderTelegramText,
  TELEGRAM_TEMPLATES,
  TelegramTemplateCode,
} from './telegram-texts';

export const TELEGRAM_SEND_POLL_MS = 5_000;
export const TELEGRAM_SEND_BATCH = 25;
export const TELEGRAM_MAX_PER_SECOND = 25;
export const TELEGRAM_CHAT_SPACING_MS = 1_100;
export const TELEGRAM_TOKEN_INVALID_PAUSE_MS = 900_000;

const TABLE = 'NotificationDelivery';
const MIN_SEND_GAP_MS = Math.ceil(1_000 / TELEGRAM_MAX_PER_SECOND);
const DEFAULT_RETRY_AFTER_SEC = 5;
/** How long the memory of "last message to this chat" is kept. */
const CHAT_MEMORY_MS = 10_000;

const CHAT_GONE_MESSAGE =
  'Telegram-Zustellung beendet. Der Bot wurde blockiert oder der Chat gelöscht.';
const TOKEN_INVALID_MESSAGE = 'Telegram: Bot-Token ungültig.';

/** Reasons a delivery is skipped; they land in lastError, so no personal data. */
const SKIP = {
  userInactive: 'user inactive',
  telegramDisabled: 'telegram disabled',
  notLinked: 'not linked',
  categoryOff: 'category off',
  permissionRevoked: 'permission revoked',
  stale: 'stale',
  taskDone: 'task already done',
  chatUnavailable: 'chat unavailable',
  unknownTemplate: 'unknown template',
  notificationGone: 'notification gone',
} as const;

/** A claimed NotificationDelivery row, as RETURNING * yields it. */
interface ClaimedDelivery {
  id: string;
  userId: string | null;
  chatId: bigint | null;
  notificationId: string | null;
  templateCode: string | null;
  attempts: number;
  claimedAt: Date;
}

/** What the send-time check reads about a delivery, in one query. */
interface DeliveryContext {
  userActive: boolean;
  telegramEnabled: boolean;
  categories: string[];
  detail: boolean;
  chatId: bigint | null;
  category: string;
  message: string;
  link: string | null;
  requiredPermissions: string[];
  stale: boolean;
  taskOpen: boolean;
}

type Target = { skip: string } | { chatId: bigint; text: string };

/** 'spaced': put back only because the chat got a message a moment ago. */
type Outcome = 'settled' | 'spaced';

interface RoundResult {
  claimed: number;
  spaced: number;
}

export interface TelegramSendFilter {
  /** Only deliveries of these users (e2e tests: no foreign rows). */
  userIds?: readonly string[];
}

function claimFilter(filter?: TelegramSendFilter): Prisma.Sql | undefined {
  if (filter?.userIds === undefined) {
    return undefined;
  }
  return filter.userIds.length === 0
    ? Prisma.sql`FALSE`
    : Prisma.sql`"userId" IN (${Prisma.join([...filter.userIds])})`;
}

function targetKey(delivery: ClaimedDelivery): string {
  return delivery.userId !== null
    ? `user:${delivery.userId}`
    : `chat:${String(delivery.chatId)}`;
}

/** The oldest delivery of each user or chat; the claim is ordered by age. */
function splitOldestPerTarget(claimed: readonly ClaimedDelivery[]): {
  oldest: ClaimedDelivery[];
  rest: ClaimedDelivery[];
} {
  const seen = new Set<string>();
  const oldest: ClaimedDelivery[] = [];
  const rest: ClaimedDelivery[] = [];
  for (const delivery of claimed) {
    const key = targetKey(delivery);
    if (seen.has(key)) {
      rest.push(delivery);
    } else {
      seen.add(key);
      oldest.push(delivery);
    }
  }
  return { oldest, rest };
}

function sleep(ms: number, signal: AbortSignal): Promise<void> {
  return new Promise((resolve) => {
    if (signal.aborted) {
      resolve();
      return;
    }
    const done = (): void => {
      clearTimeout(timer);
      signal.removeEventListener('abort', done);
      resolve();
    };
    const timer = setTimeout(done, ms);
    signal.addEventListener('abort', done, { once: true });
  });
}

function isChatGone(error: TelegramApiError): boolean {
  return (
    error.statusCode === 403 ||
    (error.statusCode === 400 && /chat not found/i.test(error.description))
  );
}

const DEVELOPMENT_ORIGIN = 'http://localhost:3000';
/** A batch stops after this long; the claim is stale after 5 minutes. */
export const TELEGRAM_BATCH_BUDGET_MS = 120_000;

/**
 * Sends the queued Telegram deliveries (Spec 9a.6): claims up to 25 due
 * rows, takes the oldest of each user or chat, checks again at send time
 * that the message may still go out, and settles each row. Bot replies
 * (templateCode) skip the checks: they answer a person in a chat.
 * Run by the worker's telegram-send loop. ASSUMPTION: one worker replica
 * (the worker deployment has replicas: 1). The per-chat spacing, the global
 * rate limit and the pause after a 401 or 429 live in this pod's memory; with
 * several pods each would keep its own (the one-delivery-per-target rule in
 * SQL and Telegram's 429 answers would still protect the bot). Never logs the token, a chat id
 * or a message text.
 */
@Injectable()
export class TelegramSender {
  private readonly logger = new Logger(TelegramSender.name);
  private pausedUntil = 0;
  private warnedMissingOrigin = false;
  private lastSendAt = 0;
  private readonly lastSendToChat = new Map<string, number>();

  constructor(
    private readonly prisma: PrismaService,
    private readonly client: TelegramClient,
    private readonly permissions: PermissionResolverService,
    private readonly notifications: NotificationsService,
  ) {}

  /** One batch. Returns how many deliveries it claimed. */
  async runOnce(
    signal: AbortSignal,
    filter?: TelegramSendFilter,
  ): Promise<number> {
    return (await this.round(signal, filter)).claimed;
  }

  /**
   * Runs batches until one claims nothing, at most `maxRounds`. After a
   * batch that put messages back for the chat spacing it waits that
   * spacing, so a burst for one person does not wait for the next poll.
   */
  async drain(
    signal: AbortSignal,
    maxRounds = DRAIN_MAX_ROUNDS,
    filter?: TelegramSendFilter,
  ): Promise<number> {
    let total = 0;
    for (let round = 0; round < maxRounds && !signal.aborted; round++) {
      const result = await this.round(signal, filter);
      if (result.claimed === 0) {
        break;
      }
      total += result.claimed;
      if (result.spaced > 0) {
        await sleep(TELEGRAM_CHAT_SPACING_MS, signal);
      }
    }
    return total;
  }

  /** Puts deliveries whose worker died back in line (or to FAILED). */
  async releaseStaleClaims(): Promise<number> {
    const released = await releaseStaleClaimRows(this.prisma, TABLE);
    const failed = released
      .filter((claim) => claim.status === 'FAILED')
      .map((claim) => claim.id);
    if (failed.length > 0) {
      this.logger.error(
        `Telegram deliveries FAILED, an expired claim used up their attempts: ${failed.join(', ')}`,
      );
    }
    return released.length;
  }

  private async round(
    signal: AbortSignal,
    filter?: TelegramSendFilter,
  ): Promise<RoundResult> {
    if (signal.aborted || this.remainingPauseMs() > 0) {
      return { claimed: 0, spaced: 0 };
    }
    const claimed = await claimBatch<ClaimedDelivery>(this.prisma, TABLE, {
      limit: TELEGRAM_SEND_BATCH,
      orderBy: 'createdAt',
      filter: claimFilter(filter),
    });
    const { oldest, rest } = splitOldestPerTarget(claimed);
    let spaced = rest.length;
    for (const delivery of rest) {
      await deferClaim(
        this.prisma,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        TELEGRAM_CHAT_SPACING_MS,
      );
    }
    const started = performance.now();
    for (const delivery of oldest) {
      if (performance.now() - started > TELEGRAM_BATCH_BUDGET_MS) {
        // Out of time: the claim would go stale. Back without an attempt.
        await deferClaim(
          this.prisma,
          TABLE,
          delivery.id,
          delivery.claimedAt,
          0,
        );
        continue;
      }
      const outcome = await this.handle(delivery, signal);
      if (outcome === 'spaced') {
        spaced++;
      }
    }
    return { claimed: claimed.length, spaced };
  }

  /** Shutdown and a paused bot put the claim back without an attempt. */
  private async handle(
    delivery: ClaimedDelivery,
    signal: AbortSignal,
  ): Promise<Outcome> {
    const pauseMs = this.remainingPauseMs();
    if (signal.aborted || pauseMs > 0) {
      await deferClaim(
        this.prisma,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        pauseMs,
      );
      return 'settled';
    }
    try {
      return await this.process(delivery, signal);
    } catch (error: unknown) {
      await this.failSafely(delivery, error);
      return 'settled';
    }
  }

  private async process(
    delivery: ClaimedDelivery,
    signal: AbortSignal,
  ): Promise<Outcome> {
    const target =
      delivery.templateCode !== null
        ? this.templateTarget(delivery)
        : await this.notificationTarget(delivery);
    if ('skip' in target) {
      await settleSkipped(
        this.prisma,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        target.skip,
      );
      return 'settled';
    }
    const waitMs = this.chatWaitMs(target.chatId);
    if (waitMs > 0) {
      await deferClaim(
        this.prisma,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        waitMs,
      );
      return 'spaced';
    }
    await this.keepRateLimit(signal);
    await this.send(delivery, target, signal);
    return 'settled';
  }

  private templateTarget(delivery: ClaimedDelivery): Target {
    const code = delivery.templateCode;
    if (
      code === null ||
      delivery.chatId === null ||
      !Object.hasOwn(TELEGRAM_TEMPLATES, code)
    ) {
      return { skip: SKIP.unknownTemplate };
    }
    return {
      chatId: delivery.chatId,
      text: TELEGRAM_TEMPLATES[code as TelegramTemplateCode],
    };
  }

  /** Spec 9a.6: everything that may have changed since the delivery was queued. */
  private async notificationTarget(delivery: ClaimedDelivery): Promise<Target> {
    const context = await this.loadContext(delivery);
    if (context === null) {
      return { skip: SKIP.notificationGone };
    }
    if (!context.userActive) {
      return { skip: SKIP.userInactive };
    }
    if (!context.telegramEnabled) {
      return { skip: SKIP.telegramDisabled };
    }
    if (context.chatId === null) {
      return { skip: SKIP.notLinked };
    }
    if (!context.categories.includes(context.category)) {
      return { skip: SKIP.categoryOff };
    }
    // Fresh, not the 30 s cache: this process never hears the invalidation
    // of the API pods, so a revoked right would keep sending for 30 s.
    const held = await this.permissions.getFreshPermissions(
      delivery.userId as string,
    );
    if (!context.requiredPermissions.every((key) => held.has(key))) {
      return { skip: SKIP.permissionRevoked };
    }
    if (context.stale) {
      return { skip: SKIP.stale };
    }
    if (!context.taskOpen) {
      return { skip: SKIP.taskDone };
    }
    return {
      chatId: context.chatId,
      text: renderTelegramText(context, context.detail, this.frontendOrigin()),
    };
  }

  /**
   * One query for user, settings, link, notification, age and task. Stale:
   * TASK_DUE and DIGEST only on the Berlin day they were written, every
   * other category for 24 hours. "createdAt" is a UTC timestamp without
   * zone; AT TIME ZONE 'UTC' makes it an instant first.
   */
  private async loadContext(
    delivery: ClaimedDelivery,
  ): Promise<DeliveryContext | null> {
    const rows = await this.prisma.$queryRaw<DeliveryContext[]>(Prisma.sql`
      SELECT
        (u."isActive" AND r."status" = 'ACTIVE') AS "userActive",
        COALESCE(p."telegramEnabled", FALSE) AS "telegramEnabled",
        COALESCE(p."telegramCategories"::text[], ARRAY[]::text[]) AS "categories",
        COALESCE(p."telegramDetail", TRUE) AS "detail",
        l."chatId" AS "chatId",
        n."category"::text AS "category",
        n."message" AS "message",
        n."link" AS "link",
        n."requiredPermissions" AS "requiredPermissions",
        CASE
          WHEN n."category" IN ('TASK_DUE', 'DIGEST') THEN
            to_char((n."createdAt" AT TIME ZONE 'UTC') AT TIME ZONE 'Europe/Berlin', 'YYYY-MM-DD')
              <> to_char(now() AT TIME ZONE 'Europe/Berlin', 'YYYY-MM-DD')
          ELSE now() - (n."createdAt" AT TIME ZONE 'UTC') > interval '24 hours'
        END AS "stale",
        CASE
          WHEN n."category" = 'TASK_DUE' AND n."relatedType" = 'Task' AND n."relatedId" IS NOT NULL THEN
            COALESCE((SELECT t."status" IN ('OPEN', 'IN_PROGRESS') FROM "Task" t WHERE t."id" = n."relatedId"), FALSE)
          ELSE TRUE
        END AS "taskOpen"
      FROM "NotificationDelivery" d
      JOIN "Notification" n ON n."id" = d."notificationId"
      JOIN "User" u ON u."id" = d."userId"
      JOIN "Role" r ON r."id" = u."roleId"
      LEFT JOIN "NotificationPreference" p ON p."userId" = u."id"
      LEFT JOIN "TelegramLink" l ON l."userId" = u."id"
      WHERE d."id" = ${delivery.id}`);
    return rows[0] ?? null;
  }

  private async send(
    delivery: ClaimedDelivery,
    target: { chatId: bigint; text: string },
    signal: AbortSignal,
  ): Promise<void> {
    try {
      await this.client.sendMessage(target.chatId, target.text, signal);
    } catch (error: unknown) {
      await this.onSendError(delivery, target.chatId, error);
      return;
    }
    this.lastSendToChat.set(String(target.chatId), performance.now());
    await settleSuccess(
      this.prisma,
      TABLE,
      delivery.id,
      delivery.claimedAt,
      'DELIVERED',
    );
  }

  private async onSendError(
    delivery: ClaimedDelivery,
    chatId: bigint,
    error: unknown,
  ): Promise<void> {
    if (!(error instanceof TelegramApiError)) {
      await this.failSafely(delivery, error);
      return;
    }
    if (error.statusCode === 429) {
      const seconds = error.retryAfterSec ?? DEFAULT_RETRY_AFTER_SEC;
      // Telegram limits the whole bot: the pod waits, not only this row.
      this.pausedUntil = Math.max(
        this.pausedUntil,
        performance.now() + seconds * 1_000,
      );
      await deferClaim(
        this.prisma,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        seconds * 1_000,
      );
      return;
    }
    if (isChatGone(error)) {
      await this.endDelivery(delivery, chatId);
      return;
    }
    if (error.statusCode === 401) {
      await this.pauseForInvalidToken(delivery);
      return;
    }
    await this.failSafely(delivery, error);
  }

  /**
   * The bot is blocked or the chat is gone: the link ends, but only if it
   * still points to this chat - a newer link of the same user to another
   * chat stays. Settings, link and message to the user change together.
   */
  private async endDelivery(
    delivery: ClaimedDelivery,
    chatId: bigint,
  ): Promise<void> {
    const userId = delivery.userId;
    if (userId === null) {
      await settleSkipped(
        this.prisma,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        SKIP.chatUnavailable,
      );
      return;
    }
    await this.prisma.$transaction(async (tx) => {
      await lockRowForUpdate(tx, 'NotificationPreference', userId);
      // Link, switch and the waiting backlog end together (only if the link
      // still points to this chat).
      if (await removeLinkIn(tx, userId, 'chat unavailable', chatId)) {
        await this.notifications.record(
          {
            userId,
            priority: 'HINWEIS',
            category: 'SYSTEM',
            message: CHAT_GONE_MESSAGE,
          },
          tx,
        );
      }
      await settleSkipped(
        tx,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        SKIP.chatUnavailable,
      );
    });
  }

  /** The token is rejected: nothing can go out, so the whole loop waits. */
  private async pauseForInvalidToken(delivery: ClaimedDelivery): Promise<void> {
    this.pausedUntil = performance.now() + TELEGRAM_TOKEN_INVALID_PAUSE_MS;
    await deferClaim(
      this.prisma,
      TABLE,
      delivery.id,
      delivery.claimedAt,
      TELEGRAM_TOKEN_INVALID_PAUSE_MS,
    );
    try {
      // One per user and day: the dedupe key carries the Berlin day.
      await this.notifications.recordForPermission('system:manage', {
        category: 'SYSTEM',
        priority: 'WARNUNG',
        message: TOKEN_INVALID_MESSAGE,
        link: '/settings/system',
        dedupeKeyPrefix: `telegram-token-invalid:${berlinToday()}`,
      });
    } catch (error: unknown) {
      this.logger.error(
        `Telegram token notice not recorded: ${loggableError(error).message}`,
      );
    }
    this.logger.error(
      'Telegram rejected the bot token (401); sending paused for 15 minutes',
    );
  }

  /** A failed attempt under the shared retry rule; the error text carries no secret. */
  private async failSafely(
    delivery: ClaimedDelivery,
    error: unknown,
  ): Promise<void> {
    const { message } = loggableError(error);
    try {
      await settleFailure(
        this.prisma,
        TABLE,
        delivery.id,
        delivery.claimedAt,
        delivery.attempts,
        message,
      );
    } catch (settleError: unknown) {
      this.logger.error(
        `Telegram delivery ${delivery.id} not settled: ${loggableError(settleError).message}`,
      );
    }
  }

  /**
   * Only development may fall back to localhost. Elsewhere a missing origin
   * (the startup check normally prevents it) sends the text without a link.
   */
  private frontendOrigin(): string | null {
    const configured = process.env.FRONTEND_ORIGIN;
    if (configured) {
      return configured;
    }
    const environment = process.env.NODE_ENV;
    if (environment === undefined || environment === 'development') {
      return DEVELOPMENT_ORIGIN;
    }
    if (!this.warnedMissingOrigin) {
      this.warnedMissingOrigin = true;
      this.logger.warn(
        'FRONTEND_ORIGIN is not set: Telegram messages go out without a link',
      );
    }
    return null;
  }

  private remainingPauseMs(): number {
    return Math.max(0, Math.ceil(this.pausedUntil - performance.now()));
  }

  /** Milliseconds until the chat may get its next message; 0: now. */
  private chatWaitMs(chatId: bigint): number {
    const now = performance.now();
    for (const [chat, sentAt] of this.lastSendToChat) {
      if (now - sentAt > CHAT_MEMORY_MS) {
        this.lastSendToChat.delete(chat);
      }
    }
    const sentAt = this.lastSendToChat.get(String(chatId));
    return sentAt === undefined
      ? 0
      : Math.max(0, Math.ceil(TELEGRAM_CHAT_SPACING_MS - (now - sentAt)));
  }

  /** At most TELEGRAM_MAX_PER_SECOND messages per second over all chats. */
  private async keepRateLimit(signal: AbortSignal): Promise<void> {
    const waitMs = MIN_SEND_GAP_MS - (performance.now() - this.lastSendAt);
    if (waitMs > 0) {
      await sleep(waitMs, signal);
    }
    this.lastSendAt = performance.now();
  }
}