fix: [BOUNTY] Webhook Event System

fix_issue_77.py

@@ -0,0 +1,3 @@
+```json
+{
+  "solution_code": "### File: app/webhooks/__init__.py\n```python\n# empty init\n```\n\n### File: app/webhooks/models.py\n```python\nfrom datetime import datetime\nfrom app.db import db\n\n\nclass WebhookEndpoint(db.Model):\n    __tablename__ = 'webhook_endpoints'\n\n    id = db.Column(db.Integer, primary_key=True)\n    user_id = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='CASCADE'), nullable=False)\n    url = db.Column(db.String(2048), nullable=False)\n    secret = db.Column(db.String(256), nullable=False)  # HMAC signing secret\n    events = db.Column(db.ARRAY(db.String), nullable=False, default=[])  # e.g. ['expense.created', 'bill.paid']\n    is_active = db.Column(db.Boolean, default=True, nullable=False)\n    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)\n    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)\n\n    deliveries = db.relationship('WebhookDelivery', backref='endpoint', lazy='dynamic', cascade='all, delete-orphan')\n\n\nclass WebhookDelivery(db.Model):\n    __tablename__ = 'webhook_deliveries'\n\n    id = db.Column(db.Integer, primary_key=True)\n    endpoint_id = db.Column(db.Integer, db.ForeignKey('webhook_endpoints.id', ondelete='CASCADE'), nullable=False)\n    event_type = db.Column(db.String(128), nullable=False)\n    payload = db.Column(db.JSON, nullable=False)\n    signature = db.Column(db.String(512), nullable=False)\n    attempt_count = db.Column(db.Integer, default=0, nullable=False)\n    max_attempts = db.Column(db.Integer, default=5, nullable=False)\n    status = db.Column(\n        db.String(32),\n        default='pending',\n        nullable=False\n    )  # pending | delivered | failed | exhausted\n    last_response_code = db.Column(db.Integer, nullable=True)\n    last_response_body = db.Column(db.Text, nullable=True)\n    next_attempt_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)\n    delivered_at = db.Column(db.DateTime, nullable=True)\n    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)\n```\n\n### File: app/webhooks/signing.py\n```python\nimport hashlib\nimport hmac\nimport json\nimport time\nfrom typing import Any, Dict\n\n\nSIGNATURE_VERSION = 'v1'\n\n\ndef build_signed_payload(payload: Dict[str, Any], secret: str, timestamp: int | None = None) -> tuple[str, str]:\n    \"\"\"\n    Build a signed payload string using HMAC-SHA256.\n    Returns (signature_header_value, canonical_string).\n\n    Header format:  t=<timestamp>,v1=<hex_digest>\n    Signed string:  <timestamp>.<json_payload>\n    \"\"\"\n    if timestamp is None:\n        timestamp = int(time.time())\n\n    body = json.dumps(payload, separators=(',', ':'), sort_keys=True)\n    signed_string = f\"{timestamp}.{body}\"\n\n    digest = hmac.new(\n        key=secret.encode('utf-8'),\n        msg=signed_string.encode('utf-8'),\n        digestmod=hashlib.sha256\n    ).hexdigest()\n\n    header_value = f\"t={timestamp},{SIGNATURE_VERSION}={digest}\"\n    return header_value, body\n\n\ndef verify_signature(raw_body: str, signature_header: str, secret: str, tolerance_seconds: int = 300) -> bool:\n    \"\"\"\n    Verify an inbound webhook signature.\n    Raises ValueError on malformed header.\n    Returns False if timestamp is outside tolerance or digest doesn't match.\n    \"\"\"\n    parts = dict(item.split('=', 1) for item in signature_header.split(','))\n    timestamp_str = parts.get('t')\n    provided_digest = parts.get(SIGNATURE_VERSION)\n\n    if not timestamp_str or not provided_digest:\n        raise ValueError('Malformed signature header')\n\n    timestamp = int(timestamp_str)\n    now = int(time.time())\n    if abs(now - timestamp) > tolerance_seconds:\n        return False\n\n    signed_string = f\"{timestamp}.{raw_body}\"\n    expected_digest = hmac.new(\n        key=secret.encode('utf-8'),\n        msg=signed_string.encode('utf-8'),\n        digestmod=hashlib.sha256\n    ).hexdigest()\n\n    return hmac.compare_digest(expected_digest, provided_digest)\n```\n\n### File: app/webhooks/events.py\n```python\n\"\"\"\nCanonical event-type registry.\n\nEvent types follow the pattern: <resource>.<action>\n\nSupported events\n----------------\nexpense.created       A new expense was recorded.\nexpense.updated       An expense was edited.\nexpense.deleted       An expense was removed.\nbill.created          A new bill was created.\nbill.updated          A bill was edited.\nbill.paid             A bill was marked as paid.\nbill.deleted          A bill was removed.\nreminder.triggered    A reminder notification was dispatched.\nbudget.alert          Spending crossed a budget threshold (emitted by insights).\n\"\"\"\n\nALL_EVENT_TYPES = [\n    'expense.created',\n    'expense.updated',\n    'expense.deleted',\n    'bill.created',\n    'bill.updated',\n    'bill.paid',\n    'bill.deleted',\n    'reminder.triggered',\n    'budget.alert',\n]\n\n\ndef validate_event_types(events: list[str]) -> list[str]:\n    \"\"\"Return list of invalid event type strings.\"\"\"\n    return [e for e in events if e not in ALL_EVENT_TYPES]\n```\n\n### File: app/webhooks/dispatcher.py\n```python\nimport logging\nfrom datetime import datetime\nfrom typing import Any, Dict\n\nimport requests\nfrom requests.exceptions import RequestException\n\nfrom app.db import db\nfrom app.webhooks.models import WebhookDelivery, WebhookEndpoint\nfrom app.webhooks.signing import build_signed_payload\n\nlogger = logging.getLogger(__name__)\n\nREQUEST_TIMEOUT = 10  # seconds\n\n\ndef emit_event(user_id: int, event_type: str, data: Dict[str, Any]) -> None:\n    \"\"\"\n    Queue webhook deliveries for all active endpoints subscribed to event_type.\n    Call this inside request handlers after committing the triggering change.\n    \"\"\"\n    endpoints = (\n        WebhookEndpoint.query\n        .filter_by(user_id=user_id, is_active=True)\n        .filter(WebhookEndpoint.events.contains([event_type]))\n        .all()\n    )\n\n    if not endpoints:\n        return\n\n    payload = {\n        'event': event_type,\n        'created_at': datetime.utcnow().isoformat() + 'Z',\n        'data': data,\n    }\n\n    for endpoint in endpoints:\n        signature_header, _ = build_signed_payload(payload, endpoint.secret)\n        delivery = WebhookDelivery(\n            endpoint_id=endpoint.id,\n            event_type=event_type,\n            payload=payload,\n            signature=signature_header,\n            status='pending',\n            next_attempt_at=datetime.utcnow(),\n        )\n        db.session.add(delivery)\n\n    db.session.commit()\n    logger.info('Queued %d webhook deliveries for event=%s user=%s', len(endpoints), event_type, user_id)\n\n\ndef deliver_webhook(delivery: WebhookDelivery) -> bool:\n    \"\"\"\n    Attempt a single HTTP delivery. Returns True on success.\n    Updates delivery record in-place; caller must commit.\n    \"\"\"\n    endpoint = delivery.endpoint\n    payload_json = delivery.payload  # already a dict (stored as JSON)\n    signature_header, body_str = build_signed_payload(payload_json, endpoint.secret)\n\n    try:\n        response = requests.post(\n            endpoint.url,\n            data=body_str,\n            headers={\n                'Content-Type': 'application/json',\n                'X-FinMind-Signature': signature_header,\n                'X-FinMind-Event': delivery.event_type,\n                'X-FinMind-Delivery': str(delivery.id),\n            },\n            timeout=REQUEST_TIMEOUT,\n        )\n        delivery.attempt_count += 1\n        delivery.last_response_code = response.status_code\n        delivery.last_response_body = response.text[:1024]  # truncate\n\n        if response.ok:\n            delivery.status = 'delivered'\n            delivery.delivered_at = datetime.utcnow()\n            logger.info('Webhook delivered: delivery_id=%s endpoint=%s', delivery.id, endpoint.url)\n            return True\n        else:\n            logger.warning(\n                'Webhook non-2xx: delivery_id=%s status=%s',\n                delivery.id, response.status_code\n            )\n\n    except RequestException as exc:\n        delivery.attempt_count += 1\n        delivery.last_response_body = str(exc)[:1024]\n        logger.warning('Webhook request failed: delivery_id=%s error=%s', delivery.id, exc)\n\n    # Schedule retry with exponential back-off (60s * 2^attempt, cap 4h)\n    _schedule_retry(delivery)\n    return False\n\n\ndef _schedule_retry(delivery: WebhookDelivery) -> None:\n    from datetime import timedelta\n\n    if delivery.attempt_count >= delivery.max_attempts:\n        delivery.status = 'exhausted'\n        logger.error('Webhook exhausted retries: delivery_id=%s', delivery.id)\n        return\n\n    delivery.status = 'pending'\n    backoff_seconds = min(60 * (2 ** delivery.attempt_count), 14400)  # max 4h\n    delivery.next_attempt_at = datetime.utcnow() + timedelta(seconds=backoff_seconds)\n    logger.info(\n        'Webhook retry scheduled: delivery_id=%s attempt=%s next_at=%s',\n        delivery.id, delivery.attempt_count + 1, delivery.next_attempt_at\n    )\n```\n\n### File: app/webhooks/retry_job.py\n```python\n\"\"\"\nAPScheduler job: poll pending webhook deliveries and attempt delivery.\nRegister once in app factory via scheduler.add_job(...).\n\"\"\"\nimport logging\nfrom datetime import datetime\n\nfrom app.db import db\nfrom app.webhooks.dispatcher import deliver_webhook\nf