✨(ingestion) handle webhooks to archive offers

env.d/ingestion-example

@@ -1,2 +1,5 @@
 TALENTSOFT_CLIENT_ID=
 TALENTSOFT_CLIENT_SECRET=
+
+WEB_BASE_URL=
+WEB_API_KEY=

env.d/web-example

@@ -38,6 +38,8 @@ WEB_ALBERT_MODEL=openweight-large
 WEB_OCR_API_KEY=verysecretkey
 WEB_OCR_BASE_URL=http://localhost:8001
 
+WEB_INGESTION_API_KEY=verysecretkey
+
 WEB_TALENTSOFT_CLIENT_ID=
 WEB_TALENTSOFT_CLIENT_SECRET=
 WEB_TALENTSOFT_BASE_URL=

src/ingestion/api/config.py

@@ -14,6 +14,9 @@ class Settings(BaseSettings):
     talentsoft_client_id: str | None = None
     talentsoft_client_secret: str | None = None
 
+    web_base_url: str | None = None
+    web_api_key: str | None = None
+
 
 class TestSettings(Settings):
     model_config = SettingsConfigDict(env_file=None)
@@ -22,6 +25,9 @@ class TestSettings(Settings):
     sentry_profiles_sample_rate: float | None = 0.0
     sentry_traces_sample_rate: float | None = 0.0
 
+    web_base_url: str | None = None
+    web_api_key: str | None = None
+
 
 def get_settings() -> Settings:
     """Get settings based on environment."""

src/ingestion/api/dependencies.py

@@ -0,0 +1,25 @@
+from collections.abc import AsyncGenerator
+
+import httpx
+from fastapi import Depends, HTTPException
+
+from api.config import Settings, get_settings
+from application.use_cases.archive_offer import ArchiveOfferUseCase
+
+
+async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
+    async with httpx.AsyncClient() as client:
+        yield client
+
+
+def get_archive_offer_use_case(
+    settings: Settings = Depends(get_settings),
+    client: httpx.AsyncClient = Depends(get_http_client),
+) -> ArchiveOfferUseCase:
+    if not settings.web_base_url or not settings.web_api_key:
+        raise HTTPException(status_code=500, detail="Web service not configured")
+    return ArchiveOfferUseCase(
+        client=client,
+        web_base_url=settings.web_base_url,
+        web_api_key=settings.web_api_key,
+    )

src/ingestion/api/routes.py

@@ -1,8 +1,15 @@
 import logging
 
 from fastapi import APIRouter, Depends, Request
+from pydantic import ValidationError
 
+from api.dependencies import get_archive_offer_use_case
 from api.talentsoft import verify_talentsoft_signature
