space_worker/config.py

import os
from dataclasses import dataclass
from pathlib import Path

from dotenv import load_dotenv


TEMPLATE_ROOT = Path(__file__).resolve().parents[1]
load_dotenv(TEMPLATE_ROOT / ".env.example", override=False)
load_dotenv(TEMPLATE_ROOT / ".env", override=True)


def env_bool(name, default=False):
    value = os.getenv(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def env_csv(name, default):
    raw_value = os.getenv(name)
    if raw_value is None:
        raw_value = os.getenv("SUPABASE_REQUEST_EVENT", default)
    return tuple(item.strip() for item in raw_value.split(",") if item.strip())


def mount_child_path(mount_path, child_name):
    return f"{mount_path.rstrip('/')}/{child_name.lstrip('/')}"


@dataclass(frozen=True)
class Settings:
    supabase_url: str
    supabase_key: str
    inbox_topic: str
    request_events: tuple[str, ...]
    response_event: str
    inbox_private: bool
    response_private: bool
    realtime_vsn: str
    broadcast_self: bool
    reply_channel_template: str
    heartbeat_seconds: int
    reconnect_seconds: int
    hang_after_worker_start: bool
    log_level: str
    model_bucket_id: str
    model_bucket_mount_path: str
    protexa_instruction_repo_id: str
    protexa_instruction_model_dir: str

    @classmethod
    def from_env(cls):
        model_mount_path = os.getenv("MODEL_BUCKET_MOUNT_PATH", "/models")
        return cls(
            supabase_url=os.getenv("SUPABASE_URL", "").rstrip("/"),
            supabase_key=os.getenv("SUPABASE_SERVICE_KEY") or os.getenv("SUPABASE_ANON_KEY", ""),
            inbox_topic=os.getenv("SUPABASE_SPACE_INBOX_TOPIC", "space-a"),
            request_events=env_csv("SUPABASE_REQUEST_EVENTS", "*"),
            response_event=os.getenv("SUPABASE_RESPONSE_EVENT", "message"),
            inbox_private=env_bool("SUPABASE_SPACE_INBOX_PRIVATE", True),
            response_private=env_bool("SUPABASE_USER_CHANNEL_PRIVATE", True),
            realtime_vsn=os.getenv("SUPABASE_REALTIME_VSN", "2.0.0"),
            broadcast_self=env_bool("SUPABASE_BROADCAST_SELF", True),
            reply_channel_template=os.getenv("SUPABASE_REPLY_CHANNEL_TEMPLATE", "user-{user_id}"),
            heartbeat_seconds=int(os.getenv("SUPABASE_HEARTBEAT_SECONDS", "20")),
            reconnect_seconds=int(os.getenv("SUPABASE_RECONNECT_SECONDS", "5")),
            hang_after_worker_start=env_bool("SPACE_HANG_AFTER_WORKER_START", True),
            log_level=os.getenv("LOG_LEVEL", "INFO").upper(),
            model_bucket_id=os.getenv("MODEL_BUCKET_ID", "admincybers2/protexa-model-space"),
            model_bucket_mount_path=model_mount_path,
            protexa_instruction_repo_id=os.getenv(
                "PROTEXA_INSTRUCTION_REPO_ID",
                "admincybers2/protexa_instruction",
            ),
            protexa_instruction_model_dir=os.getenv(
                "PROTEXA_INSTRUCTION_MODEL_DIR",
                mount_child_path(model_mount_path, "protexa_instruction"),
            ),
        )

    def require_supabase(self):
        if not self.supabase_url:
            raise RuntimeError("SUPABASE_URL is required")
        if not self.supabase_key:
            raise RuntimeError("SUPABASE_SERVICE_KEY or SUPABASE_ANON_KEY is required")

    def accepts_request_event(self, event):
        return "*" in self.request_events or event in self.request_events