feat: add webhook endpoint for receiving central-webhook payload on e…

…vent

src/backend/app/central/central_schemas.py

@@ -27,7 +27,7 @@
 from pydantic.functional_validators import field_validator, model_validator
 
 from app.config import HttpUrlStr, decrypt_value, encrypt_value
-from app.db.enums import EntityState
+from app.db.enums import EntityState, OdkWebhookEvents
 
 
 class ODKCentral(BaseModel):
@@ -299,3 +299,20 @@ def append_status_emoji(cls, value: str, info: ValidationInfo) -> str:
     def integer_status_to_string(cls, value: EntityState) -> str:
         """Convert integer status to string for ODK Entity data."""
         return str(value.value)
+
+
+class OdkCentralWebhookRequest(BaseModel):
+    """The POST data from the central webhook service."""
+
+    type: OdkWebhookEvents
+    # NOTE we cannot use UUID validation, as Central often passes uuid as 'uuid:xxx-xxx'
+    id: str
+    # NOTE do not use EntityPropertyDict or similar to allow more flexible parsing
+    # submission.create provides an XML string as the 'data'
+    data: dict
+
+
+class OdkEntitiesUpdate(BaseModel):
+    """A small base model to update the OdkEntity status field only."""
+
+    status: str  # this must be the str representation of the db enum

src/backend/app/db/enums.py

@@ -268,3 +268,11 @@ class GeomStatus(StrEnum, Enum):
 
     NEW = "NEW"
     BAD = "BAD"
+
+
+class OdkWebhookEvents(StrEnum, Enum):
+    """Types of events received from ODK Central webhook."""
+
+    UPDATE_ENTITY = "entity.update.version"
+    NEW_SUBMISSION = "submission.create"
+    REVIEW_SUBMISSION = "submission.update"

src/backend/app/db/models.py

@@ -64,6 +64,7 @@
 
 # Avoid cyclical dependencies when only type checking
 if TYPE_CHECKING:
