feat: Docuseal member agreement webhook

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."

.env.example

@@ -88,6 +88,9 @@ RESUME_EXTRACTOR_VERSION=v1
 CRM_SYNC_ENABLED=true
 CRM_SYNC_INTERVAL_SECONDS=900
 CRM_SYNC_PAGE_SIZE=200
+# Optional: restrict Docuseal agreements to this template id.
+# If unset, Docuseal agreement processing is ignored.
+DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID=
 
 # Discord bot (required for bot runtime)
 DISCORD_BOT_TOKEN=your_bot_token_here

apps/worker/src/five08/backend/api.py

@@ -54,16 +54,22 @@
 from five08.worker.config import settings
 from five08.worker.db_migrations import run_job_migrations
 from five08.worker.dispatcher import build_queue_client
+from five08.worker.masking import mask_email
 from five08.worker.jobs import (
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
 )
 from five08.worker.mailbox_resume_ingest import ResumeMailboxProcessor
-from five08.worker.models import AuditEventPayload, EspoCRMWebhookPayload
+from five08.worker.models import (
+ AuditEventPayload,
+ DocusealWebhookPayload,
+ EspoCRMWebhookPayload,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -97,8 +103,13 @@ def _is_authorized(request: Request) -> bool:
  logger.error("Rejecting request: API_SHARED_SECRET is not configured")
  return False
 
+ # TODO: security-hardening: move webhook auth to per-webhook generated secrets
+ # sourced from an admin dashboard with copyable callback URLs.
  provided_secret = request.headers.get("X-API-Secret", "")
- return secrets.compare_digest(provided_secret, settings.api_shared_secret)
+ if secrets.compare_digest(provided_secret, settings.api_shared_secret):
+ return True
+
+ return False
 
 
 def _extract_idempotency_key(value: object) -> str | None:
@@ -680,6 +691,123 @@ async def espocrm_people_sync_webhook_handler(request: Request) -> JSONResponse:
  )
 
 
+async def docuseal_webhook_handler(request: Request) -> JSONResponse:
+ """Process a Docuseal form.completed webhook and enqueue agreement job."""
+ if not _is_authorized(request):
+ return JSONResponse({"error": "unauthorized"}, status_code=401)
+
+ try:
+ payload_data = await request.json()
+ except Exception:
+ return JSONResponse({"error": "invalid_json"}, status_code=400)
+
+ if not isinstance(payload_data, dict):
+ return JSONResponse({"error": "payload_must_be_object"}, status_code=400)
+
+ try:
+ payload = DocusealWebhookPayload.model_validate(payload_data)
+ except (ValidationError, TypeError) as exc:
+ return JSONResponse(
+ {"error": "invalid_payload", "detail": str(exc)},
+ status_code=400,
+ )
+
+ if payload.event_type != "form.completed":
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": f"unhandled event_type: {payload.event_type}",
+ },
+ status_code=200,
+ )
+
+ submitter = payload.data
+ submission_id = (
+ submitter.submission_id if submitter.submission_id is not None else submitter.id
+ )
+
+ template_filter_id = settings.docuseal_member_agreement_template_id
+ if template_filter_id is None:
+ logger.info("Ignoring Docuseal agreement webhook: template filter is unset")
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_filter_not_configured",
+ },
+ status_code=200,
+ )
+
+ template_id = submitter.template.id if submitter.template else None
+ if template_id != template_filter_id:
+ logger.info(
+ "Ignoring Docuseal agreement webhook for unmatched template_id=%s"
+ " expected=%s submission_id=%s",
+ template_id,
+ template_filter_id,
+ submission_id,
+ )
+ return JSONResponse(
+ {
+ "status": "ignored",
+ "reason": "template_mismatch",
+ "submission_id": submission_id,
+ },
+ status_code=200,
+ )
+
+ email = (submitter.email or "").strip()
+
+ completed_at = submitter.completed_at or payload.timestamp
+ if isinstance(completed_at, str):
+ completed_at = completed_at.strip()
+ if not isinstance(completed_at, str) or not completed_at:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ try:
+ datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+ except ValueError:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ if not email:
+ return JSONResponse({"error": "invalid_payload"}, status_code=400)
+
+ masked_email = mask_email(email)
+
+ queue = request.app.state.queue
+ try:
+ job: EnqueuedJob = await asyncio.to_thread(
+ enqueue_job,
+ queue=queue,
+ fn=process_docuseal_agreement_job,
+ args=(email, completed_at, submission_id),
+ settings=settings,
+ idempotency_key=f"docuseal-agreement:{submission_id}",
+ )
+ except Exception:
+ logger.exception(
+ "Failed enqueueing Docuseal agreement job masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return JSONResponse({"error": "enqueue_failed"}, status_code=503)
+
+ logger.info(
+ "Enqueued Docuseal agreement job job_id=%s masked_email=%s",
+ job.id,
+ masked_email,
+ )
+ return JSONResponse(
+ {
+ "status": "queued",
+ "source": "docuseal",
+ "job_id": job.id,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ },
+ status_code=202,
+ )
+
+
 async def audit_event_handler(request: Request) -> JSONResponse:
  """Persist one human audit event."""
  if not _is_authorized(request):
@@ -1228,6 +1356,11 @@ def create_app(*, run_lifespan: bool = True) -> FastAPI:
  espocrm_people_sync_webhook_handler,
  methods=["POST"],
  )
