"""Auto-extracted from models.py — do not edit manually."""
import json
import os
import secrets
import sqlite3
import uuid
from datetime import UTC, datetime, timezone
import bcrypt
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 DB_PATH, _table_exists, get_db
from app.helpers import _get_site_owner_password_hash
def migrate_add_field_config():
"""Migration: add field_config, webhook_url, webhook_enabled, webhook_events to sites."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "sites"):
return
info = conn.execute("PRAGMA table_info(sites)").fetchall()
columns = [col["name"] for col in info]
if "field_config" not in columns:
conn.execute("ALTER TABLE sites ADD COLUMN field_config TEXT")
if "webhook_url" not in columns:
conn.execute("ALTER TABLE sites ADD COLUMN webhook_url TEXT")
if "webhook_enabled" not in columns:
conn.execute("ALTER TABLE sites ADD COLUMN webhook_enabled INTEGER DEFAULT 0")
if "webhook_events" not in columns:
conn.execute("ALTER TABLE sites ADD COLUMN webhook_events TEXT DEFAULT 'submission'")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_submission_data():
"""Migration: add data JSON column to submissions for dynamic fields."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "submissions"):
return
info = conn.execute("PRAGMA table_info(submissions)").fetchall()
columns = [col["name"] for col in info]
if "data" not in columns:
conn.execute("ALTER TABLE submissions ADD COLUMN data TEXT")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_user_id_to_sites():
"""Migration: add user_id column to sites if it doesn't exist."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "sites"):
return
info = conn.execute("PRAGMA table_info(sites)").fetchall()
columns = [col["name"] for col in info]
if "user_id" not in columns:
conn.execute("ALTER TABLE sites ADD COLUMN user_id INTEGER REFERENCES users(id)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_usage():
"""Migration: add monthly_usage table for per-user submission tracking."""
conn = None
try:
conn = get_db()
info = conn.execute("PRAGMA table_info(monthly_usage)").fetchall()
if not info:
conn.execute("""
CREATE TABLE monthly_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
month TEXT NOT NULL,
submission_count INTEGER DEFAULT 0,
UNIQUE(user_id, month)
)
""")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_abuse_protection():
"""Migration: add abuse protection columns.
- submissions: client_ip, user_agent, spam_flag
- sites: honeypot_enabled, spam_filter_enabled, recaptcha_enabled, recaptcha_threshold
"""
conn = None
try:
conn = get_db()
# Submissions columns
if not _table_exists(conn, "submissions"):
conn.commit()
return
sub_info = conn.execute("PRAGMA table_info(submissions)").fetchall()
sub_cols = [col["name"] for col in sub_info]
if "client_ip" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN client_ip TEXT")
if "user_agent" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN user_agent TEXT")
if "spam_flag" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN spam_flag INTEGER DEFAULT 0")
# Sites columns
if not _table_exists(conn, "sites"):
conn.commit()
return
site_info = conn.execute("PRAGMA table_info(sites)").fetchall()
site_cols = [col["name"] for col in site_info]
if "honeypot_enabled" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN honeypot_enabled INTEGER DEFAULT 1")
if "spam_filter_enabled" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN spam_filter_enabled INTEGER DEFAULT 1")
if "recaptcha_enabled" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN recaptcha_enabled INTEGER DEFAULT 0")
if "recaptcha_threshold" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN recaptcha_threshold REAL DEFAULT 0.5")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_rate_limit():
"""Migration: add per-site configurable rate limit columns."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "sites"):
return
site_info = conn.execute("PRAGMA table_info(sites)").fetchall()
site_cols = [col["name"] for col in site_info]
if "rate_limit_enabled" not in site_cols:
try:
conn.execute("ALTER TABLE sites ADD COLUMN rate_limit_enabled INTEGER DEFAULT 1")
except sqlite3.OperationalError:
pass
if "rate_limit_burst" not in site_cols:
try:
conn.execute("ALTER TABLE sites ADD COLUMN rate_limit_burst INTEGER DEFAULT 20")
except sqlite3.OperationalError:
pass
if "rate_limit_refill" not in site_cols:
try:
conn.execute("ALTER TABLE sites ADD COLUMN rate_limit_refill REAL DEFAULT 0.5")
except sqlite3.OperationalError:
pass
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_analytics():
"""Migration: add site_analytics table for per-site, per-day submission counts."""
conn = None
try:
conn = get_db()
info = conn.execute("PRAGMA table_info(site_analytics)").fetchall()
if not info:
conn.execute("""
CREATE TABLE site_analytics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id),
day TEXT NOT NULL,
submission_count INTEGER DEFAULT 0,
spam_count INTEGER DEFAULT 0,
UNIQUE(site_id, day)
)
""")
conn.execute("CREATE INDEX idx_site_analytics_site_day ON site_analytics(site_id, day)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_email_verification():
"""Migration: add email verification columns to users table."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "users"):
return
info = conn.execute("PRAGMA table_info(users)").fetchall()
cols = [col["name"] for col in info]
if "email_verified" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN email_verified INTEGER DEFAULT 0")
if "email_verification_token" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN email_verification_token TEXT")
if "email_verification_sent_at" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN email_verification_sent_at TIMESTAMP")
if "pending_email" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN pending_email TEXT")
if "password_reset_token" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN password_reset_token TEXT")
if "password_reset_sent_at" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN password_reset_sent_at TIMESTAMP")
if "magic_login_token" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN magic_login_token TEXT")
if "magic_login_sent_at" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN magic_login_sent_at TIMESTAMP")
if "magic_login_enabled" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN magic_login_enabled INTEGER DEFAULT 1")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_indexes():
"""Migration: add composite indexes for common query patterns.
- submissions(site_id, submitted_at DESC) — filtered+sorted listing
- monthly_usage(user_id, month) — explicit index for usage lookups
- users(stripe_customer_id) — Stripe webhook lookups (billing.py)
- users(email_hash) — email-based user lookups (auth, billing)
(UNIQUE constraint already creates an implicit index, but this makes
the intent explicit and ensures it survives schema changes.)
"""
conn = None
try:
conn = get_db()
# Submissions indexes
if _table_exists(conn, "submissions"):
indexes = conn.execute("PRAGMA index_list(submissions)").fetchall()
index_names = [idx["name"] for idx in indexes]
if "idx_submissions_site_created" not in index_names:
conn.execute("CREATE INDEX idx_submissions_site_created ON submissions(site_id, submitted_at DESC)")
# Monthly usage indexes
if _table_exists(conn, "monthly_usage"):
mu_indexes = conn.execute("PRAGMA index_list(monthly_usage)").fetchall()
mu_index_names = [idx["name"] for idx in mu_indexes]
if "idx_monthly_usage_user_month" not in mu_index_names:
conn.execute("CREATE INDEX idx_monthly_usage_user_month ON monthly_usage(user_id, month)")
# Cleanup old monthly_usage beyond retention (6 months)
from datetime import timedelta
cutoff = (datetime.now(UTC).replace(day=1) - timedelta(days=180)).strftime("%Y-%m")
conn.execute("DELETE FROM monthly_usage WHERE month < ?", (cutoff,))
# Users indexes — stripe_customer_id looked up on every Stripe webhook
if _table_exists(conn, "users"):
u_indexes = conn.execute("PRAGMA index_list(users)").fetchall()
u_index_names = [idx["name"] for idx in u_indexes]
if "idx_users_stripe_customer" not in u_index_names:
conn.execute("CREATE INDEX idx_users_stripe_customer ON users(stripe_customer_id)")
# Webhook retry queue table
wl_info = conn.execute("PRAGMA table_info(webhook_logs)").fetchall()
if not wl_info:
conn.execute("""
CREATE TABLE webhook_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id),
submission_id INTEGER NOT NULL REFERENCES submissions(id),
webhook_url TEXT NOT NULL,
attempt_count INTEGER DEFAULT 0,
status TEXT DEFAULT 'pending',
last_error TEXT,
next_retry_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_webhook_logs_retry ON webhook_logs(next_retry_at, status)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_api_keys():
"""Migration: add api_keys table + created_by/metadata columns on sites."""
conn = None
try:
conn = get_db()
# api_keys table
info = conn.execute("PRAGMA table_info(api_keys)").fetchall()
if not info:
conn.execute("""
CREATE TABLE api_keys (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
key_hash TEXT NOT NULL,
key_prefix TEXT NOT NULL,
name TEXT NOT NULL,
permissions TEXT NOT NULL DEFAULT '{"read_forms": true, "write_forms": true, "read_submissions": true, "delete_forms": false}',
expires_at TIMESTAMP,
last_used_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_api_keys_hash ON api_keys(key_hash)")
conn.execute("CREATE INDEX idx_api_keys_user ON api_keys(user_id)")
else:
# Phase 11: add expires_at column if missing
site_cols = [col["name"] for col in info]
if "expires_at" not in site_cols:
conn.execute("ALTER TABLE api_keys ADD COLUMN expires_at TIMESTAMP")
# Sites columns
if not _table_exists(conn, "sites"):
conn.commit()
return
site_info = conn.execute("PRAGMA table_info(sites)").fetchall()
site_cols = [col["name"] for col in site_info]
if "created_by" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN created_by TEXT DEFAULT 'manual'")
if "metadata" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN metadata TEXT")
conn.commit()
finally:
if conn:
conn.close()
def migrate_encrypt_existing_data():
"""Migration: encrypt existing plaintext PII data using the current ENCRYPTION_KEY.
Runs idempotently — skips rows already encrypted (detected by is_encrypted).
Called on first boot after ENCRYPTION_KEY is set.
"""
from app.crypto import is_encrypted
if not key_is_configured():
return 0
conn = None
try:
conn = get_db()
encrypted_count = 0
# Encrypt submissions (PII fields)
subs = conn.execute(
"SELECT id, site_id, customer_name, customer_phone, customer_email, customer_equipment, customer_message, data FROM submissions"
).fetchall()
for sub in subs:
sid = sub["site_id"]
has_plaintext = any(
sub[col] and not is_encrypted(str(sub[col]))
for col in (
"customer_name",
"customer_phone",
"customer_email",
"customer_equipment",
"customer_message",
)
)
if sub["data"] and not is_encrypted(str(sub["data"])):
has_plaintext = True
if not has_plaintext:
continue
conn.execute(
"UPDATE submissions SET customer_name=?, customer_phone=?, customer_email=?, customer_equipment=?, customer_message=?, data=? WHERE id=?",
(
encrypt_value(sid, sub["customer_name"]) if sub["customer_name"] else None,
encrypt_value(sid, sub["customer_phone"]) if sub["customer_phone"] else None,
encrypt_value(sid, sub["customer_email"]) if sub["customer_email"] else None,
encrypt_value(sid, sub["customer_equipment"]) if sub["customer_equipment"] else None,
encrypt_value(sid, sub["customer_message"]) if sub["customer_message"] else None,
encrypt_value(sid, sub["data"]) if sub["data"] else None,
sub["id"],
),
)
encrypted_count += 1
# Encrypt webhook_url in sites
sites = conn.execute("SELECT id, webhook_url FROM sites WHERE webhook_url IS NOT NULL").fetchall()
for site in sites:
if site["webhook_url"] and not is_encrypted(str(site["webhook_url"])):
conn.execute(
"UPDATE sites SET webhook_url=? WHERE id=?",
(encrypt_value(site["id"], site["webhook_url"]), site["id"]),
)
encrypted_count += 1
conn.commit()
if encrypted_count:
print(f"[migration] Encrypted {encrypted_count} row(s)")
return encrypted_count
finally:
if conn:
conn.close()
def migrate_rescope_to_user_keys():
"""Migration: re-encrypt submissions from site-scoped keys to user-scoped keys.
Decrypts each submission using the site-scoped key, then re-encrypts
using the site owner's password-derived key. This ensures only the site
owner can decrypt their submissions, while admin sees raw encrypted data.
Idempotent — if user-scoped decryption succeeds, the row is skipped.
"""
if not key_is_configured():
return 0
conn = None
try:
conn = get_db()
rescope_count = 0
subs = conn.execute(
"SELECT id, site_id, customer_name, customer_phone, customer_email, customer_equipment, customer_message, data FROM submissions"
).fetchall()
for sub in subs:
sid = sub["site_id"]
pw_hash = _get_site_owner_password_hash(sid)
if not pw_hash:
# No owner — keep site-scoped encryption
continue
# Check if already user-scoped encrypted (try user decrypt first)
already_user_scoped = False
if sub.get("customer_name"):
test = try_decrypt_user_submission(pw_hash, sub["customer_name"])
if test and not is_encrypted(test):
already_user_scoped = True
if already_user_scoped:
continue
# Decrypt with site-scoped key, re-encrypt with user-scoped key
new_values = {}
needs_update = False
for col in ("customer_name", "customer_phone", "customer_email", "customer_equipment", "customer_message"):
val = sub.get(col)
if val:
plaintext = try_decrypt(sid, val)
if plaintext:
enc = encrypt_user_submission(pw_hash, plaintext)
if enc:
new_values[col] = enc
needs_update = True
# Also re-encrypt the data JSON
raw_data = sub.get("data")
if raw_data:
plaintext_data = try_decrypt(sid, raw_data)
if plaintext_data:
enc_data = encrypt_user_submission(pw_hash, plaintext_data)
if enc_data:
new_values["data"] = enc_data
needs_update = True
if needs_update:
set_clause = ", ".join(f"{k}=? " for k in new_values)
values = list(new_values.values()) + [sub["id"]]
conn.execute(f"UPDATE submissions SET {set_clause}WHERE id=?", values)
rescope_count += 1
# Also re-encrypt webhook_url in sites using entity-scoped encryption
sites = conn.execute("SELECT id, user_id, webhook_url FROM sites WHERE webhook_url IS NOT NULL").fetchall()
for site in sites:
if site["webhook_url"] and is_encrypted(str(site["webhook_url"])):
# Decrypt with site-scoped, re-encrypt with entity-scoped
pw_hash = _get_site_owner_password_hash(site["id"])
if pw_hash:
plaintext_url = try_decrypt(site["id"], site["webhook_url"])
if plaintext_url:
enc_url = encrypt_entity(site["id"], plaintext_url)
if enc_url:
conn.execute("UPDATE sites SET webhook_url=? WHERE id=?", (enc_url, site["id"]))
rescope_count += 1
conn.commit()
if rescope_count:
print(f"[migration] Re-scoped {rescope_count} row(s) to user-scoped encryption")
return rescope_count
finally:
if conn:
conn.close()
def migrate_rotate_encryption_keys():
"""Migration: re-encrypt all PII under the current (V2) master key.
When ENCRYPTION_KEY_V2 is configured, this iterates all encrypted
columns, decrypts with any available key, and re-encrypts with V2.
After this runs, all data uses the V2 key and V1 can be retired.
Idempotent — safe to call even if no rotation is needed.
"""
from app.crypto import (
decrypt_value,
encrypt_entity,
encrypt_user_value,
encrypt_value,
is_encrypted,
master_key_is_rotation_active,
reset_master_key_cache,
)
if not master_key_is_rotation_active():
return 0
reset_master_key_cache()
conn = None
try:
conn = get_db()
rotated = 0
# ── sites.webhook_url (entity-scoped) ──
sites = conn.execute("SELECT id FROM sites").fetchall()
for site in sites:
sid = site["id"]
# Decrypt with current (which tries V2 first, V1 fallback)
# If it decrypts, re-encrypt (which uses V2)
row = conn.execute("SELECT webhook_url FROM sites WHERE id=?", (sid,)).fetchone()
if row and row["webhook_url"] and is_encrypted(str(row["webhook_url"])):
plaintext = decrypt_entity("site", sid, row["webhook_url"])
if plaintext:
enc = encrypt_entity("site", sid, plaintext)
if enc:
conn.execute("UPDATE sites SET webhook_url=? WHERE id=?", (enc, sid))
rotated += 1
# ── form_submissions.customer_* and data (user-scoped or site-scoped) ──
subs = conn.execute("""
SELECT s.id, s.site_id
FROM submissions s
JOIN sites sit ON s.site_id = sit.id
WHERE s.customer_name IS NOT NULL
OR s.customer_phone IS NOT NULL
OR s.customer_email IS NOT NULL
OR s.customer_equipment IS NOT NULL
OR s.customer_message IS NOT NULL
OR s.data IS NOT NULL
""").fetchall()
for sub in subs:
sid = sub["site_id"]
# Get site owner's password hash for user-scoped decryption
pw_hash = _get_site_owner_password_hash(sid)
fields = [
"customer_name",
"customer_phone",
"customer_email",
"customer_equipment",
"customer_equipment",
"data",
]
for field in fields:
row = conn.execute(f"SELECT {field} FROM submissions WHERE id=?", (sub["id"],)).fetchone()
val = row[field] if row else None
if val and is_encrypted(str(val)):
# Try user-scoped first (new data)
plaintext = None
if pw_hash:
from app.crypto import decrypt_user_submission
plaintext = decrypt_user_submission(pw_hash, val)
if plaintext == val:
plaintext = None
# Fallback: site-scoped (old data)
if plaintext is None:
plaintext = decrypt_value(sid, val)
if plaintext:
# Re-encrypt with user-scoped key (preferred)
if pw_hash:
from app.crypto import encrypt_user_submission
enc = encrypt_user_submission(pw_hash, plaintext)
else:
enc = encrypt_value(sid, plaintext)
if enc:
conn.execute(f"UPDATE submissions SET {field}=? WHERE id=?", (enc, sub["id"]))
rotated += 1
# ── users.email_encrypted, name_encrypted, display_name_encrypted (user-scoped) ──
users = conn.execute("SELECT id FROM users").fetchall()
for user in users:
uid = user["id"]
for field in ["email_encrypted", "name_encrypted", "display_name_encrypted"]:
row = conn.execute(f"SELECT {field} FROM users WHERE id=?", (uid,)).fetchone()
val = row[field] if row else None
if val and is_encrypted(str(val)):
plaintext = decrypt_user_value(uid, val)
if plaintext:
enc = encrypt_user_value(uid, plaintext)
if enc:
conn.execute(f"UPDATE users SET {field}=? WHERE id=?", (enc, uid))
rotated += 1
# ── webhook_destinations.url_encrypted (entity-scoped) ──
wd = conn.execute("SELECT id FROM webhook_destinations").fetchall()
for row in wd:
wid = row["id"]
r = conn.execute("SELECT url_encrypted FROM webhook_destinations WHERE id=?", (wid,)).fetchone()
val = r["url_encrypted"] if r else None
if val and is_encrypted(str(val)):
plaintext = decrypt_entity("webhook_dest", wid, val)
if plaintext:
enc = encrypt_entity("webhook_dest", wid, plaintext)
if enc:
conn.execute("UPDATE webhook_destinations SET url_encrypted=? WHERE id=?", (enc, wid))
rotated += 1
# ── webhook_logs.webhook_url_encrypted (entity-scoped) ──
wl = conn.execute("SELECT id FROM webhook_logs").fetchall()
for row in wl:
wid = row["id"]
r = conn.execute("SELECT webhook_url_encrypted FROM webhook_logs WHERE id=?", (wid,)).fetchone()
val = r["webhook_url_encrypted"] if r else None
if val and is_encrypted(str(val)):
plaintext = decrypt_entity("webhook_log", wid, val)
if plaintext:
enc = encrypt_entity("webhook_log", wid, plaintext)
if enc:
conn.execute("UPDATE webhook_logs SET webhook_url_encrypted=? WHERE id=?", (enc, wid))
rotated += 1
# ── invoice_schedules.customer_name, customer_email (entity-scoped) ──
inv = conn.execute("SELECT id FROM invoice_schedules").fetchall()
for row in inv:
iid = row["id"]
for field in ["customer_name", "customer_email"]:
r = conn.execute(f"SELECT {field} FROM invoice_schedules WHERE id=?", (iid,)).fetchone()
val = r[field] if r else None
if val and is_encrypted(str(val)):
plaintext = decrypt_entity("invoice_schedule", iid, val)
if plaintext:
enc = encrypt_entity("invoice_schedule", iid, plaintext)
if enc:
conn.execute(f"UPDATE invoice_schedules SET {field}=? WHERE id=?", (enc, iid))
rotated += 1
# ── team_members.invited_email_encrypted (entity-scoped) ──
tm = conn.execute("SELECT id FROM team_members").fetchall()
for row in tm:
tid = row["id"]
r = conn.execute("SELECT invited_email_encrypted FROM team_members WHERE id=?", (tid,)).fetchone()
val = r["invited_email_encrypted"] if r else None
if val and is_encrypted(str(val)):
plaintext = decrypt_entity("team_member", tid, val)
if plaintext:
enc = encrypt_entity("team_member", tid, plaintext)
if enc:
conn.execute("UPDATE team_members SET invited_email_encrypted=? WHERE id=?", (enc, tid))
rotated += 1
conn.commit()
if rotated:
print(f"[migration] Key rotation complete: {rotated} value(s) re-encrypted under V2")
return rotated
finally:
if conn:
conn.close()
def migrate_normalize_field_configs():
"""Migration: normalize existing field_configs with Phase 3 defaults.
Ensures all field definitions have default values for:
- validation: None
- condition: None
- calculation: None
- step: 1
- required: false
Idempotent — only updates fields missing these keys.
"""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "sites"):
conn.commit()
return
sites = conn.execute(
"SELECT id, field_config FROM sites WHERE field_config IS NOT NULL AND field_config != ''"
).fetchall()
updated = 0
for site in sites:
try:
fields = json.loads(site["field_config"])
if not isinstance(fields, list):
continue
changed = False
for field in fields:
if not isinstance(field, dict):
continue
if "validation" not in field:
field["validation"] = None
changed = True
if "condition" not in field:
field["condition"] = None
changed = True
if "calculation" not in field:
field["calculation"] = None
changed = True
if "step" not in field:
field["step"] = 1
changed = True
if "required" not in field:
field["required"] = False
changed = True
if changed:
conn.execute(
"UPDATE sites SET field_config = ? WHERE id = ?",
(json.dumps(fields), site["id"]),
)
updated += 1
except (json.JSONDecodeError, TypeError):
continue
conn.commit()
if updated:
print(f"[migration] Normalized field_configs for {updated} site(s)")
finally:
if conn:
conn.close()
def migrate_add_form_templates():
"""Migration: add form_templates table for Phase 4 Template Marketplace.
Schema:
- slug: unique identifier (e.g. 'contact-us')
- name: display name
- description: one-line description
- category: Business | Events | Hiring | Support | E-commerce | Onboarding
- field_config: JSON array of field definitions (Phase 3 schema)
- success_message: default success message
- created_by: 'system' | user_id
- is_featured: 0/1
- created_at: timestamp
"""
conn = None
try:
conn = get_db()
info = conn.execute("PRAGMA table_info(form_templates)").fetchall()
if not info:
conn.execute("""
CREATE TABLE form_templates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
slug TEXT UNIQUE NOT NULL,
name TEXT NOT NULL,
description TEXT NOT NULL,
category TEXT NOT NULL,
field_config TEXT NOT NULL,
success_message TEXT,
created_by TEXT DEFAULT 'system',
is_featured INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_form_templates_category ON form_templates(category)")
conn.execute("CREATE INDEX idx_form_templates_featured ON form_templates(is_featured DESC)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_teams():
"""Migration: add teams, team_members tables + sites.team_id for Phase 5 Team Collaboration.
New tables:
- teams: id, name, owner_id, tier, stripe_customer_id, stripe_subscription_id, created_at
- team_members: id, team_id, user_id, role, invited_email, status, invite_token, joined_at, created_at
New column:
- sites.team_id: NULL = personal site, team_id = team site
Team tiers:
- teams: $49/mo, unlimited sites, 25K submissions, up to 5 members
- teams_plus: $99/mo, unlimited members, custom branding
"""
conn = None
try:
conn = get_db()
# teams table
info = conn.execute("PRAGMA table_info(teams)").fetchall()
if not info:
conn.execute("""
CREATE TABLE teams (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
owner_id INTEGER NOT NULL REFERENCES users(id),
tier TEXT DEFAULT 'teams',
stripe_customer_id TEXT,
stripe_subscription_id TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_teams_owner ON teams(owner_id)")
# team_members table
info = conn.execute("PRAGMA table_info(team_members)").fetchall()
if not info:
conn.execute("""
CREATE TABLE team_members (
id INTEGER PRIMARY KEY AUTOINCREMENT,
team_id INTEGER NOT NULL REFERENCES teams(id),
user_id INTEGER REFERENCES users(id),
role TEXT NOT NULL DEFAULT 'member',
invited_email TEXT,
status TEXT NOT NULL DEFAULT 'pending',
invite_token TEXT UNIQUE,
joined_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_team_members_team ON team_members(team_id)")
conn.execute("CREATE INDEX idx_team_members_user ON team_members(user_id)")
conn.execute("CREATE INDEX idx_team_members_token ON team_members(invite_token)")
# sites.team_id column
if not _table_exists(conn, "sites"):
conn.commit()
return
site_info = conn.execute("PRAGMA table_info(sites)").fetchall()
site_cols = [col["name"] for col in site_info]
if "team_id" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN team_id INTEGER REFERENCES teams(id)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_form_analytics():
"""Migration: add analytics columns to submissions + form_impressions table for Phase 6.
New columns on submissions:
- geo_city: City from GeoIP2 Lite lookup
- geo_country: 2-letter country code from GeoIP2 Lite
- device_type: mobile | desktop | tablet | bot
- browser: Chrome, Safari, Firefox, Edge, etc.
- os: Windows, macOS, Linux, Android, iOS, etc.
New table:
- form_impressions: Track form page loads for conversion rate analysis
Tier gating:
- Free: basic submission counts only
- Starter+: geo + device + impressions
- Pro: full analytics including conversion rates
"""
conn = None
try:
conn = get_db()
# Submissions analytics columns
if not _table_exists(conn, "submissions"):
conn.commit()
return
sub_info = conn.execute("PRAGMA table_info(submissions)").fetchall()
sub_cols = [col["name"] for col in sub_info]
if "geo_city" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN geo_city TEXT")
if "geo_country" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN geo_country TEXT")
if "device_type" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN device_type TEXT")
if "browser" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN browser TEXT")
if "os" not in sub_cols:
conn.execute("ALTER TABLE submissions ADD COLUMN os TEXT")
# form_impressions table
info = conn.execute("PRAGMA table_info(form_impressions)").fetchall()
if not info:
conn.execute("""
CREATE TABLE form_impressions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id),
ip_hash TEXT NOT NULL,
user_agent TEXT,
geo_city TEXT,
geo_country TEXT,
device_type TEXT,
browser TEXT,
os TEXT,
viewed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_form_impressions_site ON form_impressions(site_id)")
conn.execute("CREATE INDEX idx_form_impressions_date ON form_impressions(site_id, viewed_at)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_ai_builder():
"""Migration: add ai_builder_prompt column to sites and ai_prompt_history table (Phase 8).
New column:
- sites.ai_builder_prompt: The natural language prompt used to AI-generate this form
New table:
- ai_prompt_history: Per-site history of AI form generation prompts and results
"""
conn = None
try:
conn = get_db()
# sites.ai_builder_prompt column
info = conn.execute("PRAGMA table_info(sites)").fetchall()
columns = [col["name"] for col in info]
if "ai_builder_prompt" not in columns:
conn.execute("ALTER TABLE sites ADD COLUMN ai_builder_prompt TEXT")
# ai_prompt_history table
if not _table_exists(conn, "ai_prompt_history"):
conn.execute("""
CREATE TABLE ai_prompt_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
prompt TEXT NOT NULL,
fields_generated INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_prompt_history_site ON ai_prompt_history(site_id)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_webhook_destinations():
"""Migration: add webhook_destinations table and destination_id to webhook_logs (Phase 7).
New table:
- webhook_destinations: Per-site webhook destination configurations
New column:
- webhook_logs.destination_id: FK to webhook_destinations (nullable, backward compat)
"""
conn = None
try:
conn = get_db()
# webhook_destinations table
if not _table_exists(conn, "webhook_destinations"):
conn.execute("""
CREATE TABLE webhook_destinations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
name TEXT NOT NULL DEFAULT 'Integration',
type TEXT NOT NULL,
url TEXT,
config TEXT,
enabled INTEGER DEFAULT 1,
last_status TEXT,
last_error TEXT,
last_delivered_at TIMESTAMP,
success_count INTEGER DEFAULT 0,
failure_count INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_destinations_site ON webhook_destinations(site_id)")
# webhook_logs.destination_id column
if _table_exists(conn, "webhook_logs"):
info = conn.execute("PRAGMA table_info(webhook_logs)").fetchall()
columns = [col["name"] for col in info]
if "destination_id" not in columns:
conn.execute(
"ALTER TABLE webhook_logs ADD COLUMN destination_id INTEGER REFERENCES webhook_destinations(id)"
)
conn.execute("CREATE INDEX idx_webhook_logs_destination ON webhook_logs(destination_id)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_form_versions():
"""Migration: add form_versions table for Phase 9 Form Versioning & Rollback.
Schema:
- site_id: FK to sites
- version: auto-incrementing version number per site
- field_config: JSON snapshot of field config at this version
- metadata: JSON snapshot of site metadata at this version
- changed_by: user email or 'system' who made the change
- changed_at: timestamp
- change_reason: free-text reason (optional)
"""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "form_versions"):
conn.execute("""
CREATE TABLE form_versions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
version INTEGER NOT NULL,
field_config TEXT NOT NULL,
metadata TEXT,
changed_by TEXT NOT NULL DEFAULT 'system',
changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
change_reason TEXT
)
""")
conn.execute("CREATE INDEX idx_form_versions_site_version ON form_versions(site_id, version DESC)")
# Seed version 1 for existing sites (snapshot their current config)
if _table_exists(conn, "sites"):
sites = conn.execute(
"SELECT id, field_config, metadata, name FROM sites WHERE field_config IS NOT NULL AND field_config != ''"
).fetchall()
for site in sites:
existing = conn.execute(
"SELECT COUNT(*) as cnt FROM form_versions WHERE site_id = ?", (site["id"],)
).fetchone()
if existing["cnt"] == 0:
conn.execute(
"INSERT INTO form_versions (site_id, version, field_config, metadata, changed_by, change_reason) VALUES (?, ?, ?, ?, ?, ?)",
(
site["id"],
1,
site["field_config"],
site["metadata"],
"system",
"Initial version from existing data",
),
)
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_form_sessions():
"""Migration: add form_sessions table for Phase 10 Advanced Analytics.
Tracks user session behavior: form load → field interactions → completion/abandonment.
Populated by SDK beacon events (session_start, session_update, session_complete).
Schema:
- session_id: UUID generated client-side, reused for same browser/form (localStorage)
- visitor_id: stable hash for return visitor tracking
- ip_hash: hashed IP for dedup
- fields_viewed: JSON array of field keys the user interacted with
- last_field_reached: the furthest field key reached before submit/abandon
- step_reached: furthest step number (for multi-step forms)
"""
conn = None
try:
conn = get_db()
info = conn.execute("PRAGMA table_info(form_sessions)").fetchall()
if not info:
conn.execute("""
CREATE TABLE form_sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id),
session_id TEXT NOT NULL,
visitor_id TEXT NOT NULL,
ip_hash TEXT,
user_agent TEXT,
referrer TEXT,
event TEXT NOT NULL DEFAULT 'session_start',
started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
completed_at TIMESTAMP,
fields_viewed TEXT,
last_field_reached TEXT,
step_reached INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_form_sessions_site ON form_sessions(site_id, started_at)")
conn.execute("CREATE INDEX idx_form_sessions_visitor ON form_sessions(site_id, visitor_id)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_form_variants():
"""Migration: add form_variants table for A/B testing.
Schema:
- variant_key: 'A' or 'B' — the variant identifier
- field_config: full field configuration for this variant
- success_message: variant-specific success message
- active: whether this test is currently running
- weight: traffic split percentage (default 50/50)
"""
conn = None
try:
conn = get_db()
info = conn.execute("PRAGMA table_info(form_variants)").fetchall()
if not info:
conn.execute("""
CREATE TABLE form_variants (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id) ON DELETE CASCADE,
variant_key TEXT NOT NULL CHECK(variant_key IN ('A', 'B')),
name TEXT,
field_config TEXT NOT NULL,
success_message TEXT,
weight INTEGER DEFAULT 50,
active INTEGER DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_form_variants_site ON form_variants(site_id, active)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_branding():
"""Migration: add branding columns to sites table for Phase 11 Multi-Channel Embedding.
- og_title: Open Graph title for social sharing
- og_description: Open Graph description for social sharing
- og_image: Open Graph image URL for social sharing
- branding_color: Hex color for branded form pages
- logo_url: Custom logo URL for hosted form header
- favicon_url: Custom favicon URL for hosted form
- custom_domain: Custom domain for hosted form (Pro tier)
"""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "sites"):
conn.commit()
return
site_info = conn.execute("PRAGMA table_info(sites)").fetchall()
site_cols = [col["name"] for col in site_info]
if "og_title" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN og_title TEXT")
if "og_description" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN og_description TEXT")
if "og_image" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN og_image TEXT")
if "branding_color" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN branding_color TEXT")
if "logo_url" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN logo_url TEXT")
if "favicon_url" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN favicon_url TEXT")
if "custom_domain" not in site_cols:
conn.execute("ALTER TABLE sites ADD COLUMN custom_domain TEXT")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_webhook_retry_locking():
"""Migration: add retrying_by column to webhook_logs for atomic row locking.
Prevents duplicate webhook deliveries when multiple retry threads
run simultaneously.
"""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "webhook_logs"):
conn.commit()
return
wl_info = conn.execute("PRAGMA table_info(webhook_logs)").fetchall()
wl_cols = [col["name"] for col in wl_info]
if "retrying_by" not in wl_cols:
try:
conn.execute("ALTER TABLE webhook_logs ADD COLUMN retrying_by TEXT")
except Exception:
pass # Column may have been added manually in parallel
conn.commit()
finally:
if conn:
conn.close()
# ─── Phase 12: Document Templates ─────────────────────────────────────────────
def migrate_add_document_templates():
"""Migration: add document_templates and documents tables.
document_templates — per-site document template configs (layout, field_mapping, style)
documents — generated document instances with PDF storage paths
"""
conn = None
try:
conn = get_db()
# document_templates table
info = conn.execute("PRAGMA table_info(document_templates)").fetchall()
if not info:
conn.execute("""
CREATE TABLE document_templates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id),
name TEXT NOT NULL,
document_type TEXT NOT NULL DEFAULT 'invoice',
layout TEXT NOT NULL DEFAULT 'classic',
field_mapping TEXT NOT NULL DEFAULT '{}',
style_config TEXT NOT NULL DEFAULT '{}',
is_active INTEGER DEFAULT 1,
auto_generate INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_doc_templates_site ON document_templates(site_id)")
else:
# Add auto_generate column to existing tables
cols = [col["name"] for col in info]
if "auto_generate" not in cols:
conn.execute("ALTER TABLE document_templates ADD COLUMN auto_generate INTEGER DEFAULT 0")
# documents table
info = conn.execute("PRAGMA table_info(documents)").fetchall()
if not info:
conn.execute("""
CREATE TABLE documents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
document_id TEXT NOT NULL,
user_id INTEGER REFERENCES users(id),
site_id INTEGER REFERENCES sites(id),
submission_id INTEGER REFERENCES submissions(id),
template_id INTEGER REFERENCES document_templates(id),
document_type TEXT NOT NULL DEFAULT 'invoice',
status TEXT NOT NULL DEFAULT 'draft',
data JSON,
pdf_path TEXT,
download_count INTEGER DEFAULT 0,
expires_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_documents_site ON documents(site_id)")
conn.execute("CREATE UNIQUE INDEX idx_documents_doc_id ON documents(document_id)")
# Add due_date column for overdue detection
cols = [col["name"] for col in conn.execute("PRAGMA table_info(documents)").fetchall()]
if "due_date" not in cols:
conn.execute("ALTER TABLE documents ADD COLUMN due_date TIMESTAMP")
# invoice_schedules table (Phase 13.5: recurring invoice scheduling)
info = conn.execute("PRAGMA table_info(invoice_schedules)").fetchall()
if not info:
conn.execute("""
CREATE TABLE invoice_schedules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id),
template_id INTEGER NOT NULL REFERENCES document_templates(id),
customer_name TEXT NOT NULL,
customer_email TEXT NOT NULL,
line_items TEXT NOT NULL DEFAULT '[]',
document_number_prefix TEXT NOT NULL DEFAULT 'INV',
interval TEXT NOT NULL DEFAULT 'monthly',
interval_count INTEGER NOT NULL DEFAULT 1,
start_date DATE NOT NULL,
next_run DATE,
last_run DATE,
status TEXT NOT NULL DEFAULT 'active',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_invoice_schedules_site ON invoice_schedules(site_id)")
conn.execute("CREATE INDEX idx_invoice_schedules_next_run ON invoice_schedules(next_run)")
else:
cols = [col["name"] for col in info]
if "line_items" not in cols:
conn.execute("ALTER TABLE invoice_schedules ADD COLUMN line_items TEXT NOT NULL DEFAULT '[]'")
if "document_number_prefix" not in cols:
conn.execute(
"ALTER TABLE invoice_schedules ADD COLUMN document_number_prefix TEXT NOT NULL DEFAULT 'INV'"
)
conn.commit()
finally:
if conn:
conn.close()
def migrate_encrypt_user_pii():
"""Migration: encrypt remaining plaintext PII across users, webhook_destinations,
invoice_schedules, webhook_logs, and team_members tables.
Adds encrypted columns alongside plaintext ones for backwards compatibility.
Hashes are added for columns used in WHERE clauses (email, invited_email).
"""
conn = None
try:
conn = get_db()
# ─── Users table ───────────────────────────────────────────────────────
columns = [row[1] for row in conn.execute("PRAGMA table_info(users)").fetchall()]
if "email_encrypted" not in columns:
conn.execute("ALTER TABLE users ADD COLUMN email_encrypted TEXT")
if "email_hash" not in columns:
conn.execute("ALTER TABLE users ADD COLUMN email_hash TEXT")
conn.execute("CREATE INDEX IF NOT EXISTS idx_users_email_hash ON users(email_hash)")
if "name_encrypted" not in columns:
conn.execute("ALTER TABLE users ADD COLUMN name_encrypted TEXT")
if "display_name_encrypted" not in columns:
conn.execute("ALTER TABLE users ADD COLUMN display_name_encrypted TEXT")
# Encrypt existing plaintext user data
try:
users = conn.execute("SELECT id, email, name, display_name FROM users").fetchall() or []
except sqlite3.OperationalError:
# Columns already migrated and dropped
users = []
for u in users:
uid = u["id"]
changed = False
if u["email"] and not _is_encrypted_value(str(u["email"])):
encrypted_email = encrypt_user_value(uid, u["email"])
email_hash = hash_value(u["email"])
if encrypted_email:
conn.execute(
"UPDATE users SET email_encrypted = ?, email_hash = ? WHERE id = ?",
(encrypted_email, email_hash, uid),
)
changed = True
if u["name"] and not _is_encrypted_value(str(u["name"])):
encrypted_name = encrypt_user_value(uid, u["name"])
if encrypted_name:
conn.execute(
"UPDATE users SET name_encrypted = ? WHERE id = ?",
(encrypted_name, uid),
)
changed = True
if u["display_name"] and not _is_encrypted_value(str(u["display_name"])):
encrypted_dn = encrypt_user_value(uid, u["display_name"])
if encrypted_dn:
conn.execute(
"UPDATE users SET display_name_encrypted = ? WHERE id = ?",
(encrypted_dn, uid),
)
changed = True
# ─── Webhook destinations table ────────────────────────────────────────
wd_cols = [row[1] for row in conn.execute("PRAGMA table_info(webhook_destinations)").fetchall()]
if "url_encrypted" not in wd_cols:
conn.execute("ALTER TABLE webhook_destinations ADD COLUMN url_encrypted TEXT")
if "url_hash" not in wd_cols:
conn.execute("ALTER TABLE webhook_destinations ADD COLUMN url_hash TEXT")
dests = conn.execute("SELECT id, site_id, url FROM webhook_destinations").fetchall()
for d in dests:
if d["url"] and not _is_encrypted_value(str(d["url"])):
encrypted_url = encrypt_entity("webhook_dest", d["id"], d["url"])
if encrypted_url:
conn.execute(
"UPDATE webhook_destinations SET url_encrypted = ? WHERE id = ?",
(encrypted_url, d["id"]),
)
# Also backfill url_hash
url_hash = hash_value(d["url"])
conn.execute(
"UPDATE webhook_destinations SET url_hash = ? WHERE id = ?",
(url_hash, d["id"]),
)
# ─── Webhook logs table ────────────────────────────────────────────────
wl_cols = [row[1] for row in conn.execute("PRAGMA table_info(webhook_logs)").fetchall()]
if "webhook_url_encrypted" not in wl_cols:
conn.execute("ALTER TABLE webhook_logs ADD COLUMN webhook_url_encrypted TEXT")
logs = conn.execute("SELECT id, webhook_url FROM webhook_logs WHERE webhook_url IS NOT NULL").fetchall()
for log in logs:
if log["webhook_url"] and not _is_encrypted_value(str(log["webhook_url"])):
encrypted_url = encrypt_entity("webhook_log", log["id"], log["webhook_url"])
if encrypted_url:
conn.execute(
"UPDATE webhook_logs SET webhook_url_encrypted = ? WHERE id = ?",
(encrypted_url, log["id"]),
)
# ─── Invoice schedules table ───────────────────────────────────────────
is_cols = [row[1] for row in conn.execute("PRAGMA table_info(invoice_schedules)").fetchall()]
if "customer_name_encrypted" not in is_cols:
conn.execute("ALTER TABLE invoice_schedules ADD COLUMN customer_name_encrypted TEXT")
if "customer_email_encrypted" not in is_cols:
conn.execute("ALTER TABLE invoice_schedules ADD COLUMN customer_email_encrypted TEXT")
schedules = conn.execute("SELECT id, customer_name, customer_email FROM invoice_schedules").fetchall()
for s in schedules:
if s["customer_name"] and not _is_encrypted_value(str(s["customer_name"])):
encrypted_name = encrypt_entity("invoice_schedule", s["id"], s["customer_name"])
if encrypted_name:
conn.execute(
"UPDATE invoice_schedules SET customer_name_encrypted = ? WHERE id = ?",
(encrypted_name, s["id"]),
)
if s["customer_email"] and not _is_encrypted_value(str(s["customer_email"])):
encrypted_email = encrypt_entity("invoice_schedule", s["id"], s["customer_email"])
if encrypted_email:
conn.execute(
"UPDATE invoice_schedules SET customer_email_encrypted = ? WHERE id = ?",
(encrypted_email, s["id"]),
)
# ─── Team members table ────────────────────────────────────────────────
tm_cols = [row[1] for row in conn.execute("PRAGMA table_info(team_members)").fetchall()]
if "invited_email_encrypted" not in tm_cols:
conn.execute("ALTER TABLE team_members ADD COLUMN invited_email_encrypted TEXT")
if "invited_email_hash" not in tm_cols:
conn.execute("ALTER TABLE team_members ADD COLUMN invited_email_hash TEXT")
members = conn.execute("SELECT id, invited_email FROM team_members WHERE invited_email IS NOT NULL").fetchall()
for m in members:
if m["invited_email"] and not _is_encrypted_value(str(m["invited_email"])):
encrypted_email = encrypt_entity("team_member", m["id"], m["invited_email"])
email_hash = hash_value(m["invited_email"])
if encrypted_email:
conn.execute(
"UPDATE team_members SET invited_email_encrypted = ?, invited_email_hash = ? WHERE id = ?",
(encrypted_email, email_hash, m["id"]),
)
conn.commit()
finally:
if conn:
conn.close()
# ─── Helper for migration ─────────────────────────────────────────────────────
def migrate_add_invite_expiration():
"""Migration: add expires_at column to team_members for invite token expiry."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "team_members"):
return
cols = [row[1] for row in conn.execute("PRAGMA table_info(team_members)").fetchall()]
if "expires_at" not in cols:
conn.execute("ALTER TABLE team_members ADD COLUMN expires_at TIMESTAMP")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_user_id_to_doc_templates():
"""Migration: add user_id to document_templates and documents tables."""
conn = None
try:
conn = get_db()
# document_templates
if _table_exists(conn, "document_templates"):
cols = [row[1] for row in conn.execute("PRAGMA table_info(document_templates)").fetchall()]
if "user_id" not in cols:
conn.execute("ALTER TABLE document_templates ADD COLUMN user_id INTEGER REFERENCES users(id)")
if "invoice_number_format" not in cols:
conn.execute(
"ALTER TABLE document_templates ADD COLUMN invoice_number_format TEXT DEFAULT 'INV-YYYY-####'"
)
if "next_invoice_number" not in cols:
conn.execute("ALTER TABLE document_templates ADD COLUMN next_invoice_number TEXT")
conn.execute("""
UPDATE document_templates SET user_id = (
SELECT user_id FROM sites WHERE sites.id = document_templates.site_id
) WHERE user_id IS NULL AND site_id IS NOT NULL
""")
# documents
if _table_exists(conn, "documents"):
cols = [row[1] for row in conn.execute("PRAGMA table_info(documents)").fetchall()]
if "user_id" not in cols:
conn.execute("ALTER TABLE documents ADD COLUMN user_id INTEGER REFERENCES users(id)")
conn.execute("""
UPDATE documents SET user_id = (
SELECT user_id FROM sites WHERE sites.id = documents.site_id
) WHERE user_id IS NULL AND site_id IS NOT NULL
""")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_user_id_to_invoice_schedules():
"""Migration: add user_id to invoice_schedules table."""
conn = None
try:
conn = get_db()
if _table_exists(conn, "invoice_schedules"):
cols = [row[1] for row in conn.execute("PRAGMA table_info(invoice_schedules)").fetchall()]
if "user_id" not in cols:
conn.execute("ALTER TABLE invoice_schedules ADD COLUMN user_id INTEGER REFERENCES users(id)")
# Backfill from sites
conn.execute("""
UPDATE invoice_schedules SET user_id = (
SELECT user_id FROM sites WHERE sites.id = invoice_schedules.site_id
) WHERE user_id IS NULL AND site_id IS NOT NULL
""")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_email_phase3():
"""Migration: add sender_domain/domain_verified to email_settings + custom_templates table."""
conn = None
try:
conn = get_db()
# Add columns to email_settings
if _table_exists(conn, "email_settings"):
info = conn.execute("PRAGMA table_info(email_settings)").fetchall()
cols = [col["name"] for col in info]
if "sender_domain" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN sender_domain TEXT")
if "domain_verified" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN domain_verified BOOLEAN DEFAULT 0")
if "default_template" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN default_template TEXT DEFAULT 'default'")
# Custom templates table
if not _table_exists(conn, "custom_templates"):
conn.execute("""
CREATE TABLE custom_templates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
name TEXT NOT NULL,
content TEXT NOT NULL,
description TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE UNIQUE INDEX idx_custom_templates_user_name ON custom_templates(user_id, name)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_email_phase4():
"""Migration: Phase 4 — reminder campaigns, A/B testing, rate limiting."""
conn = None
try:
conn = get_db()
# Campaign reminders table
if not _table_exists(conn, "campaign_reminders"):
conn.execute("""
CREATE TABLE campaign_reminders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
campaign_id INTEGER NOT NULL REFERENCES campaigns(id),
delay_hours INTEGER NOT NULL,
subject TEXT NOT NULL,
body TEXT NOT NULL,
template_id TEXT,
status TEXT DEFAULT 'scheduled' CHECK(status IN ('scheduled', 'sent', 'cancelled')),
scheduled_at TIMESTAMP,
sent_at TIMESTAMP,
recipient_count INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_reminders_campaign ON campaign_reminders(campaign_id, status)")
conn.execute("CREATE INDEX idx_reminders_scheduled ON campaign_reminders(scheduled_at, status)")
# A/B test variants table
if not _table_exists(conn, "campaign_ab_variants"):
conn.execute("""
CREATE TABLE campaign_ab_variants (
id INTEGER PRIMARY KEY AUTOINCREMENT,
campaign_id INTEGER NOT NULL REFERENCES campaigns(id),
variant_label TEXT NOT NULL,
subject TEXT NOT NULL,
body TEXT,
split_percent INTEGER NOT NULL DEFAULT 50,
sent_count INTEGER DEFAULT 0,
opened_count INTEGER DEFAULT 0,
clicked_count INTEGER DEFAULT 0,
response_count INTEGER DEFAULT 0,
is_winner BOOLEAN DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_ab_variants_campaign ON campaign_ab_variants(campaign_id)")
# Add variant_id to campaign_recipients
info = conn.execute("PRAGMA table_info(campaign_recipients)").fetchall()
cols = [col["name"] for col in info]
if "variant_id" not in cols:
conn.execute(
"ALTER TABLE campaign_recipients ADD COLUMN variant_id INTEGER REFERENCES campaign_ab_variants(id)"
)
if "reminder_sent" not in cols:
conn.execute("ALTER TABLE campaign_recipients ADD COLUMN reminder_sent INTEGER DEFAULT 0")
# Rate limiting columns on email_settings
info = conn.execute("PRAGMA table_info(email_settings)").fetchall()
cols = [col["name"] for col in info]
if "max_sends_per_hour" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN max_sends_per_hour INTEGER DEFAULT 1000")
if "max_sends_per_day" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN max_sends_per_day INTEGER DEFAULT 10000")
if "max_burst_per_campaign" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN max_burst_per_campaign INTEGER DEFAULT 100")
if "sends_this_hour" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN sends_this_hour INTEGER DEFAULT 0")
if "sends_this_day" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN sends_this_day INTEGER DEFAULT 0")
if "hour_reset_at" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN hour_reset_at TIMESTAMP")
if "day_reset_at" not in cols:
conn.execute("ALTER TABLE email_settings ADD COLUMN day_reset_at TIMESTAMP")
# Email send log table
if not _table_exists(conn, "email_send_log"):
conn.execute("""
CREATE TABLE email_send_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
sent_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
count INTEGER DEFAULT 1,
campaign_id INTEGER REFERENCES campaigns(id)
)
""")
conn.execute("CREATE INDEX idx_send_log_user_time ON email_send_log(user_id, sent_at)")
# Add submission_id and tracking_token for form notification tracking
info = conn.execute("PRAGMA table_info(email_send_log)").fetchall()
cols = [col["name"] for col in info]
if "submission_id" not in cols:
conn.execute("ALTER TABLE email_send_log ADD COLUMN submission_id INTEGER REFERENCES submissions(id)")
if "tracking_token" not in cols:
conn.execute("ALTER TABLE email_send_log ADD COLUMN tracking_token TEXT")
if "idx_email_send_log_token" not in [
idx["name"] for idx in conn.execute("PRAGMA index_list(email_send_log)").fetchall()
]:
conn.execute("CREATE INDEX IF NOT EXISTS idx_email_send_log_token ON email_send_log(tracking_token)")
if "idx_email_send_log_submission" not in [
idx["name"] for idx in conn.execute("PRAGMA index_list(email_send_log)").fetchall()
]:
conn.execute("CREATE INDEX IF NOT EXISTS idx_email_send_log_submission ON email_send_log(submission_id)")
# A/B test config on campaigns table
info = conn.execute("PRAGMA table_info(campaigns)").fetchall()
cols = [col["name"] for col in info]
if "ab_test_metric" not in cols:
conn.execute("ALTER TABLE campaigns ADD COLUMN ab_test_metric TEXT DEFAULT 'opens'")
if "ab_test_size" not in cols:
conn.execute("ALTER TABLE campaigns ADD COLUMN ab_test_size INTEGER DEFAULT 20")
if "ab_declare_after_hours" not in cols:
conn.execute("ALTER TABLE campaigns ADD COLUMN ab_declare_after_hours INTEGER DEFAULT 24")
if "ab_declared" not in cols:
conn.execute("ALTER TABLE campaigns ADD COLUMN ab_declared INTEGER DEFAULT 0")
if "ab_winner_variant_id" not in cols:
conn.execute(
"ALTER TABLE campaigns ADD COLUMN ab_winner_variant_id INTEGER REFERENCES campaign_ab_variants(id)"
)
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_email_tracking():
"""Migration: add tracking columns to campaign_recipients for open/click tracking."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "campaign_recipients"):
conn.commit()
return
info = conn.execute("PRAGMA table_info(campaign_recipients)").fetchall()
cols = [col["name"] for col in info]
if "tracking_token" not in cols:
conn.execute("ALTER TABLE campaign_recipients ADD COLUMN tracking_token TEXT")
if "opened_at" not in cols:
conn.execute("ALTER TABLE campaign_recipients ADD COLUMN opened_at TIMESTAMP")
if "clicked_at" not in cols:
conn.execute("ALTER TABLE campaign_recipients ADD COLUMN clicked_at TIMESTAMP")
if "bounce_type" not in cols:
conn.execute("ALTER TABLE campaign_recipients ADD COLUMN bounce_type TEXT")
if "bounce_reason" not in cols:
conn.execute("ALTER TABLE campaign_recipients ADD COLUMN bounce_reason TEXT")
# Index for fast tracking token lookups
indexes = conn.execute("PRAGMA index_list(campaign_recipients)").fetchall()
index_names = [idx["name"] for idx in indexes]
if "idx_campaign_recipients_tracking_token" not in index_names:
conn.execute("CREATE INDEX idx_campaign_recipients_tracking_token ON campaign_recipients(tracking_token)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_email_campaigns():
"""Migration: add campaigns, campaign_recipients, email_settings tables."""
conn = None
try:
conn = get_db()
# Campaigns table
if not _table_exists(conn, "campaigns"):
conn.execute("""
CREATE TABLE campaigns (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
name TEXT NOT NULL,
subject TEXT NOT NULL,
body TEXT NOT NULL,
status TEXT DEFAULT 'draft',
from_name TEXT,
from_email TEXT,
template_id TEXT,
scheduled_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Campaign recipients table
if not _table_exists(conn, "campaign_recipients"):
conn.execute("""
CREATE TABLE campaign_recipients (
id INTEGER PRIMARY KEY AUTOINCREMENT,
campaign_id INTEGER NOT NULL REFERENCES campaigns(id),
email TEXT NOT NULL,
status TEXT DEFAULT 'pending',
message_id TEXT,
error TEXT,
sent_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_campaign_recipients_campaign ON campaign_recipients(campaign_id)")
# Email settings table
if not _table_exists(conn, "email_settings"):
conn.execute("""
CREATE TABLE email_settings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
from_name TEXT,
from_email TEXT,
reply_to TEXT,
template_id TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
finally:
if conn:
conn.close()
# Invite token expiration (days) — override via env
INVITE_TOKEN_EXPIRY_DAYS = int(os.environ.get("INVITE_TOKEN_EXPIRY_DAYS", "7"))
def migrate_add_form_actions():
"""Migration: add form_actions table for post-submission actions.
Supports webhook, email, log, redirect, and document action types.
Each action is tied to a site and triggered on submission events.
"""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "form_actions"):
conn.execute("""
CREATE TABLE form_actions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
site_id INTEGER NOT NULL REFERENCES sites(id),
type TEXT NOT NULL,
config TEXT NOT NULL DEFAULT '{}',
trigger_event TEXT DEFAULT 'submission',
enabled INTEGER DEFAULT 1,
execution_order INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("CREATE INDEX idx_form_actions_site ON form_actions(site_id, execution_order)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_user_referral_cols():
"""Migration: add referral_code and referred_by columns to users table."""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "users"):
return
info = conn.execute("PRAGMA table_info(users)").fetchall()
cols = [col["name"] for col in info]
if "referral_code" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN referral_code TEXT")
if "referred_by" not in cols:
conn.execute("ALTER TABLE users ADD COLUMN referred_by INTEGER REFERENCES users(id)")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_agents():
"""Migration: add Phase C agent registry and call log tables.
agents — registry of external AI agents (Claude, Codex, custom).
agent_calls — execution log for agent invocations from chains.
"""
conn = None
try:
conn = get_db()
# ─── agents table ──────────────────────────────────────────────
if _table_exists(conn, "agents"):
return
conn.execute("""
CREATE TABLE agents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
name TEXT NOT NULL,
endpoint_url TEXT NOT NULL,
api_key_encrypted TEXT,
capabilities TEXT DEFAULT '[]',
metadata TEXT DEFAULT '{}',
max_timeout_seconds INTEGER DEFAULT 60,
enabled INTEGER DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE INDEX idx_agents_user_id ON agents(user_id)
""")
conn.execute("""
CREATE INDEX idx_agents_enabled ON agents(enabled)
""")
# ─── agent_calls table ─────────────────────────────────────────
conn.execute("""
CREATE TABLE agent_calls (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id),
agent_id INTEGER NOT NULL REFERENCES agents(id),
chain_action_id INTEGER REFERENCES form_actions(id),
site_id INTEGER REFERENCES sites(id),
submission_id INTEGER REFERENCES submissions(id),
request_payload TEXT,
response_payload TEXT,
status TEXT DEFAULT 'pending',
error_message TEXT,
duration_ms INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
completed_at TIMESTAMP
)
""")
conn.execute("""
CREATE INDEX idx_agent_calls_agent_id ON agent_calls(agent_id)
""")
conn.execute("""
CREATE INDEX idx_agent_calls_user_id ON agent_calls(user_id)
""")
conn.execute("""
CREATE INDEX idx_agent_calls_status ON agent_calls(status)
""")
conn.commit()
finally:
if conn:
conn.close()
def migrate_add_document_generation_status():
"""Migration: add generation_status, generation_error columns to documents table.
generation_status — tracks async PDF generation progress:
pending, processing, completed, failed
generation_error — error message when generation_status = 'failed'
"""
conn = None
try:
conn = get_db()
if not _table_exists(conn, "documents"):
return
info = conn.execute("PRAGMA table_info(documents)").fetchall()
cols = [col["name"] for col in info]
if "generation_status" not in cols:
conn.execute(
"ALTER TABLE documents ADD COLUMN generation_status TEXT DEFAULT 'completed'"
)
if "generation_error" not in cols:
conn.execute("ALTER TABLE documents ADD COLUMN generation_error TEXT")
conn.commit()
finally:
if conn:
conn.close()