Blame for backend/app/database.py

import time

from sqlalchemy import (
    create_engine,
    Column,
    Float,
    Integer,
    String,
    Text,
    DateTime,
    JSON,
    ForeignKey,
    UniqueConstraint,
    event,
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker
from datetime import datetime
import os
from pathlib import Path

# DATABASE_URL (Postgres, e.g. Azure Database for PostgreSQL) takes over when set; otherwise this
# falls back to the original SQLite-on-local-disk behavior unchanged (Hugging Face Spaces provides
# persistent storage at /data; override with HF_DATA_DIR for local development where /data isn't
# writable). Two deployments, two databases, same codebase — neither path touches the other.
DATABASE_URL = os.getenv("DATABASE_URL", "").strip()

if DATABASE_URL:
    # SQLAlchemy 1.4+/2.x rejects the "postgres://" scheme some providers still hand out (a
    # Heroku-era convention); normalize to "postgresql://" so it always works regardless of
    # which form the connection string arrives in.
    if DATABASE_URL.startswith("postgres://"):
        DATABASE_URL = "postgresql://" + DATABASE_URL[len("postgres://"):]
    engine = create_engine(DATABASE_URL, pool_pre_ping=True)
else:
    DATA_DIR = Path(os.getenv("HF_DATA_DIR", "/data"))
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    DB_PATH = os.getenv("HF_DATABASE_PATH", str(DATA_DIR / "emailout.db"))
    engine = create_engine(
        f"sqlite:///{DB_PATH}",
        connect_args={"check_same_thread": False, "timeout": 30},
    )

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

IS_POSTGRES = engine.dialect.name == "postgresql"

if not IS_POSTGRES:
    @event.listens_for(engine, "connect")
    def _sqlite_pragmas(dbapi_connection, _connection_record):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA journal_mode=WAL")
        cursor.execute("PRAGMA synchronous=NORMAL")
        cursor.execute("PRAGMA busy_timeout=30000")
        cursor.close()

# Dialect-portable SQL fragments for the additive migrations and raw inserts below — Postgres has
# no DATETIME type (use TIMESTAMP) and no datetime('now') function (use NOW()). Kept as constants
# rather than scattering dialect checks through every migration line.
SQL_DATETIME_TYPE = "TIMESTAMP" if IS_POSTGRES else "DATETIME"
SQL_NOW = "NOW()" if IS_POSTGRES else "datetime('now')"


def db_commit_with_retry(session: Session, *, attempts: int = 5) -> None:
    """Retry commits when SQLite is busy (common under concurrent webhooks + campaign ticks)."""
    last_err = None
    for i in range(max(1, attempts)):
        try:
            session.commit()
            return
        except Exception as e:
            last_err = e
            msg = str(e).lower()
            if "locked" not in msg and "busy" not in msg:
                raise
            session.rollback()
            time.sleep(min(0.05 * (2**i), 1.0))
    if last_err:
        raise last_err


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    google_sub = Column(String, unique=True, index=True, nullable=True)
    microsoft_sub = Column(String, unique=True, index=True, nullable=True)  # oid:tid — see auth_common.upsert_user_by_provider
    microsoft_tenant_id = Column(String, nullable=True)
    email = Column(String, index=True)
    name = Column(String, nullable=True)
    picture = Column(String, nullable=True)
    google_refresh_token = Column(Text, nullable=True)  # for Gmail send (invite emails); treat as secret
    microsoft_refresh_token = Column(Text, nullable=True)  # for Graph mail send (invite emails); treat as secret
    linkedin_profile_display_name = Column(String, nullable=True)
    default_unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=True)
    mailbox_profile_display_name = Column(String, nullable=True)
    default_mailbox_unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)


class Tenant(Base):
    __tablename__ = "tenants"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)