+ app.add_api_route(
+ "/webhooks/docuseal",
+ docuseal_webhook_handler,
+ methods=["POST"],
+ )
  app.add_api_route("/webhooks/{source}", ingest_handler, methods=["POST"])
 
  app.add_api_route(

apps/worker/src/five08/worker/actors.py

@@ -25,6 +25,7 @@
  apply_resume_profile_job,
  extract_resume_profile_job,
  process_contact_skills_job,
+ process_docuseal_agreement_job,
  process_webhook_event,
  sync_people_from_crm_job,
  sync_person_from_crm_job,
@@ -48,6 +49,7 @@
  apply_resume_profile_job.__name__: apply_resume_profile_job,
  sync_people_from_crm_job.__name__: sync_people_from_crm_job,
  sync_person_from_crm_job.__name__: sync_person_from_crm_job,
+ process_docuseal_agreement_job.__name__: process_docuseal_agreement_job,
 }
 
 

apps/worker/src/five08/worker/config.py

@@ -2,7 +2,7 @@
 
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import field_validator, model_validator
 
 from five08.settings import SharedSettings
 
@@ -23,6 +23,7 @@ class WorkerSettings(SharedSettings):
  openai_model: str = "gpt-4o-mini"
  resume_ai_model: str = "gpt-4o-mini"
  resume_extractor_version: str = "v1"
+ docuseal_member_agreement_template_id: int | None = None
 
  max_file_size_mb: int = 10
  allowed_file_types: str = "pdf,doc,docx,txt"
@@ -91,6 +92,23 @@ def validate_auth_cookie_samesite(self) -> "WorkerSettings":
  self.auth_cookie_samesite = normalized
  return self
 
+ @field_validator("docuseal_member_agreement_template_id", mode="before")
+ @classmethod
+ def _normalize_docuseal_member_agreement_template_id(
+ cls,
+ value: object,
+ ) -> int | None:
+ if value is None:
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, str):
+ normalized = value.strip()
+ if not normalized:
+ return None
+ return int(normalized)
+ raise TypeError("DOCUSEAL_MEMBER_AGREEMENT_TEMPLATE_ID must be an integer")
+
  @property
  def allowed_file_extensions(self) -> set[str]:
  """Allowed resume file extensions."""

apps/worker/src/five08/worker/crm/docuseal_processor.py

