"""Admin router: operational endpoints for application management.""" from datetime import datetime, timedelta, timezone from typing import Any from fastapi import APIRouter, Depends from ..db import get_conn, get_write_lock from ..dependencies import require_admin from ..services import models as models_service from ..services.models import mark_timed_out_video_jobs router = APIRouter(prefix="/admin", tags=["admin"]) @router.get("/stats") async def get_stats(_: dict = Depends(require_admin)) -> dict: """Return aggregate statistics: user counts and token counts.""" conn = get_conn() sql_user_count = "SELECT COUNT(*) FROM users" sql_user_counts = "SELECT role, COUNT(*) FROM users GROUP BY role ORDER BY role" sql_token_count = "SELECT COUNT(*) FROM refresh_tokens" sql_tokens_active = "SELECT COUNT(*) FROM refresh_tokens WHERE revoked = false AND expires_at > ?" now = datetime.now(timezone.utc) total_users_row = conn.execute(sql_user_count).fetchone() total_users = total_users_row[0] if total_users_row else 0 users_by_role = conn.execute(sql_user_counts).fetchall() total_tokens_row = conn.execute(sql_token_count).fetchone() total_tokens = total_tokens_row[0] if total_tokens_row else 0 active_tokens_row = conn.execute(sql_tokens_active, [now]).fetchone() active_tokens = active_tokens_row[0] if active_tokens_row else 0 return { "users": { "total": total_users, "by_role": {row[0]: row[1] for row in users_by_role}, }, "refresh_tokens": { "total": total_tokens, "active": active_tokens, "revoked_or_expired": total_tokens - active_tokens, }, } @router.get("/health/db") async def db_health(_: dict = Depends(require_admin)) -> dict: """Verify DuckDB is reachable.""" conn = get_conn() result_row = conn.execute("SELECT 1").fetchone() result = result_row[0] if result_row else 0 return {"status": "ok" if result == 1 else "error"} @router.post("/tokens/purge", status_code=200) async def purge_tokens(_: dict = Depends(require_admin)) -> dict: """Delete all expired or revoked refresh tokens. Returns count removed.""" conn = get_conn() lock = get_write_lock() now = datetime.now(timezone.utc) sql_count = "SELECT COUNT(*) FROM refresh_tokens" sql_delete = "DELETE FROM refresh_tokens WHERE revoked = true OR expires_at <= ?" async with lock: before_row = conn.execute(sql_count).fetchone() before = before_row[0] if before_row else 0 conn.execute(sql_delete, [now]) after_row = conn.execute(sql_count).fetchone() after = after_row[0] if after_row else 0 return {"deleted": before - after, "remaining": after} @router.get("/models/status") async def get_model_status(_: dict = Depends(require_admin)) -> dict[str, Any]: """Return model cache status: last update time and model count.""" conn = get_conn() return models_service.get_cache_status(conn) @router.get("/models") async def get_all_models(_: dict = Depends(require_admin)) -> list[dict[str, Any]]: """Return all cached models.""" conn = get_conn() return models_service.get_cached_models(conn) @router.post("/models/refresh", status_code=200) async def refresh_models( _: dict = Depends(require_admin), ) -> dict[str, str | int | None]: """Force a refresh of the model cache from OpenRouter.""" conn = get_conn() lock = get_write_lock() async with lock: count = await models_service.refresh_models_cache(conn) status = models_service.get_cache_status(conn) return { "status": "ok", "refreshed": count, "total_models": status.get("model_count"), "last_updated": status.get("last_updated"), } @router.get("/videos") async def admin_list_video_jobs(_: dict = Depends(require_admin)) -> list[dict[str, Any]]: """Return all video generation jobs across all users.""" conn = get_conn() rows = conn.execute( """ SELECT v.id, v.job_id, v.user_id, u.email, v.model_id, v.prompt, v.status, v.video_url, v.created_at, v.updated_at FROM generated_videos v LEFT JOIN users u ON v.user_id = u.id ORDER BY v.created_at DESC """ ).fetchall() return [ { "id": str(row[0]), "job_id": row[1], "user_id": str(row[2]), "user_email": row[3], "model_id": row[4], "prompt": row[5], "status": row[6], "video_url": row[7], "created_at": row[8].isoformat() if row[8] else None, "updated_at": row[9].isoformat() if row[9] else None, } for row in rows ] @router.post("/videos/{job_id}/cancel", status_code=200) async def admin_cancel_video_job(job_id: str, _: dict = Depends(require_admin)) -> dict[str, str]: """Mark a video job as 'cancelled'. Does not stop the provider job.""" conn = get_conn() lock = get_write_lock() now = datetime.now(timezone.utc) async with lock: conn.execute( "UPDATE generated_videos SET status = 'cancelled', updated_at = ? WHERE id = ?", [ now, job_id] ) return {"status": "ok", "job_id": job_id} @router.post("/videos/purge", status_code=200) async def admin_purge_video_jobs(_: dict = Depends(require_admin)) -> dict[str, Any]: """Delete all completed, failed, or cancelled jobs older than 30 days.""" conn = get_conn() lock = get_write_lock() thirty_days_ago = datetime.now( timezone.utc) - timedelta(days=30) sql_count = "SELECT COUNT(*) FROM generated_videos" sql_delete = """ DELETE FROM generated_videos WHERE status IN ('completed', 'failed', 'cancelled') AND updated_at < ? """ async with lock: before_row = conn.execute(sql_count).fetchone() before = before_row[0] if before_row else 0 conn.execute(sql_delete, [thirty_days_ago]) after_row = conn.execute(sql_count).fetchone() after = after_row[0] if after_row else 0 return {"deleted": before - after, "remaining": after} @router.post("/videos/timed-out", status_code=200) async def admin_mark_timed_out(_: dict = Depends(require_admin)) -> dict[str, int]: """Mark video jobs that have been in 'queued' or 'processing' status for too long as 'failed'.""" conn = get_conn() count = mark_timed_out_video_jobs(conn, timeout_minutes=120) return {"timed_out": count}