office_translator/_bmad-output/implementation-artifacts/3-8-webhook-envoi-post-fire-forget.md

Story 3.8: Webhook - Envoi POST Fire & Forget

Status: done

Story

En tant que système, Je veux envoyer une requête POST à l'URL du webhook quand la traduction est terminée, de sorte que les utilisateurs puissent automatiser leurs workflows.

Acceptance Criteria

1. Envoi POST à la complétion: Quand un job de traduction termine (status "completed" ou "failed"), une requête POST est envoyée au webhook_url stocké. (FR37)
2. Payload structuré: Le payload inclut: {translation_id, status, timestamp, file_name, error_message?}. (FR38)
3. Timeout de 10 secondes: La requête a un timeout de 10 secondes.
4. Fire & Forget: Si le webhook échoue, l'erreur est loggée mais la traduction réussit quand même.
5. Envoi conditionnel: Le webhook est envoyé SEULEMENT si webhook_url a été fourni dans la requête. (FR66)

Tasks / Subtasks

• Task 1: Vérifier l'implémentation existante (AC: Tous)

  • 1.1 Analyser le code existant dans _run_translation_job (lignes 680-693)
  • 1.2 Vérifier que le webhook est envoyé dans le bloc finally
  • 1.3 Confirmer que le payload correspond aux spécifications FR38

