"""
BizNode SQLite Database Layer
============================
Handles structured state and business metadata storage.
Tables: notes, note_links, leads, owner_actions, associates, agent_identity, businesses

This is part of the AI Obsidian Memory Layer.
"""

import os
import json
from typing import Dict, Any, List, Optional
from datetime import datetime

# Database path alias — kept for backwards compatibility
_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DB_PATH = os.getenv("SQLITE_PATH", os.path.join(_PROJECT_ROOT, "memory", "biznode.db"))


def get_connection():
    """Get database connection — delegates to db_adapter (supports SQLite/MySQL/PostgreSQL)."""
    from memory.db_adapter import get_connection as _get
    return _get()


def init_db():
    """Initialize database with all required tables."""
    conn = get_connection()
    cursor = conn.cursor()
    
    # === AI Obsidian Memory Tables ===
    
    # Notes table - stores business notes with summaries
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            node_id TEXT,
            title TEXT,
            content TEXT,
            summary TEXT,
            tags TEXT,
            embedding_id TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # Note links - for semantic backlinking
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS note_links (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_id INTEGER,
            target_id INTEGER,
            similarity REAL,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (source_id) REFERENCES notes(id),
            FOREIGN KEY (target_id) REFERENCES notes(id)
        )
    """)
    
    # === Business Registration Tables ===
    
    # Businesses table
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS businesses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            node_id TEXT UNIQUE,
            business_name TEXT,
            owner_telegram_id TEXT,
            owner_email TEXT,
            status TEXT DEFAULT 'active',
            wallet_address TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # === Lead Management Tables ===
    
    # Leads table - for marketing mode
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS leads (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT,
            business TEXT,
            contact_info TEXT,
            summary TEXT,
            source TEXT,
            status TEXT DEFAULT 'new',
            embedding_id TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # === Owner Authority Tables ===
    
    # Owner actions - tracks owner approvals
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS owner_actions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            action_type TEXT,
            data TEXT,
            status TEXT DEFAULT 'pending',
            owner_response TEXT,
            risk_level TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            resolved_at DATETIME
        )
    """)
    
    # === 1bz Associate Network Tables ===
    
    # Associates table - pre-registered network partners
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS associates (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            network_id TEXT UNIQUE,
            name TEXT,
            telegram_id TEXT,
            email TEXT,
            role TEXT,
            business_type TEXT,
            status TEXT DEFAULT 'active',
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # Network interactions - logs associate communications
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS network_interactions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            associate_id INTEGER,
            interaction_type TEXT,
            description TEXT,
            initiated_by TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (associate_id) REFERENCES associates(id)
        )
    """)
    
    # === Agent Identity Table ===
    
    # AI Agent identity configuration
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS agent_identity (
            id INTEGER PRIMARY KEY,
            agent_name TEXT,
            agent_email TEXT,
            telegram_bot_token TEXT,
            smtp_host TEXT,
            smtp_port INTEGER,
            smtp_user TEXT,
            smtp_password TEXT,
            owner_telegram_id TEXT,
            owner_email TEXT,
            autonomy_level INTEGER DEFAULT 1,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)
    
    # === Business Profile Migration ===
    # Add columns to existing agent_identity table (safe no-op if already present)
    for _col, _ctype in [
        ("business_type",        "TEXT"),
        ("business_description", "TEXT"),
        ("services_offered",     "TEXT"),
        ("ai_persona_name",      "TEXT"),
        ("greeting_message",     "TEXT"),
        ("tone",                 "TEXT DEFAULT 'professional'"),
        ("faqs",                 "TEXT DEFAULT '[]'"),
    ]:
        try:
            cursor.execute(f"ALTER TABLE agent_identity ADD COLUMN {_col} {_ctype}")
        except Exception:
            pass  # column already exists

    # === 1bz Network Sync Table ===

    # Sync registry for decentralized tracking
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS sync_registry (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            node_id TEXT,
            hash_data TEXT,
            signature TEXT,
            sync_status TEXT DEFAULT 'pending',
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            synced_at DATETIME
        )
    """)

    # === Decentralized Automation Engine Tables ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS scheduled_tasks (
            id TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            type TEXT NOT NULL,
            schedule_cron TEXT,
            interval_seconds INTEGER,
            action_json TEXT NOT NULL,
            owner_node_id TEXT,
            status TEXT DEFAULT 'active',
            last_run DATETIME,
            next_run DATETIME,
            run_count INTEGER DEFAULT 0,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS task_permissions (
            id TEXT PRIMARY KEY,
            task_id TEXT NOT NULL,
            shared_with TEXT NOT NULL,
            permission TEXT NOT NULL,
            mcp_key_id TEXT,
            expires_at DATETIME,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (task_id) REFERENCES scheduled_tasks(id) ON DELETE CASCADE
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS mcp_keys (
            id TEXT PRIMARY KEY,
            task_id TEXT,
            permission TEXT NOT NULL,
            issued_to TEXT NOT NULL,
            node_id TEXT,
            token_b64 TEXT NOT NULL,
            status TEXT DEFAULT 'active',
            expires_at DATETIME,
            last_used DATETIME,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS trusted_nodes (
            node_id TEXT PRIMARY KEY,
            display_name TEXT,
            public_key_b64 TEXT NOT NULL,
            endpoint TEXT,
            capabilities TEXT,
            status TEXT DEFAULT 'trusted',
            added_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            last_seen DATETIME
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS backup_config (
            id            INTEGER PRIMARY KEY,
            provider      TEXT NOT NULL UNIQUE,
            enabled       INTEGER DEFAULT 0,
            schedule_cron TEXT DEFAULT '0 2 * * *',
            retention     INTEGER DEFAULT 7,
            config_json   TEXT,
            last_backup   DATETIME,
            created_at    DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS backup_history (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            provider    TEXT NOT NULL,
            filename    TEXT NOT NULL,
            size_bytes  INTEGER DEFAULT 0,
            status      TEXT NOT NULL,
            error       TEXT,
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Marketing Hub Tables ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS query_logs (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            telegram_id TEXT NOT NULL,
            username    TEXT,
            query       TEXT NOT NULL,
            response    TEXT,
            source      TEXT DEFAULT 'telegram',
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS blocked_customers (
            telegram_id TEXT PRIMARY KEY,
            reason      TEXT,
            blocked_by  TEXT DEFAULT 'owner',
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS email_instructions (
            id            INTEGER PRIMARY KEY AUTOINCREMENT,
            raw_email     TEXT,
            parsed_action TEXT,
            status        TEXT DEFAULT 'pending',
            result        TEXT,
            created_at    DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS email_log (
            id           INTEGER PRIMARY KEY AUTOINCREMENT,
            direction    TEXT NOT NULL,
            email_type   TEXT DEFAULT 'general',
            from_addr    TEXT,
            to_addr      TEXT,
            subject      TEXT,
            body_preview TEXT,
            status       TEXT DEFAULT 'ok',
            error_msg    TEXT,
            created_at   DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS conversation_history (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            telegram_id TEXT NOT NULL,
            role        TEXT NOT NULL,
            message     TEXT NOT NULL,
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_conv_tg
        ON conversation_history(telegram_id, id DESC)
    """)

    # === Dynamic Menu System ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS menu_items (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            menu_name   TEXT NOT NULL,
            menu_type   TEXT NOT NULL,
            link        TEXT,
            task_code   TEXT,
            sort_order  INTEGER DEFAULT 0,
            is_active   INTEGER DEFAULT 1,
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS workflows (
            id               INTEGER PRIMARY KEY AUTOINCREMENT,
            workflow_code    TEXT UNIQUE NOT NULL,
            description      TEXT,
            trigger_command  TEXT,
            is_active        INTEGER DEFAULT 1,
            created_at       DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS workflow_steps (
            id            INTEGER PRIMARY KEY AUTOINCREMENT,
            workflow_code TEXT NOT NULL,
            step_order    INTEGER NOT NULL,
            step_type     TEXT NOT NULL,
            step_prompt   TEXT,
            field_name    TEXT,
            created_at    DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS user_workflow_state (
            id             INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id        TEXT NOT NULL,
            workflow_code  TEXT NOT NULL,
            current_step   INTEGER DEFAULT 1,
            collected_data TEXT DEFAULT '{}',
            status         TEXT DEFAULT 'active',
            updated_at     DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS otp_codes (
            id         INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id    TEXT NOT NULL,
            email      TEXT NOT NULL,
            otp_hash   TEXT NOT NULL,
            expires_at DATETIME NOT NULL,
            attempts   INTEGER DEFAULT 0,
            verified   INTEGER DEFAULT 0,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Payment Agent Tables ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS services (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            name        TEXT NOT NULL,
            description TEXT,
            price       REAL NOT NULL,
            currency    TEXT DEFAULT 'USD',
            duration    TEXT,
            sort_order  INTEGER DEFAULT 0,
            is_active   INTEGER DEFAULT 1,
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS payment_accounts (
            id            INTEGER PRIMARY KEY AUTOINCREMENT,
            provider_key  TEXT NOT NULL,
            display_name  TEXT NOT NULL,
            config_json   TEXT DEFAULT '{}',
            regions       TEXT DEFAULT '*',
            is_active     INTEGER DEFAULT 1,
            sort_order    INTEGER DEFAULT 0,
            created_at    DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS payment_requests (
            id             INTEGER PRIMARY KEY AUTOINCREMENT,
            reference_no   TEXT UNIQUE NOT NULL,
            telegram_id    TEXT,
            customer_name  TEXT,
            service_id     INTEGER,
            service_name   TEXT,
            amount         REAL,
            currency       TEXT DEFAULT 'USD',
            provider_key   TEXT,
            payment_link   TEXT,
            status         TEXT DEFAULT 'pending',
            customer_utr   TEXT,
            owner_notes    TEXT,
            created_at     DATETIME DEFAULT CURRENT_TIMESTAMP,
            confirmed_at   DATETIME
        )
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_pay_req_status
        ON payment_requests(status, created_at DESC)
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_pay_req_telegram
        ON payment_requests(telegram_id, created_at DESC)
    """)

    # === Client Files Table ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS client_files (
            id             INTEGER PRIMARY KEY AUTOINCREMENT,
            name           TEXT NOT NULL,
            email          TEXT,
            phone          TEXT,
            company        TEXT,
            website        TEXT,
            address        TEXT,
            telegram_id    TEXT,
            notes          TEXT,
            tags           TEXT,
            status         TEXT DEFAULT 'active',
            source         TEXT DEFAULT 'manual',
            contract_value REAL,
            created_at     DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at     DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Bot Users Table (onboarding state machine) ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS bot_users (
            telegram_id   TEXT PRIMARY KEY,
            username      TEXT DEFAULT '',
            name          TEXT DEFAULT '',
            email         TEXT DEFAULT '',
            business      TEXT DEFAULT '',
            step          TEXT DEFAULT 'new',
            verified      INTEGER DEFAULT 0,
            lead_id       INTEGER DEFAULT NULL,
            created_at    DATETIME DEFAULT CURRENT_TIMESTAMP,
            last_seen     DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # Add columns to bot_users if missing
    for _col_sql in [
        "ALTER TABLE bot_users ADD COLUMN person_type TEXT DEFAULT 'visitor'",
        "ALTER TABLE bot_users ADD COLUMN crypto_wallet TEXT DEFAULT ''",
        "ALTER TABLE bot_users ADD COLUMN wallet_verified INTEGER DEFAULT 0",
    ]:
        try:
            cursor.execute(_col_sql)
        except Exception:
            pass  # Column already exists

    # === Transactions Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS transactions (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            reference_no    TEXT UNIQUE NOT NULL,
            telegram_id     TEXT,
            person_type     TEXT DEFAULT 'customer',
            transaction_type TEXT NOT NULL,
            title           TEXT NOT NULL,
            description     TEXT,
            amount          REAL DEFAULT 0,
            currency        TEXT DEFAULT 'USD',
            status          TEXT DEFAULT 'initiated',
            payment_method  TEXT,
            payment_ref     TEXT,
            metadata_json   TEXT DEFAULT '{}',
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            completed_at    DATETIME
        )
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_txn_telegram
        ON transactions(telegram_id, created_at DESC)
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_txn_status
        ON transactions(status, created_at DESC)
    """)

    # === Affiliate Links Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS affiliate_links (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            title       TEXT NOT NULL,
            url         TEXT NOT NULL,
            description TEXT,
            category    TEXT DEFAULT 'general',
            commission  TEXT,
            is_active   INTEGER DEFAULT 1,
            click_count INTEGER DEFAULT 0,
            sort_order  INTEGER DEFAULT 0,
            created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # Add sort_order to affiliate_links if missing
    try:
        cursor.execute("""
            ALTER TABLE affiliate_links ADD COLUMN sort_order INTEGER DEFAULT 0
        """)
    except Exception:
        pass  # Column already exists

    # === Customer Engagement Log ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS engagement_log (
            id           INTEGER PRIMARY KEY AUTOINCREMENT,
            telegram_id  TEXT,
            email        TEXT,
            person_type  TEXT,
            engagement_type TEXT NOT NULL,
            subject      TEXT,
            status       TEXT DEFAULT 'sent',
            created_at   DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Group Advertisements Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS group_ads (
            id           INTEGER PRIMARY KEY AUTOINCREMENT,
            title        TEXT NOT NULL,
            message      TEXT NOT NULL,
            target_group TEXT,
            ad_type      TEXT DEFAULT 'promotion',
            schedule_cron TEXT,
            is_active    INTEGER DEFAULT 1,
            last_sent    DATETIME,
            send_count   INTEGER DEFAULT 0,
            created_at   DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === License Purchases Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS license_purchases (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            reference_no    TEXT UNIQUE NOT NULL,
            telegram_id     TEXT,
            email           TEXT NOT NULL,
            customer_name   TEXT,
            license_type    TEXT NOT NULL DEFAULT 'basic',
            license_key     TEXT,
            stripe_session_id TEXT,
            stripe_payment_intent TEXT,
            transaction_hash TEXT,
            payment_method  TEXT DEFAULT 'stripe',
            amount          REAL DEFAULT 0,
            currency        TEXT DEFAULT 'USD',
            status          TEXT DEFAULT 'pending',
            usb_serial      TEXT,
            machine_fingerprint TEXT,
            activated_at    DATETIME,
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at      DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_lic_purchase_email
        ON license_purchases(email, created_at DESC)
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_lic_purchase_key
        ON license_purchases(license_key)
    """)

    # === Crypto Payment OTP Table (separate from user OTP) ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS crypto_payment_otp (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            reference_no    TEXT NOT NULL,
            telegram_id     TEXT,
            email           TEXT NOT NULL,
            otp_hash        TEXT NOT NULL,
            amount          REAL DEFAULT 0,
            currency        TEXT DEFAULT 'USD',
            license_type    TEXT DEFAULT 'basic',
            status          TEXT DEFAULT 'pending',
            transaction_hash TEXT,
            expires_at      DATETIME NOT NULL,
            attempts        INTEGER DEFAULT 0,
            verified_at     DATETIME,
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # ── DataVault: Plugin Registry ─────────────────────────────────
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS datavault_plugins (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            plugin_id       TEXT UNIQUE NOT NULL,
            name            TEXT NOT NULL,
            description     TEXT,
            version         TEXT DEFAULT '1.0.0',
            author          TEXT,
            plugin_path     TEXT NOT NULL,
            commands_json   TEXT DEFAULT '[]',
            config_json     TEXT DEFAULT '{}',
            category        TEXT DEFAULT 'general',
            is_active       INTEGER DEFAULT 1,
            tier_required   TEXT DEFAULT 'PRO',
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at      DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # ── DataVault: Generated Reports ────────────────────────────────
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS datavault_reports (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            report_id       TEXT UNIQUE NOT NULL,
            plugin_id       TEXT NOT NULL,
            command         TEXT NOT NULL,
            title           TEXT NOT NULL,
            description     TEXT,
            input_args      TEXT DEFAULT '{}',
            md_path         TEXT,
            pdf_path        TEXT,
            file_size       INTEGER DEFAULT 0,
            category        TEXT DEFAULT 'general',
            tags            TEXT DEFAULT '',
            price           REAL DEFAULT 0.0,
            currency        TEXT DEFAULT 'BZeUSD',
            access_tier     TEXT DEFAULT 'BASIC',
            creator_node    TEXT,
            quality_score   REAL DEFAULT 0.0,
            download_count  INTEGER DEFAULT 0,
            is_public       INTEGER DEFAULT 1,
            status          TEXT DEFAULT 'ready',
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (plugin_id) REFERENCES datavault_plugins(plugin_id)
        )
    """)

    # ── DataVault: Report Purchases ─────────────────────────────────
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS datavault_purchases (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            purchase_id     TEXT UNIQUE NOT NULL,
            report_id       TEXT NOT NULL,
            buyer_node      TEXT,
            buyer_telegram  TEXT,
            buyer_email     TEXT,
            amount          REAL NOT NULL,
            currency        TEXT DEFAULT 'BZeUSD',
            payment_method  TEXT DEFAULT 'BZeUSD',
            payment_ref     TEXT,
            status          TEXT DEFAULT 'pending',
            access_key      TEXT,
            expires_at      DATETIME,
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (report_id) REFERENCES datavault_reports(report_id)
        )
    """)

    # ── DataVault: Report Packages (bundles) ─────────────────────────
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS datavault_packages (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            package_id      TEXT UNIQUE NOT NULL,
            plugin_id       TEXT NOT NULL,
            name            TEXT NOT NULL,
            description     TEXT,
            commands_csv    TEXT NOT NULL,
            price_public    REAL DEFAULT 10.0,
            currency_public TEXT DEFAULT 'BZeUSD',
            price_node      REAL DEFAULT 1.0,
            currency_node   TEXT DEFAULT 'DZIT',
            is_active       INTEGER DEFAULT 1,
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (plugin_id) REFERENCES datavault_plugins(plugin_id)
        )
    """)

    # ── DataVault: Package Orders ────────────────────────────────────
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS datavault_package_orders (
            id              INTEGER PRIMARY KEY AUTOINCREMENT,
            order_id        TEXT UNIQUE NOT NULL,
            package_id      TEXT NOT NULL,
            buyer_email     TEXT NOT NULL,
            buyer_telegram  TEXT,
            target_url      TEXT,
            is_node         INTEGER DEFAULT 0,
            amount          REAL NOT NULL,
            currency        TEXT DEFAULT 'BZeUSD',
            payment_method  TEXT DEFAULT 'BZeUSD',
            payment_ref     TEXT,
            polygonscan_url TEXT,
            status          TEXT DEFAULT 'pending',
            created_at      DATETIME DEFAULT CURRENT_TIMESTAMP,
            completed_at    DATETIME,
            FOREIGN KEY (package_id) REFERENCES datavault_packages(package_id)
        )
    """)

    # === Node Orchestrator Tables ===

    # Node registry — tracks all 1BZ network nodes
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS node_registry (
            node_id TEXT PRIMARY KEY,
            display_name TEXT NOT NULL,
            endpoint TEXT,
            capabilities TEXT,
            cpu_cores INTEGER DEFAULT 0,
            gpu_available BOOLEAN DEFAULT FALSE,
            ram_gb REAL DEFAULT 0,
            reputation_score REAL DEFAULT 50.0,
            uptime_pct REAL DEFAULT 0.0,
            last_heartbeat TEXT,
            status TEXT DEFAULT 'unknown',
            region TEXT DEFAULT 'local',
            version TEXT,
            registered_at TEXT DEFAULT CURRENT_TIMESTAMP,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Reputation System Tables ===

    # Node reputation events — audit trail of score changes
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS node_reputation (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            node_id TEXT NOT NULL,
            event_type TEXT NOT NULL,
            score_delta REAL NOT NULL,
            details TEXT,
            recorded_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # Service-level scores — per-node, per-service performance metrics
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS service_scores (
            node_id TEXT NOT NULL,
            service_name TEXT NOT NULL,
            total_requests INTEGER DEFAULT 0,
            successful INTEGER DEFAULT 0,
            failed INTEGER DEFAULT 0,
            avg_response_ms REAL DEFAULT 0,
            failure_rate REAL DEFAULT 0.0,
            last_updated TEXT DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (node_id, service_name)
        )
    """)

    # === Agent Registry (CA Verification) ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS agent_registry (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            agent_id TEXT UNIQUE NOT NULL,
            agent_name TEXT NOT NULL,
            agent_description TEXT,
            agent_endpoint TEXT,
            agent_capabilities TEXT,
            owner_name TEXT NOT NULL,
            owner_email TEXT NOT NULL,
            owner_type TEXT DEFAULT 'individual',
            legal_entity_name TEXT,
            registration_number TEXT,
            country TEXT,
            verification_status TEXT DEFAULT 'unverified',
            verification_date TEXT,
            ca_reference TEXT,
            ca_certificate_id TEXT,
            submitted_at TEXT DEFAULT CURRENT_TIMESTAMP,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
            listed_publicly BOOLEAN DEFAULT TRUE,
            node_id TEXT,
            telegram_bot_username TEXT,
            website_url TEXT
        )
    """)

    # === Support Tickets Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS support_tickets (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            email TEXT NOT NULL,
            subject TEXT,
            body TEXT,
            attachments TEXT,
            issue_category TEXT,
            auto_response TEXT,
            status TEXT DEFAULT 'open',
            priority TEXT DEFAULT 'normal',
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            resolved_at DATETIME
        )
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_support_tickets_status
        ON support_tickets(status, created_at DESC)
    """)

    # === Support Ticket Replies Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS support_ticket_replies (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ticket_id INTEGER REFERENCES support_tickets(id),
            direction TEXT NOT NULL,
            message TEXT,
            attachments TEXT,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Published Tools Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS published_tools (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            tool_id TEXT UNIQUE NOT NULL,
            agent_id TEXT NOT NULL,
            node_id TEXT NOT NULL,
            name TEXT NOT NULL,
            description TEXT,
            category TEXT,
            input_schema TEXT,
            output_schema TEXT,
            cost_amount REAL DEFAULT 0,
            cost_currency TEXT DEFAULT 'DZIT',
            access_level TEXT DEFAULT 'public',
            handler_url TEXT,
            version TEXT DEFAULT '1.0.0',
            is_active INTEGER DEFAULT 1,
            call_count INTEGER DEFAULT 0,
            total_earned REAL DEFAULT 0,
            manifest_hash TEXT,
            signature TEXT,
            published_at TEXT DEFAULT CURRENT_TIMESTAMP,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Hosted Agents Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS hosted_agents (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            agent_id TEXT UNIQUE NOT NULL,
            owner_email TEXT NOT NULL,
            owner_name TEXT,
            telegram_id TEXT,
            tier TEXT DEFAULT 'basic',
            bot_token TEXT,
            bot_username TEXT,
            business_name TEXT,
            business_description TEXT,
            industry TEXT DEFAULT 'general',
            max_bots INTEGER DEFAULT 1,
            dzit_balance REAL DEFAULT 10.0,
            bzeusd_balance REAL DEFAULT 0.0,
            tools_published INTEGER DEFAULT 0,
            workflows_published INTEGER DEFAULT 0,
            status TEXT DEFAULT 'active',
            license_key TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    # === Indexes for published_tools ===
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_published_tools_agent ON published_tools(agent_id)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_published_tools_category ON published_tools(category)")

    # === User Interactions Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS user_interactions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            agent_id TEXT NOT NULL,
            chat_id TEXT NOT NULL,
            user_name TEXT,
            user_email TEXT,
            query TEXT NOT NULL,
            response TEXT,
            topic TEXT,
            intent TEXT,
            sentiment TEXT,
            session_id TEXT,
            is_complete INTEGER DEFAULT 1,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_user_interactions_agent_chat ON user_interactions(agent_id, chat_id)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_user_interactions_topic ON user_interactions(agent_id, topic)")

    # === User Profiles Table ===
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS user_profiles (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            agent_id TEXT NOT NULL,
            chat_id TEXT NOT NULL,
            user_name TEXT,
            user_email TEXT,
            total_messages INTEGER DEFAULT 0,
            first_seen TEXT,
            last_seen TEXT,
            last_topic TEXT,
            last_query TEXT,
            top_topics TEXT,
            interests TEXT,
            engagement_score REAL DEFAULT 0,
            email_opt_in INTEGER DEFAULT 0,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
            UNIQUE(agent_id, chat_id)
        )
    """)

    # === KB Sharing Tables ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS kb_sharing_config (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            agent_id TEXT NOT NULL,
            sharing_mode TEXT DEFAULT 'private',
            cost_per_query REAL DEFAULT 0,
            cost_currency TEXT DEFAULT 'DZIT',
            referrer_percent REAL DEFAULT 0,
            referrer_agent_id TEXT,
            description TEXT,
            total_queries_served INTEGER DEFAULT 0,
            total_earned REAL DEFAULT 0,
            is_active INTEGER DEFAULT 1,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
            updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
            UNIQUE(agent_id)
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS kb_access_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            requester_agent_id TEXT NOT NULL,
            provider_agent_id TEXT NOT NULL,
            query TEXT,
            cost_charged REAL DEFAULT 0,
            cost_currency TEXT DEFAULT 'DZIT',
            biznode_fee REAL DEFAULT 0,
            provider_earned REAL DEFAULT 0,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_kb_sharing_mode ON kb_sharing_config(sharing_mode, is_active)
    """)

    # === Distributed Task Engine Tables ===

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS task_queue (
            id TEXT PRIMARY KEY,
            task_type TEXT NOT NULL,
            payload TEXT,
            status TEXT DEFAULT 'queued',
            priority INTEGER DEFAULT 5,
            agent_id TEXT,
            user_id TEXT,
            worker_node TEXT,
            retries INTEGER DEFAULT 0,
            max_retries INTEGER DEFAULT 3,
            error_message TEXT,
            result TEXT,
            scheduled_at TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
            started_at TEXT,
            completed_at TEXT
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS task_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            task_id TEXT NOT NULL,
            log_message TEXT,
            status TEXT,
            worker_node TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS worker_nodes (
            node_id TEXT PRIMARY KEY,
            capabilities TEXT,
            status TEXT DEFAULT 'idle',
            current_task TEXT,
            tasks_completed INTEGER DEFAULT 0,
            tasks_failed INTEGER DEFAULT 0,
            last_heartbeat TEXT,
            registered_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_task_queue_status ON task_queue(status, priority DESC, created_at ASC)
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_task_queue_agent ON task_queue(agent_id, status)
    """)

    cursor.execute("""
        CREATE INDEX IF NOT EXISTS idx_task_logs_task ON task_logs(task_id)
    """)

    conn.commit()
    conn.close()
    print("Database initialized successfully.")


# === Notes CRUD ===

def create_note(node_id: str, title: str, content: str, summary: str = "", tags: str = "") -> int:
    """Create a new note."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO notes (node_id, title, content, summary, tags) VALUES (?, ?, ?, ?, ?)",
        (node_id, title, content, summary, tags)
    )
    note_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return note_id


def get_note(note_id: int) -> Optional[Dict]:
    """Get note by ID."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM notes WHERE id = ?", (note_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_all_notes() -> List[Dict]:
    """Get all notes."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM notes ORDER BY created_at DESC")
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def update_note(note_id: int, **kwargs) -> bool:
    """Update note fields."""
    conn = get_connection()
    cursor = conn.cursor()
    fields = ", ".join([f"{k} = ?" for k in kwargs.keys()])
    values = list(kwargs.values()) + [note_id]
    cursor.execute(f"UPDATE notes SET {fields}, updated_at = CURRENT_TIMESTAMP WHERE id = ?", values)
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


def delete_note(note_id: int) -> bool:
    """Delete a note."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM notes WHERE id = ?", (note_id,))
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


# === Note Links CRUD ===

def create_link(source_id: int, target_id: int, similarity: float) -> int:
    """Create a link between notes."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO note_links (source_id, target_id, similarity) VALUES (?, ?, ?)",
        (source_id, target_id, similarity)
    )
    link_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return link_id


def get_links_for_note(note_id: int) -> List[Dict]:
    """Get all links for a note."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("""
        SELECT nl.*, n.title, n.summary 
        FROM note_links nl
        JOIN notes n ON nl.target_id = n.id
        WHERE nl.source_id = ? OR nl.target_id = ?
    """, (note_id, note_id))
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


# === Business CRUD ===

def create_business(node_id: str, business_name: str, owner_telegram_id: str, 
                    owner_email: str = "", wallet_address: str = "") -> int:
    """Register a new business."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO businesses (node_id, business_name, owner_telegram_id, 
           owner_email, wallet_address) VALUES (?, ?, ?, ?, ?)""",
        (node_id, business_name, owner_telegram_id, owner_email, wallet_address)
    )
    business_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return business_id


def get_business(node_id: str) -> Optional[Dict]:
    """Get business by node_id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM businesses WHERE node_id = ?", (node_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_all_businesses() -> List[Dict]:
    """Get all businesses."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM businesses ORDER BY created_at DESC")
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def check_business_exists(business_name: str) -> Optional[Dict]:
    """Check if business already exists."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM businesses WHERE business_name = ? COLLATE NOCASE",
        (business_name,)
    )
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


# === Lead Management ===

def create_lead(name: str, business: str, contact_info: str, 
                summary: str = "", source: str = "telegram") -> int:
    """Create a new lead."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO leads (name, business, contact_info, summary, source) 
           VALUES (?, ?, ?, ?, ?)""",
        (name, business, contact_info, summary, source)
    )
    lead_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return lead_id


def get_lead(lead_id: int) -> Optional[Dict]:
    """Get lead by ID."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM leads WHERE id = ?", (lead_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_all_leads(status: str = None) -> List[Dict]:
    """Get all leads, optionally filtered by status."""
    conn = get_connection()
    cursor = conn.cursor()
    if status:
        cursor.execute("SELECT * FROM leads WHERE status = ? ORDER BY created_at DESC", (status,))
    else:
        cursor.execute("SELECT * FROM leads ORDER BY created_at DESC")
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def update_lead_status(lead_id: int, status: str) -> bool:
    """Update lead status."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("UPDATE leads SET status = ? WHERE id = ?", (status, lead_id))
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


# === Owner Actions ===

def create_action(action_type: str, data: Dict, risk_level: str = "low") -> int:
    """Create a new owner action request."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO owner_actions (action_type, data, risk_level) 
           VALUES (?, ?, ?)""",
        (action_type, json.dumps(data), risk_level)
    )
    action_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return action_id


def get_pending_actions() -> List[Dict]:
    """Get all pending owner actions."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM owner_actions WHERE status = 'pending' ORDER BY created_at"
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def resolve_action(action_id: int, response: str) -> bool:
    """Resolve an owner action (approve/reject)."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """UPDATE owner_actions 
           SET status = ?, owner_response = ?, resolved_at = CURRENT_TIMESTAMP 
           WHERE id = ?""",
        ("resolved", response, action_id)
    )
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


# === Associates ===

def create_associate(network_id: str, name: str, telegram_id: str, 
                    email: str, role: str, business_type: str = "") -> int:
    """Register a new associate."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO associates (network_id, name, telegram_id, email, role, business_type)
           VALUES (?, ?, ?, ?, ?, ?)""",
        (network_id, name, telegram_id, email, role, business_type)
    )
    associate_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return associate_id


def get_associate(associate_id: int) -> Optional[Dict]:
    """Get associate by ID."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM associates WHERE id = ?", (associate_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_associates_by_role(role: str) -> List[Dict]:
    """Get all associates with a specific role."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM associates WHERE role = ? AND status = 'active'",
        (role,)
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def get_all_associates() -> List[Dict]:
    """Get all active associates."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM associates WHERE status = 'active'")
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def log_network_interaction(associate_id: int, interaction_type: str, 
                            description: str, initiated_by: str) -> int:
    """Log an interaction with an associate."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO network_interactions 
           (associate_id, interaction_type, description, initiated_by)
           VALUES (?, ?, ?, ?)""",
        (associate_id, interaction_type, description, initiated_by)
    )
    interaction_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return interaction_id


