# Task: Digest / summary notifications ## Description For popular keywords that generate many matches per cycle, users may receive 10–20 individual notifications in quick succession. This task introduces **digest mode** — instead of immediate alerts, notifications are buffered and sent as a single summary message at configurable intervals. Digest mode is complementary to mute hours: during mute hours, all messages are suppressed; with digest mode ON, messages are collected and sent as a batch at the configured interval (default: every 60 minutes). ## Architecture ``` ┌───────────────────────────────────────┐ │ user_settings table │ │ digest_mode bool DEFAULT false │ │ digest_interval int DEFAULT 60 │ └──────────┬────────────────────────────┘ │ ▼ ┌───────────────────────────────────────┐ │ Notification Pipeline (modified) │ │ │ │ For each new ad: │ │ if user.digest_mode == false: │ │ → send immediately (current) │ │ elif in mute hours: │ │ → discard (already handled) │ │ else: │ │ → insert into digest_buffer │ └──────────┬────────────────────────────┘ │ ▼ ┌───────────────────────────────────────┐ │ digest_buffer table (new) │ │ │ │ id uuid PK │ │ telegram_id text │ │ ad_id uuid REFERENCES ads │ │ keyword text │ │ title text │ │ price int │ │ url text │ │ created_at timestamptz │ │ │ │ INDEX: telegram_id, created_at │ └──────────┬────────────────────────────┘ │ ▼ ┌───────────────────────────────────────┐ │ Digest Scheduler (separate task) │ │ │ │ Runs every digest_interval per user │ │ ├─ SELECT all buffered items │ │ ├─ GROUP BY telegram_id │ │ ├─ Format summary message │ │ └─ DELETE buffered items │ │ │ │ Summary format: │ │ 📋 Digest — 5 new ads (14:30 UTC) │ │ │ │ 🔑 "rtx 3090" (3 ads): │ │ • RTX 3090 Ti - €750 [link] │ │ • ASUS RTX 3090 - €680 [link] │ │ • MSI RTX 3090 Gaming X - €720 │ │ │ │ 🔑 "gtx 1660" (2 ads): │ │ • GTX 1660 Super - €120 [link] │ │ • EVGA GTX 1660 - €95 [link] │ └───────────────────────────────────────┘ ``` ### Key design decisions - **Separate buffer table** instead of in-memory only. Survives restarts, visible via pgAdmin for debugging. - **Per-user interval**: Each user configures their own digest frequency (default 60 min). Implemented with a single scheduler task that checks all users' intervals on each cycle. - **Group by keyword** in the summary message. Makes it easy to scan relevant categories without digging through unrelated listings. ## Implementation Details ### 1. Add migration In `worker/src/migrations/04-user-settings.sql`: ```sql -- digest_buffer table for accumulating notifications CREATE TABLE IF NOT EXISTS digest_buffer ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), telegram_id text NOT NULL, ad_id uuid REFERENCES ads(id) ON DELETE CASCADE, keyword text NOT NULL, title text NOT NULL, price int, -- in cents url text, created_at timestamptz NOT NULL DEFAULT now() ); CREATE INDEX IF NOT EXISTS idx_digest_buffer ON digest_buffer(telegram_id, created_at DESC); COMMENT ON TABLE digest_buffer IS 'Buffer for digest-mode notifications. Flushed to Telegram at intervals.'; ``` ### 2. Buffer notifications instead of sending immediately In `main.py` scheduler loop, after all filters pass and mute check passes: ```python # Check if user has digest mode enabled settings = await pool.fetchrow( "SELECT digest_mode FROM user_settings WHERE telegram_id = $1", telegram_id_str, ) if settings and settings["digest_mode"]: # Buffer for digest await pool.execute( """INSERT INTO digest_buffer (telegram_id, ad_id, keyword, title, price, url) VALUES ($1, $2, $3, $4, $5, $6)""", telegram_id_str, ad_id, kw_row["keyword"], ad_dict.get("title", "Unknown"), _extract_price(ad_dict), ad_dict.get("url"), ) # Still log the notification (for stats) await log_notify(pool, ad_id, telegram_id, "new") else: # Send immediately (current behavior) await notify_new(bot, pool, kw_row["keyword"], telegram_id, ad_dict, ad_id) ``` ### 3. Add digest flushing task to scheduler In `main.py`, add a new async function and call it at the start of each cycle: ```python async def flush_digest_buffers() -> int: """Process pending digest buffers for users whose interval has elapsed.""" from db import get_pool pool = await get_pool() # Get all users with digest mode ON users = await pool.fetch(""" SELECT telegram_id, digest_interval FROM user_settings WHERE digest_mode = true """) sent_count = 0 for user in users: interval_min = user["digest_interval"] or 60 cutoff = datetime.now(tz=timezone.utc) - timedelta(minutes=interval_min) # Get buffered items older than the interval buffered = await pool.fetch(""" SELECT db.id, db.keyword, db.title, db.price, db.url FROM digest_buffer db WHERE db.telegram_id = $1 AND db.created_at <= $2 ORDER BY db.keyword, db.created_at DESC """, user["telegram_id"], cutoff) if not buffered: continue # Group by keyword from collections import defaultdict groups: dict[str, list] = defaultdict(list) for item in buffered: price_str = f"€{item['price']/100:.2f}" if item['price'] else "Free" entry = f"• {item['title']} - {price_str}" groups[item["keyword"]].append(entry) # Build summary message lines = [f"📋 Digest — {len(buffered)} new ads ({cutoff:%H:%M}–{datetime.now(tz=timezone.utc):%H:%M} UTC)\n"] for kw_name, entries in groups.items(): lines.append(f"\n🔑 \"{kw_name}\" ({len(entries)} ads):") # Limit to 10 entries per keyword to avoid spam for entry in entries[:10]: lines.append(entry) if len(entries) > 10: lines.append(f" ... and {len(entries)-10} more") message_text = "\n".join(lines) # Send the digest try: from bot import get_application_bot bot = get_application_bot() telegram_id_int = int(user["telegram_id"]) await bot.send_message(chat_id=telegram_id_int, text=message_text) sent_count += 1 # Log all buffered notifications as delivered buffer_ids = [item["id"] for item in buffered] for ad_item in buffered: await log_notify(pool, ad_item["ad_id"], telegram_id_int, "new") except TelegramError as e: logger.error("Digest send failed for %s: %s", user["telegram_id"], e) finally: # Clear the buffer (whether sent or not — if it failed, log entries remain in DB) await pool.execute( """DELETE FROM digest_buffer WHERE telegram_id = $1 AND created_at <= $2""", user["telegram_id"], cutoff, ) return sent_count # Call at the start of run_scheduler(): async def run_scheduler() -> None: while True: try: record_scheduler_run() # Flush digest buffers first digests_sent = await flush_digest_buffers() if digests_sent: logger.info("Sent %d digest summaries", digests_sent) # Process notification queue... processed = await process_notification_queue() if processed: logger.info("Retried %d queued notifications", processed) # ... existing keyword iteration ... ``` ### 4. Add bot commands in `bot.py` ```python async def cmd_digest_on(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Enable digest mode with optional interval.""" interval = 60 # default minutes if len(context.args) > 0: try: interval = int(context.args[0]) if interval < 5 or interval > 1440: raise ValueError except ValueError: await update.message.reply_text( "Usage: /digest_on [minutes]\n" "Interval must be between 5 and 1440 minutes (24h)." ) return telegram_id = str(update.effective_user.id) from db import get_pool pool = await get_pool() await pool.execute( """INSERT INTO user_settings (telegram_id, digest_mode, digest_interval) VALUES ($1, true, $2) ON CONFLICT (telegram_id) DO UPDATE SET digest_mode = EXCLUDED.digest_mode, digest_interval = EXCLUDED.digest_interval""", telegram_id, interval, ) await update.message.reply_text( f"✅ Digest mode ENABLED\n" f"Digests will be sent every {interval} minutes.\n" "Use /digest_off to return to instant notifications." ) async def cmd_digest_off(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Disable digest mode.""" telegram_id = str(update.effective_user.id) from db import get_pool pool = await get_pool() await pool.execute( """INSERT INTO user_settings (telegram_id, digest_mode) VALUES ($1, false) ON CONFLICT (telegram_id) DO UPDATE SET digest_mode = EXCLUDED.digest_mode""", telegram_id, ) # Flush any remaining buffered items immediately await pool.execute( """DELETE FROM digest_buffer WHERE telegram_id = $1""", telegram_id, ) await update.message.reply_text( "✅ Digest mode DISABLED — notifications are now instant." ) async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Show current user settings (extended from mute hours task).""" telegram_id = str(update.effective_user.id) from db import get_pool pool = await get_pool() settings = await pool.fetchrow( "SELECT mute_start, mute_end, digest_mode, digest_interval " "FROM user_settings WHERE telegram_id = $1", telegram_id, ) reply_parts = [] if not settings or not settings["mute_start"]: reply_parts.append("🔕 Mute hours: OFF") else: reply_parts.append( f"🔕 Mute hours: {settings['mute_start']} — {settings['mute_end']} UTC" ) if settings and settings["digest_mode"]: reply_parts.append( f"📋 Digest: ON (every {settings['digest_interval']} min)" ) else: reply_parts.append("📋 Digest: OFF (instant notifications)") await update.message.reply_text("\n".join(reply_parts)) ``` Register handlers: ```python dp.add_handler(CommandHandler("digest_on", cmd_digest_on)) dp.add_handler(CommandHandler("digest_off", cmd_digest_off)) ``` ## Acceptance Criteria - [ ] `/digest_on` enables digest mode with 60-minute default interval - [ ] `/digest_on 30` sets digest interval to 30 minutes - [ ] New ads are inserted into `digest_buffer` instead of being sent immediately when digest is ON - [ ] At the configured interval, all buffered items are flushed as a single summary message - [ ] The summary groups ads by keyword and includes price information - [ ] After flushing, buffered items are deleted from the table - [ ] `/digest_off` disables digest mode and sends any remaining buffered items immediately - [ ] Mute hours take precedence over digest — muted notifications are discarded, not buffered