feat: Phase 2 — user experience & advanced filtering

- Add price_min/price_max/allowed_postcode filters to keywords table (migration 05)
- Add user_settings table with mute hours + digest mode (migration 06)
- Add digest_buffer table for batching notifications
- Filter ads by price/postcode in scheduler before processing
- Mute hours check in notifier (discards muted notifications)
- Digest buffering: buffers notifications then flushes as summary per interval
- Bot commands: /price_min, /price_max, /clear_price, /postcode, /clear_postcode
- Bot commands: /mute_hours, /mute_off, /digest_on, /digest_off, /status
- Update keyword card display to show active price range and postcodes

worker/src/bot.py

@@ -188,11 +188,23 @@ def _admin_menu_keyboard() -> InlineKeyboardMarkup:
 def _format_kw_card(kw: dict) -> str:
     status_icon = "🟢 Active" if kw["is_active"] else "🔴 Stopped"
     subs_line = f"\nSubscribers: <code>{kw['subs']}</code>" if kw.get("subs", 1) > 1 else ""
-
+    
+    filters = ""
+    price_min = kw.get("price_min")
+    price_max = kw.get("price_max")
+    if price_min is not None or price_max is not None:
+        min_str = f"{(price_min or 0) / 100:.0f}" if price_min is not None else "0"
+        max_str = f"{price_max / 100:.0f}" if price_max is not None else "∞"
+        filters = f"\n💰 Price: <code>€{min_str}–€{max_str}</code>"
+    
+    postcodes = kw.get("allowed_postcodes")
+    if postcodes:
+        filters += f"\n📍 Postcodes: <code>{','.join(str(p) for p in postcodes)}</code>"
+    
     return (
         f"<b>🔍 {kw['keyword']}</b>\n"
         f"{status_icon} | Interval: <code>{kw['interval_minutes']} min</code>\n"
-        f"Last scrape: {_vienna_time(kw.get('last_scraped_at'))}{subs_line}"
+        f"Last scrape: {_vienna_time(kw.get('last_scraped_at'))}{filters}{subs_line}"
     )
 
 
