# Sprint 6: Webhook Execution

> **Priority**: 6 | **Owner**: Backend | **Complexity**: Medium

---

## Validation Status (2025-12-26)

### ✅ SPRINT COMPLETE

All Sprint 6 tasks have been implemented and verified.

| Component | Location | Status |
|-----------|----------|--------|
| Feature Flag | `config/settings.py` | ✅ `webhooks_enabled` in `FeatureFlags` |
| Domain Events | `domain/webhooks/events.py` | ✅ `WebhookEventType`, `WebhookConfig`, `WebhookDelivery`, payloads |
| Webhook Executor | `infrastructure/webhooks/executor.py` | ✅ HTTP delivery with retry + HMAC signing |
| Webhook Service | `application/services/webhook_service.py` | ✅ Full orchestration with trigger methods |
| Webhook Converter | `infrastructure/converters/webhook_converters.py` | ✅ ORM ↔ domain conversion |
| Repository | `persistence/repositories/webhook_repo.py` | ✅ `SqlAlchemyWebhookRepository` |
| Memory Repository | `persistence/memory/repositories.py` | ✅ `InMemoryWebhookRepository` |
| ORM Models | `persistence/models/integrations/webhook.py` | ✅ `WebhookConfigModel`, `WebhookDeliveryModel` |
| UoW Integration | `domain/ports/unit_of_work.py` | ✅ `webhooks` property + `supports_webhooks` |
| gRPC Integration | `grpc/_mixins/meeting.py`, `summarization.py` | ✅ Fire-and-forget webhook triggers |
| Service Injection | `grpc/service.py`, `server.py`, `protocols.py` | ✅ Three-tier injection pattern |
| Proto definitions | `noteflow.proto` | ✅ Webhook-related messages |
| Unit Tests | `tests/infrastructure/webhooks/` | ✅ Executor and service tests |

### All Acceptance Criteria Met ✓

- Webhooks fire on meeting.completed, summary.generated, recording.started/stopped
- HMAC-SHA256 signing via `X-NoteFlow-Signature` header
- Exponential backoff retry logic (configurable `max_retries`)
- Webhook failures isolated from primary RPC operations
- Feature flag controls availability

---

## Objective

Enable webhook notifications for meeting lifecycle events. Users can configure webhook URLs to receive payloads when meetings complete, summaries are generated, or recordings start/stop.

---

## Current State Analysis

### What Already Exists ✅

| Component | Location | Status |
|-----------|----------|--------|
| **ORM Models** | `models/integrations/webhook.py` | `WebhookConfigModel`, `WebhookDeliveryModel` complete |
| **Alembic Migration** | `migrations/versions/i3d4e5f6g7h8_add_webhook_tables.py` | Tables created with indexes |
| **Feature Flag** | `config/settings.py:231` | `webhooks_enabled` in `FeatureFlags` |
| **httpx Dependency** | `pyproject.toml` | Already `>=0.27` |
| Frontend UI | `client/src/components/integration-config-panel.tsx` | Webhook URL input exists |
| Meeting Events | `src/noteflow/grpc/_mixins/meeting.py` | `StopMeeting` RPC exists |
| Summary Events | `src/noteflow/grpc/_mixins/summarization.py` | `GenerateSummary` RPC exists |

### Existing ORM Model Fields

The `WebhookConfigModel` includes fields that domain entities must match:

| Field | Type | Default | Notes |
|-------|------|---------|-------|
| `id` | UUID | auto | Primary key |
| `workspace_id` | UUID | required | FK to workspaces |
| `name` | String(255) | "Webhook" | Display name |
| `url` | Text | required | HTTPS validated via CHECK |
| `events` | ARRAY(Text) | [] | Subscribed event types |
| `secret` | Text | null | HMAC signing secret |
| `enabled` | Boolean | true | Active flag |
| `timeout_ms` | Integer | 10000 | Request timeout |
| `max_retries` | Integer | 3 | Retry attempts |
| `created_at` | timestamptz | now() | — |
| `updated_at` | timestamptz | now() | Auto-updated via trigger |

### Gap (What Still Needs Building)

1. **Domain Layer**: Protocol definition, event types, payload dataclasses
2. **Repository**: `SqlAlchemyWebhookRepository` implementing domain port
3. **Converters**: ORM ↔ domain entity conversion
4. **UoW Integration**: Add webhook repository to Unit of Work
5. **Webhook Executor**: HTTP client with retry logic and HMAC signing
6. **Application Service**: `WebhookService` for orchestration
7. **gRPC Integration**: Service injection and event triggering in mixins

---

## Target/Affected Code

### Files to Create

| File | Purpose |
|------|---------|
| `src/noteflow/domain/ports/repositories.py` | Add `WebhookRepository` protocol (modify existing) |
| `src/noteflow/domain/webhooks/__init__.py` | Webhook domain package |
| `src/noteflow/domain/webhooks/events.py` | Webhook event definitions |
| `src/noteflow/infrastructure/converters/webhook_converters.py` | ORM ↔ domain conversion |
| `src/noteflow/infrastructure/persistence/repositories/webhook_repo.py` | SQLAlchemy webhook repository |
| `src/noteflow/infrastructure/persistence/memory/repositories.py` | Add `InMemoryWebhookRepository` (modify existing) |
| `src/noteflow/infrastructure/integrations/webhooks.py` | Webhook executor with retry logic |
| `src/noteflow/application/services/webhook_service.py` | Webhook application service |
| `tests/infrastructure/integrations/test_webhooks.py` | Executor unit tests |
| `tests/application/test_webhook_service.py` | Service unit tests |

### Files to Modify

