"""Webhook model functions — logs, destinations, retries, delivery."""
import json
import logging
from datetime import UTC, datetime, timezone
logger = logging.getLogger(__name__)
from app.crypto import (
decrypt_entity,
decrypt_user_value,
encrypt_entity,
encrypt_user_submission,
encrypt_user_value,
encrypt_value,
hash_value,
is_encrypted,
key_is_configured,
try_decrypt,
try_decrypt_entity,
try_decrypt_user_submission,
try_decrypt_user_value,
)
from app.db import get_db
# Import helpers
from app.helpers import (
_decrypt_webhook_dest,
_decrypt_webhook_log,
_table_exists,
)
# ─── Column-name allowlist helpers ──────────────────────────────────────
_WEBHOOK_DESTINATIONS_ALLOWED_COLUMNS = {
"url",
"active",
"events",
"secret",
"headers",
"retry_count",
"timeout",
"description",
"name",
"config",
"enabled",
"url_encrypted",
"url_hash",
}
def _validate_columns(columns: list, allowed: set) -> list:
"""Validate column names against an allowlist. Returns list of 'col = ?' clauses."""
valid = []
for col in columns:
if col in allowed:
valid.append(f"{col} = ?")
else:
logger.warning(f"Blocked column in UPDATE: {col} (not in allowlist)")
return valid
def log_webhook_attempt(site_id, submission_id, webhook_url, success, error=None, destination_id=None):
"""Log a webhook delivery attempt. Upserts if a pending/failed entry exists.
Returns the log entry ID.
Args:
site_id: Site ID.
submission_id: Submission ID.
webhook_url: The URL that was called (or will be called on retry).
success: Whether the delivery succeeded.
error: Error message if delivery failed.
destination_id: Optional FK to webhook_destinations. NULL for legacy webhooks.
"""
from datetime import timedelta
# Redact PII from error messages before logging
from app.services.webhook import redact_pii, redact_webhook_url
safe_url = redact_webhook_url(webhook_url)
safe_error = redact_pii(str(error) if error else "")
conn = None
try:
conn = get_db()
# Encrypt webhook URL before storing
url_encrypted = encrypt_entity("webhook_log", f"{site_id}_{submission_id}", webhook_url)
# Look up existing pending/failed entry — match by destination if present,
# otherwise fall back to URL-based matching (legacy).
if destination_id:
existing = conn.execute(
"""SELECT * FROM webhook_logs
WHERE site_id = ? AND submission_id = ? AND destination_id = ?
AND status IN ('pending', 'failed')
ORDER BY id DESC LIMIT 1""",
(site_id, submission_id, destination_id),
).fetchone()
else:
existing = conn.execute(
"""SELECT * FROM webhook_logs
WHERE site_id = ? AND submission_id = ? AND webhook_url_encrypted = ?
AND status IN ('pending', 'failed')
ORDER BY id DESC LIMIT 1""",
(site_id, submission_id, url_encrypted),
).fetchone()
if existing:
existing = dict(existing)
attempt = existing["attempt_count"] + 1
if success:
conn.execute(
"UPDATE webhook_logs SET status = 'delivered', attempt_count = ?, last_error = ? WHERE id = ?",
(attempt, safe_error, existing["id"]),
)
else:
backoff_min = min(30 * (2 ** (attempt - 1)), 60)
next_retry = (datetime.now(UTC) + timedelta(minutes=backoff_min)).isoformat()
conn.execute(
"UPDATE webhook_logs SET status = 'failed', attempt_count = ?, last_error = ?, next_retry_at = ? WHERE id = ?",
(attempt, safe_error, next_retry, existing["id"]),
)
conn.commit()
return existing["id"]
else:
status = "delivered" if success else "failed"
next_retry = None
if not success:
next_retry = (datetime.now(UTC) + timedelta(minutes=30)).isoformat()
cursor = conn.execute(
"""INSERT INTO webhook_logs (site_id, submission_id, webhook_url_encrypted, destination_id, attempt_count, status, last_error, next_retry_at)
VALUES (?, ?, ?, ?, 1, ?, ?, ?)""",
(site_id, submission_id, url_encrypted, destination_id, status, safe_error, next_retry),
)
log_id = cursor.lastrowid
conn.commit()
return log_id
finally:
if conn:
conn.close()
def claim_webhook_retries(thread_name, max_retries=5, limit=10):
"""Atomically claim failed webhook logs for retry by this thread.
Uses UPDATE...RETURNING for row-level locking — only one thread
can claim each row, preventing duplicate deliveries.
Returns the claimed rows as dicts with decrypted webhook_url.
"""
conn = None
try:
conn = get_db()
now = datetime.now(UTC).isoformat()
rows = conn.execute(
"""UPDATE webhook_logs
SET status = 'retrying',
retrying_by = ?
WHERE id IN (
SELECT id FROM webhook_logs
WHERE status = 'failed'
AND attempt_count < ?
AND next_retry_at <= ?
ORDER BY created_at ASC
LIMIT ?
)
RETURNING *""",
(thread_name, max_retries, now, limit),
).fetchall()
return [_decrypt_webhook_log(dict(r)) for r in rows]
finally:
if conn:
conn.close()
def release_webhook_retry(log_id, new_status, error=None):
"""Release a claimed webhook retry row by updating its status.
If new_status is 'delivered', mark as delivered.
Otherwise revert to 'failed' with updated attempt_count and error.
"""
conn = None
try:
conn = get_db()
if new_status == "delivered":
conn.execute(
"UPDATE webhook_logs SET status = 'delivered', retrying_by = NULL WHERE id = ?",
(log_id,),
)
else:
# Revert to failed, reset retrying_by so another thread can claim
conn.execute(
"UPDATE webhook_logs SET status = 'failed', retrying_by = NULL, last_error = ? WHERE id = ?",
(error, log_id),
)
conn.commit()
finally:
if conn:
conn.close()
def get_pending_webhook_retries(max_retries=5):
"""Get failed webhook logs ready for retry (past next_retry_at).
Deprecated — use claim_webhook_retries() for atomic row locking.
Kept for backward compatibility with direct calls.
"""
conn = None
try:
conn = get_db()
now = datetime.now(UTC).isoformat()
rows = conn.execute(
"""SELECT * FROM webhook_logs
WHERE status = 'failed'
AND attempt_count < ?
AND next_retry_at <= ?
ORDER BY created_at ASC
LIMIT 10""",
(max_retries, now),
).fetchall()
return [_decrypt_webhook_log(dict(r)) for r in rows]
finally:
if conn:
conn.close()
def get_permanently_failed_webhooks(max_retries=5):
"""Return webhook logs that have exhausted all retries.
These entries have attempt_count >= max_retries and status 'failed'.
Used for alerting so the operator knows deliveries are permanently down.
"""
conn = None
try:
conn = get_db()
rows = conn.execute(
"""SELECT * FROM webhook_logs
WHERE status = 'failed'
AND attempt_count >= ?
ORDER BY created_at DESC
LIMIT 50""",
(max_retries,),
).fetchall()
return [_decrypt_webhook_log(dict(r)) for r in rows]
finally:
if conn:
conn.close()
def mark_webhook_delivered(log_id):
"""Mark a webhook log entry as successfully delivered."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE webhook_logs SET status = 'delivered' WHERE id = ?",
(log_id,),
)
conn.commit()
finally:
if conn:
conn.close()
def get_site_webhook(site_id):
"""Get webhook URL for a site (for retries after restart)."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT webhook_url FROM sites WHERE id = ?",
(site_id,),
).fetchone()
if not row:
return None
result = dict(row)
result["webhook_url"] = try_decrypt(site_id, result.get("webhook_url"))
return result
finally:
if conn:
conn.close()
# ─── Webhook destinations ─────────────────────────────────────────────────────
def get_site_destinations(site_id):
"""Get all webhook destinations for a site."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"SELECT * FROM webhook_destinations WHERE site_id = ? ORDER BY created_at ASC",
(site_id,),
).fetchall()
finally:
if conn:
conn.close()
return [_decrypt_webhook_dest(dict(r)) for r in rows]
def create_destination(site_id, name, dest_type, url, config=None):
"""Create a new webhook destination. Returns destination dict."""
conn = None
try:
conn = get_db()
config_json = json.dumps(config) if config else None
url_hash = hash_value(url) if url else None
conn.execute(
"INSERT INTO webhook_destinations (site_id, name, type, url, url_encrypted, url_hash, config) VALUES (?, ?, ?, ?, ?, ?, ?)",
(site_id, name, dest_type, url, None, url_hash, config_json),
)
conn.commit()
dest_id = conn.lastrowid
# Encrypt URL after insert
if url:
enc = encrypt_entity("webhook_dest", dest_id, url)
if enc:
conn.execute(
"UPDATE webhook_destinations SET url_encrypted = ? WHERE id = ?",
(enc, dest_id),
)
conn.commit()
dest = conn.execute(
"SELECT * FROM webhook_destinations WHERE id = ?",
(dest_id,),
).fetchone()
return _decrypt_webhook_dest(dict(dest))
finally:
if conn:
conn.close()
def update_destination(dest_id, **kwargs):
"""Update destination fields. Allowed: name, url, config, enabled."""
allowed = {"name", "url", "config", "enabled"}
fields = {k: v for k, v in kwargs.items() if k in allowed}
if not fields:
return False
conn = None
try:
conn = get_db()
set_parts = []
values = []
column_names = []
for k, v in fields.items():
if k == "url" and v:
set_parts.append("url = ?")
values.append(v)
enc = encrypt_entity("webhook_dest", dest_id, v)
if enc:
set_parts.append("url_encrypted = ?")
values.append(enc)
# Also update hash
url_hash = hash_value(v)
set_parts.append("url_hash = ?")
values.append(url_hash)
elif k == "config" and v is not None:
set_parts.append("config = ?")
values.append(json.dumps(v))
else:
column_names.append(k)
values.append(v)
# Validate non-special column names against allowlist
validated = _validate_columns(column_names, _WEBHOOK_DESTINATIONS_ALLOWED_COLUMNS)
set_parts.extend(validated)
set_clause = ", ".join(set_parts)
values.append(dest_id)
conn.execute(
f"UPDATE webhook_destinations SET {set_clause} WHERE id = ?",
values,
)
conn.commit()
return True
finally:
if conn:
conn.close()
def delete_destination(dest_id):
"""Delete a destination."""
conn = None
try:
conn = get_db()
conn.execute("DELETE FROM webhook_destinations WHERE id = ?", (dest_id,))
conn.commit()
finally:
if conn:
conn.close()
def get_destination(dest_id):
"""Get a single destination by ID."""
conn = None
try:
conn = get_db()
dest = conn.execute(
"SELECT * FROM webhook_destinations WHERE id = ?",
(dest_id,),
).fetchone()
finally:
if conn:
conn.close()
if dest:
return _decrypt_webhook_dest(dict(dest))
return None
def update_destination_stats(dest_id, success, error=None):
"""Increment success/failure count, update last_status, last_delivered_at."""
now = datetime.now(UTC).isoformat()
conn = None
try:
conn = get_db()
if success:
conn.execute(
"UPDATE webhook_destinations SET success_count = success_count + 1, last_status = 'success', last_error = NULL, last_delivered_at = ? WHERE id = ?",
(now, dest_id),
)
else:
status = error or "error"
conn.execute(
"UPDATE webhook_destinations SET failure_count = failure_count + 1, last_status = ?, last_error = ?, last_delivered_at = ? WHERE id = ?",
(status, error, now, dest_id),
)
conn.commit()
finally:
if conn:
conn.close()
def get_destination_delivery_stats(site_id):
"""Get delivery stats for all destinations of a site (for UI)."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"""SELECT id, name, type, enabled, last_status, last_error,
last_delivered_at, success_count, failure_count
FROM webhook_destinations
WHERE site_id = ?
ORDER BY last_delivered_at DESC NULLS LAST""",
(site_id,),
).fetchall()
finally:
if conn:
conn.close()
return [dict(r) for r in rows]