fix: sync scraper.py with server version (proxy support + ID-based fetch)

- Remove initial_loaded/baseline logic from scheduler
- Remove initial_loaded from keywords SELECT query
- Replace synchronous notification loops with asyncio.create_task
  (fire-and-forget) for both new ads and price drops
- Add safe_notify_new_ad/safe_notify_price_drop wrapper functions
- Scraper already simplified (single page, 30 ads, no cursor)

.env.example

@@ -1,14 +1,13 @@
 # Telegram Bot Token (from @BotFather)
 TELEGRAM_BOT_TOKEN=8653489932:AAHhyOD1jtimE7kg0zoVCUVd3l0YEz_YJgg
 
-# PostgreSQL Credentials
+# Direct Supabase Postgres connection (join supabase_default network, connect to db:5432)
+POSTGRES_HOST=db
+POSTGRES_PORT=5432
 POSTGRES_USER=postgres
-POSTGRES_PASSWORD=change-me-strong-password
+POSTGRES_PASSWORD=your-supabase-db-password
 POSTGRES_DB=postgres
 
-# PostgREST JWT Secret (random string for signing tokens)
-JWT_SECRET=your-super-secret-jwt-key-change-in-production
-
 # Worker Configuration
 DEFAULT_INTERVAL_MINUTES=60
 ADMIN_TELEGRAM_IDS=298181113  # Comma-separated Telegram user IDs with admin access

docker-compose.yml

@@ -1,103 +1,12 @@
-version: "3.9"
-
 services:
-  db:
-    image: postgres:15-alpine
-    restart: unless-stopped
-    environment:
-      POSTGRES_USER: ${POSTGRES_USER:-postgres}
-      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
-      POSTGRES_DB: ${POSTGRES_DB:-postgres}
-      POSTGRES_HOST_AUTH_METHOD: trust
-    volumes:
-      - ./data/db:/var/lib/postgresql/data
-      - ./supabase/pg_hba.conf:/etc/postgresql/pg_hba.conf:ro
-      - ./supabase/migrations/00-run-init.sh:/docker-entrypoint-initdb.d/00-run-init.sh:ro
-      - ./supabase/migrations/01-schema.sql:/docker-entrypoint-initdb.d/01-schema.sql:ro
-      - ./supabase/migrations/post-boot.sql:/docker-entrypoint-initdb.d/post-boot.sql:ro
-    command: >
-      postgres
-      -c hba_file=/etc/postgresql/pg_hba.conf
-      -c wal_level=logical
-      -c max_wal_senders=0
-      -c max_replication_slots=0
-      -c idle_in_transaction_session_timeout=1min
-    ports:
-      - "55632:5432"
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-postgres} -d ${POSTGRES_DB:-postgres}"]
-      interval: 5s
-      timeout: 5s
-      retries: 10
-
-  rest:
-    image: postgrest/postgrest:v12.2.0
-    restart: unless-stopped
-    environment:
-      PGRST_DB_URI: postgres://authenticator@db:5432/${POSTGRES_DB:-postgres}
-      PGRST_DB_SCHEMAS: public
-      PGRST_DB_ANON_ROLE: authenticator
-      PGRST_JWT_SECRET: ${JWT_SECRET:-your-super-secret-jwt-key-change-in-production}
-      PGRST_DB_EXTRA_SEARCH_PATH: public
-    depends_on:
-      db:
-        condition: service_healthy
-
-  kong:
-    image: kong:2.8.1
-    restart: unless-stopped
-    environment:
-      KONG_DATABASE: "off"
-      KONG_DECLARATIVE_CONFIG: /etc/kong/kong.yml
-      KONG_PLUGINS: request-transformer,cors
-      KONG_PROXY_LISTEN: "0.0.0.0:8000"
-      KONG_NGINX_WORKER_PROCESSES: 1
-    volumes:
-      - ./supabase/kong.yml:/etc/kong/kong.yml:ro
-    ports:
-      - "55621:8000"
-    depends_on:
-      rest:
-        condition: service_started
-
-  studio:
-    image: supabase/studio
-    restart: unless-stopped
-    environment:
-      STUDIO_PG_META_URL: http://meta:8080
-      DEFAULT_ORGANIZATION_NAME: Local
-      DEFAULT_PROJECT_NAME: willhaben-tracker
-    ports:
-      - "55630:3000"
-    depends_on:
-      meta:
-        condition: service_started
-
-  meta:
-    image: supabase/postgres-meta:v0.84.2
-    restart: unless-stopped
-    environment:
-      PG_META_PORT: 8080
-      PG_META_DB_HOST: db
-      PG_META_DB_PORT: 5432
-      PG_META_DB_NAME: ${POSTGRES_DB:-postgres}
-      PG_META_DB_USER: ${POSTGRES_USER:-postgres}
-      PG_META_DB_PASSWORD: ${POSTGRES_PASSWORD}
-    depends_on:
-      db:
-        condition: service_healthy
-
   worker:
     build: ./worker
     restart: unless-stopped