# === Agent Identity ===

def save_agent_identity(identity: Dict) -> bool:
    """Save or update agent identity."""
    conn = get_connection()
    cursor = conn.cursor()
    
    # Check if exists
    cursor.execute("SELECT id FROM agent_identity WHERE id = 1")
    exists = cursor.fetchone()
    
    if exists:
        cursor.execute("""
            UPDATE agent_identity SET
                agent_name = ?, agent_email = ?, telegram_bot_token = ?,
                smtp_host = ?, smtp_port = ?, smtp_user = ?, smtp_password = ?,
                owner_telegram_id = ?, owner_email = ?, autonomy_level = ?,
                updated_at = CURRENT_TIMESTAMP
            WHERE id = 1
        """, (
            identity.get('agent_name'), identity.get('agent_email'),
            identity.get('telegram_bot_token'), identity.get('smtp_host'),
            identity.get('smtp_port'), identity.get('smtp_user'),
            identity.get('smtp_password'), identity.get('owner_telegram_id'),
            identity.get('owner_email'), identity.get('autonomy_level', 1)
        ))
    else:
        cursor.execute("""
            INSERT INTO agent_identity VALUES (
                1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
            )
        """, (
            identity.get('agent_name'), identity.get('agent_email'),
            identity.get('telegram_bot_token'), identity.get('smtp_host'),
            identity.get('smtp_port'), identity.get('smtp_user'),
            identity.get('smtp_password'), identity.get('owner_telegram_id'),
            identity.get('owner_email'), identity.get('autonomy_level', 1)
        ))
    
    conn.commit()
    conn.close()
    return True


