Pipeline/backend/app/api/gateways.py

"""Thin API wrappers for gateway CRUD and template synchronization."""

from __future__ import annotations

import asyncio
import json
from typing import TYPE_CHECKING
from uuid import UUID, uuid4

from fastapi import APIRouter, Depends, Query, Request
from sqlmodel import col
from sse_starlette.sse import EventSourceResponse

from app.api.deps import require_org_admin, require_org_member
from app.core.auth import AuthContext, get_auth_context
from app.core.config import settings
from app.core.time import utcnow
from app.db import crud
from app.db.pagination import paginate
from app.db.session import get_session
from app.models.agents import Agent
from app.models.gateways import Gateway
from app.models.skills import GatewayInstalledSkill
from app.schemas.common import OkResponse
from app.schemas.gateways import (
    GatewayAgentImportPreviewResponse,
    GatewayAgentImportRequest,
    GatewayAgentImportResponse,
    GatewayCreate,
    GatewayRead,
    GatewayTemplatesSyncResult,
    GatewayUpdate,
)
from app.schemas.gateway_ops import (
    CronJobRead,
    CronStatusResponse,
    HealthSnapshotRead,
    SystemHealthResponse,
)
from app.schemas.runtime_usage import (
    ProviderUsageResponse,
    ProviderUsageScrapeResult,
    RuntimeUsageResponse,
)
from app.services.openclaw.cron_status import (
    compute_job_status,
    fetch_cron_jobs,
)
from app.services.openclaw.system_health import (
    DEFAULT_HISTORY_WINDOW_HOURS,
    fetch_health,
    get_history,
)
from app.services.openclaw.runtime_activity import (
    HISTORY_FETCH_LIMIT,
    POLL_HISTORY_SESSIONS_MAX,
    fetch_recent_events,
)
from app.services.openclaw.runtime_usage import get_runtime_usage
from app.services.openclaw.usage_scrapers import get_provider_usage
from app.schemas.pagination import DefaultLimitOffsetPage
from app.services.openclaw.admin_service import GatewayAdminLifecycleService
from app.services.openclaw.session_service import GatewayTemplateSyncQuery

if TYPE_CHECKING:
    from fastapi_pagination.limit_offset import LimitOffsetPage
    from sqlmodel.ext.asyncio.session import AsyncSession

    from app.services.organizations import OrganizationContext


router = APIRouter(prefix="/gateways", tags=["gateways"])
SESSION_DEP = Depends(get_session)
AUTH_DEP = Depends(get_auth_context)
ORG_ADMIN_DEP = Depends(require_org_admin)
ORG_MEMBER_DEP = Depends(require_org_member)
INCLUDE_MAIN_QUERY = Query(default=True)
RESET_SESSIONS_QUERY = Query(default=False)
ROTATE_TOKENS_QUERY = Query(default=False)
FORCE_BOOTSTRAP_QUERY = Query(default=False)
OVERWRITE_QUERY = Query(default=False)
LEAD_ONLY_QUERY = Query(default=False)
BOARD_ID_QUERY = Query(default=None)
_RUNTIME_TYPE_REFERENCES = (UUID,)


def _template_sync_query(
    *,
    include_main: bool = INCLUDE_MAIN_QUERY,
    lead_only: bool = LEAD_ONLY_QUERY,
    reset_sessions: bool = RESET_SESSIONS_QUERY,
    rotate_tokens: bool = ROTATE_TOKENS_QUERY,
    force_bootstrap: bool = FORCE_BOOTSTRAP_QUERY,
    overwrite: bool = OVERWRITE_QUERY,
    board_id: UUID | None = BOARD_ID_QUERY,
) -> GatewayTemplateSyncQuery:
    return GatewayTemplateSyncQuery(
        include_main=include_main,
        lead_only=lead_only,
        reset_sessions=reset_sessions,
        rotate_tokens=rotate_tokens,
        force_bootstrap=force_bootstrap,
        overwrite=overwrite,
        board_id=board_id,
    )


SYNC_QUERY_DEP = Depends(_template_sync_query)


