feat(payments): add database schema and migrations

aisbf/database.py

@@ -144,6 +144,13 @@ class DatabaseManager:
             conn.commit()
             logger.debug(f"Recorded context dimension for {provider_id}/{model_name}")
     
+    def run_payment_migrations(self):
+        """Run payment system migrations"""
+        from aisbf.payments.migrations import PaymentMigrations
+        
+        migrations = PaymentMigrations(self)
+        migrations.run_migrations()
+    
     def get_context_dimension(
         self,
         provider_id: str,

aisbf/payments/__init__.py

+"""
+Payment system module
+"""
+from aisbf.payments.migrations import PaymentMigrations
+
+__all__ = ['PaymentMigrations']

aisbf/payments/migrations.py

+"""
+Payment system database migrations
+
+Creates all tables required for the complete payment system including:
+- Crypto wallet management (master keys, user addresses, transactions)
+- Payment methods and transactions
+- Subscriptions and billing
+- Background job management (locks, queues)
+- Configuration tables (price sources, consolidation settings, email config)
+"""
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from aisbf.database import DatabaseManager
+
+logger = logging.getLogger(__name__)
+
+
+class PaymentMigrations:
+    """Payment system database migrations"""
+    
+    def __init__(self, db_manager: 'DatabaseManager'):
+        """
+        Initialize migrations with database manager.
+        
+        Args:
+            db_manager: DatabaseManager instance
+        """
+        self.db = db_manager
+        self.db_type = db_manager.db_type
+        
+    def run_migrations(self):
+        """Run all payment system migrations"""
+        logger.info("Starting payment system migrations...")
+        
+        with self.db._get_connection() as conn:
+            cursor = conn.cursor()
+            
+            # Determine SQL syntax based on database type
+            if self.db_type == 'sqlite':
+                auto_increment = 'AUTOINCREMENT'
+                timestamp_default = 'CURRENT_TIMESTAMP'
+                boolean_type = 'BOOLEAN'
+                text_type = 'TEXT'
+                decimal_type = 'DECIMAL(18,8)'
+            else:  # mysql
+                auto_increment = 'AUTO_INCREMENT'
+                timestamp_default = 'CURRENT_TIMESTAMP'
+                boolean_type = 'TINYINT(1)'
+                text_type = 'TEXT'
+                decimal_type = 'DECIMAL(18,8)'
+            
+            # Create all payment system tables
+            self._create_crypto_tables(cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type)
+            self._create_payment_tables(cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type)
+            self._create_subscription_tables(cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type)
+            self._create_job_tables(cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type)
+            self._create_config_tables(cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type)
+            self._create_notification_tables(cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type)
+            self._insert_default_data(cursor)
+            
+            conn.commit()
+            logger.info("✅ Payment system migrations completed successfully")
+    
+    def _create_crypto_tables(self, cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type):
+        """Create cryptocurrency-related tables"""
+        
+        # Crypto master keys (encrypted BIP39 seeds)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS crypto_master_keys (
+                id INTEGER PRIMARY KEY {auto_increment},
+                crypto_type VARCHAR(20) NOT NULL UNIQUE,
+                encrypted_seed {text_type} NOT NULL,
+                encryption_key_id VARCHAR(50) NOT NULL,
+                derivation_path VARCHAR(100) NOT NULL,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default}
+            )
+        ''')
+        
+        # User crypto addresses (derived from master keys)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS user_crypto_addresses (
+                id INTEGER PRIMARY KEY {auto_increment},
+                user_id INTEGER NOT NULL,
+                crypto_type VARCHAR(20) NOT NULL,
+                address VARCHAR(255) NOT NULL UNIQUE,
+                derivation_index INTEGER NOT NULL,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                FOREIGN KEY (user_id) REFERENCES users(id),
+                UNIQUE(user_id, crypto_type)
+            )
+        ''')
+        
+        # User crypto wallets (balance tracking)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS user_crypto_wallets (
+                id INTEGER PRIMARY KEY {auto_increment},
+                user_id INTEGER NOT NULL,
+                crypto_type VARCHAR(20) NOT NULL,
+                balance_crypto {decimal_type} DEFAULT 0,
+                balance_fiat {decimal_type} DEFAULT 0,
+                last_updated TIMESTAMP DEFAULT {timestamp_default},
+                FOREIGN KEY (user_id) REFERENCES users(id),
+                UNIQUE(user_id, crypto_type)
+            )
+        ''')
+        
+        # Crypto transactions (incoming payments)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS crypto_transactions (
+                id INTEGER PRIMARY KEY {auto_increment},
+                user_id INTEGER NOT NULL,
+                address_id INTEGER NOT NULL,
+                crypto_type VARCHAR(20) NOT NULL,
+                tx_hash VARCHAR(255) NOT NULL UNIQUE,
+                amount_crypto {decimal_type} NOT NULL,
+                amount_fiat {decimal_type},
+                confirmations INTEGER DEFAULT 0,
+                required_confirmations INTEGER DEFAULT 3,
+                status VARCHAR(20) DEFAULT 'pending',
+                detected_at TIMESTAMP DEFAULT {timestamp_default},
+                confirmed_at TIMESTAMP NULL,
+                credited_at TIMESTAMP NULL,
+                FOREIGN KEY (user_id) REFERENCES users(id),
+                FOREIGN KEY (address_id) REFERENCES user_crypto_addresses(id)
+            )
+        ''')
+        
+        # Crypto webhooks (registered webhook IDs)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS crypto_webhooks (
+                id INTEGER PRIMARY KEY {auto_increment},
+                crypto_type VARCHAR(20) NOT NULL,
+                address VARCHAR(255) NOT NULL,
+                webhook_id VARCHAR(255) NOT NULL,
+                provider VARCHAR(50) NOT NULL,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                UNIQUE(crypto_type, address, provider)
+            )
+        ''')
+        
+        # Create indexes
+        try:
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_crypto_tx_user ON crypto_transactions(user_id)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_crypto_tx_status ON crypto_transactions(status)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_user_addresses_user ON user_crypto_addresses(user_id)')
+        except:
+            pass
+    
+    def _create_payment_tables(self, cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type):
+        """Create payment-related tables"""
+        
+        # Payment retry queue
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS payment_retry_queue (
+                id INTEGER PRIMARY KEY {auto_increment},
+                subscription_id INTEGER NOT NULL,
+                user_id INTEGER NOT NULL,
+                payment_method_type VARCHAR(50) NOT NULL,
+                amount {decimal_type} NOT NULL,
+                currency VARCHAR(10) DEFAULT 'USD',
+                attempt_count INTEGER DEFAULT 0,
+                max_attempts INTEGER DEFAULT 3,
+                next_retry_at TIMESTAMP NULL,
+                last_error {text_type},
+                status VARCHAR(20) DEFAULT 'pending',
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                completed_at TIMESTAMP NULL,
+                FOREIGN KEY (user_id) REFERENCES users(id)
+            )
+        ''')
+        
+        # API requests (for quota tracking)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS api_requests (
+                id INTEGER PRIMARY KEY {auto_increment},
+                user_id INTEGER NOT NULL,
+                endpoint VARCHAR(255) NOT NULL,
+                method VARCHAR(10) NOT NULL,
+                status_code INTEGER,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                FOREIGN KEY (user_id) REFERENCES users(id)
+            )
+        ''')
+        
+        # Create indexes
+        try:
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_payment_retry_status ON payment_retry_queue(status, next_retry_at)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_api_requests_user_time ON api_requests(user_id, created_at)')
+        except:
+            pass
+    
+    def _create_subscription_tables(self, cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type):
+        """Create subscription-related tables"""
+        
+        # Subscriptions table (enhanced version)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS subscriptions (
+                id INTEGER PRIMARY KEY {auto_increment},
+                user_id INTEGER NOT NULL,
+                tier_id INTEGER NOT NULL,
+                payment_method_id INTEGER,
+                status VARCHAR(20) DEFAULT 'active',
+                billing_cycle VARCHAR(20) DEFAULT 'monthly',
+                current_period_start TIMESTAMP DEFAULT {timestamp_default},
+                current_period_end TIMESTAMP NOT NULL,
+                cancel_at_period_end {boolean_type} DEFAULT 0,
+                pending_tier_id INTEGER,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default},
+                FOREIGN KEY (user_id) REFERENCES users(id),
+                FOREIGN KEY (tier_id) REFERENCES account_tiers(id),
+                FOREIGN KEY (payment_method_id) REFERENCES payment_methods(id),
+                FOREIGN KEY (pending_tier_id) REFERENCES account_tiers(id)
+            )
+        ''')
+        
+        # Create indexes
+        try:
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_subscriptions_user ON subscriptions(user_id)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_subscriptions_status ON subscriptions(status)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_subscriptions_period_end ON subscriptions(current_period_end)')
+        except:
+            pass
+    
+    def _create_job_tables(self, cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type):
+        """Create background job management tables"""
+        
+        # Job locks (distributed locking)
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS job_locks (
+                id INTEGER PRIMARY KEY {auto_increment},
+                job_name VARCHAR(100) NOT NULL UNIQUE,
+                instance_id VARCHAR(255) NOT NULL,
+                acquired_at TIMESTAMP DEFAULT {timestamp_default},
+                expires_at TIMESTAMP NOT NULL
+            )
+        ''')
+        
+        # Crypto consolidation queue
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS crypto_consolidation_queue (
+                id INTEGER PRIMARY KEY {auto_increment},
+                crypto_type VARCHAR(20) NOT NULL,
+                total_balance {decimal_type} NOT NULL,
+                address_count INTEGER NOT NULL,
+                status VARCHAR(20) DEFAULT 'pending',
+                tx_hash VARCHAR(255),
+                error_message {text_type},
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                processed_at TIMESTAMP NULL
+            )
+        ''')
+        
+        # Email notification queue
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS email_notification_queue (
+                id INTEGER PRIMARY KEY {auto_increment},
+                user_id INTEGER NOT NULL,
+                notification_type VARCHAR(50) NOT NULL,
+                recipient_email VARCHAR(255) NOT NULL,
+                subject VARCHAR(255) NOT NULL,
+                body {text_type} NOT NULL,
+                attempt_count INTEGER DEFAULT 0,
+                max_attempts INTEGER DEFAULT 3,
+                next_retry_at TIMESTAMP NULL,
+                status VARCHAR(20) DEFAULT 'pending',
+                error_message {text_type},
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                sent_at TIMESTAMP NULL,
+                FOREIGN KEY (user_id) REFERENCES users(id)
+            )
+        ''')
+        
+        # Create indexes
+        try:
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_job_locks_expires ON job_locks(expires_at)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_consolidation_status ON crypto_consolidation_queue(status)')
+            cursor.execute('CREATE INDEX IF NOT EXISTS idx_email_queue_status ON email_notification_queue(status, next_retry_at)')
+        except:
+            pass
+    
+    def _create_config_tables(self, cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type):
+        """Create configuration tables"""
+        
+        # Crypto price sources
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS crypto_price_sources (
+                id INTEGER PRIMARY KEY {auto_increment},
+                name VARCHAR(50) NOT NULL UNIQUE,
+                api_type VARCHAR(50) NOT NULL,
+                endpoint_url VARCHAR(255) NOT NULL,
+                api_key VARCHAR(255),
+                priority INTEGER DEFAULT 0,
+                is_enabled {boolean_type} DEFAULT 1,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default}
+            )
+        ''')
+        
+        # Crypto consolidation settings
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS crypto_consolidation_settings (
+                id INTEGER PRIMARY KEY {auto_increment},
+                crypto_type VARCHAR(20) NOT NULL UNIQUE,
+                threshold_amount {decimal_type} NOT NULL,
+                admin_address VARCHAR(255) NOT NULL,
+                is_enabled {boolean_type} DEFAULT 1,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default}
+            )
+        ''')
+        
+        # Payment gateway config
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS payment_gateway_config (
+                id INTEGER PRIMARY KEY {auto_increment},
+                gateway_name VARCHAR(50) NOT NULL UNIQUE,
+                config_json {text_type} NOT NULL,
+                is_enabled {boolean_type} DEFAULT 1,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default}
+            )
+        ''')
+        
+        # Crypto API config
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS crypto_api_config (
+                id INTEGER PRIMARY KEY {auto_increment},
+                crypto_type VARCHAR(20) NOT NULL,
+                api_provider VARCHAR(50) NOT NULL,
+                api_key VARCHAR(255),
+                config_json {text_type},
+                is_enabled {boolean_type} DEFAULT 1,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default},
+                UNIQUE(crypto_type, api_provider)
+            )
+        ''')
+    
+    def _create_notification_tables(self, cursor, auto_increment, timestamp_default, boolean_type, text_type, decimal_type):
+        """Create email notification configuration tables"""
+        
+        # Email notification settings
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS email_notification_settings (
+                id INTEGER PRIMARY KEY {auto_increment},
+                notification_type VARCHAR(50) NOT NULL UNIQUE,
+                is_enabled {boolean_type} DEFAULT 1,
+                subject_template VARCHAR(255) NOT NULL,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default}
+            )
+        ''')
+        
+        # Email templates
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS email_templates (
+                id INTEGER PRIMARY KEY {auto_increment},
+                notification_type VARCHAR(50) NOT NULL UNIQUE,
+                template_html {text_type} NOT NULL,
+                template_text {text_type},
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default}
+            )
+        ''')
+        
+        # Email config
+        cursor.execute(f'''
+            CREATE TABLE IF NOT EXISTS email_config (
+                id INTEGER PRIMARY KEY {auto_increment},
+                smtp_host VARCHAR(255) NOT NULL,
+                smtp_port INTEGER NOT NULL,
+                smtp_username VARCHAR(255),
+                smtp_password VARCHAR(255),
+                from_email VARCHAR(255) NOT NULL,
+                from_name VARCHAR(255),
+                use_tls {boolean_type} DEFAULT 1,
+                created_at TIMESTAMP DEFAULT {timestamp_default},
+                updated_at TIMESTAMP DEFAULT {timestamp_default}
+            )
+        ''')
+    
+    def _insert_default_data(self, cursor):
+        """Insert default configuration data"""
+        
+        # Insert default price sources
+        default_sources = [
+            ('Coinbase', 'rest', 'https://api.coinbase.com/v2/prices', None, 1),
+            ('Binance', 'rest', 'https://api.binance.com/api/v3/ticker/price', None, 2),
+            ('Kraken', 'rest', 'https://api.kraken.com/0/public/Ticker', None, 3)
+        ]
+        
+        for name, api_type, endpoint, api_key, priority in default_sources:
+            try:
+                if self.db_type == 'sqlite':
+                    cursor.execute('''
+                        INSERT OR IGNORE INTO crypto_price_sources (name, api_type, endpoint_url, api_key, priority)
+                        VALUES (?, ?, ?, ?, ?)
+                    ''', (name, api_type, endpoint, api_key, priority))
+                else:
+                    cursor.execute('''
+                        INSERT IGNORE INTO crypto_price_sources (name, api_type, endpoint_url, api_key, priority)
+                        VALUES (%s, %s, %s, %s, %s)
+                    ''', (name, api_type, endpoint, api_key, priority))
+            except:
+                pass
+        
+        # Insert default consolidation settings
+        default_consolidation = [
+            ('BTC', '0.1', ''),
+            ('ETH', '1.0', ''),
+            ('USDT', '1000.0', ''),
+            ('USDC', '1000.0', '')
+        ]
+        
+        for crypto_type, threshold, address in default_consolidation:
+            try:
+                if self.db_type == 'sqlite':
+                    cursor.execute('''
+                        INSERT OR IGNORE INTO crypto_consolidation_settings (crypto_type, threshold_amount, admin_address, is_enabled)
+                        VALUES (?, ?, ?, 0)
+                    ''', (crypto_type, threshold, address))
+                else:
+                    cursor.execute('''
+                        INSERT IGNORE INTO crypto_consolidation_settings (crypto_type, threshold_amount, admin_address, is_enabled)
+                        VALUES (%s, %s, %s, 0)
+                    ''', (crypto_type, threshold, address))
+            except:
+                pass
+        
+        # Insert default email notification settings
+        default_notifications = [
+            ('payment_failed', 'Payment Failed'),
+            ('payment_retry_success', 'Payment Successful'),
+            ('subscription_created', 'Subscription Created'),
+            ('subscription_renewed', 'Subscription Renewed'),
+            ('subscription_upgraded', 'Subscription Upgraded'),
+            ('subscription_downgrade_scheduled', 'Subscription Downgrade Scheduled'),
+            ('subscription_canceled', 'Subscription Canceled'),
+            ('subscription_downgraded', 'Subscription Downgraded'),
+            ('crypto_wallet_credited', 'Crypto Payment Received')
+        ]
+        
+        for notif_type, subject in default_notifications:
+            try:
+                if self.db_type == 'sqlite':
+                    cursor.execute('''
+                        INSERT OR IGNORE INTO email_notification_settings (notification_type, subject_template, is_enabled)
+                        VALUES (?, ?, 1)
+                    ''', (notif_type, subject))
+                else:
+                    cursor.execute('''
+                        INSERT IGNORE INTO email_notification_settings (notification_type, subject_template, is_enabled)
+                        VALUES (%s, %s, 1)
+                    ''', (notif_type, subject))
+            except:
+                pass
+        
+        logger.info("✅ Default payment system data inserted")

tests/payments/__init__.py

+"""Payment system tests"""

tests/payments/test_migrations.py

+import pytest
+from aisbf.database import DatabaseManager
+from aisbf.payments.migrations import PaymentMigrations
+
+
+def test_migrations_create_all_tables(tmp_path):
+    """Test that migrations create all required tables"""
+    db_path = tmp_path / "test.db"
+    db_config = {
+        'type': 'sqlite',
+        'sqlite_path': str(db_path)
+    }
+    
+    db = DatabaseManager(db_config)
+    migrations = PaymentMigrations(db)
+    migrations.run_migrations()
+    
+    # Check that key tables exist
+    with db._get_connection() as conn:
+        cursor = conn.cursor()
+        cursor.execute("""
+            SELECT name FROM sqlite_master 
+            WHERE type='table'
+        """)
+        tables = cursor.fetchall()
+    
+    table_names = [t[0] for t in tables]
+    
+    assert 'crypto_master_keys' in table_names
+    assert 'user_crypto_addresses' in table_names
+    assert 'user_crypto_wallets' in table_names
+    assert 'crypto_transactions' in table_names
+    assert 'payment_retry_queue' in table_names
+    assert 'subscriptions' in table_names
+    assert 'job_locks' in table_names
+    assert 'crypto_price_sources' in table_names
+    assert 'email_notification_settings' in table_names
+
+
+def test_migrations_insert_default_data(tmp_path):
+    """Test that default data is inserted"""
+    db_path = tmp_path / "test.db"
+    db_config = {
+        'type': 'sqlite',
+        'sqlite_path': str(db_path)
+    }
+    
+    db = DatabaseManager(db_config)
+    migrations = PaymentMigrations(db)
+    migrations.run_migrations()
+    
+    # Check price sources
+    with db._get_connection() as conn:
+        cursor = conn.cursor()
+        cursor.execute("SELECT * FROM crypto_price_sources")
+        sources = cursor.fetchall()
+        assert len(sources) >= 3
+        
+        # Check consolidation settings
+        cursor.execute("SELECT * FROM crypto_consolidation_settings")
+        settings = cursor.fetchall()
+        assert len(settings) == 4
+        
+        # Check email notification settings
+        cursor.execute("SELECT * FROM email_notification_settings")
+        notifications = cursor.fetchall()
+        assert len(notifications) >= 9