feat: add event-driven job status callbacks for OGC API Processes execution

app/config/settings.py

@@ -24,6 +24,10 @@ class Settings(BaseSettings):
     cors_allowed_origins: str = Field(
         default="", json_schema_extra={"env": "CORS_ALLOWED_ORIGINS"}
     )
+    dispatch_api_public_url: str = Field(
+        default="http://localhost:8000",
+        json_schema_extra={"env": "DISPATCH_API_PUBLIC_URL"},
+    )
 
     model_config = SettingsConfigDict(
         env_file=".env",

app/main.py

@@ -4,6 +4,7 @@
 from app.middleware.correlation_id import add_correlation_id
 from app.middleware.error_handling import register_exception_handlers
 from app.platforms.dispatcher import load_processing_platforms
+from app.platforms.implementations.gep.router import router as gep_callbacks_router
 from app.services.tiles.base import load_grids
 from app.config.logger import setup_logging
 from app.config.settings import settings
@@ -59,3 +60,5 @@ async def root():
 app.include_router(health.router)
 app.include_router(parameters.router)
 app.include_router(statistics.router)
+
+app.include_router(gep_callbacks_router)

app/middleware/error_handling.py

@@ -1,6 +1,10 @@
 import traceback
 from typing import Any
 
+from http_apis_problems_detail.fastapi import (
+    ProblemRegistryException,
+    problem_registry_exception_handler,
+)
 from fastapi import Request, status
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
@@ -83,6 +87,7 @@ def register_exception_handlers(app):
     Call this in main.py after creating the FastAPI() instance.
     """
 
+    app.add_exception_handler(ProblemRegistryException, problem_registry_exception_handler)
     app.add_exception_handler(DispatcherException, dispatch_exception_handler)
     app.add_exception_handler(RequestValidationError, validation_exception_handler)
     app.add_exception_handler(Exception, generic_exception_handler)

app/platforms/implementations/gep/__init__.py

@@ -0,0 +1 @@
+"""GEP stage-event callback integration for OGC API Processes."""

app/platforms/implementations/gep/auth.py

@@ -0,0 +1,92 @@
+"""Authenticate GEP callback service accounts independently of APEx user login."""
+
+from functools import lru_cache
+from typing import Annotated
+
+import jwt
+from fastapi import Depends, Request
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from http_apis_problems_detail import ErrorDetail
+from http_apis_problems_detail.fastapi import ForbiddenException, UnauthorizedException
+from jwt import PyJWKClient
+
+from app.platforms.implementations.gep.settings import get_callback_backends
+
+callback_bearer = HTTPBearer(auto_error=False, scheme_name="GepCallbackBearer")
+
+
+@lru_cache(maxsize=32)
+def _callback_jwks_client(issuer: str) -> PyJWKClient:
+    # Called only after matching a configured issuer, never an arbitrary token URL.
+    # Cache the JWKS rather than individual keys indefinitely, allowing key rotation.
+    return PyJWKClient(
+        f"{issuer.rstrip('/')}/protocol/openid-connect/certs",
+        cache_jwk_set=True,
+        lifespan=300,
+        timeout=10,
+    )
+
+
+def verify_callback_token(
+    request: Request,
+    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(callback_bearer)],
+) -> set[str]:
+    """Return the backend endpoints this verified service account may update."""
+    unauthorized = UnauthorizedException(
+        request=request,
+        errors=[ErrorDetail(
+            header="Authorization", detail="Callback Bearer Authorization token not set or invalid."
+        )],
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    if creds is None or creds.scheme.lower() != "bearer":
+        raise unauthorized
+
+    # Configuration errors are server errors, not invalid client credentials.
+    configured_backends = get_callback_backends()
+    try:
+        issuer = jwt.decode(
+            creds.credentials, options={"verify_signature": False}
+        ).get("iss")
+        candidates = {
+            endpoint: cfg for endpoint, cfg in configured_backends.items()
+            if isinstance(issuer, str) and issuer
+            and cfg.callback_issuer == issuer
+            and cfg.callback_audience and cfg.callback_clients
+        }
+        if not candidates:
+            raise unauthorized
+
+        key = _callback_jwks_client(issuer).get_signing_key_from_jwt(creds.credentials).key
+        allowed: set[str] = set()
+        valid_audience = False
+        # Require an audience match; reject any other JWT validation failure.
+        for endpoint, cfg in candidates.items():
+            try:
+                claims = jwt.decode(
+                    creds.credentials,
+                    key,
+                    algorithms=["RS256"],
+                    issuer=issuer,
+                    audience=cfg.callback_audience,
+                    options={"require": ["iss", "aud", "exp"]},
+                )
+            except jwt.InvalidAudienceError:
+                continue
+            valid_audience = True
+            client = claims.get("azp")
+            if isinstance(client, str) and client in cfg.callback_clients:
+                allowed.add(endpoint.rstrip("/"))
+        if not valid_audience:
+            raise unauthorized
+    except (jwt.PyJWTError, ValueError, TypeError, OverflowError) as exc:
+        raise unauthorized from exc
+
+    if not allowed:
+        raise ForbiddenException(
+            request=request,
+            errors=[ErrorDetail(
+                header="Authorization", detail="Callback client is not allowed."
+            )],
+        )
+    return allowed

app/platforms/implementations/gep/callbacks.py

@@ -0,0 +1,49 @@
+"""GEP event interpretation and exact platform-job correlation."""
+
+from sqlalchemy.orm import Session
+from stage_events_client.models import BaseCloudEvent
+
+from app.database.models.processing_job import ProcessingJobRecord
+from app.schemas.enum import ProcessTypeEnum, ProcessingStatusEnum
+
+STAGE_TO_APEX_STATUS: dict[str, ProcessingStatusEnum] = {
+    "submitted": ProcessingStatusEnum.QUEUED,
+    "prepared": ProcessingStatusEnum.RUNNING,
+    "staged": ProcessingStatusEnum.RUNNING,
+    "piped": ProcessingStatusEnum.FINISHED,
+    "completed": ProcessingStatusEnum.FINISHED,
+    "failed": ProcessingStatusEnum.FAILED,
+    "dismissed": ProcessingStatusEnum.CANCELED,
+}
+
+
+def extract_job_identifier(cloud_event: BaseCloudEvent) -> str | None:
+    """Extract job_identifier strictly from CloudEvent subject metadata."""
+    # CloudEvent subject pattern: "<namespace>:<workflow_uid>:<workflow_name>"
+    subject = getattr(cloud_event, "subject", None)
+    if not subject:
+        return None
+
+    parts = subject.split(":")
+    if len(parts) == 3:
+        return f"{parts[0]}:{parts[2]}"
+    return subject
+
+
+def get_gep_job(database: Session, job_identifier: str) -> ProcessingJobRecord | None:
+    """Never interpret a GEP identifier as an APEx ID or use suffix matching.
+
+    Subjects carry no trusted backend identity. If multiple OGC jobs have
+    overlapping identifiers, reject the event instead of picking the first job.
+    """
+    matches = (
+        database.query(ProcessingJobRecord)
+        .filter(
+            ProcessingJobRecord.label == ProcessTypeEnum.OGC_API_PROCESS,
+            ProcessingJobRecord.platform_job_id == job_identifier,
+        )
+        .all()
+    )
+    if len(matches) > 1:
+        raise ValueError("Ambiguous GEP job identifier")
+    return matches[0] if matches else None

app/platforms/implementations/gep/router.py

@@ -0,0 +1,209 @@
+"""Receive GEP stage CloudEvents, not arbitrary OGC callback payloads."""
+
+import json
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request, status
+from loguru import logger
+from sqlalchemy.orm import Session
+from stage_events_client.models import BaseCloudEvent
+from http_apis_problems_detail import ErrorDetail
+from http_apis_problems_detail.fastapi import (
+    PROBLEM_RESPONSES, BusinessRuleViolationException, ServerErrorException,
+    NotFoundException, ProblemRegistryException, ForbiddenException,
+)
+
+from app.platforms.implementations.gep.auth import verify_callback_token
+from app.database.db import get_db
+from app.schemas.enum import ProcessingStatusEnum
+from app.services.job_status import apply_job_status, InvalidJobTransition
+from app.platforms.implementations.gep.callbacks import (
+    STAGE_TO_APEX_STATUS, extract_job_identifier, get_gep_job,
+)
+from app.platforms.implementations.gep.schemas import (
+    ProgressEvent, SuccessEvent, FailureEvent,
+)
+from app.platforms.implementations.gep.subscriber import CALLBACK_PREFIX
+
+router = APIRouter(prefix=CALLBACK_PREFIX, tags=["GEP callbacks"])
+
+
+def _handle_callback_status_update(
+    db: Session,
+    cloud_event: BaseCloudEvent,
+    target_status: ProcessingStatusEnum,
+    request: Request,
+    caller: set[str],
+) -> None:
+    """Internal helper to update job status from callback CloudEvents."""
+    job_identifier = extract_job_identifier(cloud_event) or ""
+    endpoint_name = request.url.path.split("/")[-1]
+
+    logger.info(f"Callback '{endpoint_name}' received for job '{job_identifier}'")
+
+    try:
+        job = get_gep_job(db, job_identifier)
+        if not job:
+            if endpoint_name == "inProgress":
+                # An early inProgress callback may arrive before initial job commit finishes;
+                # acknowledge gracefully without returning 404.
+                logger.debug(
+                    f"Job '{job_identifier}' is not registered in database yet; "
+                    "acknowledging in progress"
+                )
+                return
+
+            raise NotFoundException(
+                request=request,
+                errors=[
+                    ErrorDetail(
+                        pointer="subject",
+                        detail=f"Job {job_identifier} does not exist in the Job registry",
+                    )
+                ]
+            )
+
+        # Authorize against persisted dispatch metadata, not CloudEvent claims.
+        endpoint = json.loads(job.service)["endpoint"].rstrip("/")
+        if endpoint not in caller:
+            raise ForbiddenException(
+                request=request,
+                errors=[ErrorDetail(
+                    pointer="subject",
+                    detail="Callback caller is not authorized to update this job.",
+                )],
+            )
+
+        try:
+            apply_job_status(db, job, target_status)
+        except InvalidJobTransition as exc:
+            raise BusinessRuleViolationException(
+                request=request,
+                errors=[ErrorDetail(pointer="subject", detail=str(exc))],
+            ) from exc
+        logger.info(
+            f"Successfully updated status of job '{job_identifier}' (id={job.id}) "
+            f"to {target_status}"
+        )
+    except ProblemRegistryException:
+        raise
+    except Exception as e:
+        logger.error(f"Error handling {endpoint_name} callback for job {job_identifier}: {e}")
+        raise ServerErrorException(
+            request=request,
+            errors=[
+                ErrorDetail(
+                    parameter="job_id",
+                    detail=(
+                        f"An error occurred when updating status of Job {job_identifier}: {e}"
+                    ),
+                )
+            ]
+        )
+
+
+@router.post(
+    "/success",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    summary="URIs for success callbacks for jobs.",
+    description=(
+        "Accepts GEP stage CloudEvents. The callback URL will be invoked "
+        "after a job succeeded its execution.\n"
+    ),
+    operation_id="onGepJobSuccess",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        status.HTTP_401_UNAUTHORIZED: PROBLEM_RESPONSES[status.HTTP_401_UNAUTHORIZED],
+        status.HTTP_403_FORBIDDEN: PROBLEM_RESPONSES[status.HTTP_403_FORBIDDEN],
+        status.HTTP_400_BAD_REQUEST: PROBLEM_RESPONSES[status.HTTP_400_BAD_REQUEST],
+        status.HTTP_404_NOT_FOUND: PROBLEM_RESPONSES[status.HTTP_404_NOT_FOUND],
+        status.HTTP_422_UNPROCESSABLE_CONTENT: (
+            PROBLEM_RESPONSES[status.HTTP_422_UNPROCESSABLE_CONTENT]
+        ),
+        status.HTTP_500_INTERNAL_SERVER_ERROR: (
+            PROBLEM_RESPONSES[status.HTTP_500_INTERNAL_SERVER_ERROR]
+        ),
+    },
+)
+def on_gep_job_success(
+    request: Request,
+    body: SuccessEvent,
+    caller: Annotated[set[str], Depends(verify_callback_token)],
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for success callbacks for jobs."""
+    _handle_callback_status_update(
+        db, body, ProcessingStatusEnum.FINISHED, request, caller
+    )
+
+
+@router.post(
+    "/inProgress",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    summary="URIs for progress callbacks for jobs.",
+    description=(
+        "Accepts GEP stage CloudEvents. The callback URL will be invoked while a job is changing "
+        "its status during its execution.\n"
+    ),
+    operation_id="onGepJobProgress",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        status.HTTP_401_UNAUTHORIZED: PROBLEM_RESPONSES[status.HTTP_401_UNAUTHORIZED],
+        status.HTTP_403_FORBIDDEN: PROBLEM_RESPONSES[status.HTTP_403_FORBIDDEN],
+        status.HTTP_400_BAD_REQUEST: PROBLEM_RESPONSES[status.HTTP_400_BAD_REQUEST],
+        status.HTTP_404_NOT_FOUND: PROBLEM_RESPONSES[status.HTTP_404_NOT_FOUND],
+        status.HTTP_422_UNPROCESSABLE_CONTENT: (
+            PROBLEM_RESPONSES[status.HTTP_422_UNPROCESSABLE_CONTENT]
+        ),
+        status.HTTP_500_INTERNAL_SERVER_ERROR: (
+            PROBLEM_RESPONSES[status.HTTP_500_INTERNAL_SERVER_ERROR]
+        ),
+    },
+)
+def on_gep_job_progress(
+    request: Request,
+    body: ProgressEvent,
+    caller: Annotated[set[str], Depends(verify_callback_token)],
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for progress callbacks for jobs."""
+    target_status = STAGE_TO_APEX_STATUS[body.type]
+    _handle_callback_status_update(db, body, target_status, request, caller)
+
+
+@router.post(
+    "/failed",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    summary="URIs for failed callbacks for jobs.",
+    description=(
+        "Accepts GEP stage CloudEvents. The callback URL will be invoked "
+        "after a job failed its execution.\n"
+    ),
+    operation_id="onGepJobFailed",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        status.HTTP_401_UNAUTHORIZED: PROBLEM_RESPONSES[status.HTTP_401_UNAUTHORIZED],
+        status.HTTP_403_FORBIDDEN: PROBLEM_RESPONSES[status.HTTP_403_FORBIDDEN],
+        status.HTTP_400_BAD_REQUEST: PROBLEM_RESPONSES[status.HTTP_400_BAD_REQUEST],
+        status.HTTP_404_NOT_FOUND: PROBLEM_RESPONSES[status.HTTP_404_NOT_FOUND],
+        status.HTTP_422_UNPROCESSABLE_CONTENT: (
+            PROBLEM_RESPONSES[status.HTTP_422_UNPROCESSABLE_CONTENT]
+        ),
+        status.HTTP_500_INTERNAL_SERVER_ERROR: (
+            PROBLEM_RESPONSES[status.HTTP_500_INTERNAL_SERVER_ERROR]
+        ),
+    },
+)
+def on_gep_job_failed(
+    request: Request,
+    body: FailureEvent,
+    caller: Annotated[set[str], Depends(verify_callback_token)],
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for failed callbacks for jobs."""
+    _handle_callback_status_update(
+        db, body, ProcessingStatusEnum.FAILED, request, caller
+    )