-    environment:
+    env_file:
-      TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN}
+      - .env
-      POSTGRES_HOST: db
+    networks:
-      POSTGRES_PORT: "5432"
+      - supabase_default
-      POSTGRES_USER: ${POSTGRES_USER:-postgres}
+
-      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+networks:
-      POSTGRES_DB: ${POSTGRES_DB:-postgres}
+  supabase_default:
-      DEFAULT_INTERVAL_MINUTES: ${DEFAULT_INTERVAL_MINUTES:-60}
+    external: true
-    depends_on:
-      db:
-        condition: service_healthy

supabase/migrations/supabase-migration.sql

@@ -0,0 +1,121 @@
+CREATE SCHEMA IF NOT EXISTS willhaben_tracker;
+SET search_path TO willhaben_tracker;
+
+-- -----------------------------------------------------------
+-- users (whitelisted Telegram users)
+-- -----------------------------------------------------------
+CREATE TABLE IF NOT EXISTS users (
+    id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    telegram_id bigint UNIQUE NOT NULL,
+    username    text,
+    first_name  text,
+    is_admin    boolean NOT NULL DEFAULT false,
+    is_active   boolean NOT NULL DEFAULT true,
+    created_at  timestamptz NOT NULL DEFAULT now()
+);
+
+-- -----------------------------------------------------------
+-- keywords (global search keywords — deduplicated across users)
+-- -----------------------------------------------------------
+CREATE TABLE IF NOT EXISTS keywords (
+    id               uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    keyword          text NOT NULL,
+    interval_minutes int NOT NULL DEFAULT 60,
+    is_active        boolean NOT NULL DEFAULT true,
+    initial_loaded   boolean NOT NULL DEFAULT false,
+    last_scraped_at  timestamptz,
+    created_at       timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_keywords_unique_lower
+    ON keywords(LOWER(keyword));
+
+-- -----------------------------------------------------------
+-- keyword_subscriptions (many-to-many: user <-> keyword)
+-- -----------------------------------------------------------
+CREATE TABLE IF NOT EXISTS keyword_subscriptions (
+    keyword_id uuid REFERENCES keywords(id) ON DELETE CASCADE NOT NULL,
+    user_id    uuid REFERENCES users(id) ON DELETE CASCADE NOT NULL,
+    PRIMARY KEY (keyword_id, user_id),
+    created_at timestamptz NOT NULL DEFAULT now()
+);
+
+-- -----------------------------------------------------------
+-- ads (raw ad snapshots, globally deduplicated by wh_ad_id)
+-- -----------------------------------------------------------
+CREATE TABLE IF NOT EXISTS ads (
+    id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    wh_ad_id      text UNIQUE NOT NULL,
+    raw_json      jsonb NOT NULL,
+    title         text NOT NULL,
+    price         numeric,
+    location      text,
+    url           text,
+    published_at  timestamptz,
+    first_seen_at timestamptz NOT NULL DEFAULT now(),
+    main_image_url text,
+    postcode      text,
+    modified_at   timestamptz
+);
+
+-- -----------------------------------------------------------
+-- price_history (track price changes per ad)
+-- -----------------------------------------------------------
+CREATE TABLE IF NOT EXISTS price_history (
+    id         uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    ad_id      uuid REFERENCES ads(id) ON DELETE CASCADE NOT NULL,
+    old_price  numeric NOT NULL,
+    new_price  numeric NOT NULL,
+    changed_at timestamptz NOT NULL DEFAULT now(),
+    UNIQUE (ad_id, old_price, new_price)
+);
+
+-- -----------------------------------------------------------
+-- notifications (audit log of sent Telegram messages)
+-- -----------------------------------------------------------
+CREATE TABLE IF NOT EXISTS notifications (
+    id         uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    user_id    uuid REFERENCES users(id) ON DELETE CASCADE NOT NULL,
+    ad_id      uuid REFERENCES ads(id) ON DELETE SET NULL,
+    message_id int,
+    sent_at    timestamptz NOT NULL DEFAULT now()
+);
+
+-- -----------------------------------------------------------
+-- scrape_logs (worker health / debugging)
+-- -----------------------------------------------------------
+CREATE TABLE IF NOT EXISTS scrape_logs (
+    id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+    keyword_id    uuid REFERENCES keywords(id) ON DELETE CASCADE NOT NULL,
+    status        text NOT NULL CHECK (status IN ('success', 'error', 'rate_limited')),
+    ads_found     int NOT NULL DEFAULT 0,
+    new_ads       int NOT NULL DEFAULT 0,
+    error_message text,
+    scraped_at    timestamptz NOT NULL DEFAULT now()
+);
+
+-- ============================================================
+-- Indexes
+-- ============================================================
+
+CREATE INDEX IF NOT EXISTS idx_keywords_active_scraped
+    ON keywords(is_active, last_scraped_at) WHERE is_active = true;
+
+CREATE INDEX IF NOT EXISTS idx_keyword_subscriptions_user_id
+    ON keyword_subscriptions(user_id);
+
+CREATE INDEX IF NOT EXISTS idx_price_history_ad_id
+    ON price_history(ad_id);
+
+CREATE INDEX IF NOT EXISTS idx_notifications_user_sent
+    ON notifications(user_id, sent_at DESC);
+
+CREATE INDEX IF NOT EXISTS idx_scrape_logs_keyword_at
+    ON scrape_logs(keyword_id, scraped_at DESC);
+
+-- -----------------------------------------------------------
+-- Seed: initial admin user
+-- -----------------------------------------------------------
+INSERT INTO users (telegram_id, username, first_name, is_admin, is_active)
+VALUES (298181113, NULL, 'Admin', true, true)
+ON CONFLICT (telegram_id) DO NOTHING;

worker/src/db.py

@@ -7,6 +7,10 @@ logger = logging.getLogger(__name__)
 _pool: asyncpg.Pool | None = None
 
 
+async def _init_connection(conn: asyncpg.Connection) -> None:
+    await conn.execute("SET search_path TO willhaben_tracker")
+
+
 async def get_pool() -> asyncpg.Pool:
     global _pool
     if _pool is None:
@@ -18,6 +22,7 @@ async def get_pool() -> asyncpg.Pool:
             database=os.getenv("POSTGRES_DB", "postgres"),
             min_size=2,
             max_size=10,
+            init=_init_connection
         )
         logger.info("Database pool initialized")
     return _pool