def get_agent_identity() -> Optional[Dict]:
    """Get agent identity."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM agent_identity WHERE id = 1")
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_business_profile() -> Dict:
    """Return business profile fields from agent_identity."""
    identity = get_agent_identity() or {}
    return {
        "business_name":        identity.get("agent_name", ""),
        "business_type":        identity.get("business_type", ""),
        "business_description": identity.get("business_description", ""),
        "services_offered":     identity.get("services_offered", ""),
        "ai_persona_name":      identity.get("ai_persona_name", "BizNode Assistant"),
        "greeting_message":     identity.get("greeting_message", ""),
        "tone":                 identity.get("tone", "professional"),
        "faqs":                 identity.get("faqs", "[]"),
    }


def save_business_profile(profile: Dict) -> bool:
    """Save business profile fields into agent_identity row."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT id FROM agent_identity WHERE id = 1")
    exists = cursor.fetchone()
    fields = [
        "business_type", "business_description", "services_offered",
        "ai_persona_name", "greeting_message", "tone", "faqs",
    ]
    if exists:
        set_clause = ", ".join(f"{f} = ?" for f in fields)
        set_clause += ", agent_name = ?, updated_at = CURRENT_TIMESTAMP"
        cursor.execute(
            f"UPDATE agent_identity SET {set_clause} WHERE id = 1",
            [profile.get(f, "") for f in fields] + [profile.get("business_name", "")],
        )
    else:
        cursor.execute(
            """INSERT INTO agent_identity
               (id, agent_name, business_type, business_description,
                services_offered, ai_persona_name, greeting_message, tone, faqs)
               VALUES (1, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                profile.get("business_name", ""),
                profile.get("business_type", ""),
                profile.get("business_description", ""),
                profile.get("services_offered", ""),
                profile.get("ai_persona_name", "BizNode Assistant"),
                profile.get("greeting_message", ""),
                profile.get("tone", "professional"),
                profile.get("faqs", "[]"),
            ),
        )
    conn.commit()
    conn.close()
    return True


# === Sync Registry ===

def create_sync_record(node_id: str, hash_data: str, signature: str = "") -> int:
    """Create a sync record for 1bz network."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO sync_registry (node_id, hash_data, signature) VALUES (?, ?, ?)",
        (node_id, hash_data, signature)
    )
    record_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return record_id


def get_pending_syncs() -> List[Dict]:
    """Get all pending sync records."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM sync_registry WHERE sync_status = 'pending'"
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


def mark_synced(record_id: int) -> bool:
    """Mark a sync record as synced."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """UPDATE sync_registry 
           SET sync_status = 'synced', synced_at = CURRENT_TIMESTAMP 
           WHERE id = ?""",
        (record_id,)
    )
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


# === Utility Functions ===

def search_businesses(query: str) -> List[Dict]:
    """Search businesses by name."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM businesses WHERE business_name LIKE ?",
        (f"%{query}%",)
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(row) for row in rows]


# === Scheduled Tasks CRUD ===

