Add admin panel, WebSocket support, and API versioning

Introduce an admin portal (React + Nginx), WebSocket routing, and
API versioning middleware with `/api/v1/` prefix deprecation.
Add master API key authentication, new Prometheus metrics for AI
token consumption and active WebSockets, and extend S3 config
with a public endpoint URL. Update test paths and fixtures to
align with the new routing structure.

app/config.py

@@ -12,6 +12,7 @@ class Settings(BaseSettings):
     # Application
     APP_NAME: str = "Imago"
     APP_VERSION: str = "1.0.0"
+    API_V1_SUNSET_DATE: str = ""
     DEBUG: bool = False
     SECRET_KEY: str = "changez-moi"
 
@@ -84,6 +85,7 @@ class Settings(BaseSettings):
     S3_BUCKET: str = ""
     S3_REGION: str = "us-east-1"
     S3_ENDPOINT_URL: str = ""               # vide = AWS, sinon MinIO/R2
+    S3_ENDPOINT_PUBLIC: str = ""            # URL utilisée pour les liens présignés (ex: http://localhost:9000)
     S3_ACCESS_KEY: str = ""
     S3_SECRET_KEY: str = ""
     S3_PREFIX: str = "imago"

app/dependencies/auth.py

@@ -62,6 +62,17 @@ async def verify_api_key(
             headers={"WWW-Authenticate": "Bearer"},
         )
 
