"""Email campaign service — SMTP via existing Postfix relay.
Uses the existing NotificationService (app/routes/email.py) which sends via
the configured SMTP relay (mail.agentforms.io on Hetzner). Renders templates
with Jinja2 and tracks delivery results in the campaigns tables.
"""
import logging
import os
import threading
import time
import urllib.parse
from datetime import datetime, timezone
logger = logging.getLogger("agentforms.campaigns")
logger.setLevel(logging.INFO)
if not logger.handlers:
_h = logging.StreamHandler(__import__("sys").stderr)
_h.setFormatter(logging.Formatter("[campaigns] %(message)s"))
logger.addHandler(_h)
# Default sender
DEFAULT_FROM_NAME = os.environ.get("EMAIL_FROM_NAME", "AgentForms")
DEFAULT_FROM_EMAIL = os.environ.get("EMAIL_FROM_EMAIL", "noreply@agentforms.io")
# Rate limit: seconds between sends (SMTP relay: ~1/sec is fine)
RATE_LIMIT_DELAY = float(os.environ.get("EMAIL_RATE_LIMIT_DELAY", "0.5"))
# Base URL for form links
BASE_URL = os.environ.get("APP_URL", "https://agentforms.io")
def _render_campaign_html(campaign, recipient=None):
"""Render the campaign email HTML body using the new template renderer.
Uses app.emails.renderer.EmailTemplateRenderer for personalization
support ({{recipient.email}}, {{recipient.first_name}}, {{form.field}}).
Falls back to default template if no template_id specified.
"""
from app.emails.renderer import _get_renderer
template_id = campaign.get("template_id")
body = campaign.get("body", "")
subject = campaign.get("subject", "")
name = campaign.get("name", "")
# Build personalization context
context = {
"subject": subject,
"body": body,
"form": {
"name": name,
"subject": subject,
"body": body,
},
"base_url": BASE_URL,
}
# Add recipient data for personalization
if recipient:
context["recipient"] = {
"email": recipient.get("email", ""),
"first_name": recipient.get("first_name", ""),
"last_name": recipient.get("last_name", ""),
"name": recipient.get("first_name", "") or recipient.get("email", "").split("@")[0],
**{
k: v
for k, v in recipient.items()
if k
not in (
"id",
"campaign_id",
"user_id",
"form_id",
"status",
"sent_at",
"tracking_token",
"error",
"created_at",
)
},
}
else:
context["recipient"] = {}
try:
renderer = _get_renderer()
template_name = template_id if template_id else "default"
html_body = renderer.render(template_name, context)
return subject, html_body
except Exception as e:
logger.warning("Template render failed for %s: %s, using fallback", template_id or "default", e)
# Default HTML wrapper — inline styles for email client compatibility
html_body = f"""<!DOCTYPE html>
<html>
<head>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
</head>
<body style="margin:0;padding:0;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;background-color:#f4f4f5;">
<table role="presentation" style="width:100%;max-width:600px;margin:0 auto;background-color:#ffffff;">
<tr><td style="padding:32px 24px;background-color:#f8f9fa;border-radius:8px 8px 0 0;">
<h2 style="margin:0;color:#111827;font-size:20px;">{name}</h2>
</td></tr>
<tr><td style="padding:24px;">
<p style="margin:0 0 16px 0;color:#374151;font-size:14px;line-height:1.6;">{body}</p>
</td></tr>
</table>
</body>
</html>"""
return subject, html_body
def _inject_tracking(html_body, tracking_token, base_url):
"""Inject open tracking pixel and rewrite URLs for click tracking.
Args:
html_body: HTML email body
tracking_token: Unique token for this recipient
base_url: Base URL of the application
Returns:
HTML with tracking injected
"""
import re
# Track all rewritten URLs to avoid double-rewriting
tracking_urls = set()
# Rewrite all href attributes to go through click tracking
def rewrite_url(match):
url = match.group(1)
# Skip mailto:, tel:, javascript:, and already-tracked URLs
if url.lower().startswith(("mailto:", "tel:", "javascript:")):
return match.group(0)
if url in tracking_urls:
return match.group(0)
encoded_url = urllib.parse.quote(url, safe=":/?#[]@!$&'()*+,;=-._~")
tracked_url = f"{base_url}/tracking/click/{tracking_token}?url={encoded_url}"
tracking_urls.add(url)
return f'href="{tracked_url}"'
html_body = re.sub(r'href="([^"]+)"', rewrite_url, html_body, flags=re.IGNORECASE)
html_body = re.sub(r"href='([^']+)'", rewrite_url, html_body, flags=re.IGNORECASE)
# Inject open tracking pixel before </body>
tracking_pixel = (
f'<img src="{base_url}/tracking/open/{tracking_token}" '
f'alt="" width="1" height="1" style="display:none;border:0;" />'
)
if "</body>" in html_body:
html_body = html_body.replace("</body>", f"{tracking_pixel}</body>")
else:
html_body += tracking_pixel
return html_body
def send_campaign_email(campaign_id, recipient):
"""Send a single email for a campaign recipient. Returns (success, error).
Args:
campaign_id: Campaign ID
recipient: Dict from get_pending_recipients()
Returns:
(True, None) on success, (False, error_msg) on failure
"""
from app.models import get_campaign, update_recipient_status
from app.routes.email import NotificationService
campaign = get_campaign(campaign_id, recipient["user_id"])
if not campaign:
return False, "campaign not found"
if not NotificationService.is_configured():
return False, "SMTP not configured"
recipient_id = recipient["id"]
to_email = recipient["email"]
from_name = campaign.get("from_name") or DEFAULT_FROM_NAME
from_email = campaign.get("from_email") or DEFAULT_FROM_EMAIL
from_addr = f"{from_name} <{from_email}>"
# Render template with recipient context for personalization
subject, html_body = _render_campaign_html(campaign, recipient)
# Inject tracking pixel and rewrite URLs
tracking_token = recipient.get("tracking_token")
if tracking_token:
html_body = _inject_tracking(html_body, tracking_token, BASE_URL)
# Plain-text fallback
plain_body = campaign.get("body", "")
if plain_body:
# Strip HTML tags for plain text
import re
plain_body = re.sub(r"<[^>]+>", "", plain_body)
try:
success = NotificationService.send_html(
to=to_email,
subject=subject,
html_body=html_body,
plain_body=plain_body if plain_body else None,
from_addr=from_addr,
)
if success:
logger.info("Sent campaign %s email to %s", campaign_id, to_email)
update_recipient_status(campaign_id, recipient_id, "sent")
return True, None
else:
logger.warning("SMTP send failed for campaign %s recipient %s", campaign_id, recipient_id)
update_recipient_status(campaign_id, recipient_id, "failed", error="SMTP send failed")
return False, "SMTP send failed"
except Exception as e:
error_msg = str(e)
logger.error("Failed to send campaign %s email to %s: %s", campaign_id, to_email, error_msg)
update_recipient_status(campaign_id, recipient_id, "failed", error=error_msg)
return False, error_msg
def process_campaign(campaign_id):
"""Process all pending recipients for a campaign.
Called by RQ worker — sends emails one by one with rate limiting.
Returns dict with stats.
"""
from app.models import get_campaign_by_id, get_pending_recipients, update_campaign_status
campaign = get_campaign_by_id(campaign_id)
if not campaign:
logger.warning("Campaign %s not found", campaign_id)
return {"error": "campaign not found"}
if campaign.get("status") == "cancelled":
logger.info("Campaign %s cancelled, skipping", campaign_id)
return {"skipped": True, "reason": "cancelled"}
# Update status to sending
update_campaign_status(campaign_id, campaign["user_id"], "sending")
pending = get_pending_recipients(campaign_id)
stats = {"sent": 0, "failed": 0, "total": len(pending)}
for recipient in pending:
success, error = send_campaign_email(campaign_id, recipient)
if success:
stats["sent"] += 1
else:
stats["failed"] += 1
logger.warning("Campaign %s recipient %s failed: %s", campaign_id, recipient["email"], error)
# Rate limiting between sends
if stats["total"] > 1:
time.sleep(RATE_LIMIT_DELAY)
# Update campaign status
if stats["failed"] == stats["total"]:
final_status = "failed"
elif stats["sent"] == stats["total"]:
final_status = "sent"
else:
final_status = "sent" # Partial success still counts as sent
update_campaign_status(campaign_id, campaign["user_id"], final_status)
logger.info("Campaign %s complete: %s", campaign_id, stats)
return stats
def enqueue_campaign(campaign_id):
"""Queue a campaign for processing via RQ."""
import redis as redis_lib
from rq import Queue
redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
try:
r = redis_lib.from_url(redis_url)
r.ping()
q = Queue("campaigns", connection=r)
q.enqueue(
process_campaign,
args=(campaign_id,),
job_timeout=3600,
result_ttl=86400,
)
logger.info("Campaign %s queued for processing", campaign_id)
return True
except Exception as e:
logger.warning("RQ unavailable, falling back to thread: %s", e)
thread = threading.Thread(
target=process_campaign,
args=(campaign_id,),
daemon=True,
)
thread.start()
return False
# --- Phase 4: Reminder Campaigns & A/B Testing Workers ---
def process_reminder(reminder_id):
"""Process a single reminder: send emails to eligible recipients."""
from app.models import (
get_campaign_by_id,
get_reminder,
get_reminder_eligible_recipients,
mark_reminder_sent,
)
reminder = get_reminder(reminder_id, None)
if not reminder:
logger.warning("Reminder %s not found", reminder_id)
return {"error": "reminder not found"}
campaign = get_campaign_by_id(reminder["campaign_id"])
if not campaign:
logger.warning("Campaign %s not found for reminder %s", reminder["campaign_id"], reminder_id)
return {"error": "campaign not found"}
# Get eligible recipients (havent opened/clicked, havent received reminder)
recipients = get_reminder_eligible_recipients(reminder_id)
if not recipients:
logger.info("No eligible recipients for reminder %s", reminder_id)
mark_reminder_sent(reminder_id, 0)
return {"sent": 0}
stats = {"sent": 0, "failed": 0, "total": len(recipients)}
for recipient in recipients:
success, error = send_reminder_email(reminder, recipient)
if success:
stats["sent"] += 1
else:
stats["failed"] += 1
logger.warning("Reminder %s recipient %s failed: %s", reminder_id, recipient["email"], error)
if stats["total"] > 1:
time.sleep(RATE_LIMIT_DELAY)
mark_reminder_sent(reminder_id, stats["sent"])
# Mark recipients as having received reminder
from app.models import get_db
conn = get_db()
try:
for r in recipients:
conn.execute(
"UPDATE campaign_recipients SET reminder_sent = 1 WHERE id = ?",
(r["id"],),
)
conn.commit()
finally:
if conn:
conn.close()
logger.info("Reminder %s complete: %s", reminder_id, stats)
return stats
def send_reminder_email(reminder, recipient):
"""Send a reminder email to a single recipient. Returns (success, error)."""
try:
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
campaign_id = reminder["campaign_id"]
subject = render_reminder_subject(reminder, recipient)
body = render_reminder_body(reminder, recipient)
from app.app import app
with app.app_context():
smtp_host = app.config.get("SMTP_HOST", "localhost")
smtp_port = app.config.get("SMTP_PORT", 587)
smtp_user = app.config.get("SMTP_USER", "")
smtp_pass = app.config.get("SMTP_PASS", "")
smtp_tls = app.config.get("SMTP_TLS", True)
sender_domain = app.config.get("SENDER_DOMAIN", "")
from_name = app.config.get("SENDER_NAME", "AgentForms")
from_email = app.config.get("SENDER_EMAIL", "noreply@agentforms.io")
if sender_domain:
from_email = from_email.split("@")[0] + "@" + sender_domain
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_email}>"
msg["To"] = recipient["email"]
msg["Reply-To"] = app.config.get("SMTP_REPLY_TO", from_email)
msg["X-Campaign-Id"] = str(campaign_id)
msg["X-Reminder-Id"] = str(reminder["id"])
msg.attach(MIMEText(body, "html", "utf-8"))
import smtplib
server = smtplib.SMTP(smtp_host, smtp_port, timeout=30)
try:
if smtp_tls:
server.starttls()
if smtp_user and smtp_pass:
server.login(smtp_user, smtp_pass)
server.sendmail(from_email, recipient["email"], msg.as_string())
finally:
server.quit()
from app.models import get_db
conn = get_db()
try:
conn.execute(
"UPDATE campaign_recipients SET status = ? WHERE id = ?",
("sent", recipient["id"]),
)
conn.commit()
finally:
if conn:
conn.close()
return True, None
except Exception as e:
logger.error("Failed to send reminder email: %s", e, exc_info=True)
return False, str(e)
def render_reminder_subject(reminder, recipient):
subject = reminder.get("subject", "Reminder")
try:
subject = subject.format(
first_name=recipient.get("email", "").split("@")[0],
email=recipient.get("email", ""),
)
except (KeyError, IndexError):
pass
return subject
def render_reminder_body(reminder, recipient):
body = reminder.get("body", "")
try:
body = body.format(
first_name=recipient.get("email", "").split("@")[0],
email=recipient.get("email", ""),
)
except (KeyError, IndexError):
pass
return body
def check_pending_reminders():
"""Background job: process all due reminders."""
from app.models import get_pending_reminders
reminders = get_pending_reminders()
logger.info("Checking reminders: %s due", len(reminders))
results = []
for reminder in reminders:
result = process_reminder(reminder["id"])
results.append({"reminder_id": reminder["id"], "result": result})
return {"processed": len(results), "results": results}
def send_ab_variant_email(campaign_id, recipient, variant):
"""Send an A/B test variant email. Returns (success, error)."""
try:
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
subject = variant["subject"]
body = variant.get("body", "")
try:
subject = subject.format(
first_name=recipient.get("email", "").split("@")[0],
email=recipient.get("email", ""),
)
except (KeyError, IndexError):
pass
from app.app import app
with app.app_context():
smtp_host = app.config.get("SMTP_HOST", "localhost")
smtp_port = app.config.get("SMTP_PORT", 587)
smtp_user = app.config.get("SMTP_USER", "")
smtp_pass = app.config.get("SMTP_PASS", "")
smtp_tls = app.config.get("SMTP_TLS", True)
sender_domain = app.config.get("SENDER_DOMAIN", "")
from_name = app.config.get("SENDER_NAME", "AgentForms")
from_email = app.config.get("SENDER_EMAIL", "noreply@agentforms.io")
if sender_domain:
from_email = from_email.split("@")[0] + "@" + sender_domain
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"{from_name} <{from_email}>"
msg["To"] = recipient["email"]
msg["Reply-To"] = app.config.get("SMTP_REPLY_TO", from_email)
msg["X-Campaign-Id"] = str(campaign_id)
msg["X-Variant-Id"] = str(variant["id"])
msg["X-Variant-Label"] = variant.get("variant_label", "")
msg.attach(MIMEText(body, "html", "utf-8"))
import smtplib
server = smtplib.SMTP(smtp_host, smtp_port, timeout=30)
try:
if smtp_tls:
server.starttls()
if smtp_user and smtp_pass:
server.login(smtp_user, smtp_pass)
server.sendmail(from_email, recipient["email"], msg.as_string())
finally:
server.quit()
from app.models import get_db
conn = get_db()
try:
conn.execute(
"UPDATE campaign_recipients SET status = ?, sent_at = CURRENT_TIMESTAMP WHERE id = ?",
("sent", recipient["id"]),
)
conn.commit()
finally:
if conn:
conn.close()
conn = get_db()
try:
conn.execute(
"UPDATE campaign_ab_variants SET sent_count = sent_count + 1 WHERE id = ?",
(variant["id"],),
)
conn.commit()
finally:
if conn:
conn.close()
return True, None
except Exception as e:
logger.error("Failed to send A/B variant email: %s", e, exc_info=True)
return False, str(e)
def check_pending_ab_tests():
"""Background job: declare A/B test winners and promote."""
from app.models import check_pending_ab_tests as model_check_ab_tests
from app.models import send_ab_winner_to_remaining
declared = model_check_ab_tests()
if not declared:
return {"declared": 0}
logger.info("A/B tests declared: %s", len(declared))
promoted = 0
for item in declared:
campaign_id = item["campaign_id"]
user_id = item.get("user_id")
count = send_ab_winner_to_remaining(campaign_id, user_id)
promoted += count
return {
"declared": len(declared),
"promoted": promoted,
"campaigns": [item["campaign_id"] for item in declared],
}
def _reminder_worker():
"""Periodic thread that checks for pending reminders every 60s."""
while True:
try:
check_pending_reminders()
except Exception as e:
logger.error("Reminder worker error: %s", e, exc_info=True)
time.sleep(60)
def _ab_test_worker():
"""Periodic thread that checks for pending A/B tests every 120s."""
while True:
try:
check_pending_ab_tests()
except Exception as e:
logger.error("A/B test worker error: %s", e, exc_info=True)
time.sleep(120)
def start_reminder_worker():
"""Start the reminder background thread (call once on app init)."""
thread = threading.Thread(target=_reminder_worker, daemon=True, name="reminder-worker")
thread.start()
logger.info("Reminder worker started (60s interval)")
def start_ab_test_worker():
"""Start the A-B test background thread (call once on app init)."""
thread = threading.Thread(target=_ab_test_worker, daemon=True, name="ab-test-worker")
thread.start()
logger.info("A-B test worker started (120s interval)")