@@ -201,7 +213,7 @@ async def _refresh_kw_card(bot, chat_id: int, message_id: int, kw_id_full: str):
     pool = await get_pool()
     # Accept either full UUID or truncated prefix — both work as LIKE match
     kw = await pool.fetchrow(
-        "SELECT kw.id, kw.keyword, kw.is_active, kw.interval_minutes, kw.last_scraped_at, COUNT(ks.user_id) AS subs "
+        "SELECT kw.id, kw.keyword, kw.is_active, kw.interval_minutes, kw.last_scraped_at, kw.price_min, kw.price_max, kw.allowed_postcodes, COUNT(ks.user_id) AS subs "
         "FROM keywords kw LEFT JOIN keyword_subscriptions ks ON ks.keyword_id = kw.id "
         "WHERE kw.id::text LIKE $1 || '%' GROUP BY kw.id",
         kw_id_full + "%",
@@ -226,12 +238,399 @@ async def setup_global_commands(app: Application) -> None:
     await app.bot.set_my_commands([
         ("start", "Open main menu"),
         ("admin", "Admin panel (admins only)"),
+        ("price_min", "Set min price for keyword"),
+        ("price_max", "Set max price for keyword"),
+        ("clear_price", "Clear price filters"),
+        ("postcode", "Set postcodes for keyword"),
+        ("clear_postcode", "Clear postcode filter"),
+        ("mute_hours", "Set mute hours"),
+        ("mute_off", "Disable mute hours"),
+        ("digest_on", "Enable digest mode"),
+        ("digest_off", "Disable digest mode"),
+        ("status", "Show your settings"),
     ])
 
 
 def register_handlers(app: Application) -> None:
     app.add_handler(CommandHandler("start", start_handler))
     app.add_handler(CommandHandler("admin", admin_handler))
+    app.add_handler(CommandHandler("price_min", price_min_handler))
+    app.add_handler(CommandHandler("price_max", price_max_handler))
+    app.add_handler(CommandHandler("clear_price", clear_price_handler))
+    app.add_handler(CommandHandler("postcode", postcode_handler))
+    app.add_handler(CommandHandler("clear_postcode", clear_postcode_handler))
+    app.add_handler(CommandHandler("mute_hours", mute_hours_handler))
+    app.add_handler(CommandHandler("mute_off", mute_off_handler))
+    app.add_handler(CommandHandler("digest_on", digest_on_handler))
+    app.add_handler(CommandHandler("digest_off", digest_off_handler))
+    app.add_handler(CommandHandler("status", status_handler))
+    app.add_handler(CallbackQueryHandler(callback_router))
+    # Catch all non-command text messages for conversation flows
+    app.add_handler(MessageHandler(TEXT_FILTER, text_input_handler))
+
+
+# ── /start ────────────────────────────────────────────────────────────────
+
+async def start_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    row = await _auto_register(update)
+    if not row:
+        return
+
+    name = update.effective_user.first_name or "there"
+    chat_id = update.effective_chat.id  # type: ignore[union-attr]
+    await context.bot.send_message(
+        chat_id=chat_id,
+        text=f"Hello <b>{name}</b>! I'll notify you about new willhaben listings.",
+        parse_mode="HTML",
+        reply_markup=_main_menu_keyboard(),
+    )
+
+
+# ── /admin ────────────────────────────────────────────────────────────────
+
+async def admin_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    row = await _require_admin(update)
+    if not row:
+        return
+
+    chat_id = update.effective_chat.id  # type: ignore[union-attr]
+    await context.bot.send_message(
+        chat_id=chat_id, text="<b>⚙️ Admin Panel</b>",
+        parse_mode="HTML", reply_markup=_admin_menu_keyboard(),
+    )
+
+
+# ── price filter handlers ─────────────────────────────────────────────────
+
+async def price_min_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Set minimum price for a keyword. Usage: /price_min keyword 50"""
+    await _set_price_filter(update, context, "price_min")
+
+
+async def price_max_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Set maximum price for a keyword. Usage: /price_max keyword 1000"""
+    await _set_price_filter(update, context, "price_max")
+
+
+async def _set_price_filter(update: Update, context: ContextTypes.DEFAULT_TYPE, column: str) -> None:
+    user = await _require_user(update)
+    if not user or not update.message:
+        return
+    
+    text = (update.message.text or "").strip()  # type: ignore[union-attr]
+    parts = text.split()
+    if len(parts) < 3:
+        await update.message.reply_text(  # type: ignore[union-attr]
+            f"Usage: <code>/{update.message.text.split()[0][1:]} keyword 50</code> (price in €)",
+            parse_mode="HTML")
+        return
+    
+    keyword = " ".join(parts[1:-1])
+    try:
+        price_euros = float(parts[-1])
+    except ValueError:
+        await update.message.reply_text("Invalid price. Enter a number in €.")  # type: ignore[union-attr]
+        return
+    
+    price_cents = int(round(price_euros * 100))
+    pool = await get_pool()
+    
+    # Find keyword — non-admins can only modify their own keywords
+    if user.get("is_admin"):
+        kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+    else:
+        kw = await pool.fetchrow(
+            "SELECT kw.id FROM keywords kw JOIN keyword_subscriptions ks ON ks.keyword_id = kw.id "
+            "WHERE LOWER(kw.keyword) = LOWER($1) AND ks.user_id = $2",
+            keyword, user["id"])
+    
+    if not kw:
+        await update.message.reply_text(f"Keyword <code>{keyword}</code> not found.")  # type: ignore[union-attr]
+        return
+    
+    await pool.execute(f"UPDATE keywords SET {column} = $1 WHERE id = $2", price_cents, kw["id"])
+    label = "Min" if column == "price_min" else "Max"
+    await update.message.reply_text(  # type: ignore[union-attr]
+        f"✅ <b>{label}</b> price for <code>{keyword}</code> set to <code>€{price_euros:.0f}</code>.",
+        parse_mode="HTML")
+
+
+async def clear_price_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Clear price filters for a keyword. Usage: /clear_price keyword"""
+    user = await _require_user(update)
+    if not user or not update.message:
+        return
+    
+    text = (update.message.text or "").strip()  # type: ignore[union-attr]
+    parts = text.split()
+    if len(parts) < 2:
+        await update.message.reply_text("Usage: <code>/clear_price keyword</code>", parse_mode="HTML")
+        return
+    
+    keyword = " ".join(parts[1:])
+    pool = await get_pool()
+    
+    if user.get("is_admin"):
+        kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+    else:
+        kw = await pool.fetchrow(
+            "SELECT kw.id FROM keywords kw JOIN keyword_subscriptions ks ON ks.keyword_id = kw.id "
+            "WHERE LOWER(kw.keyword) = LOWER($1) AND ks.user_id = $2",
+            keyword, user["id"])
+    
+    if not kw:
+        await update.message.reply_text(f"Keyword <code>{keyword}</code> not found.")  # type: ignore[union-attr]
+        return
+    
+    await pool.execute("UPDATE keywords SET price_min = NULL, price_max = NULL WHERE id = $1", kw["id"])
+    await update.message.reply_text(  # type: ignore[union-attr]
+        f"✅ Price filters cleared for <code>{keyword}</code>.", parse_mode="HTML")
+
+
+# ── postcode filter handlers ──────────────────────────────────────────────
+
+async def postcode_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Set allowed postcodes for a keyword. Usage: /postcode keyword 1010,1020,1030"""
+    user = await _require_user(update)
+    if not user or not update.message:
+        return
+    
+    text = (update.message.text or "").strip()  # type: ignore[union-attr]
+    parts = text.split()
+    if len(parts) < 3:
+        await update.message.reply_text("Usage: <code>/postcode keyword 1010,1020</code>", parse_mode="HTML")
+        return
+    
+    keyword = " ".join(parts[1:-1])
+    pc_str = parts[-1]
+    postcodes = [p.strip() for p in pc_str.split(",")]
+    
+    # Validate postcodes
+    for p in postcodes:
+        if not p.isdigit() or len(p) < 3:
+            await update.message.reply_text(f"Invalid postcode: <code>{p}</code>. Use numeric postcodes.")  # type: ignore[union-attr]
+            return
+    
+    pool = await get_pool()
+    
+    if user.get("is_admin"):
+        kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+    else:
+        kw = await pool.fetchrow(
+            "SELECT kw.id FROM keywords kw JOIN keyword_subscriptions ks ON ks.keyword_id = kw.id "
+            "WHERE LOWER(kw.keyword) = LOWER($1) AND ks.user_id = $2",
+            keyword, user["id"])
+    
+    if not kw:
+        await update.message.reply_text(f"Keyword <code>{keyword}</code> not found.")  # type: ignore[union-attr]
+        return
+    
+    await pool.execute("UPDATE keywords SET allowed_postcodes = $1 WHERE id = $2", postcodes, kw["id"])
+    await update.message.reply_text(  # type: ignore[union-attr]
+        f"✅ Postcodes for <code>{keyword}</code> set to: <code>{','.join(postcodes)}</code>.",
+        parse_mode="HTML")
+
+
+async def clear_postcode_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Clear postcode filter for a keyword. Usage: /clear_postcode keyword"""
+    user = await _require_user(update)
+    if not user or not update.message:
+        return
+    
+    text = (update.message.text or "").strip()  # type: ignore[union-attr]
+    parts = text.split()
+    if len(parts) < 2:
+        await update.message.reply_text("Usage: <code>/clear_postcode keyword</code>", parse_mode="HTML")
+        return
+    
+    keyword = " ".join(parts[1:])
+    pool = await get_pool()
+    
+    if user.get("is_admin"):
+        kw = await pool.fetchrow("SELECT id FROM keywords WHERE LOWER(keyword) = LOWER($1)", keyword)
+    else:
+        kw = await pool.fetchrow(
+            "SELECT kw.id FROM keywords kw JOIN keyword_subscriptions ks ON ks.keyword_id = kw.id "
+            "WHERE LOWER(kw.keyword) = LOWER($1) AND ks.user_id = $2",
+            keyword, user["id"])
+    
+    if not kw:
+        await update.message.reply_text(f"Keyword <code>{keyword}</code> not found.")  # type: ignore[union-attr]
+        return
+    
+    await pool.execute("UPDATE keywords SET allowed_postcodes = NULL WHERE id = $1", kw["id"])
+    await update.message.reply_text(  # type: ignore[union-attr]
+        f"✅ Postcode filter cleared for <code>{keyword}</code>.", parse_mode="HTML")
+
+
+# ── mute handlers ─────────────────────────────────────────────────────────
+
+async def mute_hours_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Set mute hours. Usage: /mute_hours 22:00-07:00"""
+    user = await _require_user(update)
+    if not user or not update.message:
+        return
+    
+    text = (update.message.text or "").strip()  # type: ignore[union-attr]
+    parts = text.split()
+    if len(parts) < 2:
+        await update.message.reply_text("Usage: <code>/mute_hours 22:00-07:00</code> (times in UTC)", parse_mode="HTML")
+        return
+    
+    time_range = parts[1]
+    try:
+        start_str, end_str = time_range.split("-")
+        # Validate time format
+        _ = datetime.strptime(start_str.strip(), "%H:%M")
+        _ = datetime.strptime(end_str.strip(), "%H:%M")
+    except (ValueError, AttributeError):
+        await update.message.reply_text("Invalid format. Use HH:MM-HH:MM (e.g., 22:00-07:00).")  # type: ignore[union-attr]
+        return
+    
+    pool = await get_pool()
+    await pool.execute(
+        """INSERT INTO user_settings (telegram_id, mute_start, mute_end)
+           VALUES ($1, $2, $3)
+           ON CONFLICT (telegram_id) DO UPDATE SET mute_start = $2, mute_end = $3""",
+        str(user["telegram_id"]), start_str.strip(), end_str.strip())
+    
+    await update.message.reply_text(  # type: ignore[union-attr]
+        f"✅ Mute hours set to <code>{start_str.strip()}–{end_str.strip()}</code> UTC.\nNotifications during this window will be suppressed.",
+        parse_mode="HTML")
+
+
+async def mute_off_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Disable mute hours."""
+    user = await _require_user(update)
+    if not user:
+        return
+    
+    pool = await get_pool()
+    await pool.execute(
+        """INSERT INTO user_settings (telegram_id) VALUES ($1)
+           ON CONFLICT (telegram_id) DO UPDATE SET mute_start = NULL, mute_end = NULL""",
+        str(user["telegram_id"]))
+    
+    msg = update.message or update.callback_query  # type: ignore[union-attr]
+    await msg.reply_text("✅ Mute hours disabled.", parse_mode="HTML")
+
+
+# ── digest handlers ───────────────────────────────────────────────────────
+
+async def digest_on_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Enable digest mode. Usage: /digest_on [minutes]"""
+    user = await _require_user(update)
+    if not user:
+        return
+    
+    interval = 60  # default
+    text = (update.message.text or "").strip()  # type: ignore[union-attr]
+    parts = text.split()
+    if len(parts) > 1:
+        try:
+            interval = int(parts[1])
+            if interval < 5 or interval > 1440:
+                raise ValueError
+        except ValueError:
+            msg = update.message or update.callback_query  # type: ignore[union-attr]
+            await msg.reply_text("Interval must be between 5 and 1440 minutes.")  # type: ignore[union-attr]
+            return
+    
+    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 = true, digest_interval = $2""",
+        str(user["telegram_id"]), interval)
+    
+    msg = update.message or update.callback_query  # type: ignore[union-attr]
+    await msg.reply_text(  # type: ignore[union-attr]
+        f"✅ Digest mode enabled. Notifications will be bundled every <code>{interval} min</code>.",
+        parse_mode="HTML")
+
+
+async def digest_off_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Disable digest mode and flush remaining buffer."""
+    user = await _require_user(update)
+    if not user:
+        return
+    
+    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 = false""",
+        str(user["telegram_id"]))
+    
+    # Flush remaining digest buffer — send remaining items as individual notifications
+    # This is handled by the scheduler's flush_digests function, but we also clear the buffer here
+    await pool.execute("DELETE FROM digest_buffer WHERE telegram_id = $1", str(user["telegram_id"]))
+    
+    msg = update.message or update.callback_query  # type: ignore[union-attr]
+    await msg.reply_text("✅ Digest mode disabled. Notifications will be sent immediately again.")  # type: ignore[union-attr]
+
+
+# ── status handler ────────────────────────────────────────────────────────
+
+async def status_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    """Show current user settings (mute + digest)."""
+    user = await _require_user(update)
+    if not user:
+        return
+    
+    pool = await get_pool()
+    settings = await pool.fetchrow(
+        "SELECT mute_start, mute_end, digest_mode, digest_interval FROM user_settings WHERE telegram_id = $1",
+        str(user["telegram_id"]))
+    
+    lines = ["<b>⚙️ Your Settings</b>"]
+    
+    if settings and (settings["mute_start"] is not None):
+        lines.append(f"\n🔇 Mute hours: <code>{settings['mute_start']}–{settings['mute_end']}</code> UTC")
+    else:
+        lines.append("\n🔇 Mute hours: off")
+    
+    if settings and settings["digest_mode"]:
+        lines.append(f"📦 Digest: on (<code>{settings['digest_interval']} min</code>)")
+    else:
+        lines.append("📦 Digest: off")
+    
+    msg = update.message or update.callback_query  # type: ignore[union-attr]
+    await msg.reply_text("\n".join(lines), parse_mode="HTML")  # type: ignore[union-attr]
+
+
+# ── handler registration + global commands ────────────────────────────────
+
+async def setup_global_commands(app: Application) -> None:
+    """Call this from main() after building the app to register bot commands."""
+    await app.bot.set_my_commands([
+        ("start", "Open main menu"),
+        ("admin", "Admin panel (admins only)"),
+        ("price_min", "Set min price for keyword"),
+        ("price_max", "Set max price for keyword"),
+        ("clear_price", "Clear price filters"),
+        ("postcode", "Set postcodes for keyword"),
+        ("clear_postcode", "Clear postcode filter"),
+        ("mute_hours", "Set mute hours"),
+        ("mute_off", "Disable mute hours"),
+        ("digest_on", "Enable digest mode"),
+        ("digest_off", "Disable digest mode"),
+        ("status", "Show your settings"),
+    ])
+
+
+def register_handlers(app: Application) -> None:
+    app.add_handler(CommandHandler("start", start_handler))
+    app.add_handler(CommandHandler("admin", admin_handler))
+    app.add_handler(CommandHandler("price_min", price_min_handler))
+    app.add_handler(CommandHandler("price_max", price_max_handler))
+    app.add_handler(CommandHandler("clear_price", clear_price_handler))
+    app.add_handler(CommandHandler("postcode", postcode_handler))
+    app.add_handler(CommandHandler("clear_postcode", clear_postcode_handler))
+    app.add_handler(CommandHandler("mute_hours", mute_hours_handler))
+    app.add_handler(CommandHandler("mute_off", mute_off_handler))
+    app.add_handler(CommandHandler("digest_on", digest_on_handler))
+    app.add_handler(CommandHandler("digest_off", digest_off_handler))
+    app.add_handler(CommandHandler("status", status_handler))
     app.add_handler(CallbackQueryHandler(callback_router))
     # Catch all non-command text messages for conversation flows
     app.add_handler(MessageHandler(TEXT_FILTER, text_input_handler))
@@ -485,7 +884,8 @@ async def _handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -
         elif sub == "list":
             subs = await pool.fetch(
                 "SELECT kw.id, kw.keyword, kw.is_active, kw.interval_minutes, "
-                "kw.last_scraped_at, COUNT(ks2.user_id) AS subs "
+                "kw.last_scraped_at, kw.price_min, kw.price_max, kw.allowed_postcodes, "
+                "COUNT(ks2.user_id) AS subs "
                 "FROM keyword_subscriptions ks "
                 "JOIN keywords kw ON kw.id = ks.keyword_id "
                 "LEFT JOIN keyword_subscriptions ks2 ON ks2.keyword_id = ks.keyword_id "

worker/src/main.py

@@ -4,6 +4,7 @@ import logging
 import os
 import signal
 import sys
+from collections import defaultdict
 from contextlib import suppress
 from datetime import datetime, timedelta, timezone
 from typing import Any
@@ -24,6 +25,30 @@ logger = logging.getLogger(__name__)
 load_dotenv()
 
 
+def _ad_passes_filters(fields: dict, kw_row: dict) -> bool:
+    """Check if an ad passes the keyword's price and postcode filters."""
+    price = fields.get("price")
+
+    # Price filter (stored in cents, ad price is in euros as float)
+    if price is not None:
+        price_cents = int(round(price * 100))
+        price_min = kw_row.get("price_min")
+        price_max = kw_row.get("price_max")
+        if price_min is not None and price_cents < price_min:
+            return False
+        if price_max is not None and price_cents > price_max:
+            return False
+
+    # Postcode filter
+    allowed_postcodes = kw_row.get("allowed_postcodes")
+    if allowed_postcodes is not None:
+        ad_postcode = fields.get("postcode")
+        if ad_postcode is None or ad_postcode not in allowed_postcodes:
+            return False
+
+    return True
+
+
 async def process_notification_queue(pool: asyncpg.Pool, bot: ExtBot) -> int:
     """Process pending notifications from the retry queue."""
     rows = await pool.fetch("""
@@ -89,6 +114,83 @@ async def process_notification_queue(pool: asyncpg.Pool, bot: ExtBot) -> int:
     return processed
 
 
+async def flush_digests(pool: asyncpg.Pool, bot: ExtBot) -> int:
+    """Flush digest buffers for users whose interval has elapsed."""
+    users = await pool.fetch("""
+        SELECT us.telegram_id, us.digest_interval, us.last_digest_flush
+        FROM user_settings us
+        WHERE us.digest_mode = true
+        AND (us.last_digest_flush IS NULL 
+             OR us.last_digest_flush < now() - (us.digest_interval || ' minutes')::interval)
+    """)
+    
+    if not u
+            digested = await flush_digests(pool, bot)
+            if digested:
+                logger.info("Flushed %d digest summaries", digested)
+        except Exception:
+            logger.exception("Error flushing digests")
+
+        try:sers:
+        return 0
+    
+    flushed = 0
+    
+    for user_row in users:
+        tg_id = user_row["telegram_id"]
+        
+        # Get all buffered notifications for this user
+        buffered = await pool.fetch("""
+            SELECT db.id, db.ad_id, db.keyword, db.title, db.price, db.url
+            FROM digest_buffer db
+            WHERE db.telegram_id = $1
+            ORDER BY db.created_at DESC
+        """, tg_id)
+        
+        if not buffered:
+            # Update flush time even if no items (to keep tracking)
+            await pool.execute(
+                """INSERT INTO user_settings (telegram_id, last_digest_flush) VALUES ($1, now())
+                   ON CONFLICT (telegram_id) DO UPDATE SET last_digest_flush = now()""",
+                tg_id)
+            continue
+        
+        # Build digest message grouped by keyword
+        by_keyword = defaultdict(list)
+        for item in buffered:
+            by_keyword[item["keyword"]].append(item)
+        
+        lines = ["📦 <b>Digest Summary</b>"]
+        
+        for keyword, items in by_keyword.items():
+            lines.append(f"\n<b>🔍 {keyword}</b> ({len(items)} ad{'s' if len(items) != 1 else ''})")
+            for item in items[:20]:  # Limit to 20 ads per keyword to avoid message too long
+                price_str = f"€{item['price'] / 100:.0f}" if item["price"] else "N/A"
+                lines.append(f"  • {item['title']} — {price_str}")
+        
+        text = "\n".join(lines)
+        
+        try:
+            await bot.send_message(
+                chat_id=int(tg_id),
+                text=text,
+                parse_mode="HTML",
+            )
+            flushed += 1
+            logger.info("Sent digest to %s (%d items)", tg_id, len(buffered))
+        except Exception as e:
+            logger.error("Failed to send digest to %s: %s", tg_id, e)
+        
+        # Clear buffered items and update flush time
+        await pool.execute("DELETE FROM digest_buffer WHERE telegram_id = $1", tg_id)
+        await pool.execute(
+            """INSERT INTO user_settings (telegram_id, last_digest_flush) VALUES ($1, now())
+               ON CONFLICT (telegram_id) DO UPDATE SET last_digest_flush = now()""",
+            tg_id)
+    
+    return flushed
+
+
 async def scheduler_task(pool: object, bot: ExtBot) -> None:
     while True:
         record_scheduler_run()  # mark this cycle as started
@@ -101,7 +203,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
 
         try:
             rows = await pool.fetch(
-                "SELECT id, keyword, interval_minutes, initial_loaded, ads_cursor FROM keywords "
+                "SELECT id, keyword, interval_minutes, initial_loaded, ads_cursor, price_min, price_max, allowed_postcodes FROM keywords "
                 "WHERE is_active = true "
                 "AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)"
             )
@@ -135,6 +237,9 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
 
                     for ad_data in ads_raw:
                         fields = extract_ad_fields(ad_data)
+                        # Skip ads that don't match keyword filters
+                        if not _ad_passes_filters(fields, row):
+                            continue
                         wh_ad_id = fields["wh_ad_id"]
                         is_price_drop = False
                         old_price = None
@@ -163,7 +268,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                             if initial_loaded:
                                 notify_fields = {**fields, "keyword": keyword}
                                 for tg_id in telegram_ids:
-                                    msg_id_val = await notify_new_ad(bot, tg_id, notify_fields, ad_uuid=ad_uuid)
+                                    msg_id_val = await notify_new_ad(bot, tg_id, notify_fields, ad_uuid=ad_uuid, pool=pool)
                                     if msg_id_val:
                                         user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
                                         if user_row:
@@ -197,7 +302,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                         if is_price_drop:
                             notify_fields = {**fields, "keyword": keyword}
                             for tg_id in telegram_ids:
-                                msg_id_val = await notify_price_drop(bot, tg_id, notify_fields, ad_uuid=ad_uuid)
+                                msg_id_val = await notify_price_drop(bot, tg_id, notify_fields, ad_uuid=ad_uuid, pool=pool)
                                 if msg_id_val:
                                     user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
                                     if user_row:

worker/src/migrations/05-keyword-filters.sql

@@ -0,0 +1,9 @@
+-- Price and postcode filters for keywords
+
+ALTER TABLE keywords ADD COLUMN IF NOT EXISTS price_min int;
+ALTER TABLE keywords ADD COLUMN IF NOT EXISTS price_max int;
+ALTER TABLE keywords ADD COLUMN IF NOT EXISTS allowed_postcodes text[];
+
+COMMENT ON COLUMN keywords.price_min IS 'Minimum price filter in cents (e.g., 5000 = €50.00)';
+COMMENT ON COLUMN keywords.price_max IS 'Maximum price filter in cents (e.g., 5000 = €50.00)';
+COMMENT ON COLUMN keywords.allowed_postcodes IS 'Array of allowed postcode strings (e.g., ARRAY[''1010'', ''1020''])';

worker/src/migrations/06-user-settings.sql

@@ -0,0 +1,49 @@
+-- User settings and digest buffer tables
+
+CREATE TABLE IF NOT EXISTS user_settings (
+    telegram_id       text PRIMARY KEY,
+    mute_start        time,
+    mute_end          time,
+    digest_mode       boolean NOT NULL DEFAULT false,
+    digest_interval   int NOT NULL DEFAULT 60,
+    last_digest_flush timestamptz,
+    CONSTRAINT chk_mute_hours CHECK (
+        (mute_start IS NULL AND mute_end IS NULL)
+        OR (mute_start IS NOT NULL AND mute_end IS NOT NULL)
+    )
+);
+
+COMMENT ON TABLE user_settings IS
+    'Per-user preferences for mute hours and digest notifications';
+
+COMMENT ON COLUMN user_settings.mute_start IS
+    'UTC time at which notifications are silenced each day';
+
+COMMENT ON COLUMN user_settings.mute_end IS
+    'UTC time at which notifications are re-enabled each day';
+
+COMMENT ON COLUMN user_settings.digest_mode IS
+    'When true, notifications are batched into periodic digest messages';
+
+COMMENT ON COLUMN user_settings.digest_interval IS
+    'Interval in minutes between digest flushes (default 60)';
+
+COMMENT ON COLUMN user_settings.last_digest_flush IS
+    'Timestamp of the last digest sent to this user';
+
+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,
+    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
+    'Temporary holding area for notifications pending digest flush';

worker/src/notifier.py

@@ -1,5 +1,5 @@
 import logging
-from datetime import datetime
+from datetime import datetime, time, timezone
 from typing import Any
 
 import asyncpg
@@ -10,6 +10,46 @@ from telegram.ext import ExtBot
 logger = logging.getLogger(__name__)
 
 
+async def is_user_muted(pool: asyncpg.Pool, telegram_id: str) -> bool:
+    """Check if a user is currently in their mute window (UTC)."""
+    settings = await pool.fetchrow(
+        "SELECT mute_start, mute_end FROM user_settings WHERE telegram_id = $1",
+        telegram_id)
+    
+    if not settings or not settings["mute_start"]:
+        return False
+    
+    now_utc = datetime.now(tz=timezone.utc).time()
+    start_t = settings["mute_start"]
+    end_t = settings["mute_end"]
+    
+    # Handle cross-midnight windows (e.g., 22:00-07:00)
+    if start_t > end_t:
+        return now_utc >= start_t or now_utc < end_t
+    else:
+        return start_t <= now_utc < end_t
+
+
+async def get_user_settings(pool: asyncpg.Pool, telegram_id: str) -> dict | None:
+    """Get user settings row, or None if not set."""
+    return await pool.fetchrow(
+        "SELECT mute_start, mute_end, digest_mode, digest_interval FROM user_settings WHERE telegram_id = $1",
+        telegram_id)
+
+
+async def buffer_for_digest(pool: asyncpg.Pool, telegram_id: str, ad: dict, ad_uuid: str) -> None:
+    """Buffer a notification for digest mode instead of sending immediately."""
+    await pool.execute(
+        """INSERT INTO digest_buffer (telegram_id, ad_id, keyword, title, price, url)
+           VALUES ($1, $2, $3, $4, $5, $6)""",
+        telegram_id, ad_uuid,
+        ad.get("keyword", ""),
+        ad.get("title", "Untitled"),
+        int(round(ad["price"] * 100)) if ad.get("price") else None,
+        ad.get("url"))
+    logger.info("Buffered notification for digest: user=%s ad=%s", telegram_id, ad.get("wh_ad_id", ""))
+
+
 def _build_keyboard(ad: dict[str, Any]) -> InlineKeyboardMarkup | None:
     keyboard: list[list[InlineKeyboardButton]] = []
     if ad.get("url"):
@@ -76,7 +116,21 @@ async def notify_new_ad(
     telegram_id: int,
     ad: dict[str, Any],
     ad_uuid: str | None = None,
+    pool: asyncpg.Pool | None = None,
 ) -> int | None:
+    # Check mute hours and digest mode
+    if pool:
+        tg_id_str = str(telegram_id)
+        muted = await is_user_muted(pool, tg_id_str)
+        if muted:
+            return None  # Muted — discard (mute takes precedence over digest)
+        
+        # Not muted, check digest mode
+        settings = await get_user_settings(pool, tg_id_str)
+        if settings and settings["digest_mode"] and ad_uuid:
+            await buffer_for_digest(pool, tg_id_str, ad, ad_uuid)
+            return None  # Buffered — will be sent in next digest flush
+
     text = _format_text("🆕 New listing found!", ad)
     reply_markup = _build_keyboard(ad)
 
@@ -127,7 +181,21 @@ async def notify_price_drop(
     telegram_id: int,
     ad: dict[str, Any],
     ad_uuid: str | None = None,
+    pool: asyncpg.Pool | None = None,
 ) -> int | None:
+    # Check mute hours and digest mode
+    if pool:
+        tg_id_str = str(telegram_id)
+        muted = await is_user_muted(pool, tg_id_str)
+        if muted:
+            return None  # Muted — discard (mute takes precedence over digest)
+        
+        # Not muted, check digest mode
+        settings = await get_user_settings(pool, tg_id_str)
+        if settings and settings["digest_mode"] and ad_uuid:
+            await buffer_for_digest(pool, tg_id_str, ad, ad_uuid)
+            return None  # Buffered — will be sent in next digest flush
+
     text = _format_text("⚠️ Price drop!", ad)
     reply_markup = _build_keyboard(ad)