Feature Development Workflow

  # Feature: Order Status Webhooks

## What
Implement webhooks to notify merchants when order statuses change.

## Why
Merchants need real-time notifications to integrate with their fulfilment systems.
Currently, they poll our API every 5 minutes, causing unnecessary load.

## User Story
As a merchant, I want to receive webhook notifications when order statuses change,
so that I can update my fulfilment system in real-time without polling.

## Success Metrics
- 80% of merchants enable webhooks within 1 month
- API polling requests decrease by 60%
- Webhook delivery success rate > 95%

## Acceptance Criteria
- [ ] Merchants can register webhook URLs
- [ ] Webhooks fire on order status changes (pending → processing → completed)
- [ ] Webhooks include order ID, old status, new status, timestamp
- [ ] Webhook signature verification for security
- [ ] Retry mechanism (3 attempts with exponential backoff)
- [ ] Webhook logs viewable in merchant dashboard
  

  ## Feature Breakdown

### Must Have (MVP)
1. Webhook registration endpoint
2. Webhook firing on status change
3. Signature verification
4. Basic retry logic (3 attempts)

### Should Have
1. Webhook logs in dashboard
2. Test webhook endpoint
3. Webhook disable/enable toggle

### Nice to Have
1. Custom retry configuration
2. Webhook filtering (only specific statuses)
3. Multiple webhook URLs per merchant
  

  Table webhook_subscriptions {
  id integer [primary key, increment]
  user_id integer [not null, ref: > users.id]
  url varchar(500) [not null]
  is_active boolean [default: true]
  secret varchar(255) [not null]
  events varchar(100)[] [note: 'Array of event types to subscribe to']
  created_at timestamp [default: `now()`]
  updated_at timestamp

  indexes {
    user_id
    is_active
    (user_id, is_active) [name: 'idx_user_active']
  }
}

Table webhook_deliveries {
  id integer [primary key, increment]
  subscription_id integer [not null, ref: > webhook_subscriptions.id]
  event_type varchar(100) [not null]
  payload jsonb [not null]
  status varchar(50) [not null, default: 'pending']
  attempts integer [default: 0]
  last_attempt_at timestamp
  response_code integer
  response_body text
  created_at timestamp [default: `now()`]

  indexes {
    subscription_id
    status
    event_type
    created_at
    (subscription_id, status) [name: 'idx_subscription_status']
  }
}

Table orders {
  id integer [primary key]
  user_id integer [ref: > users.id]
  status varchar(50) [not null]
  // ... other fields
}
  

  ### Register Webhook Subscription

**Endpoint**: `POST /api/v1/webhooks/subscriptions`

**Authentication**: Required (JWT)

**Description**: Register a webhook subscription for order status changes.

**Request Body**:
```json
{
  "url": "https://merchant.example.com/webhooks/orders",
  "events": ["order.created", "order.status_changed"],
  "secret": "optional-custom-secret"  // Optional, auto-generated if not provided
}
  

  {
  "id": 123,
  "url": "https://merchant.example.com/webhooks/orders",
  "events": ["order.created", "order.status_changed"],
  "secret": "whsec_a1b2c3d4...",
  "is_active": true,
  "created_at": "2025-01-15T10:30:00Z"
}
  

  curl -X POST https://api.yourapp.com/api/v1/webhooks/subscriptions \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://merchant.example.com/webhooks", "events": ["order.status_changed"]}'
  

  
### Estimate Response Times

For each endpoint, estimate:
- Expected response time (target: <500ms)
- If >500ms, plan async processing

**Example**:

```markdown
| Endpoint | Expected Time | Approach |
|----------|---------------|----------|
| POST /webhooks/subscriptions | <100ms | Synchronous (just DB insert) |
| POST /webhooks/test | 2-5 seconds | Async (queue background job) |
| GET /webhooks/deliveries | <200ms | Synchronous (paginated query) |
  

  # app/models/webhook_subscription.py