| File | Change Type | Lines Est. |
|------|-------------|------------|
| `src/noteflow/domain/ports/repositories.py` | Add `WebhookRepository` protocol | +25 |
| `src/noteflow/domain/ports/unit_of_work.py` | Add `webhooks` property + `supports_webhooks` | +10 |
| `src/noteflow/domain/ports/__init__.py` | Export `WebhookRepository` | +2 |
| `src/noteflow/infrastructure/persistence/unit_of_work.py` | Add webhook repository integration | +20 |
| `src/noteflow/infrastructure/persistence/memory/unit_of_work.py` | Add memory webhook repository | +15 |
| `src/noteflow/infrastructure/persistence/repositories/__init__.py` | Export webhook repo | +2 |
| `src/noteflow/infrastructure/converters/__init__.py` | Export `WebhookConverter` | +2 |
| `src/noteflow/grpc/_mixins/protocols.py` | Add `_webhook_service` to `ServicerHost` | +3 |
| `src/noteflow/grpc/service.py` | Accept and store webhook service | +5 |
| `src/noteflow/grpc/server.py` | Initialize and inject webhook service | +25 |
| `src/noteflow/grpc/_mixins/meeting.py` | Trigger on StopMeeting | +10 |
| `src/noteflow/grpc/_mixins/summarization.py` | Trigger on GenerateSummary | +10 |
| `src/noteflow/grpc/_mixins/streaming.py` | Trigger on recording start | +10 |
| `tests/conftest.py` | Add shared webhook fixtures | +15 |

### Already Complete (No Changes Needed)

| File | Status |
|------|--------|
| `models/integrations/webhook.py` | ✅ ORM models exist |
| `migrations/versions/i3d4e5f6g7h8_add_webhook_tables.py` | ✅ Migration exists |
| `models/integrations/__init__.py` | ✅ Already exports webhook models |
| `pyproject.toml` | ✅ httpx already `>=0.27` |
| `docker/db/schema.sql` | ✅ Tables already defined |

---

## Implementation Tasks

### Task 0: Domain Ports & Repository Protocol

**PREREQUISITE**: Define domain contracts before implementing infrastructure.

#### Domain Port Protocol

**File**: `src/noteflow/domain/ports/repositories.py` (add to existing file)

```python
class WebhookRepository(Protocol):
    """Repository protocol for webhook operations."""

    async def get_all_enabled(self, workspace_id: UUID | None = None) -> Sequence[WebhookConfig]:
        """Retrieve all enabled webhook configurations.

        Args:
            workspace_id: Optional filter by workspace.

        Returns:
            Sequence of enabled webhook configurations.
        """
        ...

    async def get_by_id(self, webhook_id: UUID) -> WebhookConfig | None:
        """Retrieve webhook by ID.

        Args:
            webhook_id: Unique webhook identifier.

        Returns:
            Webhook configuration or None if not found.
        """
        ...

    async def create(self, config: WebhookConfig) -> WebhookConfig:
        """Persist a new webhook configuration.

        Args:
            config: Webhook configuration to create.

        Returns:
            Created webhook with generated ID.
        """
        ...

    async def update(self, config: WebhookConfig) -> WebhookConfig:
        """Update an existing webhook configuration.

        Args:
            config: Webhook configuration with updated values.

        Returns:
            Updated webhook configuration.
        """
        ...

    async def delete(self, webhook_id: UUID) -> bool:
        """Delete a webhook configuration.

        Args:
            webhook_id: Unique webhook identifier.

        Returns:
            True if deleted, False if not found.
        """
        ...

    async def add_delivery(self, delivery: WebhookDelivery) -> WebhookDelivery:
        """Record a webhook delivery attempt.

        Args:
            delivery: Delivery record to persist.

        Returns:
            Persisted delivery record.
        """
        ...

    async def get_deliveries(
        self,
        webhook_id: UUID,
        limit: int = 50,
    ) -> Sequence[WebhookDelivery]:
        """Retrieve delivery history for a webhook.

        Args:
            webhook_id: Webhook to get deliveries for.
            limit: Maximum number of records.

        Returns:
            Sequence of delivery records, newest first.
        """
        ...
```

#### Unit of Work Protocol Update

**File**: `src/noteflow/domain/ports/unit_of_work.py` (add to existing protocol)

```python
# Add property to UnitOfWork protocol:
@property
def webhooks(self) -> WebhookRepository:
    """Access the webhooks repository."""
    ...

@property
def supports_webhooks(self) -> bool:
    """Check if webhook operations are supported."""
    ...
```

#### SQLAlchemy Unit of Work Integration

**File**: `src/noteflow/infrastructure/persistence/unit_of_work.py` (modify existing)

```python
# Add import:
from .repositories import SqlAlchemyWebhookRepository

# Add to __init__:
self._webhooks_repo: SqlAlchemyWebhookRepository | None = None

# Add property:
@property
def webhooks(self) -> SqlAlchemyWebhookRepository:
    """Get webhooks repository."""
    if self._webhooks_repo is None:
        raise RuntimeError("UnitOfWork not in context")
    return self._webhooks_repo

@property
def supports_webhooks(self) -> bool:
    """Webhook persistence is fully supported with database."""
    return True

# Add to __aenter__:
self._webhooks_repo = SqlAlchemyWebhookRepository(self._session)

# Add to __aexit__:
self._webhooks_repo = None
```

#### Memory Unit of Work Integration

**File**: `src/noteflow/infrastructure/persistence/memory/unit_of_work.py` (modify existing)

```python
# Add to __init__:
self._webhooks = InMemoryWebhookRepository()

# Add property:
@property
def webhooks(self) -> InMemoryWebhookRepository:
    """Get webhooks repository."""
    return self._webhooks

@property
def supports_webhooks(self) -> bool:
    """Webhook persistence supported in memory mode."""
    return True
```

---

