356 lines
13 KiB
Markdown
356 lines
13 KiB
Markdown
# 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
|