"""Onboarding email sequence for new AgentForms users.
Sends a series of emails after signup:
- Welcome + getting started (immediate)
- Tips: create your first form (after 1 day)
- Advanced features (after 3 days)
- Invite friends + referral (after 7 days)
Uses RQ delayed jobs for scheduling.
"""
import os
from app.services.logging import log
_APP_URL = os.environ.get("APP_URL", "https://agentforms.io")
def send_onboarding_email(user_id: int, step: str, email: str) -> bool:
"""Send an onboarding email step.
Args:
user_id: User ID
step: One of welcome, tips, features, invite
email: Recipient email address
Returns:
True if sent successfully, False otherwise
"""
from app.routes.email import NotificationService
if not NotificationService.is_configured():
log.warning("onboarding", "SMTP not configured, skipping onboarding emails")
return False
# Get referral code for invite email
from app.models import generate_referral_code
referral_code = generate_referral_code(user_id)
templates = {
"welcome": {
"subject": "Welcome to AgentForms — let's get started!",
"body": f"""Hello!
Welcome to AgentForms — the form builder that doesn't track you.
Here's what to do first:
1. Create your first form: {_APP_URL}/sites/new
2. Customize your form fields and design
3. Embed it on your website with the embed code
4. Get notified on every submission via email or webhooks
Need help? Check out our docs at {_APP_URL}/docs
Thanks for joining,
The AgentForms Team
P.S. You can invite friends and earn credits: {_APP_URL}/referral/
---
To unsubscribe from these emails, visit: {_APP_URL}/settings/""",
},
"tips": {
"subject": "Quick tip: make your forms stand out",
"body": f"""Hey there!
A few tips to make your forms more effective:
1. Keep it short — each extra field reduces submissions by ~10%
2. Use conditional logic to show relevant questions based on previous answers
3. Enable spam protection (honeypot + CAPTCHA) to reduce junk submissions
4. Customize your form design to match your brand
Check out our templates for inspiration: {_APP_URL}/sites/templates
The AgentForms Team
---
To unsubscribe, visit: {_APP_URL}/settings/""",
},
"features": {
"subject": "Did you know? AgentForms has A/B testing + AI form builder",
"body": f"""Hi!
You've been using AgentForms for a few days — here are some features you might not have tried:
• A/B Testing — test different form variants to see which converts better
• Document Generation — turn form submissions into invoices, contracts, or certificates
• Teams — collaborate with your team on form management
Upgrade to Starter to unlock AI and custom branding:
{_APP_URL}/billing/upgrade
The AgentForms Team
---
To unsubscribe, visit: {_APP_URL}/settings/""",
},
"invite": {
"subject": "Invite friends, earn credits",
"body": f"""Hey!
Love AgentForms? Invite friends and earn credits!
Your referral code: {referral_code}
Share this link: {_APP_URL}/auth/register?ref={referral_code}
When someone signs up with your code, you both get 50 credits.
View your referrals: {_APP_URL}/referral/
The AgentForms Team
---
To unsubscribe, visit: {_APP_URL}/settings/""",
},
}
if step not in templates:
log.warning("onboarding", f"Unknown step: {step}")
return False
template = templates[step]
result = NotificationService.send(
to=email,
subject=template["subject"],
body=template["body"],
)
if result:
log.info("onboarding", f"Sent {step} email to {email} (user {user_id})")
else:
log.error("onboarding", f"Failed to send {step} email to {email} (user {user_id})")
return result
def schedule_onboarding_sequence(user_id: int, email: str):
"""Schedule the full onboarding sequence for a new user.
Args:
user_id: User ID
email: Recipient email address
"""
import os
from redis import Redis
from rq import Queue
redis_url = os.environ.get("REDIS_URL")
if not redis_url:
log.warning("onboarding", "Redis not configured, skipping onboarding schedule")
return
try:
redis_conn = Redis.from_url(redis_url, decode_responses=True)
q = Queue("onboarding", connection=redis_conn)
# Send welcome email immediately
q.enqueue_at(
int(__import__("time").time()), # now
send_onboarding_email,
user_id,
"welcome",
email,
job_description=f"onboarding:welcome:{user_id}",
)
# Tips after 1 day
q.enqueue_at(
int(__import__("time").time()) + 86400,
send_onboarding_email,
user_id,
"tips",
email,
job_description=f"onboarding:tips:{user_id}",
)
# Advanced features after 3 days
q.enqueue_at(
int(__import__("time").time()) + 86400 * 3,
send_onboarding_email,
user_id,
"features",
email,
job_description=f"onboarding:features:{user_id}",
)
# Invite friends after 7 days
q.enqueue_at(
int(__import__("time").time()) + 86400 * 7,
send_onboarding_email,
user_id,
"invite",
email,
job_description=f"onboarding:invite:{user_id}",
)
log.info("onboarding", f"Scheduled onboarding sequence for user {user_id}")
except Exception as e:
log.error("onboarding", f"Failed to schedule onboarding: {e}")