"""Usage tracking and tier enforcement — monthly limits, site counts, tier checks."""
import json
from datetime import datetime, timezone
from app.db import get_db
# Import helpers
from app.helpers import (
_check_spam_content,
_current_month,
_decrypt_submission_fields,
_decrypt_user_row,
_encrypt_submission_field,
_get_site_owner_password_hash,
_is_encrypted_value,
parse_user_agent,
resolve_geo,
)
from app.models_user import get_user_site_count
# Import TIERS from config
from app.models_config import TIERS
def get_monthly_submission_count(user_id):
"""Get submission count for the current month for a user."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT submission_count FROM monthly_usage WHERE user_id = ? AND month = ?",
(user_id, _current_month()),
).fetchone()
finally:
if conn:
conn.close()
return row["submission_count"] if row else 0
def increment_submission_count(user_id):
"""Increment the monthly submission counter.
DEPRECATED: Use `accept_submission` which atomically combines limit check,
submission insert, and counter increment in a single transaction.
"""
conn = None
try:
conn = get_db()
month = _current_month()
conn.execute(
"""INSERT INTO monthly_usage (user_id, month, submission_count)
VALUES (?, ?, 1)
ON CONFLICT(user_id, month) DO UPDATE SET submission_count = submission_count + 1""",
(user_id, month),
)
conn.commit()
finally:
if conn:
conn.close()
def can_create_site(user_id, user_tier):
"""Check if user can create another site based on tier limits."""
tier_config = TIERS.get(user_tier, TIERS["free"])
max_sites = tier_config["max_sites"]
if max_sites == 0:
return True # unlimited
current_count = get_user_site_count(user_id)
return current_count < max_sites
def accept_site_creation(name, owner_email, user_id, user_tier, smtp_from=None, field_config=None, webhook_url=None):
"""Atomically check site limit and insert new site.
All operations happen inside a single BEGIN IMMEDIATE transaction.
If the user is at their limit, nothing is inserted.
Args:
name: Site name.
owner_email: Owner email.
user_id: User owning the site.
user_tier: User's current tier string.
smtp_from: Optional SMTP from address.
field_config: Optional field config list.
webhook_url: Optional webhook URL.
Returns:
(success: bool, site: dict or None, current_count: int, max_sites: int)
"""
import secrets
tier_config = TIERS.get(user_tier, TIERS["free"])
max_sites = tier_config["max_sites"]
conn = None
try:
conn = get_db()
conn.execute("BEGIN IMMEDIATE")
# 1. Check limit under write lock
row = conn.execute(
"SELECT COUNT(*) as cnt FROM sites WHERE user_id = ?",
(user_id,),
).fetchone()
current_count = row["cnt"]
if max_sites > 0 and current_count >= max_sites:
conn.execute("ROLLBACK")
return False, None, current_count, max_sites
# 2. Insert the site
token = f"fr-{secrets.token_hex(6)}"
fc_json = json.dumps(field_config) if field_config else None
conn.execute(
"INSERT INTO sites (token, name, owner_email, smtp_from, user_id, field_config, webhook_url) VALUES (?, ?, ?, ?, ?, ?, ?)",
(token, name, owner_email, smtp_from, user_id, fc_json, webhook_url),
)
conn.commit()
# 3. Return the inserted site
site = conn.execute("SELECT * FROM sites WHERE token = ?", (token,)).fetchone()
return True, dict(site), current_count + 1, max_sites
except Exception:
conn.execute("ROLLBACK")
raise
finally:
if conn:
conn.close()
def can_accept_submission(user_id, user_tier):
"""Check if user's tier can accept more submissions this month.
Returns (allowed: bool, current_count, max_count).
"""
tier_config = TIERS.get(user_tier, TIERS["free"])
max_subs = tier_config["max_submissions"]
current = get_monthly_submission_count(user_id)
return current < max_subs, current, max_subs
def get_user_usage(user_id, user_tier):
"""Get full usage snapshot for a user.
Returns dict with site count, submission count, limits, and percentages.
"""
tier_config = TIERS.get(user_tier, TIERS["free"])
max_sites = tier_config["max_sites"]
max_subs = tier_config["max_submissions"]
site_count = get_user_site_count(user_id)
sub_count = get_monthly_submission_count(user_id)
return {
"tier": user_tier,
"tier_name": tier_config["name"],
"tier_price": tier_config["price"],
"site_count": site_count,
"max_sites": max_sites, # 0 = unlimited
"submissions": sub_count,
"max_submissions": max_subs,
"sites_pct": (site_count / max_sites * 100) if max_sites > 0 else 0,
"subs_pct": (sub_count / max_subs * 100) if max_subs > 0 else 0,
"sites_unlimited": max_sites == 0,
}
def seed_monthly_usage():
"""Ensure all users have a monthly_usage row for the current month.
Called on app startup and periodically. This is the automatic billing
cycle reset — when the month changes, all users get a fresh row with
submission_count=0.
"""
from datetime import timedelta
conn = None
try:
conn = get_db()
month = _current_month()
user_ids = [r["id"] for r in conn.execute("SELECT id FROM users").fetchall()]
for uid in user_ids:
conn.execute(
"""INSERT INTO monthly_usage (user_id, month, submission_count)
VALUES (?, ?, 0)
ON CONFLICT(user_id, month) DO NOTHING""",
(uid, month),
)
conn.commit()
finally:
if conn:
conn.close()