entrypoint.py

#!/usr/bin/env python3
"""
Container entrypoint for BRIDGE Modul-X on Hugging Face Spaces.

Responsibilities
----------------
1. Ensure Redis for cache + Celery broker/results, in priority order:
   a. Space secrets win ONLY if they answer PING:
      REDIS_URL / CELERY_BROKER_URL / CELERY_RESULT_BACKEND.
      (A stale secret — e.g. an expired Upstash DB whose domain no longer
      resolves — must never shadow the sidecar; it used to crash-loop the
      whole app.)
   b. Otherwise run a local redis-server sidecar on 127.0.0.1:6379
      (installed in the image; volatile memory, no persistence). This is the
      default production path — no external dependency, no expiry.
   c. If the sidecar binary is unavailable, fall back to Upstash free-tier
      provisioning (deterministic idempotency keys; DBs expire in ~3 days
      unless claimed at the console URL printed here).
2. Supervise child processes:
   - redis-server (sidecar mode only)
   - Celery worker  (bulk validation queue)
   - uvicorn        (FastAPI web app on $PORT / 7860)
   Sidecar/worker are restarted if they die; if the web process dies the
   container exits and the platform restarts it.
"""
import os
import signal
import subprocess
import sys
import time
import uuid

# Fixed namespace so the per-store Idempotency-Keys are deterministic.
_NS = uuid.UUID("7c9e6679-7425-40de-944b-e07fc1f90ae7")

SIDECAR_PORT = 6379


def _stable_db_id(store: str) -> str:
    """Deterministic UUIDv4-FORM id (Upstash rejects non-v4 idempotency keys):
    derive bytes from uuid5, then patch version/variant bits to v4/RFC-4122."""
    b = bytearray(uuid.uuid5(_NS, f"bridge-modul-x/{store}").bytes)
    b[6] = (b[6] & 0x0F) | 0x40
    b[8] = (b[8] & 0x3F) | 0x80
    return str(uuid.UUID(bytes=bytes(b)))


# --------------------------------------------------------------------------- #
# Upstash free-tier fallback (only used when no local redis-server exists)
# --------------------------------------------------------------------------- #
def _provision_upstash(store: str):
    from app.utils.upstash_free import create_database

    for attempt, db_id in enumerate([_stable_db_id(store), ""]):
        try:
            db = create_database(db_id=db_id)
            print(
                f"[ENTRYPOINT] upstash {store}: db {db.db_id} ok "
                f"(expires {db.expires}) - claim to keep it: {db.console_url}",
                flush=True,
            )
            return db.connection_string(0)
        except Exception as exc:
            print(f"[ENTRYPOINT] upstash {store}: failed (try {attempt}): {exc}", flush=True)
    return None


# --------------------------------------------------------------------------- #
# Redis resolution
# --------------------------------------------------------------------------- #
def _secret_url(env_name: str):
    """An externally-provided (Space secret) URL wins over any fallback."""
    url = os.environ.get(env_name, "").strip()
    if url and "localhost" not in url and "127.0.0.1" not in url:
        return url
    return None


def _wait_for_redis(url: str, seconds: int = 30) -> bool:
    import redis as _redis

    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        try:
            client = _redis.Redis.from_url(url, socket_connect_timeout=2, socket_timeout=2)
            if client.ping():
                client.close()
                return True
        except Exception:
            time.sleep(0.5)
    return False


def _resolve_redis(children: dict):
    """Return (cache, broker, result) URLs and populate `children` with a
    redis-server sidecar process when one is started."""
    cache = _secret_url("REDIS_URL")
    broker = _secret_url("CELERY_BROKER_URL")
    result = _secret_url("CELERY_RESULT_BACKEND")
    if cache or broker or result:
        # Trust-but-verify: every provided secret URL must answer PING,
        # otherwise ignore all of them and use the sidecar instead.
        dead = [
            u for u in (cache, broker, result) if u and not _wait_for_redis(u, seconds=6)
        ]
        if not dead:
            print("[ENTRYPOINT] using Redis from Space secrets (PING ok)", flush=True)
            cache = cache or broker or result
            broker = broker or cache
            result = result or broker
            return cache, broker, result
        print(
            "[ENTRYPOINT] WARNING: Space secret Redis unreachable "
            f"({len(dead)} of the URLs failed PING) - ignoring secrets",
            flush=True,
        )
        cache = broker = result = None

    # Local sidecar: same image, same container, always reachable.
    try:
        proc = subprocess.Popen(
            [
                "redis-server",
                "--port", str(SIDECAR_PORT),
                "--bind", "127.0.0.1",
                "--save", "",
                "--appendonly", "no",
                "--maxmemory", "120mb",
                "--maxmemory-policy", "allkeys-lru",
            ],
            env=os.environ,
        )
        children["redis"] = proc
        print(f"[ENTRYPOINT] redis-server sidecar started (pid {proc.pid})", flush=True)
    except FileNotFoundError:
        print("[ENTRYPOINT] redis-server binary missing - using Upstash fallback", flush=True)
    except Exception as exc:
        print(f"[ENTRYPOINT] sidecar spawn failed ({exc}) - using Upstash fallback", flush=True)

    if "redis" in children:
        base = f"redis://127.0.0.1:{SIDECAR_PORT}"
        if _wait_for_redis(f"{base}/0"):
            print("[ENTRYPOINT] sidecar PONG ok", flush=True)
            # Separate logical DBs for cache / broker / results.
            return f"{base}/0", f"{base}/1", f"{base}/2"
        print("[ENTRYPOINT] sidecar never answered PING - falling back", flush=True)
        try:
            children.pop("redis").terminate()
        except Exception:
            pass

    cache = _provision_upstash("cache")
    broker = _provision_upstash("celery-broker") or cache
    result = _provision_upstash("celery-result") or broker
    return cache, broker, result