### Task 0.5: Converters

**File**: `src/noteflow/infrastructure/converters/webhook_converters.py`

```python
"""Webhook ORM ↔ domain entity converters."""

from __future__ import annotations

from typing import TYPE_CHECKING

from noteflow.domain.webhooks.events import (
    WebhookConfig,
    WebhookDelivery,
    WebhookEventType,
)

if TYPE_CHECKING:
    from noteflow.infrastructure.persistence.models.integrations.webhook import (
        WebhookConfigModel,
        WebhookDeliveryModel,
    )


class WebhookConverter:
    """Convert between webhook ORM models and domain entities."""

    @staticmethod
    def config_to_domain(model: WebhookConfigModel) -> WebhookConfig:
        """Convert ORM WebhookConfigModel to domain WebhookConfig."""
        return WebhookConfig(
            id=model.id,
            workspace_id=model.workspace_id,
            name=model.name,
            url=model.url,
            events=frozenset(WebhookEventType(e) for e in model.events),
            secret=model.secret,
            enabled=model.enabled,
            timeout_ms=model.timeout_ms,
            max_retries=model.max_retries,
            created_at=model.created_at,
            updated_at=model.updated_at,
        )

    @staticmethod
    def config_to_orm_kwargs(entity: WebhookConfig) -> dict[str, object]:
        """Convert domain WebhookConfig to ORM kwargs."""
        return {
            "id": entity.id,
            "workspace_id": entity.workspace_id,
            "name": entity.name,
            "url": entity.url,
            "events": [e.value for e in entity.events],
            "secret": entity.secret,
            "enabled": entity.enabled,
            "timeout_ms": entity.timeout_ms,
            "max_retries": entity.max_retries,
        }

    @staticmethod
    def delivery_to_domain(model: WebhookDeliveryModel) -> WebhookDelivery:
        """Convert ORM WebhookDeliveryModel to domain WebhookDelivery."""
        return WebhookDelivery(
            id=model.id,
            webhook_id=model.webhook_id,
            event_type=WebhookEventType(model.event_type),
            payload=dict(model.payload),
            status_code=model.status_code,
            response_body=model.response_body,
            error_message=model.error_message,
            attempt_count=model.attempt_count,
            duration_ms=model.duration_ms,
            delivered_at=model.delivered_at,
        )

    @staticmethod
    def delivery_to_orm_kwargs(entity: WebhookDelivery) -> dict[str, object]:
        """Convert domain WebhookDelivery to ORM kwargs."""
        return {
            "id": entity.id,
            "webhook_id": entity.webhook_id,
            "event_type": entity.event_type.value,
            "payload": entity.payload,
            "status_code": entity.status_code,
            "response_body": entity.response_body,
            "error_message": entity.error_message,
            "attempt_count": entity.attempt_count,
            "duration_ms": entity.duration_ms,
        }
```

**File**: `src/noteflow/infrastructure/converters/__init__.py` (add export)

```python
from noteflow.infrastructure.converters.webhook_converters import WebhookConverter

__all__ = [
    # ... existing exports ...
    "WebhookConverter",
]
```

---

### Task 1: Domain Event Definitions

**File**: `src/noteflow/domain/webhooks/events.py`

> **Note**: Domain entities must match existing ORM model fields in `models/integrations/webhook.py`.

```python
"""Webhook event definitions."""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
from uuid import UUID, uuid4

from noteflow.domain.utils.time import utc_now


class WebhookEventType(Enum):
    """Types of webhook trigger events."""

    MEETING_COMPLETED = "meeting.completed"
    SUMMARY_GENERATED = "summary.generated"
    RECORDING_STARTED = "recording.started"
    RECORDING_STOPPED = "recording.stopped"


@dataclass(frozen=True, slots=True)
class WebhookConfig:
    """Webhook configuration.

    Fields match WebhookConfigModel ORM for seamless conversion.
    """

    id: UUID
    workspace_id: UUID
    url: str
    events: frozenset[WebhookEventType]
    name: str = "Webhook"
    secret: str | None = None
    enabled: bool = True
    timeout_ms: int = 10000
    max_retries: int = 3
    created_at: datetime = field(default_factory=utc_now)
    updated_at: datetime = field(default_factory=utc_now)

    @classmethod
    def create(
        cls,
        workspace_id: UUID,
        url: str,
        events: list[WebhookEventType],
        *,
        name: str = "Webhook",
        secret: str | None = None,
        timeout_ms: int = 10000,
        max_retries: int = 3,
    ) -> WebhookConfig:
        """Create a new webhook configuration."""
        now = utc_now()
        return cls(
            id=uuid4(),
            workspace_id=workspace_id,
            url=url,
            events=frozenset(events),
            name=name,
            secret=secret,
            timeout_ms=timeout_ms,
            max_retries=max_retries,
            created_at=now,
            updated_at=now,
        )


@dataclass(frozen=True, slots=True)
class WebhookDelivery:
    """Record of a webhook delivery attempt.

    Fields match WebhookDeliveryModel ORM for seamless conversion.
    """

    id: UUID
    webhook_id: UUID
    event_type: WebhookEventType
    payload: dict[str, Any]
    status_code: int | None
    response_body: str | None
    error_message: str | None
    attempt_count: int
    duration_ms: int | None
    delivered_at: datetime

    @property
    def succeeded(self) -> bool:
        """Check if delivery was successful."""
        return self.status_code is not None and 200 <= self.status_code < 300


@dataclass(frozen=True, slots=True)
class WebhookPayload:
    """Base webhook payload."""

    event: str
    timestamp: str
    meeting_id: str

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        return {
            "event": self.event,
            "timestamp": self.timestamp,
            "meeting_id": self.meeting_id,
        }


@dataclass(frozen=True, slots=True)
class MeetingCompletedPayload(WebhookPayload):
    """Payload for meeting.completed event."""

    title: str
    duration_seconds: float
    segment_count: int
    has_summary: bool

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        return {
            **super().to_dict(),
            "title": self.title,
            "duration_seconds": self.duration_seconds,
            "segment_count": self.segment_count,
            "has_summary": self.has_summary,
        }


@dataclass(frozen=True, slots=True)
class SummaryGeneratedPayload(WebhookPayload):
    """Payload for summary.generated event."""

    title: str
    executive_summary: str
    key_points_count: int
    action_items_count: int

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        return {
            **super().to_dict(),
            "title": self.title,
            "executive_summary": self.executive_summary,
            "key_points_count": self.key_points_count,
            "action_items_count": self.action_items_count,
        }
```

