noteflow/docs/sprints/phase-3-integrations/sprint-6-webhooks/README.md

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)

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)

# 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)

# 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)

# 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

"""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)

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.

"""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

"""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

"""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

# Add to ServicerHost protocol class:
from noteflow.application.services.webhooks import WebhookService

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

    # Sprint 6: Webhook service
    _webhook_service: WebhookService | None

Tier 2: NoteFlowServicer

File: src/noteflow/grpc/service.py

from noteflow.application.services.webhooks 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

from noteflow.application.services.webhooks 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)):

# 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:

# 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:

# 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:

from noteflow.application.services.webhooks 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:

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

# 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

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:

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

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

import pytest
from unittest.mock import AsyncMock, Mock

from noteflow.application.services.webhooks 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

@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:

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

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