@@ -0,0 +1,98 @@
+"""Docuseal member agreement processing workflow."""
+
+import logging
+from typing import Any
+
+from five08.clients.espo import EspoAPI, EspoAPIError
+from five08.worker.config import settings
+from five08.worker.masking import mask_email
+
+logger = logging.getLogger(__name__)
+
+
+class DocusealAgreementProcessor:
+ """Look up a CRM contact by email and mark their member agreement as signed."""
+
+ def __init__(self) -> None:
+ api_url = settings.espo_base_url.rstrip("/") + "/api/v1"
+ self.api = EspoAPI(api_url, settings.espo_api_key)
+
+ def process_agreement(
+ self,
+ email: str,
+ completed_at: str,
+ submission_id: int,
+ ) -> dict[str, Any]:
+ """Search for the signer by email and update cMemberAgreementSignedAt."""
+ masked_email = mask_email(email)
+
+ try:
+ result = self.api.request(
+ "GET",
+ "Contact",
+ {
+ "where": [
+ {
+ "type": "equals",
+ "attribute": "emailAddress",
+ "value": email,
+ }
+ ],
+ "maxSize": 1,
+ "select": "id,name,emailAddress",
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM search failed for masked_email=%s: %s", masked_email, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": f"CRM search failed: {exc}",
+ }
+
+ contacts = result.get("list", [])
+ if not contacts:
+ logger.warning(
+ "No CRM contact found for masked_email=%s submission_id=%s",
+ masked_email,
+ submission_id,
+ )
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "error": "contact_not_found",
+ }
+
+ contact = contacts[0]
+ contact_id = contact["id"]
+
+ try:
+ self.api.request(
+ "PUT",
+ f"Contact/{contact_id}",
+ {
+ "cMemberAgreementSignedAt": completed_at,
+ },
+ )
+ except EspoAPIError as exc:
+ logger.error("CRM update failed for contact_id=%s: %s", contact_id, exc)
+ return {
+ "success": False,
+ "masked_email": masked_email,
+ "submission_id": submission_id,
+ "contact_id": contact_id,
+ "error": f"CRM update failed: {exc}",
+ }
+
+ logger.info(
+ "Marked member agreement signed contact_id=%s masked_email=%s",
+ contact_id,
+ masked_email,
+ )
+ return {
+ "success": True,
+ "masked_email": masked_email,
+ "contact_id": contact_id,
+ "submission_id": submission_id,
+ "completed_at": completed_at,
+ }

apps/worker/src/five08/worker/jobs.py

@@ -4,9 +4,11 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from five08.worker.crm.docuseal_processor import DocusealAgreementProcessor
 from five08.worker.crm.people_sync import PeopleSyncProcessor
 from five08.worker.crm.processor import ContactSkillsProcessor
 from five08.worker.crm.resume_profile_processor import ResumeProfileProcessor
+from five08.worker.masking import mask_email
 
 logger = logging.getLogger(__name__)
 
@@ -68,6 +70,21 @@ def apply_resume_profile_job(
  return result.model_dump()
 
 
+def process_docuseal_agreement_job(
+ email: str,
+ completed_at: str,
+ submission_id: int,
+) -> dict[str, Any]:
+ """Mark a CRM contact as having signed the member agreement via Docuseal."""
+ logger.info(
+ "Processing Docuseal agreement job masked_email=%s submission_id=%s",
+ mask_email(email),
+ submission_id,
+ )
+ processor = DocusealAgreementProcessor()
+ return processor.process_agreement(email, completed_at, submission_id)
+
+
 def sync_people_from_crm_job() -> dict[str, Any]:
  """Sync a full contacts page-set from CRM into the local people cache."""
  logger.info("Processing CRM people full-sync job")

apps/worker/src/five08/worker/masking.py

@@ -0,0 +1,15 @@
+"""PII masking helpers used across worker modules."""
+
+
+def mask_email(email: str) -> str:
+ """Return a deterministic redacted email representation for logs and responses."""
+ local, at, domain = email.partition("@")
+ if not at:
+ return "***"
+
+ masked_local = (local[:1] if local else "*") + "***"
+
+ if not domain:
+ return f"{masked_local}@****..."
+
+ return f"{masked_local}@{domain[:1]}****..."