"""User model functions — auth, profile, tokens, API keys, quotas, referral."""
import json
import logging
import secrets
import sqlite3
from datetime import UTC, datetime, timedelta, timezone
import bcrypt
from app.models import API_KEY_PREFIX
logger = logging.getLogger(__name__)
from app.crypto import (
decrypt_entity,
decrypt_user_value,
encrypt_entity,
encrypt_user_submission,
encrypt_user_value,
encrypt_value,
hash_value,
is_encrypted,
key_is_configured,
try_decrypt,
try_decrypt_entity,
try_decrypt_user_submission,
try_decrypt_user_value,
)
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,
)
# Import TIERS from config
from app.models_config import TIERS
from app.model_documents import count_user_documents
# ─── Column-name allowlist helpers ──────────────────────────────────────
_USERS_ALLOWED_COLUMNS = {
"name",
"email",
"password_hash",
"password_salt",
"role",
"tier",
"stripe_customer_id",
"profile_data",
"settings",
"api_key",
"avatar_url",
"language",
"timezone",
"phone",
"company",
"title",
"bio",
"social_links",
"notification_preferences",
"usage_limit",
"usage_count",
"created_at",
"updated_at",
}
def _validate_columns(columns: list, allowed: set) -> list:
"""Validate column names against an allowlist. Returns list of 'col = ?' clauses."""
valid = []
for col in columns:
if col in allowed:
valid.append(f"{col} = ?")
else:
logger.warning(f"Blocked column in UPDATE: {col} (not in allowlist)")
return valid
def get_user_tier(user_id):
"""Return the tier key for a user, defaulting to 'free'.
Returns:
tuple: (tier_key, tier_config) or ("free", TIERS["free"])
"""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT tier FROM users WHERE id = ?", (user_id,)).fetchone()
finally:
if conn:
conn.close()
if row and row["tier"] and row["tier"] in TIERS:
return row["tier"], TIERS[row["tier"]]
return "free", TIERS["free"]
def get_site_owner_tier(site_id):
"""Return the tier for a site's owner.
Returns:
tuple: (tier_key, tier_config) or ("free", TIERS["free"])
"""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT user_id FROM sites WHERE id = ?", (site_id,)).fetchone()
finally:
if conn:
conn.close()
if not row:
return "free", TIERS["free"]
return get_user_tier(row["user_id"])
def register_user(email, password, name=None, magic_login_enabled=True):
"""Register a new user. Returns user dict or raises on duplicate email."""
password_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
email = email.lower().strip()
email_hash = hash_value(email)
conn = None
try:
conn = get_db()
cursor = conn.execute(
"INSERT INTO users (email_hash, email_encrypted, password_hash, name_encrypted, tier, magic_login_enabled) "
"VALUES (?, ?, ?, ?, 'free', ?)",
(email_hash, None, password_hash, None, magic_login_enabled),
)
conn.commit()
user_id = cursor.lastrowid
# Encrypt and store
if email:
enc = encrypt_user_value(user_id, email)
if enc:
conn.execute("UPDATE users SET email_encrypted = ? WHERE id = ?", (enc, user_id))
if name:
enc = encrypt_user_value(user_id, name)
if enc:
conn.execute("UPDATE users SET name_encrypted = ? WHERE id = ?", (enc, user_id))
conn.commit()
user = conn.execute(
"SELECT id, email_encrypted, name_encrypted, password_hash, tier, magic_login_enabled, created_at "
"FROM users WHERE id = ?",
(user_id,),
).fetchone()
return _decrypt_user_row(dict(user))
except sqlite3.IntegrityError:
raise ValueError("Email already registered")
finally:
if conn:
conn.close()
# ─── User decryption helpers ──────────────────────────────────────────────────
def authenticate_user(email, password):
"""Verify credentials. Returns user dict or None.
Timing-safe: always runs bcrypt.checkpw even when user not found
to prevent email enumeration via response timing.
"""
email = email.lower().strip()
email_hash = hash_value(email)
conn = None
try:
conn = get_db()
row = conn.execute("SELECT * FROM users WHERE email_hash = ?", (email_hash,)).fetchone()
finally:
if conn:
conn.close()
if row is None:
# Run bcrypt against a dummy hash to prevent timing-based enumeration
bcrypt.checkpw(password.encode("utf-8"), b"$2b$12$92IXUNpkjO0rOQ5byMi.Ye4oKoEa3Ro9llC/.og/at2.uheWG/igi")
return None
row = dict(row)
if not bcrypt.checkpw(password.encode("utf-8"), row["password_hash"].encode("utf-8")):
return None
# Decrypt PII
email_val = try_decrypt_user_value(row["id"], row.get("email_encrypted"))
row["email"] = email_val
return row
def get_user(user_id):
"""Get user by ID (without password hash)."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT id, email_encrypted, email_hash, name_encrypted, display_name_encrypted, "
"tier, stripe_customer_id, stripe_subscription_id, "
"magic_login_enabled, email_verified, created_at FROM users WHERE id = ?",
(user_id,),
).fetchone()
finally:
if conn:
conn.close()
if not row:
return None
return _decrypt_user_row(dict(row))
def find_user_by_email(email):
"""Find user by email (with password hash for auth flows)."""
email = email.lower().strip()
email_hash = hash_value(email)
conn = None
try:
conn = get_db()
row = conn.execute("SELECT * FROM users WHERE email_hash = ?", (email_hash,)).fetchone()
finally:
if conn:
conn.close()
if not row:
return None
row = dict(row)
email_val = try_decrypt_user_value(row["id"], row.get("email_encrypted"))
row["email"] = email_val
return row
def update_user_tier(user_id, tier, stripe_subscription_id=None):
"""Update user tier and stripe subscription reference."""
conn = None
try:
conn = get_db()
if stripe_subscription_id:
conn.execute(
"UPDATE users SET tier = ?, stripe_subscription_id = ? WHERE id = ?",
(tier, stripe_subscription_id, user_id),
)
else:
conn.execute("UPDATE users SET tier = ? WHERE id = ?", (tier, user_id))
conn.commit()
finally:
if conn:
conn.close()
def update_user_magic_login_pref(user_id, enabled):
"""Enable or disable magic login for a user."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE users SET magic_login_enabled = ? WHERE id = ?",
(1 if enabled else 0, user_id),
)
conn.commit()
finally:
if conn:
conn.close()
def update_user_profile(user_id, name=None, display_name=None):
"""Update user profile fields. Returns dict with updated values."""
conn = None
try:
conn = get_db()
updates = []
params = []
if name is not None:
updates.append("name = ?")
params.append(name)
enc = encrypt_user_value(user_id, name)
if enc:
updates.append("name_encrypted = ?")
params.append(enc)
if display_name is not None:
updates.append("display_name = ?")
params.append(display_name)
enc = encrypt_user_value(user_id, display_name)
if enc:
updates.append("display_name_encrypted = ?")
params.append(enc)
if not updates:
conn.close()
return {"success": False, "error": "No fields to update"}
params.append(user_id)
conn.execute(f"UPDATE users SET {', '.join(updates)} WHERE id = ?", params)
conn.commit()
finally:
if conn:
conn.close()
return {"success": True}
def change_user_password(user_id, current_password, new_password):
"""Change user password after verifying current one. Returns success dict."""
if len(new_password) < 8:
return {"success": False, "error": "Password must be at least 8 characters"}
conn = None
try:
conn = get_db()
row = conn.execute("SELECT password_hash FROM users WHERE id = ?", (user_id,)).fetchone()
if not row:
conn.close()
return {"success": False, "error": "User not found"}
# Verify current password
if not bcrypt.checkpw(current_password.encode("utf-8"), row["password_hash"].encode("utf-8")):
conn.close()
return {"success": False, "error": "Current password is incorrect"}
# Set new password
new_hash = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (new_hash, user_id))
conn.commit()
finally:
if conn:
conn.close()
return {"success": True}
def change_user_email(user_id, current_password, new_email, new_password_hash=None):
"""Change user email. Requires current password verification.
Also updates the email_verification_token so the new email needs verification.
If new_password_hash is provided, sets it (for unauthenticated password resets).
"""
new_email = new_email.lower().strip()
conn = None
try:
conn = get_db()
# Verify current password
row = conn.execute("SELECT password_hash FROM users WHERE id = ?", (user_id,)).fetchone()
if not row:
conn.close()
return {"success": False, "error": "User not found"}
if not bcrypt.checkpw(current_password.encode("utf-8"), row["password_hash"].encode("utf-8")):
conn.close()
return {"success": False, "error": "Current password is incorrect"}
# Check email not taken (use email_hash)
existing_hash = hash_value(new_email)
existing = conn.execute(
"SELECT id FROM users WHERE email_hash = ? AND id != ?", (existing_hash, user_id)
).fetchone()
if existing:
conn.close()
return {"success": False, "error": "Email already in use by another account"}
# Generate verification token for new email
token = secrets.token_urlsafe(32)
email_enc = encrypt_user_value(user_id, new_email)
conn.execute(
"""UPDATE users SET email_hash = ?, email_encrypted = ?, email_verified = 0, email_verification_token = ?
WHERE id = ?""",
(existing_hash, email_enc, token, user_id),
)
conn.commit()
finally:
if conn:
conn.close()
return {"success": True, "verification_token": token}
def is_magic_login_enabled(user_id):
"""Check if magic login is enabled for a user."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT magic_login_enabled FROM users WHERE id = ?",
(user_id,),
).fetchone()
finally:
if conn:
conn.close()
if row is None:
return True # default enabled
return bool(row["magic_login_enabled"])
def update_stripe_customer(user_id, stripe_customer_id):
"""Store Stripe customer ID for a user."""
conn = None
try:
conn = get_db()
conn.execute(
"UPDATE users SET stripe_customer_id = ? WHERE id = ?",
(stripe_customer_id, user_id),
)
conn.commit()
finally:
if conn:
conn.close()
def get_user_site_count(user_id):
"""Count sites owned by a user."""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT COUNT(*) as cnt FROM sites WHERE user_id = ?", (user_id,)).fetchone()
finally:
if conn:
conn.close()
return row["cnt"] if row else 0
def get_user_chain_count(user_id):
"""Count action chains across all sites owned by a user."""
conn = None
try:
conn = get_db()
row = conn.execute(
"""SELECT COUNT(*) as cnt FROM form_actions fa
INNER JOIN sites s ON fa.site_id = s.id
WHERE s.user_id = ? AND fa.type = 'chain'""",
(user_id,),
).fetchone()
finally:
if conn:
conn.close()
return row["cnt"] if row else 0
def check_chain_quota(user_id):
"""Check if a user can create another chain based on tier limits.
Returns:
(allowed: bool, current_count: int, max_chains: int)
max_chains is -1 for unlimited, 0 for blocked.
"""
conn = None
try:
conn = get_db()
user_row = conn.execute("SELECT tier FROM users WHERE id = ?", (user_id,)).fetchone()
if not user_row:
return False, 0, 0
user_tier = dict(user_row).get("tier", "free")
tier_config = TIERS.get(user_tier, TIERS["free"])
max_chains = tier_config.get("max_chains", 0)
current_count = get_user_chain_count(user_id)
if max_chains == 0:
return False, current_count, 0
if max_chains == -1:
return True, current_count, -1
return current_count < max_chains, current_count, max_chains
finally:
if conn:
conn.close()
def get_user_document_count(user_id):
"""Count total documents for a user across all statuses."""
counts = count_user_documents(user_id)
return sum(counts.values())
def check_document_quota(user_id):
"""Check if a user can create another document based on tier limits.
Returns:
(allowed: bool, current_count: int, max_documents: int)
max_documents is -1 for unlimited, 0 for blocked.
"""
conn = None
try:
conn = get_db()
user_row = conn.execute("SELECT tier FROM users WHERE id = ?", (user_id,)).fetchone()
if not user_row:
return False, 0, 0
user_tier = dict(user_row).get("tier", "free")
tier_config = TIERS.get(user_tier, TIERS["free"])
max_documents = tier_config.get("max_documents", 0)
current_count = get_user_document_count(user_id)
if max_documents == 0:
return False, current_count, 0
if max_documents == -1:
return True, current_count, -1
return current_count < max_documents, current_count, max_documents
finally:
if conn:
conn.close()
def list_all_users():
"""List all users (admin)."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"SELECT id, email_encrypted, name_encrypted, tier, created_at FROM users ORDER BY created_at DESC"
).fetchall()
finally:
if conn:
conn.close()
result = []
for r in rows:
row = dict(r)
uid = row.get("id")
if row.get("email_encrypted"):
row["email"] = try_decrypt_user_value(uid, row["email_encrypted"])
if row.get("name_encrypted"):
row["name"] = try_decrypt_user_value(uid, row["name_encrypted"])
result.append(row)
return result
def generate_verification_token(user_id):
"""Generate and store an email verification token. Returns the token."""
token = secrets.token_urlsafe(32)
conn = None
try:
conn = get_db()
conn.execute(
"""UPDATE users SET email_verification_token = ?, email_verification_sent_at = CURRENT_TIMESTAMP
WHERE id = ?""",
(token, user_id),
)
conn.commit()
finally:
if conn:
conn.close()
return token
def verify_email_token(token):
"""Verify an email using the token. Returns user_id or None."""
conn = None
try:
conn = get_db()
row = conn.execute(
"""SELECT id, email_verification_sent_at FROM users
WHERE email_verification_token = ?""",
(token,),
).fetchone()
if not row:
conn.close()
return None
# Token expires after 24 hours
from datetime import timedelta
sent_at = row["email_verification_sent_at"]
if sent_at:
sent_dt = datetime.fromisoformat(sent_at).replace(tzinfo=UTC)
if datetime.now(UTC) - sent_dt > timedelta(hours=24):
conn.execute(
"UPDATE users SET email_verification_token = NULL WHERE id = ?",
(row["id"],),
)
conn.commit()
conn.close()
return None
# Mark as verified
conn.execute(
"""UPDATE users SET email_verified = 1, email_verification_token = NULL
WHERE id = ?""",
(row["id"],),
)
conn.commit()
finally:
if conn:
conn.close()
return row["id"]
# ─── Password reset ───────────────────────────────────────────────────────────
def generate_password_reset_token(email):
"""Generate a password reset token for a user. Returns token or None."""
email = email.lower().strip()
email_hash = hash_value(email)
conn = None
try:
conn = get_db()
user = conn.execute("SELECT id FROM users WHERE email_hash = ?", (email_hash,)).fetchone()
if not user:
conn.close()
return None
token = secrets.token_urlsafe(32)
conn.execute(
"""UPDATE users SET password_reset_token = ?, password_reset_sent_at = CURRENT_TIMESTAMP
WHERE id = ?""",
(token, user["id"]),
)
conn.commit()
finally:
if conn:
conn.close()
return token
def reset_password_with_token(token, new_password):
"""Reset a user's password using a token. Returns user_id or None."""
conn = None
try:
conn = get_db()
row = conn.execute(
"""SELECT id, password_reset_sent_at FROM users
WHERE password_reset_token = ?""",
(token,),
).fetchone()
if not row:
conn.close()
return None
# Token expires after 1 hour
sent_at = row["password_reset_sent_at"]
if sent_at:
from datetime import timedelta
sent_dt = datetime.fromisoformat(sent_at).replace(tzinfo=UTC)
if datetime.now(UTC) - sent_dt > timedelta(hours=1):
conn.execute(
"UPDATE users SET password_reset_token = NULL WHERE id = ?",
(row["id"],),
)
conn.commit()
conn.close()
return None
# Hash new password
pw_hash = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt())
conn.execute(
"""UPDATE users SET password_hash = ?, password_reset_token = NULL
WHERE id = ?""",
(pw_hash.decode("utf-8"), row["id"]),
)
conn.commit()
finally:
if conn:
conn.close()
return row["id"]
# ─── Magic login (passwordless) ───────────────────────────────────────────────
def generate_magic_login_token(email):
"""Generate a magic login token for a user. Returns token or None."""
email = email.lower().strip()
email_hash = hash_value(email)
conn = None
try:
conn = get_db()
user = conn.execute("SELECT id FROM users WHERE email_hash = ?", (email_hash,)).fetchone()
if not user:
conn.close()
return None
token = secrets.token_urlsafe(32)
conn.execute(
"""UPDATE users SET magic_login_token = ?, magic_login_sent_at = CURRENT_TIMESTAMP
WHERE id = ?""",
(token, user["id"]),
)
conn.commit()
finally:
if conn:
conn.close()
return token
def validate_magic_login_token(token):
"""Validate a magic login token and consume it. Returns user_id or None.
Token expires after 15 minutes. Consumed tokens are immediately invalidated.
"""
conn = None
try:
conn = get_db()
row = conn.execute(
"""SELECT id, magic_login_sent_at FROM users
WHERE magic_login_token = ?""",
(token,),
).fetchone()
if not row:
conn.close()
return None
# Token expires after 15 minutes
sent_at = row["magic_login_sent_at"]
if sent_at:
from datetime import timedelta
sent_dt = datetime.fromisoformat(sent_at).replace(tzinfo=UTC)
if datetime.now(UTC) - sent_dt > timedelta(minutes=15):
conn.execute(
"UPDATE users SET magic_login_token = NULL WHERE id = ?",
(row["id"],),
)
conn.commit()
conn.close()
return None
# Consume the token
conn.execute(
"UPDATE users SET magic_login_token = NULL, magic_login_sent_at = NULL WHERE id = ?",
(row["id"],),
)
conn.commit()
finally:
if conn:
conn.close()
return row["id"]
# ─── Admin functions ──────────────────────────────────────────────────────────
def generate_api_key(name, user_id, permissions=None, expires_in_days=None):
"""Generate a new API key for a user.
Args:
name: Human-readable name for the key
user_id: Owner user ID
permissions: Optional permissions dict
expires_in_days: Optional expiration in days (None = never expires)
Returns dict with: id, key_prefix, full_key (shown once), expires_at.
The full_key is NOT stored — only the hash.
"""
if permissions is None:
permissions = {
"read_forms": True,
"write_forms": True,
"read_submissions": True,
"delete_forms": False,
}
# Generate raw key
raw_key = f"{API_KEY_PREFIX}{secrets.token_hex(24)}"
key_hash = bcrypt.hashpw(raw_key.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
key_prefix = raw_key[:12] # afk_live_xx
# Phase 11: calculate expiration
from datetime import timedelta
expires_at = None
if expires_in_days is not None and expires_in_days > 0:
expires_at = (datetime.now(UTC) + timedelta(days=expires_in_days)).isoformat()
conn = None
try:
conn = get_db()
# Check for duplicate name
existing = conn.execute(
"SELECT id FROM api_keys WHERE user_id = ? AND name = ?",
(user_id, name),
).fetchone()
if existing:
raise ValueError(f"Key with name '{name}' already exists for this user")
cursor = conn.execute(
"INSERT INTO api_keys (user_id, key_hash, key_prefix, name, permissions, expires_at) VALUES (?, ?, ?, ?, ?, ?)",
(user_id, key_hash, key_prefix, name, json.dumps(permissions), expires_at),
)
conn.commit()
key_id = cursor.lastrowid
return {
"id": key_id,
"key_prefix": key_prefix,
"full_key": raw_key,
"name": name,
"permissions": permissions,
"expires_at": expires_at,
}
finally:
if conn:
conn.close()
def validate_api_key(raw_key):
"""Validate an API key. Returns (user_id, permissions) or (None, None)."""
from datetime import timedelta as _td
conn = None
try:
conn = get_db()
# Prefix lookup first, then verify hash
prefix = raw_key[:12] if len(raw_key) >= 12 else ""
if not prefix:
return None, None
rows = conn.execute(
"SELECT ak.*, u.id as user_id FROM api_keys ak JOIN users u ON ak.user_id = u.id WHERE ak.key_prefix = ?",
(prefix,),
).fetchall()
for row in rows:
row = dict(row)
if bcrypt.checkpw(raw_key.encode("utf-8"), row["key_hash"].encode("utf-8")):
# Phase 11: check expiration
if row.get("expires_at"):
expires_at = datetime.fromisoformat(row["expires_at"])
if datetime.now(UTC) > expires_at:
# Key expired — log but don't update last_used_at
_admin_logger.warning("Expired API key used: id=%s user=%s", row.get("id"), row.get("user_id"))
return None, None
# Update last_used_at
conn.execute(
"UPDATE api_keys SET last_used_at = CURRENT_TIMESTAMP WHERE id = ?",
(row["id"],),
)
conn.commit()
permissions = json.loads(row["permissions"]) if row["permissions"] else {}
return row["user_id"], permissions
return None, None
finally:
if conn:
conn.close()
def list_api_keys(user_id):
"""List all API keys for a user (without hashes)."""
conn = None
try:
conn = get_db()
rows = conn.execute(
"SELECT id, key_prefix, name, permissions, expires_at, last_used_at, created_at FROM api_keys WHERE user_id = ? ORDER BY created_at DESC",
(user_id,),
).fetchall()
finally:
if conn:
conn.close()
result = []
for row in rows:
d = dict(row)
try:
d["permissions"] = json.loads(d["permissions"]) if d["permissions"] else {}
except (json.JSONDecodeError, TypeError):
d["permissions"] = {}
result.append(d)
return result
def revoke_api_key(key_id, user_id):
"""Revoke an API key (must belong to user)."""
conn = None
try:
conn = get_db()
result = conn.execute(
"DELETE FROM api_keys WHERE id = ? AND user_id = ?",
(key_id, user_id),
)
conn.commit()
finally:
if conn:
conn.close()
return result.rowcount > 0
def renew_api_key(key_id, user_id, expires_in_days):
"""Extend the expiration of an API key. Returns True if updated."""
from datetime import timedelta as _td
conn = None
try:
conn = get_db()
new_expiry = (datetime.now(UTC) + timedelta(days=expires_in_days)).isoformat()
result = conn.execute(
"UPDATE api_keys SET expires_at = ? WHERE id = ? AND user_id = ?",
(new_expiry, key_id, user_id),
)
conn.commit()
finally:
if conn:
conn.close()
return result.rowcount > 0
def check_api_permission(permissions, action):
"""Check if an API key has the required permission.
Actions: 'read_forms', 'write_forms', 'read_submissions', 'delete_forms'
"""
if not permissions:
return False
return bool(permissions.get(action, False))
# ─── Template model functions ────────────────────────────────────────────────
def export_user_data(user_id):
"""Export all user data as dict (GDPR right to data portability)."""
conn = None
try:
conn = get_db()
# Get user record
user_row = conn.execute("SELECT id, tier FROM users WHERE id = ?", (user_id,)).fetchone()
if not user_row:
return None
# Get user's sites
sites = [dict(r) for r in conn.execute("SELECT * FROM sites WHERE user_id = ?", (user_id,)).fetchall()]
# Get monthly usage
usage = [dict(r) for r in conn.execute("SELECT * FROM monthly_usage WHERE user_id = ?", (user_id,)).fetchall()]
return {
"user": {"id": user_row["id"], "tier": user_row["tier"]},
"sites": sites,
"monthly_usage": usage,
}
finally:
if conn:
conn.close()
def delete_user(user_id):
"""Delete a user and all associated data. Returns dict with success/error."""
conn = None
try:
conn = get_db()
conn.execute("BEGIN IMMEDIATE")
# Delete cascading: sites, submissions, form_sessions, etc.
conn.execute(
"DELETE FROM form_submissions WHERE site_id IN (SELECT id FROM sites WHERE user_id = ?)", (user_id,)
)
conn.execute("DELETE FROM form_sessions WHERE site_id IN (SELECT id FROM sites WHERE user_id = ?)", (user_id,))
conn.execute(
"DELETE FROM form_impressions WHERE site_id IN (SELECT id FROM sites WHERE user_id = ?)", (user_id,)
)
conn.execute("DELETE FROM webhook_logs WHERE site_id IN (SELECT id FROM sites WHERE user_id = ?)", (user_id,))
conn.execute("DELETE FROM sites WHERE user_id = ?", (user_id,))
conn.execute("DELETE FROM api_keys WHERE user_id = ?", (user_id,))
conn.execute("DELETE FROM monthly_usage WHERE user_id = ?", (user_id,))
conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
conn.commit()
return {"success": True}
except Exception as e:
conn.execute("ROLLBACK")
return {"success": False, "error": str(e)}
finally:
if conn:
conn.close()
def generate_referral_code(user_id):
"""Generate or return existing referral code for user. Returns string."""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT referral_code FROM users WHERE id = ?", (user_id,)).fetchone()
if row and row["referral_code"]:
return row["referral_code"]
code = secrets.token_urlsafe(8)[:12].upper()
conn.execute("UPDATE users SET referral_code = ? WHERE id = ?", (code, user_id))
conn.commit()
return code
finally:
if conn:
conn.close()
def get_referral_stats(user_id):
"""Get referral stats for user. Returns dict."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT COUNT(*) as cnt FROM users WHERE referred_by = ?",
(user_id,),
).fetchone()
return {"total_referrals": row["cnt"] if row else 0}
finally:
if conn:
conn.close()
def try_decrypt_user_value(user_id, encrypted_value):
"""Try to decrypt a user value. Returns decrypted string or original value on failure."""
if not encrypted_value:
return None
if not is_encrypted(encrypted_value):
return encrypted_value
try:
result = decrypt_user_value(user_id, encrypted_value)
return result if result is not None else encrypted_value
except Exception:
return encrypted_value
# ─── Email Campaigns ────────────────────────────────────────────────────────────
# ─── Agent I/O Helpers ─────────────────────────────────────────────────────────
def check_rate_limit(user_id, limit_type="api"):
"""Check if user has exceeded rate limits. Returns True if allowed."""
# For now, just return True — rate limiting is handled by Flask middleware
# This is a placeholder for per-user rate limiting
return True
def record_api_usage(user_id, api_key_id, action_type):
"""Record an API usage event for tracking/analytics."""
conn = None
try:
conn = get_db()
conn.execute(
"""INSERT INTO api_usage (user_id, api_key_id, action_type, created_at)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)""",
(user_id, api_key_id, action_type),
)
conn.commit()
except Exception as e:
print(f"[model] record_api_usage error: {e}")
finally:
if conn:
conn.close()