---

### Task 2: Webhook Executor

**File**: `src/noteflow/infrastructure/integrations/webhooks.py`

```python
"""Webhook execution infrastructure."""

from __future__ import annotations

import asyncio
import hashlib
import hmac
import json
import logging
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any
from uuid import uuid4

import httpx

from noteflow.domain.webhooks.events import (
    WebhookConfig,
    WebhookDelivery,
    WebhookEventType,
)

if TYPE_CHECKING:
    pass

_logger = logging.getLogger(__name__)


class WebhookExecutor:
    """Execute webhooks with retry logic and HMAC signing.

    Implements exponential backoff for failed deliveries and
    supports optional HMAC-SHA256 signature verification.
    """

    DEFAULT_TIMEOUT = 10.0
    DEFAULT_MAX_RETRIES = 3
    BACKOFF_BASE = 2.0

    def __init__(
        self,
        max_retries: int = DEFAULT_MAX_RETRIES,
        timeout_seconds: float = DEFAULT_TIMEOUT,
    ) -> None:
        """Initialize webhook executor.

        Args:
            max_retries: Maximum delivery attempts.
            timeout_seconds: HTTP request timeout.
        """
        self._max_retries = max_retries
        self._timeout = timeout_seconds
        self._client: httpx.AsyncClient | None = None

    async def _ensure_client(self) -> httpx.AsyncClient:
        """Lazy-initialize HTTP client."""
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=self._timeout)
        return self._client

    async def deliver(
        self,
        config: WebhookConfig,
        event_type: WebhookEventType,
        payload: dict[str, Any],
    ) -> WebhookDelivery:
        """Deliver webhook with retries.

        Args:
            config: Webhook configuration.
            event_type: Type of event being delivered.
            payload: Event payload data.

        Returns:
            Delivery record with status information.
        """
        if not config.enabled:
            return self._create_delivery(
                config=config,
                event_type=event_type,
                payload=payload,
                status_code=None,
                error_message="Webhook disabled",
                attempt_count=0,
            )

        if event_type not in config.events:
            return self._create_delivery(
                config=config,
                event_type=event_type,
                payload=payload,
                status_code=None,
                error_message=f"Event {event_type.value} not subscribed",
                attempt_count=0,
            )

        headers = self._build_headers(config, event_type, payload)
        client = await self._ensure_client()

        last_error: str | None = None
        attempt = 0

        for attempt in range(1, self._max_retries + 1):
            try:
                _logger.debug(
                    "Webhook delivery attempt %d/%d to %s",
                    attempt,
                    self._max_retries,
                    config.url,
                )

                response = await client.post(
                    config.url,
                    json=payload,
                    headers=headers,
                )

                return self._create_delivery(
                    config=config,
                    event_type=event_type,
                    payload=payload,
                    status_code=response.status_code,
                    error_message=None if response.is_success else response.text[:500],
                    attempt_count=attempt,
                )

            except httpx.TimeoutException:
                last_error = "Request timed out"
                _logger.warning("Webhook timeout (attempt %d): %s", attempt, config.url)

            except httpx.RequestError as e:
                last_error = str(e)
                _logger.warning(
                    "Webhook request error (attempt %d): %s - %s",
                    attempt,
                    config.url,
                    e,
                )

            # Exponential backoff before retry
            if attempt < self._max_retries:
                delay = self.BACKOFF_BASE ** (attempt - 1)
                await asyncio.sleep(delay)

        return self._create_delivery(
            config=config,
            event_type=event_type,
            payload=payload,
            status_code=None,
            error_message=f"Max retries exceeded: {last_error}",
            attempt_count=attempt,
        )

    def _build_headers(
        self,
        config: WebhookConfig,
        event_type: WebhookEventType,
        payload: dict[str, Any],
    ) -> dict[str, str]:
        """Build HTTP headers for webhook request."""
        headers = {
            "Content-Type": "application/json",
            "X-NoteFlow-Event": event_type.value,
            "X-NoteFlow-Delivery": str(uuid4()),
        }

        if config.secret:
            body = json.dumps(payload, separators=(",", ":"))
            signature = hmac.new(
                config.secret.encode(),
                body.encode(),
                hashlib.sha256,
            ).hexdigest()
            headers["X-NoteFlow-Signature"] = f"sha256={signature}"

        return headers

    def _create_delivery(
        self,
        config: WebhookConfig,
        event_type: WebhookEventType,
        payload: dict[str, Any],
        status_code: int | None,
        error_message: str | None,
        attempt_count: int,
    ) -> WebhookDelivery:
        """Create a delivery record."""
        return WebhookDelivery(
            id=uuid4(),
            webhook_id=config.id,
            event_type=event_type,
            payload=payload,
            status_code=status_code,
            error_message=error_message,
            attempt_count=attempt_count,
            delivered_at=datetime.now(timezone.utc),
        )

    async def close(self) -> None:
        """Close HTTP client and release resources."""
        if self._client is not None:
            await self._client.aclose()
            self._client = None
```