from datetime import datetime
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, ARRAY
from sqlalchemy.orm import relationship

from app.db.base_class import Base


class WebhookSubscription(Base):
    """Webhook subscription for receiving event notifications."""

    __tablename__ = "webhook_subscriptions"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
    url = Column(String(500), nullable=False)
    is_active = Column(Boolean, default=True, index=True)
    secret = Column(String(255), nullable=False)
    events = Column(ARRAY(String), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, index=True)
    updated_at = Column(DateTime, onupdate=datetime.utcnow)

    # Relationships
    user = relationship("User", back_populates="webhook_subscriptions")
    deliveries = relationship("WebhookDelivery", back_populates="subscription")
  

  # app/schemas/webhook.py
from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, HttpUrl, Field


class WebhookSubscriptionCreate(BaseModel):
    """Schema for creating webhook subscription."""

    url: HttpUrl = Field(..., description="Webhook URL to receive notifications")
    events: List[str] = Field(..., description="Event types to subscribe to")
    secret: Optional[str] = Field(None, description="Custom webhook secret (auto-generated if not provided)")


class WebhookSubscriptionResponse(BaseModel):
    """Schema for webhook subscription response."""

    id: int
    url: str
    events: List[str]
    secret: str
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True
  

  # app/services/webhook_service.py
import secrets
from typing import List
from sqlalchemy.orm import Session

from app.models.webhook_subscription import WebhookSubscription
from app.schemas.webhook import WebhookSubscriptionCreate


class WebhookService:
    """Service for managing webhook subscriptions."""

    @staticmethod
    def create_subscription(
        db: Session,
        user_id: int,
        subscription_in: WebhookSubscriptionCreate
    ) -> WebhookSubscription:
        """
        Create a webhook subscription.

        Args:
            db: Database session
            user_id: ID of user creating subscription
            subscription_in: Subscription data

        Returns:
            WebhookSubscription: Created subscription
        """
        # Generate secret if not provided
        secret = subscription_in.secret or f"whsec_{secrets.token_urlsafe(32)}"

        subscription = WebhookSubscription(
            user_id=user_id,
            url=str(subscription_in.url),
            events=subscription_in.events,
            secret=secret,
            is_active=True,
        )

        db.add(subscription)
        db.commit()
        db.refresh(subscription)

        return subscription
  

  # app/api/v1/endpoints/webhooks.py
from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session

from app.api import deps
from app.schemas.webhook import WebhookSubscriptionCreate, WebhookSubscriptionResponse
from app.services.webhook_service import WebhookService

router = APIRouter()


@router.post("/subscriptions", response_model=WebhookSubscriptionResponse, status_code=status.HTTP_201_CREATED)
def create_webhook_subscription(
    subscription_in: WebhookSubscriptionCreate,
    db: Session = Depends(deps.get_db),
    current_user: User = Depends(deps.get_current_user),
) -> WebhookSubscriptionResponse:
    """
    Create a webhook subscription.

    Args:
        subscription_in: Subscription creation data
        db: Database session
        current_user: Authenticated user

    Returns:
        WebhookSubscriptionResponse: Created subscription
    """
    subscription = WebhookService.create_subscription(
        db,
        current_user.id,
        subscription_in
    )
    return subscription
  

  # tests/test_api/test_webhooks.py
def test_create_webhook_subscription(client, auth_headers):
    """Test creating webhook subscription."""
    response = client.post(
        "/api/v1/webhooks/subscriptions",
        json={
            "url": "https://example.com/webhook",
            "events": ["order.status_changed"]
        },
        headers=auth_headers
    )

    assert response.status_code == 201
    data = response.json()
    assert data["url"] == "https://example.com/webhook"
    assert "whsec_" in data["secret"]
    assert data["is_active"] is True
  

  feat(webhooks): add webhook subscription model
feat(webhooks): add webhook subscription schemas
feat(webhooks): add webhook service create method
feat(webhooks): add create subscription endpoint
test(webhooks): add subscription creation tests