# services/order_service.py from __future__ import annotations import asyncio import logging import sqlite3 from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Iterable, Optional logger = logging.getLogger(__name__) # ============================================================ # Statuses # ============================================================ STATUS_PENDING_OWNER = "pending_owner_approval" STATUS_OWNER_PROCESSING = "owner_processing" STATUS_REJECTED_BY_OWNER = "rejected_by_owner" STATUS_CANCELLED_BY_USER = "cancelled_by_user" STATUS_PRICE_CHANGED = "price_changed" STATUS_INSUFFICIENT_BALANCE = "insufficient_balance" STATUS_PROVIDER_PROCESSING = "provider_processing" STATUS_PROVIDER_UNKNOWN = "provider_status_unknown" STATUS_COMPLETED = "completed" STATUS_FAILED_REFUNDED = "failed_refunded" STATUS_FAILED_REFUND_REQUIRED = "failed_refund_required" # Legacy statuses are retained for compatibility with old rows/handlers. LEGACY_STATUSES = { "pending", "processing", "paid", "failed", "cancelled", "refunded", "created", } VALID_STATUSES = { STATUS_PENDING_OWNER, STATUS_OWNER_PROCESSING, STATUS_REJECTED_BY_OWNER, STATUS_CANCELLED_BY_USER, STATUS_PRICE_CHANGED, STATUS_INSUFFICIENT_BALANCE, STATUS_PROVIDER_PROCESSING, STATUS_PROVIDER_UNKNOWN, STATUS_COMPLETED, STATUS_FAILED_REFUNDED, STATUS_FAILED_REFUND_REQUIRED, *LEGACY_STATUSES, } # ============================================================ # Models # ============================================================ @dataclass(slots=True) class Order: id: int user_id: int product_id: str target: Optional[str] price: int status: str gateway: str market_order_id: Optional[str] created_at: str updated_at: str owner_id: Optional[int] = None provider_required_price: Optional[int] = None provider_error: Optional[str] = None provider_started_at: Optional[str] = None completed_at: Optional[str] = None refunded_at: Optional[str] = None def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass(slots=True) class WalletReservationResult: success: bool reason: str order_id: Optional[int] = None user_id: Optional[int] = None amount: int = 0 balance_before: int = 0 balance_after: int = 0 @dataclass(slots=True) class PendingOrderCreateResult: success: bool reason: str order: Optional[Order] = None user_id: Optional[int] = None required_price: int = 0 balance: int = 0 # ============================================================ # Exceptions # ============================================================ class OrderServiceError(Exception): pass class OrderNotFoundError(OrderServiceError): pass class InvalidOrderError(OrderServiceError): pass class InvalidOrderStatusError(OrderServiceError): pass class WalletError(OrderServiceError): pass # ============================================================ # Order Service # ============================================================ class OrderService: """ Order service for the main bot database. Database contract used by the current project: users.user_id users.balance_toman (preferred) users.balance (legacy fallback) orders.id INTEGER PRIMARY KEY AUTOINCREMENT orders.user_id -> users.user_id Purchase flow: pending_owner_approval -> owner_processing -> provider_processing -> completed Alternative terminal/intermediate states: rejected_by_owner cancelled_by_user price_changed insufficient_balance failed_refunded failed_refund_required provider_status_unknown """ def __init__( self, db_path: str | Path = "matrix_bot.db", ) -> None: self.db_path = str(db_path) if not self.db_path.strip(): raise ValueError( "db_path cannot be empty." ) # ======================================================== # SQLite # ======================================================== 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 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 _wallet_column( cls, conn: sqlite3.Connection, ) -> Optional[str]: columns = cls._column_names( conn, "users", ) if "balance_toman" in columns: return "balance_toman" if "balance" in columns: return "balance" return None async def _run( self, func, *args, ): return await asyncio.to_thread( func, *args, ) # ======================================================== # Schema # ======================================================== async def ensure_schema( self, ) -> None: """ Verify/repair the orders table. Repairs the known historical FK bug: orders.user_id -> users.id to: orders.user_id -> users.user_id The migration is performed with foreign_keys disabled, inside BEGIN IMMEDIATE, and preserves known order columns. """ await self._run( self._ensure_schema_sync ) def _ensure_schema_sync( self, ) -> None: # Foreign keys must be OFF before BEGIN for table repair. conn = self._connect( foreign_keys=False ) try: conn.execute( "BEGIN IMMEDIATE" ) if not self._table_exists( conn, "users", ): raise OrderServiceError( "users table does not exist. " "Initialize DatabaseManager first." ) user_columns = self._column_names( conn, "users", ) if "user_id" not in user_columns: raise OrderServiceError( "users.user_id is missing." ) if not self._table_exists( conn, "orders", ): self._create_orders_table( conn ) self._create_indexes( conn ) conn.commit() logger.info( "orders table created." ) return order_columns = ( self._column_names( conn, "orders", ) ) required_columns = { "id", "user_id", "product_id", "target", "price", "status", "gateway", "market_order_id", "created_at", "updated_at", } fk_rows = conn.execute( "PRAGMA foreign_key_list(orders)" ).fetchall() has_correct_user_fk = any( str(row["table"]) == "users" and str(row["from"]) == "user_id" and str(row["to"]) == "user_id" for row in fk_rows ) # If the basic schema + FK is already good, # only add optional owner-approval columns. if ( required_columns.issubset( order_columns ) and has_correct_user_fk ): self._add_optional_columns( conn, order_columns, ) self._create_indexes( conn ) conn.commit() return # ------------------------------------------------ # Repair current orders table # ------------------------------------------------ legacy_name = ( "orders_legacy_order_service" ) if self._table_exists( conn, legacy_name, ): # A previous failed migration left a temp table. raise OrderServiceError( f"{legacy_name} already exists. " "Manual inspection is required before migration." ) conn.execute( ( "ALTER TABLE orders " f"RENAME TO {legacy_name}" ) ) self._create_orders_table( conn ) old_columns = ( self._column_names( conn, legacy_name, ) ) 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", "created_at", "updated_at", "owner_id", "provider_required_price", "provider_error", "provider_started_at", "completed_at", "refunded_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_name} """ ) conn.execute( f"DROP TABLE {legacy_name}" ) self._create_indexes( conn ) conn.commit() logger.info( "orders schema repaired | " "db=%s", self.db_path, ) except Exception: try: conn.rollback() except Exception: pass logger.exception( "Could not initialize/repair orders schema." ) raise finally: conn.close() @staticmethod def _create_orders_table( 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 ) """ ) @staticmethod def _add_optional_columns( conn: sqlite3.Connection, columns: set[str], ) -> None: 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}" ) ) @staticmethod def _create_indexes( conn: sqlite3.Connection, ) -> None: conn.execute( """ CREATE INDEX IF NOT EXISTS idx_orders_user_id ON orders(user_id) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_orders_user_status ON orders(user_id, status) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_orders_market_order_id ON orders(market_order_id) """ ) # ======================================================== # Row Conversion # ======================================================== @staticmethod def _row_to_order( row: sqlite3.Row, ) -> Order: keys = set( row.keys() ) def optional_int( key: str, ) -> Optional[int]: if ( key not in keys or row[key] is None ): return None return int( row[key] ) def optional_str( key: str, ) -> Optional[str]: if ( key not in keys or row[key] is None ): return None return str( row[key] ) return Order( id=int( row["id"] ), user_id=int( row["user_id"] ), product_id=str( row["product_id"] ), target=( optional_str( "target" ) ), price=int( row["price"] or 0 ), status=str( row["status"] or "" ), gateway=str( row["gateway"] or "wallet" ), market_order_id=( optional_str( "market_order_id" ) ), created_at=str( row["created_at"] or "" ), updated_at=str( row["updated_at"] or "" ), owner_id=( optional_int( "owner_id" ) ), provider_required_price=( optional_int( "provider_required_price" ) ), provider_error=( optional_str( "provider_error" ) ), provider_started_at=( optional_str( "provider_started_at" ) ), completed_at=( optional_str( "completed_at" ) ), refunded_at=( optional_str( "refunded_at" ) ), ) # ======================================================== # Read # ======================================================== async def get_order( self, order_id: int, *, user_id: Optional[int] = None, ) -> Order: await self.ensure_schema() return await self._run( self._get_order_sync, int(order_id), ( int(user_id) if user_id is not None else None ), ) def _get_order_sync( self, order_id: int, user_id: Optional[int], ) -> Order: conn = self._connect() try: query = ( "SELECT * " "FROM orders " "WHERE id = ?" ) params: list[Any] = [ order_id ] if user_id is not None: query += ( " AND user_id = ?" ) params.append( user_id ) query += " LIMIT 1" row = conn.execute( query, tuple(params), ).fetchone() if not row: raise OrderNotFoundError( f"Order {order_id} not found." ) return self._row_to_order( row ) finally: conn.close() async def list_orders( self, user_id: int, *, status: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> list[Order]: await self.ensure_schema() return await self._run( self._list_orders_sync, int(user_id), status, int(limit), int(offset), ) def _list_orders_sync( self, user_id: int, status: Optional[str], limit: int, offset: int, ) -> list[Order]: limit = max( 1, min( limit, 100, ), ) offset = max( 0, offset, ) conn = self._connect() try: query = ( "SELECT * " "FROM orders " "WHERE user_id = ?" ) params: list[Any] = [ user_id ] if status is not None: query += ( " AND status = ?" ) params.append( str(status) ) query += ( " ORDER BY id DESC " "LIMIT ? OFFSET ?" ) params.extend( [ limit, offset, ] ) rows = conn.execute( query, tuple(params), ).fetchall() return [ self._row_to_order( row ) for row in rows ] finally: conn.close() async def list_by_status( self, status: str, *, limit: int = 100, ) -> list[Order]: await self.ensure_schema() return await self._run( self._list_by_status_sync, str(status), int(limit), ) def _list_by_status_sync( self, status: str, limit: int, ) -> list[Order]: limit = max( 1, min( limit, 500, ), ) conn = self._connect() try: rows = conn.execute( """ SELECT * FROM orders WHERE status = ? ORDER BY id ASC LIMIT ? """, ( status, limit, ), ).fetchall() return [ self._row_to_order( row ) for row in rows ] finally: conn.close() # ======================================================== # Create # ======================================================== async def create_order( self, *, user_id: int, product_id: str, target: Optional[str], price: int, status: str = STATUS_PENDING_OWNER, gateway: str = "wallet", ) -> Order: await self.ensure_schema() if int(user_id) <= 0: raise InvalidOrderError( "user_id must be positive." ) product_id = str( product_id or "" ).strip() if not product_id: raise InvalidOrderError( "product_id is required." ) price = int( price ) if price < 0: raise InvalidOrderError( "price cannot be negative." ) status = str( status ).strip() if status not in VALID_STATUSES: raise InvalidOrderStatusError( f"Unsupported status: {status}" ) gateway = str( gateway or "wallet" ).strip() if not gateway: gateway = "wallet" target_value = ( str(target).strip() if target is not None else None ) return await self._run( self._create_order_sync, int(user_id), product_id, target_value, price, status, gateway, ) def _create_order_sync( self, user_id: int, product_id: str, target: Optional[str], price: int, status: str, gateway: str, ) -> Order: conn = self._connect() try: cursor = conn.execute( """ INSERT INTO orders ( user_id, product_id, target, price, status, gateway ) VALUES (?, ?, ?, ?, ?, ?) """, ( user_id, product_id, target, price, status, gateway, ), ) order_id = int( cursor.lastrowid ) conn.commit() finally: conn.close() return self._get_order_sync( order_id, None, ) async def create_pending_owner_order( self, *, user_id: int, product_id: str, target: str, price: int, ) -> Order: return await self.create_order( user_id=user_id, product_id=product_id, target=target, price=price, status=( STATUS_PENDING_OWNER ), gateway="wallet", ) async def create_pending_owner_order_if_balance_sufficient( self, *, user_id: int, product_id: str, target: str, price: int, ) -> PendingOrderCreateResult: """ Atomically checks wallet balance and creates the pending owner order. IMPORTANT: - Does NOT debit the wallet. - If balance < price, NO order row is created. - This prevents an insufficient-balance request from reaching admins. """ await self.ensure_schema() user_id = int( user_id ) product_id = str( product_id or "" ).strip() target = str( target or "" ).strip() price = int( price ) if user_id <= 0: raise InvalidOrderError( "user_id must be positive." ) if not product_id: raise InvalidOrderError( "product_id is required." ) if not target: raise InvalidOrderError( "target is required." ) if price <= 0: raise InvalidOrderError( "price must be positive." ) return await self._run( self._create_pending_owner_order_if_balance_sufficient_sync, user_id, product_id, target, price, ) def _create_pending_owner_order_if_balance_sufficient_sync( self, user_id: int, product_id: str, target: str, price: int, ) -> PendingOrderCreateResult: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) wallet_column = self._wallet_column( conn ) if not wallet_column: conn.rollback() return PendingOrderCreateResult( success=False, reason="wallet_column_missing", user_id=user_id, required_price=price, balance=0, ) user_row = conn.execute( f""" SELECT COALESCE({wallet_column}, 0) AS balance FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not user_row: conn.rollback() return PendingOrderCreateResult( success=False, reason="user_not_found", user_id=user_id, required_price=price, balance=0, ) balance = max( 0, int( user_row["balance"] or 0 ), ) if balance < price: conn.rollback() return PendingOrderCreateResult( success=False, reason="insufficient_balance", user_id=user_id, required_price=price, balance=balance, ) cursor = conn.execute( """ INSERT INTO orders ( user_id, product_id, target, price, status, gateway ) VALUES (?, ?, ?, ?, ?, ?) """, ( user_id, product_id, target, price, STATUS_PENDING_OWNER, "wallet", ), ) order_id = int( cursor.lastrowid ) row = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() if not row: conn.rollback() return PendingOrderCreateResult( success=False, reason="order_insert_failed", user_id=user_id, required_price=price, balance=balance, ) order = self._row_to_order( row ) conn.commit() return PendingOrderCreateResult( success=True, reason="created", order=order, user_id=user_id, required_price=price, balance=balance, ) except Exception: conn.rollback() raise finally: conn.close() # ======================================================== # Owner Approval # ======================================================== async def claim_for_owner( self, order_id: int, owner_id: int, ) -> Optional[Order]: """ Atomic claim. Returns None if another owner already handled the order. """ await self.ensure_schema() return await self._run( self._claim_for_owner_sync, int(order_id), int(owner_id), ) def _claim_for_owner_sync( self, order_id: int, owner_id: int, ) -> Optional[Order]: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) cursor = conn.execute( """ UPDATE orders SET status = ?, owner_id = ?, provider_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_OWNER_PROCESSING, owner_id, order_id, STATUS_PENDING_OWNER, ), ) if cursor.rowcount != 1: conn.rollback() return None row = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() conn.commit() return ( self._row_to_order( row ) if row else None ) except Exception: conn.rollback() raise finally: conn.close() async def reject_by_owner( self, order_id: int, owner_id: int, ) -> bool: await self.ensure_schema() return await self._run( self._reject_by_owner_sync, int(order_id), int(owner_id), ) def _reject_by_owner_sync( self, order_id: int, owner_id: int, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, owner_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_REJECTED_BY_OWNER, owner_id, order_id, STATUS_PENDING_OWNER, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def restore_owner_pending( self, order_id: int, *, error: Optional[str] = None, ) -> bool: await self.ensure_schema() return await self._run( self._restore_owner_pending_sync, int(order_id), ( str(error)[:1000] if error else None ), ) def _restore_owner_pending_sync( self, order_id: int, error: Optional[str], ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PENDING_OWNER, error, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def mark_price_changed( self, order_id: int, *, required_price: int, ) -> bool: await self.ensure_schema() return await self._run( self._mark_price_changed_sync, int(order_id), int(required_price), ) def _mark_price_changed_sync( self, order_id: int, required_price: int, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, provider_required_price = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PRICE_CHANGED, required_price, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def accept_price_change( self, order_id: int, *, user_id: int, ) -> bool: """ User accepts provider_required_price after a price increase. Atomic: price_changed -> price = provider_required_price -> pending_owner_approval No wallet debit occurs here. """ await self.ensure_schema() return await self._run( self._accept_price_change_sync, int(order_id), int(user_id), ) def _accept_price_change_sync( self, order_id: int, user_id: int, ) -> bool: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) row = conn.execute( """ SELECT status, user_id, provider_required_price FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() if not row: conn.rollback() return False if ( str( row["status"] ) != STATUS_PRICE_CHANGED or int( row["user_id"] ) != user_id ): conn.rollback() return False required_price = int( row[ "provider_required_price" ] or 0 ) if required_price <= 0: conn.rollback() return False cursor = conn.execute( """ UPDATE orders SET price = ?, status = ?, owner_id = NULL, provider_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? AND user_id = ? """, ( required_price, STATUS_PENDING_OWNER, order_id, STATUS_PRICE_CHANGED, user_id, ), ) conn.commit() return ( cursor.rowcount == 1 ) except Exception: conn.rollback() raise finally: conn.close() async def cancel_price_change( self, order_id: int, *, user_id: int, ) -> bool: """ User rejects the increased price. No wallet debit can have happened while status=price_changed. """ await self.ensure_schema() return await self._run( self._cancel_price_change_sync, int(order_id), int(user_id), ) def _cancel_price_change_sync( self, order_id: int, user_id: int, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? AND user_id = ? """, ( STATUS_CANCELLED_BY_USER, "User rejected updated price.", order_id, STATUS_PRICE_CHANGED, user_id, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() # ======================================================== # Final MarketApp Price # ======================================================== async def set_owner_final_price( self, order_id: int, *, final_price: int, ) -> bool: """ Persist the exact converted MarketApp price after owner approval. This is allowed only while the order is owner_processing and before any wallet debit/provider buy starts. """ await self.ensure_schema() final_price = int(final_price) if final_price <= 0: raise InvalidOrderError( "final_price must be positive." ) return await self._run( self._set_owner_final_price_sync, int(order_id), final_price, ) def _set_owner_final_price_sync( self, order_id: int, final_price: int, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE orders SET price = ?, provider_required_price = ?, provider_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( final_price, final_price, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return cursor.rowcount == 1 except Exception: conn.rollback() raise finally: conn.close() # ======================================================== # Wallet Reserve # ======================================================== async def reserve_wallet_for_provider( self, order_id: int, ) -> WalletReservationResult: """ Atomic: owner_processing + balance check + balance debit -> provider_processing """ await self.ensure_schema() return await self._run( self._reserve_wallet_sync, int(order_id), ) def _reserve_wallet_sync( self, order_id: int, ) -> WalletReservationResult: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) order = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() if not order: conn.rollback() return WalletReservationResult( success=False, reason="order_not_found", order_id=order_id, ) if str( order["status"] ) != STATUS_OWNER_PROCESSING: conn.rollback() return WalletReservationResult( success=False, reason="invalid_order_status", order_id=order_id, ) wallet_column = ( self._wallet_column( conn ) ) if not wallet_column: conn.rollback() return WalletReservationResult( success=False, reason="wallet_column_missing", order_id=order_id, ) user_id = int( order["user_id"] ) amount = int( order["price"] or 0 ) user = conn.execute( f""" SELECT {wallet_column} AS balance FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not user: conn.rollback() return WalletReservationResult( success=False, reason="user_not_found", order_id=order_id, user_id=user_id, amount=amount, ) before = int( user["balance"] or 0 ) if before < amount: conn.execute( """ UPDATE orders SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_INSUFFICIENT_BALANCE, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return WalletReservationResult( success=False, reason="insufficient_balance", order_id=order_id, user_id=user_id, amount=amount, balance_before=before, balance_after=before, ) after = ( before - amount ) conn.execute( f""" UPDATE users SET {wallet_column} = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( after, user_id, ), ) self._insert_wallet_transaction( conn, user_id=user_id, tx_type="order_reserve", amount=-amount, balance_before=before, balance_after=after, description=( f"رزرو موجودی سفارش #{order_id}" ), reference_id=( f"order:{order_id}:reserve" ), ) conn.execute( """ UPDATE orders SET status = ?, provider_started_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PROVIDER_PROCESSING, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return WalletReservationResult( success=True, reason="reserved", order_id=order_id, user_id=user_id, amount=amount, balance_before=before, balance_after=after, ) except Exception: conn.rollback() logger.exception( "Wallet reserve failed | " "order=%s", order_id, ) return WalletReservationResult( success=False, reason="database_error", order_id=order_id, ) finally: conn.close() # ======================================================== # Provider Outcome # ======================================================== async def mark_provider_unknown( self, order_id: int, *, error: Optional[str] = None, ) -> bool: await self.ensure_schema() return await self._run( self._mark_provider_unknown_sync, int(order_id), ( str(error)[:1000] if error else None ), ) def _mark_provider_unknown_sync( self, order_id: int, error: Optional[str], ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PROVIDER_UNKNOWN, error, order_id, STATUS_PROVIDER_PROCESSING, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def complete( self, order_id: int, *, market_order_id: str, ) -> bool: await self.ensure_schema() market_order_id = str( market_order_id or "" ).strip() if not market_order_id: raise InvalidOrderError( "market_order_id is required." ) return await self._run( self._complete_sync, int(order_id), market_order_id, ) def _complete_sync( self, order_id: int, market_order_id: str, ) -> bool: conn = self._connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, market_order_id = ?, completed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status IN (?, ?) """, ( STATUS_COMPLETED, market_order_id, order_id, STATUS_PROVIDER_PROCESSING, STATUS_PROVIDER_UNKNOWN, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def refund_explicit_provider_failure( self, order_id: int, *, error: Optional[str] = None, ) -> bool: """ Refund only for an explicit provider/business failure. Do NOT call for timeout/connection/5xx ambiguity. """ await self.ensure_schema() return await self._run( self._refund_explicit_failure_sync, int(order_id), ( str(error)[:1000] if error else None ), ) def _refund_explicit_failure_sync( self, order_id: int, error: Optional[str], ) -> bool: conn = self._connect() try: conn.execute( "BEGIN IMMEDIATE" ) order = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() if not order: conn.rollback() return False if str( order["status"] ) != STATUS_PROVIDER_PROCESSING: conn.rollback() return False wallet_column = ( self._wallet_column( conn ) ) if not wallet_column: conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( STATUS_FAILED_REFUND_REQUIRED, error, order_id, ), ) conn.commit() return False user_id = int( order["user_id"] ) amount = int( order["price"] or 0 ) user = conn.execute( f""" SELECT {wallet_column} AS balance FROM users WHERE user_id = ? LIMIT 1 """, ( user_id, ), ).fetchone() if not user: conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( STATUS_FAILED_REFUND_REQUIRED, error, order_id, ), ) conn.commit() return False before = int( user["balance"] or 0 ) after = ( before + amount ) conn.execute( f""" UPDATE users SET {wallet_column} = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( after, user_id, ), ) self._insert_wallet_transaction( conn, user_id=user_id, tx_type="order_refund", amount=amount, balance_before=before, balance_after=after, description=( f"بازگشت وجه سفارش #{order_id}" ), reference_id=( f"order:{order_id}:refund" ), ) conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, refunded_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_FAILED_REFUNDED, error, order_id, STATUS_PROVIDER_PROCESSING, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Provider failure refund failed | " "order=%s", order_id, ) return False finally: conn.close() # ======================================================== # Generic Compatibility Update # ======================================================== async def update_order_status( self, order_id: int, status: str, *, market_order_id: Optional[str] = None, ) -> Order: await self.ensure_schema() status = str( status ).strip() if status not in VALID_STATUSES: raise InvalidOrderStatusError( f"Unsupported status: {status}" ) return await self._run( self._update_status_sync, int(order_id), status, market_order_id, ) def _update_status_sync( self, order_id: int, status: str, market_order_id: Optional[str], ) -> Order: conn = self._connect() try: current = conn.execute( """ SELECT id FROM orders WHERE id = ? LIMIT 1 """, ( order_id, ), ).fetchone() if not current: raise OrderNotFoundError( f"Order {order_id} not found." ) if market_order_id is None: conn.execute( """ UPDATE orders SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( status, order_id, ), ) else: conn.execute( """ UPDATE orders SET status = ?, market_order_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( status, str( market_order_id ), order_id, ), ) conn.commit() finally: conn.close() return self._get_order_sync( order_id, None, ) # ======================================================== # Wallet Transaction # ======================================================== @classmethod def _insert_wallet_transaction( cls, conn: sqlite3.Connection, *, user_id: int, tx_type: str, amount: int, balance_before: int, balance_after: int, description: str, reference_id: str, ) -> None: if not cls._table_exists( conn, "wallet_transactions", ): return columns = cls._column_names( conn, "wallet_transactions", ) required = { "user_id", "type", "amount", "balance_before", "balance_after", } if not required.issubset( columns ): return insert_columns = [ "user_id", "type", "amount", "balance_before", "balance_after", ] values: list[Any] = [ user_id, tx_type, amount, balance_before, balance_after, ] if "description" in columns: insert_columns.append( "description" ) values.append( description ) if "reference_id" in columns: # Prevent duplicate bookkeeping when a unique index # exists in a future schema. The order state itself # is still the primary idempotency guard. insert_columns.append( "reference_id" ) values.append( reference_id ) placeholders = ", ".join( "?" for _ in insert_columns ) conn.execute( ( "INSERT INTO wallet_transactions (" + ", ".join( insert_columns ) + ") VALUES (" + placeholders + ")" ), tuple( values ), ) # ============================================================ # Backward-compatible module singleton # ============================================================ # Legacy code may still do: # from services.order_service import order_service # # New main.py should use: # OrderService(db_path=...) # # No DB work happens at import time. order_service = OrderService() # ============================================================ # Compatibility module-level async wrappers # ============================================================ async def ensure_schema() -> None: await order_service.ensure_schema() async def get_order( order_id: int, *, user_id: Optional[int] = None, ) -> Order: return await order_service.get_order( order_id, user_id=user_id, ) async def list_orders( user_id: int, *, status: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> list[Order]: return await order_service.list_orders( user_id, status=status, limit=limit, offset=offset, ) async def create_pending_owner_order( *, user_id: int, product_id: str, target: str, price: int, ) -> Order: return ( await order_service .create_pending_owner_order( user_id=user_id, product_id=product_id, target=target, price=price, ) ) __all__ = [ "Order", "WalletReservationResult", "PendingOrderCreateResult", "OrderService", "order_service", "OrderServiceError", "OrderNotFoundError", "InvalidOrderError", "InvalidOrderStatusError", "WalletError", "STATUS_PENDING_OWNER", "STATUS_OWNER_PROCESSING", "STATUS_REJECTED_BY_OWNER", "STATUS_CANCELLED_BY_USER", "STATUS_PRICE_CHANGED", "STATUS_INSUFFICIENT_BALANCE", "STATUS_PROVIDER_PROCESSING", "STATUS_PROVIDER_UNKNOWN", "STATUS_COMPLETED", "STATUS_FAILED_REFUNDED", "STATUS_FAILED_REFUND_REQUIRED", "VALID_STATUSES", "ensure_schema", "get_order", "list_orders", "create_pending_owner_order", ]