+from application.use_cases.archive_offer import ArchiveOfferUseCase
+from presentation.dtos.talentsoft_webhook import (
+    TalentsoftWebhookPayload,
+    should_archive,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -17,7 +24,28 @@ def health():
 @public_router.post(
     "/webhooks/talentsoft", dependencies=[Depends(verify_talentsoft_signature)]
 )
-async def talentsoft_webhook(request: Request):
+async def talentsoft_webhook(
+    request: Request,
+    use_case: ArchiveOfferUseCase = Depends(get_archive_offer_use_case),
+):
     body = await request.body()
     logger.info("TalentSoft webhook received", extra={"body": body.decode()})
+
+    if not body:
+        return {"status": "ok"}
+
+    try:
+        payload = TalentsoftWebhookPayload.model_validate_json(body)
+    except ValidationError:
+        logger.warning("Unrecognised TalentSoft webhook payload, ignoring")
+        return {"status": "ok"}
+
+    if not should_archive(payload):
+        return {"status": "ok"}
+
+    if not payload.reference:
+        logger.warning("Archive event received without reference, skipping")
+        return {"status": "ok"}
+
+    await use_case.execute(payload.reference)
     return {"status": "ok"}

src/ingestion/application/use_cases/archive_offer.py

@@ -0,0 +1,21 @@
+import httpx
+
+
+class ArchiveOfferUseCase:
+    def __init__(
+        self,
+        client: httpx.AsyncClient,
+        web_base_url: str,
+        web_api_key: str,
+    ) -> None:
+        self._client = client
+        self._web_base_url = web_base_url
+        self._web_api_key = web_api_key
+
+    async def execute(self, reference: str) -> None:
+        url = f"{self._web_base_url}/api/offers/{reference}/archive"
+        response = await self._client.post(
+            url,
+            headers={"Authorization": f"Api-Key {self._web_api_key}"},
+        )
+        response.raise_for_status()

src/ingestion/presentation/dtos/talentsoft_webhook.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel, ConfigDict, Field
+
+_TS_ARCHIVED = "_TS_Archived"
+
+
+class TalentsoftWebhookPayload(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    event_type: str
+    reference: str | None = None
+    status_id: str | None = Field(None, alias="statusId")
+
+
+def should_archive(payload: TalentsoftWebhookPayload) -> bool:
+    if payload.event_type == "vacancy_deleted":
+        return True
+    return payload.event_type == "vacancy_status" and payload.status_id == _TS_ARCHIVED

src/ingestion/pyproject.toml

@@ -37,7 +37,7 @@ url = "https://pypi.org/simple"
 ingestion = { path = ".", editable = true }
 
 [tool.setuptools.packages.find]
-include = ["api*"]
+include = ["api*", "application*", "presentation*"]
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 python_files = [
@@ -76,6 +76,9 @@ select = [
 "api/routes.py" = [
     "B008", # Function call in argument defaults (acceptable for FastAPI)
 ]
+"api/dependencies.py" = [
+    "B008", # Function call in argument defaults (acceptable for FastAPI)
+]
 "**/tests/*" = [
     "S101", # use of assert
     "S105", # hardcoded passwords (acceptable for test fixtures)

src/ingestion/tests/integration/conftest.py

@@ -0,0 +1,116 @@
+import base64
+import hashlib
+import hmac
+import json
+import time
+import urllib.parse
+
+import pytest
+from fastapi.testclient import TestClient
+from httpx import Response
+
+from api.main import create_app
+
+TALENTSOFT_CLIENT_ID = "test_client_id"
+TALENTSOFT_CLIENT_SECRET = "test_client_secret"
+WEB_BASE_URL = "https://web.example.com"
+WEB_API_KEY = "test-web-api-key"
+WEBHOOK_PATH = "/webhooks/talentsoft"
+
+
+@pytest.fixture
+def test_client(monkeypatch):
+    monkeypatch.setenv("TESTING", "true")
+    monkeypatch.delenv("TALENTSOFT_CLIENT_ID", raising=False)
+    monkeypatch.delenv("TALENTSOFT_CLIENT_SECRET", raising=False)
+    app = create_app()
+    return TestClient(app)
+
+
+@pytest.fixture
+def talentsoft_client(monkeypatch):
+    monkeypatch.setenv("TESTING", "true")
+    monkeypatch.setenv("TALENTSOFT_CLIENT_ID", TALENTSOFT_CLIENT_ID)
+    monkeypatch.setenv("TALENTSOFT_CLIENT_SECRET", TALENTSOFT_CLIENT_SECRET)
+    monkeypatch.setenv("WEB_BASE_URL", WEB_BASE_URL)
+    monkeypatch.setenv("WEB_API_KEY", WEB_API_KEY)
+    app = create_app()
+    return TestClient(app)
+
+
+def make_signature(
+    path: str,
+    query_items: list[tuple[str, str]],
+    content_type: str = "",
+    body: bytes = b"",
+    ts_rec_headers: dict[str, str] | None = None,
+) -> str:
+    all_ts_rec = ts_rec_headers or {}
+    expires = next(v for k, v in all_ts_rec.items() if k.lower() == "x-ts-rec-expires")
+
+    content_md5 = (
+        base64.b64encode(hashlib.md5(body).digest()).decode()  # noqa: S324
+        if body
+        else ""
+    )
+
+    canonicalized_headers_list = sorted(
+        (name.lower(), value.strip())
+        for name, value in all_ts_rec.items()
+        if name.lower().startswith("x-ts-rec-")
+    )
+    canonicalized_headers = "".join(
+        f"{name}:{value}\n" for name, value in canonicalized_headers_list
+    )
+
+    params_no_sig = [(k, v) for k, v in query_items if k != "signature"]
+    query_string = urllib.parse.urlencode(params_no_sig)
+    canonicalized_resource = f"{path}?{query_string}" if query_string else path
+
+    string_to_sign = (
+        "POST\n"
+        + content_md5
+        + "\n"
+        + content_type
+        + "\n"
+        + expires
+        + "\n"
+        + canonicalized_headers
+        + canonicalized_resource
+    )
+
+    secret = TALENTSOFT_CLIENT_SECRET.encode("utf-8")
+    digest = hmac.new(secret, string_to_sign.encode("utf-8"), hashlib.sha1).digest()
+    return base64.b64encode(digest).decode("utf-8")
+
+
+def valid_query_items() -> list[tuple[str, str]]:
+    return [("client_id", TALENTSOFT_CLIENT_ID)]
+
+
+def valid_ts_rec_headers(expires: int | None = None) -> dict[str, str]:
+    if expires is None:
+        expires = int(time.time()) + 300
+    return {"X-TS-REC-Expires": str(expires)}
+
+
+def make_signed_request(client: TestClient, body: dict) -> Response:
+    body_bytes = json.dumps(body).encode()
+    content_type = "application/json"
+    ts_rec_headers = valid_ts_rec_headers()
+
+    query_items = valid_query_items()
+    signature = make_signature(
+        WEBHOOK_PATH,
+        query_items,
+        content_type=content_type,
+        ts_rec_headers=ts_rec_headers,
+    )
+    query_items.append(("signature", signature))
+
+    return client.post(
+        WEBHOOK_PATH,
+        params=dict(query_items),
+        content=body_bytes,
+        headers={"Content-Type": content_type, **ts_rec_headers},
+    )