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/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/ogc_api_process.py

@@ -4,6 +4,7 @@
 
 import requests
 from app.auth import exchange_token, get_current_user_claims
+from app.config.settings import settings
 from fastapi import Response
 from loguru import logger
 
@@ -50,7 +51,7 @@ class OGCAPIProcessPlatform(BaseProcessingPlatform):
     }
 
     status_mapping = {
-        StatusCode.ACCEPTED: ProcessingStatusEnum.CREATED,
+        StatusCode.ACCEPTED: ProcessingStatusEnum.QUEUED,
         StatusCode.RUNNING: ProcessingStatusEnum.RUNNING,
         StatusCode.DISMISSED: ProcessingStatusEnum.CANCELED,
         StatusCode.SUCCESSFUL: ProcessingStatusEnum.FINISHED,
@@ -210,9 +211,17 @@ async def execute_job(
         if user_claims.get("email"):
             properties["email"] = user_claims["email"]
 
+        base_url = settings.dispatch_api_public_url.rstrip("/")
+        subscriber = {
+            "successUri": f"{base_url}/unit_jobs/success",
+            "inProgressUri": f"{base_url}/unit_jobs/inProgress",
+            "failedUri": f"{base_url}/unit_jobs/failed",
+        }
+
         data = {
             "inputs": parameters,
             "properties": properties,
+            "subscriber": subscriber,
         }
 
         content = api_client.execute_simple(

app/routers/unit_jobs.py

@@ -1,5 +1,5 @@
 from typing import Annotated
-from fastapi import Body, APIRouter, Depends, status
+from fastapi import Body, APIRouter, Depends, Request, status
 from loguru import logger
 from sqlalchemy.orm import Session
 
@@ -12,7 +12,26 @@
     JobNotFoundException,
 )
 from app.middleware.error_handling import get_dispatcher_error_response
-from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum
+from http_apis_problems_detail import ErrorDetail
+from http_apis_problems_detail.fastapi import (
+    PROBLEM_RESPONSES,
+    BusinessRuleViolationException,
+    ServerErrorException,
+    NotFoundException,
+    ProblemRegistryException
+)
+from stage_events_client.models import (
+    BaseCloudEvent,
+    CompletedCloudEvent,
+    DismissedCloudEvent,
+    FailedCloudEvent,
+    PipedCloudEvent,
+    PreparedCloudEvent,
+    StagedCloudEvent,
+    SubmittedCloudEvent,
+)
+from app.schemas.callbacks import STAGE_TO_APEX_STATUS
+from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum, ProcessingStatusEnum
 from app.schemas.unit_job import (
     BaseJobRequest,
     ProcessingJob,
@@ -24,8 +43,10 @@
     delete_processing_job,
     get_processing_job_by_user_id,
     get_processing_job_results,
+    get_job_by_platform_or_id,
+    update_job_status_by_id
 )
-
+from app.services.processing_state_machine import ProcessingStateMachine
 from stac_pydantic import Collection
 
 # from app.auth import get_current_user
@@ -272,3 +293,189 @@ async def delete_job(
             message="An error occurred while deleting the processing job.",
             details={"error": str(e)}
         )
+
+
+def _extract_job_identifier_from_cloudevent(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 _handle_callback_status_update(
+    db: Session,
+    cloud_event: BaseCloudEvent,
+    target_status: ProcessingStatusEnum,
+    request: Request,
+) -> None:
+    """Internal helper to update job status from callback CloudEvents."""
+    job_identifier = _extract_job_identifier_from_cloudevent(cloud_event) or ""
+    endpoint_name = request.url.path.split("/")[-1]
+
+    logger.info(f"Callback '{endpoint_name}' received for job '{job_identifier}'")
+
+    try:
+        job = get_job_by_platform_or_id(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",
+                    )
+                ]
+            )
+
+        if not ProcessingStateMachine.can_transition(job.status, target_status):
+            raise BusinessRuleViolationException(
+                request=request,
+                errors=[
+                    ErrorDetail(
+                        pointer="subject",
+                        detail=(
+                            f"Job {job_identifier} can not transition from status "
+                            f"{job.status} to {target_status}"
+                        ),
+                    )
+                ]
+            )
+
+        update_job_status_by_id(db, job.id, target_status)
+        job.status = target_status
+        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(
+    "/unit_jobs/success",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    tags=["Unit Jobs"],
+    summary="URIs for success callbacks for jobs.",
+    description="The callback URL will be invoked after a job succeeded its execution.\n",
+    operation_id="onSuccessJob",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        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_success_job(
+    request: Request,
+    body: PipedCloudEvent | CompletedCloudEvent,
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for success callbacks for jobs."""
+    _handle_callback_status_update(
+        db, body, ProcessingStatusEnum.FINISHED, request
+    )
+
+
+@router.post(
+    "/unit_jobs/inProgress",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    tags=["Unit Jobs"],
+    summary="URIs for progress callbacks for jobs.",
+    description=(
+        "The callback URL will be invoked while a job is changing "
+        "its status during its execution.\n"
+    ),
+    operation_id="onProgressJob",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        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_progress_job(
+    request: Request,
+    body: (
+        SubmittedCloudEvent
+        | DismissedCloudEvent
+        | PreparedCloudEvent
+        | StagedCloudEvent
+    ),
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for progress callbacks for jobs."""
+    target_status = STAGE_TO_APEX_STATUS.get(
+        body.type, ProcessingStatusEnum.RUNNING
+    )
+    _handle_callback_status_update(db, body, target_status, request)
+
+
+@router.post(
+    "/unit_jobs/failed",
+    response_model=None,
+    status_code=status.HTTP_200_OK,
+    tags=["Unit Jobs"],
+    summary="URIs for failed callbacks for jobs.",
+    description="The callback URL will be invoked after a job failed its execution.\n",
+    operation_id="onFailedJob",
+    responses={
+        status.HTTP_200_OK: {"description": "Job status updated successfully."},
+        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_failed_job(
+    request: Request,
+    body: FailedCloudEvent,
+    db: Session = Depends(get_db),
+) -> None:
+    """URIs for failed callbacks for jobs."""
+    _handle_callback_status_update(
+        db, body, ProcessingStatusEnum.FAILED, request
+    )