worker/src/main.py

@@ -19,11 +19,33 @@ logger = logging.getLogger(__name__)
 load_dotenv()
 
 
+async def safe_notify_new_ad(bot, pool, tg_id, notify_fields, ad_uuid):
+    try:
+        msg_id_val = await notify_new_ad(bot, tg_id, notify_fields)
+        if msg_id_val:
+            user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
+            if user_row:
+                await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
+    except Exception:
+        logger.exception("Failed to send new ad notification for %s", ad_uuid)
+
+
+async def safe_notify_price_drop(bot, pool, tg_id, notify_fields, ad_uuid):
+    try:
+        msg_id_val = await notify_price_drop(bot, tg_id, notify_fields)
+        if msg_id_val:
+            user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
+            if user_row:
+                await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
+    except Exception:
+        logger.exception("Failed to send price drop notification for %s", ad_uuid)
+
+
 async def scheduler_task(pool: object, bot: ExtBot) -> None:
     while True:
         try:
             rows = await pool.fetch(
-                "SELECT id, keyword, interval_minutes, initial_loaded FROM keywords "
+                "SELECT id, keyword, interval_minutes FROM keywords "
                 "WHERE is_active = true "
                 "AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)"
             )
@@ -31,7 +53,6 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
             for row in rows:
                 kw_id = str(row["id"])
                 keyword = row["keyword"]