---

### Task 3: Webhook Application Service

**File**: `src/noteflow/application/services/webhook_service.py`

```python
"""Webhook application service."""

from __future__ import annotations

import logging
from datetime import datetime, timezone
from typing import TYPE_CHECKING

from noteflow.domain.webhooks.events import (
    MeetingCompletedPayload,
    SummaryGeneratedPayload,
    WebhookConfig,
    WebhookDelivery,
    WebhookEventType,
)
from noteflow.infrastructure.integrations.webhooks import WebhookExecutor

if TYPE_CHECKING:
    from noteflow.domain.entities.meeting import Meeting

_logger = logging.getLogger(__name__)


class WebhookService:
    """Orchestrates webhook delivery for meeting events.

    Manages webhook configurations and coordinates delivery
    across all registered webhooks for each event type.
    """

    def __init__(
        self,
        executor: WebhookExecutor | None = None,
    ) -> None:
        """Initialize webhook service.

        Args:
            executor: Webhook executor instance (created if not provided).
        """
        self._executor = executor or WebhookExecutor()
        self._configs: list[WebhookConfig] = []

    def register_webhook(self, config: WebhookConfig) -> None:
        """Register a webhook configuration.

        Args:
            config: Webhook configuration to register.
        """
        self._configs.append(config)
        _logger.info(
            "Registered webhook: %s for events: %s",
            config.url,
            [e.value for e in config.events],
        )

    def unregister_webhook(self, webhook_id: str) -> bool:
        """Unregister a webhook by ID.

        Args:
            webhook_id: UUID of webhook to remove.

        Returns:
            True if webhook was found and removed.
        """
        initial_count = len(self._configs)
        self._configs = [c for c in self._configs if str(c.id) != webhook_id]
        return len(self._configs) < initial_count

    def get_webhooks(self) -> list[WebhookConfig]:
        """Get all registered webhooks."""
        return list(self._configs)

    async def trigger_meeting_completed(
        self,
        meeting: Meeting,
    ) -> list[WebhookDelivery]:
        """Trigger webhooks for meeting completion.

        Args:
            meeting: Completed meeting entity.

        Returns:
            List of delivery records for all webhook attempts.
        """
        payload = MeetingCompletedPayload(
            event=WebhookEventType.MEETING_COMPLETED.value,
            timestamp=datetime.now(timezone.utc).isoformat(),
            meeting_id=str(meeting.id),
            title=meeting.title or "Untitled",
            duration_seconds=meeting.duration_seconds or 0.0,
            segment_count=len(meeting.segments),
            has_summary=meeting.summary is not None,
        )

        return await self._deliver_to_all(
            WebhookEventType.MEETING_COMPLETED,
            payload.to_dict(),
        )

    async def trigger_summary_generated(
        self,
        meeting: Meeting,
    ) -> list[WebhookDelivery]:
        """Trigger webhooks for summary generation.

        Args:
            meeting: Meeting with generated summary.

        Returns:
            List of delivery records for all webhook attempts.
        """
        summary = meeting.summary
        payload = SummaryGeneratedPayload(
            event=WebhookEventType.SUMMARY_GENERATED.value,
            timestamp=datetime.now(timezone.utc).isoformat(),
            meeting_id=str(meeting.id),
            title=meeting.title or "Untitled",
            executive_summary=summary.executive_summary if summary else "",
            key_points_count=len(summary.key_points) if summary else 0,
            action_items_count=len(summary.action_items) if summary else 0,
        )

        return await self._deliver_to_all(
            WebhookEventType.SUMMARY_GENERATED,
            payload.to_dict(),
        )

    async def trigger_recording_started(
        self,
        meeting_id: str,
        title: str,
    ) -> list[WebhookDelivery]:
        """Trigger webhooks for recording start.

        Args:
            meeting_id: ID of meeting being recorded.
            title: Meeting title.

        Returns:
            List of delivery records.
        """
        payload = {
            "event": WebhookEventType.RECORDING_STARTED.value,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "meeting_id": meeting_id,
            "title": title,
        }

        return await self._deliver_to_all(
            WebhookEventType.RECORDING_STARTED,
            payload,
        )

    async def trigger_recording_stopped(
        self,
        meeting_id: str,
        title: str,
        duration_seconds: float,
    ) -> list[WebhookDelivery]:
        """Trigger webhooks for recording stop.

        Args:
            meeting_id: ID of meeting.
            title: Meeting title.
            duration_seconds: Recording duration.

        Returns:
            List of delivery records.
        """
        payload = {
            "event": WebhookEventType.RECORDING_STOPPED.value,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "meeting_id": meeting_id,
            "title": title,
            "duration_seconds": duration_seconds,
        }

        return await self._deliver_to_all(
            WebhookEventType.RECORDING_STOPPED,
            payload,
        )

    async def _deliver_to_all(
        self,
        event_type: WebhookEventType,
        payload: dict,
    ) -> list[WebhookDelivery]:
        """Deliver event to all registered webhooks.

        Args:
            event_type: Type of event.
            payload: Event payload.

        Returns:
            List of delivery records.
        """
        deliveries: list[WebhookDelivery] = []

        for config in self._configs:
            try:
                delivery = await self._executor.deliver(config, event_type, payload)
                deliveries.append(delivery)

                if delivery.succeeded:
                    _logger.info(
                        "Webhook delivered: %s -> %s (status=%d)",
                        event_type.value,
                        config.url,
                        delivery.status_code,
                    )
                else:
                    _logger.warning(
                        "Webhook failed: %s -> %s (error=%s)",
                        event_type.value,
                        config.url,
                        delivery.error_message,
                    )

            except Exception:
                _logger.exception("Unexpected error delivering webhook to %s", config.url)

        return deliveries

    async def close(self) -> None:
        """Clean up resources."""
        await self._executor.close()
```