@router.get("", response_model=DefaultLimitOffsetPage[GatewayRead])
async def list_gateways(
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> LimitOffsetPage[GatewayRead]:
    """List gateways for the caller's organization."""
    statement = (
        Gateway.objects.filter_by(organization_id=ctx.organization.id)
        .order_by(col(Gateway.created_at).desc())
        .statement
    )

    return await paginate(session, statement)


@router.post("", response_model=GatewayRead)
async def create_gateway(
    payload: GatewayCreate,
    session: AsyncSession = SESSION_DEP,
    auth: AuthContext = AUTH_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> Gateway:
    """Create a gateway and provision or refresh its main agent."""
    service = GatewayAdminLifecycleService(session)
    await service.assert_gateway_runtime_compatible(
        url=payload.url,
        token=payload.token,
        allow_insecure_tls=payload.allow_insecure_tls,
        disable_device_pairing=payload.disable_device_pairing,
    )
    data = payload.model_dump()
    gateway_id = uuid4()
    data["id"] = gateway_id
    data["organization_id"] = ctx.organization.id
    gateway = await crud.create(session, Gateway, **data)
    await service.ensure_main_agent(gateway, auth, action="provision")
    return gateway


@router.get("/{gateway_id}", response_model=GatewayRead)
async def get_gateway(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> Gateway:
    """Return one gateway by id for the caller's organization."""
    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    return gateway


@router.patch("/{gateway_id}", response_model=GatewayRead)
async def update_gateway(
    gateway_id: UUID,
    payload: GatewayUpdate,
    session: AsyncSession = SESSION_DEP,
    auth: AuthContext = AUTH_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> Gateway:
    """Patch a gateway and refresh the main-agent provisioning state."""
    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    updates = payload.model_dump(exclude_unset=True)
    if (
        "url" in updates
        or "token" in updates
        or "allow_insecure_tls" in updates
        or "disable_device_pairing" in updates
    ):
        raw_next_url = updates.get("url", gateway.url)
        next_url = raw_next_url.strip() if isinstance(raw_next_url, str) else ""
        next_token = updates.get("token", gateway.token)
        next_allow_insecure_tls = bool(
            updates.get("allow_insecure_tls", gateway.allow_insecure_tls),
        )
        next_disable_device_pairing = bool(
            updates.get("disable_device_pairing", gateway.disable_device_pairing),
        )
        if next_url:
            await service.assert_gateway_runtime_compatible(
                url=next_url,
                token=next_token,
                allow_insecure_tls=next_allow_insecure_tls,
                disable_device_pairing=next_disable_device_pairing,
            )
    await crud.patch(session, gateway, updates)
    await service.ensure_main_agent(gateway, auth, action="update")
    return gateway


@router.post("/{gateway_id}/templates/sync", response_model=GatewayTemplatesSyncResult)
async def sync_gateway_templates(
    gateway_id: UUID,
    sync_query: GatewayTemplateSyncQuery = SYNC_QUERY_DEP,
    session: AsyncSession = SESSION_DEP,
    auth: AuthContext = AUTH_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> GatewayTemplatesSyncResult:
    """Sync templates for a gateway and optionally rotate runtime settings."""
    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    return await service.sync_templates(gateway, query=sync_query, auth=auth)


@router.get(
    "/{gateway_id}/agents/import-preview",
    response_model=GatewayAgentImportPreviewResponse,
)
async def preview_import_gateway_agents(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> GatewayAgentImportPreviewResponse:
    """Preview existing gateway runtime agents that can be imported."""
    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    return await service.preview_gateway_agent_import(gateway=gateway)


@router.post(
    "/{gateway_id}/agents/import",
    response_model=GatewayAgentImportResponse,
)
async def import_gateway_agents(
    gateway_id: UUID,
    payload: GatewayAgentImportRequest,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> GatewayAgentImportResponse:
    """Import selected existing gateway runtime agents into Pipeline."""
    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    return await service.import_gateway_agents(
        gateway=gateway,
        gateway_agent_ids=payload.gateway_agent_ids,
        reconcile_after_import=payload.reconcile_after_import,
        rotate_tokens=payload.rotate_tokens,
        reset_sessions=payload.reset_sessions,
        force_bootstrap=payload.force_bootstrap,
    )


@router.get(
    "/{gateway_id}/runtime-usage",
    response_model=RuntimeUsageResponse,
    summary="Gateway runtime usage",
    description=(
        "Return model usage, token counts, estimated spend, burn rate, and "
        "time-remaining predictions for the specified gateway.  "
        "Visible to all organisation members."
    ),
)
async def get_gateway_runtime_usage(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_MEMBER_DEP,
) -> RuntimeUsageResponse:
    """Aggregate runtime usage from the gateway's usage.cost / usage.status RPC methods."""
    from app.services.openclaw.gateway_rpc import GatewayConfig as GatewayClientConfig

    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    config = GatewayClientConfig(
        url=gateway.url,
        token=gateway.token,
        allow_insecure_tls=gateway.allow_insecure_tls,
        disable_device_pairing=gateway.disable_device_pairing,
    )
    account_key = gateway.name.lower().replace(" ", "-") if gateway.name else "default"
    return await get_runtime_usage(
        gateway_id=gateway.id,
        config=config,
        account_key=account_key,
    )


@router.get(
    "/{gateway_id}/provider-usage",
    response_model=ProviderUsageResponse,
    summary="Gateway provider-native usage (opt-in scraper)",
    description=(
        "Return provider-native subscription usage data scraped from the CLI "
        "(e.g. ``claude /usage``).  Returns an empty results list when "
        "USAGE_SCRAPER_ENABLED=false (the default).  "
        "Enable with USAGE_SCRAPER_ENABLED=true and ensure the required "
        "prerequisites (tmux, claude binary) are accessible from the backend process."
    ),
)
async def get_gateway_provider_usage(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_MEMBER_DEP,
) -> ProviderUsageResponse:
    """Scrape provider-native usage for the specified gateway (opt-in)."""
    service = GatewayAdminLifecycleService(session)
    await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    now = utcnow()

    if not settings.usage_scraper_enabled:
        return ProviderUsageResponse(
            gateway_id=gateway_id,
            generated_at=now,
            scraper_enabled=False,
            results=[],
        )

    enabled_providers = [
        p.strip()
        for p in settings.usage_scraper_providers.split(",")
        if p.strip()
    ]
    scrape_results = await get_provider_usage(
        gateway_id=str(gateway_id),
        enabled_providers=enabled_providers,
        tmux_socket=settings.usage_scraper_tmux_socket,
        include_raw=settings.usage_scraper_include_raw,
    )

    results = [
        ProviderUsageScrapeResult(
            provider=r.provider,
            source_name=r.source_name,
            scraped_at=r.scraped_at,
            fresh=r.fresh,
            freshness_ttl_seconds=r.freshness_ttl_seconds,
            current_pct=r.parsed.current_pct,
            remaining_ms=r.parsed.remaining_ms,
            remaining_label=r.parsed.remaining_label,
            weekly_messages_used=r.parsed.weekly_messages_used,
            weekly_messages_limit=r.parsed.weekly_messages_limit,
            weekly_tokens_used=r.parsed.weekly_tokens_used,
            weekly_cost_usd=r.parsed.weekly_cost_usd,
            raw_text=r.parsed.raw_text,
            error=r.error or r.parsed.error,
        )
        for r in scrape_results
    ]

    return ProviderUsageResponse(
        gateway_id=gateway_id,
        generated_at=now,
        scraper_enabled=True,
        results=results,
    )


@router.get(
    "/{gateway_id}/cron",
    response_model=CronStatusResponse,
    summary="Gateway cron job status",
    description="Return the list of cron jobs configured on the gateway with their last-run status.",
)
async def get_gateway_cron(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_MEMBER_DEP,
) -> CronStatusResponse:
    """Read cron job status from the gateway (read-only)."""
    from app.services.openclaw.gateway_rpc import GatewayConfig as GatewayClientConfig

    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    config = GatewayClientConfig(
        url=gateway.url,
        token=gateway.token,
        allow_insecure_tls=gateway.allow_insecure_tls,
        disable_device_pairing=gateway.disable_device_pairing,
    )
    jobs = await fetch_cron_jobs(config)
    return CronStatusResponse(
        gateway_id=gateway_id,
        generated_at=utcnow(),
        jobs=[
            CronJobRead(
                name=j.name,
                schedule=j.schedule,
                enabled=j.enabled,
                status=compute_job_status(j),
                last_run=j.last_run,
                next_run=j.next_run,
                last_duration_ms=j.last_duration_ms,
                last_error=j.last_error,
            )
            for j in jobs
        ],
    )


@router.get(
    "/{gateway_id}/health",
    response_model=SystemHealthResponse,
    summary="Gateway system health",
    description="Return current CPU, RAM, disk, and uptime stats plus a 24-hour history.",
)
async def get_gateway_health(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_MEMBER_DEP,
) -> SystemHealthResponse:
    """Read system health from the gateway and append to the rolling history."""
    from app.services.openclaw.gateway_rpc import GatewayConfig as GatewayClientConfig

    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    config = GatewayClientConfig(
        url=gateway.url,
        token=gateway.token,
        allow_insecure_tls=gateway.allow_insecure_tls,
        disable_device_pairing=gateway.disable_device_pairing,
    )
    snapshot = await fetch_health(str(gateway_id), config, record=True)
    history = get_history(str(gateway_id))

    def _snap_read(s) -> HealthSnapshotRead:
        return HealthSnapshotRead(
            recorded_at=s.recorded_at,
            cpu_pct=s.cpu_pct,
            memory_pct=s.memory_pct,
            memory_used_bytes=s.memory_used_bytes,
            memory_total_bytes=s.memory_total_bytes,
            disk_pct=s.disk_pct,
            disk_used_bytes=s.disk_used_bytes,
            disk_total_bytes=s.disk_total_bytes,
            uptime_seconds=s.uptime_seconds,
            load_avg_1m=s.load_avg_1m,
            load_avg_5m=s.load_avg_5m,
            load_avg_15m=s.load_avg_15m,
            hostname=s.hostname,
            platform=s.platform,
        )

    return SystemHealthResponse(
        gateway_id=gateway_id,
        generated_at=utcnow(),
        current=_snap_read(snapshot),
        history=[_snap_read(s) for s in history],
        history_window_hours=DEFAULT_HISTORY_WINDOW_HOURS,
    )


@router.get(
    "/{gateway_id}/runtime-activity",
    summary="Recent gateway runtime messages (REST snapshot)",
    description=(
        "Return the most recent messages from all active sessions on the gateway, "
        "normalised and redacted.  Use the streaming endpoint for a live feed."
    ),
)
async def get_gateway_runtime_activity(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_MEMBER_DEP,
) -> dict:
    """Non-streaming snapshot of recent gateway activity."""
    from app.services.openclaw.gateway_rpc import GatewayConfig as GatewayClientConfig

    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    config = GatewayClientConfig(
        url=gateway.url,
        token=gateway.token,
        allow_insecure_tls=gateway.allow_insecure_tls,
        disable_device_pairing=gateway.disable_device_pairing,
    )
    events = await fetch_recent_events(
        config,
        max_sessions=POLL_HISTORY_SESSIONS_MAX,
        history_limit=HISTORY_FETCH_LIMIT,
    )
    return {
        "gateway_id": str(gateway_id),
        "generated_at": utcnow().isoformat(),
        "events": [e.to_dict() for e in events],
    }


_RUNTIME_ACTIVITY_POLL_SECONDS = 4
_RUNTIME_ACTIVITY_SEEN_MAX = 1000


@router.get(
    "/{gateway_id}/runtime-activity/stream",
    summary="Live gateway runtime activity (SSE)",
    description=(
        "Stream normalised gateway session messages as server-sent events. "
        "Each event is of type ``runtime_message``.  The stream polls the "
        "gateway every few seconds and emits only new messages."
    ),
)
async def stream_gateway_runtime_activity(
    request: Request,
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_MEMBER_DEP,
) -> EventSourceResponse:
    """SSE stream of live gateway session messages."""
    from app.services.openclaw.gateway_rpc import GatewayConfig as GatewayClientConfig

    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    config = GatewayClientConfig(
        url=gateway.url,
        token=gateway.token,
        allow_insecure_tls=gateway.allow_insecure_tls,
        disable_device_pairing=gateway.disable_device_pairing,
    )

    async def event_generator():
        from collections import deque
        seen_ids: set[str] = set()
        seen_queue: deque[str] = deque()
        initial = True

        while True:
            if await request.is_disconnected():
                break

            events = await fetch_recent_events(
                config,
                since_ids=None if initial else seen_ids,
                max_sessions=POLL_HISTORY_SESSIONS_MAX,
                history_limit=HISTORY_FETCH_LIMIT,
            )
            initial = False

            for event in events:
                if event.event_id in seen_ids:
                    continue
                seen_ids.add(event.event_id)
                seen_queue.append(event.event_id)
                if len(seen_queue) > _RUNTIME_ACTIVITY_SEEN_MAX:
                    oldest = seen_queue.popleft()
                    seen_ids.discard(oldest)
                yield {
                    "event": "runtime_message",
                    "data": json.dumps(event.to_dict()),
                }

            await asyncio.sleep(_RUNTIME_ACTIVITY_POLL_SECONDS)

    return EventSourceResponse(event_generator(), ping=15)


@router.delete("/{gateway_id}", response_model=OkResponse)
async def delete_gateway(
    gateway_id: UUID,
    session: AsyncSession = SESSION_DEP,
    ctx: OrganizationContext = ORG_ADMIN_DEP,
) -> OkResponse:
    """Delete a gateway in the caller's organization."""
    service = GatewayAdminLifecycleService(session)
    gateway = await service.require_gateway(
        gateway_id=gateway_id,
        organization_id=ctx.organization.id,
    )
    main_agent = await service.find_main_agent(gateway)
    if main_agent is not None:
        await service.clear_agent_foreign_keys(agent_id=main_agent.id)
        await session.delete(main_agent)

    duplicate_main_agents = await Agent.objects.filter_by(
        gateway_id=gateway.id,
        board_id=None,
    ).all(session)
    for agent in duplicate_main_agents:
        if main_agent is not None and agent.id == main_agent.id:
            continue
        await service.clear_agent_foreign_keys(agent_id=agent.id)
        await session.delete(agent)

    # NOTE: The migration declares `ondelete="CASCADE"` for gateway_installed_skills.gateway_id,
    # but some backends/test environments (e.g. SQLite without FK pragma) may not
    # enforce cascades. Delete rows explicitly to guarantee cleanup semantics.
    installed_skills = await GatewayInstalledSkill.objects.filter_by(
        gateway_id=gateway.id,
    ).all(session)
    for installed_skill in installed_skills:
        await session.delete(installed_skill)

    await session.delete(gateway)
    await session.commit()
    return OkResponse()