linkbay-webhooks 0.1.0
Lightweight webhook event system for LinkBay CMS with pub/sub dispatcher and retry logic
License: MIT
Requires Python: >=3.8
Keywords: webhooks, fastapi, event-system, pub-sub, linkbay

# LinkBay Webhooks v0.1.0

[![License](https://img.shields.io/badge/license-MIT-blue)]() [![Python](https://img.shields.io/badge/python-3.8+-blue)]()

Libreria leggera per sistema di webhook con pub/sub dispatcher e retry logic per LinkBay CMS.

## Caratteristiche

- 🚀 **Event Dispatcher**: Pub/Sub pattern con supporto wildcards (`order.*`, `product.*`)
- 🔄 **Retry Logic**: 3 tentativi con exponential backoff (5s, 25s, 125s)
- 🔐 **HMAC Signature**: Firma HMAC-SHA256 per sicurezza webhook
- 📊 **Delivery Logs**: Tracciamento completo di ogni tentativo di consegna
- ⚡ **Rate Limiting**: Supporto per limitazione eventi (1000/ora per tenant)
- 🎯 **Zero Dependencies sul DB**: Libreria agnostica, implementa le tue interfacce di storage

## Prima di iniziare

**Questa libreria è adatta se:**
- Stai costruendo un sistema che deve inviare webhook agli utenti
- Hai bisogno di retry logic affidabile e configurabile
- Vuoi un sistema pub/sub con pattern matching
- Necessiti di sicurezza HMAC per webhook

**NON è adatta se:**
- Cerchi una soluzione chiavi-in-mano con DB incluso (questa è solo la logica)
- Non vuoi implementare le interfacce di storage

## Installazione

```bash
pip install linkbay-webhooks
```

Oppure l'ultima versione dal repository:

```bash
pip install git+https://github.com/AlessioQuagliara/linkbay-webhooks
```

## Architettura

La libreria è **completamente agnostica** rispetto al database e ai modelli. Fornisce:

1. **Interfacce astratte** (`WebhookStorage`, `RateLimiter`) che implementi con il tuo backend
2. **Logica di business**: dispatching, delivery, retry, HMAC signing
3. **Router FastAPI** pronto all'uso

**Non include**:
- ❌ Modelli SQLAlchemy/Pydantic per il database
- ❌ Migrazioni database
- ❌ Implementazioni di storage specifiche

Questo la rende leggera, flessibile e facile da integrare in qualsiasi progetto.

## Quick Start

### 1. Implementa lo Storage Backend

```python
from linkbay_webhooks import WebhookStorage, WebhookEndpoint, WebhookEvent, DeliveryLog
from typing import Optional

class MyWebhookStorage(WebhookStorage):
    """Implementazione con il tuo DB (SQL, MongoDB, etc.)"""
    
    async def get_endpoint(self, endpoint_id: str) -> Optional[WebhookEndpoint]:
        # Recupera dal tuo database
        row = await db.execute("SELECT * FROM webhooks WHERE id = ?", endpoint_id)
        if not row:
            return None
        return WebhookEndpoint(
            id=row.id,
            tenant_id=row.tenant_id,
            url=row.url,
            events=row.events,
            secret=row.secret,
            active=row.active,
        )
    
    async def create_endpoint(self, endpoint: WebhookEndpoint) -> WebhookEndpoint:
        # Salva nel tuo database
        await db.execute(
            "INSERT INTO webhooks (id, tenant_id, url, events, secret, active) VALUES (?, ?, ?, ?, ?, ?)",
            endpoint.id, endpoint.tenant_id, endpoint.url, endpoint.events, endpoint.secret, endpoint.active
        )
        return endpoint
    
    async def list_endpoints(self, tenant_id: str, active_only: bool = True) -> list[WebhookEndpoint]:
        query = "SELECT * FROM webhooks WHERE tenant_id = ?"
        if active_only:
            query += " AND active = true"
        rows = await db.execute(query, tenant_id)
        return [WebhookEndpoint(...) for row in rows]
    
    # ... implementa gli altri metodi
```

### 2. Setup FastAPI Application

```python
from fastapi import FastAPI, Depends
from linkbay_webhooks import EventDispatcher, create_webhook_router

app = FastAPI()

# Inizializza storage e dispatcher
storage = MyWebhookStorage()
dispatcher = EventDispatcher(storage)

# Funzione per ottenere tenant corrente
async def get_current_tenant() -> str:
    return "tenant_123"

# Crea e monta il router
webhook_router = create_webhook_router(
    storage=storage,
    dispatcher=dispatcher,
    get_current_tenant=get_current_tenant,
)

app.include_router(webhook_router)
```

### 3. Emetti Eventi

```python
# Emetti un evento
await dispatcher.emit(
    tenant_id="tenant_123",
    event_name="order.created",
    payload={
        "order_id": "ord_12345",
        "customer": "John Doe",
        "total": 99.99,
    }
)
```

## API Endpoints

```http
POST   /webhooks              # Crea endpoint webhook
GET    /webhooks              # Lista endpoint
GET    /webhooks/{id}         # Dettagli endpoint
PATCH  /webhooks/{id}         # Aggiorna endpoint
DELETE /webhooks/{id}         # Elimina endpoint
POST   /webhooks/{id}/test    # Testa endpoint
GET    /webhooks/deliveries   # Lista delivery logs
POST   /webhooks/deliveries/{id}/retry # Riprova delivery
```

## Event Patterns

```python
{
    "url": "https://myapp.com/webhooks",
    "events": [
        "order.*",           # Tutti gli eventi order
        "product.created",   # Solo product.created
        "*"                  # Tutti gli eventi
    ]
}
```

## Sicurezza HMAC

```python
from linkbay_webhooks import HMACSignature

# Nel tuo webhook receiver
def verify_webhook(request):
    signature = request.headers["X-LinkBay-Signature"]
    is_valid = HMACSignature.validate(
        payload=request.json(),
        secret=YOUR_SECRET,
        signature=signature
    )
    if not is_valid:
        raise HTTPException(status_code=401)
```

## Retry Logic

- **Timeout**: 30 secondi
- **Max retries**: 3 tentativi
- **Backoff**: Esponenziale (5s → 25s → 125s)

```python
from linkbay_webhooks import WebhookDelivery

delivery = WebhookDelivery(
    storage=storage,
    timeout=60,
    max_retries=5,
    backoff_base=10,
)
```

## Rate Limiting

```python
from linkbay_webhooks import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    max_events=1000,      # 1000 eventi
    window_seconds=3600,  # per ora
)

dispatcher = EventDispatcher(
    storage=storage,
    rate_limiter=rate_limiter,
)
```

## Payload Webhook

```json
{
    "event": "order.created",
    "tenant_id": "tenant_123",
    "event_id": "evt_xyz",
    "created_at": "2025-11-24T10:30:00Z",
    "data": {
        "order_id": "ord_12345",
        "customer": "John Doe",
        "total": 99.99
    }
}
```

## Testing

```python
import pytest
from linkbay_webhooks import EventDispatcher, HMACSignature

@pytest.mark.asyncio
async def test_webhook_delivery():
    storage = MockWebhookStorage()
    dispatcher = EventDispatcher(storage)
    
    event = await dispatcher.emit(
        tenant_id="test",
        event_name="test.event",
        payload={"test": True}
    )
    
    assert event.event_name == "test.event"
```

## Licenza

MIT

## LinkBay Ecosystem

Fa parte dell'ecosistema LinkBay CMS:
- `linkbay-core` - Core utilities
- `linkbay-multitenant` - Multi-tenancy support
- `linkbay-webhooks` - Event system (questa libreria)

## Limitazioni note


## Roadmap consigliata


## Licenza

MIT. Puoi usarla e modificarla liberamente, assicurandoti di valutarne i rischi nel tuo contesto.

## Supporto

- Issues: https://github.com/AlessioQuagliara/----/issues
- Email: quagliara.alessio@gmail.com

Contribuisci, apri una issue o raccontaci come stai usando la libreria 🧡