+    # ── 1.5. Vérification Master Key (ADMIN_API_KEY) ───────
+    from app.config import settings
+    if settings.ADMIN_API_KEY and raw_key == settings.ADMIN_API_KEY:
+        # Retourne un client virtuel avec tous les droits
+        return APIClient(
+            id="admin-master",
+            name="Imago Master Admin",
+            scopes=["admin", "images:read", "images:write", "ai:use"],
+            plan="premium"
+        )
+
     # ── Lookup par hash ───────────────────────────────────────
     key_hash = hash_api_key(raw_key)
     result = await db.execute(

app/main.py

@@ -12,9 +12,10 @@ from slowapi.errors import RateLimitExceeded
 from app.config import settings
 from app.database import init_db
 from app.logging_config import configure_logging
-from app.routers import images_router, ai_router, auth_router, files_router
+from app.routers import images_router, ai_router, auth_router, files_router, ws_router, admin_router
 from app.middleware import limiter
 from app.middleware.logging_middleware import LoggingMiddleware
+from app.middleware.versioning import APIVersioningMiddleware
 from app.workers.redis_client import get_redis_pool, close_redis_pool
 
 # Configure le logging structuré dès l'import
@@ -144,13 +145,15 @@ class _FallbackArqPool:
 # ─────────────────────────────────────────────────────────────
 
 app = FastAPI(
-    title=settings.APP_NAME,
+    title=f"{settings.APP_NAME} API",
     version=settings.APP_VERSION,
     description="""
-## Imago
+## Imago API (v1)
 
 Backend de gestion d'images et fonctionnalités AI pour l'interface Shaarli.
 
+> **Note de versionnement :** L'API principale est désormais servie sous le préfixe `/api/v1/`. Les anciennes routes sans ce préfixe sont dépréciées.
+
 ### Fonctionnalités
 - 📸 **Upload et stockage d'images** avec génération de thumbnails
 - 🔍 **Extraction EXIF** automatique (appareil, GPS, paramètres de prise de vue)
@@ -182,6 +185,12 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+# ─────────────────────────────────────────────────────────────
+# Middleware de Versionnement API
+# ─────────────────────────────────────────────────────────────
+
+app.add_middleware(APIVersioningMiddleware)
+
 # ─────────────────────────────────────────────────────────────
 # Middleware Logging HTTP
 # ─────────────────────────────────────────────────────────────
@@ -216,12 +225,24 @@ if settings.STORAGE_BACKEND == "local":
     app.mount("/static/thumbnails", StaticFiles(directory=str(settings.thumbnails_path)), name="thumbnails")
 
 # ─────────────────────────────────────────────────────────────
-# Routers
+# Routers Versionnés (/api/v1/)
+# ─────────────────────────────────────────────────────────────
+from fastapi import APIRouter
+
+api_v1_router = APIRouter(prefix="/api/v1")
+api_v1_router.include_router(images_router)
+api_v1_router.include_router(ai_router)
+api_v1_router.include_router(auth_router)
+
+# Mount the versioned router on the app
+app.include_router(api_v1_router)
+
+# ─────────────────────────────────────────────────────────────
+# Routers Non-Versionnés (WebSockets, Fichiers, Santé, Metrics)
 # ─────────────────────────────────────────────────────────────
 
-app.include_router(images_router)
-app.include_router(ai_router)
-app.include_router(auth_router)
+app.include_router(ws_router)
+app.include_router(admin_router)
 
 
 # ─────────────────────────────────────────────────────────────
@@ -261,7 +282,14 @@ async def health_detailed(request: Request):
     """Endpoint de santé détaillé pour monitoring avancé."""
     checks = {}
 
-    # DB check
+    # 1. Backend check
+    checks["backend"] = {
+        "status": "ok",
+        "version": settings.APP_VERSION,
+        "uptime": "N/A" # Optionnel: ajouter un tracker d'uptime si nécessaire
+    }
+
+    # 2. Database check
     try:
         from app.database import AsyncSessionLocal
         from sqlalchemy import text
@@ -271,7 +299,7 @@ async def health_detailed(request: Request):
     except Exception as e:
         checks["database"] = {"status": "error", "error": str(e)}
 
-    # Redis check
+    # 3. Redis check
     redis = getattr(request.app.state, "redis", None)
     if redis:
         try:
@@ -282,25 +310,77 @@ async def health_detailed(request: Request):
     else:
         checks["redis"] = {"status": "not_configured"}
 
-    # ARQ check
+    # 4. Worker & Queue check (ARQ)
     arq_pool = getattr(request.app.state, "arq_pool", None)
     if arq_pool and not isinstance(arq_pool, _FallbackArqPool):
-        checks["arq"] = {"status": "ok"}
+        try:
+            # On essaye de voir le nombre de jobs en attente
+            queued_jobs = await arq_pool.zcard("arq:queue")
+            checks["queue"] = {
+                "status": "ok",
+                "pending_jobs": queued_jobs
+            }
+            
+            # Pour le worker, on vérifie la clé de health check déposée par ARQ
+            # Par défaut c'est <queue_name>:health-check
+            hc_key = f"{settings.S3_PREFIX or 'arq'}:queue:health-check"
+            # On ré-essaye avec la queue par défaut si le préfixe est différent
+            if not await redis.exists(hc_key):
+                hc_key = "arq:queue:health-check"
+            
+            worker_active = await redis.exists(hc_key)
+            checks["worker"] = {
+                "status": "ok" if worker_active else "error",
+                "active_workers": 1 if worker_active else 0,
+                "detail": "Worker opérationnel" if worker_active else "Aucun worker détecté"
+            }
+        except Exception as e:
+            checks["queue"] = {"status": "error", "error": str(e)}
+            checks["worker"] = {"status": "error", "error": str(e)}
     else:
-        checks["arq"] = {"status": "fallback"}
+        checks["queue"] = {"status": "fallback"}
+        checks["worker"] = {"status": "error", "detail": "ARQ non disponible"}
 
-    # OCR check
-    checks["ocr"] = {"status": "enabled" if settings.OCR_ENABLED else "disabled"}
+    # 5. MinIO / Storage check
+    from app.services.storage_backend import get_storage_backend, S3Storage
+    storage = get_storage_backend()
+    try:
+        if isinstance(storage, S3Storage):
+            # On vérifie la connexion réelle à MinIO en listant les objets (ou head bucket)
+            # Pour faire simple/rapide on vérifie juste si on peut accéder au bucket
+            # via l'interface du storage qui a déjà les sessions asynchrones
+            exists = await storage.exists(".health-check") # path arbitraire
+            checks["minio"] = {
+                "status": "ok",
+                "backend": "s3",
+                "bucket": settings.S3_BUCKET
+            }
+        else:
+            checks["minio"] = {
+                "status": "ok",
+                "backend": "local",
+                "path": settings.UPLOAD_DIR
+            }
+    except Exception as e:
+        checks["minio"] = {"status": "error", "error": str(e)}
 
-    # Storage check
-    checks["storage"] = {
-        "backend": settings.STORAGE_BACKEND,
-        "status": "ok",
+    # 6. OCR check (Tesseract)
+    checks["tesseract"] = {"status": "ok" if settings.OCR_ENABLED else "disabled"}
+
+    # 7. AI Provider check
+    ai_configured = (
+        (settings.AI_PROVIDER == "gemini" and bool(settings.GEMINI_API_KEY)) or
+        (settings.AI_PROVIDER == "openrouter" and bool(settings.OPENROUTER_API_KEY))
+    )
+    checks["ai"] = {
+        "status": "ok" if ai_configured else "warning",
+        "provider": settings.AI_PROVIDER,
+        "configured": ai_configured
     }
 
     overall = "healthy" if all(
         c.get("status") in ("ok", "enabled", "disabled", "not_configured", "fallback")
-        for c in checks.values()
+        for k, c in checks.items() if k != "ai" # On tolère AI warning
     ) else "degraded"
 
     return {

app/metrics.py

@@ -9,12 +9,20 @@ from prometheus_client import Counter, Histogram, Gauge
 hub_images_uploaded = Counter(
     "hub_images_uploaded_total",
     "Nombre total d'images uploadées",
-    ["client_plan"],
+    ["client_id"],
 )
 
 hub_images_deleted = Counter(
     "hub_images_deleted_total",
     "Nombre total d'images supprimées",
+    ["client_id"],
+)
+
+# ── AI ────────────────────────────────────────────────────────
+hub_ai_tokens_consumed = Counter(
+    "hub_ai_tokens_consumed_total",
+    "Nombre total de tokens consommés",
+    ["client_id", "model"],
 )
 
 # ── Pipeline ──────────────────────────────────────────────────
@@ -44,6 +52,12 @@ hub_storage_used_bytes = Gauge(
     ["client_id"],
 )
 
+# ── WebSockets ────────────────────────────────────────────────
+hub_active_websockets = Gauge(
+    "hub_active_websockets",
+    "Nombre de connexions WebSocket actives",
+)
+
 # ── ARQ ───────────────────────────────────────────────────────
 hub_arq_jobs_enqueued = Counter(
     "hub_arq_jobs_enqueued_total",

app/middleware/versioning.py

@@ -0,0 +1,39 @@
+"""
+Middleware de versionnement d'API.
+
+Ajoute l'en-tête `X-API-Version` aux réponses des routes sous `/api/v1/`
+et gère un éventuel avertissement de dépréciation via Sunset.
+"""
+import logging
+from typing import Awaitable, Callable
+
+from fastapi import Request, Response
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class APIVersioningMiddleware(BaseHTTPMiddleware):
+    async def dispatch(
+        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
+    ) -> Response:
+        response = await call_next(request)
+
+        # Appliquer X-API-Version seulement aux routes versionnées (ex: /api/v1/)
+        path = request.url.path
+        if path.startswith("/api/v1/"):
+            response.headers["X-API-Version"] = "v1"
+
+            # Si une date de fin de support est configurée, l'ajouter en en-tête
+            if settings.API_V1_SUNSET_DATE:
+                response.headers["Sunset"] = settings.API_V1_SUNSET_DATE
+                # Logger un avertissement si on l'utilise
+                logger.warning("api.v1_sunset_warning", extra={
+                    "client": request.client.host if request.client else "unknown",
+                    "path": path,
+                    "sunset": settings.API_V1_SUNSET_DATE
+                })
+
+        return response

app/routers/__init__.py

@@ -2,5 +2,7 @@ from app.routers.images import router as images_router
 from app.routers.ai import router as ai_router
 from app.routers.auth import router as auth_router
 from app.routers.files import router as files_router
+from app.routers.websocket import router as ws_router
+from app.routers.admin import router as admin_router
 
-__all__ = ["images_router", "ai_router", "auth_router", "files_router"]
+__all__ = ["images_router", "ai_router", "auth_router", "files_router", "ws_router", "admin_router"]

app/routers/admin.py

@@ -0,0 +1,165 @@
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+from typing import Dict, Any, List
+from pathlib import Path
+
+from app.database import get_db
+from app.models.image import Image
+from app.models.client import APIClient
+from app.dependencies.auth import require_scope
+
+router = APIRouter(prefix="/admin/api", tags=["Admin"])
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+ALLOWED_DOCS = {
+    "README.md": PROJECT_ROOT / "README.md",
+    "USER_GUIDE.md": PROJECT_ROOT / "docs" / "USER_GUIDE.md",
+    "ARCHITECTURE.md": PROJECT_ROOT / "docs" / "ARCHITECTURE.md",
+    "API_GUIDE.md": PROJECT_ROOT / "docs" / "API_GUIDE.md",
+    "SDK.md": PROJECT_ROOT / "docs" / "SDK.md",
+    "SHAARLI-INTEGRATION.md": PROJECT_ROOT / "docs" / "SHAARLI-INTEGRATION.md",
+    "WEBSOCKET.md": PROJECT_ROOT / "docs" / "WEBSOCKET.md",
+}
+
+@router.get("/stats")
+async def get_stats(
+    _: APIClient = Depends(require_scope("admin")),
+    db: AsyncSession = Depends(get_db)
+) -> Dict[str, Any]:
+    """Get global system statistics."""
+    # Count images
+    total_images_query = await db.execute(select(func.count(Image.id)))
+    total_images = total_images_query.scalar() or 0
+    
+    # Storage used (sum of file_size)
+    storage_query = await db.execute(select(func.sum(Image.file_size)))
+    storage_bytes = storage_query.scalar() or 0
+    
+    # Total AI tokens (sum of prompt + output tokens)
+    tokens_query = await db.execute(
+        select(func.sum(Image.ai_prompt_tokens) + func.sum(Image.ai_output_tokens))
+        .where(
+            Image.ai_prompt_tokens.isnot(None) | Image.ai_output_tokens.isnot(None)
+        )
+    )
+    total_tokens = tokens_query.scalar() or 0
+
+    # Total clients
+    clients_query = await db.execute(select(func.count(APIClient.id)))
+    total_clients = clients_query.scalar() or 0
+
+    return {
+        "total_images": total_images,
+        "total_storage_bytes": storage_bytes,
+        "total_ai_tokens": total_tokens,
+        "total_clients": total_clients
+    }
+
+@router.get("/clients")
+async def get_clients(
+    _: APIClient = Depends(require_scope("admin")),
+    db: AsyncSession = Depends(get_db)
+) -> List[Dict[str, Any]]:
+    """List all clients with metrics."""
+    result = await db.execute(select(APIClient).order_by(APIClient.created_at.desc()))
+    clients = result.scalars().all()
+    
+    return [
+        {
+            "id": c.id,
+            "name": c.name,
+            "is_active": c.is_active,
+            "is_premium": c.plan == "premium",
+            "storage_used": c.storage_used_bytes,
+            "storage_quota": c.quota_storage_mb,
+            "created_at": c.created_at
+        } for c in clients
+    ]
+
+@router.get("/queue/status")
+async def get_queue_status(
+    request: Request,
+    _: APIClient = Depends(require_scope("admin"))
+) -> Dict[str, Any]:
+    """Get ARQ queue status using the redis connection."""
+    redis = request.app.state.redis
+    
+    # Note: A real implementation would parse the ARQ keys here. 
+    # For now, we return basic statistics by probing redis directly with arq known queues.
+    # Count pending jobs in arq:queue
+    pending_count = await redis.llen("arq:queue") if hasattr(redis, "llen") else 0
+    # In ARQ, active jobs are harder to count without querying the worker sets, 
+    # so we return pending count.
+    
+    return {
+        "pending_jobs": pending_count,
+        "status": "active"
+    }
+
+@router.post("/clients/{client_id}/toggle")
+async def toggle_client(
+    client_id: str,
+    _: APIClient = Depends(require_scope("admin")),
+    db: AsyncSession = Depends(get_db)
+) -> Dict[str, Any]:
+    """Activate or deactivate a client."""
+    result = await db.execute(select(APIClient).where(APIClient.id == client_id))
+    client = result.scalar_one_or_none()
+    if not client:
+        raise HTTPException(status_code=404, detail="Client not found")
+        
+    client.is_active = not client.is_active
+    await db.commit()
+    
+    return {"id": client.id, "is_active": client.is_active}
+
+@router.post("/clients/{client_id}/reset-quota")
+async def reset_client_quota(
+    client_id: str,
+    _: APIClient = Depends(require_scope("admin")),
+    db: AsyncSession = Depends(get_db)
+) -> Dict[str, Any]:
+    """Reset the quota counters for a client."""
+    result = await db.execute(select(APIClient).where(APIClient.id == client_id))
+    client = result.scalar_one_or_none()
+    if not client:
+        raise HTTPException(status_code=404, detail="Client not found")
+        
+    client.storage_used_bytes = 0
+    await db.commit()
+    
+    return {"id": client.id, "storage_used": 0}
+
+@router.get("/docs")
+async def list_docs(
+    _: APIClient = Depends(require_scope("admin"))
+) -> List[Dict[str, str]]:
+    """List available documentation files."""
+    docs = []
+    for name, path in ALLOWED_DOCS.items():
+        if path.exists():
+            docs.append({"name": name, "title": name.replace(".md", "").replace("-", " ").title()})
+        else:
+            docs.append({"name": name, "title": name.replace(".md", "").replace("-", " ").title()}) # we might still want to list them, or omit
+    return docs
+
+@router.get("/docs/{filename}")
+async def get_doc(
+    filename: str,
+    _: APIClient = Depends(require_scope("admin"))
+) -> Dict[str, str]:
+    """Get the content of a specific documentation file."""
+    if filename not in ALLOWED_DOCS:
+        raise HTTPException(status_code=404, detail="Document not found or access denied")
+    
+    path = ALLOWED_DOCS[filename]
+    if not path.exists():
+        raise HTTPException(status_code=404, detail=f"Document file {filename} not found on server")
+        
+    try:
+        content = path.read_text(encoding="utf-8")
+        return {"filename": filename, "content": content}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error reading document: {str(e)}")
+

app/routers/ai.py

@@ -14,6 +14,7 @@ from app.services.scraper import fetch_page_content
 from app.services.ai_vision import summarize_url, draft_task
 from app.config import settings
 from app.middleware import limiter
+from app.metrics import hub_ai_tokens_consumed
 
 
 router = APIRouter(prefix="/ai", tags=["Intelligence Artificielle"])
@@ -62,7 +63,7 @@ async def summarize_link(
         language=body.language,
     )
 
-    return SummarizeResponse(
+    resp = SummarizeResponse(
         url=body.url,
         title=page.get("title"),
         summary=result["summary"],
@@ -70,6 +71,13 @@ async def summarize_link(
         model=result["model"],
     )
 
+    # Metrics
+    if result.get("prompt_tokens") or result.get("output_tokens"):
+        total = (result.get("prompt_tokens") or 0) + (result.get("output_tokens") or 0)
+        hub_ai_tokens_consumed.labels(client_id=client.id, model=result["model"]).inc(total)
+
+    return resp
+
 
 @router.post(
     "/draft-task",
@@ -99,4 +107,9 @@ async def generate_task(
     if not result.get("title"):
         raise HTTPException(status_code=500, detail="Échec de la génération de la tâche")
 
+    # Metrics
+    if result.get("prompt_tokens") or result.get("output_tokens"):
+        total = (result.get("prompt_tokens") or 0) + (result.get("output_tokens") or 0)
+        hub_ai_tokens_consumed.labels(client_id=client.id, model=result.get("model", "unknown")).inc(total)
+
     return DraftTaskResponse(**result)

app/routers/images.py

@@ -25,6 +25,9 @@ from app.schemas import (
 from app.services import storage
 from app.middleware import limiter, get_upload_rate_limit
 from app.workers.image_worker import QUEUE_STANDARD, QUEUE_PREMIUM
+from app.metrics import (
+    hub_images_uploaded, hub_images_deleted, hub_storage_used_bytes, hub_arq_jobs_enqueued
+)
 
 logger = logging.getLogger(__name__)
 
@@ -102,13 +105,14 @@ async def upload_image(
     await db.commit()
     await db.refresh(image)
 
-    # Enqueue dans ARQ (persistant, avec retry)
+    # Enqueue dans ARQ
     arq_pool = request.app.state.arq_pool
-    await arq_pool.enqueue_job(
-        "process_image_task",
-        image.id,
-        str(client.id)
-    )
+    await arq_pool.enqueue_job("process_image_task", image.id, str(client.id))
+
+    # Metrics
+    hub_images_uploaded.labels(client_id=client.id).inc()
+    hub_storage_used_bytes.labels(client_id=client.id).set(client.storage_used_bytes)
+    hub_arq_jobs_enqueued.labels(queue="default").inc()
 
     return UploadResponse(
         id=image.id,
@@ -190,6 +194,8 @@ async def list_images(
                 height=img.height,
                 uploaded_at=img.uploaded_at,
                 processing_status=img.processing_status,
+                client_id=img.client_id,
+                client_name=client.name,
                 ai_tags=img.ai_tags,
                 ai_description=img.ai_description,
                 thumbnail_path=img.thumbnail_path,
@@ -444,6 +450,10 @@ async def delete_image(
     await db.delete(image)
     await db.commit()
 
+    # Metrics
+    hub_images_deleted.labels(client_id=client.id).inc()
+    hub_storage_used_bytes.labels(client_id=client.id).set(client.storage_used_bytes)
+
     return DeleteResponse(deleted_id=image_id)
 
 

app/routers/websocket.py

@@ -0,0 +1,251 @@
+"""
+Router WebSocket — suivi temps réel du pipeline de traitement d'images.
+
+Endpoints :
+- WS /ws/pipeline/{image_id}?token=<api_key>  → événements d'un pipeline
+- WS /ws/admin/monitor?token=<admin_api_key>   → monitoring admin global
+"""
+import json
+import logging
+from typing import Any
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database import AsyncSessionLocal, get_db
+from app.dependencies.auth import hash_api_key
+from app.metrics import hub_active_websockets
+from app.models.client import APIClient
+from app.models.image import Image, ProcessingStatus
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["WebSocket"])
+
+
+# ─────────────────────────────────────────────────────────────
+# Helpers
+# ─────────────────────────────────────────────────────────────
+
+async def _authenticate_ws(websocket: WebSocket, db: AsyncSession) -> APIClient | None:
+    """
+    Authentifie une connexion WebSocket via le query param `token`.
+    Retourne le client ou None si invalide.
+    """
+    token = websocket.query_params.get("token")
+    if not token:
+        return None
+
+    # Vérification Master Key
+    from app.config import settings
+    if settings.ADMIN_API_KEY and token == settings.ADMIN_API_KEY:
+        return APIClient(
+            id="admin-master",
+            name="Imago Master Admin",
+            scopes=["admin", "images:read", "images:write", "ai:use"],
+            plan="premium",
+        )
+
+    key_hash = hash_api_key(token)
+    result = await db.execute(
+        select(APIClient).where(APIClient.api_key_hash == key_hash)
+    )
+    client = result.scalar_one_or_none()
+    if client and client.is_active:
+        return client
+    return None
+
+
+async def _get_image(image_id: int, db: AsyncSession) -> Image | None:
+    """Charge une image depuis la BDD."""
+    result = await db.execute(select(Image).where(Image.id == image_id))
+    return result.scalar_one_or_none()
+
+
+# ─────────────────────────────────────────────────────────────
+# WS /ws/pipeline/{image_id} — suivi d'un pipeline
+# ─────────────────────────────────────────────────────────────
+
+from fastapi import Depends
+
+@router.websocket("/ws/pipeline/{image_id}")
+async def ws_pipeline(
+    websocket: WebSocket,
+    image_id: int,
+    db: AsyncSession = Depends(get_db)
+) -> None:
+    """
+    WebSocket de suivi temps réel d'un pipeline image.
+
+    - Authentification via query param `token`
+    - Vérifie que l'image appartient au client
+    - Envoie le buffer de reconnexion puis les événements live
+    - Ferme après pipeline.done ou pipeline.error
+    """
+    # ── Authentification ──────────────────────────────────────
+    client = await _authenticate_ws(websocket, db)
+    if client is None:
+        await websocket.close(code=4001, reason="Token manquant ou invalide")
+        return
+
+    # ── Vérification propriété de l'image ─────────────────────
+    image = await _get_image(image_id, db)
+
+    if image is None:
+        await websocket.accept()
+        await websocket.close(code=4004, reason="Image introuvable")
+        return
+
+    # Admin peut voir toutes les images, sinon vérifier ownership
+    if not client.has_scope("admin") and image.client_id != client.id:
+        await websocket.close(code=4003, reason="Accès interdit")
+        return
+
+    # ── Accepter la connexion ─────────────────────────────────
+    await websocket.accept()
+    hub_active_websockets.inc()
+
+    try:
+        # ── Image déjà terminée → message synthétique ─────────
+        if image.processing_status == ProcessingStatus.DONE:
+            await websocket.send_json({
+                "event": "pipeline.done",
+                "image_id": image_id,
+                "status": "done",
+                "synthetic": True,
+            })
+            return
+
+        if image.processing_status == ProcessingStatus.ERROR:
+            await websocket.send_json({
+                "event": "pipeline.error",
+                "image_id": image_id,
+                "error": image.processing_error or "Erreur inconnue",
+                "synthetic": True,
+            })
+            return
+
+        # ── Récupérer le buffer de reconnexion depuis Redis ───
+        redis = getattr(websocket.app.state, "redis", None)
+        if redis is not None:
+            try:
+                buffer_key = f"pipeline:buffer:{image_id}"
+                buffered = await redis.lrange(buffer_key, 0, -1)
+                for raw_event in buffered:
+                    try:
+                        event_data = json.loads(raw_event)
+                        await websocket.send_json(event_data)
+                    except (json.JSONDecodeError, Exception):
+                        pass
+            except Exception as e:
+                logger.warning("ws.buffer_read_error", extra={"error": str(e)})
+
+        # ── S'abonner au channel Redis et écouter les événements
+        if redis is not None:
+            pubsub = redis.pubsub()
+            try:
+                await pubsub.subscribe(f"pipeline:{image_id}")
+
+                async for message in pubsub.listen():
+                    if message["type"] != "message":
+                        continue
+
+                    try:
+                        data = json.loads(message["data"])
+                    except (json.JSONDecodeError, TypeError):
+                        continue
+
+                    await websocket.send_json(data)
+
+                    # Fermer après pipeline.done ou pipeline.error
+                    event_type = data.get("event", "")
+                    if event_type in ("pipeline.done", "pipeline.error"):
+                        break
+            finally:
+                await pubsub.unsubscribe(f"pipeline:{image_id}")
+                await pubsub.close()
+        else:
+            # Pas de Redis — envoyer un message d'info et fermer
+            await websocket.send_json({
+                "event": "error",
+                "message": "Redis indisponible — utilisez le polling GET /images/{id}/status",
+            })
+
+    except WebSocketDisconnect:
+        logger.info("ws.client_disconnected", extra={
+            "image_id": image_id,
+            "client_id": client.id,
+        })
+    except Exception as e:
+        logger.error("ws.unexpected_error", extra={
+            "image_id": image_id,
+            "error": str(e),
+        })
+    finally:
+        hub_active_websockets.dec()
+
+
+# ─────────────────────────────────────────────────────────────
+# WS /ws/admin/monitor — monitoring admin global
+# ─────────────────────────────────────────────────────────────
+
+@router.websocket("/ws/admin/monitor")
+async def ws_admin_monitor(
+    websocket: WebSocket,
+    db: AsyncSession = Depends(get_db)
+) -> None:
+    """
+    WebSocket admin pour surveiller tous les pipelines en temps réel.
+
+    Nécessite le scope `admin`. Pousse un événement à chaque démarrage
+    ou fin de pipeline sur n'importe quelle image.
+    """
+    # ── Authentification ──────────────────────────────────────
+    client = await _authenticate_ws(websocket, db)
+    if client is None:
+        await websocket.close(code=4001, reason="Token manquant ou invalide")
+        return
+
+    if not client.has_scope("admin"):
+        await websocket.close(code=4003, reason="Scope admin requis")
+        return
+
+    # ── Accepter la connexion ─────────────────────────────────
+    await websocket.accept()
+    hub_active_websockets.inc()
+
+    try:
+        redis = getattr(websocket.app.state, "redis", None)
+        if redis is None:
+            await websocket.send_json({
+                "event": "error",
+                "message": "Redis indisponible",
+            })
+            return
+
+        pubsub = redis.pubsub()
+        try:
+            await pubsub.subscribe("pipeline:admin")
+
+            async for message in pubsub.listen():
+                if message["type"] != "message":
+                    continue
+
+                try:
+                    data = json.loads(message["data"])
+                except (json.JSONDecodeError, TypeError):
+                    continue
+
+                await websocket.send_json(data)
+
+        finally:
+            await pubsub.unsubscribe("pipeline:admin")
+            await pubsub.close()
+
+    except WebSocketDisconnect:
+        logger.info("ws.admin_disconnected", extra={"client_id": client.id})
+    except Exception as e:
+        logger.error("ws.admin_error", extra={"error": str(e)})
+    finally:
+        hub_active_websockets.dec()

app/schemas/__init__.py

@@ -79,6 +79,8 @@ class ImageBase(BaseModel):
     height: Optional[int] = None
     uploaded_at: Optional[datetime] = None
     processing_status: ProcessingStatus
+    client_id: str
+    client_name: Optional[str] = None
 
 
 class ImageSummary(ImageBase):
@@ -107,6 +109,8 @@ class ImageDetail(ImageBase):
             height=img.height,
             uploaded_at=img.uploaded_at,
             processing_status=img.processing_status,
+            client_id=img.client_id,
+            client_name=img.client.name if img.client else None,
             thumbnail_path=img.thumbnail_path,
             exif=ExifData(
                 camera=ExifCamera(

app/services/ai_vision.py

@@ -338,6 +338,8 @@ async def summarize_url(url: str, content: str, language: str = "français") ->
         "summary": "",
         "tags": [],
         "model": settings.AI_PROVIDER,
+        "prompt_tokens": None,
+        "output_tokens": None,
     }
 
     if not settings.AI_ENABLED:
@@ -370,6 +372,7 @@ Retourne UNIQUEMENT ce JSON :
             if parsed:
                 result["summary"] = parsed.get("summary", "")
                 result["tags"] = parsed.get("tags", [])
+                result["prompt_tokens"], result["output_tokens"] = response.get("usage")
 
     except Exception as e:
         logger.error("ai.summarize_url.error", extra={"error": str(e)})
@@ -385,6 +388,9 @@ async def draft_task(description: str, context: Optional[str], language: str = "
         "steps": [],
         "estimated_time": None,
         "priority": None,
+        "model": settings.AI_PROVIDER,
+        "prompt_tokens": None,
+        "output_tokens": None,
     }
 
     if not settings.AI_ENABLED:
@@ -416,6 +422,7 @@ Retourne UNIQUEMENT ce JSON :
             parsed = _extract_json(text)
             if parsed:
                 result.update(parsed)
+                result["prompt_tokens"], result["output_tokens"] = response.get("usage")
 
     except Exception as e:
         logger.error("ai.draft_task.error", extra={"error": str(e)})

app/services/pipeline.py

@@ -17,6 +17,10 @@ from app.models.image import Image, ProcessingStatus
 from app.services.exif_service import extract_exif
 from app.services.ocr_service import extract_text
 from app.services.ai_vision import analyze_image, extract_text_with_ai
+from app.metrics import (
+    hub_pipeline_duration, hub_pipeline_step_duration, hub_pipeline_errors,
+    hub_ai_tokens_consumed
+)
 import asyncio
 
 logger = logging.getLogger(__name__)
@@ -25,14 +29,33 @@ logger = logging.getLogger(__name__)
 async def _publish_event(
     redis: Any, image_id: int, event: str, data: dict | None = None
 ) -> None:
-    """Publie un événement sur le channel Redis pipeline:{image_id}."""
+    """
+    Publie un événement sur le channel Redis pipeline:{image_id}.
+
+    Stocke aussi les 10 derniers événements dans un buffer Redis
+    (clé pipeline:buffer:{image_id}, TTL 60s) pour la reconnexion WebSocket.
+    Publie sur pipeline:admin pour le monitoring admin.
+    """
     if redis is None:
         return
     try:
         payload = {"event": event, "image_id": image_id, "timestamp": time.time()}
         if data:
             payload["data"] = data
-        await redis.publish(f"pipeline:{image_id}", json.dumps(payload))
+        raw = json.dumps(payload)
+
+        # Publier sur le channel dédié à l'image
+        await redis.publish(f"pipeline:{image_id}", raw)
+
+        # Buffer de reconnexion (10 derniers événements, TTL 60s)
+        buffer_key = f"pipeline:buffer:{image_id}"
+        await redis.rpush(buffer_key, raw)
+        await redis.ltrim(buffer_key, -10, -1)
+        await redis.expire(buffer_key, 60)
+
+        # Publier sur le channel admin pour le monitoring global
+        if event in ("pipeline.started", "pipeline.done", "pipeline.error"):
+            await redis.publish("pipeline:admin", raw)
     except Exception:
         pass  # Pub/Sub non critique — ne doit pas bloquer le pipeline
 
@@ -61,6 +84,7 @@ async def process_image_pipeline(
     # ── Démarrage ─────────────────────────────────────────────
     image.processing_status = ProcessingStatus.PROCESSING
     image.processing_started_at = datetime.now(timezone.utc)
+    t_start = time.time()
     await db.commit()
     await db.refresh(image)
 
@@ -97,6 +121,8 @@ async def process_image_pipeline(
 
         await db.commit()
         elapsed = int((time.time() - t0) * 1000)
+        hub_pipeline_step_duration.labels(step="exif").observe(time.time() - t0)
+        
         logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "exif", "duration_ms": elapsed, "camera": image.exif_make})
         await _publish_event(redis, image_id, "step.completed", {
             "step": "exif", "duration_ms": elapsed, "camera": image.exif_make,
@@ -105,6 +131,7 @@ async def process_image_pipeline(
     except Exception as e:
         msg = f"EXIF : {str(e)}"
         errors.append(msg)
+        hub_pipeline_errors.labels(step="exif").inc()
         logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "exif", "error": str(e)})
 
     # ════════════════════════════════════════════════════════════
@@ -135,6 +162,8 @@ async def process_image_pipeline(
 
         await db.commit()
         elapsed = int((time.time() - t0) * 1000)
+        hub_pipeline_step_duration.labels(step="ocr").observe(time.time() - t0)
+
         logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text})
         await _publish_event(redis, image_id, "step.completed", {
             "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text,
@@ -143,6 +172,7 @@ async def process_image_pipeline(
     except Exception as e:
         msg = f"OCR : {str(e)}"
         errors.append(msg)
+        hub_pipeline_errors.labels(step="ocr").inc()
         logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ocr", "error": str(e)})
 
     # ════════════════════════════════════════════════════════════
@@ -166,6 +196,16 @@ async def process_image_pipeline(
 
         await db.commit()
         elapsed = int((time.time() - t0) * 1000)
+        hub_pipeline_step_duration.labels(step="ai").observe(time.time() - t0)
+        
+        # Token metrics
+        if image.ai_prompt_tokens or image.ai_output_tokens:
+            total_tokens = (image.ai_prompt_tokens or 0) + (image.ai_output_tokens or 0)
+            hub_ai_tokens_consumed.labels(
+                client_id=image.client_id, 
+                model=image.ai_model_used or "unknown"
+            ).inc(total_tokens)
+
         logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or [])})
         await _publish_event(redis, image_id, "step.completed", {
             "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or []),
@@ -174,6 +214,7 @@ async def process_image_pipeline(
     except Exception as e:
         msg = f"AI Vision : {str(e)}"
         errors.append(msg)
+        hub_pipeline_errors.labels(step="ai").inc()
         logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ai", "error": str(e)})
 
     # ════════════════════════════════════════════════════════════
@@ -193,6 +234,8 @@ async def process_image_pipeline(
         image.processing_error = None
 
     await db.commit()
+    hub_pipeline_duration.observe(time.time() - t_start)
+    
     logger.info("pipeline.completed", extra={
         "image_id": image_id,
         "status": image.processing_status.value,

app/services/storage_backend.py

@@ -127,6 +127,7 @@ class S3Storage(StorageBackend):
         endpoint_url: str = "",
         access_key: str = "",
         secret_key: str = "",
+        endpoint_public: str = "",
     ) -> None:
         self._bucket = bucket
         self._prefix = prefix.rstrip("/")
@@ -134,6 +135,7 @@ class S3Storage(StorageBackend):
         self._endpoint_url = endpoint_url or None
         self._access_key = access_key
         self._secret_key = secret_key
+        self._endpoint_public = endpoint_public or self._endpoint_url
 
     def _s3_key(self, path: str) -> str:
         if self._prefix:
@@ -178,7 +180,7 @@ class S3Storage(StorageBackend):
     async def get_signed_url(self, path: str, expires_in: int = 900) -> str:
         """Génère une URL présignée S3."""
         session = self._get_session()
-        async with session.client("s3", endpoint_url=self._endpoint_url) as client:
+        async with session.client("s3", endpoint_url=self._endpoint_public) as client:
             url = await client.generate_presigned_url(
                 "get_object",
                 Params={"Bucket": self._bucket, "Key": self._s3_key(path)},
@@ -238,6 +240,7 @@ def get_storage_backend() -> StorageBackend:
             endpoint_url=settings.S3_ENDPOINT_URL,
             access_key=settings.S3_ACCESS_KEY,
             secret_key=settings.S3_SECRET_KEY,
+            endpoint_public=settings.S3_ENDPOINT_PUBLIC,
         )
     else:
         _backend = LocalStorage(