def main():
    children = {}
    state = {"stopping": False}

    cache_url, broker_url, result_url = _resolve_redis(children)
    if cache_url:
        os.environ["REDIS_URL"] = cache_url
    if broker_url:
        os.environ["CELERY_BROKER_URL"] = broker_url
    if result_url:
        os.environ["CELERY_RESULT_BACKEND"] = result_url
    if not broker_url:
        print("[ENTRYPOINT] WARNING: no Redis reachable - bulk jobs use the synchronous fallback", flush=True)

    # Hugging Face Spaces always proxies to container port 7860; a PORT
    # env/secret with another value (e.g. 8000) must not steer uvicorn,
    # or the platform never sees the app start.
    port = "7860"

    def spawn(kind):
        if kind == "worker":
            cmd = [
                sys.executable, "-m", "celery",
                "-A", "app.tasks.celery_tasks:celery_app", "worker",
                "--loglevel=info", "--concurrency=4",
            ]
        elif kind == "redis":
            cmd = [
                "redis-server", "--port", str(SIDECAR_PORT), "--bind", "127.0.0.1",
                "--save", "", "--appendonly", "no",
                "--maxmemory", "120mb", "--maxmemory-policy", "allkeys-lru",
            ]
        else:
            cmd = [
                sys.executable, "-m", "uvicorn", "app.main:app",
                "--host", "0.0.0.0", "--port", port,
            ]
        proc = subprocess.Popen(cmd, env=os.environ)
        children[kind] = proc
        print(f"[ENTRYPOINT] started {kind} (pid {proc.pid})", flush=True)
        return proc

    def on_signal(signum, frame):
        state["stopping"] = True
        for proc in list(children.values()):
            try:
                proc.terminate()
            except Exception:
                pass

    signal.signal(signal.SIGTERM, on_signal)
    signal.signal(signal.SIGINT, on_signal)

    # Celery worker (bulk validation queue) starts ONLY when bulk is routed
    # through Celery on this host. With CELERY_BULK_ENABLED=false (the Space
    # default) bulk runs synchronously in-process, and spawning a worker
    # against a misconfigured broker (e.g. a rediss:// URL without
    # ssl_cert_reqs) produced a pointless 5s crash-restart loop.
    run_celery_worker = (
        os.environ.get("CELERY_BULK_ENABLED", "").strip().lower() in ("1", "true", "yes")
    )
    if broker_url and run_celery_worker:
        spawn("worker")
    elif broker_url:
        print(
            "[ENTRYPOINT] CELERY_BULK_ENABLED!=true — bulk runs sync; "
            "celery worker NOT started",
            flush=True,
        )
    web = spawn("web")

    while not state["stopping"]:
        if web.poll() is not None:
            print(f"[ENTRYPOINT] web exited ({web.returncode}) - shutting down", flush=True)
            break
        if "redis" in children and children["redis"].poll() is not None:
            print("[ENTRYPOINT] redis sidecar exited - restarting in 2s", flush=True)
            time.sleep(2)
            if not state["stopping"]:
                spawn("redis")
        if run_celery_worker and broker_url and children.get("worker") and children["worker"].poll() is not None:
            print(f"[ENTRYPOINT] worker exited ({children['worker'].returncode}) - restarting in 5s", flush=True)
            time.sleep(5)
            if not state["stopping"]:
                spawn("worker")
        time.sleep(2)

    for proc in list(children.values()):
        try:
            proc.terminate()
            proc.wait(timeout=10)
        except Exception:
            try:
                proc.kill()
            except Exception:
                pass

    sys.exit(web.returncode or 0)


if __name__ == "__main__":
    main()