class TenantMembership(Base):
    __tablename__ = "tenant_memberships"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    role = Column(String, default="member", nullable=False)  # admin | member
    team_id = Column(Integer, ForeignKey("teams.id"), nullable=True, index=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    __table_args__ = (UniqueConstraint("user_id", "tenant_id", name="uq_user_tenant"),)


class Team(Base):
    """A sales team headed by one manager (TenantMembership.team_id points here). The manager
    is themself a member of their own team, so 'my team' queries naturally include their own
    records alongside the members they head."""

    __tablename__ = "teams"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    name = Column(String, nullable=False)
    manager_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class Invitation(Base):
    __tablename__ = "invitations"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    email = Column(String, nullable=False, index=True)
    token_hash = Column(String(64), unique=True, nullable=False, index=True)
    role = Column(String, default="member")
    team_id = Column(Integer, ForeignKey("teams.id"), nullable=True)
    invited_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
    expires_at = Column(DateTime, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    accepted_at = Column(DateTime, nullable=True)


class UploadedFile(Base):
    __tablename__ = "uploaded_files"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    file_id = Column(String, unique=True, index=True)
    filename = Column(String)
    contact_count = Column(Integer)
    file_path = Column(String)
    sequence_plan_json = Column(Text, nullable=True)
    campaign_name = Column(String, nullable=True)  # user-facing campaign name — the backend record of "this campaign exists"
    status = Column(String, default="draft")  # draft | generating | ready | launched
    # Email send pacing (campaign-setup-time config) — NULL timezone means no business-hours restriction.
    email_timezone = Column(String, nullable=True)  # IANA zone name, e.g. "America/New_York"
    email_business_hours_start = Column(Integer, default=8)  # local hour, inclusive, 0-23
    email_business_hours_end = Column(Integer, default=18)  # local hour, exclusive, 0-23
    email_business_days = Column(String, default="1,2,3,4,5")  # CSV ISO weekdays, Mon=1..Sun=7
    email_send_interval_seconds = Column(Integer, default=60)  # min gap between two email sends for this campaign
    generation_cancel_requested = Column(Integer, default=0)  # 0/1 flag checked mid-stream by the SSE generator
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class Prompt(Base):
    __tablename__ = "prompts"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    file_id = Column(String, index=True)
    product_name = Column(String)
    prompt_template = Column(Text)
    prompt_kind = Column(String, default="email")  # email | linkedin
    created_at = Column(DateTime, default=datetime.utcnow)


class GeneratedSequence(Base):
    __tablename__ = "generated_sequences"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    file_id = Column(String, index=True)
    sequence_id = Column(Integer)  # Contact sequence number
    email_number = Column(Integer, default=1)  # Email number in sequence (1-10)
    channel = Column(String, default="email")  # email | linkedin
    first_name = Column(String)
    last_name = Column(String)
    email = Column(String)
    company = Column(String)
    title = Column(String)
    product = Column(String)
    subject = Column(String)
    email_content = Column(Text)
    step_order = Column(Integer, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)


class Contact(Base):
    __tablename__ = "contacts"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    file_id = Column(String, index=True)
    row_index = Column(Integer)
    first_name = Column(String)
    last_name = Column(String)
    email = Column(String, index=True)
    company = Column(String)
    title = Column(String)
    phone = Column(String, default="")
    source = Column(String, default="apollo_csv")
    raw_data = Column(JSON)
    # UniPile LinkedIn outreach (set when a campaign invite/DM is executed)
    unipile_provider_id = Column(String, nullable=True, index=True)
    linkedin_invite_sent_at = Column(DateTime, nullable=True)
    linkedin_invite_pending = Column(Integer, default=0)  # 1 = waiting on acceptance webhook
    linkedin_connection_accepted_at = Column(DateTime, nullable=True)
    linkedin_last_followup_sent_at = Column(DateTime, nullable=True)
    linkedin_followup_next_email_number = Column(Integer, nullable=True)
    outreach_stopped_at = Column(DateTime, nullable=True)
    outreach_stop_reason = Column(String, nullable=True)
    owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    created_at = Column(DateTime, default=datetime.utcnow())


class Company(Base):
    """Account-level record — either created manually or synced (by name match, one-directionally)
    from distinct Contact.company values. Not linked to Contact via a foreign key: association is
    a case-insensitive name match, kept intentionally soft so this table is fully additive and
    never touches existing Contact/CrmLead/CrmDeal rows or behavior."""

    __tablename__ = "companies"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    name = Column(String, nullable=False, index=True)
    category = Column(String, default="")  # A | B | C | ""
    type = Column(String, default="")  # Existing | New | ""
    company_name_for_emails = Column(String, default="")
    industry = Column(String, default="")
    employees = Column(String, default="")
    annual_revenue = Column(String, default="")
    last_raised_at = Column(String, default="")
    website = Column(String, default="")
    city = Column(String, default="")
    state = Column(String, default="")
    country = Column(String, default="")
    phone = Column(String, default="")  # main company line — company-level, synced like the rest
    source = Column(String, default="manual")  # manual | synced_from_contacts
    owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class ContactActivity(Base):
    """A logged Call or Visit against a Contact — who logged it, what was discussed, when.
    Powers the per-contact and per-company activity timelines. Always tied to a real Contact
    row (never a bare Lead) so company-level rollups can reuse the same case-insensitive
    Contact.company name match already used for associated-contacts lists."""

    __tablename__ = "contact_activities"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    contact_id = Column(Integer, ForeignKey("contacts.id"), nullable=False, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
    activity_type = Column(String, nullable=False)  # "call" | "visit"
    notes = Column(Text, default="")
    created_at = Column(DateTime, default=datetime.utcnow, index=True)


class CrmLead(Base):
    """Lead synced from Smartlead replies (webhook) — CRM pipeline status is local."""

    __tablename__ = "crm_leads"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    smartlead_lead_id = Column(String, index=True)
    campaign_id = Column(String, index=True)
    campaign_name = Column(String)
    email = Column(String, index=True)
    first_name = Column(String)
    last_name = Column(String)
    company_name = Column(String)
    title = Column(String)
    phone = Column(String, default="")  # the lead's own personal phone — not company-synced
    last_reply_subject = Column(String)
    last_reply_body = Column(Text)
    last_reply_at = Column(DateTime, nullable=True)
    crm_status = Column(String, default="new_lead")  # new_lead|contacted|qualified|unqualified|none
    contact_id = Column(Integer, nullable=True)  # links to contacts.id after "Move to Contacts"
    owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    raw_webhook = Column(JSON)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class CrmDeal(Base):
    """Pipeline deal (often converted from a lead)."""

    __tablename__ = "crm_deals"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    name = Column(String, index=True)
    stage = Column(String, default="new")  # new|discovery|proposal|negotiation|won|lost
    owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    owner_initials = Column(String, default="")  # display cache; derived from User when owner_user_id set
    deal_value = Column(Integer, nullable=True)  # whole USD (nullable)
    revenue_type = Column(String, default="arr")  # arr|qrr|mrr|one_time
    contact_display = Column(String, default="")  # primary person label
    account_name = Column(String, default="")
    expected_close_date = Column(DateTime, nullable=True)
    close_probability = Column(Integer, default=10)  # 0–100
    country = Column(String, default="")
    last_interaction_at = Column(DateTime, nullable=True)
    source_lead_id = Column(Integer, nullable=True)
    source_campaign_name = Column(String, default="")
    contact_id = Column(Integer, nullable=True)
    # When stage is lost: reason + optional detail fields
    lost_reason = Column(String, default="")  # price|no_decision|chose_another|didnt_meet_expectation|unknown
    lost_price_discount_pct = Column(Float, nullable=True)
    lost_chose_product_name = Column(String, default="")
    # When stage is won: invoicing / PO snapshot
    won_po_number = Column(String, default="")
    won_customer_legal_name = Column(String, default="")
    won_customer_address = Column(Text, default="")
    won_contact_person_name = Column(String, default="")
    won_channel_partner_name = Column(String, default="")
    won_note_to_customer = Column(Text, default="")
    won_note_to_accounts = Column(Text, default="")
    won_line_items = Column(JSON, nullable=True)
    won_at = Column(DateTime, nullable=True)  # first transition to won (for quarterly attribution)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class SmartleadRun(Base):
    __tablename__ = "smartlead_runs"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    file_id = Column(String, index=True)
    run_id = Column(String, unique=True, index=True)
    campaign_id = Column(String, index=True)
    campaign_name = Column(String)
    mode = Column(String)  # 'existing' or 'new'
    steps_count = Column(Integer)
    dry_run = Column(Integer, default=0)  # 0 = false, 1 = true
    total_leads = Column(Integer)
    added_leads = Column(Integer, default=0)
    skipped_leads = Column(Integer, default=0)
    failed_leads = Column(Integer, default=0)
    error_details = Column(Text)  # JSON string of errors
    status = Column(String)  # 'pending', 'completed', 'failed'
    created_at = Column(DateTime, default=datetime.utcnow)
    completed_at = Column(DateTime, nullable=True)


class UnipileAccount(Base):
    __tablename__ = "unipile_accounts"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    label = Column(String, default="")
    provider = Column(String, default="LINKEDIN")
    unipile_account_id = Column(String, index=True)
    status = Column(String, default="pending")
    auth_mode = Column(String, default="hosted")
    metadata_json = Column(JSON, nullable=True)
    # Safe-sending caps (LinkedIn accounts only, harmless/unread on mailbox accounts). Defaults are
    # conservative research-backed numbers — Unipile's own platform ceiling is ~80-100 connects/day,
    # but sustained volume well under that is what actually avoids LinkedIn's restriction flags.
    linkedin_daily_connection_cap = Column(Integer, default=20)
    linkedin_daily_message_cap = Column(Integer, default=40)
    linkedin_weekly_connection_cap = Column(Integer, default=100)
    linkedin_weekly_message_cap = Column(Integer, default=200)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class LinkedinCampaign(Base):
    __tablename__ = "linkedin_campaigns"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    name = Column(String, index=True)
    status = Column(String, default="draft")  # draft | generated | running | completed | failed
    unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=False, index=True)
    file_id = Column(String, index=True, nullable=True)
    contact_count = Column(Integer, default=0)
    prompt_template = Column(Text, default="")
    execution_result = Column(JSON, nullable=True)
    followup_interval_hours = Column(Integer, default=72)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    executed_at = Column(DateTime, nullable=True)


class FileOutreachExecution(Base):
    """Server-side orchestration for wizard email+LinkedIn sequences (per uploaded CSV file)."""

    __tablename__ = "file_outreach_executions"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    file_id = Column(String, index=True, nullable=False)
    status = Column(String, default="idle")  # idle | running | paused | completed
    progress_done = Column(Integer, default=0)
    progress_total = Column(Integer, default=0)
    stats_json = Column(Text, nullable=True)
    cursor_json = Column(Text, nullable=True)  # delivery_started_at, etc.
    last_error = Column(Text, nullable=True)
    linkedin_followup_interval_hours = Column(Integer, default=72)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    __table_args__ = (UniqueConstraint("tenant_id", "file_id", name="uq_outreach_exec_tenant_file"),)


class OutreachSendReceipt(Base):
    """One row per (file, contact, sequence action slot) send attempt."""

    __tablename__ = "outreach_send_receipts"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    file_id = Column(String, index=True, nullable=False)
    contact_id = Column(Integer, ForeignKey("contacts.id"), nullable=False, index=True)
    step_slot = Column(Integer, nullable=False)
    channel = Column(String, nullable=False)
    action = Column(String, default="")
    status = Column(String, default="pending")  # sent | failed | skipped
    detail = Column(Text, nullable=True)
    unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=True, index=True)
    unipile_message_id = Column(String, nullable=True)
    unipile_tracking_id = Column(String, nullable=True, index=True)
    smtp_message_id = Column(String, nullable=True, index=True)
    tracking_label = Column(String(220), nullable=True, index=True)
    opened_at = Column(DateTime, nullable=True)
    open_count = Column(Integer, default=0)
    replied_at = Column(DateTime, nullable=True)
    bounced_at = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    __table_args__ = (
        UniqueConstraint("tenant_id", "file_id", "contact_id", "step_slot", name="uq_outreach_receipt_slot"),
    )


class UnipileHostedAuthState(Base):
    __tablename__ = "unipile_hosted_auth_states"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    token = Column(String, unique=True, index=True)
    provider = Column(String, default="LINKEDIN")
    label = Column(String, default="")
    expires_at = Column(DateTime, nullable=True)
    used_at = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)