+    from app.central.central_schemas import OdkEntitiesUpdate
     from app.organisations.organisation_schemas import (
         OrganisationIn,
         OrganisationUpdate,
@@ -308,7 +309,8 @@ async def create(
             SET
                 role = EXCLUDED.role,
                 mapping_level = EXCLUDED.mapping_level,
-                name = EXCLUDED.name
+                name = EXCLUDED.name,
+                api_key = EXCLUDED.api_key
         """
 
         sql = f"""
@@ -1535,6 +1537,34 @@ async def upsert(
 
         return bool(result)
 
+    @classmethod
+    async def update(
+        cls, db: Connection, entity_uuid: str, entity_update: "OdkEntitiesUpdate"
+    ) -> bool:
+        """Update the entity value in the FMTM db."""
+        model_dump = dump_and_check_model(entity_update)
+        placeholders = [f"{key} = %({key})s" for key in model_dump.keys()]
+        sql = f"""
+            UPDATE odk_entities
+            SET {", ".join(placeholders)}
+            WHERE entity_id = %(entity_uuid)s
+            RETURNING entity_id;
+        """
+
+        async with db.cursor() as cur:
+            await cur.execute(
+                sql,
+                {"entity_uuid": entity_uuid, **model_dump},
+            )
+            success = await cur.fetchone()
+
+        if not success:
+            msg = f"Failed to update entity with UUID: {entity_uuid}"
+            log.error(msg)
+            return False
+
+        return True
+
 
 class DbBackgroundTask(BaseModel):
     """Table background_tasks.

src/backend/app/integrations/integration_crud.py

@@ -19,11 +19,25 @@
 
 from secrets import token_urlsafe
 
+from fastapi.exceptions import HTTPException
+from fastapi.responses import Response
 from loguru import logger as log
 from psycopg import Connection
 from psycopg.rows import class_row
 
-from app.db.models import DbUser
+# from app.central.central_crud import update_entity_mapping_status
+from app.central.central_schemas import (
+    OdkCentralWebhookRequest,
+    OdkEntitiesUpdate,
+    # EntityMappingStatusIn,
+)
+from app.db.enums import (
+    EntityState,
+    HTTPStatus,
+    OdkWebhookEvents,
+    # ReviewStateEnum,
+)
+from app.db.models import DbOdkEntities, DbUser
 
 
 async def generate_api_token(
@@ -48,3 +62,84 @@ async def generate_api_token(
             raise ValueError(msg)
 
     return db_user.api_key
+
+
+async def update_entity_status_in_fmtm(
+    db: Connection,
+    odk_event: OdkCentralWebhookRequest,
+):
+    """Update the status for an Entity in the FMTM db."""
+    log.debug(f"Webhook called with event ({odk_event.type.value})")
+
+    if odk_event.type == OdkWebhookEvents.UPDATE_ENTITY:
+        # insert state into db
+        new_state = odk_event.data.get("status")
+
+        if new_state is None:
+            log.warning(f"Missing entity state in webhook event: {odk_event.data}")
+            return HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST,
+                content="Missing entity state property",
+            )
+
+        try:
+            new_state_int = int(new_state)
+            # the string name is inserted in the db
+            new_entity_state = EntityState(new_state_int).name
+        except (ValueError, TypeError):
+            log.warning(
+                f"Invalid entity state '{new_state}' in webhook event: {odk_event.data}"
+            )
+            return HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST,
+                content="Invalid entity state",
+            )
+
+        log.debug(
+            f"Updating entity ({str(odk_event.id)}) status "
+            f"in FMTM db to ({new_entity_state})"
+        )
+        update_success = await DbOdkEntities.update(
+            db,
+            str(odk_event.id),
+            OdkEntitiesUpdate(status=new_entity_state),
+        )
+        if not update_success:
+            raise HTTPException(
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+                detail=f"Error updating entity with UUID ({str(odk_event.id)})",
+            )
+        return Response(status_code=HTTPStatus.OK)
+
+
+# async def update_entity_status_in_odk(
+#     odk_event: OdkCentralWebhookRequest,
+# ):
+#     # FIXME here we don't have the project id and odkid to submit the updates!
+#     # FIXME perhaps this needs an update to the webhook, to include the
+#     # related entity details, so we can extract the project id, and then
+#     # get the related ODK credentials?
+#     # Else we need another workaround
+
+#     review_state = odk_event.data.get("reviewState")
+#     if review_state not in [ReviewStateEnum.HASISSUES, ReviewStateEnum.REJECTED]:
+#         log.debug(f"Submission ({odk_event.id}) reviewed and marked 'approved'")
+#         return Response(status_code=HTTPStatus.OK)
+
+#     new_entity_label = f"Task {odk_event.data.get("task_id")} "
+#     new_entity_label += f"Feature {odk_event.data.get("osm_id")}"
+
+#     # We parse as EntityMappingStatusIn to ensure the status
+#     # emoji is appended to the label
+#     entity_update = EntityMappingStatusIn(
+#         entity_id=str(odk_event.id),
+#         status=EntityState.MARKED_BAD,
+#         label=new_entity_label,
+#     )
+#     return await update_entity_mapping_status(
+#         project.odk_credentials,
+#         project.odkid,
+#         entity_update.entity_id,
+#         entity_update.label,
+#         entity_update.status,
+#     )

src/backend/app/integrations/integration_routes.py

@@ -31,19 +31,22 @@
 )
 from fastapi.exceptions import HTTPException
 from fastapi.responses import JSONResponse
+from loguru import logger as log
 from psycopg import Connection
 
 from app.auth.roles import super_admin
-from app.central.central_crud import update_entity_mapping_status
-from app.central.central_schemas import EntityMappingStatus, EntityMappingStatusIn
+from app.central.central_schemas import (
+    OdkCentralWebhookRequest,
+)
 from app.db.database import db_conn
-from app.db.enums import HTTPStatus
-from app.db.models import DbProject, DbUser
+from app.db.enums import HTTPStatus, OdkWebhookEvents
+from app.db.models import DbUser
 from app.integrations.integration_crud import (
     generate_api_token,
+    update_entity_status_in_fmtm,
+    # update_entity_status_in_odk,
 )
 from app.integrations.integration_deps import valid_api_token
-from app.projects.project_deps import get_project
 
 router = APIRouter(
     prefix="/integrations",
@@ -52,12 +55,12 @@
 )
 
 
-@router.get("/api-token")
-async def get_api_token(
+@router.get("/api-key")
+async def get_api_key(
     current_user: Annotated[DbUser, Depends(super_admin)],
     db: Annotated[Connection, Depends(db_conn)],
 ):
-    """Generate and return a new API token.
+    """Generate and return a new API key.
 
     This can only be accessed once, and is regenerated on
     each call to this endpoint.
@@ -67,32 +70,54 @@ async def get_api_token(
     NOTE currently requires super admin permission.
     """
     try:
-        api_key = await generate_api_token(db, current_user.id)
+        api_token = await generate_api_token(db, current_user.id)
     except ValueError as e:
         raise HTTPException(
             status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
             detail=str(e),
         ) from e
     return JSONResponse(
         status_code=HTTPStatus.OK,
-        content={"api_key": api_key},
+        content={"api_token": api_token},
     )
 
 
 @router.post(
     "/webhooks/entity-status",
-    response_model=EntityMappingStatus,
+    # response_model=EntityMappingStatus,
 )
-async def update_entity_status(
+async def update_entity_status_from_webhook(
+    db: Annotated[Connection, Depends(db_conn)],
     current_user: Annotated[DbUser, Depends(valid_api_token)],
-    project: Annotated[DbProject, Depends(get_project)],
-    entity_details: EntityMappingStatusIn,
+    odk_event: OdkCentralWebhookRequest,
 ):
-    """Update the status for an Entity."""
-    return await update_entity_mapping_status(
-        project.odk_credentials,
-        project.odkid,
-        entity_details.entity_id,
-        entity_details.label,
-        entity_details.status,
-    )
+    """ODK Central webhook triggers.
+
+    These are required to trigger the replication to users via electric-sql.
+
+    TODO perhaps these should be separated out, as the review action
+    does not require a db connection.
+    """
+    log.debug(f"Webhook called with event ({odk_event.type.value})")
+
+    if odk_event.type == OdkWebhookEvents.UPDATE_ENTITY:
+        # insert state into db
+        await update_entity_status_in_fmtm(db, odk_event)
+
+    elif odk_event.type == OdkWebhookEvents.REVIEW_SUBMISSION:
+        # update entity status in odk
+        # await update_entity_status_in_odk(db, odk_event)
+        log.warning(
+            "The handling of submission reviews via webhook is not implemented yet."
+        )
+
+    elif odk_event.type == OdkWebhookEvents.NEW_SUBMISSION:
+        # unsupported for now
+        log.debug("The handling of new submissions via webhook is not implemented yet.")
+
+    else:
+        msg = (
+            f"Webhook was called for an unsupported event type ({odk_event.type.value})"
+        )
+        log.warning(msg)
+        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=msg)