feat: T4 wave 11 – bottling form with article values and confirmation, Telegram in the app

- Bottling form: jar size and best-before from the product with source,
  differs-hints before saving, prefill status, water-content confirmation
  naming the measured value, idempotency key renewed on body change;
  product form gets shelf life (months) and whole-gram jar weight.
- Telegram frontend: link card with step-up dialog, t.me-only deep link,
  status polling, unlink; system Telegram card with receiver/sender state
  and confirmed retry; StepUpDialog shared with the backup import.
- Reauth calls never refresh-and-retry on 401 (no double lockout count);
  German texts for expired session and 429; service worker never caches
  Telegram, notification settings and system responses.
- Telegram backend round 2: shared link removal on every disconnect path,
  lease released on shutdown, connection errors not counted as poison.
- ARCHITECTURE.md §2, §6 (Annahmen 56–58), §10, §11.12, §11.13; TASKS.md T4.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

ApiCore-Backend/src/telegram/telegram-updates.service.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from 'node:crypto';
-import { Injectable, Logger } from '@nestjs/common';
+import { Inject, Injectable, Logger, Optional } from '@nestjs/common';
 import { Prisma } from '../../generated/prisma';
 import { loggableError } from '../common/loggable-error';
 import { PrismaService } from '../prisma/prisma.service';
@@ -18,6 +18,17 @@ export const NO_LEASE_WAIT_MS = 30_000;
 export const HELP_REPLIES_PER_CHAT = 3;
 export const HELP_WINDOW_MS = 600_000;
 
+/** Tunables of the receiver; tests inject their own (limits, clock). */
+export interface TelegramUpdatesOptions {
+  /** Attempts after which a failing update is skipped. */
+  attempts: number;
+  /** Time after which a failing update is skipped, in ms of `now()`. */
+  windowMs: number;
+  /** Monotonic clock in ms. */
+  now: () => number;
+}
+export const TELEGRAM_UPDATES_OPTIONS = Symbol('TELEGRAM_UPDATES_OPTIONS');
+
 /** An update that keeps failing is skipped after this many attempts or this long. */
 export const MAX_UPDATE_ATTEMPTS = 5;
 export const MAX_UPDATE_RETRY_MS = 600_000;
@@ -59,6 +70,34 @@ function describeError(error: unknown): string {
   return error instanceof Error ? error.name : 'unknown error';
 }
 