class CtaForm(Base):
    """Embeddable lead-capture form (WordPress / external sites → Leads)."""

    __tablename__ = "cta_forms"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    public_id = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False, default="")
    fields = Column(JSON, nullable=False, default=list)
    embed_width = Column(String, default="100%")
    embed_height = Column(String, default="480")
    is_active = Column(Integer, default=1)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class LandingPage(Base):
    """AI-generated public landing pages with CTA form embed."""

    __tablename__ = "landing_pages"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
    public_id = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False, default="")
    prompt = Column(String, default="")
    content = Column(JSON, nullable=False, default=dict)
    cta_form_id = Column(Integer, ForeignKey("cta_forms.id"), nullable=True, index=True)
    banner_asset = Column(String, nullable=True)
    logo_asset = Column(String, nullable=True)
    is_active = Column(Integer, default=1)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class ChatAgentConfig(Base):
    """One row per tenant: Chat Agent widget config, Drive connection, website source, model choice."""

    __tablename__ = "chat_agent_configs"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, unique=True, index=True)
    is_enabled = Column(Integer, default=1)
    public_id = Column(String, unique=True, index=True, nullable=False)
    greeting_message = Column(
        Text,
        default="Hi! 👋 I'm happy to help answer any questions. What's your name and email, in case we get disconnected?",
    )
    chat_model_provider = Column(String, default="openai")  # openai | gemini

    website_base_url = Column(String, default="")
    website_status = Column(String, default="not_configured")  # not_configured|idle|running|success|error
    website_last_indexed_at = Column(DateTime, nullable=True)

    drive_folder_id = Column(String, default="")
    drive_folder_url = Column(String, default="")
    drive_refresh_token = Column(Text, nullable=True)
    drive_connected_email = Column(String, default="")
    drive_status = Column(String, default="not_connected")  # not_connected|idle|running|success|error
    drive_last_indexed_at = Column(DateTime, nullable=True)

    reindex_interval_hours = Column(Integer, default=24)

    # Opt-in, off by default: lets the agentic loop call a Gemini web-search tool when the
    # knowledge base doesn't have enough for a detailed answer (e.g. a named competitor,
    # industry/technology background). Results are always presented as unverified external
    # content — see SKILLS-SECURITY.md for the residual risk this doesn't eliminate (a visitor's
    # message can still influence what query gets researched).
    enable_web_search = Column(Integer, default=0)

    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class ChatAgentSkill(Base):
    """
    Admin-editable skill, following Anthropic's Agent Skills anatomy (SKILL.md: YAML
    frontmatter name+description + markdown instructions, plus references/assets/templates/
    scripts/evals). `instructions` (the "brain") is always compiled into the chat system prompt;
    references, assets, and templates are indexed into the same searchable knowledge base as
    Drive/website content so the agentic search_knowledge_base tool can surface them on demand.
    `scripts_json` can only be executed via the admin-only, sandboxed "Run script" action
    (chat_agent_script_runner.py) — the live public chat agent never executes it (see
    SKILLS-SECURITY.md for why).
    """

    __tablename__ = "chat_agent_skills"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    text = Column(Text, nullable=False, default="")  # legacy; superseded by `instructions` below
    name = Column(String, default="")
    description = Column(Text, default="")
    instructions = Column(Text, default="")
    references_json = Column(JSON, nullable=True)  # [{title, content}]
    assets_json = Column(JSON, nullable=True)  # [{name, content}]
    templates_json = Column(JSON, nullable=True)  # [{name, content}]
    scripts_json = Column(JSON, nullable=True)  # [{name, description, language, content, last_run_*}]
    evals_json = Column(JSON, nullable=True)  # [{prompt, assertion_type, assertion_value, last_run_*}]
    is_default = Column(Integer, default=0)
    is_active = Column(Integer, default=1)
    sort_order = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class ChatAgentDocument(Base):
    """One ingested source item (a Drive file or a crawled website page), converted to markdown."""

    __tablename__ = "chat_agent_documents"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    source_type = Column(String, nullable=False, index=True)  # drive | website
    source_ref = Column(String, nullable=False, index=True)  # drive file id | page URL
    title = Column(String, default="")
    content_markdown = Column(Text, default="")
    content_hash = Column(String, default="", index=True)
    external_modified_at = Column(DateTime, nullable=True)
    indexed_at = Column(DateTime, nullable=True)
    status = Column(String, default="ok")  # ok | error
    error_detail = Column(Text, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    __table_args__ = (
        UniqueConstraint("tenant_id", "source_type", "source_ref", name="uq_chat_doc_tenant_source"),
    )


class ChatAgentChunk(Base):
    """Embedded retrieval unit for a ChatAgentDocument."""

    __tablename__ = "chat_agent_chunks"

    id = Column(Integer, primary_key=True, index=True)
    document_id = Column(Integer, ForeignKey("chat_agent_documents.id"), nullable=False, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    chunk_index = Column(Integer, default=0)
    text = Column(Text, default="")
    embedding = Column(JSON, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)


class ChatAgentConversation(Base):
    """One website-visitor chat session (public widget)."""

    __tablename__ = "chat_agent_conversations"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    session_id = Column(String, unique=True, index=True, nullable=False)
    visitor_name = Column(String, default="")
    visitor_email = Column(String, default="", index=True)
    visitor_company = Column(String, default="")
    lead_id = Column(Integer, ForeignKey("crm_leads.id"), nullable=True, index=True)
    page_url = Column(String, default="")
    started_at = Column(DateTime, default=datetime.utcnow)
    last_message_at = Column(DateTime, default=datetime.utcnow)


class ChatAgentMessage(Base):
    __tablename__ = "chat_agent_messages"

    id = Column(Integer, primary_key=True, index=True)
    conversation_id = Column(Integer, ForeignKey("chat_agent_conversations.id"), nullable=False, index=True)
    role = Column(String, nullable=False)  # user | assistant
    content = Column(Text, default="")
    # True when this reply used the search_the_web tool — surfaced to admins in the Test Chat
    # and Conversations views so they can tell live-web-sourced answers apart from knowledge-base
    # answers at a glance.
    used_web_search = Column(Integer, default=0)
    # Total LLM tokens (prompt + completion, summed across every round of the agentic loop) spent
    # generating this reply — 0 for user messages and for any reply where the provider didn't
    # report usage. Surfaced per-message and summed per-conversation in the Conversations tab.
    total_tokens = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)


class ChatAgentIndexLog(Base):
    __tablename__ = "chat_agent_index_logs"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    source_type = Column(String, nullable=False)  # drive | website
    status = Column(String, default="running")  # running | success | error
    documents_processed = Column(Integer, default=0)
    chunks_created = Column(Integer, default=0)
    started_at = Column(DateTime, default=datetime.utcnow)
    finished_at = Column(DateTime, nullable=True)
    detail = Column(Text, nullable=True)


# ---------------------------------------------------------------------------
# Qualifier Agent (RFQ Qualifier) — a deliberately SEPARATE skill/ingestion system from the
# Chat Agent tables above, per design decision: each background agent (Qualifier, and later Quote
# Estimator) gets its own independent skill model + knowledge base rather than a shared/generalized
# one. Structurally the same anatomy as ChatAgentSkill/Document/Chunk/IndexLog (SKILL.md-compatible
# skill row, hash-diffed embedded chunks, index-run log) but intentionally not reusing those tables
# or any cross-FK to them — this whole block could be lifted into a standalone service/repo without
# touching the Chat Agent feature. See qualifier_common.py / qualifier_ingest.py / qualifier_agent.py.
# ---------------------------------------------------------------------------


class QualifierSkill(Base):
    """
    Admin-editable skill for the RFQ Qualifier agent — same SKILL.md anatomy as ChatAgentSkill
    (name+description+instructions, references/assets/templates/scripts/evals), compiled/parsed by
    qualifier_common.py. `instructions` (the qualify policy) is always injected into the agent's
    system prompt; references/assets/templates are indexed into qualifier_chunks alongside the
    Wittur pricelist so the agentic loop's search_pricelist-adjacent lookups can surface them.
    `scripts_json` is admin-run-only via qualifier_script_runner.py — the background agent never
    executes it.
    """

    __tablename__ = "qualifier_skills"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    name = Column(String, default="")
    description = Column(Text, default="")
    instructions = Column(Text, default="")
    references_json = Column(JSON, nullable=True)  # [{title, content}]
    assets_json = Column(JSON, nullable=True)  # [{name, content}]
    templates_json = Column(JSON, nullable=True)  # [{name, content}]
    scripts_json = Column(JSON, nullable=True)  # [{name, description, language, content, last_run_*}]
    evals_json = Column(JSON, nullable=True)  # [{name, input_ref, expected_decision, expected_project_type, assertion_type, assertion_value, last_run_*}]
    is_default = Column(Integer, default=0)
    is_active = Column(Integer, default=1)
    sort_order = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class QualifierDocument(Base):
    """One ingested source item for the Qualifier's knowledge base — currently only source_type
    'pricelist' (the Wittur pricelist PDF, re-ingested on demand) plus 'skill_reference' /
    'skill_asset' / 'skill_template' (a QualifierSkill's own content, indexed the same way)."""

    __tablename__ = "qualifier_documents"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    source_type = Column(String, nullable=False, index=True)  # pricelist | skill_reference | skill_asset | skill_template
    source_ref = Column(String, nullable=False, index=True)  # e.g. "pricelist:page:3" or "skill:1:reference:0"
    title = Column(String, default="")
    content_markdown = Column(Text, default="")
    content_hash = Column(String, default="", index=True)
    indexed_at = Column(DateTime, nullable=True)
    status = Column(String, default="ok")  # ok | error
    error_detail = Column(Text, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    __table_args__ = (
        UniqueConstraint("tenant_id", "source_type", "source_ref", name="uq_qualifier_doc_tenant_source"),
    )


class QualifierChunk(Base):
    """Embedded retrieval unit for a QualifierDocument — searched by the agent's search_pricelist
    tool call (qualifier_retrieval.py)."""

    __tablename__ = "qualifier_chunks"

    id = Column(Integer, primary_key=True, index=True)
    document_id = Column(Integer, ForeignKey("qualifier_documents.id"), nullable=False, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    chunk_index = Column(Integer, default=0)
    text = Column(Text, default="")
    embedding = Column(JSON, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)


class QualifierIndexLog(Base):
    __tablename__ = "qualifier_index_logs"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    source_type = Column(String, nullable=False)  # pricelist
    status = Column(String, default="running")  # running | success | error
    documents_processed = Column(Integer, default=0)
    chunks_created = Column(Integer, default=0)
    started_at = Column(DateTime, default=datetime.utcnow)
    finished_at = Column(DateTime, nullable=True)
    detail = Column(Text, nullable=True)


class QualifierRun(Base):
    """One qualification run: an RFQ input (uploaded .eml, or PDF/DOCX spec, or a future
    inbox-polling trigger) processed through the Qualifier agent end to end. This is both the
    audit trail and the source of the "past runs" list in the dashboard."""

    __tablename__ = "qualifier_runs"

    id = Column(Integer, primary_key=True, index=True)
    tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True)
    skill_id = Column(Integer, ForeignKey("qualifier_skills.id"), nullable=True, index=True)
    created_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)

    trigger_type = Column(String, default="manual_upload")  # manual_upload | api | inbox_poll (future)
    input_type = Column(String, nullable=False, default="")  # eml | pdf | docx
    input_filename = Column(String, default="")
    input_file_path = Column(String, default="")  # storage path/blob ref to the raw upload, see qualifier_storage.py

    project_name = Column(String, default="")
    sender_email = Column(String, default="")
    deadline_text = Column(String, default="")

    status = Column(String, default="running")  # running | complete | error
    decision = Column(String, nullable=True)  # qualify | disqualify | needs_review
    project_type = Column(String, nullable=True)  # modernization | new_construction | unknown
    confidence = Column(Float, nullable=True)
    result_json = Column(JSON, nullable=True)  # full structured QualifierDecision object
    extracted_text_excerpt = Column(Text, nullable=True)  # what was actually handed to the model, for debugging
    error_detail = Column(Text, nullable=True)
    total_tokens = Column(Integer, default=0)

    started_at = Column(DateTime, default=datetime.utcnow)
    finished_at = Column(DateTime, nullable=True)


def run_migrations(connection_engine):
    """Add tenant_id to legacy SQLite tables and attach rows to the default tenant."""
    from sqlalchemy import inspect, text

    insp = inspect(connection_engine)
    tenant_tables = (
        "uploaded_files",
        "prompts",
        "generated_sequences",
        "contacts",
        "crm_leads",
        "crm_deals",
        "smartlead_runs",
        "unipile_accounts",
        "linkedin_campaigns",
        "unipile_hosted_auth_states",
        "cta_forms",
        "landing_pages",
        "chat_agent_configs",
        "chat_agent_skills",
        "chat_agent_documents",
        "chat_agent_chunks",
        "chat_agent_conversations",
        "chat_agent_index_logs",
    )

    with connection_engine.begin() as conn:
        if insp.has_table("tenants"):
            n = conn.execute(text("SELECT COUNT(*) FROM tenants")).scalar()
            if n == 0:
                conn.execute(
                    text(
                        f"INSERT INTO tenants (name, created_at) VALUES (:name, {SQL_NOW})"
                    ),
                    {"name": "Default workspace"},
                )

        default_tid = conn.execute(text("SELECT id FROM tenants ORDER BY id LIMIT 1")).scalar()
        if default_tid is None:
            default_tid = 1

        insp = inspect(connection_engine)
        for tbl in tenant_tables:
            if not insp.has_table(tbl):
                continue
            cols = [c["name"] for c in insp.get_columns(tbl)]
            if "tenant_id" not in cols:
                conn.execute(text(f"ALTER TABLE {tbl} ADD COLUMN tenant_id INTEGER"))

        for tbl in tenant_tables:
            if not insp.has_table(tbl):
                continue
            conn.execute(
                text(f"UPDATE {tbl} SET tenant_id = :tid WHERE tenant_id IS NULL"),
                {"tid": default_tid},
            )

        insp = inspect(connection_engine)
        if insp.has_table("users"):
            ucols = [c["name"] for c in insp.get_columns("users")]
            if "google_refresh_token" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN google_refresh_token TEXT"))
            if "linkedin_profile_display_name" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN linkedin_profile_display_name TEXT"))
            if "default_unipile_account_ref_id" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN default_unipile_account_ref_id INTEGER"))
            if "mailbox_profile_display_name" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN mailbox_profile_display_name TEXT"))
            if "default_mailbox_unipile_account_ref_id" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN default_mailbox_unipile_account_ref_id INTEGER"))
            if "microsoft_sub" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN microsoft_sub TEXT"))
                conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_users_microsoft_sub ON users (microsoft_sub)"))
            if "microsoft_tenant_id" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN microsoft_tenant_id TEXT"))
            if "microsoft_refresh_token" not in ucols:
                conn.execute(text("ALTER TABLE users ADD COLUMN microsoft_refresh_token TEXT"))
            # google_sub was NOT NULL from day one (every user signed in with Google); a
            # Microsoft-only user has no Google identity, so relax it. Postgres-only (production)
            # — instant, no table rewrite. Skipped on SQLite (dev-only; ALTER COLUMN isn't
            # supported there and no production data is ever on SQLite).
            if IS_POSTGRES:
                col_info = next((c for c in insp.get_columns("users") if c["name"] == "google_sub"), None)
                if col_info is not None and not col_info.get("nullable", True):
                    conn.execute(text("ALTER TABLE users ALTER COLUMN google_sub DROP NOT NULL"))

        insp = inspect(connection_engine)
        if insp.has_table("crm_deals"):
            dcols = [c["name"] for c in insp.get_columns("crm_deals")]
            if "owner_user_id" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN owner_user_id INTEGER"))
            if "revenue_type" not in dcols:
                conn.execute(
                    text("ALTER TABLE crm_deals ADD COLUMN revenue_type TEXT DEFAULT 'arr'")
                )
                conn.execute(
                    text(
                        "UPDATE crm_deals SET revenue_type = 'arr' WHERE revenue_type IS NULL OR revenue_type = ''"
                    )
                )
            if "lost_reason" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN lost_reason TEXT DEFAULT ''"))
            if "lost_price_discount_pct" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN lost_price_discount_pct REAL"))
            if "lost_chose_product_name" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN lost_chose_product_name TEXT DEFAULT ''"))
            if "won_po_number" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_po_number TEXT DEFAULT ''"))
            if "won_customer_legal_name" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_customer_legal_name TEXT DEFAULT ''"))
            if "won_customer_address" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_customer_address TEXT DEFAULT ''"))
            if "won_contact_person_name" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_contact_person_name TEXT DEFAULT ''"))
            if "won_channel_partner_name" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_channel_partner_name TEXT DEFAULT ''"))
            if "won_note_to_customer" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_note_to_customer TEXT DEFAULT ''"))
            if "won_note_to_accounts" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_note_to_accounts TEXT DEFAULT ''"))
            if "won_line_items" not in dcols:
                conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_line_items TEXT"))
            if "won_at" not in dcols:
                conn.execute(text(f"ALTER TABLE crm_deals ADD COLUMN won_at {SQL_DATETIME_TYPE}"))
                conn.execute(
                    text(
                        "UPDATE crm_deals SET won_at = updated_at WHERE stage = 'won' AND won_at IS NULL"
                    )
                )

        insp = inspect(connection_engine)
        if insp.has_table("prompts"):
            pcols = [c["name"] for c in insp.get_columns("prompts")]
            if "prompt_kind" not in pcols:
                conn.execute(text("ALTER TABLE prompts ADD COLUMN prompt_kind TEXT DEFAULT 'email'"))
                conn.execute(
                    text(
                        "UPDATE prompts SET prompt_kind = 'email' WHERE prompt_kind IS NULL OR prompt_kind = ''"
                    )
                )

        insp = inspect(connection_engine)
        if insp.has_table("generated_sequences"):
            gcols = [c["name"] for c in insp.get_columns("generated_sequences")]
            if "channel" not in gcols:
                conn.execute(text("ALTER TABLE generated_sequences ADD COLUMN channel TEXT DEFAULT 'email'"))
                conn.execute(
                    text(
                        "UPDATE generated_sequences SET channel = 'email' WHERE channel IS NULL OR channel = ''"
                    )
                )
            gcols2 = [c["name"] for c in insp.get_columns("generated_sequences")]
            if "step_order" not in gcols2:
                conn.execute(text("ALTER TABLE generated_sequences ADD COLUMN step_order INTEGER"))

        insp = inspect(connection_engine)
        if insp.has_table("uploaded_files"):
            ufcols = [c["name"] for c in insp.get_columns("uploaded_files")]
            if "sequence_plan_json" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN sequence_plan_json TEXT"))
            if "campaign_name" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN campaign_name TEXT"))
            if "status" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN status TEXT DEFAULT 'draft'"))
            if "updated_at" not in ufcols:
                conn.execute(text(f"ALTER TABLE uploaded_files ADD COLUMN updated_at {SQL_DATETIME_TYPE}"))
            if "email_timezone" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_timezone TEXT"))
            if "email_business_hours_start" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_business_hours_start INTEGER DEFAULT 8"))
            if "email_business_hours_end" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_business_hours_end INTEGER DEFAULT 18"))
            if "email_business_days" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_business_days TEXT DEFAULT '1,2,3,4,5'"))
            if "email_send_interval_seconds" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_send_interval_seconds INTEGER DEFAULT 60"))
            if "generation_cancel_requested" not in ufcols:
                conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN generation_cancel_requested INTEGER DEFAULT 0"))

        insp = inspect(connection_engine)
        if insp.has_table("chat_agent_conversations"):
            caccols = [c["name"] for c in insp.get_columns("chat_agent_conversations")]
            if "visitor_company" not in caccols:
                conn.execute(text("ALTER TABLE chat_agent_conversations ADD COLUMN visitor_company TEXT DEFAULT ''"))

        insp = inspect(connection_engine)
        if insp.has_table("unipile_accounts"):
            uacols = [c["name"] for c in insp.get_columns("unipile_accounts")]
            if "user_id" not in uacols:
                conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN user_id INTEGER"))
            if "linkedin_daily_connection_cap" not in uacols:
                conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_daily_connection_cap INTEGER DEFAULT 20"))
            if "linkedin_daily_message_cap" not in uacols:
                conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_daily_message_cap INTEGER DEFAULT 40"))
            if "linkedin_weekly_connection_cap" not in uacols:
                conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_weekly_connection_cap INTEGER DEFAULT 100"))
            if "linkedin_weekly_message_cap" not in uacols:
                conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_weekly_message_cap INTEGER DEFAULT 200"))

        insp = inspect(connection_engine)
        if insp.has_table("linkedin_campaigns"):
            lccols = [c["name"] for c in insp.get_columns("linkedin_campaigns")]
            if "user_id" not in lccols:
                conn.execute(text("ALTER TABLE linkedin_campaigns ADD COLUMN user_id INTEGER"))
            if "followup_interval_hours" not in lccols:
                conn.execute(
                    text("ALTER TABLE linkedin_campaigns ADD COLUMN followup_interval_hours INTEGER DEFAULT 72")
                )
                conn.execute(
                    text(
                        "UPDATE linkedin_campaigns SET followup_interval_hours = 72 WHERE followup_interval_hours IS NULL"
                    )
                )

        insp = inspect(connection_engine)
        if insp.has_table("contacts"):
            ctcols = [c["name"] for c in insp.get_columns("contacts")]
            if "unipile_provider_id" not in ctcols:
                conn.execute(text("ALTER TABLE contacts ADD COLUMN unipile_provider_id TEXT"))
            if "linkedin_invite_sent_at" not in ctcols:
                conn.execute(text(f"ALTER TABLE contacts ADD COLUMN linkedin_invite_sent_at {SQL_DATETIME_TYPE}"))
            if "linkedin_invite_pending" not in ctcols:
                conn.execute(text("ALTER TABLE contacts ADD COLUMN linkedin_invite_pending INTEGER DEFAULT 0"))
            if "linkedin_connection_accepted_at" not in ctcols:
                conn.execute(text(f"ALTER TABLE contacts ADD COLUMN linkedin_connection_accepted_at {SQL_DATETIME_TYPE}"))
            if "linkedin_last_followup_sent_at" not in ctcols:
                conn.execute(text(f"ALTER TABLE contacts ADD COLUMN linkedin_last_followup_sent_at {SQL_DATETIME_TYPE}"))
            if "linkedin_followup_next_email_number" not in ctcols:
                conn.execute(text("ALTER TABLE contacts ADD COLUMN linkedin_followup_next_email_number INTEGER"))
            if "outreach_stopped_at" not in ctcols:
                conn.execute(text(f"ALTER TABLE contacts ADD COLUMN outreach_stopped_at {SQL_DATETIME_TYPE}"))
            if "outreach_stop_reason" not in ctcols:
                conn.execute(text("ALTER TABLE contacts ADD COLUMN outreach_stop_reason TEXT"))
            if "phone" not in ctcols:
                conn.execute(text("ALTER TABLE contacts ADD COLUMN phone TEXT DEFAULT ''"))
            if "owner_user_id" not in ctcols:
                conn.execute(text("ALTER TABLE contacts ADD COLUMN owner_user_id INTEGER"))

        insp = inspect(connection_engine)
        if insp.has_table("crm_leads"):
            leadcols = [c["name"] for c in insp.get_columns("crm_leads")]
            if "phone" not in leadcols:
                conn.execute(text("ALTER TABLE crm_leads ADD COLUMN phone TEXT DEFAULT ''"))
            if "owner_user_id" not in leadcols:
                conn.execute(text("ALTER TABLE crm_leads ADD COLUMN owner_user_id INTEGER"))

        insp = inspect(connection_engine)
        if insp.has_table("tenant_memberships"):
            tmcols = [c["name"] for c in insp.get_columns("tenant_memberships")]
            if "team_id" not in tmcols:
                conn.execute(text("ALTER TABLE tenant_memberships ADD COLUMN team_id INTEGER"))

        insp = inspect(connection_engine)
        if insp.has_table("companies"):
            compcols = [c["name"] for c in insp.get_columns("companies")]
            if "phone" not in compcols:
                conn.execute(text("ALTER TABLE companies ADD COLUMN phone TEXT DEFAULT ''"))
            if "type" not in compcols:
                conn.execute(text("ALTER TABLE companies ADD COLUMN type TEXT DEFAULT ''"))
            if "owner_user_id" not in compcols:
                conn.execute(text("ALTER TABLE companies ADD COLUMN owner_user_id INTEGER"))

        insp = inspect(connection_engine)
        if insp.has_table("file_outreach_executions"):
            ecols = [c["name"] for c in insp.get_columns("file_outreach_executions")]
            if "linkedin_followup_interval_hours" not in ecols:
                conn.execute(
                    text(
                        "ALTER TABLE file_outreach_executions ADD COLUMN linkedin_followup_interval_hours INTEGER DEFAULT 72"
                    )
                )

        insp = inspect(connection_engine)
        if insp.has_table("outreach_send_receipts"):
            rcols = [c["name"] for c in insp.get_columns("outreach_send_receipts")]
            if "tracking_label" not in rcols:
                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN tracking_label TEXT"))
            if "unipile_tracking_id" not in rcols:
                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN unipile_tracking_id TEXT"))
            if "unipile_account_ref_id" not in rcols:
                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN unipile_account_ref_id INTEGER"))
            if "smtp_message_id" not in rcols:
                conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN smtp_message_id TEXT"))
            if "opened_at" not in rcols:
                conn.execute(text(f"ALTER TABLE outreach_send_receipts ADD COLUMN opened_at {SQL_DATETIME_TYPE}"))
            if "open_count" not in rcols:
                conn.execute(
                    text("ALTER TABLE outreach_send_receipts ADD COLUMN open_count INTEGER DEFAULT 0")
                )
            if "replied_at" not in rcols:
                conn.execute(text(f"ALTER TABLE outreach_send_receipts ADD COLUMN replied_at {SQL_DATETIME_TYPE}"))
            if "bounced_at" not in rcols:
                conn.execute(text(f"ALTER TABLE outreach_send_receipts ADD COLUMN bounced_at {SQL_DATETIME_TYPE}"))

        insp = inspect(connection_engine)
        if insp.has_table("cta_forms"):
            ccols = [c["name"] for c in insp.get_columns("cta_forms")]
            if "embed_width" not in ccols:
                conn.execute(text("ALTER TABLE cta_forms ADD COLUMN embed_width TEXT DEFAULT '100%'"))
            if "embed_height" not in ccols:
                conn.execute(text("ALTER TABLE cta_forms ADD COLUMN embed_height TEXT DEFAULT '480'"))

        insp = inspect(connection_engine)
        if insp.has_table("chat_agent_skills"):
            skcols = [c["name"] for c in insp.get_columns("chat_agent_skills")]
            if "name" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN name TEXT DEFAULT ''"))
            if "description" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN description TEXT"))
            if "instructions" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN instructions TEXT"))
            if "references_json" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN references_json TEXT"))
            if "assets_json" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN assets_json TEXT"))
            if "templates_json" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN templates_json TEXT"))
            if "scripts_json" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN scripts_json TEXT"))
            if "evals_json" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN evals_json TEXT"))
            if "sort_order" not in skcols:
                conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN sort_order INTEGER DEFAULT 0"))
            # Backfill legacy rows (created before `instructions` existed) so every skill has a
            # real name/description/instructions triple, matching the SKILL.md anatomy.
            conn.execute(
                text(
                    "UPDATE chat_agent_skills SET instructions = text "
                    "WHERE (instructions IS NULL OR instructions = '') AND text IS NOT NULL"
                )
            )
            conn.execute(
                text(
                    "UPDATE chat_agent_skills SET name = 'skill-' || id "
                    "WHERE name IS NULL OR name = ''"
                )
            )
            conn.execute(
                text(
                    "UPDATE chat_agent_skills SET description = 'Legacy skill — edit to add a name and description' "
                    "WHERE description IS NULL OR description = ''"
                )
            )
            # One-time consolidation: the Skills feature moved from 3 small default guardrail
            # skills to one comprehensive default skill. A tenant whose skills are EXACTLY the old
            # 3 defaults (all is_default=1, no custom skills added) gets them cleared here —
            # ensure_default_skills() (chat_agent_common.py) already reseeds automatically once a
            # tenant has zero skills, so it picks up the new comprehensive one on next access. Any
            # tenant with a custom (is_default=0) skill is left untouched, so real admin edits are
            # never silently destroyed.
            _old_default_names = sorted(["no-internal-pricing", "no-client-names", "no-sensitive-info"])
            _rows = conn.execute(text("SELECT tenant_id, name, is_default FROM chat_agent_skills")).fetchall()
            _by_tenant: dict = {}
            for _tid, _name, _is_default in _rows:
                _by_tenant.setdefault(_tid, []).append((_name, _is_default))
            for _tid, _items in _by_tenant.items():
                _names = sorted(n for n, _ in _items)
                _all_default = all(bool(d) for _, d in _items)
                if _all_default and _names == _old_default_names:
                    conn.execute(
                        text("DELETE FROM chat_agent_skills WHERE tenant_id = :tid"),
                        {"tid": _tid},
                    )

        insp = inspect(connection_engine)
        if insp.has_table("chat_agent_configs"):
            cfgcols = [c["name"] for c in insp.get_columns("chat_agent_configs")]
            if "enable_web_search" not in cfgcols:
                conn.execute(
                    text("ALTER TABLE chat_agent_configs ADD COLUMN enable_web_search INTEGER DEFAULT 0")
                )

        if insp.has_table("chat_agent_messages"):
            msgcols = [c["name"] for c in insp.get_columns("chat_agent_messages")]
            if "used_web_search" not in msgcols:
                conn.execute(
                    text("ALTER TABLE chat_agent_messages ADD COLUMN used_web_search INTEGER DEFAULT 0")
                )
            if "total_tokens" not in msgcols:
                conn.execute(
                    text("ALTER TABLE chat_agent_messages ADD COLUMN total_tokens INTEGER DEFAULT 0")
                )

        insp = inspect(connection_engine)
        if insp.has_table("invitations"):
            invcols = [c["name"] for c in insp.get_columns("invitations")]
            if "team_id" not in invcols:
                conn.execute(text("ALTER TABLE invitations ADD COLUMN team_id INTEGER"))


# Create tables then migrate legacy SQLite schemas
Base.metadata.create_all(bind=engine)
run_migrations(engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()