• Task 2: Améliorer la gestion des erreurs (AC: #4)

  • 2.1 Logger les erreurs webhook avec plus de contexte (status code, response body)
  • 2.2 Ajouter des métriques pour le suivi des webhooks (optionnel)
  • 2.3 S'assurer que l'exception ne remonte jamais au caller

• Task 3: Améliorer le payload (AC: #2)

  • 3.1 Vérifier tous les champs requis: translation_id, status, timestamp, file_name, error_message
  • 3.2 Ajouter source_lang et target_lang au payload (utile pour l'utilisateur)
  • 3.3 Documenter le format du payload dans la docstring du endpoint

• Task 4: Tests d'intégration (AC: Tous)

  • 4.1 Tester un webhook réussi avec un serveur mock
  • 4.2 Tester un webhook qui timeout (devrait logger mais pas échouer)
  • 4.3 Tester un webhook qui retourne 4xx/5xx (devrait logger mais pas échouer)
  • 4.4 Tester une traduction sans webhook_url (pas de POST envoyé)
  • 4.5 Vérifier le payload envoyé correspond au format attendu

• Task 5: Documentation OpenAPI (AC: #2)

  • 5.1 Documenter le format du payload webhook dans la description du endpoint
  • 5.2 Ajouter un exemple de payload dans ERROR_EXAMPLES ou docstring

Dev Notes

🔥 État Actuel de l'Implémentation

⚠️ IMPORTANT: Le code webhook est DÉJÀ IMPLÉMENTÉ dans routes/translate_routes.py!

# Lignes 680-693 dans _run_translation_job()
finally:
    if webhook_url:
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                await client.post(
                    webhook_url,
                    json={
                        "translation_id": job_id,
                        "status": job["status"],
                        "timestamp": datetime.now(timezone.utc).isoformat(),
                        "file_name": job.get("file_name"),
                        "error_message": job.get("error_message"),
                    },
                )
        except Exception as e:
            logger.warning(f"Job {job_id}: Webhook notification failed - {e}")

CE QUI EXISTE DÉJÀ ✅:

• ✅ Envoi POST dans le bloc finally (garanti d'être exécuté)
• ✅ Condition if webhook_url (envoi seulement si URL fournie - FR66)
• ✅ Timeout de 10 secondes
• ✅ Payload avec: translation_id, status, timestamp, file_name, error_message
• ✅ Exception catchée et loggée (Fire & Forget)
• ✅ La traduction n'échoue pas si le webhook échoue

CE QUI MANQUE / À AMÉLIORER ❌:

• ❌ Pas de logging du status code HTTP de la réponse webhook
• ❌ Pas de logging du body de réponse en cas d'erreur
• ❌ Payload ne contient pas source_lang et target_lang (utile pour l'utilisateur)
• ❌ Pas de tests dédiés pour le webhook
• ❌ Pas de documentation du format payload dans OpenAPI

Architecture Patterns

Pattern Fire & Forget (depuis architecture.md):

Webhooks (FR36-FR38) → modules/webhooks/service.py → Fire & Forget

Le pattern Fire & Forget signifie:

1. Le système envoie le webhook de manière asynchrone
2. Aucune attente de confirmation de la part du destinataire
3. L'échec du webhook n'affecte pas le succès de la traduction
4. L'erreur est loggée pour debugging ultérieur

Format de Payload Recommandé:

{
  "translation_id": "tr_abc123def456",
  "status": "completed",
  "timestamp": "2024-01-15T10:30:00Z",
  "file_name": "report.xlsx",
  "source_lang": "en",
  "target_lang": "fr",
  "error_message": null
}

En cas d'erreur:

{
  "translation_id": "tr_abc123def456",
  "status": "failed",
  "timestamp": "2024-01-15T10:30:00Z",
  "file_name": "report.xlsx",
  "source_lang": "en",
  "target_lang": "fr",
  "error_message": "Provider unavailable: connection timeout"
}

Sécurité - Considérations

SSRF déjà géré dans la Story 3.7:

• La validation webhook_validator bloque les IPs privées et localhost
• Seules les URLs HTTP/HTTPS sont acceptées
• Pas de credentials dans l'URL

Timeout de 10 secondes:

• Suffisant pour la plupart des webhooks
• Évite de bloquer indéfiniment
• En cas de timeout, l'exception est catchée et loggée

Project Structure Notes

• Fichier principal: routes/translate_routes.py (vérifier/améliorer implémentation existante)
• Validation webhook: middleware/validation.py (déjà implémenté dans Story 3.7)
• Tests: tests/test_webhook_notification.py (nouveau fichier à créer)

Références

• [Source: _bmad-output/planning-artifacts/epics.md#Story 3.8]
• [Source: _bmad-output/planning-artifacts/architecture.md#Webhooks]
• [Source: FR36-FR38, FR65-FR66 - Webhook requirements]
• [Source: routes/translate_routes.py - Implémentation actuelle lignes 680-693]

Intelligence des Stories Précédentes (Epic 3)

Story 3.7 (Webhook - Spécification URL) - Enseignements

1. WebhookURLValidator déjà implémenté dans middleware/validation.py ✅
2. Code d'erreur INVALID_WEBHOOK_URL déjà utilisé ✅
3. Validation SSRF bloque localhost et IPs privées ✅
4. webhook_url stocké dans _translation_jobs[job_id] ✅
5. Paramètre optionnel - la traduction fonctionne sans webhook ✅

Story 3.6 (Documentation OpenAPI) - Enseignements

1. Schémas Pydantic dans schemas/ pour documentation ✅
2. Codes d'erreur documentés dans schemas/errors.py avec ERROR_EXAMPLES ✅
3. Descriptions complètes dans les docstrings des endpoints

Story 3.5 (API Versioning) - Enseignements

1. Tous les endpoints sont sous /api/v1/ ✅
2. Routeur principal routes/api_v1_router.py

Story 3.4 (Authentification X-API-Key) - Enseignements

1. Coexistence JWT + API Key dans get_authenticated_user
2. webhook_url accessible pour les utilisateurs authentifiés (JWT ou API Key)

Contexte Métier

Epic 3: API & Automation (Pro)

Cette story est la huitième de l'Epic 3 qui permet aux utilisateurs Pro d'automatiser les traductions:

1. API Keys - Génération (Story 3.1 ✅)
2. API Keys - Révocation User (Story 3.2 ✅)
3. API Keys - Révocation Admin (Story 3.3 ✅)
4. Authentification X-API-Key (Story 3.4 ✅)
5. API Versioning (Story 3.5 ✅)
6. Documentation OpenAPI (Story 3.6 ✅)
7. Webhook - Spécification URL (Story 3.7 ✅)
8. Webhook - Envoi POST Fire & Forget (cette story)
9. Glossaires (Stories 3.9-3.10 - backlog)
10. Custom Prompts (Stories 3.11-3.12 - backlog)

Valeur Business

Le webhook Fire & Forget est critique pour:

• Automation: Thomas (Pro user) peut enchaîner les traductions dans ses workflows n8n/Zapier
• Intégration CI/CD: Notification automatique quand une traduction termine
• Architecture événementielle: Base pour les futures fonctionnalités temps réel
• Expérience Pro: Justification de l'abonnement Pro

Dépendances

• Stories 3.1-3.7 (prérequis): API versionnée, authentifiée, documentée, avec validation webhook ✅
• Story 3.7 (prérequis direct): webhook_url stocké dans le job ✅
• Stories futures: Glossaires et Custom Prompts pourront aussi utiliser les webhooks

Guardrails Développeur

❌ À NE PAS FAIRE

1. NE PAS faire échouer la traduction si le webhook échoue (Fire & Forget!)
2. NE PAS attendre indéfiniment le webhook (timeout de 10s max)
3. NE PAS envoyer le webhook si webhook_url n'est pas fourni
4. NE PAS inclure de données sensibles dans le payload (pas de contenu de document)
5. NE PAS créer un nouveau module webhooks/ pour cette story - le code est déjà dans translate_routes.py
6. NE PAS oublier de logger les erreurs webhook pour debugging

✅ À FAIRE

1. VÉRIFIER que l'implémentation existante fonctionne correctement
2. AMÉLIORER le logging avec status code et response body
3. AJOUTER source_lang et target_lang au payload
4. CRÉER des tests dédiés pour le webhook
5. DOCUMENTER le format du payload dans OpenAPI
6. CONFIRMER que le webhook est envoyé dans le bloc finally (garanti d'être exécuté)

Code Suggéré

Amélioration du logging webhook dans routes/translate_routes.py

# Remacer le bloc finally existant (lignes 680-693) par:

finally:
    if webhook_url:
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                response = await client.post(
                    webhook_url,
                    json={
                        "translation_id": job_id,
                        "status": job["status"],
                        "timestamp": datetime.now(timezone.utc).isoformat(),
                        "file_name": job.get("file_name"),
                        "source_lang": job.get("source_lang"),
                        "target_lang": job.get("target_lang"),
                        "error_message": job.get("error_message"),
                    },
                )
                
                # Log successful webhook delivery
                if response.is_success:
                    logger.info(
                        f"Job {job_id}: Webhook notification sent successfully to {webhook_url} "
                        f"(status={response.status_code})"
                    )
                else:
                    # Log non-2xx response but don't fail
                    logger.warning(
                        f"Job {job_id}: Webhook returned non-success status "
                        f"(status={response.status_code}, url={webhook_url})"
                    )
                    
        except httpx.TimeoutException:
            logger.warning(
                f"Job {job_id}: Webhook notification timed out after 10s (url={webhook_url})"
            )
        except httpx.RequestError as e:
            logger.warning(
                f"Job {job_id}: Webhook notification failed - {type(e).__name__}: {e} "
                f"(url={webhook_url})"
            )
        except Exception as e:
            logger.warning(
                f"Job {job_id}: Unexpected webhook error - {type(e).__name__}: {e}"
            )

Fichier de tests tests/test_webhook_notification.py

"""
Tests for webhook notification functionality.
Story 3.8: Webhook - Envoi POST Fire & Forget
"""

import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from datetime import datetime, timezone
import httpx

from routes.translate_routes import _run_translation_job, _translation_jobs


class TestWebhookNotification:
    """Tests for webhook notification in translation jobs."""

    @pytest.fixture
    def mock_job(self, tmp_path):
        """Create a mock translation job."""
        job_id = "tr_test_webhook_123"
        input_path = tmp_path / "test_input.xlsx"
        input_path.write_bytes(b"PK\x03\x04" + b"\x00" * 100)  # Mock Office file
        
        _translation_jobs[job_id] = {
            "id": job_id,
            "status": "queued",
            "progress_percent": 0,
            "current_step": "Initializing",
            "file_name": "test.xlsx",
            "source_lang": "en",
            "target_lang": "fr",
            "created_at": datetime.now(timezone.utc).isoformat(),
            "input_path": str(input_path),
            "file_extension": ".xlsx",
            "provider": "google",
            "webhook_url": None,
        }
        
        yield job_id
        
        # Cleanup
        if job_id in _translation_jobs:
            del _translation_jobs[job_id]

    @pytest.mark.asyncio
    async def test_webhook_sent_on_completion(self, mock_job, tmp_path):
        """Webhook should be sent when translation completes."""
        job_id = mock_job
        webhook_url = "https://example.com/webhook"
        
        # Setup job with webhook URL
        _translation_jobs[job_id]["webhook_url"] = webhook_url
        
        # Mock the translation to complete successfully
        with patch("routes.translate_routes.excel_translator") as mock_translator:
            mock_translator.translate_file = MagicMock()
            
            with patch("httpx.AsyncClient") as mock_client:
                mock_response = AsyncMock()
                mock_response.is_success = True
                mock_response.status_code = 200
                
                mock_post = AsyncMock(return_value=mock_response)
                mock_client.return_value.__aenter__.return_value.post = mock_post
                
                # Run the job
                await _run_translation_job(
                    job_id=job_id,
                    input_path=tmp_path / "test_input.xlsx",
                    file_extension=".xlsx",
                    target_lang="fr",
                    source_lang="en",
                    provider="google",
                    user_id=None,
                    custom_prompt=None,
                    glossary_id=None,
                    webhook_url=webhook_url,
                )
                
                # Verify webhook was called
                mock_post.assert_called_once()
                call_args = mock_post.call_args
                
                assert call_args[0][0] == webhook_url
                payload = call_args[1]["json"]
                assert payload["translation_id"] == job_id
                assert payload["status"] == "completed"
                assert "timestamp" in payload
                assert payload["file_name"] == "test.xlsx"
                assert payload["source_lang"] == "en"
                assert payload["target_lang"] == "fr"

    @pytest.mark.asyncio
    async def test_webhook_not_sent_without_url(self, mock_job, tmp_path):
        """Webhook should NOT be sent if no URL provided."""
        job_id = mock_job
        
        # Ensure no webhook URL
        _translation_jobs[job_id]["webhook_url"] = None
        
        with patch("routes.translate_routes.excel_translator") as mock_translator:
            mock_translator.translate_file = MagicMock()
            
            with patch("httpx.AsyncClient") as mock_client:
                # Run the job without webhook URL
                await _run_translation_job(
                    job_id=job_id,
                    input_path=tmp_path / "test_input.xlsx",
                    file_extension=".xlsx",
                    target_lang="fr",
                    source_lang="en",
                    provider="google",
                    user_id=None,
                    custom_prompt=None,
                    glossary_id=None,
                    webhook_url=None,
                )
                
                # Verify webhook was NOT called
                mock_client.assert_not_called()

    @pytest.mark.asyncio
    async def test_webhook_timeout_does_not_fail_translation(self, mock_job, tmp_path):
        """Translation should succeed even if webhook times out."""
        job_id = mock_job
        webhook_url = "https://example.com/webhook"
        
        _translation_jobs[job_id]["webhook_url"] = webhook_url
        
        with patch("routes.translate_routes.excel_translator") as mock_translator:
            mock_translator.translate_file = MagicMock()
            
            with patch("httpx.AsyncClient") as mock_client:
                # Simulate timeout
                mock_client.return_value.__aenter__.return_value.post = AsyncMock(
                    side_effect=httpx.TimeoutException("Timeout")
                )
                
                # Run the job
                await _run_translation_job(
                    job_id=job_id,
                    input_path=tmp_path / "test_input.xlsx",
                    file_extension=".xlsx",
                    target_lang="fr",
                    source_lang="en",
                    provider="google",
                    user_id=None,
                    custom_prompt=None,
                    glossary_id=None,
                    webhook_url=webhook_url,
                )
                
                # Translation should still be marked as completed
                assert _translation_jobs[job_id]["status"] == "completed"

    @pytest.mark.asyncio
    async def test_webhook_error_response_does_not_fail_translation(self, mock_job, tmp_path):
        """Translation should succeed even if webhook returns error."""
        job_id = mock_job
        webhook_url = "https://example.com/webhook"
        
        _translation_jobs[job_id]["webhook_url"] = webhook_url
        
        with patch("routes.translate_routes.excel_translator") as mock_translator:
            mock_translator.translate_file = MagicMock()
            
            with patch("httpx.AsyncClient") as mock_client:
                # Simulate 500 error from webhook
                mock_response = AsyncMock()
                mock_response.is_success = False
                mock_response.status_code = 500
                
                mock_client.return_value.__aenter__.return_value.post = AsyncMock(
                    return_value=mock_response
                )
                
                # Run the job
                await _run_translation_job(
                    job_id=job_id,
                    input_path=tmp_path / "test_input.xlsx",
                    file_extension=".xlsx",
                    target_lang="fr",
                    source_lang="en",
                    provider="google",
                    user_id=None,
                    custom_prompt=None,
                    glossary_id=None,
                    webhook_url=webhook_url,
                )
                
                # Translation should still be marked as completed
                assert _translation_jobs[job_id]["status"] == "completed"

    @pytest.mark.asyncio
    async def test_webhook_payload_on_failure(self, mock_job, tmp_path):
        """Webhook payload should include error_message on translation failure."""
        job_id = mock_job
        webhook_url = "https://example.com/webhook"
        
        _translation_jobs[job_id]["webhook_url"] = webhook_url
        
        with patch("routes.translate_routes.excel_translator") as mock_translator:
            # Simulate translation failure
            mock_translator.translate_file = MagicMock(
                side_effect=Exception("Provider unavailable")
            )
            
            with patch("httpx.AsyncClient") as mock_client:
                mock_response = AsyncMock()
                mock_response.is_success = True
                mock_response.status_code = 200
                
                mock_post = AsyncMock(return_value=mock_response)
                mock_client.return_value.__aenter__.return_value.post = mock_post
                
                # Run the job
                await _run_translation_job(
                    job_id=job_id,
                    input_path=tmp_path / "test_input.xlsx",
                    file_extension=".xlsx",
                    target_lang="fr",
                    source_lang="en",
                    provider="google",
                    user_id=None,
                    custom_prompt=None,
                    glossary_id=None,
                    webhook_url=webhook_url,
                )
                
                # Verify webhook was called with error
                mock_post.assert_called_once()
                payload = mock_post.call_args[1]["json"]
                assert payload["status"] == "failed"
                assert "Provider unavailable" in payload["error_message"]


class TestWebhookPayloadFormat:
    """Tests for webhook payload format compliance."""

    def test_payload_has_required_fields(self):
        """Verify payload has all required fields per FR38."""
        required_fields = [
            "translation_id",
            "status",
            "timestamp",
            "file_name",
            "error_message",  # Optional but must be present (can be null)
        ]
        
        # This test documents the expected fields
        # Actual validation happens in integration tests
        assert len(required_fields) == 5

    def test_payload_extended_fields(self):
        """Verify payload includes useful extended fields."""
        extended_fields = [
            "source_lang",
            "target_lang",
        ]
        
        # These fields are recommended for better UX
        assert len(extended_fields) == 2

Documentation du payload dans la docstring du endpoint

Ajouter dans routes/translate_routes.py docstring:

"""
Submit a document for translation.

...

**Webhook Notification:**
If `webhook_url` is provided, a POST request will be sent when translation completes.

**Webhook Payload (Success):**
```json
{
  "translation_id": "tr_abc123def456",
  "status": "completed",
  "timestamp": "2024-01-15T10:30:00Z",
  "file_name": "report.xlsx",
  "source_lang": "en",
  "target_lang": "fr",
  "error_message": null
}

Webhook Payload (Failure):

{
  "translation_id": "tr_abc123def456",
  "status": "failed",
  "timestamp": "2024-01-15T10:30:00Z",
  "file_name": "report.xlsx",
  "source_lang": "en",
  "target_lang": "fr",
  "error_message": "Provider unavailable: connection timeout"
}

Webhook Behavior:

• Timeout: 10 seconds
• Fire & Forget: Translation succeeds even if webhook fails
• Retries: None (implement retry logic on your server if needed) """

## Dev Agent Record

### Agent Model Used

Claude 3.5 Sonnet (claude-3-5-sonnet)

### Debug Log References

- Tests exécutés: `pytest tests/test_webhook_notification.py -v`
- Résultat: 11 tests passés

### Completion Notes List

- ✅ Analyse exhaustive du contexte terminée
- ✅ Code existant identifié dans `routes/translate_routes.py`
- ✅ Logging amélioré avec status code et types d'erreur spécifiques
- ✅ Payload étendu avec `source_lang` et `target_lang`
- ✅ Documentation webhook ajoutée dans la docstring du endpoint
- ✅ 11 tests créés et passent tous

### File List

- `routes/translate_routes.py` - MODIFIÉ - Logging webhook amélioré, payload étendu, documentation ajoutée
- `tests/test_webhook_notification.py` - CRÉÉ - 11 tests pour webhook notification

## Change Log

- 2026-02-22: Story créée avec contexte complet et analyse exhaustive
- 2026-02-22: Code existant déjà implémenté identifié
- 2026-02-22: Implémentation terminée - logging amélioré, payload étendu, tests ajoutés

## Checklist de Validation

Avant de marquer cette story comme terminée, vérifier:

- [x] Le webhook est envoyé quand la traduction termine (completed ou failed)
- [x] Le payload contient: translation_id, status, timestamp, file_name, error_message
- [x] Le timeout est de 10 secondes
- [x] L'échec du webhook n'affecte pas la traduction (Fire & Forget)
- [x] Le webhook n'est envoyé que si webhook_url est fourni
- [x] Les erreurs webhook sont loggées avec contexte
- [x] Les tests passent (11/11)
- [x] La documentation OpenAPI est mise à jour