+const CONNECTION_ERROR_CODES = new Set(['P1001', 'P1002', 'P1008', 'P1017']);
+const SOCKET_ERROR_CODES = new Set([
+  'ECONNREFUSED',
+  'ECONNRESET',
+  'ETIMEDOUT',
+  'EHOSTUNREACH',
+]);
+
+/**
+ * The database or the network is unreachable (Prisma P1001, P1002, P1008,
+ * P1017, an initialization error, a refused or timed-out socket, a Telegram
+ * call without an answer): that says nothing about the update itself, so it
+ * does not count toward skipping it.
+ */
+function isConnectionError(error: unknown): boolean {
+  if (error instanceof Prisma.PrismaClientInitializationError) {
+    return true;
+  }
+  if (error instanceof Prisma.PrismaClientKnownRequestError) {
+    return CONNECTION_ERROR_CODES.has(error.code);
+  }
+  if (error instanceof TelegramApiError) {
+    return error.statusCode === null;
+  }
+  const code = (error as { code?: unknown } | null)?.code;
+  return typeof code === 'string' && SOCKET_ERROR_CODES.has(code);
+}
+
 function parseCommand(text: string | null): Command {
   const match = START_COMMAND.exec(text?.trim() ?? '');
   const name = match?.[1].toLowerCase();
@@ -77,6 +116,8 @@ function parseCommand(text: string | null): Command {
 class ReplyLimiter {
   private readonly sentAt = new Map<string, number[]>();
 
+  constructor(private readonly now: () => number) {}
+
   /** Would one more reply to this chat be within the limit? Counts nothing. */
   allows(chatId: bigint): boolean {
     this.prune();
@@ -88,11 +129,11 @@ class ReplyLimiter {
   /** Counts a reply that was queued and committed. */
   record(chatId: bigint): void {
     const key = String(chatId);
-    this.sentAt.set(key, [...(this.sentAt.get(key) ?? []), performance.now()]);
+    this.sentAt.set(key, [...(this.sentAt.get(key) ?? []), this.now()]);
   }
 
   private prune(): void {
-    const now = performance.now();
+    const now = this.now();
     for (const [chat, times] of this.sentAt) {
       const recent = times.filter((time) => now - time < HELP_WINDOW_MS);
       if (recent.length === 0) {
@@ -119,7 +160,12 @@ export class TelegramUpdatesService {
   private readonly logger = new Logger(TelegramUpdatesService.name);
   /** Names this receiver in the lease. */
   private readonly instanceId = randomUUID();
-  private readonly replies = new ReplyLimiter();
+  private readonly replies: ReplyLimiter;
+  private readonly limits: Pick<
+    TelegramUpdatesOptions,
+    'attempts' | 'windowMs'
+  >;
+  private readonly now: () => number;
   private holdsLease = false;
   /** Monotonic time until which Telegram asked us to stay away (429). */
   private rateLimitedUntil = 0;
@@ -128,38 +174,32 @@ export class TelegramUpdatesService {
     bigint,
     { attempts: number; firstAt: number }
   >();
-  /** Test seam: after how many attempts / how long a failing update is skipped. */
-  retryLimits = {
-    attempts: MAX_UPDATE_ATTEMPTS,
-    windowMs: MAX_UPDATE_RETRY_MS,
-  };
 
   constructor(
     private readonly prisma: PrismaService,
     private readonly client: TelegramClient,
     private readonly links: TelegramLinkService,
-  ) {}
+    @Optional()
+    @Inject(TELEGRAM_UPDATES_OPTIONS)
+    options?: Partial<TelegramUpdatesOptions>,
+  ) {
+    this.now = options?.now ?? (() => performance.now());
+    this.limits = {
+      attempts: options?.attempts ?? MAX_UPDATE_ATTEMPTS,
+      windowMs: options?.windowMs ?? MAX_UPDATE_RETRY_MS,
+    };
+    this.replies = new ReplyLimiter(this.now);
+  }
 
   /**
    * One cycle: take or extend the lease, fetch updates (long poll), handle
-   * them one by one. 'polled' means the next cycle may start at once. A
-   * cycle that ends because of shutdown hands the lease back.
+   * them one by one. 'polled' means the next cycle may start at once.
    */
   async pollOnce(signal: AbortSignal): Promise<PollOutcome> {
-    try {
-      return await this.cycle(signal);
-    } finally {
-      if (signal.aborted && this.holdsLease) {
-        await this.releaseLease();
-      }
-    }
-  }
-
-  private async cycle(signal: AbortSignal): Promise<PollOutcome> {
     if (!this.client.isConfigured()) {
       return 'disabled';
     }
-    if (signal.aborted || performance.now() < this.rateLimitedUntil) {
+    if (signal.aborted || this.now() < this.rateLimitedUntil) {
       return 'error';
     }
     const offset = await this.acquireLease();
@@ -172,6 +212,12 @@ export class TelegramUpdatesService {
     } catch (error: unknown) {
       return this.onFetchError(error, signal);
     }
+    if (batch.received > 0 && batch.highestUpdateId === null) {
+      // An answer nobody can move the offset past would be fetched again at
+      // once: report it, so the 30 s wait applies.
+      await this.setLastError('getUpdates answer without update ids');
+      return 'error';
+    }
     await this.markPolled();
     return this.applyBatch(batch, offset, signal);
   }
@@ -217,12 +263,23 @@ export class TelegramUpdatesService {
     if (dropped !== 'ok') {
       return dropped;
     }
+    this.pruneFailures(current);
     if (!skipped) {
       await this.clearLastError();
     }
     return 'polled';
   }
 
+  /** Forgets updates behind the offset and failures older than the window. */
+  private pruneFailures(offset: bigint): void {
+    const now = this.now();
+    for (const [updateId, state] of this.failures) {
+      if (updateId < offset || now - state.firstAt >= this.limits.windowMs) {
+        this.failures.delete(updateId);
+      }
+    }
+  }
+
   /** Dropped updates (no chat, wrong shape): move the offset past them, never fetch them again. */
   private async skipDropped(
     batch: TelegramUpdateBatch,
@@ -256,8 +313,14 @@ export class TelegramUpdatesService {
     return rows[0]?.updateOffset ?? null;
   }
 
-  /** Shutdown: the next receiver need not wait for the lease to run out. */
-  private async releaseLease(): Promise<void> {
+  /**
+   * Shutdown: the next receiver need not wait for the lease to run out.
+   * Call it after the loop has stopped; it does nothing without a lease.
+   */
+  async releaseLease(): Promise<void> {
+    if (!this.holdsLease) {
+      return;
+    }
     this.holdsLease = false;
     try {
       await this.prisma.$executeRaw`
@@ -307,7 +370,7 @@ export class TelegramUpdatesService {
       const seconds =
         error instanceof TelegramApiError ? error.retryAfterSec : null;
       this.rateLimitedUntil =
-        performance.now() +
+        this.now() +
         Math.min(
           (seconds ?? NO_LEASE_WAIT_MS / 1_000) * 1_000,
           MAX_RATE_LIMIT_PAUSE_MS,
@@ -342,7 +405,12 @@ export class TelegramUpdatesService {
     update: TelegramUpdate,
     error: unknown,
   ): Promise<boolean> {
-    const now = performance.now();
+    if (isConnectionError(error)) {
+      // The database or network is down: the update is not at fault.
+      await this.onUpdateError(error);
+      return false;
+    }
+    const now = this.now();
     const known = this.failures.get(update.updateId);
     const state = {
       attempts: (known?.attempts ?? 0) + 1,
@@ -350,8 +418,8 @@ export class TelegramUpdatesService {
     };
     this.failures.set(update.updateId, state);
     const exhausted =
-      state.attempts >= this.retryLimits.attempts ||
-      now - state.firstAt >= this.retryLimits.windowMs;
+      state.attempts >= this.limits.attempts ||
+      now - state.firstAt >= this.limits.windowMs;
     if (!exhausted) {
       await this.onUpdateError(error);
     }