"""Extension initialization for AgentForms.
Handles database, Redis, rate limiter connection, and background worker setup.
"""
import fcntl
import os
from app.models import get_db, init_db
from app.services.logging import log
from app.services.ratelimit import limiter
_worker_lock_path = "/tmp/agentforms-workers.lock"
def _start_background_workers():
"""Only the first gunicorn worker starts background threads."""
lock_fd = open(_worker_lock_path, "w")
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
# We got the lock — we are the first worker
lock_fd.write(str(os.getpid()))
lock_fd.flush()
return True
except BlockingIOError:
# Another worker already has the lock
lock_fd.close()
return False
def init_extensions(app):
"""Initialize database, Redis, rate limiter, and background workers.
Called from create_app() after middleware is registered.
"""
# Init DB (includes migrations)
init_db()
from app.models import seed_monthly_usage
seed_monthly_usage() # Ensure all users have current month rows
# ─── Redis setup ──────────────────────────────────────────────────────────
redis_client = None
redis_url = os.environ.get("REDIS_URL")
if redis_url:
try:
import redis as redis_lib
redis_client = redis_lib.from_url(redis_url)
redis_client.ping()
log.info("relay", "Redis connected", url=redis_url)
except Exception as e:
log.warning("relay", "Redis unavailable", error=str(e))
# Connect rate limiter to Redis
if redis_client:
limiter.connect(redis_client)
from app.services.backup import start_backup_worker, start_vacuum_worker
from app.services.campaign_emails import start_ab_test_worker, start_reminder_worker
from app.services.usage_alerts import start_usage_alert_worker
from app.services.webhook import start_retry_worker
# Only the first gunicorn worker starts background threads (prevents duplicate workers)
if _start_background_workers():
log.info("app", f"Starting background workers (PID {os.getpid()})")
start_retry_worker()
start_backup_worker()
start_vacuum_worker()
start_usage_alert_worker()
start_reminder_worker()
start_ab_test_worker()
else:
log.info("app", f"Skipping background workers (PID {os.getpid()}) — another worker has the lock")