willhaben-tracker/docs/phase-1/task-notification-retry-queue.md

Task: Retry queue for failed notifications

Description

The current notifier.notify_new() and notify_drop() call _send_message() directly. If the Telegram API returns an error (rate limiting, network hiccup, user deleted the bot), the notification is silently logged and never retried:

async def _send_message(bot: Bot, chat_id: int, message_text):
    try:
        await bot.send_message(chat_id=chat_id, text=message_text)
    except TelegramError as e:
        logger.warning("Failed to send notification ...")  # ← notification LOST forever

This task introduces a persistent retry queue backed by the notification_queue table. Failed notifications are stored with an attempt counter and retried on subsequent scheduler cycles with exponential backoff. After 5 failed attempts, they're marked as dead and logged — not silently dropped.

Architecture

┌───────────────────────────────────────┐
│  notification_queue table (new)       │
│                                       │
│  id            uuid PK                │
│  ad_id         uuid                   │
│  telegram_id   text                   │
│  message_text  text                   │
│  type          enum('new','drop')     │
│  attempts      int DEFAULT 0          │
│  max_attempts  int DEFAULT 5          │
│  last_error    text                   │
│  status        enum                  │
│              ('pending','sent',       │
│               'failed','dead')        │
│  created_at    timestamptz            │
│  updated_at    timestamptz            │
│                                       │
│  INDEX: status, attempts (composite)  │
└───────────┬───────────────────────────┘
            │
            ▼
┌───────────────────────────────────────┐
│  notifier.py flow:                    │
│                                       │
│  send_notification():                 │
│    try:                               │
│      await bot.send_message(...)     │
│      → log_notify() (as before)       │
│    except TelegramError as e:         │
│      INSERT INTO notification_queue   │
│        (ad_id, telegram_id, ...)      │
│        VALUES (...)                   │
└───────────┬───────────────────────────┘
            │
            ▼
┌───────────────────────────────────────┐
│  scheduler.py / main.py:              │
│  At the START of each cycle:          │
│                                       │
│    for item in pending_queue:         │
│      if attempts < max_attempts:      │
│        backoff = 2^attempts minutes   │
│        if now >= updated_at + backoff:│
│          try send again              │
│            success → mark 'sent'     │
│            fail    → increment count │
│      elif attempts >= max_attempts:   │
│        mark as 'dead'                │
│        log warning                   │
└───────────────────────────────────────┘

Exponential backoff schedule:
  Attempt 1 → wait 1 min (2^0)
  Attempt 2 → wait 2 min  (2^1)
  Attempt 3 → wait 4 min  (2^2)
  Attempt 4 → wait 8 min  (2^3)
  Attempt 5 → wait 16 min (2^4)
  Total worst case: ~31 min before giving up

Key design decisions

• Store full message text in the queue table so we can retry without re-rendering. This is important because ad data might change or be removed from willhaben by the time we retry.
• Process at start of scheduler cycle — ensures queued items are attempted before new scraping starts, prioritizing user notifications over fresh data collection.
• Backoff based on updated_at, not wall-clock from first attempt. Each retry resets the backoff timer. This handles edge cases where a failure was transient but then another transient follows.

Implementation Details

1. Add migration for notification_queue table

In worker/src/migrations/02-notification-queue.sql:

-- Notification retry queue table

CREATE TYPE notification_type AS ENUM ('new', 'drop');
CREATE TYPE notification_status AS ENUM ('pending', 'sent', 'failed', 'dead');

CREATE TABLE IF NOT EXISTS notification_queue (
    id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    ad_id         uuid NOT NULL REFERENCES ads(id) ON DELETE SET NULL,
    telegram_id   text NOT NULL,
    message_text  text NOT NULL,
    type          notification_type NOT NULL,
    attempts      int NOT NULL DEFAULT 0,
    max_attempts  int NOT NULL DEFAULT 5,
    last_error    text,
    status        notification_status NOT NULL DEFAULT 'pending',
    created_at    timestamptz NOT NULL DEFAULT now(),
    updated_at    timestamptz NOT NULL DEFAULT now()
);

-- Index for efficient queue polling
CREATE INDEX IF NOT EXISTS idx_notif_queue_poll 
    ON notification_queue(status, attempts, updated_at)
    WHERE status IN ('pending', 'failed');

COMMENT ON TABLE notification_queue IS 
    'Persistent retry queue for failed Telegram notifications';

2. Update notifier.py — enqueue on failure

Current:

async def _send_message(bot: Bot, chat_id: int, message_text):
    try:
        await bot.send_message(chat_id=chat_id, text=message_text)
    except TelegramError as e:
        logger.warning(
            "Failed to send notification to %d: %s", chat_id, str(e)[:30]
        )

After:

