"""Usage alerts for AgentForms users.
Sends email alerts when users approach their tier limits:
- 75% of submission limit — warning
- 90% of submission limit — urgent warning
Run as a daily cron job or scheduled task.
"""
import os
import time
from app.services.logging import log
_APP_URL = os.environ.get("APP_URL", "https://agentforms.io")
def check_usage_alerts():
"""Check all users for usage alerts.
Returns:
List of users who received alerts today.
"""
from app.models import get_db
from app.routes.email import NotificationService
if not NotificationService.is_configured():
log.warning("usage_alerts", "SMTP not configured, skipping usage alerts")
return []
conn = get_db()
try:
# Get users with their usage for current month
from datetime import datetime
current_month = datetime.now().strftime("%Y-%m")
users = conn.execute(
"""
SELECT u.id, u.email_encrypted, u.tier, mu.submission_count
FROM users u
JOIN monthly_usage mu ON u.id = mu.user_id
WHERE mu.month = ?
""",
(current_month,),
).fetchall()
from app.models import TIERS, try_decrypt_user_value
alerted = []
for user in users:
user_id = user[0]
email = try_decrypt_user_value(user_id, user[1])
tier = user[2]
submission_count = user[3]
if not email or not tier:
continue
tier_config = TIERS.get(tier, TIERS["free"])
max_subs = tier_config.get("max_submissions", 0)
if max_subs <= 0:
continue # Unlimited tier
usage_pct = (submission_count / max_subs) * 100
# Check if we already alerted today
last_alert = conn.execute(
"SELECT last_alert_pct FROM usage_alerts WHERE user_id = ? AND DATE(last_alerted_at) = DATE('now')",
(user_id,),
).fetchone()
if last_alert and last_alert[0] >= usage_pct:
continue # Already alerted at this or higher level today
if usage_pct >= 90:
subject = f"Urgent: You're at {usage_pct:.0f}% of your submission limit"
body = f"""Hey there!
You've used {submission_count} of your {max_subs} submissions this month ({usage_pct:.0f}%).
If you hit the limit, new form submissions will be blocked until next month.
Upgrade now to avoid interruption:
{_APP_URL}/billing/upgrade
The AgentForms Team
---
To unsubscribe, visit: {_APP_URL}/settings/"""
elif usage_pct >= 75:
subject = f"You're using {usage_pct:.0f}% of your submission limit"
body = f"""Hey there!
Just a heads up — you've used {submission_count} of your {max_subs} submissions this month ({usage_pct:.0f}%).
Consider upgrading to avoid hitting the limit:
{_APP_URL}/billing/upgrade
The AgentForms Team
---
To unsubscribe, visit: {_APP_URL}/settings/"""
else:
continue
result = NotificationService.send(to=email, subject=subject, body=body)
if result:
# Record the alert
try:
conn.execute(
"INSERT OR IGNORE INTO usage_alerts (user_id, last_alerted_at, last_alert_pct) VALUES (?, datetime('now'), ?)",
(user_id, usage_pct),
)
conn.execute(
"UPDATE usage_alerts SET last_alert_pct = ?, last_alerted_at = datetime('now') WHERE user_id = ? AND DATE(last_alerted_at) != DATE('now')",
(usage_pct, user_id),
)
alerted.append({"user_id": user_id, "email": email, "usage_pct": usage_pct})
log.info("usage_alerts", f"Alerted user {user_id} at {usage_pct:.0f}%")
except Exception as e:
log.error("usage_alerts", f"Failed to record alert: {e}")
else:
log.error("usage_alerts", f"Failed to send alert to {email}")
conn.commit()
return alerted
finally:
conn.close()
def start_usage_alert_worker():
"""Start the usage alert background worker thread."""
import threading
import time
thread = threading.Thread(
target=_usage_alert_worker,
name="usage-alert-worker",
daemon=True,
)
thread.start()
log.info("usage_alerts", "Usage alert worker started")
def _usage_alert_worker():
"""Background thread that checks usage alerts daily."""
from app.services.logging import log
log.info("usage_alerts", "Usage alert worker started")
while True:
try:
alerted = check_usage_alerts()
if alerted:
log.info("usage_alerts", f"Sent {len(alerted)} usage alert(s)")
except Exception as e:
log.error("usage_alerts", f"Error checking usage alerts: {e}")
# Check every 6 hours
time.sleep(6 * 3600)