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:

```python
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`:

```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:**
```python
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:**
```python
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):

```python
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:

```python
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

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