---

### Task 4: gRPC Service Injection (Three-Tier Pattern)

The NoteFlow gRPC layer uses a three-tier service injection pattern. All three tiers must be updated.

#### Tier 1: ServicerHost Protocol

**File**: `src/noteflow/grpc/_mixins/protocols.py`

```python
# Add to ServicerHost protocol class:
from noteflow.application.services.webhook_service import WebhookService

class ServicerHost(Protocol):
    # ... existing fields ...

    # Sprint 6: Webhook service
    _webhook_service: WebhookService | None
```

#### Tier 2: NoteFlowServicer

**File**: `src/noteflow/grpc/service.py`

```python
from noteflow.application.services.webhook_service import WebhookService

class NoteFlowServicer(
    # ... existing mixins ...
):
    def __init__(
        self,
        # ... existing params ...
        webhook_service: WebhookService | None = None,
    ) -> None:
        # ... existing assignments ...
        self._webhook_service = webhook_service
```

#### Tier 3: NoteFlowServer

**File**: `src/noteflow/grpc/server.py`

```python
from noteflow.application.services.webhook_service import WebhookService

class NoteFlowServer:
    def __init__(
        self,
        # ... existing params ...
        webhook_service: WebhookService | None = None,
    ) -> None:
        # ... existing assignments ...
        self._webhook_service = webhook_service

    async def start(self) -> None:
        # ... existing code ...
        self._servicer = NoteFlowServicer(
            # ... existing args ...
            webhook_service=self._webhook_service,
        )
```

#### Event Triggering in Mixins

**File**: `src/noteflow/grpc/_mixins/meeting.py`

Add to `StopMeeting` method after successful completion (after `return meeting_to_proto(meeting)`):

```python
# After successfully stopping the meeting, trigger webhook (fire-and-forget)
if self._webhook_service is not None:
    try:
        await self._webhook_service.trigger_meeting_completed(meeting)
    except Exception:
        _logger.exception("Failed to trigger meeting.completed webhooks")
        # Don't fail the RPC for webhook errors
```

**File**: `src/noteflow/grpc/_mixins/summarization.py`

Add to `GenerateSummary` method after successful generation:

```python
# After successfully generating summary
if self._webhook_service is not None:
    try:
        await self._webhook_service.trigger_summary_generated(meeting)
    except Exception:
        _logger.exception("Failed to trigger summary.generated webhooks")
        # Don't fail the RPC for webhook errors
```

**File**: `src/noteflow/grpc/_mixins/streaming.py`

Add to `StreamAudioSegments` method when streaming begins:

```python
# After successfully starting audio stream (first audio chunk received)
if self._webhook_service is not None:
    try:
        await self._webhook_service.trigger_recording_started(
            meeting_id=meeting_id,
            title=meeting.title or "Untitled",
        )
    except Exception:
        _logger.exception("Failed to trigger recording.started webhook")
```

---

### Task 5: Server Initialization in run_server_with_config()

**File**: `src/noteflow/grpc/server.py`

Add to `run_server_with_config()` function:

```python
from noteflow.application.services.webhook_service import WebhookService
from noteflow.infrastructure.integrations.webhooks import WebhookExecutor
from noteflow.infrastructure.converters import WebhookConverter

# Create webhook service if feature enabled
webhook_service: WebhookService | None = None
settings = get_settings()

if settings.features.webhooks_enabled and session_factory is not None:
    logger.info("Initializing webhook service...")

    # Create executor with config from database settings
    webhook_executor = WebhookExecutor()

    # Create UoW factory for loading configs
    uow_factory = SqlAlchemyUnitOfWork.factory_from_settings(settings)

    # Create service with executor and UoW factory
    webhook_service = WebhookService(
        executor=webhook_executor,
        uow_factory=uow_factory,
    )

    # Load existing webhook configs from database
    async with uow_factory() as uow:
        if uow.supports_webhooks:
            configs = await uow.webhooks.get_all_enabled()
            for config in configs:
                webhook_service.register_webhook(config)
            logger.info("Loaded %d webhook configurations", len(configs))

    logger.info("Webhook service initialized")

# Create server with webhook service
server = NoteFlowServer(
    port=config.port,
    asr_model=config.asr.model,
    asr_device=config.asr.device,
    asr_compute_type=config.asr.compute_type,
    session_factory=session_factory,
    summarization_service=summarization_service,
    diarization_engine=diarization_engine,
    diarization_refinement_enabled=config.diarization.refinement_enabled,
    ner_service=ner_service,
    webhook_service=webhook_service,  # Add this
)

# Add cleanup handler
async def cleanup() -> None:
    if webhook_service is not None:
        await webhook_service.close()
    # ... existing cleanup ...
```

---

## Code Segments to Reuse

### Persistence Layer Patterns (CRITICAL)

**Location**: `src/noteflow/infrastructure/persistence/`

Follow existing patterns for consistency:

| Pattern | Reference Location | Usage |
|---------|-------------------|-------|
| ORM Model | `models/integrations/integration.py` | Copy structure for `WebhookConfigModel`, `WebhookDeliveryModel` |
| Repository | `repositories/meeting_repo.py` | Base class usage, async query patterns |
| Unit of Work | `unit_of_work.py` | Add webhook repository to UoW |
| Converters | `converters/orm_converters.py` | ORM ↔ domain entity conversion |

**Key classes to import**:
```python
from noteflow.infrastructure.persistence.models._base import Base
from noteflow.infrastructure.persistence.repositories._base import BaseRepository
from noteflow.domain.utils.time import utc_now
```

### Database Schema Reference

**Location**: `docker/db/schema.sql`

Follow the existing table patterns:
- UUID primary keys with `gen_random_uuid()`
- `created_at`/`updated_at` timestamp columns
- Foreign key constraints with `ON DELETE CASCADE`
- Trigger for `updated_at` using `noteflow.set_updated_at()`

### Existing Integration Models

**Location**: `models/integrations/integration.py`

Reference for:
- `IntegrationModel` structure (workspace scoping, status tracking)
- `IntegrationSyncRunModel` (tracking operation history)
- Relationship patterns between models

### HMAC Signing Pattern

**Location**: `src/noteflow/infrastructure/security/`

The keystore uses HMAC for token verification - similar pattern for webhook signatures.

### HTTP Client Pattern

**Location**: External reference - follow httpx best practices

```python
# Async context manager pattern for cleanup
async with httpx.AsyncClient() as client:
    response = await client.post(url, json=payload)
```

### Event Payload Building

**Location**: `src/noteflow/grpc/_mixins/converters.py`

Pattern for converting domain entities to serializable formats.

### Error Handling in gRPC

**Location**: `src/noteflow/grpc/_mixins/meeting.py`

```python
try:
    # Operation that might fail
except Exception:
    _logger.exception("Error message")
    # Don't fail the main RPC
```

---

## Dependencies

> ✅ **No changes needed** — `httpx>=0.27` already exists in `pyproject.toml`.

---

## Acceptance Criteria

### Functional

- [ ] Webhooks fire on meeting.completed event
- [ ] Webhooks fire on summary.generated event
- [ ] Webhooks fire on recording.started event
- [ ] Webhooks fire on recording.stopped event
- [ ] Failed webhooks retry with exponential backoff
- [ ] HMAC signature included when secret configured
- [ ] Disabled webhooks don't fire
- [ ] Unsubscribed events don't trigger delivery

### Technical

- [ ] Webhook failures don't fail primary RPC operations
- [ ] Delivery attempts logged at DEBUG level
- [ ] Delivery failures logged at WARNING level
- [ ] HTTP client properly closed on shutdown
- [ ] Payload size reasonable (< 10KB)

### Quality Gates

- [ ] `pytest tests/quality/` passes
- [ ] `ruff check src/noteflow` clean
- [ ] `mypy src/noteflow` clean
- [ ] No `# type: ignore` without justification
- [ ] All public functions have docstrings
- [ ] Test coverage > 80% for webhook module

---

## Test Plan

### Shared Fixtures (Add to tests/conftest.py)

Add these shared fixtures to `tests/conftest.py` for reuse across test files:

```python
from uuid import uuid4

from noteflow.domain.webhooks.events import WebhookConfig, WebhookEventType


@pytest.fixture
def webhook_config() -> WebhookConfig:
    """Create test webhook configuration."""
    return WebhookConfig.create(
        workspace_id=uuid4(),
        url="https://example.com/webhook",
        events=[WebhookEventType.MEETING_COMPLETED],
        secret="test-secret",
    )


@pytest.fixture
def webhook_config_all_events() -> WebhookConfig:
    """Create webhook subscribed to all events."""
    return WebhookConfig.create(
        workspace_id=uuid4(),
        url="https://example.com/webhook",
        events=[
            WebhookEventType.MEETING_COMPLETED,
            WebhookEventType.SUMMARY_GENERATED,
            WebhookEventType.RECORDING_STARTED,
            WebhookEventType.RECORDING_STOPPED,
        ],
        secret="test-secret",
    )
```

### Unit Tests

**File**: `tests/infrastructure/integrations/test_webhooks.py`

```python
import pytest
from unittest.mock import AsyncMock, patch

import httpx

from noteflow.domain.webhooks.events import WebhookConfig, WebhookEventType
from noteflow.infrastructure.integrations.webhooks import WebhookExecutor


# Use shared fixture from conftest.py: webhook_config

async def test_deliver_success(webhook_config: WebhookConfig) -> None:
    """Successful delivery returns 200 status."""
    executor = WebhookExecutor(max_retries=1)

    with patch.object(executor, "_ensure_client") as mock_client:
        mock_response = AsyncMock()
        mock_response.status_code = 200
        mock_response.is_success = True
        mock_client.return_value.post = AsyncMock(return_value=mock_response)

        delivery = await executor.deliver(
            webhook_config,
            WebhookEventType.MEETING_COMPLETED,
            {"meeting_id": "123"},
        )

        assert delivery.succeeded
        assert delivery.status_code == 200
        assert delivery.attempt_count == 1


async def test_deliver_disabled_webhook(webhook_config: WebhookConfig) -> None:
    """Disabled webhooks skip delivery."""
    disabled_config = WebhookConfig(
        id=webhook_config.id,
        url=webhook_config.url,
        events=webhook_config.events,
        enabled=False,
    )
    executor = WebhookExecutor()

    delivery = await executor.deliver(
        disabled_config,
        WebhookEventType.MEETING_COMPLETED,
        {"meeting_id": "123"},
    )

    assert not delivery.succeeded
    assert delivery.error_message == "Webhook disabled"
    assert delivery.attempt_count == 0


async def test_deliver_unsubscribed_event(webhook_config: WebhookConfig) -> None:
    """Unsubscribed events skip delivery."""
    executor = WebhookExecutor()

    delivery = await executor.deliver(
        webhook_config,
        WebhookEventType.SUMMARY_GENERATED,  # Not subscribed
        {"meeting_id": "123"},
    )

    assert not delivery.succeeded
    assert "not subscribed" in delivery.error_message


async def test_deliver_retries_on_failure(webhook_config: WebhookConfig) -> None:
    """Failed requests are retried with backoff."""
    executor = WebhookExecutor(max_retries=3)

    call_count = 0

    async def failing_post(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        raise httpx.TimeoutException("Timeout")

    with patch.object(executor, "_ensure_client") as mock_client:
        mock_client.return_value.post = failing_post

        delivery = await executor.deliver(
            webhook_config,
            WebhookEventType.MEETING_COMPLETED,
            {"meeting_id": "123"},
        )

        assert not delivery.succeeded
        assert delivery.attempt_count == 3
        assert "Max retries exceeded" in delivery.error_message


def test_hmac_signature_generation(webhook_config: WebhookConfig) -> None:
    """HMAC signature is correctly generated."""
    executor = WebhookExecutor()
    payload = {"meeting_id": "123"}

    headers = executor._build_headers(
        webhook_config,
        WebhookEventType.MEETING_COMPLETED,
        payload,
    )

    assert "X-NoteFlow-Signature" in headers
    assert headers["X-NoteFlow-Signature"].startswith("sha256=")
```