def create_scheduled_task(id: str, name: str, type: str, action_json: str,
                          schedule_cron: str = None, interval_seconds: int = None,
                          owner_node_id: str = None) -> str:
    conn = get_connection()
    conn.execute(
        """INSERT INTO scheduled_tasks (id, name, type, action_json, schedule_cron,
           interval_seconds, owner_node_id) VALUES (?, ?, ?, ?, ?, ?, ?)""",
        (id, name, type, action_json, schedule_cron, interval_seconds, owner_node_id)
    )
    conn.commit()
    conn.close()
    return id


def get_scheduled_tasks(status: str = None) -> List[Dict]:
    conn = get_connection()
    if status:
        rows = conn.execute(
            "SELECT * FROM scheduled_tasks WHERE status = ? ORDER BY created_at DESC", (status,)
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM scheduled_tasks ORDER BY created_at DESC"
        ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_scheduled_task(task_id: str) -> Optional[Dict]:
    conn = get_connection()
    row = conn.execute(
        "SELECT * FROM scheduled_tasks WHERE id = ?", (task_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def update_scheduled_task(task_id: str, **fields) -> bool:
    if not fields:
        return False
    conn = get_connection()
    sets = ", ".join(f"{k} = ?" for k in fields)
    conn.execute(f"UPDATE scheduled_tasks SET {sets} WHERE id = ?",
                 list(fields.values()) + [task_id])
    conn.commit()
    success = conn.execute(
        "SELECT changes()"
    ).fetchone()[0] > 0
    conn.close()
    return success


def delete_scheduled_task(task_id: str) -> bool:
    conn = get_connection()
    conn.execute("DELETE FROM scheduled_tasks WHERE id = ?", (task_id,))
    conn.commit()
    conn.close()
    return True


def record_task_run(task_id: str, last_run: str, next_run: str = None) -> bool:
    conn = get_connection()
    conn.execute(
        """UPDATE scheduled_tasks
           SET last_run = ?, next_run = ?, run_count = run_count + 1
           WHERE id = ?""",
        (last_run, next_run, task_id)
    )
    conn.commit()
    conn.close()
    return True


# === Task Permissions CRUD ===

def create_task_permission(id: str, task_id: str, shared_with: str,
                           permission: str, mcp_key_id: str = None,
                           expires_at: str = None) -> str:
    conn = get_connection()
    conn.execute(
        """INSERT INTO task_permissions (id, task_id, shared_with, permission,
           mcp_key_id, expires_at) VALUES (?, ?, ?, ?, ?, ?)""",
        (id, task_id, shared_with, permission, mcp_key_id, expires_at)
    )
    conn.commit()
    conn.close()
    return id


def get_task_permissions(task_id: str) -> List[Dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM task_permissions WHERE task_id = ? ORDER BY created_at",
        (task_id,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def revoke_task_permission(perm_id: str) -> bool:
    conn = get_connection()
    conn.execute("DELETE FROM task_permissions WHERE id = ?", (perm_id,))
    conn.commit()
    conn.close()
    return True


# === MCP Keys CRUD ===

def create_mcp_key(id: str, permission: str, issued_to: str, token_b64: str,
                   task_id: str = None, node_id: str = None,
                   expires_at: str = None) -> str:
    conn = get_connection()
    conn.execute(
        """INSERT INTO mcp_keys (id, task_id, permission, issued_to, node_id,
           token_b64, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?)""",
        (id, task_id, permission, issued_to, node_id, token_b64, expires_at)
    )
    conn.commit()
    conn.close()
    return id


def get_mcp_key(key_id: str) -> Optional[Dict]:
    conn = get_connection()
    row = conn.execute(
        "SELECT * FROM mcp_keys WHERE id = ?", (key_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def get_all_mcp_keys(status: str = None) -> List[Dict]:
    conn = get_connection()
    if status:
        rows = conn.execute(
            "SELECT * FROM mcp_keys WHERE status = ? ORDER BY created_at DESC", (status,)
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM mcp_keys ORDER BY created_at DESC"
        ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def revoke_mcp_key(key_id: str) -> bool:
    conn = get_connection()
    conn.execute(
        "UPDATE mcp_keys SET status = 'revoked' WHERE id = ?", (key_id,)
    )
    conn.commit()
    conn.close()
    return True


def mark_mcp_key_used(key_id: str) -> bool:
    conn = get_connection()
    conn.execute(
        "UPDATE mcp_keys SET last_used = CURRENT_TIMESTAMP WHERE id = ?", (key_id,)
    )
    conn.commit()
    conn.close()
    return True


# === Trusted Nodes CRUD ===

def register_trusted_node(node_id: str, public_key_b64: str,
                          display_name: str = None, endpoint: str = None,
                          capabilities: str = None) -> str:
    conn = get_connection()
    conn.execute(
        """INSERT INTO trusted_nodes (node_id, display_name, public_key_b64,
           endpoint, capabilities)
           VALUES (?, ?, ?, ?, ?)
           ON CONFLICT(node_id) DO UPDATE SET
             display_name = excluded.display_name,
             public_key_b64 = excluded.public_key_b64,
             endpoint = excluded.endpoint,
             capabilities = excluded.capabilities,
             last_seen = CURRENT_TIMESTAMP""",
        (node_id, display_name, public_key_b64, endpoint, capabilities)
    )
    conn.commit()
    conn.close()
    return node_id


def get_trusted_nodes() -> List[Dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM trusted_nodes WHERE status = 'trusted' ORDER BY added_at DESC"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_trusted_node(node_id: str) -> Optional[Dict]:
    conn = get_connection()
    row = conn.execute(
        "SELECT * FROM trusted_nodes WHERE node_id = ?", (node_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def remove_trusted_node(node_id: str) -> bool:
    conn = get_connection()
    conn.execute("DELETE FROM trusted_nodes WHERE node_id = ?", (node_id,))
    conn.commit()
    conn.close()
    return True


def update_node_last_seen(node_id: str) -> bool:
    conn = get_connection()
    conn.execute(
        "UPDATE trusted_nodes SET last_seen = CURRENT_TIMESTAMP WHERE node_id = ?",
        (node_id,)
    )
    conn.commit()
    conn.close()
    return True


# === Backup Config CRUD ===

def upsert_backup_config(
    provider: str,
    enabled: int = 0,
    schedule_cron: str = "0 2 * * *",
    retention: int = 7,
    config_json: Optional[str] = None,
) -> str:
    conn = get_connection()
    conn.execute(
        """INSERT INTO backup_config (provider, enabled, schedule_cron, retention, config_json)
           VALUES (?, ?, ?, ?, ?)
           ON CONFLICT(provider) DO UPDATE SET
             enabled = excluded.enabled,
             schedule_cron = excluded.schedule_cron,
             retention = excluded.retention,
             config_json = COALESCE(excluded.config_json, backup_config.config_json)""",
        (provider, enabled, schedule_cron, retention, config_json),
    )
    conn.commit()
    conn.close()
    return provider


def get_backup_config(provider: str) -> Optional[Dict]:
    conn = get_connection()
    row = conn.execute(
        "SELECT * FROM backup_config WHERE provider = ?", (provider,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None


def get_all_backup_configs() -> List[Dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM backup_config ORDER BY provider"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_backup_last_run(provider: str, last_backup: str) -> bool:
    conn = get_connection()
    conn.execute(
        "UPDATE backup_config SET last_backup = ? WHERE provider = ?",
        (last_backup, provider),
    )
    conn.commit()
    conn.close()
    return True


def record_backup_history(
    provider: str,
    filename: str,
    size_bytes: int,
    status: str,
    error: Optional[str] = None,
) -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO backup_history (provider, filename, size_bytes, status, error)
           VALUES (?, ?, ?, ?, ?)""",
        (provider, filename, size_bytes, status, error),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_backup_history(limit: int = 50) -> List[Dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM backup_history ORDER BY created_at DESC LIMIT ?", (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_backup_stats() -> Dict:
    conn = get_connection()
    total = conn.execute("SELECT COUNT(*) FROM backup_history").fetchone()[0]
    success = conn.execute(
        "SELECT COUNT(*) FROM backup_history WHERE status = 'success'"
    ).fetchone()[0]
    last_success_row = conn.execute(
        "SELECT created_at, provider FROM backup_history WHERE status = 'success' ORDER BY created_at DESC LIMIT 1"
    ).fetchone()
    conn.close()
    return {
        "total_runs": total,
        "successful_runs": success,
        "failed_runs": total - success,
        "last_successful_backup": last_success_row["created_at"] if last_success_row else None,
        "last_successful_provider": last_success_row["provider"] if last_success_row else None,
    }


# === Marketing Hub CRUD ===

def log_query(
    telegram_id: str,
    username: str = "",
    query: str = "",
    response: str = "",
    source: str = "telegram",
) -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO query_logs (telegram_id, username, query, response, source) VALUES (?,?,?,?,?)",
        (telegram_id, username, query, response, source),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_query_logs(date: Optional[str] = None, limit: int = 100) -> List[Dict]:
    conn = get_connection()
    if date:
        rows = conn.execute(
            "SELECT * FROM query_logs WHERE DATE(created_at) = ? ORDER BY created_at DESC LIMIT ?",
            (date, limit),
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM query_logs ORDER BY created_at DESC LIMIT ?", (limit,)
        ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_query_log_stats() -> Dict:
    conn = get_connection()
    today = datetime.now().strftime("%Y-%m-%d")
    total_today = conn.execute(
        "SELECT COUNT(*) FROM query_logs WHERE DATE(created_at) = ?", (today,)
    ).fetchone()[0]
    unique_today = conn.execute(
        "SELECT COUNT(DISTINCT telegram_id) FROM query_logs WHERE DATE(created_at) = ?", (today,)
    ).fetchone()[0]
    total_all = conn.execute("SELECT COUNT(*) FROM query_logs").fetchone()[0]
    conn.close()
    return {
        "total_today":  total_today,
        "unique_today": unique_today,
        "total_all":    total_all,
        "date":         today,
    }


def block_customer(telegram_id: str, reason: str = "", blocked_by: str = "owner") -> bool:
    conn = get_connection()
    conn.execute(
        """INSERT INTO blocked_customers (telegram_id, reason, blocked_by)
           VALUES (?,?,?)
           ON CONFLICT(telegram_id) DO UPDATE SET reason=excluded.reason, blocked_by=excluded.blocked_by""",
        (str(telegram_id), reason, blocked_by),
    )
    conn.commit()
    conn.close()
    return True


def unblock_customer(telegram_id: str) -> bool:
    conn = get_connection()
    conn.execute("DELETE FROM blocked_customers WHERE telegram_id = ?", (str(telegram_id),))
    conn.commit()
    conn.close()
    return True


def get_blocked_customers() -> List[Dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM blocked_customers ORDER BY created_at DESC"
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def is_customer_blocked(telegram_id: str) -> bool:
    conn = get_connection()
    row = conn.execute(
        "SELECT 1 FROM blocked_customers WHERE telegram_id = ?", (str(telegram_id),)
    ).fetchone()
    conn.close()
    return row is not None


def log_email_instruction(
    raw_email: str = "",
    parsed_action: str = "",
    status: str = "pending",
    result: str = "",
) -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO email_instructions (raw_email, parsed_action, status, result) VALUES (?,?,?,?)",
        (raw_email, parsed_action, status, result),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_email_instructions(limit: int = 50) -> List[Dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM email_instructions ORDER BY created_at DESC LIMIT ?", (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


# === Email Log CRUD ===

def log_email_sent(
    to_addr: str,
    subject: str,
    body_preview: str = "",
    email_type: str = "general",
    status: str = "ok",
    error_msg: str = "",
    from_addr: str = "",
) -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO email_log
           (direction, email_type, from_addr, to_addr, subject, body_preview, status, error_msg)
           VALUES (?,?,?,?,?,?,?,?)""",
        ("sent", email_type, from_addr, to_addr, subject, body_preview[:500], status, error_msg),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def log_email_received(
    from_addr: str,
    subject: str,
    body_preview: str = "",
    email_type: str = "incoming",
) -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO email_log
           (direction, email_type, from_addr, to_addr, subject, body_preview, status)
           VALUES (?,?,?,?,?,?,?)""",
        ("received", email_type, from_addr, "", subject, body_preview[:500], "ok"),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_email_log(direction: str = None, limit: int = 100, date: str = None) -> List[Dict]:
    conn = get_connection()
    conditions = []
    params: list = []
    if direction:
        conditions.append("direction = ?")
        params.append(direction)
    if date:
        conditions.append("DATE(created_at) = ?")
        params.append(date)
    where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
    params.append(limit)
    rows = conn.execute(
        f"SELECT * FROM email_log {where} ORDER BY created_at DESC LIMIT ?", params
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_email_log_stats() -> dict:
    conn = get_connection()
    today = __import__("datetime").date.today().isoformat()
    cur = conn.cursor()
    cur.execute(
        """SELECT
             SUM(CASE WHEN direction='sent'     AND DATE(created_at)=? THEN 1 ELSE 0 END) AS sent_today,
             SUM(CASE WHEN direction='received' AND DATE(created_at)=? THEN 1 ELSE 0 END) AS received_today,
             SUM(CASE WHEN direction='sent'     THEN 1 ELSE 0 END) AS sent_total,
             SUM(CASE WHEN direction='received' THEN 1 ELSE 0 END) AS received_total
           FROM email_log""",
        (today, today),
    )
    row = cur.fetchone() or {}
    conn.close()
    return {
        "sent_today":     int(row.get("sent_today") or 0),
        "received_today": int(row.get("received_today") or 0),
        "sent_total":     int(row.get("sent_total") or 0),
        "received_total": int(row.get("received_total") or 0),
        "date": today,
    }


# === Conversation History CRUD ===

def add_conversation_message(telegram_id: str, role: str, message: str, max_history: int = 20):
    """Append a message to a customer's conversation history and trim oldest beyond max_history."""
    conn = get_connection()
    cur = conn.cursor()
    cur.execute(
        "INSERT INTO conversation_history (telegram_id, role, message) VALUES (?, ?, ?)",
        (str(telegram_id), role, message),
    )
    # Keep only the most recent max_history messages per user
    cur.execute(
        """
        DELETE FROM conversation_history
        WHERE telegram_id = ? AND id NOT IN (
            SELECT id FROM conversation_history
            WHERE telegram_id = ?
            ORDER BY id DESC
            LIMIT ?
        )
        """,
        (str(telegram_id), str(telegram_id), max_history),
    )
    conn.commit()
    conn.close()


def get_conversation_history(telegram_id: str, limit: int = 10) -> list:
    """Return the last `limit` messages for a customer, ordered oldest-first."""
    conn = get_connection()
    rows = conn.execute(
        """
        SELECT role, message, created_at FROM (
            SELECT role, message, created_at, id
            FROM conversation_history
            WHERE telegram_id = ?
            ORDER BY id DESC
            LIMIT ?
        ) AS subq ORDER BY subq.id ASC
        """,
        (str(telegram_id), limit),
    ).fetchall()
    conn.close()
    return [{"role": r["role"], "message": r["message"], "created_at": str(r["created_at"])} for r in rows]


def clear_conversation_history(telegram_id: str):
    """Delete all conversation history for a customer."""
    conn = get_connection()
    conn.execute(
        "DELETE FROM conversation_history WHERE telegram_id = ?",
        (str(telegram_id),),
    )
    conn.commit()
    conn.close()


# === Menu Items CRUD ===

def create_menu_item(menu_name: str, menu_type: str, link: str = None,
                     task_code: str = None, sort_order: int = 0) -> int:
    """Create a new menu item. Returns the new row id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO menu_items (menu_name, menu_type, link, task_code, sort_order) VALUES (?, ?, ?, ?, ?)",
        (menu_name, menu_type, link, task_code, sort_order),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_menu_items(active_only: bool = True) -> list:
    """Return all menu items ordered by sort_order."""
    conn = get_connection()
    cursor = conn.cursor()
    if active_only:
        cursor.execute("SELECT * FROM menu_items WHERE is_active = 1 ORDER BY sort_order ASC, id ASC")
    else:
        cursor.execute("SELECT * FROM menu_items ORDER BY sort_order ASC, id ASC")
    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows


def get_menu_item(item_id: int) -> dict:
    """Return a single menu item or None."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM menu_items WHERE id = ?", (item_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def update_menu_item(item_id: int, **fields) -> bool:
    """Update one or more fields of a menu item."""
    if not fields:
        return False
    allowed = {"menu_name", "menu_type", "link", "task_code", "sort_order", "is_active"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return False
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [item_id]
    conn = get_connection()
    conn.execute(f"UPDATE menu_items SET {cols} WHERE id = ?", vals)
    conn.commit()
    conn.close()
    return True


def delete_menu_item(item_id: int) -> bool:
    """Delete a menu item. Returns True if a row was deleted."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM menu_items WHERE id = ?", (item_id,))
    deleted = cursor.rowcount > 0
    conn.commit()
    conn.close()
    return deleted


# === Workflows CRUD ===

def create_workflow(workflow_code: str, description: str = None,
                    trigger_command: str = None) -> int:
    """Create a new workflow. Returns the new row id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO workflows (workflow_code, description, trigger_command) VALUES (?, ?, ?)",
        (workflow_code.strip(), description, trigger_command),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_workflows() -> list:
    """Return all workflows."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM workflows ORDER BY id ASC")
    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows


def get_workflow_by_command(command_text: str) -> dict:
    """Find an active workflow by trigger_command (case-insensitive). Returns dict or None."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM workflows WHERE LOWER(trigger_command) = ? AND is_active = 1",
        (command_text.lower().strip(),),
    )
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def update_workflow(workflow_code: str, **fields) -> bool:
    """Update workflow fields."""
    if not fields:
        return False
    allowed = {"description", "trigger_command", "is_active"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return False
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [workflow_code]
    conn = get_connection()
    conn.execute(f"UPDATE workflows SET {cols} WHERE workflow_code = ?", vals)
    conn.commit()
    conn.close()
    return True


def delete_workflow(workflow_code: str):
    """Delete a workflow and all its steps."""
    conn = get_connection()
    conn.execute("DELETE FROM workflow_steps WHERE workflow_code = ?", (workflow_code,))
    conn.execute("DELETE FROM workflows WHERE workflow_code = ?", (workflow_code,))
    conn.commit()
    conn.close()


# === Workflow Steps CRUD ===

def create_workflow_step(workflow_code: str, step_order: int, step_type: str,
                         step_prompt: str = None, field_name: str = None) -> int:
    """Add a step to a workflow. Returns new step id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO workflow_steps (workflow_code, step_order, step_type, step_prompt, field_name) VALUES (?, ?, ?, ?, ?)",
        (workflow_code, step_order, step_type, step_prompt, field_name),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_workflow_steps(workflow_code: str) -> list:
    """Return all steps for a workflow ordered by step_order."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM workflow_steps WHERE workflow_code = ? ORDER BY step_order ASC",
        (workflow_code,),
    )
    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows


def update_workflow_step(step_id: int, **fields) -> bool:
    """Update workflow step fields."""
    if not fields:
        return False
    allowed = {"step_order", "step_type", "step_prompt", "field_name"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return False
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [step_id]
    conn = get_connection()
    conn.execute(f"UPDATE workflow_steps SET {cols} WHERE id = ?", vals)
    conn.commit()
    conn.close()
    return True


def delete_workflow_step(step_id: int) -> bool:
    """Delete a single workflow step."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM workflow_steps WHERE id = ?", (step_id,))
    deleted = cursor.rowcount > 0
    conn.commit()
    conn.close()
    return deleted


def delete_all_workflow_steps(workflow_code: str):
    """Delete all steps for a workflow."""
    conn = get_connection()
    conn.execute("DELETE FROM workflow_steps WHERE workflow_code = ?", (workflow_code,))
    conn.commit()
    conn.close()


# === User Workflow State CRUD ===

def get_active_workflow_state(user_id: str) -> dict:
    """Return the current active workflow state for a user, or None."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM user_workflow_state WHERE user_id = ? AND status = 'active' ORDER BY id DESC LIMIT 1",
        (str(user_id),),
    )
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def start_workflow_state(user_id: str, workflow_code: str):
    """Expire any existing active state for the user, then create a fresh one."""
    conn = get_connection()
    conn.execute(
        "UPDATE user_workflow_state SET status = 'expired' WHERE user_id = ? AND status = 'active'",
        (str(user_id),),
    )
    conn.execute(
        "INSERT INTO user_workflow_state (user_id, workflow_code, current_step, collected_data, status) VALUES (?, ?, 1, '{}', 'active')",
        (str(user_id), workflow_code),
    )
    conn.commit()
    conn.close()


def update_workflow_state(user_id: str, current_step: int = None,
                          collected_data: str = None, status: str = None):
    """Update fields on the active workflow state row."""
    parts, vals = [], []
    if current_step is not None:
        parts.append("current_step = ?"); vals.append(current_step)
    if collected_data is not None:
        parts.append("collected_data = ?"); vals.append(collected_data)
    if status is not None:
        parts.append("status = ?"); vals.append(status)
    parts.append("updated_at = CURRENT_TIMESTAMP")
    if not vals:
        return
    vals.append(str(user_id))
    sql = f"UPDATE user_workflow_state SET {', '.join(parts)} WHERE user_id = ? AND status = 'active'"
    conn = get_connection()
    conn.execute(sql, vals)
    conn.commit()
    conn.close()


def clear_workflow_state(user_id: str):
    """Mark any active workflow state as expired for a user."""
    conn = get_connection()
    conn.execute(
        "UPDATE user_workflow_state SET status = 'expired' WHERE user_id = ? AND status = 'active'",
        (str(user_id),),
    )
    conn.commit()
    conn.close()


# === OTP CRUD ===

def store_otp(user_id: str, email: str, otp_hash: str, expires_at: str):
    """Delete any existing OTPs for the user and insert a fresh one."""
    conn = get_connection()
    conn.execute("DELETE FROM otp_codes WHERE user_id = ?", (str(user_id),))
    conn.execute(
        "INSERT INTO otp_codes (user_id, email, otp_hash, expires_at) VALUES (?, ?, ?, ?)",
        (str(user_id), email, otp_hash, expires_at),
    )
    conn.commit()
    conn.close()


def get_active_otp(user_id: str) -> dict:
    """Return the current unexpired, unverified OTP row for a user, or None."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM otp_codes WHERE user_id = ? AND verified = 0 AND expires_at > NOW() ORDER BY id DESC LIMIT 1",
        (str(user_id),),
    )
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def increment_otp_attempts(user_id: str):
    """Increment attempt counter for the user's latest OTP."""
    conn = get_connection()
    conn.execute(
        "UPDATE otp_codes SET attempts = attempts + 1 WHERE user_id = ? AND verified = 0 AND id = (SELECT id FROM otp_codes WHERE user_id = ? AND verified = 0 ORDER BY id DESC LIMIT 1)",
        (str(user_id), str(user_id)),
    )
    conn.commit()
    conn.close()


def mark_otp_verified(user_id: str):
    """Mark the user's OTP as verified."""
    conn = get_connection()
    conn.execute(
        "UPDATE otp_codes SET verified = 1 WHERE user_id = ? AND verified = 0",
        (str(user_id),),
    )
    conn.commit()
    conn.close()


def clear_otp(user_id: str):
    """Delete all OTP records for a user."""
    conn = get_connection()
    conn.execute("DELETE FROM otp_codes WHERE user_id = ?", (str(user_id),))
    conn.commit()
    conn.close()


# === Payment Agent: Services ===

def create_service(name: str, description: str = "", price: float = 0.0,
                   currency: str = "USD", duration: str = "", sort_order: int = 0) -> int:
    """Create a service in the catalog."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO services (name, description, price, currency, duration, sort_order) VALUES (?, ?, ?, ?, ?, ?)",
        (name, description, price, currency, duration, sort_order),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_services(active_only: bool = True) -> List[Dict]:
    """Return all services ordered by sort_order."""
    conn = get_connection()
    cursor = conn.cursor()
    if active_only:
        cursor.execute("SELECT * FROM services WHERE is_active = 1 ORDER BY sort_order ASC, id ASC")
    else:
        cursor.execute("SELECT * FROM services ORDER BY sort_order ASC, id ASC")
    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows


def get_service(service_id: int) -> Optional[Dict]:
    """Return a single service or None."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM services WHERE id = ?", (service_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def update_service(service_id: int, **fields) -> bool:
    """Update service fields."""
    allowed = {"name", "description", "price", "currency", "duration", "sort_order", "is_active"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return False
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [service_id]
    conn = get_connection()
    conn.execute(f"UPDATE services SET {cols} WHERE id = ?", vals)
    conn.commit()
    conn.close()
    return True


def delete_service(service_id: int) -> bool:
    """Delete a service."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM services WHERE id = ?", (service_id,))
    deleted = cursor.rowcount > 0
    conn.commit()
    conn.close()
    return deleted


# === Payment Agent: Payment Accounts ===

def create_payment_account(provider_key: str, display_name: str,
                           config_json: str = "{}", regions: str = "*",
                           sort_order: int = 0) -> int:
    """Register an owner payment account for a provider."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO payment_accounts (provider_key, display_name, config_json, regions, sort_order) VALUES (?, ?, ?, ?, ?)",
        (provider_key, display_name, config_json, regions, sort_order),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_payment_accounts(active_only: bool = True) -> List[Dict]:
    """Return all payment accounts."""
    conn = get_connection()
    cursor = conn.cursor()
    if active_only:
        cursor.execute(
            "SELECT * FROM payment_accounts WHERE is_active = 1 ORDER BY sort_order ASC, id ASC"
        )
    else:
        cursor.execute("SELECT * FROM payment_accounts ORDER BY sort_order ASC, id ASC")
    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows


def get_payment_account(account_id: int) -> Optional[Dict]:
    """Return a single payment account."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM payment_accounts WHERE id = ?", (account_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_payment_accounts_for_region(region_code: str) -> List[Dict]:
    """Return active payment accounts whose regions include region_code or '*'."""
    accounts = get_payment_accounts(active_only=True)
    result = []
    for acc in accounts:
        regions = acc.get("regions", "*")
        if regions == "*":
            result.append(acc)
        else:
            try:
                region_list = json.loads(regions) if regions.startswith("[") else [r.strip() for r in regions.split(",")]
                if region_code.upper() in [r.upper() for r in region_list]:
                    result.append(acc)
            except Exception:
                result.append(acc)
    return result


def update_payment_account(account_id: int, **fields) -> bool:
    """Update payment account fields."""
    allowed = {"display_name", "config_json", "regions", "is_active", "sort_order"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return False
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [account_id]
    conn = get_connection()
    conn.execute(f"UPDATE payment_accounts SET {cols} WHERE id = ?", vals)
    conn.commit()
    conn.close()
    return True


def delete_payment_account(account_id: int) -> bool:
    """Delete a payment account."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM payment_accounts WHERE id = ?", (account_id,))
    deleted = cursor.rowcount > 0
    conn.commit()
    conn.close()
    return deleted


# === Payment Agent: Payment Requests ===

def create_payment_request(reference_no: str, telegram_id: str, customer_name: str,
                           service_id: int, service_name: str, amount: float,
                           currency: str, provider_key: str, payment_link: str) -> int:
    """Create a new payment request record."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO payment_requests
           (reference_no, telegram_id, customer_name, service_id, service_name,
            amount, currency, provider_key, payment_link)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (reference_no, str(telegram_id), customer_name, service_id, service_name,
         amount, currency, provider_key, payment_link),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_payment_request(reference_no: str) -> Optional[Dict]:
    """Return a payment request by reference number."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM payment_requests WHERE reference_no = ?", (reference_no,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_payment_requests(status: str = None, limit: int = 100) -> List[Dict]:
    """Return payment requests, optionally filtered by status."""
    conn = get_connection()
    cursor = conn.cursor()
    if status:
        cursor.execute(
            "SELECT * FROM payment_requests WHERE status = ? ORDER BY created_at DESC LIMIT ?",
            (status, limit),
        )
    else:
        cursor.execute(
            "SELECT * FROM payment_requests ORDER BY created_at DESC LIMIT ?", (limit,)
        )
    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows


def get_payment_requests_for_customer(telegram_id: str) -> List[Dict]:
    """Return all payment requests for a customer."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM payment_requests WHERE telegram_id = ? ORDER BY created_at DESC",
        (str(telegram_id),),
    )
    rows = [dict(r) for r in cursor.fetchall()]
    conn.close()
    return rows


def update_payment_request_status(reference_no: str, status: str,
                                  customer_utr: str = None, owner_notes: str = None) -> bool:
    """Update payment request status and optional fields."""
    parts = ["status = ?"]
    vals = [status]
    if customer_utr is not None:
        parts.append("customer_utr = ?")
        vals.append(customer_utr)
    if owner_notes is not None:
        parts.append("owner_notes = ?")
        vals.append(owner_notes)
    if status in ("owner_confirmed", "paid"):
        parts.append("confirmed_at = CURRENT_TIMESTAMP")
    vals.append(reference_no)
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        f"UPDATE payment_requests SET {', '.join(parts)} WHERE reference_no = ?", vals
    )
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


def generate_reference_no() -> str:
    """Generate a unique payment reference number like BN-20260307-0042."""
    from datetime import datetime as _dt
    date_str = _dt.now().strftime("%Y%m%d")
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT COUNT(*) FROM payment_requests WHERE reference_no LIKE ?",
        (f"BN-{date_str}-%",),
    )
    row = cursor.fetchone()
    count = (list(row.values())[0] if isinstance(row, dict) else row[0]) + 1
    conn.close()
    return f"BN-{date_str}-{count:04d}"


# === Client Files CRUD ===

def create_client_file(
    name: str,
    email: str = "",
    phone: str = "",
    company: str = "",
    website: str = "",
    address: str = "",
    telegram_id: str = "",
    notes: str = "",
    tags: str = "",
    status: str = "active",
    source: str = "manual",
    contract_value: float = None,
) -> int:
    """Create a new client file. Returns the new row id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO client_files
           (name, email, phone, company, website, address, telegram_id,
            notes, tags, status, source, contract_value)
           VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
        (name, email, phone, company, website, address, telegram_id,
         notes, tags, status, source, contract_value),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_client_file(client_id: int) -> Optional[Dict]:
    """Get a single client file by id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM client_files WHERE id = ?", (client_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_all_client_files(status: str = None) -> List[Dict]:
    """Return all client files, optionally filtered by status."""
    conn = get_connection()
    if status:
        rows = conn.execute(
            "SELECT * FROM client_files WHERE status = ? ORDER BY created_at DESC", (status,)
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT * FROM client_files ORDER BY created_at DESC"
        ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_client_file(client_id: int, **fields) -> bool:
    """Update one or more fields of a client file."""
    allowed = {
        "name", "email", "phone", "company", "website", "address",
        "telegram_id", "notes", "tags", "status", "contract_value",
    }
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return False
    cols = ", ".join(f"{k} = ?" for k in updates)
    cols += ", updated_at = CURRENT_TIMESTAMP"
    vals = list(updates.values()) + [client_id]
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(f"UPDATE client_files SET {cols} WHERE id = ?", vals)
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


def delete_client_file(client_id: int) -> bool:
    """Delete a client file."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM client_files WHERE id = ?", (client_id,))
    conn.commit()
    success = cursor.rowcount > 0
    conn.close()
    return success


def search_client_files(query: str) -> List[Dict]:
    """Search client files by name, email, company, or telegram_id."""
    like = f"%{query}%"
    conn = get_connection()
    rows = conn.execute(
        """SELECT * FROM client_files
           WHERE name LIKE ? OR email LIKE ? OR company LIKE ? OR telegram_id LIKE ?
           ORDER BY created_at DESC""",
        (like, like, like, like),
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


# === Bot Users CRUD ===

def get_bot_user(telegram_id: str) -> Optional[Dict]:
    """Get bot user state by telegram_id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM bot_users WHERE telegram_id = ?", (str(telegram_id),))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def upsert_bot_user(telegram_id: str, **fields) -> None:
    """Insert or update a bot user's fields."""
    allowed = {"username", "name", "email", "business", "step", "verified", "lead_id",
                "person_type", "crypto_wallet", "wallet_verified"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT telegram_id FROM bot_users WHERE telegram_id = ?", (str(telegram_id),))
    exists = cursor.fetchone()
    if exists:
        if updates:
            cols = ", ".join(f"{k} = ?" for k in updates)
            cols += ", last_seen = CURRENT_TIMESTAMP"
            vals = list(updates.values()) + [str(telegram_id)]
            cursor.execute(f"UPDATE bot_users SET {cols} WHERE telegram_id = ?", vals)
    else:
        updates["telegram_id"] = str(telegram_id)
        cols = ", ".join(updates.keys())
        placeholders = ", ".join("?" for _ in updates)
        cursor.execute(
            f"INSERT INTO bot_users ({cols}) VALUES ({placeholders})",
            list(updates.values()),
        )
    conn.commit()
    conn.close()


def get_lead_by_telegram_id(telegram_id: str) -> Optional[Dict]:
    """Find a lead whose contact_info contains the given telegram_id."""
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM leads WHERE contact_info LIKE ? ORDER BY id DESC LIMIT 1",
        (f"%{telegram_id}%",),
    ).fetchall()
    conn.close()
    return dict(rows[0]) if rows else None


# === Transactions CRUD ===

def create_transaction(reference_no: str, telegram_id: str, person_type: str,
                       transaction_type: str, title: str, description: str = "",
                       amount: float = 0, currency: str = "USD",
                       payment_method: str = "", metadata_json: str = "{}") -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO transactions (reference_no, telegram_id, person_type,
           transaction_type, title, description, amount, currency, payment_method, metadata_json)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (reference_no, str(telegram_id), person_type, transaction_type,
         title, description, amount, currency, payment_method, metadata_json),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_transaction(reference_no: str) -> Optional[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM transactions WHERE reference_no = ?", (reference_no,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_transactions_by_user(telegram_id: str, limit: int = 20) -> List[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM transactions WHERE telegram_id = ? ORDER BY created_at DESC LIMIT ?",
        (str(telegram_id), limit),
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_all_transactions(status: str = None, limit: int = 100) -> List[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    if status:
        cursor.execute(
            "SELECT * FROM transactions WHERE status = ? ORDER BY created_at DESC LIMIT ?",
            (status, limit),
        )
    else:
        cursor.execute("SELECT * FROM transactions ORDER BY created_at DESC LIMIT ?", (limit,))
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_transaction_status(reference_no: str, status: str, payment_ref: str = None,
                               completed_at: str = None) -> None:
    conn = get_connection()
    cursor = conn.cursor()
    parts = ["status = ?", "updated_at = CURRENT_TIMESTAMP"]
    vals = [status]
    if payment_ref:
        parts.append("payment_ref = ?")
        vals.append(payment_ref)
    if completed_at:
        parts.append("completed_at = ?")
        vals.append(completed_at)
    vals.append(reference_no)
    cursor.execute(f"UPDATE transactions SET {', '.join(parts)} WHERE reference_no = ?", vals)
    conn.commit()
    conn.close()


# === Affiliate Links CRUD ===

def create_affiliate_link(title: str, url: str, description: str = "",
                          category: str = "general", commission: str = "") -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO affiliate_links (title, url, description, category, commission) VALUES (?, ?, ?, ?, ?)",
        (title, url, description, category, commission),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_affiliate_links(active_only: bool = True) -> List[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    if active_only:
        cursor.execute("SELECT * FROM affiliate_links WHERE is_active = 1 ORDER BY sort_order, id")
    else:
        cursor.execute("SELECT * FROM affiliate_links ORDER BY id")
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_affiliate_link(link_id: int, **fields) -> None:
    allowed = {"title", "url", "description", "category", "commission", "is_active"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return
    conn = get_connection()
    cursor = conn.cursor()
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [link_id]
    cursor.execute(f"UPDATE affiliate_links SET {cols} WHERE id = ?", vals)
    conn.commit()
    conn.close()


def delete_affiliate_link(link_id: int) -> None:
    conn = get_connection()
    conn.execute("DELETE FROM affiliate_links WHERE id = ?", (link_id,))
    conn.commit()
    conn.close()


def increment_affiliate_click(link_id: int) -> None:
    conn = get_connection()
    conn.execute("UPDATE affiliate_links SET click_count = click_count + 1 WHERE id = ?", (link_id,))
    conn.commit()
    conn.close()


# === Engagement Log CRUD ===

def log_engagement(telegram_id: str, email: str, person_type: str,
                   engagement_type: str, subject: str = "", status: str = "sent") -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO engagement_log (telegram_id, email, person_type, engagement_type, subject, status) VALUES (?, ?, ?, ?, ?, ?)",
        (str(telegram_id), email, person_type, engagement_type, subject, status),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_engagement_log(telegram_id: str = None, limit: int = 50) -> List[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    if telegram_id:
        cursor.execute(
            "SELECT * FROM engagement_log WHERE telegram_id = ? ORDER BY created_at DESC LIMIT ?",
            (str(telegram_id), limit),
        )
    else:
        cursor.execute("SELECT * FROM engagement_log ORDER BY created_at DESC LIMIT ?", (limit,))
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_last_engagement(telegram_id: str, engagement_type: str) -> Optional[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM engagement_log WHERE telegram_id = ? AND engagement_type = ? ORDER BY created_at DESC LIMIT 1",
        (str(telegram_id), engagement_type),
    )
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


# === Group Ads CRUD ===

def create_group_ad(title: str, message: str, target_group: str = "",
                    ad_type: str = "promotion", schedule_cron: str = "") -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO group_ads (title, message, target_group, ad_type, schedule_cron) VALUES (?, ?, ?, ?, ?)",
        (title, message, target_group, ad_type, schedule_cron),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_group_ads(active_only: bool = True) -> List[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    if active_only:
        cursor.execute("SELECT * FROM group_ads WHERE is_active = 1 ORDER BY id")
    else:
        cursor.execute("SELECT * FROM group_ads ORDER BY id")
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_group_ad(ad_id: int, **fields) -> None:
    allowed = {"title", "message", "target_group", "ad_type", "schedule_cron", "is_active", "last_sent", "send_count"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return
    conn = get_connection()
    cursor = conn.cursor()
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [ad_id]
    cursor.execute(f"UPDATE group_ads SET {cols} WHERE id = ?", vals)
    conn.commit()
    conn.close()


def delete_group_ad(ad_id: int) -> None:
    conn = get_connection()
    conn.execute("DELETE FROM group_ads WHERE id = ?", (ad_id,))
    conn.commit()
    conn.close()


# === All Verified Bot Users (for engagement emails) ===

def get_all_verified_bot_users() -> List[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM bot_users WHERE verified = 1 ORDER BY last_seen DESC")
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


# === License Purchases CRUD ===

def create_license_purchase(reference_no: str, email: str, license_type: str = "basic",
                            telegram_id: str = "", customer_name: str = "",
                            payment_method: str = "stripe", amount: float = 0,
                            currency: str = "USD", stripe_session_id: str = "",
                            license_key: str = "") -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO license_purchases
           (reference_no, telegram_id, email, customer_name, license_type,
            license_key, stripe_session_id, payment_method, amount, currency, status)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending')""",
        (reference_no, str(telegram_id), email, customer_name, license_type,
         license_key, stripe_session_id, payment_method, amount, currency),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_license_purchase(reference_no: str) -> Optional[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM license_purchases WHERE reference_no = ?", (reference_no,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_license_purchase_by_key(license_key: str) -> Optional[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM license_purchases WHERE license_key = ?", (license_key,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_license_purchases(status: str = None, limit: int = 100) -> List[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    if status:
        cursor.execute(
            "SELECT * FROM license_purchases WHERE status = ? ORDER BY created_at DESC LIMIT ?",
            (status, limit),
        )
    else:
        cursor.execute("SELECT * FROM license_purchases ORDER BY created_at DESC LIMIT ?", (limit,))
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_license_purchase(reference_no: str, **fields) -> None:
    allowed = {"status", "license_key", "stripe_payment_intent", "transaction_hash",
               "usb_serial", "machine_fingerprint", "activated_at"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return
    updates["updated_at"] = datetime.now().isoformat()
    conn = get_connection()
    cursor = conn.cursor()
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [reference_no]
    cursor.execute(f"UPDATE license_purchases SET {cols} WHERE reference_no = ?", vals)
    conn.commit()
    conn.close()


# === Crypto Payment OTP CRUD ===

def create_crypto_payment_otp(reference_no: str, email: str, otp_hash: str,
                               expires_at: str, telegram_id: str = "",
                               amount: float = 0, currency: str = "USD",
                               license_type: str = "basic") -> int:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO crypto_payment_otp
           (reference_no, telegram_id, email, otp_hash, amount, currency,
            license_type, expires_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
        (reference_no, str(telegram_id), email, otp_hash, amount, currency,
         license_type, expires_at),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_active_crypto_otp(reference_no: str) -> Optional[Dict]:
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT * FROM crypto_payment_otp
           WHERE reference_no = ? AND status = 'pending'
           ORDER BY created_at DESC LIMIT 1""",
        (reference_no,),
    )
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def update_crypto_otp(otp_id: int, **fields) -> None:
    allowed = {"status", "transaction_hash", "verified_at", "attempts"}
    updates = {k: v for k, v in fields.items() if k in allowed}
    if not updates:
        return
    conn = get_connection()
    cursor = conn.cursor()
    cols = ", ".join(f"{k} = ?" for k in updates)
    vals = list(updates.values()) + [otp_id]
    cursor.execute(f"UPDATE crypto_payment_otp SET {cols} WHERE id = ?", vals)
    conn.commit()
    conn.close()


# === Published Tools CRUD ===

def create_published_tool(tool_id: str, agent_id: str, node_id: str, name: str,
                          description: str = "", category: str = "general",
                          input_schema: str = "", output_schema: str = "",
                          cost_amount: float = 0, cost_currency: str = "DZIT",
                          access_level: str = "public", handler_url: str = "",
                          version: str = "1.0.0", manifest_hash: str = "",
                          signature: str = "") -> int:
    """Create a published tool entry."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()
    cursor.execute(
        """INSERT INTO published_tools
           (tool_id, agent_id, node_id, name, description, category,
            input_schema, output_schema, cost_amount, cost_currency,
            access_level, handler_url, version, manifest_hash, signature,
            published_at, updated_at)
           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (tool_id, agent_id, node_id, name, description, category,
         input_schema, output_schema, cost_amount, cost_currency,
         access_level, handler_url, version, manifest_hash, signature,
         now, now),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_published_tool(tool_id: str) -> Optional[Dict]:
    """Get published tool by tool_id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM published_tools WHERE tool_id = ?", (tool_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_published_tools_by_agent(agent_id: str) -> List[Dict]:
    """Get all published tools for an agent."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM published_tools WHERE agent_id = ? ORDER BY published_at DESC",
        (agent_id,),
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def list_published_tools(category: str = None, active_only: bool = True) -> List[Dict]:
    """List all published tools, optionally filtered."""
    conn = get_connection()
    cursor = conn.cursor()
    query = "SELECT * FROM published_tools WHERE 1=1"
    params: list = []
    if active_only:
        query += " AND is_active = 1"
    if category:
        query += " AND category = ?"
        params.append(category)
    query += " ORDER BY call_count DESC, published_at DESC"
    cursor.execute(query, params)
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_published_tool(tool_id: str, **kwargs) -> bool:
    """Update published tool fields."""
    kwargs["updated_at"] = datetime.utcnow().isoformat()
    conn = get_connection()
    cursor = conn.cursor()
    fields = ", ".join(f"{k} = ?" for k in kwargs)
    values = list(kwargs.values()) + [tool_id]
    cursor.execute(f"UPDATE published_tools SET {fields} WHERE tool_id = ?", values)
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


def delete_published_tool(tool_id: str) -> bool:
    """Delete a published tool."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM published_tools WHERE tool_id = ?", (tool_id,))
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


def increment_tool_call(tool_id: str, earned: float = 0) -> bool:
    """Increment call count and add to total earned."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """UPDATE published_tools
           SET call_count = call_count + 1,
               total_earned = total_earned + ?,
               updated_at = ?
           WHERE tool_id = ?""",
        (earned, datetime.utcnow().isoformat(), tool_id),
    )
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


# === Hosted Agents CRUD ===

def create_hosted_agent(agent_id: str, owner_email: str, owner_name: str = "",
                        telegram_id: str = "", tier: str = "basic",
                        bot_token: str = "", bot_username: str = "",
                        business_name: str = "", business_description: str = "",
                        industry: str = "general", max_bots: int = 1,
                        license_key: str = "") -> int:
    """Create a hosted agent entry."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()
    cursor.execute(
        """INSERT INTO hosted_agents
           (agent_id, owner_email, owner_name, telegram_id, tier,
            bot_token, bot_username, business_name, business_description,
            industry, max_bots, license_key, created_at, updated_at)
           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (agent_id, owner_email, owner_name, telegram_id, tier,
         bot_token, bot_username, business_name, business_description,
         industry, max_bots, license_key, now, now),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_hosted_agent(agent_id: str) -> Optional[Dict]:
    """Get hosted agent by agent_id."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM hosted_agents WHERE agent_id = ?", (agent_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_hosted_agent_by_email(email: str) -> Optional[Dict]:
    """Get hosted agent by owner email."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM hosted_agents WHERE owner_email = ?", (email,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def list_hosted_agents(tier: str = None, status: str = None) -> List[Dict]:
    """List all hosted agents."""
    conn = get_connection()
    cursor = conn.cursor()
    query = "SELECT * FROM hosted_agents WHERE 1=1"
    params: list = []
    if tier:
        query += " AND tier = ?"
        params.append(tier)
    if status:
        query += " AND status = ?"
        params.append(status)
    query += " ORDER BY created_at DESC"
    cursor.execute(query, params)
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def update_hosted_agent(agent_id: str, **kwargs) -> bool:
    """Update hosted agent fields."""
    kwargs["updated_at"] = datetime.utcnow().isoformat()
    conn = get_connection()
    cursor = conn.cursor()
    fields = ", ".join(f"{k} = ?" for k in kwargs)
    values = list(kwargs.values()) + [agent_id]
    cursor.execute(f"UPDATE hosted_agents SET {fields} WHERE agent_id = ?", values)
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


def adjust_agent_balance(agent_id: str, dzit_delta: float = 0, bzeusd_delta: float = 0) -> bool:
    """Adjust DZIT and/or BZeUSD balance for a hosted agent."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """UPDATE hosted_agents
           SET dzit_balance = dzit_balance + ?,
               bzeusd_balance = bzeusd_balance + ?,
               updated_at = ?
           WHERE agent_id = ?""",
        (dzit_delta, bzeusd_delta, datetime.utcnow().isoformat(), agent_id),
    )
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


# === User Interactions CRUD ===

def log_user_interaction(agent_id: str, chat_id: str, query: str, response: str = "",
                         user_name: str = "", topic: str = "", intent: str = "",
                         sentiment: str = "", session_id: str = "",
                         is_complete: int = 1) -> int:
    """Log a user interaction with a hosted bot."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO user_interactions
           (agent_id, chat_id, user_name, query, response, topic, intent,
            sentiment, session_id, is_complete, created_at)
           VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
        (agent_id, chat_id, user_name, query, response, topic, intent,
         sentiment, session_id, is_complete, datetime.utcnow().isoformat()),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_user_interactions(agent_id: str, chat_id: str, limit: int = 20) -> List[Dict]:
    """Get recent interactions for a user with a specific bot."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT * FROM user_interactions
           WHERE agent_id = ? AND chat_id = ?
           ORDER BY created_at DESC LIMIT ?""",
        (agent_id, chat_id, limit),
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_last_interaction(agent_id: str, chat_id: str) -> Optional[Dict]:
    """Get the most recent interaction for a user."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT * FROM user_interactions
           WHERE agent_id = ? AND chat_id = ?
           ORDER BY created_at DESC LIMIT 1""",
        (agent_id, chat_id),
    )
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_user_topics(agent_id: str, chat_id: str) -> List[Dict]:
    """Get topic frequency for a user."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT topic, COUNT(*) as count FROM user_interactions
           WHERE agent_id = ? AND chat_id = ? AND topic != '' AND topic IS NOT NULL
           GROUP BY topic ORDER BY count DESC LIMIT 10""",
        (agent_id, chat_id),
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_interactions_since(agent_id: str, since: str) -> List[Dict]:
    """Get all interactions for an agent since a timestamp."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT * FROM user_interactions
           WHERE agent_id = ? AND created_at >= ?
           ORDER BY created_at DESC""",
        (agent_id, since),
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


# === User Profiles CRUD ===

def upsert_user_profile(agent_id: str, chat_id: str, user_name: str = "",
                         user_email: str = "", topic: str = "",
                         query: str = "") -> bool:
    """Create or update a user profile with latest interaction data."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()

    # Try to get existing profile
    cursor.execute(
        "SELECT * FROM user_profiles WHERE agent_id = ? AND chat_id = ?",
        (agent_id, chat_id),
    )
    existing = cursor.fetchone()

    if existing:
        row = dict(existing)
        updates = {
            "last_seen": now,
            "total_messages": row["total_messages"] + 1,
            "updated_at": now,
        }
        if user_name:
            updates["user_name"] = user_name
        if user_email:
            updates["user_email"] = user_email
        if topic:
            updates["last_topic"] = topic
            # Update top_topics JSON
            try:
                top = json.loads(row.get("top_topics") or "{}")
            except (json.JSONDecodeError, TypeError):
                top = {}
            top[topic] = top.get(topic, 0) + 1
            updates["top_topics"] = json.dumps(top)
        if query:
            updates["last_query"] = query

        fields = ", ".join(f"{k} = ?" for k in updates)
        values = list(updates.values()) + [agent_id, chat_id]
        cursor.execute(
            f"UPDATE user_profiles SET {fields} WHERE agent_id = ? AND chat_id = ?",
            values,
        )
    else:
        top_topics = json.dumps({topic: 1}) if topic else "{}"
        cursor.execute(
            """INSERT INTO user_profiles
               (agent_id, chat_id, user_name, user_email, total_messages,
                first_seen, last_seen, last_topic, last_query, top_topics,
                interests, created_at, updated_at)
               VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
            (agent_id, chat_id, user_name, user_email, 1,
             now, now, topic, query, top_topics,
             "[]", now, now),
        )

    conn.commit()
    conn.close()
    return True


def get_user_profile(agent_id: str, chat_id: str) -> Optional[Dict]:
    """Get user profile for a specific bot."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM user_profiles WHERE agent_id = ? AND chat_id = ?",
        (agent_id, chat_id),
    )
    row = cursor.fetchone()
    conn.close()
    if row:
        d = dict(row)
        for field in ("top_topics", "interests"):
            if d.get(field) and isinstance(d[field], str):
                try:
                    d[field] = json.loads(d[field])
                except (json.JSONDecodeError, TypeError):
                    pass
        return d
    return None


def get_profiles_with_email(agent_id: str) -> List[Dict]:
    """Get all user profiles that have an email address for a bot."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT * FROM user_profiles
           WHERE agent_id = ? AND user_email IS NOT NULL AND user_email != ''
           AND email_opt_in = 1
           ORDER BY last_seen DESC""",
        (agent_id,),
    )
    rows = cursor.fetchall()
    conn.close()
    results = []
    for row in rows:
        d = dict(row)
        for field in ("top_topics", "interests"):
            if d.get(field) and isinstance(d[field], str):
                try:
                    d[field] = json.loads(d[field])
                except (json.JSONDecodeError, TypeError):
                    pass
        results.append(d)
    return results


def update_user_profile(agent_id: str, chat_id: str, **kwargs) -> bool:
    """Update user profile fields."""
    kwargs["updated_at"] = datetime.utcnow().isoformat()
    for field in ("top_topics", "interests"):
        if field in kwargs and isinstance(kwargs[field], (dict, list)):
            kwargs[field] = json.dumps(kwargs[field])
    conn = get_connection()
    cursor = conn.cursor()
    fields = ", ".join(f"{k} = ?" for k in kwargs)
    values = list(kwargs.values()) + [agent_id, chat_id]
    cursor.execute(
        f"UPDATE user_profiles SET {fields} WHERE agent_id = ? AND chat_id = ?",
        values,
    )
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


# === KB Sharing CRUD ===

def set_kb_sharing(agent_id: str, sharing_mode: str = "private",
                   cost_per_query: float = 0, cost_currency: str = "DZIT",
                   description: str = "") -> bool:
    """Set KB sharing config for an agent. Modes: shared, private, charged."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()
    cursor.execute(
        "SELECT id FROM kb_sharing_config WHERE agent_id = ?", (agent_id,)
    )
    if cursor.fetchone():
        cursor.execute(
            """UPDATE kb_sharing_config
               SET sharing_mode = ?, cost_per_query = ?, cost_currency = ?,
                   description = ?, updated_at = ?
               WHERE agent_id = ?""",
            (sharing_mode, cost_per_query, cost_currency, description, now, agent_id),
        )
    else:
        cursor.execute(
            """INSERT INTO kb_sharing_config
               (agent_id, sharing_mode, cost_per_query, cost_currency, description,
                created_at, updated_at)
               VALUES (?,?,?,?,?,?,?)""",
            (agent_id, sharing_mode, cost_per_query, cost_currency, description, now, now),
        )
    conn.commit()
    conn.close()
    return True


def get_kb_sharing(agent_id: str) -> Optional[Dict]:
    """Get KB sharing config for an agent."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM kb_sharing_config WHERE agent_id = ?", (agent_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_shared_kbs(include_charged: bool = True) -> List[Dict]:
    """Get all agents with shared or charged KBs."""
    conn = get_connection()
    cursor = conn.cursor()
    if include_charged:
        cursor.execute(
            "SELECT * FROM kb_sharing_config WHERE sharing_mode IN ('shared','charged') AND is_active = 1"
        )
    else:
        cursor.execute(
            "SELECT * FROM kb_sharing_config WHERE sharing_mode = 'shared' AND is_active = 1"
        )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def log_kb_access(requester_agent_id: str, provider_agent_id: str, query: str = "",
                  cost_charged: float = 0, cost_currency: str = "DZIT",
                  biznode_fee: float = 0, provider_earned: float = 0) -> int:
    """Log a KB access event."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO kb_access_log
           (requester_agent_id, provider_agent_id, query, cost_charged,
            cost_currency, biznode_fee, provider_earned, created_at)
           VALUES (?,?,?,?,?,?,?,?)""",
        (requester_agent_id, provider_agent_id, query, cost_charged,
         cost_currency, biznode_fee, provider_earned, datetime.utcnow().isoformat()),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def increment_kb_queries_served(agent_id: str, earned: float = 0) -> bool:
    """Increment query count and earnings for a KB provider."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """UPDATE kb_sharing_config
           SET total_queries_served = total_queries_served + 1,
               total_earned = total_earned + ?,
               updated_at = ?
           WHERE agent_id = ?""",
        (earned, datetime.utcnow().isoformat(), agent_id),
    )
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


# === Task Queue CRUD ===

def create_task(task_id: str, task_type: str, payload: str = "", priority: int = 5,
                agent_id: str = "", user_id: str = "", max_retries: int = 3,
                scheduled_at: str = "") -> str:
    """Create a new task in the queue."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()
    cursor.execute(
        """INSERT INTO task_queue
           (id, task_type, payload, priority, agent_id, user_id, max_retries,
            scheduled_at, created_at)
           VALUES (?,?,?,?,?,?,?,?,?)""",
        (task_id, task_type, payload, priority, agent_id, user_id, max_retries,
         scheduled_at or now, now),
    )
    conn.commit()
    conn.close()
    return task_id


def fetch_next_task(worker_node: str = "", task_types: list = None) -> Optional[Dict]:
    """Fetch and lock the next queued task (highest priority, oldest first)."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()

    query = """SELECT * FROM task_queue
               WHERE status = 'queued' AND (scheduled_at IS NULL OR scheduled_at <= ?)"""
    params = [now]

    if task_types:
        placeholders = ",".join("?" for _ in task_types)
        query += f" AND task_type IN ({placeholders})"
        params.extend(task_types)

    query += " ORDER BY priority DESC, created_at ASC LIMIT 1"

    cursor.execute(query, params)
    row = cursor.fetchone()
    if row:
        task = dict(row)
        cursor.execute(
            """UPDATE task_queue SET status = 'running', started_at = ?, worker_node = ?
               WHERE id = ? AND status = 'queued'""",
            (now, worker_node, task["id"]),
        )
        conn.commit()
        conn.close()
        task["status"] = "running"
        return task
    conn.close()
    return None


def complete_task(task_id: str, result: str = "") -> bool:
    """Mark a task as completed."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()
    cursor.execute(
        """UPDATE task_queue SET status = 'completed', completed_at = ?, result = ?
           WHERE id = ?""",
        (now, result, task_id),
    )
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


def fail_task(task_id: str, error: str = "") -> bool:
    """Mark a task as failed or requeue for retry."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT retries, max_retries FROM task_queue WHERE id = ?", (task_id,))
    row = cursor.fetchone()
    if not row:
        conn.close()
        return False

    retries = row[0] + 1
    max_retries = row[1]
    now = datetime.utcnow().isoformat()

    if retries < max_retries:
        cursor.execute(
            """UPDATE task_queue SET status = 'queued', retries = ?, error_message = ?,
               started_at = NULL, worker_node = NULL WHERE id = ?""",
            (retries, error, task_id),
        )
    else:
        cursor.execute(
            """UPDATE task_queue SET status = 'failed', retries = ?, error_message = ?,
               completed_at = ? WHERE id = ?""",
            (retries, error, now, task_id),
        )
    conn.commit()
    conn.close()
    return True


def get_task(task_id: str) -> Optional[Dict]:
    """Get a task by ID."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM task_queue WHERE id = ?", (task_id,))
    row = cursor.fetchone()
    conn.close()
    return dict(row) if row else None


def get_tasks_by_status(status: str = "queued", limit: int = 50) -> List[Dict]:
    """Get tasks by status."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM task_queue WHERE status = ? ORDER BY priority DESC, created_at ASC LIMIT ?",
        (status, limit),
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_task_stats() -> Dict:
    """Get task queue statistics."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """SELECT status, COUNT(*) as count FROM task_queue GROUP BY status"""
    )
    rows = cursor.fetchall()
    stats = {}
    for row in rows:
        d = dict(row)
        stats[d["status"]] = d["count"]
    cursor.execute("SELECT COUNT(*) as total FROM task_queue")
    total_row = dict(cursor.fetchone())
    stats["total"] = total_row.get("total", 0)
    conn.close()
    return stats


def log_task_event(task_id: str, log_message: str, status: str = "", worker_node: str = "") -> int:
    """Log a task event."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        """INSERT INTO task_logs (task_id, log_message, status, worker_node, created_at)
           VALUES (?,?,?,?,?)""",
        (task_id, log_message, status, worker_node, datetime.utcnow().isoformat()),
    )
    row_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return row_id


def get_task_logs(task_id: str) -> List[Dict]:
    """Get logs for a task."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM task_logs WHERE task_id = ? ORDER BY created_at ASC", (task_id,)
    )
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


# === Worker Nodes CRUD ===

def register_worker(node_id: str, capabilities: str = "[]") -> bool:
    """Register or update a worker node."""
    conn = get_connection()
    cursor = conn.cursor()
    now = datetime.utcnow().isoformat()
    cursor.execute("SELECT node_id FROM worker_nodes WHERE node_id = ?", (node_id,))
    if cursor.fetchone():
        cursor.execute(
            "UPDATE worker_nodes SET capabilities = ?, status = 'idle', last_heartbeat = ? WHERE node_id = ?",
            (capabilities, now, node_id),
        )
    else:
        cursor.execute(
            """INSERT INTO worker_nodes (node_id, capabilities, status, last_heartbeat, registered_at)
               VALUES (?,?,?,?,?)""",
            (node_id, capabilities, "idle", now, now),
        )
    conn.commit()
    conn.close()
    return True


def worker_heartbeat(node_id: str, status: str = "idle", current_task: str = "") -> bool:
    """Update worker heartbeat."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE worker_nodes SET last_heartbeat = ?, status = ?, current_task = ? WHERE node_id = ?",
        (datetime.utcnow().isoformat(), status, current_task, node_id),
    )
    affected = cursor.rowcount
    conn.commit()
    conn.close()
    return affected > 0


def get_active_workers() -> List[Dict]:
    """Get all active worker nodes."""
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM worker_nodes ORDER BY last_heartbeat DESC")
    rows = cursor.fetchall()
    conn.close()
    return [dict(r) for r in rows]


if __name__ == "__main__":
    init_db()
    print("Database setup complete.")
