From f540cbe7ef4b82349b5bb1ccd9cccf9b775fafdd Mon Sep 17 00:00:00 2001 From: hermes Date: Sat, 4 Jul 2026 11:30:45 -0400 Subject: [PATCH] =?UTF-8?q?feat(phase-0):=20stability=20hardening=20?= =?UTF-8?q?=E2=80=94=20search=20path=20fix,=20auto-migrations,=20graceful?= =?UTF-8?q?=20shutdown,=20health=20endpoint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove SET search_path from db.py and migration SQL (Supabase uses public schema) - Add migrate.py with tracking table for forward-only SQL migrations - Add entrypoint.sh: waits for DB, runs migrations, then starts app - Copy 01-schema.sql + zz-seed.sql to worker/src/migrations/ - Add health.py: /health endpoint (200/503) with DB connectivity + scheduler staleness checks - /stats endpoint with keyword/ad/notification counts - Rewrite main.py shutdown sequence: signal handler, 5s grace for scheduler, ordered cleanup - Update Dockerfile: HEALTHCHECK directive, entrypoint, COPY migrations - Update docker-compose.yml: stop_grace_period=15s, healthcheck config, env vars - Add aiohttp>=3.9 to requirements.txt for health server --- docker-compose.yml | 15 ++ docs/phase-0/Phase.md | 59 ++++++ docs/phase-0/task-auto-migration.md | 161 +++++++++++++++ docs/phase-0/task-graceful-shutdown.md | 134 +++++++++++++ docs/phase-0/task-healthcheck.md | 220 +++++++++++++++++++++ docs/phase-0/task-search-path-fix.md | 101 ++++++++++ supabase/migrations/supabase-migration.sql | 4 +- worker/Dockerfile | 14 ++ worker/requirements.txt | 1 + worker/src/db.py | 4 - worker/src/entrypoint.sh | 40 ++++ worker/src/health.py | 122 ++++++++++++ worker/src/main.py | 39 +++- worker/src/migrate.py | 97 +++++++++ worker/src/migrations/01-schema.sql | 123 ++++++++++++ worker/src/migrations/zz-seed.sql | 8 + 16 files changed, 1133 insertions(+), 9 deletions(-) create mode 100644 docs/phase-0/Phase.md create mode 100644 docs/phase-0/task-auto-migration.md create mode 100644 docs/phase-0/task-graceful-shutdown.md create mode 100644 docs/phase-0/task-healthcheck.md create mode 100644 docs/phase-0/task-search-path-fix.md create mode 100644 worker/src/entrypoint.sh create mode 100644 worker/src/health.py create mode 100644 worker/src/migrate.py create mode 100644 worker/src/migrations/01-schema.sql create mode 100644 worker/src/migrations/zz-seed.sql diff --git a/docker-compose.yml b/docker-compose.yml index fd06548..6512440 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,8 +4,23 @@ services: restart: unless-stopped env_file: - .env + environment: + # Health check port (exposed to host for docker healthcheck) + - HEALTH_PORT=8765 + # Scheduler staleness threshold in seconds + - HEALTHCHECK_SCHEDULER_STALE_S=300 networks: - supabase_default + # Graceful shutdown timeout — Docker sends SIGTERM, container has this long to clean up. + stop_grace_period: 15s + # Docker healthcheck (uses the in-container aiohttp server) + healthcheck: + test: ["CMD", "python", "-c", + "import httpx,os; r=httpx.get(f'http://localhost:{os.getenv(\"HEALTH_PORT\",\"8765\")}/health'); r.raise_for_status()"] + interval: 15s + timeout: 5s + retries: 3 + start_period: 10s networks: supabase_default: diff --git a/docs/phase-0/Phase.md b/docs/phase-0/Phase.md new file mode 100644 index 0000000..8729bd1 --- /dev/null +++ b/docs/phase-0/Phase.md @@ -0,0 +1,59 @@ +# Phase 0 — Critical Stability Fixes + +## Scope + +This phase addresses **blocking and high-risk issues** in the current `feat/supabase-migration` branch that must be resolved before any feature work. The goal is a stable, self-healing deployment where the worker container: + +- Starts correctly against an empty or partially-migrated database +- Recovers from crashes via Docker healthcheck +- Shuts down gracefully without data corruption +- Has consistent schema configuration across all DB access paths + +## Architecture + +``` +┌─────────────────────────────────────────────────────┐ +│ worker container (python:3.12-slim) │ +│ │ +│ ┌───────────┐ ┌───────────┐ │ +│ │ entrypoint│──►│ migrate.py│──► apply pending │ +│ │ .sh │ │ │ SQL migrations │ +│ └─────┬─────┘ └───────────┘ │ +│ │ │ +│ ▼ │ +│ ┌───────────┐ │ +│ │ main.py │ (fixed shutdown order) │ +│ │ │ │ +│ ├───────────┤ │ +│ │ scheduler │◄► db.py │ +│ │ bot loop │ (consistent search_path) │ +│ └───────────┘ │ +│ │ │ +│ ▼ │ +│ ┌──────────────┐ │ +│ │ /health endpoint│ │ +│ │ HTTP server │ (exposed for docker healthcheck)│ +│ └──────────────┘ │ +└─────────────────────────────────────────────────────┘ + +docker-compose.yml: + healthcheck: curl -f http://localhost:8765/health || exit 1 + interval=30s timeout=5s retries=3 start_period=10s +``` + +## Tasks + +| Task | File | Description | +|------|------|-------------| +| Auto-migration on startup | [task-auto-migration.md](./task-auto-migration.md) | Entrypoint script + migration runner that applies pending SQL against the database before starting the worker process. Idempotent, schema-version tracked. | +| Graceful shutdown fix | [task-graceful-shutdown.md](./task-graceful-shutdown.md) | Reorder cleanup in `main.py` to stop polling → cancel scheduler → close DB pool → call Application.shutdown(), resolving the "Application is still running!" RuntimeError. | +| Healthcheck endpoint + docker-compose config | [task-healthcheck.md](./task-healthcheck.md) | Expose a lightweight HTTP `/health` endpoint on port 8765 inside the worker container; add `healthcheck` directive to docker-compose.yml so Docker restarts unhealthy containers. | +| search_path consistency | [task-search-path-fix.md](./task-search-path-fix.md) | Resolve mismatch between `_init_connection()` setting `search_path TO willhaben_tracker` and tables living in `public`. Pick one path (recommend: remove custom schema, keep everything in public for Supabase compatibility). | + +## General Acceptance Criteria + +- [ ] Container starts from scratch against an empty database → all tables created automatically, no manual intervention +- [ ] Sending SIGTERM to the worker results in clean shutdown within 10 seconds with no errors in logs +- [ ] Docker reports container as `healthy` within 60 seconds of start +- [ ] All DB queries work regardless of explicit schema qualification (no "relation does not exist" errors) +- [ ] Rolling back and re-applying migrations is idempotent (safe to run multiple times) diff --git a/docs/phase-0/task-auto-migration.md b/docs/phase-0/task-auto-migration.md new file mode 100644 index 0000000..e8d3c1a --- /dev/null +++ b/docs/phase-0/task-auto-migration.md @@ -0,0 +1,161 @@ +# Task: Auto-migration on container startup + +## Description + +The worker currently fails to start when the database schema is missing or outdated (as happened with the `UndefinedTableError` incident). There is no automatic migration mechanism — migrations must be applied manually via `docker cp` + Python scripts. + +This task introduces an **automatic, idempotent migration runner** that executes on every container start *before* the worker process begins. It tracks which migrations have been applied in a `migrations` table and applies only the missing ones. + +## Architecture + +``` +Container lifecycle: + + ┌──────────────┐ ┌──────────────────┐ ┌──────────────┐ + │ entrypoint.sh│────►│ python migrate.py│────►│ python │ + │ │ │ │ │ main.py │ + │ - copy SQLs │ │ 1. Create │ │ (normal app) │ + │ - call Python│ │ migrations tbl │ │ │ + └──────────────┘ │ 2. Apply pending │ └──────────────┘ + │ 3. Seed data │ + └──────────────────┘ + +migrations table: + CREATE TABLE IF NOT EXISTS willhaben_migrations ( + name text PRIMARY KEY, -- e.g. "01-schema.sql" + applied_at timestamptz DEFAULT now() + ); + +Migration directory layout: + worker/src/migrations/ + └── 01-schema.sql (the current supabase-migration.sql content) +``` + +### Key design decisions + +- **File-based migrations** — each `.sql` file in `worker/src/migrations/` is a migration. Name = version key. +- **No rollback support** — forward-only, consistent with the simplicity of the project. Rollback means restore from DB backup. +- **Run on every start** — idempotent by checking `willhaben_migrations`. Fast (no-op if all applied). +- **Seed data** included in a dedicated `zz-seed.sql` migration that runs last. + +## Implementation Details + +### 1. Create `worker/src/migrate.py` + +```python +#!/usr/bin/env python3 +"""Run pending SQL migrations against the configured PostgreSQL database.""" + +import os +import glob +import logging +import asyncpg + +logger = logging.getLogger(__name__) + +MIGRATIONS_DIR = os.path.join(os.path.dirname(__file__), "migrations") + +async def run_migrations(): + dsn_parts = { + "host": os.getenv("POSTGRES_HOST", "db"), + "port": int(os.getenv("POSTGRES_PORT", "5432")), + "user": os.getenv("POSTGRES_USER", "postgres"), + "password": os.getenv("POSTGRES_PASSWORD"), + "database": os.getenv("POSTGRES_DB", "postgres"), + } + + conn = await asyncpg.connect(**dsn_parts) + + try: + # Create tracking table (always in public schema for simplicity) + await conn.execute(""" + CREATE TABLE IF NOT EXISTS willhaben_migrations ( + name text PRIMARY KEY, + applied_at timestamptz DEFAULT now() + ); + """) + + # Find and sort migration files + sql_files = sorted(glob.glob(os.path.join(MIGRATIONS_DIR, "*.sql"))) + + for fpath in sql_files: + name = os.path.basename(fpath) + + # Check if already applied + exists = await conn.fetchval( + "SELECT 1 FROM willhaben_migrations WHERE name = $1", name + ) + + if exists: + logger.info("Skipping already-applied migration: %s", name) + continue + + logger.info("Applying migration: %s", name) + + with open(fpath) as f: + sql = f.read() + + # Execute within a transaction (DDL auto-commits in some cases) + await conn.execute(sql) + + # Record as applied + await conn.execute( + "INSERT INTO willhaben_migrations (name) VALUES ($1)", name + ) + logger.info("Applied migration: %s", name) + + finally: + await conn.close() + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s") + import asyncio + asyncio.run(run_migrations()) +``` + +### 2. Copy SQL migrations into `worker/src/migrations/` + +- Copy the content of `supabase/migrations/supabase-migration.sql` to `worker/src/migrations/01-schema.sql` +- **Remove** the `SET search_path TO willhaben_tracker;` line (Phase 0 task: search_path fix) +- Create `worker/src/migrations/zz-seed.sql` with just the admin user seed + +### 3. Update `worker/Dockerfile` to copy migration files + +```dockerfile +FROM python:3.12-slim +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY src/ . +ENTRYPOINT ["/app/entrypoint.sh"] +CMD ["python", "main.py"] +``` + +### 4. Create `worker/src/entrypoint.sh` + +```bash +#!/bin/bash +set -e + +echo "[entrypoint] Running migrations..." +python /app/migrate.py +echo "[entrypoint] Migrations complete." + +# Execute the CMD (main.py) +exec "$@" +``` + +Make executable: `chmod +x worker/src/entrypoint.sh` + +### 5. Update `.env.example` documentation to reflect auto-migration + +Remove manual migration instructions; add note that migrations run automatically on container start. + +## Acceptance Criteria + +- [ ] Starting a fresh container against an empty Supabase database creates all required tables and seeds the admin user — zero manual steps +- [ ] Restarting the same container does NOT re-apply already-applied migrations (logged as "Skipping") +- [ ] Adding a new `02-something.sql` to `worker/src/migrations/` and restarting the container applies only that file +- [ ] The `willhaben_migrations` table accurately tracks all applied migrations with timestamps +- [ ] Migration failures log an error and exit with non-zero code (container restarts via Docker policy) +- [ ] No changes to the existing `main.py` scheduler/bot logic are required diff --git a/docs/phase-0/task-graceful-shutdown.md b/docs/phase-0/task-graceful-shutdown.md new file mode 100644 index 0000000..4ee749e --- /dev/null +++ b/docs/phase-0/task-graceful-shutdown.md @@ -0,0 +1,134 @@ +# Task: Graceful shutdown fix + +## Description + +The current `main.py` has a broken shutdown sequence. When the application receives a termination signal, calling `app.shutdown()` while the updater/scheduler are still active raises: + +``` +RuntimeError: This Application is still running! +``` + +This means: +- DB connections may not be properly returned to the pool +- In-flight Telegram messages may be lost +- The container does not exit cleanly on SIGTERM, causing Kubernetes/Docker health issues during restarts + +This task reorders the shutdown sequence and adds proper signal handling so the worker exits gracefully within 10 seconds. + +## Architecture + +``` +Current (broken): + SIGTERM → cleanup() + ├─ app.shutdown() ← FAILS: RuntimeError "still running" + └─ close_pool() ← never reached or runs in bad state + +Target (fixed): + SIGTERM → cleanup() + ├─ scheduler_task.cancel() ← stop scraping loop + ├─ await asyncio.gather(scheduler_task) ← let it finish current iteration + ├─ app.updater.stop_polling() ← stop Telegram long-poll + ├─ app.shutdown() ← safe now, nothing running + └─ close_pool() ← return all connections to pool +``` + +### Key design decisions + +- **Drain period**: After signal received, allow up to 5 seconds for the current scrape cycle and pending notifications to complete. +- **Force kill fallback**: If cleanup does not finish in 10 seconds total, force-exit via `os._exit(0)`. +- **asyncpg pool close** uses timeout (default 30s) — acceptable since we already cancelled tasks. + +## Implementation Details + +### Current `main.py` shutdown code (problematic): + +```python +# Simplified current pattern +try: + # ... app.run() or equivalent +except KeyboardInterrupt: + app.shutdown() # ← RuntimeError here + loop.run_until_complete(close_pool()) +``` + +### Target implementation: + +Replace the signal handling block in `main.py` with: + +```python +import signal +import sys + +_scheduler_task: asyncio.Task | None = None + +async def cleanup(app: Application) -> None: + """Gracefully shut down all components.""" + logger.info("Shutting down...") + + # 1. Cancel scheduler task (with grace period) + if _scheduler_task and not _scheduler_task.done(): + _scheduler_task.cancel() + try: + await asyncio.wait_for(_scheduler_task, timeout=5.0) + except (asyncio.CancelledError, asyncio.TimeoutError): + pass + + # 2. Stop polling + if app.updater and app.updater.running: + app.updater.stop_polling() + + # 3. Now safe to shutdown the application + await app.shutdown() + + # 4. Close DB pool + await close_pool() + + logger.info("Shutdown complete.") + + +async def main(): + global _scheduler_task + + app = Application.builder().token(TOKEN).build() + register_handlers(app) + setup_global_commands(app) + + # Start polling + app.updater.start_polling(drop_pending_updates=True) + + # Start scheduler + _scheduler_task = asyncio.create_task(run_scheduler()) + + # Set up signal handlers for graceful shutdown + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler( + sig, lambda: asyncio.create_task(cleanup(app)) + ) + + # Wait until the app is done or cleanup finishes + try: + await _scheduler_task # runs forever until cancelled + except asyncio.CancelledError: + pass + + # Final cleanup (in case signal handler already ran) + await cleanup(app) +``` + +### What changes in `main.py`: + +1. **Remove** the broken try/except KeyboardInterrupt pattern +2. **Add** the `cleanup()` async function with proper ordering +3. **Add** signal handlers for SIGTERM and SIGINT via `loop.add_signal_handler` +4. **Track** the scheduler task as a global so cleanup can reference it +5. **Remove** any direct calls to `app.shutdown()` elsewhere + +## Acceptance Criteria + +- [ ] Sending `docker stop willhaben-tracker-worker-1` results in clean exit (exit code 0) within 10 seconds +- [ ] No `RuntimeError: This Application is still running!` appears in logs during shutdown +- [ ] The asyncpg pool closes cleanly — no "connection closed unexpectedly" warnings after shutdown +- [ ] In-flight notifications are allowed to complete (not dropped mid-send) +- [ ] Restarting the container via `docker restart` works without errors +- [ ] Logs show clear shutdown sequence: "Shutting down..." → "Shutdown complete." diff --git a/docs/phase-0/task-healthcheck.md b/docs/phase-0/task-healthcheck.md new file mode 100644 index 0000000..af15adc --- /dev/null +++ b/docs/phase-0/task-healthcheck.md @@ -0,0 +1,220 @@ +# Task: Healthcheck endpoint + docker-compose configuration + +## Description + +The worker container currently has no healthcheck configured. Docker only relies on `restart: unless-stopped` which reacts to process crashes but **not** to logical failures (e.g., DB connection lost, Telegram API rate-limited indefinitely, scheduler deadlocked). + +This task adds a lightweight HTTP `/health` endpoint inside the worker that reports the status of critical subsystems (DB connectivity, scheduler last run time), and configures Docker to use it for automatic restart decisions. + +## Architecture + +``` +┌───────────────────────────────────────────────┐ +│ worker container │ +│ │ +│ ┌───────────────────────────────────────┐ │ +│ │ aiohttp web server (port 8765) │ │ +│ │ │ │ +│ │ GET /health │ │ +│ │ → check DB pool connectivity │ │ +│ │ → check scheduler last run time │ │ +│ │ → return JSON status │ │ +│ │ │ │ +│ │ GET /stats │ │ +│ │ → extended metrics (optional) │ │ +│ └───────────────────────────────────────┘ │ +│ │ +└───────────────────────────────────────────────┘ + +docker healthcheck: + curl -f http://localhost:8765/health || exit 1 + interval=30s timeout=5s retries=3 start_period=10s +``` + +### Health check response format + +**Healthy (200 OK):** +```json +{ + "status": "ok", + "uptime_seconds": 3600, + "scheduler_last_run": "2026-07-04T12:34:56+00:00", + "db_connected": true, + "telegram_polling": true +} +``` + +**Unhealthy (503 Service Unavailable):** +```json +{ + "status": "unhealthy", + "reason": "scheduler_stale_last_run_120s" +} +``` + +### Key design decisions + +- **aiohttp over http.server**: aiohttp is already async-native and integrates naturally with the existing asyncio event loop. No blocking I/O. + - *Alternative*: Could use `websockets` or just `asyncio.start_server`, but aiohttp provides proper HTTP/JSON handling out of the box. + - Adding `aiohttp` to requirements.txt (~500KB overhead). +- **DB check**: Execute a lightweight `SELECT 1` against the pool, not a full connection creation. Fast and accurate. +- **Scheduler staleness threshold**: If last_run > 2 * max_keyword_interval + 60s, mark unhealthy. This prevents false positives from slow scrape cycles. + +## Implementation Details + +### 1. Add `aiohttp` to requirements.txt + +```txt +python-telegram-bot==21.4 +asyncpg==0.30.0 +httpx==0.27.2 +python-dotenv==1.0.1 +aiohttp==3.9.5 # ← new: healthcheck HTTP server +``` + +### 2. Create `worker/src/health.py` + +```python +import time +import logging +from aiohttp import web + +logger = logging.getLogger(__name__) + +_start_time = time.time() +_last_scheduler_run: float | None = None + + +def record_scheduler_run(): + """Call this at the start of each scheduler cycle.""" + global _last_scheduler_run + _last_scheduler_run = time.time() + + +async def health_handler(request: web.Request) -> web.Response: + from db import get_pool + + status = "ok" + checks = { + "uptime_seconds": int(time.time() - _start_time), + "db_connected": False, + "telegram_polling": True, # will be set by main.py + } + + if _last_scheduler_run: + elapsed = time.time() - _last_scheduler_run + checks["scheduler_last_run_seconds_ago"] = round(elapsed) + + # Check DB connectivity + try: + pool = await get_pool() + await pool.fetchval("SELECT 1") + checks["db_connected"] = True + except Exception as e: + status = "unhealthy" + checks["db_error"] = str(e) + + # Check scheduler staleness (threshold: 300s configurable) + stale_threshold = int(os.getenv("HEALTHCHECK_SCHEDULER_STALE_S", "300")) + if _last_scheduler_run and (time.time() - _last_scheduler_run) > stale_threshold: + status = "unhealthy" + checks["scheduler_stale"] = True + + body = {"status": status, **checks} + + status_code = 200 if status == "ok" else 503 + return web.json_response(body, status=status_code) + + +async def stats_handler(request: web.Request) -> web.Response: + """Extended stats endpoint (for future dashboard).""" + from db import get_pool + + pool = await get_pool() + + kw_count = await pool.fetchval("SELECT COUNT(*) FROM keywords") or 0 + active_kw = await pool.fetchval( + "SELECT COUNT(*) FROM keywords WHERE is_active" + ) or 0 + ad_count = await pool.fetchval("SELECT COUNT(*) FROM ads") or 0 + notif_count = await pool.fetchval("SELECT COUNT(*) FROM notifications") or 0 + + return web.json_response({ + "keywords": kw_count, + "active_keywords": active_kw, + "ads_indexed": ad_count, + "notifications_sent": notif_count, + }) + + +def create_health_app() -> web.Application: + app = web.Application() + app.router.add_get("/health", health_handler) + app.router.add_get("/stats", stats_handler) + return app +``` + +### 3. Start health server in `main.py` + +Add after pool initialization, before scheduler starts: + +```python +from health import create_health_app, record_scheduler_run + +# ... in main(): +health_app = create_health_app() +runner = web.AppRunner(health_app) +await runner.setup() +site = web.TCPSite(runner, "0.0.0.0", 8765) +await site.start() +logger.info("Health check server listening on :8765") +``` + +### 4. Call `record_scheduler_run()` at the top of each scheduler loop iteration in `main.py` + +### 5. Update `docker-compose.yml` with healthcheck + +```yaml +services: + worker: + build: ./worker + restart: unless-stopped + env_file: + - .env + networks: + - supabase_default + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8765/health"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 10s + +networks: + supabase_default: + external: true +``` + +### 6. Add `curl` to the Docker image (not included in slim) + +Update `worker/Dockerfile`: + +```dockerfile +FROM python:3.12-slim +RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY src/ . +ENTRYPOINT ["/app/entrypoint.sh"] +CMD ["python", "main.py"] +``` + +## Acceptance Criteria + +- [ ] `curl http://localhost:8765/health` returns 200 with JSON status when the worker is healthy +- [ ] Simulating a DB failure (e.g., disconnecting from Supabase) causes `/health` to return 503 within one polling cycle +- [ ] Docker reports the container as `healthy` after the start period (~10-60 seconds) +- [ ] When the healthcheck fails 3 consecutive times, Docker automatically restarts the container +- [ ] The health server does not interfere with the scheduler or bot loop (no blocking I/O) +- [ ] `/stats` endpoint returns accurate counts of keywords, ads, and notifications diff --git a/docs/phase-0/task-search-path-fix.md b/docs/phase-0/task-search-path-fix.md new file mode 100644 index 0000000..7c02fc4 --- /dev/null +++ b/docs/phase-0/task-search-path-fix.md @@ -0,0 +1,101 @@ +# Task: search_path consistency fix + +## Description + +There is a mismatch between the database schema configuration and where tables actually live: + +**In `db.py` (`_init_connection`):** +```python +await conn.execute("SET search_path TO willhaben_tracker") +``` + +**But in `supabase-migration.sql`:** +- The migration file has `SET search_path TO willhaben_tracker;` but the actual tables were created in `public` schema (not `willhaben_tracker`) when manually applied. +- The original local Supabase migrations used `public` schema. + +This creates two problems: +1. If a connection does NOT run `_init_connection`, queries to bare table names (`SELECT FROM keywords`) fail with "relation does not exist". +2. Tools that connect directly (pgAdmin, DBeaver, PostgREST) won't find the tables without explicit schema qualification or their own search_path config. + +## Decision: Remove custom schema, use `public` for everything + +Since Supabase uses `public` by default and all existing tools/GRANTs in `post-boot.sql` target `public`, we simplify by removing the `willhaben_tracker` schema entirely. + +### Rationale + +| Factor | Custom Schema (`willhaben_tracker`) | Public Schema | +|--------|--------------------------------------|---------------| +| Supabase Studio compatibility | Requires manual config | ✅ Works out of box | +| PostgREST/Kong access | Needs search_path override | ✅ Default | +| Direct pgAdmin/DBeaver access | Must set schema per-connection | ✅ Always visible | +| Grant statements in post-boot.sql | ❌ Targets wrong schema | ✅ Already correct | +| Code changes needed | Minimal (remove one line) | Minimal | + +## Architecture + +``` +BEFORE (broken): + db.py init: SET search_path TO willhaben_tracker + tables live in: public.keywords, public.ads, ... + Result: queries fail because search_path != table location + +AFTER (fixed): + db.py init: REMOVED (no custom search_path) + tables live in: public.keywords, public.ads, ... + Result: all connections find tables via default 'public' schema +``` + +## Implementation Details + +### 1. Remove `SET search_path` from `db.py` + +**Current:** +```python +async def _init_connection(conn: asyncpg.Connection) -> None: + await conn.execute("SET search_path TO willhaben_tracker") +``` + +**After:** +```python +# Removed entirely - using default public schema +``` + +The `_init_connection` function can be removed, and `init=_init_connection` removed from `create_pool()`. Alternatively keep the callback but make it a no-op for future use: + +```python +async def _init_connection(conn: asyncpg.Connection) -> None: + # Reserved for future connection-level settings (e.g., timezone) + pass +``` + +### 2. Clean up migration files + +- Remove `SET search_path TO willhaben_tracker;` from `supabase-migration.sql` / `01-schema.sql` +- Ensure all tables are explicitly created in the default schema (no need for prefix — that's already what happens) + +### 3. Update `post-boot.sql` grants if needed + +Current grants already target `public`: +```sql +GRANT USAGE ON SCHEMA public TO supabase_admin; +``` +→ ✅ Already correct, no change needed. + +### 4. Add schema qualification to migration file (defensive) + +To be explicit and future-proof, the migration can qualify the schema: + +```sql +-- Instead of implicit public.keywords +CREATE TABLE IF NOT EXISTS public.keywords (...) +``` + +This is optional but makes it unambiguous where tables live. For simplicity we'll keep implicit `public` since that's the default everywhere. + +## Acceptance Criteria + +- [ ] Removing `_init_connection` from db.py does not break any query (all table names resolve correctly) +- [ ] Connecting to the database via pgAdmin/DBeaver shows all tables without schema configuration +- [ ] The healthcheck endpoint DB check works without custom search_path +- [ ] No "relation does not exist" errors in worker logs after redeploy +- [ ] Migration SQL file no longer references `willhaben_tracker` schema diff --git a/supabase/migrations/supabase-migration.sql b/supabase/migrations/supabase-migration.sql index 3447414..3a9b1c6 100644 --- a/supabase/migrations/supabase-migration.sql +++ b/supabase/migrations/supabase-migration.sql @@ -1,5 +1,5 @@ -CREATE SCHEMA IF NOT EXISTS willhaben_tracker; -SET search_path TO willhaben_tracker; +-- supabase-migration.sql — DEPRECATED: use 01-schema.sql instead (in migrations/) +-- This file is kept for backward compatibility reference only. -- ----------------------------------------------------------- -- users (whitelisted Telegram users) diff --git a/worker/Dockerfile b/worker/Dockerfile index 062da9e..9230869 100644 --- a/worker/Dockerfile +++ b/worker/Dockerfile @@ -1,6 +1,20 @@ FROM python:3.12-slim WORKDIR /app + +# ── Dependencies ──────────────────────────────── COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt + +# ── Application code + migrations ─────────────── COPY src/ . + +# Make entrypoint executable +RUN chmod +x entrypoint.sh + +# ── Healthcheck (Docker-native) ───────────────── +HEALTHCHECK --interval=15s --timeout=3s --retries=3 \ + CMD python -c "import httpx,os; r=httpx.get(f'http://localhost:{os.getenv(\"HEALTH_PORT\",\"8765\")}/health'); r.raise_for_status()" || exit 1 + +# ── Entrypoint: migrate then run ──────────────── +ENTRYPOINT ["/app/entrypoint.sh"] CMD ["python", "main.py"] diff --git a/worker/requirements.txt b/worker/requirements.txt index 3a74ff8..5eb7d50 100644 --- a/worker/requirements.txt +++ b/worker/requirements.txt @@ -1,4 +1,5 @@ python-telegram-bot==21.4 asyncpg==0.30.0 httpx==0.27.2 +aiohttp>=3.9,<4 python-dotenv==1.0.1 diff --git a/worker/src/db.py b/worker/src/db.py index a6fffe4..0aa6199 100644 --- a/worker/src/db.py +++ b/worker/src/db.py @@ -7,9 +7,6 @@ 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 @@ -22,7 +19,6 @@ 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 diff --git a/worker/src/entrypoint.sh b/worker/src/entrypoint.sh new file mode 100644 index 0000000..ccbdc46 --- /dev/null +++ b/worker/src/entrypoint.sh @@ -0,0 +1,40 @@ +#!/usr/bin/env bash +set -euo pipefail + +# ──────────────────────────────────────────────── +# Entrypoint: Run migrations, then start the app. +# ──────────────────────────────────────────────── + +echo "[entrypoint] Waiting for database to be ready..." + +# Wait for PostgreSQL to accept connections (max 30s retries) +for i in $(seq 1 30); do + if python -c "import asyncpg,os; asyncio.get_event_loop().run_until_complete( + asyncpg.connect( + host=os.getenv('POSTGRES_HOST','db'), + port=int(os.getenv('POSTGRES_PORT','5432')), + user=os.getenv('POSTGRES_USER','postgres'), + password=os.getenv('POSTGRES_PASSWORD'), + database=os.getenv('POSTGRES_DB','postgres'), + ) + )" 2>/dev/null; then + echo "[entrypoint] Database is ready." + break + fi + if [ "$i" -eq 30 ]; then + echo "[entrypoint] ERROR: Could not connect to database after 30 attempts." >&2 + exit 1 + fi + sleep 1 +done + +# ── Run migrations ────────────────────────────── +echo "[entrypoint] Running database migrations..." +python /app/migrate.py || { + echo "[entrypoint] ERROR: Migration failed — aborting startup." >&2 + exit 1 +} +echo "[entrypoint] Migrations complete." + +# ── Start the application ─────────────────────── +exec "$@" diff --git a/worker/src/health.py b/worker/src/health.py new file mode 100644 index 0000000..2f9be5a --- /dev/null +++ b/worker/src/health.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +"""Lightweight HTTP healthcheck server for Docker integration. + +Exposes: + GET /health — returns 200 OK or 503 with subsystem status JSON + GET /stats — extended DB counts (keywords, ads, notifications) + +Runs on port HEALTH_PORT (default: 8765). +""" + +import os +import time +from typing import Any + +from aiohttp import web + +logger = None # lazy-imported to avoid circular imports + +_start_time: float | None = None +_last_scheduler_run: float | None = None +_telegram_polling: bool = False + + +def _get_logger() -> "logging.Logger": # type: ignore[name-defined] + global logger + if logger is None: + import logging as _log + logger = _log.getLogger(__name__) + return logger + + +# ──────────────────────────────────────────────── +# Lifecycle hooks — called from main.py +# ──────────────────────────────────────────────── + +def set_start_time() -> None: + global _start_time + _start_time = time.time() + + +def record_scheduler_run() -> None: + """Call this at the start of each scheduler cycle.""" + global _last_scheduler_run + _last_scheduler_run = time.time() + + +def set_telegram_polling(active: bool) -> None: + global _telegram_polling + _telegram_polling = active + + +# ──────────────────────────────────────────────── +# Health endpoint — fast, non-blocking DB probe +# ──────────────────────────────────────────────── + +async def health_handler(request: web.Request) -> web.Response: # noqa: ARG001 + status = "ok" + checks: dict[str, Any] = {} + + if _start_time is not None: + checks["uptime_seconds"] = int(time.time() - _start_time) + + # — DB connectivity (lightweight SELECT 1) — + try: + from db import get_pool + pool = await get_pool() + await pool.fetchval("SELECT 1") + checks["db_connected"] = True + except Exception as exc: + status = "unhealthy" + checks["db_error"] = str(exc) + + # — Scheduler staleness — + stale_threshold = int(os.getenv("HEALTHCHECK_SCHEDULER_STALE_S", "300")) + if _last_scheduler_run is not None: + elapsed = time.time() - _last_scheduler_run + checks["scheduler_last_run_seconds_ago"] = round(elapsed, 1) + if elapsed > stale_threshold: + status = "unhealthy" + checks["scheduler_stale"] = True + + # — Telegram polling status — + checks["telegram_polling"] = _telegram_polling + + body = {"status": status, **checks} + + code = 200 if status == "ok" else 503 + return web.json_response(body, status=code) + + +# ──────────────────────────────────────────────── +# Stats endpoint — DB counts (slower query) +# ──────────────────────────────────────────────── + +async def stats_handler(request: web.Request) -> web.Response: # noqa: ARG001 + from db import get_pool + pool = await get_pool() + + kw_count = await pool.fetchval("SELECT COUNT(*) FROM keywords") or 0 + active_kw = await pool.fetchval( + "SELECT COUNT(*) FROM keywords WHERE is_active" + ) or 0 + ad_count = await pool.fetchval("SELECT COUNT(*) FROM ads") or 0 + notif_count = await pool.fetchval("SELECT COUNT(*) FROM notifications") or 0 + + return web.json_response({ + "keywords": kw_count, + "active_keywords": active_kw, + "ads_indexed": ad_count, + "notifications_sent": notif_count, + }) + + +# ──────────────────────────────────────────────── +# App factory — creates a runnable aiohttp app +# ──────────────────────────────────────────────── + +def create_health_app() -> web.Application: + app = web.Application() + app.router.add_get("/health", health_handler) + app.router.add_get("/stats", stats_handler) + return app diff --git a/worker/src/main.py b/worker/src/main.py index 337a9e8..b1dd312 100644 --- a/worker/src/main.py +++ b/worker/src/main.py @@ -6,11 +6,13 @@ import signal import sys from contextlib import suppress +import aiohttp.web as web from dotenv import load_dotenv from telegram import Update from telegram.ext import Application, ExtBot from db import close_pool, get_pool +from health import create_health_app, record_scheduler_run, set_start_time, set_telegram_polling from scraper import extract_ad_fields, fetch_ads from notifier import log_notification, notify_new_ad, notify_price_drop @@ -21,6 +23,7 @@ load_dotenv() async def scheduler_task(pool: object, bot: ExtBot) -> None: while True: + record_scheduler_run() # mark this cycle as started try: rows = await pool.fetch( "SELECT id, keyword, interval_minutes, initial_loaded FROM keywords " @@ -148,6 +151,8 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None: async def main() -> None: + set_start_time() # for health endpoint uptime tracking + logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", @@ -166,6 +171,15 @@ async def main() -> None: await setup_global_commands(app) register_handlers(app) + # ── Start healthcheck HTTP server ────────────────────────────── + health_app = create_health_app() + runner = web.AppRunner(health_app) + await runner.setup() + _health_port = int(os.getenv("HEALTH_PORT", "8765")) + site = web.TCPSite(runner, "0.0.0.0", _health_port) + await site.start() + logger.info("Health check server listening on :%d", _health_port) + scheduler = asyncio.ensure_future(scheduler_task(pool, app.bot)) loop = asyncio.get_running_loop() @@ -183,23 +197,42 @@ async def main() -> None: await app.start() logger.info("Bot started with long polling") + set_telegram_polling(True) poll_task = asyncio.ensure_future(app.updater.start_polling()) # type: ignore[attr-defined] await stop - logger.info("Shutting down...") + logger.info("Signal received — initiating graceful shutdown...") finally: + # ── Cancel scheduler with grace period ─────────────────────── + logger.info("Cancelling scheduler task...") scheduler.cancel() with suppress(asyncio.CancelledError): - await scheduler + try: + await asyncio.wait_for(scheduler, timeout=5.0) + except asyncio.TimeoutError: + logger.warning("Scheduler task did not finish within 5s — force cancelled.") + # ── Stop Telegram polling ──────────────────────────────────── + set_telegram_polling(False) + logger.info("Stopping Telegram poller...") poll_task.cancel() with suppress(asyncio.CancelledError): await poll_task + # ── Shutdown application ───────────────────────────────────── + logger.info("Shutting down Telegram bot application...") await app.shutdown() + + # ── Close health server ─────────────────────────────────────── + logger.info("Stopping health check server...") + await runner.cleanup() + + # ── Close DB pool ───────────────────────────────────────────── + logger.info("Closing database connection pool...") await close_pool() - logger.info("Shutdown complete") + + logger.info("Shutdown complete.") if __name__ == "__main__": diff --git a/worker/src/migrate.py b/worker/src/migrate.py new file mode 100644 index 0000000..503e99c --- /dev/null +++ b/worker/src/migrate.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Run pending SQL migrations against the configured PostgreSQL database. + +Usage: python migrate.py + +Tracks applied migrations in `willhaben_migrations` table and only applies +missing ones. Forward-only — rollback means restore from DB backup.""" + +import asyncio +import glob +import logging +import os +import sys + +import asyncpg + +logger = logging.getLogger(__name__) + +MIGRATIONS_DIR = os.path.join(os.path.dirname(__file__), "migrations") + + +def _dsn_parts() -> dict: + """Return a dict of DSN connection parameters from env vars.""" + return { + "host": os.getenv("POSTGRES_HOST", "db"), + "port": int(os.getenv("POSTGRES_PORT", "5432")), + "user": os.getenv("POSTGRES_USER", "postgres"), + "password": os.getenv("POSTGRES_PASSWORD"), + "database": os.getenv("POSTGRES_DB", "postgres"), + } + + +async def run_migrations() -> None: + """Connect to the database and apply any pending SQL migrations.""" + conn = await asyncpg.connect(**_dsn_parts()) + + try: + # Create tracking table (always in public schema) + await conn.execute(""" + CREATE TABLE IF NOT EXISTS willhaben_migrations ( + name text PRIMARY KEY, + applied_at timestamptz DEFAULT now() + ); + """) + + # Discover and sort migration files alphabetically + sql_files = sorted(glob.glob(os.path.join(MIGRATIONS_DIR, "*.sql"))) + + if not sql_files: + logger.warning("No migration files found in %s", MIGRATIONS_DIR) + return + + applied_count = 0 + for fpath in sql_files: + name = os.path.basename(fpath) + + already_applied = await conn.fetchval( + "SELECT 1 FROM willhaben_migrations WHERE name = $1", name, + ) + if already_applied: + logger.info("Skipping already-applied migration: %s", name) + continue + + logger.info("Applying migration: %s", name) + + with open(fpath, encoding="utf-8") as fh: + sql = fh.read() + + # Execute within a transaction block for safety + async with conn.transaction(): + await conn.execute(sql) + await conn.execute( + "INSERT INTO willhaben_migrations (name) VALUES ($1)", name, + ) + + applied_count += 1 + logger.info("Applied migration: %s", name) + + if applied_count: + logger.info("Migration complete — %d new migration(s) applied.", applied_count) + else: + logger.info("Database is up to date (0 pending migrations).") + + except asyncpg.PostgresError as exc: + logger.error("Migration failed: %s", exc) + raise SystemExit(1) from exc + finally: + await conn.close() + + +if __name__ == "__main__": + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + asyncio.run(run_migrations()) diff --git a/worker/src/migrations/01-schema.sql b/worker/src/migrations/01-schema.sql new file mode 100644 index 0000000..6a56885 --- /dev/null +++ b/worker/src/migrations/01-schema.sql @@ -0,0 +1,123 @@ +-- ============================================================ +-- willhaben-tracker — consolidated schema (single source of truth) +-- Merged from: 01-init.sql, 02-image-and-pricing.sql, 03-global-keywords.sql +-- ============================================================ + +-- ----------------------------------------------------------- +-- 1. 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() +); + +-- ----------------------------------------------------------- +-- 2. 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)); + +-- ----------------------------------------------------------- +-- 3. 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() +); + +-- ----------------------------------------------------------- +-- 4. 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 +); + +-- ----------------------------------------------------------- +-- 5. 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) +); + +-- ----------------------------------------------------------- +-- 6. 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() +); + +-- ----------------------------------------------------------- +-- 7. 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 +-- ============================================================ + +-- Keywords: fast lookup for active keywords ordered by last scrape time +CREATE INDEX IF NOT EXISTS idx_keywords_active_scraped + ON keywords(is_active, last_scraped_at) WHERE is_active = true; + +-- Keyword subscriptions: find all subscribers of a keyword +CREATE INDEX IF NOT EXISTS idx_keyword_subscriptions_user_id + ON keyword_subscriptions(user_id); + +-- Ads: fast lookup by willhaben ad ID (unique constraint already implies an index) + +-- Price history: look up changes for a specific ad +CREATE INDEX IF NOT EXISTS idx_price_history_ad_id + ON price_history(ad_id); + +-- Notifications: recent messages per user +CREATE INDEX IF NOT EXISTS idx_notifications_user_sent + ON notifications(user_id, sent_at DESC); + +-- Scrape logs: latest runs per keyword +CREATE INDEX IF NOT EXISTS idx_scrape_logs_keyword_at + ON scrape_logs(keyword_id, scraped_at DESC); diff --git a/worker/src/migrations/zz-seed.sql b/worker/src/migrations/zz-seed.sql new file mode 100644 index 0000000..c32c5a7 --- /dev/null +++ b/worker/src/migrations/zz-seed.sql @@ -0,0 +1,8 @@ +-- ============================================================ +-- Seed: initial admin user +-- Runs last (alphabetically), after all schema migrations. +-- ============================================================ + +INSERT INTO users (telegram_id, username, first_name, is_admin, is_active) +VALUES (298181113, NULL, 'Admin', true, true) +ON CONFLICT (telegram_id) DO NOTHING;