Velocity-OS/core/main.py

"""
Velocity — Unified FastAPI Backend
Covers: Catalyst (Meta Marketing), Sentinel (QD Engine), Vault (Trackable Links), Auth

GPU partitioning on AWS:
  - NemoClaw / Ollama → CUDA devices 0, 1  (enforced in nemoclaw.service systemd unit)
  - ComfyUI / Wan 2.2 → CUDA devices 2, 3  (enforced in comfyui.service systemd unit)
"""

import os
import json
import asyncio
import logging
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Set

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from dotenv import load_dotenv

def _load_velocity_env() -> None:
    repo_root = Path(__file__).resolve().parent.parent
    backend_root = repo_root / "backend"

    explicit_env = os.getenv("VELOCITY_ENV_FILE", "").strip()
    candidate_paths = []

    if explicit_env:
        candidate_paths.append(Path(explicit_env))

    candidate_paths.extend(
        [
            backend_root / ".env",
            repo_root / ".env",
        ]
    )

    loaded_any = False
    seen: set[Path] = set()
    for candidate in candidate_paths:
        resolved = candidate.resolve()
        if resolved in seen or not candidate.exists():
            continue
        load_dotenv(candidate, override=not loaded_any)
        loaded_any = True
        seen.add(resolved)

    if not loaded_any:
        load_dotenv()


_load_velocity_env()

from backend.api.routes_catalyst import router as catalyst_router
from backend.api.routes_crm import crm_router, analytics_router
from backend.api.routes_dashboard import router as dashboard_router
from backend.api.routes_oracle import router as oracle_helper_router
from backend.api.routes_mobile_edge import router as mobile_edge_router
from backend.api.routes_inventory import router as inventory_router
from backend.api.routes_admin_surface import router as admin_surface_router
from backend.api.routes_oracle_templates import router as oracle_templates_router
from backend.api.routes_observability import router as observability_router
from backend.api.routes_crm_imports import router as crm_imports_router
from backend.api.routes_comms import router as comms_router
from backend.api.routes_runtime_llm import router as runtime_llm_router
from backend.auth.routes import router as auth_router
from backend.auth.user_directory import ensure_user_directory_schema
from backend.crm.canonical_schema import ensure_canonical_crm_schema
from backend.db.seed_synthetic_crm import seed as seed_synthetic_crm
from backend.db.pool import create_pool, close_pool
from backend.migrations.runner import apply_migrations
from backend.observability import RequestObservabilityMiddleware
from backend.oracle.router_v1 import router as oracle_v1_router
from backend.routers.cctv import router as cctv_router
from backend.routers.scenes import router as scenes_router
from backend.routers.videos import router as videos_router
from backend.routers.vault import router as vault_router
from backend.routers.sentinel import router as sentinel_router, broadcast_sentinel_event

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("velocity.main")

# ── Lifespan: DB pool init / teardown ─────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    try:
        app.state.db_pool = await create_pool()
        logger.info("asyncpg pool created")
        async with app.state.db_pool.acquire() as conn:
            applied = await apply_migrations(conn)
            if applied:
                logger.info("Applied backend migrations: %s", ", ".join(applied))
            has_crm_schema = await conn.fetchval("SELECT to_regclass('public.crm_people') IS NOT NULL")
            if not has_crm_schema:
                schema_path = Path(__file__).resolve().parent / "db" / "db" / "schema_crm_canonical.sql"
                logger.info("Canonical CRM schema missing; applying %s", schema_path)
                await conn.execute(schema_path.read_text(encoding="utf-8"))
        await ensure_user_directory_schema(app)
        await ensure_canonical_crm_schema(app)
        if os.getenv("VELOCITY_AUTO_SEED_SYNTHETIC_CRM", "1").strip().lower() not in {"0", "false", "no"}:
            async with app.state.db_pool.acquire() as conn:
                people_count = await conn.fetchval("SELECT COUNT(*) FROM crm_people")
            if int(people_count or 0) == 0:
                logger.info("CRM is empty; seeding bundled synthetic CRM v2 data")
                counts = await seed_synthetic_crm(pool=app.state.db_pool)
                logger.info("Synthetic CRM v2 seed complete: %s", counts)
    except Exception as exc:
        logger.error("Failed to create DB pool: %s", exc)
        app.state.db_pool = None

    app.state.broadcast_sentinel_event = broadcast_sentinel_event

    yield

    # Shutdown
    await close_pool()
    logger.info("asyncpg pool closed")

# ── App ───────────────────────────────────────────────────────────────────────

app = FastAPI(
    title="Velocity — Neural Core",
    description="Unified backend: Catalyst, Sentinel QD Engine, Vault, Oracle, Auth.",
    version="2.0.0",
    lifespan=lifespan,
)

# ── CORS ──────────────────────────────────────────────────────────────────────

origins = [o.strip() for o in os.getenv("CORS_ORIGINS", "http://localhost:5173").split(",")]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(RequestObservabilityMiddleware)

# ── Static asset serving (Vault files) ───────────────────────────────────────