### Service Tests

**File**: `tests/application/test_webhook_service.py`

```python
import pytest
from unittest.mock import AsyncMock, Mock

from noteflow.application.services.webhook_service import WebhookService
from noteflow.domain.entities.meeting import Meeting, MeetingId
from noteflow.domain.webhooks.events import WebhookConfig, WebhookEventType


# Use shared fixtures from conftest.py: webhook_config, mock_uow

@pytest.fixture
def sample_meeting() -> Meeting:
    """Create test meeting for webhook tests."""
    return Meeting(
        id=MeetingId.create(),
        title="Test Meeting",
        duration_seconds=3600.0,
        segments=[],
        summary=None,
    )


async def test_trigger_meeting_completed(
    webhook_config: WebhookConfig,
    sample_meeting: Meeting,
) -> None:
    """Meeting completion triggers registered webhooks."""
    mock_executor = Mock()
    mock_executor.deliver = AsyncMock(return_value=Mock(succeeded=True))

    service = WebhookService(executor=mock_executor)
    service.register_webhook(webhook_config)

    deliveries = await service.trigger_meeting_completed(sample_meeting)

    assert len(deliveries) == 1
    mock_executor.deliver.assert_called_once()


async def test_no_webhooks_registered(sample_meeting: Meeting) -> None:
    """No deliveries when no webhooks registered."""
    service = WebhookService()

    deliveries = await service.trigger_meeting_completed(sample_meeting)

    assert deliveries == []


async def test_webhook_failure_does_not_raise(
    webhook_config: WebhookConfig,
    sample_meeting: Meeting,
) -> None:
    """Webhook delivery failure does not propagate exception."""
    mock_executor = Mock()
    mock_executor.deliver = AsyncMock(side_effect=Exception("Network error"))

    service = WebhookService(executor=mock_executor)
    service.register_webhook(webhook_config)

    # Should not raise despite executor failure
    deliveries = await service.trigger_meeting_completed(sample_meeting)

    assert deliveries == []  # No successful deliveries
```

### Integration Tests

**File**: `tests/integration/test_webhook_integration.py`

```python
@pytest.mark.integration
async def test_meeting_stop_triggers_webhook(
    grpc_client: NoteFlowClient,
    running_meeting: Meeting,
    webhook_server: MockWebhookServer,
) -> None:
    """StopMeeting RPC triggers webhook delivery."""
    # Stop the meeting
    await grpc_client.StopMeeting(
        noteflow_pb2.StopMeetingRequest(meeting_id=str(running_meeting.id))
    )

    # Verify webhook was called
    await asyncio.sleep(0.5)  # Allow async delivery
    assert webhook_server.received_events == ["meeting.completed"]
```

---

## Rollback Plan

If issues arise:

1. **Webhook failures**: Already isolated - don't fail primary RPCs
2. **HTTP client issues**: Replace httpx with aiohttp if needed
3. **Signature verification**: Make secret optional, graceful fallback
4. **Performance impact**: Add circuit breaker for repeated failures

---

## Security Considerations

### HMAC Signature Verification

Webhook receivers should verify signatures:

```python
import hmac
import hashlib

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify webhook signature."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)
```

### Best Practices

- Store webhook secrets securely (KeyStore)
- Use HTTPS endpoints only
- Validate URL format before registration
- Rate limit webhook registrations
- Log deliveries for audit trail

### Webhook Secret Storage

The existing ORM model stores `secret` as plaintext in `webhook_configs.secret`. Two approaches are available:

#### Option A: Use Existing Schema (Simpler)

Keep the current `secret` column as-is. This is acceptable for HMAC signing secrets because:
- Secrets are user-provided, not system-generated
- They're used for outbound signing, not authentication
- Database access already requires authentication

#### Option B: Use IntegrationSecretModel Pattern (More Secure)

Follow the existing `IntegrationSecretModel` pattern for encrypted storage:

**File**: `src/noteflow/infrastructure/persistence/models/integrations/webhook.py`

```python
class WebhookSecretModel(Base):
    """Store encrypted secrets for a webhook (follows IntegrationSecretModel pattern)."""

    __tablename__ = "webhook_secrets"
    __table_args__: ClassVar[dict[str, str]] = {"schema": "noteflow"}

    webhook_id: Mapped[PyUUID] = mapped_column(
        UUID(as_uuid=True),
        ForeignKey("noteflow.webhook_configs.id", ondelete="CASCADE"),
        primary_key=True,
    )
    secret_value: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        default=utc_now,
    )
```

**Recommendation**: Start with Option A for MVP. Migrate to Option B if security requirements increase.

---

## Post-Sprint

- Add webhook delivery history UI in Settings
- Implement webhook testing (manual trigger)
- Add Slack/Discord webhook presets
- Consider webhook batching for high-volume events
- Add dead letter queue for persistent failures
