13 KiB
13 KiB
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:
-- 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:
# 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:
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
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:
dp.add_handler(CommandHandler("digest_on", cmd_digest_on))
dp.add_handler(CommandHandler("digest_off", cmd_digest_off))
Acceptance Criteria
/digest_onenables digest mode with 60-minute default interval/digest_on 30sets digest interval to 30 minutes- New ads are inserted into
digest_bufferinstead 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_offdisables digest mode and sends any remaining buffered items immediately- Mute hours take precedence over digest — muted notifications are discarded, not buffered