ASSET_DIR = os.getenv("VELOCITY_ASSET_DIR", "/opt/dlami/nvme/assets")
if os.path.isdir(ASSET_DIR):
    app.mount("/assets", StaticFiles(directory=ASSET_DIR), name="assets")

# ── Routers ───────────────────────────────────────────────────────────────────

app.include_router(catalyst_router, prefix="/api/catalyst", tags=["Catalyst"])
app.include_router(dashboard_router, prefix="/api", tags=["Dashboard"])
app.include_router(crm_router, prefix="/api", tags=["CRM"])
app.include_router(analytics_router, prefix="/api/analytics", tags=["Analytics"])
app.include_router(oracle_helper_router, prefix="/api/oracle", tags=["Oracle"])
app.include_router(oracle_v1_router, prefix="/api/oracle/v1", tags=["Oracle V1"])
app.include_router(oracle_templates_router, prefix="/api/oracle", tags=["Oracle Templates"])
app.include_router(sentinel_router, prefix="/api/sentinel", tags=["Sentinel"])
app.include_router(cctv_router, prefix="/api/cctv", tags=["CCTV"])
app.include_router(scenes_router, prefix="/api/scenes", tags=["Scenes"])
app.include_router(videos_router, prefix="/api/videos", tags=["Videos"])
app.include_router(vault_router, prefix="/api/vault", tags=["Vault"])
app.include_router(mobile_edge_router, prefix="/api/mobile-edge", tags=["Mobile Edge"])
app.include_router(inventory_router, prefix="/api/inventory", tags=["Inventory"])
app.include_router(admin_surface_router, prefix="/api/admin-surface", tags=["Admin Surface"])
app.include_router(observability_router, prefix="/api", tags=["Observability"])
app.include_router(crm_imports_router, prefix="/api", tags=["CRM Canonical"])
app.include_router(comms_router, prefix="/api/comms", tags=["Comms"])
app.include_router(runtime_llm_router, prefix="/api/runtime/llm", tags=["Runtime LLM"])
app.include_router(auth_router)

# Public vault link (no /api prefix — shared externally with prospects)
from backend.routers.vault import router as public_vault_router
app.include_router(public_vault_router, prefix="/vault", tags=["Vault Public"])

# ── Catalyst WebSocket (preserved from v1) ────────────────────────────────────

class _CatalystManager:
    def __init__(self) -> None:
        self.active: Set[WebSocket] = set()

    async def connect(self, ws: WebSocket) -> None:
        await ws.accept()
        self.active.add(ws)

    def disconnect(self, ws: WebSocket) -> None:
        self.active.discard(ws)

    async def broadcast(self, payload: dict) -> None:
        dead: Set[WebSocket] = set()
        for ws in self.active:
            try:
                await ws.send_text(json.dumps(payload))
            except Exception:
                dead.add(ws)
        self.active -= dead


_catalyst_mgr = _CatalystManager()


class _CRMManager:
    def __init__(self) -> None:
        self.active: Set[WebSocket] = set()

    async def connect(self, ws: WebSocket) -> None:
        await ws.accept()
        self.active.add(ws)

    def disconnect(self, ws: WebSocket) -> None:
        self.active.discard(ws)

    async def broadcast(self, payload: dict) -> None:
        dead: Set[WebSocket] = set()
        for ws in self.active:
            try:
                await ws.send_text(json.dumps(payload))
            except Exception:
                dead.add(ws)
        self.active -= dead


_crm_mgr = _CRMManager()


@app.websocket("/ws/catalyst")
async def catalyst_ws(ws: WebSocket) -> None:
    await _catalyst_mgr.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(json.dumps({"type": "ack", "data": data}))
    except WebSocketDisconnect:
        _catalyst_mgr.disconnect(ws)


@app.websocket("/ws/crm")
async def crm_ws(ws: WebSocket) -> None:
    await _crm_mgr.connect(ws)
    await _crm_mgr.broadcast(
        {
                "type": "crm_presence",
                "connected_clients": len(_crm_mgr.active),
                "timestamp": datetime.now(timezone.utc).isoformat(),
        }
    )
    try:
        while True:
            message = await ws.receive_text()
            await ws.send_text(json.dumps({"type": "crm_ack", "data": message}))
    except WebSocketDisconnect:
        _crm_mgr.disconnect(ws)


async def broadcast_live_event(event_type, message, campaign_name=None, value=None):
    payload = {
        "type": event_type,
        "message": message,
        "campaignName": campaign_name,
        "value": value,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    await _catalyst_mgr.broadcast(payload)


app.state.broadcast_live_event = broadcast_live_event


async def broadcast_crm_event(payload: dict) -> None:
    enriched = {
        **payload,
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }
    await _crm_mgr.broadcast(enriched)


app.state.broadcast_crm_event = broadcast_crm_event


# ── Health ─────────────────────────────────────────────────────────────────────

@app.get("/health", tags=["Health"])
async def health() -> dict:
    pool = app.state.db_pool
    db_ok = pool is not None
    return {
        "status": "ok",
        "service": "velocity-backend",
        "version": "2.0.0",
        "db_pool": "connected" if db_ok else "unavailable",
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }