75b95106ba
The retain outbox callback was passing context.tenant_id (raw UUID like 0f3ad4ec-8b88-...) instead of the PostgreSQL schema name (tenant_0f3ad4ec_...). This caused the webhook manager to query a non-existent schema, triggering a PostgreSQL error that silently aborted the entire retain transaction — rolling back all inserted memory data with no clear indication of data loss. Fixed both the async worker path (memory_engine.py) and sync HTTP path (http.py) to use _current_schema.get() which holds the correct tenant-prefixed schema name. Also changed fire_event_with_conn to re-raise exceptions instead of swallowing them, since errors inside a caller's transaction poison it irreversibly.
242 lines
9.1 KiB
Python
242 lines
9.1 KiB
Python
"""Webhook manager for delivering event notifications."""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import TYPE_CHECKING
|
|
|
|
import asyncpg
|
|
|
|
from .models import WebhookConfig, WebhookEvent, WebhookHttpConfig
|
|
|
|
if TYPE_CHECKING:
|
|
from hindsight_api.extensions.tenant import TenantExtension
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Retry delay schedule in seconds: 5 retries after the first attempt.
|
|
# Fast early retries catch transient failures; later retries handle longer outages.
|
|
RETRY_DELAYS = [5, 300, 1800, 7200, 18000]
|
|
MAX_ATTEMPTS = len(RETRY_DELAYS) + 1 # first attempt + len(RETRY_DELAYS) retries
|
|
|
|
|
|
def _fq_table(table: str, schema: str | None = None) -> str:
|
|
"""Get fully-qualified table name with optional schema prefix."""
|
|
if schema:
|
|
return f'"{schema}".{table}'
|
|
return table
|
|
|
|
|
|
def _parse_http_config(value: str | dict | None) -> WebhookHttpConfig:
|
|
"""Parse http_config column value (JSONB returned as text or dict) into a model."""
|
|
if value is None:
|
|
return WebhookHttpConfig()
|
|
if isinstance(value, str):
|
|
return WebhookHttpConfig.model_validate_json(value)
|
|
return WebhookHttpConfig.model_validate(value)
|
|
|
|
|
|
class WebhookManager:
|
|
"""
|
|
Manages webhook registration and event firing.
|
|
|
|
Supports both global webhooks (configured via env vars) and per-bank
|
|
webhooks stored in the database. Deliveries are queued as async_operations
|
|
tasks (operation_type='webhook_delivery') and picked up by the worker poller.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
pool: asyncpg.Pool,
|
|
global_webhooks: list[WebhookConfig],
|
|
tenant_extension: "TenantExtension | None" = None,
|
|
):
|
|
self._pool = pool
|
|
self._global_webhooks = global_webhooks
|
|
self._tenant_extension = tenant_extension
|
|
|
|
def _sign_payload(self, secret: str, payload_bytes: bytes) -> str:
|
|
"""Compute HMAC-SHA256 signature for a payload."""
|
|
return "sha256=" + hmac.new(secret.encode(), payload_bytes, hashlib.sha256).hexdigest()
|
|
|
|
async def fire_event(self, event: WebhookEvent, schema: str | None = None) -> None:
|
|
"""
|
|
Queue webhook deliveries for an event as async_operations tasks.
|
|
|
|
Loads per-bank and global webhooks, inserts pending webhook_delivery tasks for
|
|
any webhook whose event_types list matches the fired event type. The worker
|
|
poller picks these up and calls MemoryEngine._handle_webhook_delivery().
|
|
|
|
Args:
|
|
event: The event to deliver.
|
|
schema: Database schema (for multi-tenant). None = default schema.
|
|
"""
|
|
webhook_table = _fq_table("webhooks", schema)
|
|
ops_table = _fq_table("async_operations", schema)
|
|
now = datetime.now(timezone.utc)
|
|
payload_str = event.model_dump_json()
|
|
|
|
try:
|
|
# Load per-bank webhooks from DB (bank-specific + global NULL rows)
|
|
rows = await self._pool.fetch(
|
|
f"""
|
|
SELECT id, bank_id, url, secret, event_types, enabled, http_config::text
|
|
FROM {webhook_table}
|
|
WHERE (bank_id = $1 OR bank_id IS NULL) AND enabled = true
|
|
""",
|
|
event.bank_id,
|
|
)
|
|
|
|
db_webhooks = [
|
|
WebhookConfig(
|
|
id=str(row["id"]),
|
|
bank_id=row["bank_id"],
|
|
url=row["url"],
|
|
secret=row["secret"],
|
|
event_types=list(row["event_types"]) if row["event_types"] else [],
|
|
enabled=row["enabled"],
|
|
http_config=_parse_http_config(row["http_config"]),
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
# Merge with global webhooks from env config
|
|
all_webhooks = self._global_webhooks + db_webhooks
|
|
matched = 0
|
|
|
|
for webhook in all_webhooks:
|
|
if not webhook.enabled:
|
|
continue
|
|
if event.event.value not in webhook.event_types:
|
|
continue
|
|
|
|
operation_id = uuid.uuid4()
|
|
webhook_id = webhook.id if webhook.id else None
|
|
|
|
task_payload = json.dumps(
|
|
{
|
|
"type": "webhook_delivery",
|
|
"operation_id": str(operation_id),
|
|
"bank_id": event.bank_id,
|
|
"url": webhook.url,
|
|
"secret": webhook.secret,
|
|
"event_type": event.event.value,
|
|
"payload": payload_str,
|
|
"webhook_id": webhook_id,
|
|
"http_config": webhook.http_config.model_dump(),
|
|
}
|
|
)
|
|
|
|
await self._pool.execute(
|
|
f"""
|
|
INSERT INTO {ops_table}
|
|
(operation_id, bank_id, operation_type, status, task_payload, result_metadata, created_at, updated_at)
|
|
VALUES ($1, $2, 'webhook_delivery', 'pending', $3::jsonb, '{{}}'::jsonb, $4, $4)
|
|
""",
|
|
operation_id,
|
|
event.bank_id,
|
|
task_payload,
|
|
now,
|
|
)
|
|
matched += 1
|
|
|
|
logger.debug(f"Fired webhook event {event.event} for bank {event.bank_id}: {matched} delivery(ies) queued")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to queue webhook deliveries for event {event.event}: {e}")
|
|
|
|
async def fire_event_with_conn(
|
|
self, event: WebhookEvent, conn: asyncpg.Connection, schema: str | None = None
|
|
) -> None:
|
|
"""
|
|
Queue webhook deliveries within an existing database connection/transaction.
|
|
|
|
Identical to fire_event() but uses the provided connection instead of acquiring
|
|
one from the pool. Use this to atomically insert delivery tasks in the same
|
|
transaction as the primary operation (transactional outbox pattern).
|
|
|
|
Args:
|
|
event: The event to deliver.
|
|
conn: Existing asyncpg connection (may be inside an active transaction).
|
|
schema: Database schema (for multi-tenant). None = default schema.
|
|
"""
|
|
webhook_table = _fq_table("webhooks", schema)
|
|
ops_table = _fq_table("async_operations", schema)
|
|
now = datetime.now(timezone.utc)
|
|
payload_str = event.model_dump_json()
|
|
|
|
try:
|
|
rows = await conn.fetch(
|
|
f"""
|
|
SELECT id, bank_id, url, secret, event_types, enabled, http_config::text
|
|
FROM {webhook_table}
|
|
WHERE (bank_id = $1 OR bank_id IS NULL) AND enabled = true
|
|
""",
|
|
event.bank_id,
|
|
)
|
|
|
|
db_webhooks = [
|
|
WebhookConfig(
|
|
id=str(row["id"]),
|
|
bank_id=row["bank_id"],
|
|
url=row["url"],
|
|
secret=row["secret"],
|
|
event_types=list(row["event_types"]) if row["event_types"] else [],
|
|
enabled=row["enabled"],
|
|
http_config=_parse_http_config(row["http_config"]),
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
all_webhooks = self._global_webhooks + db_webhooks
|
|
matched = 0
|
|
|
|
for webhook in all_webhooks:
|
|
if not webhook.enabled:
|
|
continue
|
|
if event.event.value not in webhook.event_types:
|
|
continue
|
|
|
|
operation_id = uuid.uuid4()
|
|
webhook_id = webhook.id if webhook.id else None
|
|
|
|
task_payload = json.dumps(
|
|
{
|
|
"type": "webhook_delivery",
|
|
"operation_id": str(operation_id),
|
|
"bank_id": event.bank_id,
|
|
"url": webhook.url,
|
|
"secret": webhook.secret,
|
|
"event_type": event.event.value,
|
|
"payload": payload_str,
|
|
"webhook_id": webhook_id,
|
|
"http_config": webhook.http_config.model_dump(),
|
|
}
|
|
)
|
|
|
|
await conn.execute(
|
|
f"""
|
|
INSERT INTO {ops_table}
|
|
(operation_id, bank_id, operation_type, status, task_payload, result_metadata, created_at, updated_at)
|
|
VALUES ($1, $2, 'webhook_delivery', 'pending', $3::jsonb, '{{}}'::jsonb, $4, $4)
|
|
""",
|
|
operation_id,
|
|
event.bank_id,
|
|
task_payload,
|
|
now,
|
|
)
|
|
matched += 1
|
|
|
|
logger.debug(
|
|
f"Fired webhook event {event.event} for bank {event.bank_id}: {matched} delivery(ies) queued (in-transaction)"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Failed to queue webhook deliveries (in-transaction) for event {event.event}: {e}. "
|
|
"CRITICAL: The enclosing database transaction is now aborted and will roll back all changes."
|
|
)
|
|
raise
|