feat(phase-0): stability hardening — search path fix, auto-migrations, graceful shutdown, health endpoint
- 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
This commit is contained in:
hermes
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 "$@"
|
||||
@@ -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
|
||||
+36
-3
@@ -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__":
|
||||
|
||||
@@ -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())
|
||||
@@ -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);
|
||||
@@ -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;
|
||||
Reference in New Issue
Block a user