async def _send_message(
    bot: Bot, 
    chat_id: int, 
    message_text: str,
    ad_id: uuid.UUID | None = None,
    notif_type: str = "new",
):
    try:
        await bot.send_message(chat_id=chat_id, text=message_text)
    except TelegramError as e:
        error_msg = str(e)[:300]  # cap length
        logger.warning("Telegram send failed for %s: %s", chat_id, error_msg)
        
        if ad_id:
            await _enqueue_retry(
                ad_id=ad_id,
                telegram_id=str(chat_id),
                message_text=message_text,
                notif_type=notif_type,
                error_msg=error_msg,
            )


async def _enqueue_retry(
    ad_id: uuid.UUID,
    telegram_id: str,
    message_text: str,
    notif_type: str,
    error_msg: str,
) -> None:
    """Store a failed notification for later retry."""
    from db import get_pool
    
    pool = await get_pool()
    
    # Check if already queued (avoid duplicates for same ad+user)
    existing = await pool.fetchval(
        """SELECT id FROM notification_queue 
           WHERE ad_id = $1 AND telegram_id = $2 AND status IN ('pending', 'failed')""",
        ad_id, telegram_id,
    )
    
    if existing:
        logger.info("Already queued: ad=%s user=%s", ad_id[:8], telegram_id)
        return
    
    await pool.execute(
        """INSERT INTO notification_queue 
           (ad_id, telegram_id, message_text, type, last_error, status)
           VALUES ($1, $2, $3, $4, $5, 'pending')""",
        ad_id, telegram_id, message_text, notif_type, error_msg,
    )
    
    logger.info("Queued for retry: ad=%s user=%s", ad_id[:8], telegram_id)

3. Add queue processing to scheduler in main.py

At the start of each scheduler cycle (before keyword scraping):

async def process_notification_queue() -> int:
    """Process pending notifications from the retry queue."""
    from db import get_pool
    
    pool = await get_pool()
    
    # Get items eligible for retry (backoff respected)
    rows = await pool.fetch("""
        SELECT id, ad_id, telegram_id, message_text, type, attempts, 
               max_attempts, last_error, updated_at
        FROM notification_queue
        WHERE status IN ('pending', 'failed')
          AND updated_at + ($1 || ' minutes')::interval <= now()
        ORDER BY attempts ASC, updated_at ASC
    """, "2^attempts" if pool.is_pg else 0)

    # Actually use a computed backoff in Python since PG expressions are tricky:
    rows = await pool.fetch("""
        SELECT id, ad_id, telegram_id, message_text, type, attempts, 
               max_attempts, last_error, updated_at
        FROM notification_queue
        WHERE status IN ('pending', 'failed')
        ORDER BY attempts ASC, updated_at ASC
        LIMIT 50
    """)

    processed = 0
    
    for row in rows:
        backoff_min = min(2 ** row["attempts"], 60)  # cap at 60 min
        retry_after = row["updated_at"] + timedelta(minutes=backoff_min)
        
        if datetime.now(tz=timezone.utc) < retry_after:
            continue  # not yet eligible
        
        try:
            from bot import get_application_bot
            bot = get_application_bot()
            
            await bot.send_message(
                chat_id=int(row["telegram_id"]), 
                text=row["message_text"]
            )
            
            await pool.execute(
                "UPDATE notification_queue SET status = 'sent', updated_at = now() WHERE id = $1",
                row["id"],
            )
            
            await log_notify(pool, row["ad_id"], int(row["telegram_id"]), row["type"])
            
            processed += 1
            
        except TelegramError as e:
            new_attempts = row["attempts"] + 1
            
            if new_attempts >= row["max_attempts"]:
                await pool.execute(
                    """UPDATE notification_queue 
                       SET status = 'dead', attempts = $2, last_error = $3, updated_at = now()
                       WHERE id = $1""",
                    row["id"], new_attempts, str(e)[:300],
                )
                logger.error(
                    "Notification DEAD after %d attempts: ad=%s user=%s err=%s",
                    new_attempts, row["ad_id"][:8], row["telegram_id"], e,
                )
            else:
                await pool.execute(
                    """UPDATE notification_queue 
                       SET status = 'failed', attempts = $2, last_error = $3, updated_at = now()
                       WHERE id = $1""",
                    row["id"], new_attempts, str(e)[:300],
                )
    
    return processed

4. Call process_notification_queue() in the scheduler loop

In main.py, before iterating keywords:

async def run_scheduler() -> None:
    while True:
        try:
            record_scheduler_run()  # healthcheck
            
            # Process pending notifications FIRST
            processed = await process_notification_queue()
            if processed:
                logger.info("Retried %d queued notifications", processed)
            
            # ... existing keyword iteration ...

Acceptance Criteria

• When _send_message() raises TelegramError, the notification is INSERTed into notification_queue with status='pending'
• On the next scheduler cycle, pending items are attempted (respecting backoff)
• After 5 failed attempts, the notification status becomes 'dead' and a warning is logged
• The queue processes at most 50 items per cycle to avoid blocking the scheduler
• Duplicate enqueue prevention works: calling _enqueue_retry twice for the same ad+user creates only one queue entry
• Successful retries update log_notifications table (same as direct notifications)
• The /health endpoint or logs can show the current count of pending/dead items