# core/database.py from __future__ import annotations import asyncio import logging import sqlite3 from pathlib import Path from typing import Any, Optional logger = logging.getLogger(__name__) # ============================================================ # Database Manager # ============================================================ class DatabaseManager: """ Async-friendly SQLite database manager for the current project schema. Canonical user primary key: users.user_id Important: - All SQLite work is synchronous internally and runs through asyncio.to_thread() from async methods. - No aiosqlite connection is awaited twice. - Existing databases are migrated non-destructively where possible. - Broken historical foreign keys to users(id) are repaired. """ VALID_ORDER_STATUSES = { # Owner approval workflow "pending_owner_approval", "owner_processing", "rejected_by_owner", "cancelled_by_user", "price_changed", "insufficient_balance", "provider_processing", "provider_status_unknown", "completed", "failed_refunded", "failed_refund_required", # Legacy compatibility "pending", "processing", "paid", "failed", "cancelled", "refunded", "created", } def __init__( self, db_path: str = "matrix_bot.db", ) -> None: self.db_path = str( Path(db_path) ) # Compatibility attribute used by some handlers. self.database_path = ( self.db_path ) self._initialized = False self._init_lock = asyncio.Lock() # ======================================================== # Connection / Introspection # ======================================================== def _connect( self, *, foreign_keys: bool = True, ) -> sqlite3.Connection: conn = sqlite3.connect( self.db_path, timeout=30, check_same_thread=False, ) conn.row_factory = ( sqlite3.Row ) conn.execute( "PRAGMA journal_mode = WAL" ) conn.execute( "PRAGMA synchronous = NORMAL" ) conn.execute( "PRAGMA busy_timeout = 30000" ) conn.execute( ( "PRAGMA foreign_keys = ON" if foreign_keys else "PRAGMA foreign_keys = OFF" ) ) return conn @staticmethod def _table_exists( conn: sqlite3.Connection, table_name: str, ) -> bool: row = conn.execute( """ SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1 """, ( table_name, ), ).fetchone() return row is not None @classmethod def _column_names( cls, conn: sqlite3.Connection, table_name: str, ) -> set[str]: if not cls._table_exists( conn, table_name, ): return set() return { str( row["name"] ) for row in conn.execute( f"PRAGMA table_info({table_name})" ).fetchall() } @classmethod def _foreign_keys( cls, conn: sqlite3.Connection, table_name: str, ) -> list[sqlite3.Row]: if not cls._table_exists( conn, table_name, ): return [] return list( conn.execute( f"PRAGMA foreign_key_list({table_name})" ).fetchall() ) @classmethod def _has_user_id_fk( cls, conn: sqlite3.Connection, table_name: str, from_column: str = "user_id", ) -> bool: return any( str(row["table"]) == "users" and str(row["from"]) == from_column and str(row["to"]) == "user_id" for row in cls._foreign_keys( conn, table_name, ) ) # ======================================================== # Initialization # ======================================================== async def init( self, ) -> None: if self._initialized: return async with self._init_lock: if self._initialized: return await asyncio.to_thread( self._create_tables ) self._initialized = True logger.info( "Database initialized: %s", self.db_path, ) async def create_tables( self, ) -> None: await self.init() async def _ensure_initialized( self, ) -> None: if not self._initialized: await self.init() def _create_tables( self, ) -> None: """ Main schema migration. foreign_keys is disabled only for the migration connection because historical databases can contain: orders.user_id -> users.id wallet_transactions.user_id -> users.id After migration, normal connections always enable foreign keys. """ conn = self._connect( foreign_keys=False ) legacy_users_table: Optional[ str ] = None try: conn.execute( "BEGIN IMMEDIATE" ) legacy_users_table = ( self._ensure_users_schema( conn ) ) self._ensure_orders_schema( conn ) self._ensure_wallet_transactions_schema( conn ) self._ensure_referrals_schema( conn ) self._ensure_kyc_schema( conn ) self._ensure_audit_schema( conn ) self._ensure_finance_schema( conn ) self._create_indexes( conn ) if ( legacy_users_table and self._table_exists( conn, legacy_users_table, ) ): conn.execute( f"DROP TABLE {legacy_users_table}" ) conn.commit() except Exception: conn.rollback() logger.exception( "Failed to initialize/migrate database schema." ) raise finally: conn.close() # ======================================================== # Users Schema # ======================================================== @staticmethod def _create_canonical_users( conn: sqlite3.Connection, table_name: str = "users", ) -> None: conn.execute( f""" CREATE TABLE IF NOT EXISTS {table_name} ( user_id INTEGER PRIMARY KEY, username TEXT, full_name TEXT, is_banned INTEGER NOT NULL DEFAULT 0, is_admin INTEGER NOT NULL DEFAULT 0, user_tier TEXT NOT NULL DEFAULT 'standard', is_kyc_verified INTEGER NOT NULL DEFAULT 0, balance_toman INTEGER NOT NULL DEFAULT 0, balance_ton REAL NOT NULL DEFAULT 0, balance_usdt REAL NOT NULL DEFAULT 0, invited_by INTEGER, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) """ ) def _ensure_users_schema( self, conn: sqlite3.Connection, ) -> Optional[str]: """ Returns the temporary legacy users table name if users(id) had to be migrated. It is dropped only after dependent tables are rebuilt. """ if not self._table_exists( conn, "users", ): self._create_canonical_users( conn ) return None columns = self._column_names( conn, "users", ) # ---------------------------------------------------- # Current schema: users.user_id # ---------------------------------------------------- if "user_id" in columns: additions = { "username": "TEXT", "full_name": "TEXT", "is_banned": "INTEGER NOT NULL DEFAULT 0", "is_admin": "INTEGER NOT NULL DEFAULT 0", "user_tier": "TEXT NOT NULL DEFAULT 'standard'", "is_kyc_verified": "INTEGER NOT NULL DEFAULT 0", "balance_toman": "INTEGER NOT NULL DEFAULT 0", "balance_ton": "REAL NOT NULL DEFAULT 0", "balance_usdt": "REAL NOT NULL DEFAULT 0", "invited_by": "INTEGER", "created_at": "TIMESTAMP", "updated_at": "TIMESTAMP", } for ( column, definition, ) in additions.items(): if column not in columns: conn.execute( ( "ALTER TABLE users " f"ADD COLUMN {column} {definition}" ) ) # Compatibility migration from legacy columns that may coexist. refreshed = self._column_names( conn, "users", ) if ( "role" in refreshed and "is_admin" in refreshed ): conn.execute( """ UPDATE users SET is_admin = 1 WHERE LOWER( COALESCE(role, '') ) IN ( 'admin', 'owner', 'administrator' ) """ ) if ( "kyc_status" in refreshed and "is_kyc_verified" in refreshed ): conn.execute( """ UPDATE users SET is_kyc_verified = 1 WHERE LOWER( COALESCE(kyc_status, '') ) IN ( 'approved', 'verified', 'accepted' ) """ ) return None # ---------------------------------------------------- # Historical schema: users.id # ---------------------------------------------------- if "id" not in columns: raise RuntimeError( "users table has neither user_id nor id." ) legacy_name = ( "users_legacy_database_manager" ) if self._table_exists( conn, legacy_name, ): raise RuntimeError( f"{legacy_name} already exists. " "Inspect the database before retrying migration." ) conn.execute( ( "ALTER TABLE users " f"RENAME TO {legacy_name}" ) ) self._create_canonical_users( conn ) old_columns = self._column_names( conn, legacy_name, ) id_expr = "id" username_expr = ( "username" if "username" in old_columns else "NULL" ) if "full_name" in old_columns: full_name_expr = ( "full_name" ) elif ( "first_name" in old_columns or "last_name" in old_columns ): first = ( "COALESCE(first_name, '')" if "first_name" in old_columns else "''" ) last = ( "COALESCE(last_name, '')" if "last_name" in old_columns else "''" ) full_name_expr = ( "TRIM(" + first + " || ' ' || " + last + ")" ) else: full_name_expr = "NULL" is_banned_expr = ( "COALESCE(is_banned, 0)" if "is_banned" in old_columns else ( "COALESCE(is_blocked, 0)" if "is_blocked" in old_columns else "0" ) ) if "is_admin" in old_columns: is_admin_expr = ( "COALESCE(is_admin, 0)" ) elif "role" in old_columns: is_admin_expr = ( "CASE " "WHEN LOWER(COALESCE(role, '')) " "IN ('admin', 'owner', 'administrator') " "THEN 1 ELSE 0 END" ) else: is_admin_expr = "0" user_tier_expr = ( "COALESCE(user_tier, 'standard')" if "user_tier" in old_columns else ( "COALESCE(tier, 'standard')" if "tier" in old_columns else "'standard'" ) ) if "is_kyc_verified" in old_columns: kyc_expr = ( "COALESCE(is_kyc_verified, 0)" ) elif "kyc_status" in old_columns: kyc_expr = ( "CASE " "WHEN LOWER(COALESCE(kyc_status, '')) " "IN ('approved', 'verified', 'accepted') " "THEN 1 ELSE 0 END" ) else: kyc_expr = "0" balance_toman_expr = ( "COALESCE(balance_toman, 0)" if "balance_toman" in old_columns else ( "COALESCE(balance, 0)" if "balance" in old_columns else "0" ) ) balance_ton_expr = ( "COALESCE(balance_ton, 0)" if "balance_ton" in old_columns else "0" ) balance_usdt_expr = ( "COALESCE(balance_usdt, 0)" if "balance_usdt" in old_columns else "0" ) invited_by_expr = ( "invited_by" if "invited_by" in old_columns else "NULL" ) created_at_expr = ( "created_at" if "created_at" in old_columns else "CURRENT_TIMESTAMP" ) updated_at_expr = ( "updated_at" if "updated_at" in old_columns else "CURRENT_TIMESTAMP" ) conn.execute( f""" INSERT OR IGNORE INTO users ( user_id, username, full_name, is_banned, is_admin, user_tier, is_kyc_verified, balance_toman, balance_ton, balance_usdt, invited_by, created_at, updated_at ) SELECT {id_expr}, {username_expr}, {full_name_expr}, {is_banned_expr}, {is_admin_expr}, {user_tier_expr}, {kyc_expr}, {balance_toman_expr}, {balance_ton_expr}, {balance_usdt_expr}, {invited_by_expr}, {created_at_expr}, {updated_at_expr} FROM {legacy_name} """ ) logger.info( "Migrated legacy users.id schema " "to users.user_id." ) return legacy_name # ======================================================== # Orders Schema # ======================================================== @staticmethod def _create_canonical_orders( conn: sqlite3.Connection, ) -> None: conn.execute( """ CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, product_id TEXT NOT NULL, target TEXT, price INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'pending_owner_approval', gateway TEXT NOT NULL DEFAULT 'wallet', market_order_id TEXT, owner_id INTEGER, provider_required_price INTEGER, provider_error TEXT, provider_started_at TIMESTAMP, completed_at TIMESTAMP, refunded_at TIMESTAMP, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE ) """ ) def _ensure_orders_schema( self, conn: sqlite3.Connection, ) -> None: if not self._table_exists( conn, "orders", ): self._create_canonical_orders( conn ) return columns = self._column_names( conn, "orders", ) required_base = { "id", "user_id", "product_id", "target", "price", "status", "gateway", "market_order_id", "created_at", "updated_at", } correct_fk = ( self._has_user_id_fk( conn, "orders", ) ) if ( required_base.issubset( columns ) and correct_fk ): additions = { "owner_id": "INTEGER", "provider_required_price": "INTEGER", "provider_error": "TEXT", "provider_started_at": "TIMESTAMP", "completed_at": "TIMESTAMP", "refunded_at": "TIMESTAMP", } for ( column, definition, ) in additions.items(): if column not in columns: conn.execute( ( "ALTER TABLE orders " f"ADD COLUMN {column} {definition}" ) ) return legacy = ( "orders_legacy_database_manager" ) if self._table_exists( conn, legacy, ): raise RuntimeError( f"{legacy} already exists." ) conn.execute( ( "ALTER TABLE orders " f"RENAME TO {legacy}" ) ) self._create_canonical_orders( conn ) old_columns = self._column_names( conn, legacy, ) new_columns = self._column_names( conn, "orders", ) copy_columns = [ column for column in ( "id", "user_id", "product_id", "target", "price", "status", "gateway", "market_order_id", "owner_id", "provider_required_price", "provider_error", "provider_started_at", "completed_at", "refunded_at", "created_at", "updated_at", ) if ( column in old_columns and column in new_columns ) ] if copy_columns: names = ", ".join( copy_columns ) conn.execute( f""" INSERT OR IGNORE INTO orders ( {names} ) SELECT {names} FROM {legacy} """ ) conn.execute( f"DROP TABLE {legacy}" ) logger.info( "Repaired orders foreign key " "to users(user_id)." ) # ======================================================== # Wallet Transactions Schema # ======================================================== @staticmethod def _create_canonical_wallet_transactions( conn: sqlite3.Connection, ) -> None: conn.execute( """ CREATE TABLE IF NOT EXISTS wallet_transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, type TEXT NOT NULL, amount INTEGER NOT NULL, balance_before INTEGER NOT NULL, balance_after INTEGER NOT NULL, description TEXT, reference_id TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE ) """ ) def _ensure_wallet_transactions_schema( self, conn: sqlite3.Connection, ) -> None: if not self._table_exists( conn, "wallet_transactions", ): self._create_canonical_wallet_transactions( conn ) return columns = self._column_names( conn, "wallet_transactions", ) required = { "id", "user_id", "type", "amount", "balance_before", "balance_after", "description", "reference_id", "created_at", } if ( required.issubset( columns ) and self._has_user_id_fk( conn, "wallet_transactions", ) ): return legacy = ( "wallet_transactions_legacy_database_manager" ) if self._table_exists( conn, legacy, ): raise RuntimeError( f"{legacy} already exists." ) conn.execute( ( "ALTER TABLE wallet_transactions " f"RENAME TO {legacy}" ) ) self._create_canonical_wallet_transactions( conn ) old_columns = self._column_names( conn, legacy, ) new_columns = self._column_names( conn, "wallet_transactions", ) copy_columns = [ column for column in ( "id", "user_id", "type", "amount", "balance_before", "balance_after", "description", "reference_id", "created_at", ) if ( column in old_columns and column in new_columns ) ] if copy_columns: names = ", ".join( copy_columns ) conn.execute( f""" INSERT OR IGNORE INTO wallet_transactions ( {names} ) SELECT {names} FROM {legacy} """ ) conn.execute( f"DROP TABLE {legacy}" ) logger.info( "Repaired wallet_transactions foreign key " "to users(user_id)." ) # ======================================================== # Referrals Schema # ======================================================== @staticmethod def _create_canonical_referrals( conn: sqlite3.Connection, ) -> None: conn.execute( """ CREATE TABLE IF NOT EXISTS referrals ( id INTEGER PRIMARY KEY AUTOINCREMENT, referrer_id INTEGER NOT NULL, referred_id INTEGER NOT NULL, reward_earned INTEGER NOT NULL DEFAULT 0, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (referrer_id) REFERENCES users(user_id) ON DELETE CASCADE, FOREIGN KEY (referred_id) REFERENCES users(user_id) ON DELETE CASCADE ) """ ) def _ensure_referrals_schema( self, conn: sqlite3.Connection, ) -> None: if not self._table_exists( conn, "referrals", ): self._create_canonical_referrals( conn ) return columns = self._column_names( conn, "referrals", ) required = { "id", "referrer_id", "referred_id", "reward_earned", "created_at", } fks = self._foreign_keys( conn, "referrals", ) correct_referrer = any( str(row["table"]) == "users" and str(row["from"]) == "referrer_id" and str(row["to"]) == "user_id" for row in fks ) correct_referred = any( str(row["table"]) == "users" and str(row["from"]) == "referred_id" and str(row["to"]) == "user_id" for row in fks ) if ( required.issubset( columns ) and correct_referrer and correct_referred ): return legacy = ( "referrals_legacy_database_manager" ) if self._table_exists( conn, legacy, ): raise RuntimeError( f"{legacy} already exists." ) conn.execute( ( "ALTER TABLE referrals " f"RENAME TO {legacy}" ) ) self._create_canonical_referrals( conn ) old_columns = self._column_names( conn, legacy, ) copy_columns = [ column for column in ( "id", "referrer_id", "referred_id", "reward_earned", "created_at", ) if column in old_columns ] if copy_columns: names = ", ".join( copy_columns ) conn.execute( f""" INSERT OR IGNORE INTO referrals ( {names} ) SELECT {names} FROM {legacy} """ ) conn.execute( f"DROP TABLE {legacy}" ) # ======================================================== # KYC / Audit / Finance Schema # ======================================================== @staticmethod def _ensure_kyc_schema( conn: sqlite3.Connection, ) -> None: conn.execute( """ CREATE TABLE IF NOT EXISTS kyc_requests ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, file_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', reviewed_by INTEGER, rejection_reason TEXT, warning_text TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, reviewed_at TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE CASCADE ) """ ) @staticmethod def _ensure_audit_schema( conn: sqlite3.Connection, ) -> None: conn.execute( """ CREATE TABLE IF NOT EXISTS audit_logs ( log_id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, action TEXT NOT NULL, details TEXT, ip_address TEXT, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) """ ) @staticmethod def _ensure_finance_schema( conn: sqlite3.Connection, ) -> None: conn.execute( """ CREATE TABLE IF NOT EXISTS bot_finance_rates ( rate_key TEXT PRIMARY KEY, rate_value TEXT NOT NULL, source TEXT NOT NULL DEFAULT 'admin', updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) """ ) # ======================================================== # Indexes # ======================================================== @staticmethod def _create_indexes( conn: sqlite3.Connection, ) -> None: statements = [ """ CREATE INDEX IF NOT EXISTS idx_users_username ON users(username) """, """ CREATE INDEX IF NOT EXISTS idx_users_invited_by ON users(invited_by) """, """ CREATE INDEX IF NOT EXISTS idx_orders_user_id ON orders(user_id) """, """ CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status) """, """ CREATE INDEX IF NOT EXISTS idx_orders_user_status ON orders(user_id, status) """, """ CREATE INDEX IF NOT EXISTS idx_orders_created_at ON orders(created_at) """, """ CREATE INDEX IF NOT EXISTS idx_orders_market_order_id ON orders(market_order_id) """, """ CREATE INDEX IF NOT EXISTS idx_wallet_transactions_user_id ON wallet_transactions(user_id) """, """ CREATE INDEX IF NOT EXISTS idx_wallet_transactions_reference ON wallet_transactions(reference_id) """, """ CREATE INDEX IF NOT EXISTS idx_referrals_referrer ON referrals(referrer_id) """, """ CREATE INDEX IF NOT EXISTS idx_referrals_referred ON referrals(referred_id) """, """ CREATE INDEX IF NOT EXISTS idx_kyc_requests_user_status ON kyc_requests(user_id, status) """, """ CREATE INDEX IF NOT EXISTS idx_kyc_requests_status ON kyc_requests(status) """, ] for statement in statements: conn.execute( statement ) # ======================================================== # User Helpers # ======================================================== @staticmethod def _clean_username( username: Optional[str], ) -> Optional[str]: if username is None: return None value = str( username ).strip() if value in { "", "ندارد", "none", "None", }: return None if value.startswith("@"): value = value[1:] return value or None @staticmethod def _clean_full_name( full_name: Optional[str], ) -> Optional[str]: if full_name is None: return None value = str( full_name ).strip() if value in { "", "بدون نام", }: return None return value or None @staticmethod def _user_row_to_dict( row: sqlite3.Row, ) -> dict[str, Any]: data = dict( row ) user_id = int( data.get( "user_id", 0, ) or 0 ) is_admin = bool( int( data.get( "is_admin", 0, ) or 0 ) ) is_banned = bool( int( data.get( "is_banned", 0, ) or 0 ) ) is_kyc_verified = bool( int( data.get( "is_kyc_verified", 0, ) or 0 ) ) # Compatibility aliases for older handlers/middleware. data["id"] = user_id data["user_id"] = user_id data["is_admin"] = is_admin data["is_banned"] = is_banned data["is_blocked"] = is_banned data["is_active"] = ( not is_banned ) data["role"] = ( "admin" if is_admin else "user" ) data["tier"] = str( data.get( "user_tier", "standard", ) or "standard" ) data["kyc_status"] = ( "approved" if is_kyc_verified else "none" ) return data # ======================================================== # Create / Upsert User # ======================================================== async def create_user( self, user_id: int, username: Optional[str] = None, first_name: Optional[str] = None, last_name: Optional[str] = None, ) -> bool: await self._ensure_initialized() full_name = " ".join( part for part in ( str(first_name or "").strip(), str(last_name or "").strip(), ) if part ) or None return await asyncio.to_thread( self._create_user_sync, int(user_id), self._clean_username( username ), full_name, ) def _create_user_sync( self, user_id: int, username: Optional[str], full_name: Optional[str], ) -> bool: if user_id <= 0: return False conn = self._connect() try: conn.execute( """ INSERT INTO users ( user_id, username, full_name ) VALUES (?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET username = excluded.username, full_name = excluded.full_name, updated_at = CURRENT_TIMESTAMP """, ( user_id, username, full_name, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Could not create/update user %s", user_id, ) return False finally: conn.close() async def upsert_user_activity( self, *, user_id: int, username: Optional[str] = None, full_name: Optional[str] = None, referrer_id: Optional[int] = None, ) -> bool: """ Backward-compatible /start upsert. Referral rules: - only when the user is first created; - self-referral is rejected; - referrer must already exist; - inviter can never be replaced later; - users.invited_by and referrals are written atomically. """ await self._ensure_initialized() return await asyncio.to_thread( self._upsert_user_activity_sync, int(user_id), self._clean_username( username ), self._clean_full_name( full_name ), ( int(referrer_id) if referrer_id is not None else None ), ) def _upsert_user_activity_sync( self, user_id: int, username: Optional[str], full_name: Optional[str], referrer_id: Optional[int], ) -> bool: if user_id <= 0: return False conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) existing = conn.execute( """ SELECT user_id, invited_by FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if existing: conn.execute( """ UPDATE users SET username = ?, full_name = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( username, full_name, user_id, ), ) conn.commit() return True valid_referrer: Optional[ int ] = None if ( referrer_id is not None and referrer_id > 0 and referrer_id != user_id ): referrer = conn.execute( """ SELECT user_id, is_banned FROM users WHERE user_id = ? LIMIT 1 """, ( referrer_id, ), ).fetchone() if ( referrer and int( referrer["is_banned"] or 0 ) == 0 ): valid_referrer = ( referrer_id ) conn.execute( """ INSERT INTO users ( user_id, username, full_name, invited_by ) VALUES (?, ?, ?, ?) """, ( user_id, username, full_name, valid_referrer, ), ) if valid_referrer is not None: existing_referral = ( conn.execute( """ SELECT 1 FROM referrals WHERE referred_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() ) if not existing_referral: conn.execute( """ INSERT INTO referrals ( referrer_id, referred_id, reward_earned ) VALUES (?, ?, 0) """, ( valid_referrer, user_id, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Could not upsert user activity | " "user=%s referrer=%s", user_id, referrer_id, ) return False finally: conn.close() # ======================================================== # Get User / Auth Context # ======================================================== async def get_user( self, user_id: int, ) -> Optional[dict[str, Any]]: await self._ensure_initialized() return await asyncio.to_thread( self._get_user_sync, int(user_id), ) def _get_user_sync( self, user_id: int, ) -> Optional[dict[str, Any]]: conn = self._connect() try: row = conn.execute( """ SELECT * FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not row: return None return self._user_row_to_dict( row ) finally: conn.close() async def get_user_auth_context( self, user_id: int, ) -> Optional[dict[str, Any]]: user = await self.get_user( user_id ) if user is None: return None kyc_status = ( await self.get_user_kyc_status( user_id ) ) user[ "kyc_status" ] = kyc_status return user async def get_user_role( self, user_id: int, ) -> Optional[dict[str, Any]]: """ Historical method name retained. It returns the rich user dict expected by older wallet handlers, not merely a string. """ return await self.get_user( user_id ) async def set_user_role( self, user_id: int, role: str, ) -> bool: await self._ensure_initialized() role = str( role or "" ).strip().lower() if role not in { "user", "member", "admin", "administrator", "owner", }: return False is_admin = ( 1 if role in { "admin", "administrator", "owner", } else 0 ) return await asyncio.to_thread( self._set_user_admin_sync, int(user_id), is_admin, ) def _set_user_admin_sync( self, user_id: int, is_admin: int, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE users SET is_admin = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( int( bool( is_admin ) ), user_id, ), ) conn.commit() return ( cursor.rowcount == 1 ) except Exception: conn.rollback() logger.exception( "Could not set admin role | user=%s", user_id, ) return False finally: conn.close() async def get_user_tier( self, user_id: int, ) -> str: user = await self.get_user( user_id ) if not user: return "standard" return str( user.get( "user_tier", "standard", ) or "standard" ) async def set_user_tier( self, user_id: int, tier: str, ) -> bool: await self._ensure_initialized() tier = str( tier or "" ).strip().lower() if not tier: return False return await asyncio.to_thread( self._set_user_tier_sync, int(user_id), tier, ) def _set_user_tier_sync( self, user_id: int, tier: str, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE users SET user_tier = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( tier, user_id, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def check_user_ban_status( self, user_id: int, ) -> tuple[bool, Optional[str]]: user = await self.get_user( user_id ) if not user: return ( False, None, ) return ( bool( user.get( "is_banned", False, ) ), None, ) async def set_user_ban_status( self, user_id: int, is_banned: bool, ) -> bool: await self._ensure_initialized() return await asyncio.to_thread( self._set_user_ban_sync, int(user_id), int( bool( is_banned ) ), ) def _set_user_ban_sync( self, user_id: int, is_banned: int, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE users SET is_banned = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( is_banned, user_id, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() # ======================================================== # KYC # ======================================================== async def get_user_kyc_status( self, user_id: int, ) -> str: await self._ensure_initialized() return await asyncio.to_thread( self._get_user_kyc_status_sync, int(user_id), ) async def get_kyc_status( self, user_id: int, ) -> str: return await self.get_user_kyc_status( user_id ) def _get_user_kyc_status_sync( self, user_id: int, ) -> str: conn = self._connect() try: user = conn.execute( """ SELECT is_kyc_verified FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if ( user and int( user[ "is_kyc_verified" ] or 0 ) == 1 ): return "approved" request = conn.execute( """ SELECT status FROM kyc_requests WHERE user_id = ? ORDER BY id DESC LIMIT 1 """, ( user_id, ), ).fetchone() if not request: return "none" status = str( request[ "status" ] or "none" ).strip().lower() aliases = { "verified": "approved", "accepted": "approved", "declined": "rejected", "waiting": "pending", "under_review": "pending", } return aliases.get( status, status, ) finally: conn.close() async def save_kyc_request( self, user_id: int, file_id: str, status: str = "pending", ) -> Optional[int]: await self._ensure_initialized() return await asyncio.to_thread( self._save_kyc_request_sync, int(user_id), str( file_id or "" ).strip(), str( status or "pending" ).strip().lower(), ) def _save_kyc_request_sync( self, user_id: int, file_id: str, status: str, ) -> Optional[int]: if ( user_id <= 0 or not file_id ): return None if status not in { "pending", "approved", "rejected", }: status = "pending" conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) user = conn.execute( """ SELECT user_id FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not user: conn.rollback() return None if status == "pending": existing = conn.execute( """ SELECT id FROM kyc_requests WHERE user_id = ? AND status = 'pending' ORDER BY id DESC LIMIT 1 """, ( user_id, ), ).fetchone() if existing: conn.rollback() return int( existing[ "id" ] ) cursor = conn.execute( """ INSERT INTO kyc_requests ( user_id, file_id, status ) VALUES (?, ?, ?) """, ( user_id, file_id, status, ), ) if status == "approved": verified = 1 else: verified = 0 conn.execute( """ UPDATE users SET is_kyc_verified = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( verified, user_id, ), ) conn.commit() return int( cursor.lastrowid ) except Exception: conn.rollback() logger.exception( "Could not save KYC request | user=%s", user_id, ) return None finally: conn.close() async def set_user_kyc_status( self, user_id: int, status: str, *, reviewed_by: Optional[int] = None, rejection_reason: Optional[str] = None, ) -> bool: await self._ensure_initialized() return await asyncio.to_thread( self._set_user_kyc_status_sync, int(user_id), str( status or "" ).strip().lower(), ( int(reviewed_by) if reviewed_by is not None else None ), rejection_reason, ) def _set_user_kyc_status_sync( self, user_id: int, status: str, reviewed_by: Optional[int], rejection_reason: Optional[str], ) -> bool: if status not in { "approved", "rejected", "pending", }: return False conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) request = conn.execute( """ SELECT id FROM kyc_requests WHERE user_id = ? ORDER BY id DESC LIMIT 1 """, ( user_id, ), ).fetchone() if request: conn.execute( """ UPDATE kyc_requests SET status = ?, reviewed_by = ?, rejection_reason = ?, reviewed_at = CASE WHEN ? IN ( 'approved', 'rejected' ) THEN CURRENT_TIMESTAMP ELSE reviewed_at END, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( status, reviewed_by, rejection_reason, status, int( request[ "id" ] ), ), ) conn.execute( """ UPDATE users SET is_kyc_verified = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( ( 1 if status == "approved" else 0 ), user_id, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Could not update KYC status | user=%s", user_id, ) return False finally: conn.close() # ======================================================== # User Statistics # ======================================================== async def get_user_full_stats( self, user_id: int, ) -> dict[str, Any]: await self._ensure_initialized() return await asyncio.to_thread( self._get_user_full_stats_sync, int(user_id), ) def _get_user_full_stats_sync( self, user_id: int, ) -> dict[str, Any]: conn = self._connect() try: user = conn.execute( """ SELECT created_at, invited_by FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() total_orders = conn.execute( """ SELECT COUNT(*) AS count FROM orders WHERE user_id = ? """, ( user_id, ), ).fetchone()[ "count" ] completed_orders = conn.execute( """ SELECT COUNT(*) AS count FROM orders WHERE user_id = ? AND status = 'completed' """, ( user_id, ), ).fetchone()[ "count" ] referral_count = conn.execute( """ SELECT COUNT( DISTINCT referred_id ) AS count FROM referrals WHERE referrer_id = ? """, ( user_id, ), ).fetchone()[ "count" ] return { "total_orders": int( total_orders or 0 ), "completed_orders": int( completed_orders or 0 ), "referral_count": int( referral_count or 0 ), "joined_date": ( str( user[ "created_at" ] ) if user and user[ "created_at" ] else None ), "invited_by": ( int( user[ "invited_by" ] ) if user and user[ "invited_by" ] is not None else None ), } finally: conn.close() # ======================================================== # Balance # ======================================================== async def get_balance( self, user_id: int, ) -> int: user = await self.get_user( user_id ) if not user: return 0 try: return int( user.get( "balance_toman", 0, ) or 0 ) except ( TypeError, ValueError, ): return 0 async def get_user_balance( self, user_id: int, ) -> int: return await self.get_balance( user_id ) @staticmethod def _reference_already_exists( conn: sqlite3.Connection, reference_id: Optional[str], ) -> bool: if not reference_id: return False row = conn.execute( """ SELECT 1 FROM wallet_transactions WHERE reference_id = ? LIMIT 1 """, ( reference_id, ), ).fetchone() return row is not None async def increase_user_balance( self, user_id: int, amount: int, description: str = "Wallet charge", reference_id: Optional[str] = None, ) -> bool: await self._ensure_initialized() try: amount = int( amount ) except ( TypeError, ValueError, ): return False if amount <= 0: return False return await asyncio.to_thread( self._increase_balance_sync, int(user_id), amount, str( description or "Wallet charge" ), ( str(reference_id) if reference_id else None ), ) def _increase_balance_sync( self, user_id: int, amount: int, description: str, reference_id: Optional[str], ) -> bool: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) if self._reference_already_exists( conn, reference_id, ): conn.rollback() return True row = conn.execute( """ SELECT balance_toman FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not row: conn.rollback() return False before = int( row[ "balance_toman" ] or 0 ) after = ( before + amount ) conn.execute( """ UPDATE users SET balance_toman = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( after, user_id, ), ) conn.execute( """ INSERT INTO wallet_transactions ( user_id, type, amount, balance_before, balance_after, description, reference_id ) VALUES ( ?, 'credit', ?, ?, ?, ?, ? ) """, ( user_id, amount, before, after, description, reference_id, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Could not increase balance | " "user=%s amount=%s", user_id, amount, ) return False finally: conn.close() async def decrease_user_balance( self, user_id: int, amount: int, description: str = "Order payment", reference_id: Optional[str] = None, ) -> bool: await self._ensure_initialized() try: amount = int( amount ) except ( TypeError, ValueError, ): return False if amount <= 0: return False return await asyncio.to_thread( self._decrease_balance_sync, int(user_id), amount, str( description or "Order payment" ), ( str(reference_id) if reference_id else None ), ) def _decrease_balance_sync( self, user_id: int, amount: int, description: str, reference_id: Optional[str], ) -> bool: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) if self._reference_already_exists( conn, reference_id, ): conn.rollback() return True row = conn.execute( """ SELECT balance_toman FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not row: conn.rollback() return False before = int( row[ "balance_toman" ] or 0 ) if before < amount: conn.rollback() return False after = ( before - amount ) conn.execute( """ UPDATE users SET balance_toman = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( after, user_id, ), ) conn.execute( """ INSERT INTO wallet_transactions ( user_id, type, amount, balance_before, balance_after, description, reference_id ) VALUES ( ?, 'debit', ?, ?, ?, ?, ? ) """, ( user_id, amount, before, after, description, reference_id, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Could not decrease balance | " "user=%s amount=%s", user_id, amount, ) return False finally: conn.close() # ======================================================== # Atomic Legacy Order + Payment # ======================================================== async def create_paid_order( self, user_id: int, product_id: str, target: Optional[str], price: int, gateway: str = "wallet", ) -> Optional[int]: """ Legacy compatibility only. New Premium/Stars workflow should use OrderService and pending_owner_approval instead. """ await self._ensure_initialized() try: price = int( price ) except ( TypeError, ValueError, ): return None if price <= 0: return None return await asyncio.to_thread( self._create_paid_order_sync, int(user_id), str( product_id or "" ), ( str(target) if target is not None else None ), price, str( gateway or "wallet" ), ) def _create_paid_order_sync( self, user_id: int, product_id: str, target: Optional[str], price: int, gateway: str, ) -> Optional[int]: if not product_id: return None conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) user = conn.execute( """ SELECT balance_toman FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not user: conn.rollback() return None balance = int( user[ "balance_toman" ] or 0 ) if balance < price: conn.rollback() return None new_balance = ( balance - price ) conn.execute( """ UPDATE users SET balance_toman = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( new_balance, user_id, ), ) cursor = conn.execute( """ INSERT INTO orders ( user_id, product_id, target, price, status, gateway ) VALUES ( ?, ?, ?, ?, 'processing', ? ) """, ( user_id, product_id, target, price, gateway, ), ) order_id = int( cursor.lastrowid ) conn.execute( """ INSERT INTO wallet_transactions ( user_id, type, amount, balance_before, balance_after, description, reference_id ) VALUES ( ?, 'debit', ?, ?, ?, 'Order payment', ? ) """, ( user_id, price, balance, new_balance, str( order_id ), ), ) conn.commit() return order_id except Exception: conn.rollback() logger.exception( "Atomic legacy order/payment failed | " "user=%s product=%s", user_id, product_id, ) return None finally: conn.close() # ======================================================== # Refund # ======================================================== async def refund_order( self, order_id: int, description: str = "Order refund", ) -> bool: """ Manual/legacy refund. provider_status_unknown is intentionally NOT refundable here. Ambiguous provider outcomes must be reconciled manually first. """ await self._ensure_initialized() return await asyncio.to_thread( self._refund_order_sync, int(order_id), str( description or "Order refund" ), ) def _refund_order_sync( self, order_id: int, description: str, ) -> bool: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) order = conn.execute( """ SELECT user_id, price, status FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() if not order: conn.rollback() return False status = str( order[ "status" ] or "" ).lower() if status in { "refunded", "failed_refunded", }: conn.rollback() return True if status not in { "processing", "failed", "failed_refund_required", }: conn.rollback() return False user_id = int( order[ "user_id" ] ) amount = int( order[ "price" ] or 0 ) user = conn.execute( """ SELECT balance_toman FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not user: conn.rollback() return False reference_id = ( f"order:{order_id}:legacy_refund" ) if self._reference_already_exists( conn, reference_id, ): conn.rollback() return True before = int( user[ "balance_toman" ] or 0 ) after = ( before + amount ) conn.execute( """ UPDATE users SET balance_toman = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( after, user_id, ), ) conn.execute( """ UPDATE orders SET status = 'refunded', refunded_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( order_id, ), ) conn.execute( """ INSERT INTO wallet_transactions ( user_id, type, amount, balance_before, balance_after, description, reference_id ) VALUES ( ?, 'credit', ?, ?, ?, ?, ? ) """, ( user_id, amount, before, after, description, reference_id, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Refund failed | order=%s", order_id, ) return False finally: conn.close() # ======================================================== # Orders # ======================================================== async def save_order( self, user_id: int, product_id: str, target: Optional[str], price: int, status: str = "pending", gateway: str = "wallet", market_order_id: Optional[str] = None, ) -> Optional[int]: await self._ensure_initialized() try: price = int( price ) except ( TypeError, ValueError, ): return None if price < 0: return None status = str( status or "pending" ).strip() if status not in self.VALID_ORDER_STATUSES: return None return await asyncio.to_thread( self._save_order_sync, int(user_id), str( product_id or "" ).strip(), ( str(target) if target is not None else None ), price, status, str( gateway or "wallet" ), ( str(market_order_id) if market_order_id else None ), ) def _save_order_sync( self, user_id: int, product_id: str, target: Optional[str], price: int, status: str, gateway: str, market_order_id: Optional[str], ) -> Optional[int]: if not product_id: return None conn = self._connect() try: cursor = conn.execute( """ INSERT INTO orders ( user_id, product_id, target, price, status, gateway, market_order_id ) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( user_id, product_id, target, price, status, gateway, market_order_id, ), ) conn.commit() return int( cursor.lastrowid ) except Exception: conn.rollback() logger.exception( "Could not save order." ) return None finally: conn.close() async def get_order( self, order_id: int, ) -> Optional[dict[str, Any]]: await self._ensure_initialized() return await asyncio.to_thread( self._get_order_sync, int(order_id), ) def _get_order_sync( self, order_id: int, ) -> Optional[dict[str, Any]]: conn = self._connect() try: row = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() return ( dict( row ) if row else None ) finally: conn.close() async def get_user_orders( self, user_id: int, limit: int = 20, ) -> list[dict[str, Any]]: await self._ensure_initialized() try: limit = max( 1, min( int( limit ), 200, ), ) except ( TypeError, ValueError, ): limit = 20 return await asyncio.to_thread( self._get_user_orders_sync, int(user_id), limit, ) def _get_user_orders_sync( self, user_id: int, limit: int, ) -> list[dict[str, Any]]: conn = self._connect() try: rows = conn.execute( """ SELECT * FROM orders WHERE user_id = ? ORDER BY id DESC LIMIT ? """, ( user_id, limit, ), ).fetchall() return [ dict( row ) for row in rows ] finally: conn.close() async def update_order_status( self, order_id: int, status: str, market_order_id: Optional[str] = None, ) -> bool: await self._ensure_initialized() status = str( status or "" ).strip() if status not in self.VALID_ORDER_STATUSES: return False return await asyncio.to_thread( self._update_order_status_sync, int(order_id), status, ( str(market_order_id) if market_order_id else None ), ) def _update_order_status_sync( self, order_id: int, status: str, market_order_id: Optional[str], ) -> bool: conn = self._connect() try: if market_order_id is None: cursor = conn.execute( """ UPDATE orders SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( status, order_id, ), ) else: cursor = conn.execute( """ UPDATE orders SET status = ?, market_order_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( status, market_order_id, order_id, ), ) conn.commit() return ( cursor.rowcount == 1 ) except Exception: conn.rollback() logger.exception( "Could not update order %s", order_id, ) return False finally: conn.close() # ======================================================== # Wallet Transactions # ======================================================== async def get_wallet_transactions( self, user_id: int, limit: int = 20, ) -> list[dict[str, Any]]: await self._ensure_initialized() try: limit = max( 1, min( int( limit ), 200, ), ) except ( TypeError, ValueError, ): limit = 20 return await asyncio.to_thread( self._get_wallet_transactions_sync, int(user_id), limit, ) def _get_wallet_transactions_sync( self, user_id: int, limit: int, ) -> list[dict[str, Any]]: conn = self._connect() try: rows = conn.execute( """ SELECT * FROM wallet_transactions WHERE user_id = ? ORDER BY id DESC LIMIT ? """, ( user_id, limit, ), ).fetchall() return [ dict( row ) for row in rows ] finally: conn.close() # ======================================================== # Admin Reads # ======================================================== async def get_all_users( self, limit: int = 100, ) -> list[dict[str, Any]]: await self._ensure_initialized() try: limit = max( 1, min( int( limit ), 1000, ), ) except ( TypeError, ValueError, ): limit = 100 return await asyncio.to_thread( self._get_all_users_sync, limit, ) def _get_all_users_sync( self, limit: int, ) -> list[dict[str, Any]]: conn = self._connect() try: rows = conn.execute( """ SELECT * FROM users ORDER BY user_id DESC LIMIT ? """, ( limit, ), ).fetchall() return [ self._user_row_to_dict( row ) for row in rows ] finally: conn.close() async def get_all_orders( self, limit: int = 100, ) -> list[dict[str, Any]]: await self._ensure_initialized() try: limit = max( 1, min( int( limit ), 1000, ), ) except ( TypeError, ValueError, ): limit = 100 return await asyncio.to_thread( self._get_all_orders_sync, limit, ) def _get_all_orders_sync( self, limit: int, ) -> list[dict[str, Any]]: conn = self._connect() try: rows = conn.execute( """ SELECT * FROM orders ORDER BY id DESC LIMIT ? """, ( limit, ), ).fetchall() return [ dict( row ) for row in rows ] finally: conn.close() # ======================================================== # Statistics # ======================================================== async def get_statistics( self, ) -> dict[str, Any]: await self._ensure_initialized() return await asyncio.to_thread( self._get_statistics_sync ) def _get_statistics_sync( self, ) -> dict[str, Any]: conn = self._connect() try: users = conn.execute( """ SELECT COUNT(*) AS count FROM users """ ).fetchone()[ "count" ] orders = conn.execute( """ SELECT COUNT(*) AS count FROM orders """ ).fetchone()[ "count" ] completed = conn.execute( """ SELECT COUNT(*) AS count FROM orders WHERE status = 'completed' """ ).fetchone()[ "count" ] revenue = conn.execute( """ SELECT COALESCE( SUM(price), 0 ) AS total FROM orders WHERE status = 'completed' """ ).fetchone()[ "total" ] wallet = conn.execute( """ SELECT COALESCE( SUM(balance_toman), 0 ) AS total FROM users """ ).fetchone()[ "total" ] pending_owner = conn.execute( """ SELECT COUNT(*) AS count FROM orders WHERE status = 'pending_owner_approval' """ ).fetchone()[ "count" ] provider_unknown = conn.execute( """ SELECT COUNT(*) AS count FROM orders WHERE status = 'provider_status_unknown' """ ).fetchone()[ "count" ] return { "users": int( users or 0 ), "orders": int( orders or 0 ), "completed_orders": int( completed or 0 ), "revenue_toman": int( revenue or 0 ), "wallet_balance_toman": int( wallet or 0 ), "pending_owner_orders": int( pending_owner or 0 ), "provider_unknown_orders": int( provider_unknown or 0 ), } finally: conn.close() # ======================================================== # Audit # ======================================================== async def add_audit_log( self, *, user_id: Optional[int], action: str, details: Optional[str] = None, ip_address: Optional[str] = None, ) -> bool: await self._ensure_initialized() return await asyncio.to_thread( self._add_audit_log_sync, ( int(user_id) if user_id is not None else None ), str( action or "" ).strip(), ( str(details) if details is not None else None ), ( str(ip_address) if ip_address is not None else None ), ) def _add_audit_log_sync( self, user_id: Optional[int], action: str, details: Optional[str], ip_address: Optional[str], ) -> bool: if not action: return False conn = self._connect() try: conn.execute( """ INSERT INTO audit_logs ( user_id, action, details, ip_address ) VALUES (?, ?, ?, ?) """, ( user_id, action, details, ip_address, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Could not add audit log | action=%s", action, ) return False finally: conn.close() # ======================================================== # Health / Close # ======================================================== async def health_check( self, ) -> bool: try: await self._ensure_initialized() return await asyncio.to_thread( self._health_check_sync ) except Exception: logger.exception( "Database health check failed." ) return False def _health_check_sync( self, ) -> bool: conn = self._connect() try: row = conn.execute( "SELECT 1 AS ok" ).fetchone() # Also validate the canonical user key exists. columns = self._column_names( conn, "users", ) return bool( row and row[ "ok" ] == 1 and "user_id" in columns ) finally: conn.close() async def close( self, ) -> None: # Connections are short-lived, so there is no open shared # SQLite handle to close. self._initialized = False logger.info( "Database manager closed." ) __all__ = [ "DatabaseManager", ]