"""Email notifications via Resend, plus the retry-queue processor.
A notification row is the unit of delivery. `process_notification()` attempts one
send and records the outcome (status / attempts / last_error / provider_message_id)
so the dashboard can show delivery state and offer a manual resend. The background
worker (started in main.py) drains pending/failed rows with exponential backoff.
"""
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
from html import escape
import httpx
import deps
from db import execute, query_one
RESEND_ENDPOINT = "https://api.resend.com/emails"
BACKOFF_BASE_SECONDS = 60
BACKOFF_CAP_SECONDS = 3600
def _send_via_resend(to: str, subject: str, html: str, text: str,
from_addr: str, reply_to: str | None) -> dict:
"""Returns {ok, message_id, error}. In DEV_EMAIL_PRINT mode, logs and succeeds."""
if deps.DEV_EMAIL_PRINT:
print(
"\n===== [DEV_EMAIL_PRINT] email would be sent =====\n"
f"From: {from_addr}\nTo: {to}\nReply-To: {reply_to or '-'}\n"
f"Subject: {subject}\n---\n{text}\n================================================\n",
flush=True,
)
return {"ok": True, "message_id": "dev-print", "error": None}
payload = {"from": from_addr, "to": [to], "subject": subject, "html": html, "text": text}
if reply_to:
payload["reply_to"] = reply_to
try:
resp = httpx.post(
RESEND_ENDPOINT,
headers={"Authorization": f"Bearer {deps.RESEND_API_KEY}",
"Content-Type": "application/json"},
json=payload,
timeout=20,
)
if resp.status_code in (200, 201):
return {"ok": True, "message_id": resp.json().get("id"), "error": None}
return {"ok": False, "message_id": None,
"error": f"resend {resp.status_code}: {resp.text[:400]}"}
except Exception as exc: # noqa: BLE001 — record any transport error for retry
return {"ok": False, "message_id": None, "error": f"{type(exc).__name__}: {exc}"}
def _render_notification(form_row, sub_row, payload: dict) -> tuple[str, str, str]:
"""Build (subject, html, text) for an owner/recipient notification email.
Deliverability notes (these matter — form-notification mail is easy to mis-file):
* A COMPLETE html document (doctype + charset) — bare fragments score worse.
* Plain, conventional subject; no decorative punctuation.
* A "why you're receiving this" line, which is what distinguishes legitimate
transactional mail from bulk in both filters' and humans' eyes.
* text/plain mirrors the html exactly.
"""
form_name = form_row["name"]
site = sub_row["origin"] or ""
subject = f"New submission: {form_name}"
lines = [f'New submission on the "{form_name}" form'
+ (f" ({site})" if site else "") + ".", ""]
rows_html = []
for key, val in payload.items():
if key.startswith("_"):
continue
val_str = val if isinstance(val, str) else json.dumps(val)
lines.append(f"{key}: {val_str}")
rows_html.append(
f"| "
f"{escape(key)} | "
f""
f"{escape(val_str).replace(chr(10), '
')} |
"
)
why = (f'You are receiving this because you are listed as a notification '
f'recipient for the "{form_name}" form.')
lines += ["", f"Entry #{sub_row['id']} · IP {sub_row['submitter_ip'] or '-'}",
f"View in dashboard: {deps.APP_BASE_URL}/entries/{sub_row['id']}", "", why]
text = "\n".join(lines)
html = (
""
""
f"{escape(subject)}"
""
""
f"
New submission
"
f"
{escape(form_name)}"
+ (f" · {escape(site)}" if site else "") + "
"
f"
"
f"
Entry #{sub_row['id']} · "
f"IP {escape(sub_row['submitter_ip'] or '-')} · "
f"view in dashboard
{escape(why)}
"
"
{escape(text).replace(chr(10), '
')}
"
reply_to = None
else:
subject, html, text = _render_notification(form, sub, payload)
rt = payload.get(form["reply_to_field"])
reply_to = rt if isinstance(rt, str) and "@" in rt else None
attempt_no = n["attempts"] + 1
deps.log_event(f"send attempt {attempt_no} → {n['recipient']} ({n['kind']})",
submission_id=sub["id"], form_id=form["id"])
result = _send_via_resend(n["recipient"], subject, html, text, from_addr, reply_to)
if result["ok"]:
execute(
"UPDATE notifications SET status='sent', attempts=?, provider_message_id=?, "
"last_error=NULL, updated_at=CURRENT_TIMESTAMP WHERE id=?",
(attempt_no, result["message_id"], notif_id),
)
deps.log_event(f"delivered to {n['recipient']} (msg {result['message_id']})",
submission_id=sub["id"], form_id=form["id"])
return True
# Failure: schedule retry or give up.
delay = min(BACKOFF_BASE_SECONDS * (2 ** n["attempts"]), BACKOFF_CAP_SECONDS)
next_at = datetime.now(timezone.utc) + timedelta(seconds=delay)
give_up = attempt_no >= n["max_attempts"]
execute(
"UPDATE notifications SET status=?, attempts=?, last_error=?, next_attempt_at=?, "
"updated_at=CURRENT_TIMESTAMP WHERE id=?",
("failed", attempt_no, result["error"], None if give_up else next_at, notif_id),
)
deps.log_event(
f"send failed (attempt {attempt_no}): {result['error']}"
+ ("" if give_up else f" — retry in {delay}s"),
submission_id=sub["id"], form_id=form["id"], level="error",
data={"give_up": give_up},
)
return False
def enqueue_notification(submission_id: int, recipient: str, kind: str = "notify") -> int:
return execute(
"INSERT INTO notifications (submission_id, recipient, kind, next_attempt_at) "
"VALUES (?, ?, ?, CURRENT_TIMESTAMP)",
(submission_id, recipient, kind),
)