"""Email tracking endpoints — open pixel, click redirect, bounce webhook."""
import html
import urllib.parse
from flask import Blueprint, jsonify, make_response, redirect, request
from app.models import (
get_recipient_by_token,
record_click,
record_open,
)
from app.services.logging import log
tracking_bp = Blueprint("tracking", __name__)
@tracking_bp.route("/tracking/open/<token>", methods=["GET"])
def track_open(token):
"""1x1 transparent GIF — fires when email is opened."""
# First check campaign recipients
recipient = get_recipient_by_token(token)
if recipient:
record_open(recipient["id"])
log.info("tracking", f"Email opened: campaign={recipient['campaign_id']}, recipient={recipient['email']}")
else:
# Check form notification emails
from app.model_email_campaigns import get_notification_email_by_token, record_notification_open
notif = get_notification_email_by_token(token)
if notif:
record_notification_open(notif["id"])
log.info(
"tracking", f"Form notification opened: submission={notif['submission_id']}, user={notif['user_id']}"
)
# Return a 1x1 transparent GIF (22 bytes)
transparent_gif = (
b"\x47\x49\x46\x38\x39\x61\x01\x00\x01\x00\x80\x00\x00\x00\x00\x00"
b"\xff\x00\xff\x00\x00\xff\x00\xff\x21\xf9\x04\x01\x00\x00\x00\x00"
b"\x2c\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02\x44\x01\x00\x3b"
)
response = make_response(transparent_gif)
response.headers["Content-Type"] = "image/gif"
response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
response.headers["Pragma"] = "no-cache"
return response
@tracking_bp.route("/tracking/click/<token>", methods=["GET"])
def track_click(token):
"""Track a link click, then redirect to the destination URL."""
url = request.args.get("url", "")
if not url:
log.warning("tracking", f"Click tracked but no URL: token={token[:8]}...")
return redirect("/", code=302)
# First check campaign recipients
recipient = get_recipient_by_token(token)
if recipient:
record_click(recipient["id"])
log.info(
"tracking",
f"Link clicked: campaign={recipient['campaign_id']}, recipient={recipient['email']}, url={urllib.parse.unquote(url)[:200]}",
)
else:
# Check form notification emails
from app.model_email_campaigns import get_notification_email_by_token, record_notification_click
notif = get_notification_email_by_token(token)
if notif:
record_notification_click(notif["id"])
log.info(
"tracking",
f"Form notification clicked: submission={notif['submission_id']}, user={notif['user_id']}, url={urllib.parse.unquote(url)[:200]}",
)
# Validate URL before redirecting
try:
parsed = urllib.parse.urlparse(url)
if parsed.scheme in ("http", "https") and parsed.netloc:
return redirect(url, code=302)
else:
log.warning("tracking", f"Invalid redirect URL: {url[:200]}")
return redirect("/", code=302)
except Exception:
log.warning("tracking", f"Failed to parse redirect URL: {url[:200]}")
return redirect("/", code=302)
@tracking_bp.route("/tracking/bounce", methods=["POST"])
def track_bounce():
"""Receive bounce notifications from SMTP relay.
Accepts JSON with:
- email: recipient email address
- bounce_type: "hard" | "soft"
- reason: bounce reason text (optional)
Also parses RFC 3464 DSN format (multipart/report) from raw POST data.
"""
from app.models import get_campaign_recipients_by_email, record_bounce
data = request.get_json(silent=True)
if data:
email = data.get("email", "").strip().lower()
bounce_type = data.get("bounce_type", "hard")
bounce_reason = data.get("reason", "")
if not email:
return jsonify({"error": "Missing email"}), 400
# Find the recipient across all campaigns
recipients = get_campaign_recipients_by_email(email)
for recipient in recipients:
if recipient.get("status") in ("sent", "pending"):
record_bounce(recipient["id"], bounce_type, bounce_reason)
log.info(
"tracking",
f"Bounce recorded: campaign={recipient['campaign_id']}, "
f"recipient={email}, type={bounce_type}, reason={bounce_reason[:200]}",
)
return jsonify({"success": True, "bounced": len(recipients)})
# Fallback: try to parse raw RFC 3464 DSN
content_type = request.content_type or ""
if "message/rfc822" in content_type or "multipart/report" in content_type:
return jsonify({"warning": "Raw DSN parsing not yet supported — use JSON format"}), 200
return jsonify({"error": "Unsupported content type — send JSON"}), 400