forked from sagnik/Project_Velocity
feat/#24 WebOS Completion (#25)
#24 WebOS Completion Co-authored-by: Sayan Datta <sayan@Sayans-MacBook-Air.local> Reviewed-on: sagnik/Project_Velocity#25
This commit is contained in:
529
backend/api/routes_admin_surface.py
Normal file
529
backend/api/routes_admin_surface.py
Normal file
@@ -0,0 +1,529 @@
|
||||
"""
|
||||
routes_admin_surface.py
|
||||
───────────────────────
|
||||
Admin Control Plane API
|
||||
|
||||
Roles: Only 'admin' or 'superadmin' may access these endpoints.
|
||||
|
||||
Endpoints:
|
||||
GET /admin-surface/health — system health overview
|
||||
GET /admin-surface/queues — queue depth snapshot
|
||||
GET /admin-surface/installs — surface session / install overview
|
||||
POST /admin-surface/actions — submit an admin action
|
||||
GET /admin-surface/actions — list admin action history
|
||||
GET /admin-surface/actions/{id} — get a specific action
|
||||
GET /admin-surface/logs — recent audit event log
|
||||
GET /admin-surface/templates — template catalog summary (admin view)
|
||||
POST /admin-surface/templates/{id}/publish — publish a template
|
||||
POST /admin-surface/templates/{id}/archive — archive a template
|
||||
GET /admin-surface/synthetic-jobs — list synthetic generation jobs
|
||||
POST /admin-surface/synthetic-jobs/{id}/cancel — cancel a synthetic job
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from backend.auth.dependencies import get_current_user
|
||||
|
||||
logger = logging.getLogger("velocity.admin_surface")
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
# ── RBAC guard ────────────────────────────────────────────────────────────────
|
||||
|
||||
ADMIN_ROLES = {"admin", "superadmin", "ADMIN", "SUPERADMIN"}
|
||||
|
||||
|
||||
def require_admin(user=Depends(get_current_user)):
|
||||
normalized_role = user.role.upper()
|
||||
if normalized_role not in {"ADMIN", "SUPERADMIN"}:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin access required.",
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def _pool(request: Request):
|
||||
pool = request.app.state.db_pool
|
||||
if pool is None:
|
||||
raise HTTPException(status_code=503, detail="Database unavailable.")
|
||||
return pool
|
||||
|
||||
|
||||
# ── Pydantic Models ───────────────────────────────────────────────────────────
|
||||
|
||||
VALID_ACTION_TYPES = {
|
||||
"user_create", "user_deactivate", "user_role_change",
|
||||
"tenant_config_update", "inventory_batch_approve", "inventory_batch_reject",
|
||||
"template_publish", "template_archive",
|
||||
"synthetic_job_trigger", "synthetic_job_cancel",
|
||||
"system_health_check", "queue_drain", "debug_event_export",
|
||||
"install_register", "install_deregister",
|
||||
}
|
||||
|
||||
|
||||
class AdminActionRequest(BaseModel):
|
||||
action_type: str
|
||||
target_type: str
|
||||
target_id: str
|
||||
payload: dict = Field(default_factory=dict)
|
||||
idempotency_key: Optional[str] = None
|
||||
|
||||
|
||||
# ── System Health ─────────────────────────────────────────────────────────────
|
||||
|
||||
@router.get("/health", summary="System health overview")
|
||||
async def get_health(
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
"""
|
||||
Returns an aggregated health snapshot covering DB pool, queue depths,
|
||||
and basic surface session counts.
|
||||
"""
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
# DB round-trip latency
|
||||
import time
|
||||
t0 = time.monotonic()
|
||||
await conn.fetchval("SELECT 1")
|
||||
db_latency_ms = round((time.monotonic() - t0) * 1000, 2)
|
||||
|
||||
# Pending jobs
|
||||
pending_transcriptions = await conn.fetchval(
|
||||
"SELECT COUNT(*) FROM edge_transcription_jobs WHERE status='pending'"
|
||||
)
|
||||
pending_synthetic_jobs = await conn.fetchval(
|
||||
"SELECT COUNT(*) FROM oracle_synthetic_generation_jobs WHERE status IN ('pending','running')"
|
||||
)
|
||||
pending_admin_actions = await conn.fetchval(
|
||||
"SELECT COUNT(*) FROM admin_action_events WHERE status='pending'"
|
||||
)
|
||||
pending_inventory_batches = await conn.fetchval(
|
||||
"SELECT COUNT(*) FROM inventory_import_batches WHERE status IN ('pending','validating','processing')"
|
||||
)
|
||||
|
||||
# Active surface sessions (last 30 min)
|
||||
active_sessions = await conn.fetchval(
|
||||
"SELECT COUNT(*) FROM surface_sessions WHERE last_active_at > NOW() - INTERVAL '30 minutes'"
|
||||
)
|
||||
|
||||
# Surface breakdown
|
||||
surface_breakdown = await conn.fetch(
|
||||
"""
|
||||
SELECT surface_type, COUNT(*) as count
|
||||
FROM surface_sessions
|
||||
WHERE last_active_at > NOW() - INTERVAL '30 minutes'
|
||||
GROUP BY surface_type
|
||||
"""
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "ok",
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
"database": {
|
||||
"connected": True,
|
||||
"latency_ms": db_latency_ms,
|
||||
},
|
||||
"queues": {
|
||||
"pending_transcriptions": pending_transcriptions,
|
||||
"pending_synthetic_jobs": pending_synthetic_jobs,
|
||||
"pending_admin_actions": pending_admin_actions,
|
||||
"pending_inventory_batches": pending_inventory_batches,
|
||||
},
|
||||
"active_sessions": {
|
||||
"total": active_sessions,
|
||||
"by_surface": {r["surface_type"]: r["count"] for r in surface_breakdown},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# ── Queue Visibility ──────────────────────────────────────────────────────────
|
||||
|
||||
@router.get("/queues", summary="Queue depth snapshot")
|
||||
async def get_queues(
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
transcription_queue = await conn.fetch(
|
||||
"""
|
||||
SELECT status, COUNT(*) as count
|
||||
FROM edge_transcription_jobs
|
||||
GROUP BY status ORDER BY status
|
||||
"""
|
||||
)
|
||||
synthetic_queue = await conn.fetch(
|
||||
"""
|
||||
SELECT status, COUNT(*) as count
|
||||
FROM oracle_synthetic_generation_jobs
|
||||
GROUP BY status ORDER BY status
|
||||
"""
|
||||
)
|
||||
inventory_queue = await conn.fetch(
|
||||
"""
|
||||
SELECT status, COUNT(*) as count
|
||||
FROM inventory_import_batches
|
||||
GROUP BY status ORDER BY status
|
||||
"""
|
||||
)
|
||||
admin_queue = await conn.fetch(
|
||||
"""
|
||||
SELECT status, COUNT(*) as count
|
||||
FROM admin_action_events
|
||||
GROUP BY status ORDER BY status
|
||||
"""
|
||||
)
|
||||
return {
|
||||
"transcription_jobs": {r["status"]: r["count"] for r in transcription_queue},
|
||||
"synthetic_jobs": {r["status"]: r["count"] for r in synthetic_queue},
|
||||
"inventory_batches": {r["status"]: r["count"] for r in inventory_queue},
|
||||
"admin_actions": {r["status"]: r["count"] for r in admin_queue},
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
|
||||
|
||||
# ── Install / Surface Overview ────────────────────────────────────────────────
|
||||
|
||||
@router.get("/installs", summary="Surface session and install overview")
|
||||
async def get_installs(
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
"""
|
||||
SELECT surface_type, app_version, COUNT(*) as session_count,
|
||||
MAX(last_active_at) as last_seen
|
||||
FROM surface_sessions
|
||||
GROUP BY surface_type, app_version
|
||||
ORDER BY surface_type, app_version
|
||||
"""
|
||||
)
|
||||
return {
|
||||
"installs": [dict(r) for r in rows],
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
|
||||
|
||||
# ── Admin Actions ─────────────────────────────────────────────────────────────
|
||||
|
||||
@router.post("/actions", status_code=status.HTTP_201_CREATED, summary="Submit an admin action")
|
||||
async def submit_action(
|
||||
request: Request,
|
||||
body: AdminActionRequest,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
"""
|
||||
Submit a bounded admin action. All actions are persisted with full audit trail.
|
||||
Supported action_types are enumerated in VALID_ACTION_TYPES.
|
||||
|
||||
Actions are not auto-executed — they transition to 'pending' and must be
|
||||
processed by the appropriate backend job or confirmed by a second admin.
|
||||
(This prevents destructive mass-actions from running unreviewed.)
|
||||
"""
|
||||
if body.action_type not in VALID_ACTION_TYPES:
|
||||
raise HTTPException(400, f"Invalid action_type. Valid: {sorted(VALID_ACTION_TYPES)}")
|
||||
|
||||
action_id = body.idempotency_key or str(uuid.uuid4())
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
try:
|
||||
row = await conn.fetchrow(
|
||||
"""
|
||||
INSERT INTO admin_action_events (
|
||||
tenant_id, action_id, action_type, target_type, target_id,
|
||||
requested_by, payload
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7::jsonb)
|
||||
RETURNING action_event_id, status, created_at
|
||||
""",
|
||||
admin.role, action_id, body.action_type, body.target_type,
|
||||
body.target_id, admin.user_id, json.dumps(body.payload),
|
||||
)
|
||||
except Exception as exc:
|
||||
if "unique" in str(exc).lower():
|
||||
raise HTTPException(409, "Action with this idempotency key already exists")
|
||||
raise
|
||||
|
||||
logger.info(
|
||||
"Admin action submitted: %s by %s → %s/%s",
|
||||
body.action_type, admin.user_id, body.target_type, body.target_id,
|
||||
)
|
||||
return {
|
||||
"action_event_id": str(row["action_event_id"]),
|
||||
"action_id": action_id,
|
||||
"status": row["status"],
|
||||
"created_at": str(row["created_at"]),
|
||||
}
|
||||
|
||||
|
||||
@router.get("/actions", summary="List admin action history")
|
||||
async def list_actions(
|
||||
request: Request,
|
||||
action_type: Optional[str] = Query(None),
|
||||
status_filter: Optional[str] = Query(None, alias="status"),
|
||||
limit: int = Query(50, ge=1, le=200),
|
||||
offset: int = Query(0, ge=0),
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
where = "WHERE tenant_id = $1"
|
||||
params: list[Any] = [admin.role]
|
||||
idx = 2
|
||||
|
||||
if action_type:
|
||||
where += f" AND action_type = ${idx}"; params.append(action_type); idx += 1
|
||||
if status_filter:
|
||||
where += f" AND status = ${idx}"; params.append(status_filter); idx += 1
|
||||
|
||||
async with pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
f"""
|
||||
SELECT action_event_id, action_id, action_type, target_type, target_id,
|
||||
requested_by, status, result_message, executed_at, created_at
|
||||
FROM admin_action_events
|
||||
{where}
|
||||
ORDER BY created_at DESC
|
||||
LIMIT ${idx} OFFSET ${idx+1}
|
||||
""",
|
||||
*params, limit, offset,
|
||||
)
|
||||
total = await conn.fetchval(
|
||||
f"SELECT COUNT(*) FROM admin_action_events {where}", *params,
|
||||
)
|
||||
return {"total": total, "limit": limit, "offset": offset, "actions": [dict(r) for r in rows]}
|
||||
|
||||
|
||||
@router.get("/actions/{action_event_id}", summary="Get a specific admin action")
|
||||
async def get_action(
|
||||
action_event_id: str,
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
row = await conn.fetchrow(
|
||||
"SELECT * FROM admin_action_events WHERE action_event_id=$1 AND tenant_id=$2",
|
||||
action_event_id, admin.role,
|
||||
)
|
||||
if not row:
|
||||
raise HTTPException(404, "Admin action not found")
|
||||
return dict(row)
|
||||
|
||||
|
||||
# ── Audit Log ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@router.get("/logs", summary="Recent Oracle audit events")
|
||||
async def get_audit_logs(
|
||||
request: Request,
|
||||
entity_type: Optional[str] = Query(None),
|
||||
limit: int = Query(100, ge=1, le=500),
|
||||
offset: int = Query(0, ge=0),
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
if entity_type:
|
||||
rows = await conn.fetch(
|
||||
"""
|
||||
SELECT audit_event_id, entity_type, entity_id, action, actor_id,
|
||||
actor_type, correlation_id, details, created_at
|
||||
FROM oracle_audit_events
|
||||
WHERE tenant_id=$1 AND entity_type=$2
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $3 OFFSET $4
|
||||
""",
|
||||
admin.role, entity_type, limit, offset,
|
||||
)
|
||||
else:
|
||||
rows = await conn.fetch(
|
||||
"""
|
||||
SELECT audit_event_id, entity_type, entity_id, action, actor_id,
|
||||
actor_type, correlation_id, details, created_at
|
||||
FROM oracle_audit_events
|
||||
WHERE tenant_id=$1
|
||||
ORDER BY created_at DESC
|
||||
LIMIT $2 OFFSET $3
|
||||
""",
|
||||
admin.role, limit, offset,
|
||||
)
|
||||
return {"logs": [dict(r) for r in rows]}
|
||||
|
||||
|
||||
# ── Template Administration ───────────────────────────────────────────────────
|
||||
|
||||
@router.get("/templates", summary="Template catalog admin view")
|
||||
async def get_templates_admin(
|
||||
request: Request,
|
||||
status_filter: Optional[str] = Query(None, alias="status"),
|
||||
limit: int = Query(50, ge=1, le=200),
|
||||
offset: int = Query(0, ge=0),
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
where = "WHERE tenant_id = $1"
|
||||
params: list[Any] = [admin.role]
|
||||
idx = 2
|
||||
|
||||
if status_filter:
|
||||
where += f" AND status = ${idx}"; params.append(status_filter); idx += 1
|
||||
|
||||
async with pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
f"""
|
||||
SELECT t.template_id, t.name, t.category, t.status, t.origin,
|
||||
t.version, t.use_count, t.chapter_id, t.subchapter_id,
|
||||
ch.name as chapter_name, sub.name as subchapter_name,
|
||||
t.created_at, t.updated_at
|
||||
FROM oracle_component_templates t
|
||||
LEFT JOIN oracle_template_chapters ch ON ch.chapter_id = t.chapter_id
|
||||
LEFT JOIN oracle_template_subchapters sub ON sub.subchapter_id = t.subchapter_id
|
||||
{where}
|
||||
ORDER BY t.updated_at DESC
|
||||
LIMIT ${idx} OFFSET ${idx+1}
|
||||
""",
|
||||
*params, limit, offset,
|
||||
)
|
||||
total = await conn.fetchval(
|
||||
f"SELECT COUNT(*) FROM oracle_component_templates {where}", *params,
|
||||
)
|
||||
return {"total": total, "limit": limit, "offset": offset, "templates": [dict(r) for r in rows]}
|
||||
|
||||
|
||||
@router.post("/templates/{template_id}/publish", summary="Publish a template")
|
||||
async def publish_template(
|
||||
template_id: str,
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
result = await conn.execute(
|
||||
"""
|
||||
UPDATE oracle_component_templates
|
||||
SET status='catalog_active', updated_at=NOW()
|
||||
WHERE template_id=$1 AND tenant_id=$2
|
||||
""",
|
||||
template_id, admin.role,
|
||||
)
|
||||
if result == "UPDATE 0":
|
||||
raise HTTPException(404, "Template not found")
|
||||
logger.info("Template %s published by admin %s", template_id, admin.user_id)
|
||||
return {"status": "published"}
|
||||
|
||||
|
||||
@router.post("/templates/{template_id}/archive", summary="Archive a template")
|
||||
async def archive_template(
|
||||
template_id: str,
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
result = await conn.execute(
|
||||
"""
|
||||
UPDATE oracle_component_templates
|
||||
SET status='archived', updated_at=NOW()
|
||||
WHERE template_id=$1 AND tenant_id=$2
|
||||
""",
|
||||
template_id, admin.role,
|
||||
)
|
||||
if result == "UPDATE 0":
|
||||
raise HTTPException(404, "Template not found")
|
||||
logger.info("Template %s archived by admin %s", template_id, admin.user_id)
|
||||
return {"status": "archived"}
|
||||
|
||||
|
||||
# ── Template Chapter Admin ────────────────────────────────────────────────────
|
||||
|
||||
@router.get("/template-chapters", summary="List template chapters (admin view)")
|
||||
async def list_chapters_admin(
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
"""
|
||||
SELECT ch.chapter_id, ch.name, ch.description, ch.sort_order, ch.is_active,
|
||||
COUNT(sub.subchapter_id) as subchapter_count
|
||||
FROM oracle_template_chapters ch
|
||||
LEFT JOIN oracle_template_subchapters sub ON sub.chapter_id = ch.chapter_id
|
||||
WHERE ch.tenant_id=$1
|
||||
GROUP BY ch.chapter_id
|
||||
ORDER BY ch.sort_order ASC
|
||||
""",
|
||||
admin.role,
|
||||
)
|
||||
return {"chapters": [dict(r) for r in rows]}
|
||||
|
||||
|
||||
# ── Synthetic Jobs Admin ──────────────────────────────────────────────────────
|
||||
|
||||
@router.get("/synthetic-jobs", summary="List synthetic generation jobs")
|
||||
async def list_synthetic_jobs(
|
||||
request: Request,
|
||||
status_filter: Optional[str] = Query(None, alias="status"),
|
||||
limit: int = Query(50, ge=1, le=200),
|
||||
offset: int = Query(0, ge=0),
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
if status_filter:
|
||||
rows = await conn.fetch(
|
||||
"""
|
||||
SELECT job_id, template_id, model, status, requested_count,
|
||||
accepted_count, created_by, started_at, completed_at, created_at
|
||||
FROM oracle_synthetic_generation_jobs
|
||||
WHERE tenant_id=$1 AND status=$2
|
||||
ORDER BY created_at DESC LIMIT $3 OFFSET $4
|
||||
""",
|
||||
admin.role, status_filter, limit, offset,
|
||||
)
|
||||
else:
|
||||
rows = await conn.fetch(
|
||||
"""
|
||||
SELECT job_id, template_id, model, status, requested_count,
|
||||
accepted_count, created_by, started_at, completed_at, created_at
|
||||
FROM oracle_synthetic_generation_jobs
|
||||
WHERE tenant_id=$1
|
||||
ORDER BY created_at DESC LIMIT $2 OFFSET $3
|
||||
""",
|
||||
admin.role, limit, offset,
|
||||
)
|
||||
return {"jobs": [dict(r) for r in rows]}
|
||||
|
||||
|
||||
@router.post("/synthetic-jobs/{job_id}/cancel", summary="Cancel a synthetic generation job")
|
||||
async def cancel_synthetic_job(
|
||||
job_id: str,
|
||||
request: Request,
|
||||
admin=Depends(require_admin),
|
||||
):
|
||||
pool = _pool(request)
|
||||
async with pool.acquire() as conn:
|
||||
result = await conn.execute(
|
||||
"""
|
||||
UPDATE oracle_synthetic_generation_jobs
|
||||
SET status='cancelled', updated_at=NOW()
|
||||
WHERE job_id=$1 AND tenant_id=$2 AND status IN ('pending','running')
|
||||
""",
|
||||
job_id, admin.role,
|
||||
)
|
||||
if result == "UPDATE 0":
|
||||
raise HTTPException(404, "Job not found or already in terminal state")
|
||||
return {"status": "cancelled"}
|
||||
Reference in New Issue
Block a user