-                initial_loaded = row["initial_loaded"]
 
                 subs = await pool.fetch(
                     "SELECT telegram_id FROM users u JOIN keyword_subscriptions ks ON u.id = ks.user_id "
@@ -50,15 +71,9 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                     ads_raw, total_hits = await fetch_ads(keyword)
                     new_count = 0
 
-                    if not initial_loaded and len(ads_raw) > 0:
-                        logger.info("Initial baseline load for '%s' — indexing %d ads, no notifications", keyword, len(ads_raw))
-
                     for ad_data in ads_raw:
                         fields = extract_ad_fields(ad_data)
                         wh_ad_id = fields["wh_ad_id"]
-                        is_price_drop = False
-                        old_price = None
-                        new_price = None
 
                         existing = await pool.fetchrow(
                             "SELECT id, price FROM ads WHERE wh_ad_id = $1",
@@ -66,6 +81,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                         )
 
                         if not existing:
+                            # NEW AD - insert and fire-and-forget notify
                             ad_row = await pool.fetchrow(
                                 "INSERT INTO ads (wh_ad_id, raw_json, title, price, location, url, published_at, main_image_url, postcode, modified_at) "
                                 "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING id",
@@ -75,20 +91,12 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                             )
                             ad_uuid = str(ad_row["id"])
 
-                            # Only notify for genuinely new ads after baseline load is done
+                            notify_fields = {**fields, "keyword": keyword}
-                            if initial_loaded:
+                            for tg_id in telegram_ids:
-                                notify_fields = {**fields, "keyword": keyword}
+                                asyncio.create_task(safe_notify_new_ad(bot, pool, tg_id, notify_fields, ad_uuid))
-                                for tg_id in telegram_ids:
+                            new_count += 1
-                                    msg_id_val = await notify_new_ad(bot, tg_id, notify_fields)
-                                    if msg_id_val:
-                                        user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
-                                        if user_row:
-                                            try:
-                                                await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
-                                            except Exception:
-                                                logger.exception("Failed to log new ad notification")
-                                new_count += 1
                         else:
+                            # EXISTING AD - check price drop
                             ad_uuid = str(existing["id"])
                             old_price = existing["price"]
                             new_price = fields["price"]
@@ -102,29 +110,17 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                                     "INSERT INTO price_history (ad_id, old_price, new_price) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
                                     ad_uuid, old_price, new_price,
                                 )
-                                is_price_drop = True
+                                notify_fields = {**fields, "keyword": keyword}
+                                for tg_id in telegram_ids:
+                                    asyncio.create_task(safe_notify_price_drop(bot, pool, tg_id, notify_fields, ad_uuid))
                             else:
+                                # Update metadata if missing
                                 if fields.get("main_image_url") or fields.get("postcode"):
                                     await pool.execute(
                                         "UPDATE ads SET main_image_url = COALESCE($1, main_image_url), postcode = COALESCE($2, postcode) WHERE id = $3 AND (main_image_url IS NULL OR postcode IS NULL)",
                                         fields.get("main_image_url"), fields.get("postcode"), ad_uuid,
                                     )
 
-                        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)
-                                if msg_id_val:
-                                    user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
-                                    if user_row:
-                                        try:
-                                            await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
-                                        except Exception:
-                                            logger.exception("Failed to log price drop notification")
-
-                    if not initial_loaded:
-                        await pool.execute("UPDATE keywords SET initial_loaded = true WHERE id = $1", kw_id)
-
                     await pool.execute("UPDATE keywords SET last_scraped_at = now() WHERE id = $1", kw_id)
 
                     await pool.execute(
@@ -203,4 +199,4 @@ async def main() -> None:
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())

worker/src/scraper.py

@@ -1,13 +1,195 @@
 import asyncio
 import logging
-from datetime import datetime, timezone
+import os
+from datetime import datetime
 from typing import Any
-from urllib.parse import quote_plus
 
 import httpx
 
+from db import get_pool
+from settings import get_effective_proxy_url, get_proxy_url_from_env, proxy_available
+
 logger = logging.getLogger(__name__)
 
+_client: httpx.AsyncClient | None = None
+_client_proxy_url: str | None = None
+_proxy_ip_logged = False
+_system_public_ip: str | None = None
+_proxy_public_ip: str | None = None
+_last_used_public_ip: str | None = None
+_proxy_enabled_effective: bool | None = None
+
+
+def _get_proxy_url() -> str | None:
+    return get_proxy_url_from_env()
+
+
+async def _get_effective_proxy_url() -> str | None:
+    global _proxy_enabled_effective
+
+    pool = await get_pool()
+    proxy_url = await get_effective_proxy_url(pool)
+    _proxy_enabled_effective = proxy_url is not None
+    return proxy_url
+
+
+def _redact_proxy_url(proxy_url: str) -> str:
+    try:
+        parsed = httpx.URL(proxy_url)
+        host = parsed.host or "unknown-host"
+        port = parsed.port or 80
+        user = parsed.username or "unknown-user"
+        return f"{host}:{port} (user={user}, credentials=set)"
+    except Exception:
+        return "<unparseable proxy>"
+
+
+async def _fetch_public_ip(client: httpx.AsyncClient) -> str | None:
+    try:
+        resp = await client.get("https://api.ipify.org?format=json")
+        resp.raise_for_status()
+        data = resp.json()
+        ip = data.get("ip")
+        return str(ip) if ip else None
+    except Exception as exc:
+        logger.warning("Could not resolve public IP: %s", exc)
+        return None
+
+
+async def _log_proxy_ip_comparison(proxy_url: str) -> None:
+    global _proxy_ip_logged, _system_public_ip, _proxy_public_ip, _last_used_public_ip
+
+    if _proxy_ip_logged:
+        return
+
+    _proxy_ip_logged = True
+
+    direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False)
+    proxy_client = httpx.AsyncClient(timeout=10.0, trust_env=False, proxy=proxy_url)
+
+    try:
+        direct_ip = await _fetch_public_ip(direct_client)
+        proxy_ip = await _fetch_public_ip(proxy_client)
+        _system_public_ip = direct_ip
+        _proxy_public_ip = proxy_ip
+        _last_used_public_ip = proxy_ip or direct_ip
+        logger.info(
+            "HTTPS proxy enabled: %s | public_ip_without_proxy=%s | public_ip_with_proxy=%s",
+            _redact_proxy_url(proxy_url),
+            direct_ip or "unknown",
+            proxy_ip or "unknown",
+        )
+    finally:
+        await direct_client.aclose()
+        await proxy_client.aclose()
+
+
+async def get_client() -> httpx.AsyncClient:
+    """Return a shared AsyncClient with keepalive connection pool."""
+    global _client, _client_proxy_url, _system_public_ip, _last_used_public_ip
+
+    proxy_url = await _get_effective_proxy_url()
+
+    if _client is not None and not _client.is_closed and _client_proxy_url != proxy_url:
+        await _client.aclose()
+        logger.info(
+            "Recreated httpx client because proxy changed: %s -> %s",
+            "enabled" if _client_proxy_url else "disabled",
+            "enabled" if proxy_url else "disabled",
+        )
+        _client = None
+
+    if _client is None or _client.is_closed:
+        max_conns = int(os.getenv("HTTP_MAX_CONNECTIONS", "10"))
+        max_keepalive = int(os.getenv("HTTP_KEEPALIVE_CONNECTIONS", "5"))
+
+        if proxy_url:
+            await _log_proxy_ip_comparison(proxy_url)
+        elif _system_public_ip is None:
+            direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False)
+            try:
+                _system_public_ip = await _fetch_public_ip(direct_client)
+                _last_used_public_ip = _system_public_ip
+            finally:
+                await direct_client.aclose()
+
+        _client = httpx.AsyncClient(
+            timeout=float(os.getenv("HTTP_TIMEOUT_S", "30.0")),
+            limits=httpx.Limits(
+                max_connections=max_conns,
+                max_keepalive_connections=max_keepalive,
+                keepalive_expiry=60,
+            ),
+            proxy=proxy_url,
+            trust_env=False,
+        )
+        _client_proxy_url = proxy_url
+        logger.info(
+            "Created httpx client: max_conns=%d, keepalive=%d, proxy=%s",
+            max_conns,
+            max_keepalive,
+            "enabled" if proxy_url else "disabled",
+        )
+
+    return _client
+
+
+async def refresh_network_status() -> dict[str, str | bool | None]:
+    """Ensure network status has best-effort values even before first scrape."""
+    global _system_public_ip, _proxy_public_ip, _last_used_public_ip, _proxy_enabled_effective
+
+    proxy_url = await _get_effective_proxy_url()
+    _proxy_enabled_effective = proxy_url is not None
+
+    if _system_public_ip is None:
+        direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False)
+        try:
+            _system_public_ip = await _fetch_public_ip(direct_client)
+        finally:
+            await direct_client.aclose()
+
+    if proxy_url and _proxy_public_ip is None:
+        proxy_client = httpx.AsyncClient(timeout=10.0, trust_env=False, proxy=proxy_url)
+        try:
+            _proxy_public_ip = await _fetch_public_ip(proxy_client)
+        finally:
+            await proxy_client.aclose()
+
+    if proxy_url:
+        _last_used_public_ip = _proxy_public_ip or _system_public_ip
+    else:
+        _last_used_public_ip = _system_public_ip
+
+    return get_network_status()
+
+
+def get_network_status() -> dict[str, str | bool | None]:
+    """Return best-effort network status for UI display."""
+    available = proxy_available()
+    proxy_enabled = bool(_proxy_enabled_effective) if _proxy_enabled_effective is not None else available
+    last_used = _last_used_public_ip
+    if last_used is None:
+        last_used = _proxy_public_ip if proxy_enabled else _system_public_ip
+
+    return {
+        "proxy_available": available,
+        "proxy_enabled": proxy_enabled,
+        "system_public_ip": _system_public_ip,
+        "proxy_public_ip": _proxy_public_ip,
+        "last_used_public_ip": last_used,
+    }
+
+
+async def close_client() -> None:
+    """Close the shared AsyncClient. Call during shutdown."""
+    global _client, _client_proxy_url
+    if _client and not _client.is_closed:
+        await _client.aclose()
+        logger.info("Closed httpx client")
+    _client = None
+    _client_proxy_url = None
+
+
 _API_URL = (
     "https://www.willhaben.at/webapi/ad-search/search/atz/seo/"
     "kaufen-und-verkaufen/marktplatz"
@@ -21,25 +203,27 @@ _HEADERS = {
 
 
 async def fetch_ads(keyword: str) -> tuple[list[dict[str, Any]], int]:
+    """Fetch the latest 30 ads for a keyword (single page, newest first)."""
     params = {
         "keyword": keyword,
         "rows": 30,
         "sort": 1,
     }
 
-    async with httpx.AsyncClient(timeout=30.0) as client:
+    client = await get_client()
-        for attempt in range(1, 4):
+
-            try:
+    for attempt in range(1, 4):
-                resp = await client.get(_API_URL, headers=_HEADERS, params=params)
+        try:
-                resp.raise_for_status()
+            resp = await client.get(_API_URL, headers=_HEADERS, params=params)
-                data = resp.json()
+            resp.raise_for_status()
-                break
+            data = resp.json()
-            except Exception as exc:
+            break
-                logger.warning("fetch_ads attempt %d failed: %s", attempt, exc)
+        except Exception as exc:
-                if attempt < 3:
+            logger.warning("fetch_ads attempt %d failed for '%s': %s", attempt, keyword, exc)
-                    await asyncio.sleep(2 ** attempt)
+            if attempt < 3:
-                    continue
+                await asyncio.sleep(2 ** attempt)
-                raise
+                continue
+            raise
 
     ads_raw = data.get("advertSummaryList", {}).get("advertSummary", [])
     total_hits = int(data.get("rowsFound", 0))
@@ -111,4 +295,4 @@ def extract_ad_fields(ad_dict: dict[str, Any]) -> dict[str, Any]:
         "main_image_url": main_image_url,
         "postcode": postcode,
         "modified_at": modified_at,
-    }
+    }