credi layer

CREDIT_SERVICE_EXAMPLE.py

@@ -0,0 +1,181 @@
+"""
+Example app.py configuration for Credit Service
+
+This shows how to configure the new middleware-based credit system.
+"""
+
+# ============================================================================
+# Credit Service Registration
+# ============================================================================
+
+from services.credit_service import CreditServiceConfig
+
+# Register credit service with route configurations
+CreditServiceConfig.register(
+    route_configs={
+        # Synchronous operations - credits confirmed/refunded immediately
+        "/gemini/generate-animation-prompt": {
+            "cost": 1,
+            "type": "sync"
+        },
+        "/gemini/edit-image": {
+            "cost": 1,
+            "type": "sync"
+        },
+        "/gemini/generate-text": {
+            "cost": 1,
+            "type": "sync"
+        },
+        "/gemini/analyze-image": {
+            "cost": 1,
+            "type": "sync"
+        },
+        
+        # Asynchronous operations - credits kept reserved until job completes
+        "/gemini/generate-video": {
+            "cost": 10,
+            "type": "async",
+            "status_endpoint": "/gemini/job/{job_id}"  # Optional metadata
+        },
+        
+        # Status check endpoints - no additional cost, but can trigger confirm/refund
+        "/gemini/job/{job_id}": {
+            "cost": 0,
+            "type": "async"
+        }
+    }
+)
+
+# ============================================================================
+# Middleware Chain (IMPORTANT: Order matters!)
+# ============================================================================
+
+# 1. AuthMiddleware MUST come first (sets request.state.user)
+app.add_middleware(AuthServiceConfig.get_middleware())
+
+# 2. CreditMiddleware comes after auth (needs request.state.user)
+app.add_middleware(CreditServiceConfig.get_middleware())
+
+# 3. Other middleware can follow
+# app.add_middleware(...)
+
+# ============================================================================
+# How It Works - Examples
+# ============================================================================
+
+"""
+SYNCHRONOUS OPERATION EXAMPLE:
+==============================
+
+1. User requests: POST /gemini/analyze-image
+   → Middleware: Reserve 1 credit
+   → Application: Process image analysis
+   → Application: Return 200 + analysis result
+   → Middleware: Confirm credit (success)
+
+2. User requests: POST /gemini/analyze-image  
+   → Middleware: Reserve 1 credit
+   → Application: Image validation fails
+   → Application: Return 400 + error
+   → Middleware: Refund credit (request error)
+
+
+ASYNCHRONOUS OPERATION EXAMPLE:
+===============================
+
+1. User requests: POST /gemini/generate-video
+   → Middleware: Reserve 10 credits
+   → Application: Create job, return job_id
+   → Application: Return 200 + {"job_id": "abc123", "status": "queued"}
+   → Middleware: Keep credits reserved (job pending)
+
+2. User polls: GET /gemini/job/abc123
+   → Middleware: No additional cost (cost=0)
+   → Application: Fetch job status
+   → Application: Return 200 + {"job_id": "abc123", "status": "processing"}
+   → Middleware: Keep credits reserved (still processing)
+
+3. User polls: GET /gemini/job/abc123
+   → Middleware: No additional cost
+   → Application: Fetch job status
+   → Application: Return 200 + {"job_id": "abc123", "status": "completed", "video_url": "..."}
+   → Middleware: Confirm 10 credits (job completed)
+
+4. Alternative - Job fails with refundable error:
+   GET /gemini/job/abc123
+   → Application: Return 200 + {"status": "failed", "error_message": "API_KEY_INVALID"}
+   → Middleware: Refund 10 credits (refundable server error)
+
+5. Alternative - Job fails with user error:
+   GET /gemini/job/abc123
+   → Application: Return 200 + {"status": "failed", "error_message": "safety filter"}  
+   → Middleware: Confirm 10 credits (non-refundable user error)
+
+
+TRANSACTION HISTORY:
+===================
+
+All operations are recorded in credit_transactions table:
+
+| transaction_id | type    | amount | balance_before | balance_after | reason                       |
+|----------------|---------|--------|----------------|---------------|------------------------------|
+| ctx_abc123     | reserve | -10    | 100            | 90            | POST /gemini/generate-video  |
+| cfm_def456     | confirm | 0      | 90             | 90            | Confirmed usage for ctx_abc  |
+
+User can query their history via: GET /credits/transactions
+"""
+
+# ============================================================================
+# Payment Integration (Still uses transaction manager directly)
+# ============================================================================
+
+"""
+In routers/payments.py, use CreditTransactionManager directly:
+
+from services.credit_service import CreditTransactionManager
+
+async def process_successful_payment(...):
+    await CreditTransactionManager.add_credits(
+        session=db,
+        user=user,
+        amount=credits_purchased,
+        source="payment",
+        reference_type="payment",
+        reference_id=transaction_id,
+        reason=f"Purchase: {package_id}",
+        metadata={"gateway": "razorpay", ...}
+    )
+"""
+
+# ============================================================================
+# Application Layer Changes
+# ============================================================================
+
+"""
+NO CHANGES NEEDED in application routers!
+
+Before (old approach):
+    @router.post("/gemini/analyze-image")
+    async def analyze_image(
+        request: Request,
+        user: User = Depends(verify_credits)  # ← Remove this
+    ):
+        # Had to manually handle credits
+        pass
+
+After (new approach):
+    @router.post("/gemini/analyze-image")
+    async def analyze_image(request: Request):
+        user = request.state.user  # Set by AuthMiddleware
+        # Credits automatically handled by middleware!
+        # Just focus on business logic
+        pass
+
+Old dependencies to REMOVE:
+- Depends(verify_credits)
+- Depends(verify_video_credits)
+- Manual credit deductions: user.credits -= X
+- Manual refunds in endpoints
+
+Everything is now handled automatically by middleware!
+"""

add_credit_transactions_table.py

@@ -0,0 +1,120 @@
+"""
+Database migration script to add credit_transactions table.
+
+Run this script to add the credit transaction tracking table to your database.
+
+Usage:
+    python add_credit_transactions_table.py
+"""
+import sqlite3
+import sys
+from pathlib import Path
+
+# Find database file
+DB_FILE = "apigateway_dev.db"
+if not Path(DB_FILE).exists():
+    DB_FILE = "apigateway_development.db"
+    if not Path(DB_FILE).exists():
+        print(f"Error: Database file not found. Looking for 'apigateway_dev.db' or 'apigateway_development.db'")
+        sys.exit(1)
+
+print(f"Using database: {DB_FILE}")
+
+# SQL to create credit_transactions table
+CREATE_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS credit_transactions (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    transaction_id VARCHAR(100) UNIQUE NOT NULL,
+    user_id INTEGER NOT NULL,
+    
+    -- Transaction details
+    transaction_type VARCHAR(20) NOT NULL,
+    amount INTEGER NOT NULL,
+    balance_before INTEGER NOT NULL,
+    balance_after INTEGER NOT NULL,
+    
+    -- Context
+    source VARCHAR(50) NOT NULL,
+    reference_type VARCHAR(30),
+    reference_id VARCHAR(100),
+    
+    -- Request/Response metadata
+    request_path VARCHAR(500),
+    request_method VARCHAR(10),
+    response_status INTEGER,
+    
+    -- Additional details
+    reason TEXT,
+    metadata JSON,
+    
+    -- Timestamps
+    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+    deleted_at DATETIME,
+    
+    FOREIGN KEY (user_id) REFERENCES users(id)
+);
+"""
+
+# SQL to create indexes
+CREATE_INDEXES_SQL = [
+    "CREATE INDEX IF NOT EXISTS ix_credit_transactions_transaction_id ON credit_transactions(transaction_id);",
+    "CREATE INDEX IF NOT EXISTS ix_user_transactions ON credit_transactions(user_id, created_at);",
+    "CREATE INDEX IF NOT EXISTS ix_transaction_type ON credit_transactions(transaction_type);",
+    "CREATE INDEX IF NOT EXISTS ix_reference ON credit_transactions(reference_type, reference_id);",
+    "CREATE INDEX IF NOT EXISTS ix_credit_transactions_created_at ON credit_transactions(created_at);",
+    "CREATE INDEX IF NOT EXISTS ix_credit_transactions_deleted_at ON credit_transactions(deleted_at);",
+]
+
+def run_migration():
+    """Run the migration to add credit_transactions table."""
+    try:
+        conn = sqlite3.connect(DB_FILE)
+        cursor = conn.cursor()
+        
+        # Check if table already exists
+        cursor.execute("""
+            SELECT name FROM sqlite_master 
+            WHERE type='table' AND name='credit_transactions';
+        """)
+        
+        if cursor.fetchone():
+            print("✅ Table 'credit_transactions' already exists. Skipping creation.")
+        else:
+            print("Creating 'credit_transactions' table...")
+            cursor.execute(CREATE_TABLE_SQL)
+            print("✅ Table created successfully!")
+        
+        # Create indexes
+        print("\nCreating indexes...")
+        for index_sql in CREATE_INDEXES_SQL:
+            cursor.execute(index_sql)
+        print("✅ Indexes created successfully!")
+        
+        # Commit changes
+        conn.commit()
+        
+        # Verify table
+        cursor.execute("PRAGMA table_info(credit_transactions);")
+        columns = cursor.fetchall()
+        
+        print(f"\nTable schema (columns: {len(columns)}):")
+        for col in columns:
+            print(f"  - {col[1]} ({col[2]})")
+        
+        conn.close()
+        
+        print("\n✅ Migration completed successfully!")
+        print("\nNext steps:")
+        print("  1. Update app.py to register credit service with route configs")
+        print("  2. The middleware will automatically track all credit operations")
+        print("  3. Check transaction history via /credits/transactions endpoint")
+        
+    except Exception as e:
+        print(f"\n❌ Migration failed: {e}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    print("="*60)
+    print("Credit Transactions Table Migration")
+    print("="*60)
+    run_migration()

app.py

@@ -84,14 +84,36 @@ async def lifespan(app: FastAPI):
     logger.info("✅ Auth Service configured")
     
     # Register Credit Service configuration
-    from services.credit_service import register_credit_service
-    register_credit_service(
-        route_costs={
-            "/gemini/generate-animation-prompt": 1,
-            "/gemini/edit-image": 1,
-            "/gemini/generate-video": 10,
-            "/gemini/generate-text": 1,
-            "/gemini/analyze-image": 1,
+    from services.credit_service import CreditServiceConfig
+    CreditServiceConfig.register(
+        route_configs={
+            # Synchronous operations - credits confirmed/refunded immediately
+            "/gemini/generate-animation-prompt": {
+                "cost": 1,
+                "type": "sync"
+            },
+            "/gemini/edit-image": {
+                "cost": 1,
+                "type": "sync"
+            },
+            "/gemini/generate-text": {
+                "cost": 1,
+                "type": "sync"
+            },
+            "/gemini/analyze-image": {
+                "cost": 1,
+                "type": "sync"
+            },
+            # Asynchronous operations - credits reserved until job completes
+            "/gemini/generate-video": {
+                "cost": 10,
+                "type": "async"
+            },
+            #  Status check endpoints - inspect response for job completion
+            "/gemini/job/{job_id}": {
+                "cost": 0,  # No additional cost for status checks
+                "type": "async"
+            }
         }
     )
     logger.info("✅ Credit Service configured")

core/models.py

@@ -297,3 +297,65 @@ class ApiKeyUsage(Base):
 
     def __repr__(self):
         return f"<ApiKeyUsage(index={self.key_index}, total={self.total_requests}, success={self.success_count}, failed={self.failure_count})>"
+
+
+# =============================================================================
+# Credit Transactions
+# =============================================================================
+
+class CreditTransaction(Base):
+    """
+    Complete audit trail for all credit operations.
+    
+    All credit changes (reserve, refund, confirm, purchase) are recorded here.
+    Enables balance verification, transaction history, and debugging.
+    
+    Transaction types:
+    - reserve: Credits deducted when request arrives (negative amount)
+    - refund: Credits returned due to failure (positive amount)
+    - confirm: Credits confirmed as used (amount = 0, marks finalization)
+    - purchase: Credits added via payment (positive amount)
+    - adjustment: Manual admin adjustment (positive or negative)
+    """
+    __tablename__ = "credit_transactions"
+    
+    # Composite indexes for efficient queries
+    __table_args__ = (
+        Index('ix_user_transactions', 'user_id', 'created_at'),
+        Index('ix_transaction_type', 'transaction_type'),
+        Index('ix_reference', 'reference_type', 'reference_id'),
+    )
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    transaction_id = Column(String(100), unique=True, index=True, nullable=False)  # ctx_<uuid>, ref_<uuid>, etc.
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)  # FK to users.id
+    
+    # Transaction details
+    transaction_type = Column(String(20), nullable=False)  # reserve, refund, confirm, purchase, adjustment
+    amount = Column(Integer, nullable=False)  # Positive for credits added, negative for removed
+    balance_before = Column(Integer, nullable=False)  # User balance before transaction
+    balance_after = Column(Integer, nullable=False)  # User balance after transaction
+    
+    # Context
+    source = Column(String(50), nullable=False)  # middleware, payment, job_completion, manual, admin
+    reference_type = Column(String(30), nullable=True)  # job, payment, request, NULL
+    reference_id = Column(String(100), nullable=True)  # job_id, transaction_id, request endpoint, etc.
+    
+    # Request/Response metadata (for middleware transactions)
+    request_path = Column(String(500), nullable=True)
+    request_method = Column(String(10), nullable=True)
+    response_status = Column(Integer, nullable=True)
+    
+    # Additional details
+    reason = Column(Text, nullable=True)  # Human-readable reason
+    metadata = Column(JSON, nullable=True)  # Additional context (endpoint_type, error_message, etc.)
+    
+    # Timestamps
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete
+    
+    # Relationship
+    user = relationship("User", backref="credit_transactions")
+
+    def __repr__(self):
+        return f"<CreditTransaction(id={self.transaction_id}, type={self.transaction_type}, amount={self.amount}, user={self.user_id})>"

routers/credits.py

@@ -122,3 +122,169 @@ async def get_credit_history(
         page=page,
         limit=limit
     )
+
+
+# =============================================================================
+# New Transaction History Endpoints
+# =============================================================================
+
+class CreditTransactionItem(BaseModel):
+    """Single credit transaction record."""
+    transaction_id: str
+    transaction_type: str  # reserve, refund, confirm, purchase
+    amount: int
+    balance_before: int
+    balance_after: int
+    source: str
+    reference_type: Optional[str] = None
+    reference_id: Optional[str] = None
+    request_path: Optional[str] = None
+    request_method: Optional[str] = None
+    response_status: Optional[int] = None
+    reason: Optional[str] = None
+    created_at: str
+
+
+class TransactionHistoryResponse(BaseModel):
+    """Response for transaction history endpoint."""
+    user_id: str
+    current_balance: int
+    transactions: List[CreditTransactionItem]
+    total_count: int
+    page: int
+    limit: int
+
+
+@router.get("/transactions", response_model=TransactionHistoryResponse)
+async def get_transaction_history(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    transaction_type: Optional[str] = Query(None, description="Filter by type (reserve, refund, confirm, purchase)"),
+    page: int = Query(1, ge=1, description="Page number"),
+    limit: int = Query(50, ge=1, le=100, description="Items per page")
+):
+    """
+    Get detailed credit transaction history.
+    
+    Returns complete audit trail of all credit operations including:
+    - Reservations (when credits were deducted for API calls)
+    - Refunds (when credits were returned due to failures)
+    - Confirmations (when credits were confirmed as used)
+    - Purchases (when credits were added via payments)
+    
+    Auth is handled by AuthMiddleware - user is in request.state.user
+    """
+    from core.models import CreditTransaction
+    from sqlalchemy import func
+    
+    user = request.state.user
+    offset = (page - 1) * limit
+    
+    # Build query
+    query = select(CreditTransaction).where(
+        CreditTransaction.user_id == user.id
+    )
+    
+    if transaction_type:
+        query = query.where(CreditTransaction.transaction_type == transaction_type)
+    
+    query = query.order_by(desc(CreditTransaction.created_at)).offset(offset).limit(limit)
+    
+    # Execute query
+    result = await db.execute(query)
+    transactions = result.scalars().all()
+    
+    # Get total count
+    count_query = select(func.count(CreditTransaction.id)).where(
+        CreditTransaction.user_id == user.id
+    )
+    if transaction_type:
+        count_query = count_query.where(CreditTransaction.transaction_type == transaction_type)
+    
+    count_result = await db.execute(count_query)
+    total_count = count_result.scalar() or 0
+    
+    # Build response
+    transaction_items = []
+    for tx in transactions:
+        transaction_items.append(CreditTransactionItem(
+            transaction_id=tx.transaction_id,
+            transaction_type=tx.transaction_type,
+            amount=tx.amount,
+            balance_before=tx.balance_before,
+            balance_after=tx.balance_after,
+            source=tx.source,
+            reference_type=tx.reference_type,
+            reference_id=tx.reference_id,
+            request_path=tx.request_path,
+            request_method=tx.request_method,
+            response_status=tx.response_status,
+            reason=tx.reason,
+            created_at=tx.created_at.isoformat() if tx.created_at else None
+        ))
+    
+    return TransactionHistoryResponse(
+        user_id=user.user_id,
+        current_balance=user.credits,
+        transactions=transaction_items,
+        total_count=total_count,
+        page=page,
+        limit=limit
+    )
+
+
+class BalanceVerificationResponse(BaseModel):
+    """Response for balance verification endpoint."""
+    user_id: str
+    current_balance: int
+    calculated_balance: int
+    is_valid: bool
+    discrepancy: int
+    last_transaction_at: Optional[str] = None
+
+
+@router.get("/balance/verify", response_model=BalanceVerificationResponse)
+async def verify_balance(
+    request: Request,
+    db: AsyncSession = Depends(get_db)
+):
+    """
+    Verify user balance against transaction history.
+    
+    Calculates balance from all transactions and compares with stored balance.
+    Useful for debugging credit discrepancies.
+    
+    Auth is handled by AuthMiddleware - user is in request.state.user
+    """
+    from core.models import CreditTransaction
+    from sqlalchemy import func
+    
+    user = request.state.user
+    
+    # Calculate balance from transactions
+    result = await db.execute(
+        select(func.sum(CreditTransaction.amount)).where(
+            CreditTransaction.user_id == user.id
+        )
+    )
+    calculated_balance = result.scalar() or 0
+    
+    # Get last transaction time
+    last_tx_result = await db.execute(
+        select(CreditTransaction).where(
+            CreditTransaction.user_id == user.id
+        ).order_by(desc(CreditTransaction.created_at)).limit(1)
+    )
+    last_tx = last_tx_result.scalar_one_or_none()
+    
+    is_valid = (calculated_balance == user.credits)
+    discrepancy = user.credits - calculated_balance
+    
+    return BalanceVerificationResponse(
+        user_id=user.user_id,
+        current_balance=user.credits,
+        calculated_balance=calculated_balance,
+        is_valid=is_valid,
+        discrepancy=discrepancy,
+        last_transaction_at=last_tx.created_at.isoformat() if last_tx and last_tx.created_at else None
+    )

routers/payments.py

@@ -174,7 +174,27 @@ async def process_successful_payment(
     credits_added = 0
     
     if transaction.status != "paid":
-        # First time processing - add credits
+        # First time processing - add credits using transaction manager
+        from services.credit_service import CreditTransactionManager
+        
+        await CreditTransactionManager.add_credits(
+            session=db,
+            user=user,
+            amount=transaction.credits_amount,
+            source="payment",
+            reference_type="payment",
+            reference_id=transaction.transaction_id,
+            reason=f"Purchase: {transaction.package_id}",
+            metadata={
+                "package_id": transaction.package_id,
+                "gateway": "razorpay",
+                "gateway_payment_id": payment_id,
+                "amount_paise": transaction.amount_paise,
+                "verified_by": source,
+                "signature": signature
+            }
+        )
+        
         transaction.status = "paid"
         transaction.gateway_payment_id = payment_id
         transaction.paid_at = datetime.utcnow()
@@ -182,8 +202,6 @@ async def process_successful_payment(
             transaction.razorpay_signature = signature
         
         update_verified_by(transaction, source)
-        
-        user.credits += transaction.credits_amount
         credits_added = transaction.credits_amount
         
         logger.info(

services/credit_service/__init__.py

@@ -1,75 +1,61 @@
 """
-Credit Service - Credit validation middleware for API Gateway
+Credit Service - Middleware-based credit management system.
 
-Provides plug-and-play credit management with:
-- Per-route cost configuration
-- Credit reservation and validation
-- Request middleware for credit checks
-- Automatic refund on errors
+Provides automatic credit tracking via middleware with complete audit trail.
+Application layer is completely unaware of credit management.
 
-Usage:
-    # In app.py startup
-    from services.credit_service import register_credit_service
+Components:
+- CreditServiceConfig: Route configuration with credit costs and types
+- CreditMiddleware: Request/response-based credit handling
+- CreditTransactionManager: Core credit transaction operations
+- ResponseInspector: Response analysis for credit decisions
+- credit_manager: Legacy refund logic (used by middleware)
+
+Usage in app.py:
+    from services.credit_service import CreditServiceConfig
     
-    register_credit_service(
-        route_costs={
-            "/gemini/generate-animation-prompt": 1,
-            "/gemini/edit-image": 1,
-            "/gemini/generate-video": 10,
-            "/gemini/generate-text": 1,
-            "/gemini/analyze-image": 1,
+    CreditServiceConfig.register(
+        route_configs={
+            "/gemini/analyze-image": {"cost": 1, "type": "sync"},
+            "/gemini/generate-video": {"cost": 10, "type": "async"},
         }
     )
     
-    # In routers
-    from fastapi import Request
-    
-    @router.post("/api/endpoint")
-    async def endpoint(request: Request):
-        user = request.state.user  # From AuthMiddleware
-        credits_reserved = request.state.credits_reserved  # From CreditMiddleware
-        return {"credits_remaining": user.credits}
+    app.add_middleware(CreditServiceConfig.get_middleware())
 """
 
 from services.credit_service.config import CreditServiceConfig
 from services.credit_service.middleware import CreditMiddleware
+from services.credit_service.transaction_manager import (
+    CreditTransactionManager,
+    InsufficientCreditsError,
+    TransactionNotFoundError,
+    UserNotFoundError
+)
+from services.credit_service.response_inspector import ResponseInspector
 from services.credit_service.credit_manager import (
-    reserve_credit,
-    confirm_credit,
-    refund_credit,
-    handle_job_completion,
     is_refundable_error,
     REFUNDABLE_ERROR_PATTERNS,
     NON_REFUNDABLE_ERROR_PATTERNS,
 )
 
-
-def register_credit_service(
-    route_costs: dict = None,
-) -> None:
-    """
-    Register the credit service with application configuration.
-    
-    Args:
-        route_costs: Dictionary mapping route paths to credit costs
-                    Example: {"/gemini/generate-video": 10, "/gemini/edit-image": 1}
-    """
-    CreditServiceConfig.register(
-        route_costs=route_costs or {},
-    )
-
-
 __all__ = [
-    # Registration
-    'register_credit_service',
+    # Configuration
     'CreditServiceConfig',
+    
+    # Middleware
     'CreditMiddleware',
     
-    # Credit Management
-    'reserve_credit',
-    'confirm_credit',
-    'refund_credit',
-    'handle_job_completion',
+    # Transaction Manager
+    'CreditTransactionManager',
+    'InsufficientCreditsError',
+    'TransactionNotFoundError',
+    'UserNotFoundError',
+    
+    # Response Inspector
+    'ResponseInspector',
+    
+    # Refund logic helpers
     'is_refundable_error',
     'REFUNDABLE_ERROR_PATTERNS',
     'NON_REFUNDABLE_ERROR_PATTERNS',

services/credit_service/config.py

@@ -1,7 +1,7 @@
 """
 Credit Service Configuration
 
-Manages credit cost configuration for API routes.
+Manages credit cost configuration for API routes with endpoint type classification.
 """
 
 import logging
@@ -15,25 +15,41 @@ class CreditServiceConfig(BaseService):
     """
     Configuration for the credit service.
     
-    Controls which routes require credits and how much they cost.
+    Controls which routes require credits, how much they cost,
+    and how to handle credits based on response (sync vs async).
     """
     
     SERVICE_NAME = "credit_service"
     
-    # Route cost configuration
-    _route_costs: Dict[str, int] = {}
+    # Route configuration: path -> {cost, type, ...}
+    _route_configs: Dict[str, dict] = {}
     
     @classmethod
     def register(
         cls,
-        route_costs: Dict[str, int] = None,
+        route_configs: Dict[str, dict] = None,
     ) -> None:
         """
         Register credit service configuration.
         
         Args:
-            route_costs: Dictionary mapping route paths to credit costs
-                        Example: {"/gemini/generate-video": 10, "/gemini/edit-image": 1}
+            route_configs: Dictionary mapping route paths to configuration
+                Example: {
+                    "/gemini/generate-video": {
+                        "cost": 10,
+                        "type": "async",  # or "sync", "free"
+                        "status_endpoint": "/gemini/job/{job_id}"  # optional
+                    },
+                    "/gemini/analyze-image": {
+                        "cost": 1,
+                        "type": "sync"
+                    }
+                }
+        
+        Endpoint types:
+        - "sync": Synchronous operation, confirm/refund immediately based on response status
+        - "async": Asynchronous operation (job), keep reserved until status check
+        - "free": No credits required (cost = 0)
         
         Raises:
             RuntimeError: If service is already registered
@@ -41,15 +57,15 @@ class CreditServiceConfig(BaseService):
         if cls._registered:
             raise RuntimeError(f"{cls.SERVICE_NAME} is already registered")
         
-        # Store route costs
-        cls._route_costs = route_costs or {}
+        # Store route configs
+        cls._route_configs = route_configs or {}
         
         cls._registered = True
         
         logger.info(f"✅ {cls.SERVICE_NAME} registered successfully")
-        logger.info(f"   Routes with credit costs: {len(cls._route_costs)}")
-        for route, cost in cls._route_costs.items():
-            logger.info(f"     {route}: {cost} credits")
+        logger.info(f"   Routes with credit requirements: {len(cls._route_configs)}")
+        for route, config in cls._route_configs.items():
+            logger.info(f"     {route}: {config.get('cost', 0)} credits ({config.get('type', 'free')})")
     
     @classmethod
     def get_middleware(cls):
@@ -57,6 +73,21 @@ class CreditServiceConfig(BaseService):
         from services.credit_service.middleware import CreditMiddleware
         return CreditMiddleware
     
+    @classmethod
+    def get_config(cls, path: str) -> dict:
+        """
+        Get the full configuration for a given path.
+        
+        Args:
+            path: URL path to check
+            
+        Returns:
+            Configuration dict with 'cost' and 'type' keys
+            Default: {"cost": 0, "type": "free"}
+        """
+        cls.assert_registered()
+        return cls._route_configs.get(path, {"cost": 0, "type": "free"})
+    
     @classmethod
     def get_cost(cls, path: str) -> int:
         """
@@ -68,20 +99,18 @@ class CreditServiceConfig(BaseService):
         Returns:
             Credit cost (0 if route doesn't require credits)
         """
-        cls.assert_registered()
-        return cls._route_costs.get(path, 0)
+        return cls.get_config(path).get("cost", 0)
     
     @classmethod
     def requires_credits(cls, path: str) -> bool:
         """Check if a URL path requires credits."""
-        cls.assert_registered()
         return cls.get_cost(path) > 0
     
     @classmethod
-    def get_all_costs(cls) -> Dict[str, int]:
-        """Get all route costs."""
+    def get_all_configs(cls) -> Dict[str, dict]:
+        """Get all route configurations."""
         cls.assert_registered()
-        return cls._route_costs.copy()
+        return cls._route_configs.copy()
 
 
 __all__ = ['CreditServiceConfig']

services/credit_service/middleware.py

@@ -1,17 +1,26 @@
 """
-Credit Middleware - Request credit validation layer
+Credit Middleware - Automatic request/response-based credit management.
 
-Intercepts requests to validate and reserve credits for paid endpoints.
+Intercepts requests to reserve credits and inspects responses to automatically
+confirm or refund based on operation success/failure.
+
+Application layer is completely unaware of credit management.
 """
 
 import logging
+import json
 from datetime import datetime
-from fastapi import Request, HTTPException, status
-from fastapi.responses import JSONResponse
+from fastapi import Request, status
+from fastapi.responses import JSONResponse, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from core.database import async_session_maker
 from services.credit_service.config import CreditServiceConfig
+from services.credit_service.transaction_manager import (
+    CreditTransactionManager,
+    InsufficientCreditsError
+)
+from services.credit_service.response_inspector import ResponseInspector
 from services.base_service.middleware_chain import (
     BaseServiceMiddleware,
     get_request_context,
@@ -22,18 +31,15 @@ logger = logging.getLogger(__name__)
 
 class CreditMiddleware(BaseServiceMiddleware):
     """
-    Credit validation middleware for request validation.
+    Credit middleware with automatic request/response-based credit management.
     
-    Flow:
-    1. Check if route requires credits based on URL
-    2. Get authenticated user from request.state (set by AuthMiddleware)
-    3. Check if user has sufficient credits
-    4. Reserve credits (deduct from balance)
-    5. Attach credit info to request.state
-    6. Continue to next middleware/route
+    Application layer is completely unaware of credits.
+    Credits are reserved on request, confirmed/refunded based on response.
     
-    Credits are reserved but tracked - they can be refunded if job fails
-    with a server-side error.
+    Flow:
+    1. REQUEST PHASE: Reserve credits if endpoint requires them
+    2. APPLICATION: Process request (unaware of credits)
+    3. RESPONSE PHASE: Inspect response and confirm/refund credits
     
     NOTE: This middleware MUST run AFTER AuthMiddleware since it needs
     the authenticated user from request.state.user
@@ -43,88 +49,171 @@ class CreditMiddleware(BaseServiceMiddleware):
     
     async def dispatch(self, request: Request, call_next):
         """Process request through credit middleware."""
+        
         # Skip OPTIONS requests (CORS preflight)
         if request.method == "OPTIONS":
             return await call_next(request)
         
-        # Initialize request context
-        ctx = get_request_context(request)
-        
-        # Get path from request
         path = request.url.path
         
-        # Check if route requires credits
-        credit_cost = CreditServiceConfig.get_cost(path)
+        # Get endpoint configuration
+        config = CreditServiceConfig.get_config(path)
+        credit_cost = config.get("cost", 0)
+        endpoint_type = config.get("type", "free")
         
         if credit_cost == 0:
-            # Route doesn't require credits, skip
-            self.log_request(request, f"Route doesn't require credits")
-            ctx.set_credits(0, 0)
-            response = await call_next(request)
-            return response
-        
-        # Route requires credits - user MUST be authenticated
-        # (AuthMiddleware should have already validated this)
-        user = request.state.user if hasattr(request.state, 'user') else None
+            # Free endpoint, no credit handling needed
+            return await call_next(request)
         
+        # User must be authenticated (set by AuthMiddleware)
+        user = getattr(request.state, 'user', None)
         if not user:
-            # This shouldn't happen if auth is configured correctly
-            self.log_error(request, "Credit-required route accessed without authentication")
             return JSONResponse(
                 status_code=status.HTTP_401_UNAUTHORIZED,
-                content={"detail": "Authentication required for this endpoint"},
+                content={"detail": "Authentication required for this endpoint"}
             )
         
-        # Check if user has sufficient credits
-        if user.credits < credit_cost:
-            self.log_request(
-                request,
-                f"Insufficient credits: has {user.credits}, needs {credit_cost}"
-            )
-            return JSONResponse(
-                status_code=status.HTTP_402_PAYMENT_REQUIRED,
-                content={
-                    "detail": f"Insufficient credits. This operation requires {credit_cost} credits. You have {user.credits}.",
-                    "credits_required": credit_cost,
-                    "credits_available": user.credits,
-                },
-            )
-        
-        # Reserve credits (deduct from user balance)
+        # ===================================================================
+        # REQUEST PHASE: Reserve Credits
+        # ===================================================================
         async with async_session_maker() as db:
             try:
-                # Deduct credits
-                user.credits -= credit_cost
-                user.last_used_at = datetime.utcnow()
+                # Reserve credits
+                transaction = await CreditTransactionManager.reserve_credits(
+                    session=db,
+                    user=user,
+                    amount=credit_cost,
+                    source="middleware",
+                    reference_type="request",
+                    reference_id=f"{request.method}:{path}",
+                    reason=f"{request.method} {path}",
+                    metadata={
+                        "endpoint_type": endpoint_type,
+                        "cost": credit_cost
+                    },
+                    request=request
+                )
                 
-                # Update in database
-                db.add(user)
                 await db.commit()
-                await db.refresh(user)
                 
-                # Attach credit info to request state
-                ctx.set_credits(credit_cost, user.credits)
-                request.state.credits_reserved = credit_cost
-                request.state.credits_remaining = user.credits
+                # Attach transaction info to request for response phase
+                request.state.credit_transaction_id = transaction.transaction_id
+                request.state.endpoint_type = endpoint_type
+                request.state.credit_cost = credit_cost
                 
                 self.log_request(
                     request,
-                    f"Reserved {credit_cost} credits for {user.user_id}, remaining: {user.credits}"
+                    f"Reserved {credit_cost} credits ({endpoint_type}), txn: {transaction.transaction_id}"
                 )
                 
-                # Continue to next middleware/route
-                response = await call_next(request)
-                return response
-            
+            except InsufficientCreditsError as e:
+                await db.rollback()
+                return JSONResponse(
+                    status_code=status.HTTP_402_PAYMENT_REQUIRED,
+                    content={
+                        "detail": f"Insufficient credits. Required: {credit_cost}, Available: {user.credits}",
+                        "credits_required": credit_cost,
+                        "credits_available": user.credits
+                    }
+                )
             except Exception as e:
                 await db.rollback()
-                self.log_error(request, f"Error reserving credits: {e}")
+                logger.error(f"Credit reservation failed: {e}", exc_info=True)
                 return JSONResponse(
                     status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                    content={"detail": "Failed to reserve credits. Please try again."},
+                    content={"detail": "Failed to reserve credits"}
                 )
-            finally:
-                await db.close()
+        
+        # ===================================================================
+        # CALL APPLICATION LAYER (completely unaware of credits)
+        # ===================================================================
+        response = await call_next(request)
+        
+        # ===================================================================
+        # RESPONSE PHASE: Confirm or Refund Based on Response
+        # ===================================================================
+        transaction_id = getattr(request.state, 'credit_transaction_id', None)
+        if not transaction_id:
+            # No transaction to finalize (shouldn't happen, but be safe)
+            return response
+        
+        # We need to read the response body to inspect it
+        # This requires special handling to not consume the response
+        response_body = b""
+        async for chunk in response.body_iterator:
+            response_body += chunk
+        
+        # Parse response for inspection
+        response_data = ResponseInspector.parse_response_body(response_body)
+        inspector = ResponseInspector()
+        
+        # Determine credit action based on response
+        async with async_session_maker() as db:
+            try:
+                should_confirm = inspector.should_confirm(
+                    response, endpoint_type, response_data
+                )
+                should_refund = inspector.should_refund(
+                    response, endpoint_type, response_data
+                )
+                
+                if should_confirm:
+                    # Operation successful - confirm credits
+                    await CreditTransactionManager.confirm_credits(
+                        session=db,
+                        transaction_id=transaction_id,
+                        metadata={
+                            "response_status": response.status_code,
+                            "endpoint_type": endpoint_type
+                        }
+                    )
+                    await db.commit()
+                    
+                    self.log_request(
+                        request,
+                        f"Credits confirmed for {transaction_id} (success)"
+                    )
+                
+                elif should_refund:
+                    # Operation failed - refund credits
+                    reason = inspector.get_refund_reason(response, response_data)
+                    await CreditTransactionManager.refund_credits(
+                        session=db,
+                        transaction_id=transaction_id,
+                        reason=reason,
+                        metadata={
+                            "response_status": response.status_code,
+                            "endpoint_type": endpoint_type,
+                            "error": response_data.get("detail") if response_data else None
+                        }
+                    )
+                    await db.commit()
+                    
+                    self.log_request(
+                        request,
+                        f"Credits refunded for {transaction_id}: {reason}"
+                    )
+                
+                else:
+                    # Keep reserved (async operation pending completion)
+                    self.log_request(
+                        request,
+                        f"Credits kept reserved for {transaction_id} (async pending)"
+                    )
+                
+            except Exception as e:
+                logger.error(f"Response phase credit handling failed: {e}", exc_info=True)
+                # Don't fail the actual response, just log the error
+                # The transaction remains reserved and can be manually resolved
+        
+        # Reconstruct response with original body
+        # This is necessary because we consumed the body_iterator above
+        return Response(
+            content=response_body,
+            status_code=response.status_code,
+            headers=dict(response.headers),
+            media_type=response.media_type
+        )
 
 
 __all__ = ['CreditMiddleware']

services/credit_service/response_inspector.py

@@ -0,0 +1,143 @@
+"""
+Response Inspector - Helper for credit middleware to inspect responses.
+
+Determines credit actions based on response status and content.
+"""
+import json
+import logging
+from typing import Optional
+from fastapi import Response
+
+logger = logging.getLogger(__name__)
+
+
+class ResponseInspector:
+    """Helper class to inspect responses and determine credit actions."""
+    
+    @staticmethod
+    def should_confirm(
+        response: Response,
+        endpoint_type: str,
+        response_data: Optional[dict]
+    ) -> bool:
+        """
+        Determine if credits should be confirmed.
+        
+        Args:
+            response: FastAPI Response object
+            endpoint_type: Type of endpoint ("sync", "async", "free")
+            response_data: Parsed response body (if JSON)
+            
+        Returns:
+            True if credits should be confirmed (operation successful)
+        """
+        if endpoint_type == "sync":
+            # Synchronous: confirm on 2xx
+            return 200 <= response.status_code < 300
+        
+        if endpoint_type == "async":
+            # Asynchronous: check job status in response
+            if not response_data:
+                return False
+            
+            # For job creation endpoints (POST /gemini/generate-video)
+            if "job_id" in response_data and response.status_code < 300:
+                # Job created successfully, keep reserved (don't confirm yet)
+                return False
+            
+            # For status check endpoints (GET /gemini/job/{id})
+            job_status = response_data.get("status")
+            if job_status ==  "completed":
+                logger.debug(f"Job completed, confirming credits")
+                return True
+        
+        return False
+    
+    @staticmethod
+    def should_refund(
+        response: Response,
+        endpoint_type: str,
+        response_data: Optional[dict]
+    ) -> bool:
+        """
+        Determine if credits should be refunded.
+        
+        Args:
+            response: FastAPI Response object
+            endpoint_type: Type of endpoint ("sync", "async", "free")
+            response_data: Parsed response body (if JSON)
+            
+        Returns:
+            True if credits should be refunded (operation failed)
+        """
+        if endpoint_type == "sync":
+            # Synchronous: refund on 4xx/5xx
+            return response.status_code >= 400
+        
+        if endpoint_type == "async":
+            # Asynchronous: check job status
+            if not response_data:
+                # Failed to parse response, likely an error
+                return response.status_code >= 400
+            
+            job_status = response_data.get("status")
+            if job_status == "failed":
+                # Check if error is refundable
+                error_message = response_data.get("error_message", "")
+                from services.credit_service.credit_manager import is_refundable_error
+                is_refundable = is_refundable_error(error_message)
+                logger.debug(
+                    f"Job failed with error: {error_message[:100]}, "
+                    f"refundable: {is_refundable}"
+                )
+                return is_refundable
+        
+        return False
+    
+    @staticmethod
+    def get_refund_reason(
+        response: Response,
+        response_data: Optional[dict]
+    ) -> str:
+        """
+        Get human-readable refund reason.
+        
+        Args:
+            response: FastAPI Response object
+            response_data: Parsed response body (if JSON)
+            
+        Returns:
+            Human-readable reason string
+        """
+        if response.status_code >= 500:
+            return f"Server error: {response.status_code}"
+        
+        if response.status_code >= 400:
+            detail = response_data.get("detail") if response_data else None
+            return f"Request error: {detail or response.status_code}"
+        
+        if response_data:
+            error_message = response_data.get("error_message")
+            if error_message:
+                return f"Job failed: {error_message[:100]}"
+        
+        return "Unknown error"
+    
+    @staticmethod
+    def parse_response_body(response_body: bytes) -> Optional[dict]:
+        """
+        Safely parse response body as JSON.
+        
+        Args:
+            response_body: Raw response bytes
+            
+        Returns:
+            Parsed dict or None if not JSON
+        """
+        try:
+            return json.loads(response_body.decode())
+        except:
+            return None
+
+
+__all__ = ['ResponseInspector']

services/credit_service/transaction_manager.py

@@ -0,0 +1,425 @@
+"""
+Credit Transaction Manager - Core credit operations service.
+
+All credit operations flow through this service:
+- Reserve credits (deduct from balance)
+- Refund credits (add back to balance)
+- Confirm credits (mark as used)
+- Add credits (purchases, bonuses)
+
+Provides complete audit trail with request/response context.
+"""
+import uuid
+import logging
+from datetime import datetime
+from typing import Optional, List
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, desc, func
+from fastapi import Request
+
+from core.models import User, CreditTransaction
+
+logger = logging.getLogger(__name__)
+
+
+# =============================================================================
+# Custom Exceptions
+# =============================================================================
+
+class InsufficientCreditsError(Exception):
+    """Raised when user doesn't have enough credits."""
+    pass
+
+
+class TransactionNotFoundError(Exception):
+    """Raised when a transaction cannot be found."""
+    pass
+
+
+class UserNotFoundError(Exception):
+    """Raised when a user cannot be found."""
+    pass
+
+
+# =============================================================================
+# Credit Transaction Manager
+# =============================================================================
+
+class CreditTransactionManager:
+    """
+    Centralized credit transaction management.
+    All credit operations flow through this service.
+    """
+    
+    @staticmethod
+    async def reserve_credits(
+        session: AsyncSession,
+        user: User,
+        amount: int,
+        source: str,
+        reference_type: Optional[str] = None,
+        reference_id: Optional[str] = None,
+        reason: Optional[str] = None,
+        metadata: Optional[dict] = None,
+        request: Optional[Request] = None
+    ) -> CreditTransaction:
+        """
+        Reserve credits (deduct from balance).
+        
+        Called by middleware when request arrives.
+        
+        Args:
+            session: Database session
+            user: User model instance
+            amount: Number of credits to reserve
+            source: Source of transaction (e.g., "middleware")
+            reference_type: Type of reference (e.g., "request", "job")
+            reference_id: Reference identifier (e.g., "POST:/gemini/video")
+            reason: Human-readable reason
+            metadata: Additional context
+            request: FastAPI request object (for extracting metadata)
+            
+        Returns:
+            CreditTransaction record
+            
+        Raises:
+            InsufficientCreditsError: If user doesn't have enough credits
+        """
+        if user.credits < amount:
+            raise InsufficientCreditsError(
+                f"User has {user.credits} credits, needs {amount}"
+            )
+        
+        # Generate transaction ID
+        transaction_id = f"ctx_{uuid.uuid4().hex[:16]}"
+        
+        # Capture request metadata
+        request_metadata = {}
+        request_path = None
+        request_method = None
+        
+        if request:
+            request_path = request.url.path
+            request_method = request.method
+            request_metadata = {
+                "path": request_path,
+                "method": request_method,
+                "ip": request.client.host if request.client else None,
+                "user_agent": request.headers.get("user-agent")
+            }
+        
+        # Create transaction record
+        transaction = CreditTransaction(
+            transaction_id=transaction_id,
+            user_id=user.id,
+            transaction_type="reserve",
+            amount=-amount,  # Negative for deduction
+            balance_before=user.credits,
+            balance_after=user.credits - amount,
+            source=source,
+            reference_type=reference_type,
+            reference_id=reference_id,
+            request_path=request_path,
+            request_method=request_method,
+            reason=reason or f"Reserved for {request_path or 'operation'}",
+            metadata={**request_metadata, **(metadata or {})}
+        )
+        
+        # Update user balance
+        user.credits -= amount
+        user.last_used_at = datetime.utcnow()
+        
+        session.add(transaction)
+        await session.flush()  # Don't commit yet - let caller handle
+        
+        logger.info(
+            f"Reserved {amount} credits for user {user.user_id}, "
+            f"transaction: {transaction_id}, new balance: {user.credits}"
+        )
+        
+        return transaction
+    
+    @staticmethod
+    async def confirm_credits(
+        session: AsyncSession,
+        transaction_id: str,
+        metadata: Optional[dict] = None
+    ) -> CreditTransaction:
+        """
+        Confirm credits were legitimately used.
+        
+        Called by middleware after successful response.
+        Creates a confirmation record (amount = 0).
+        
+        Args:
+            session: Database session
+            transaction_id: Original reservation transaction ID
+            metadata: Additional context (response_status, etc.)
+            
+        Returns:
+            CreditTransaction record
+            
+        Raises:
+            TransactionNotFoundError: If original transaction not found
+        """
+        # Find original reservation
+        result = await session.execute(
+            select(CreditTransaction).where(
+                CreditTransaction.transaction_id == transaction_id
+            )
+        )
+        original = result.scalar_one_or_none()
+        
+        if not original:
+            raise TransactionNotFoundError(f"Transaction {transaction_id} not found")
+        
+        # Create confirmation transaction
+        confirm_tx = CreditTransaction(
+            transaction_id=f"cfm_{uuid.uuid4().hex[:16]}",
+            user_id=original.user_id,
+            transaction_type="confirm",
+            amount=0,  # No balance change
+            balance_before=original.balance_after,
+            balance_after=original.balance_after,
+            source="middleware",
+            reference_type=original.reference_type,
+            reference_id=original.reference_id,
+            request_path=original.request_path,
+            request_method=original.request_method,
+            reason=f"Confirmed usage for {original.transaction_id}",
+            metadata={
+                "original_transaction_id": transaction_id,
+                **(metadata or {})
+            }
+        )
+        
+        session.add(confirm_tx)
+        await session.flush()
+        
+        logger.info(f"Confirmed credits for transaction {transaction_id}")
+        
+        return confirm_tx
+    
+    @staticmethod
+    async def refund_credits(
+        session: AsyncSession,
+        transaction_id: str,
+        reason: str,
+        metadata: Optional[dict] = None
+    ) -> CreditTransaction:
+        """
+        Refund reserved credits.
+        
+        Called by middleware after failure or refundable error.
+        
+        Args:
+            session: Database session
+            transaction_id: Original reservation transaction ID
+            reason: Reason for refund
+            metadata: Additional context
+            
+        Returns:
+            CreditTransaction record
+            
+        Raises:
+            TransactionNotFoundError: If original transaction not found
+            UserNotFoundError: If user not found
+        """
+        # Find original reservation
+        result = await session.execute(
+            select(CreditTransaction).where(
+                CreditTransaction.transaction_id == transaction_id
+            )
+        )
+        original = result.scalar_one_or_none()
+        
+        if not original:
+            raise TransactionNotFoundError(f"Transaction {transaction_id} not found")
+        
+        # Get user
+        user_result = await session.execute(
+            select(User).where(User.id == original.user_id)
+        )
+        user = user_result.scalar_one_or_none()
+        
+        if not user:
+            raise UserNotFoundError(f"User {original.user_id} not found")
+        
+        # Calculate refund amount (reverse of original deduction)
+        refund_amount = abs(original.amount)
+        
+        # Create refund transaction
+        refund_tx = CreditTransaction(
+            transaction_id=f"ref_{uuid.uuid4().hex[:16]}",
+            user_id=user.id,
+            transaction_type="refund",
+            amount=refund_amount,  # Positive for addition
+            balance_before=user.credits,
+            balance_after=user.credits + refund_amount,
+            source="middleware",
+            reference_type=original.reference_type,
+            reference_id=original.reference_id,
+            request_path=original.request_path,
+            request_method=original.request_method,
+            reason=reason,
+            metadata={
+                "original_transaction_id": transaction_id,
+                **(metadata or {})
+            }
+        )
+        
+        # Update user balance
+        user.credits += refund_amount
+        
+        session.add(refund_tx)
+        await session.flush()
+        
+        logger.info(
+            f"Refunded {refund_amount} credits for transaction {transaction_id}, "
+            f"reason: {reason[:100]}, new balance: {user.credits}"
+        )
+        
+        return refund_tx
+    
+    @staticmethod
+    async def add_credits(
+        session: AsyncSession,
+        user: User,
+        amount: int,
+        source: str,
+        reference_type: Optional[str] = None,
+        reference_id: Optional[str] = None,
+        reason: Optional[str] = None,
+        metadata: Optional[dict] = None
+    ) -> CreditTransaction:
+        """
+        Add credits (purchase, bonus, etc).
+        
+        Used by payment router only.
+        
+        Args:
+            session: Database session
+            user: User model instance
+            amount: Number of credits to add
+            source: Source of transaction (e.g., "payment")
+            reference_type: Type of reference (e.g., "payment")
+            reference_id: Reference identifier (e.g., transaction_id)
+            reason: Human-readable reason
+            metadata: Additional context
+            
+        Returns:
+            CreditTransaction record
+        """
+        transaction_id = f"add_{uuid.uuid4().hex[:16]}"
+        
+        transaction = CreditTransaction(
+            transaction_id=transaction_id,
+            user_id=user.id,
+            transaction_type="purchase",
+            amount=amount,
+            balance_before=user.credits,
+            balance_after=user.credits + amount,
+            source=source,
+            reference_type=reference_type,
+            reference_id=reference_id,
+            reason=reason,
+            metadata=metadata
+        )
+        
+        user.credits += amount
+        
+        session.add(transaction)
+        await session.flush()
+        
+        logger.info(
+            f"Added {amount} credits to user {user.user_id}, "
+            f"source: {source}, new balance: {user.credits}"
+        )
+        
+        return transaction
+    
+    @staticmethod
+    async def get_balance(
+        session: AsyncSession,
+        user_id: int,
+        verify: bool = False
+    ) -> int:
+        """
+        Get current balance, optionally verify against transactions.
+        
+        Args:
+            session: Database session
+            user_id: User ID
+            verify: If True, calculate balance from transactions and compare
+            
+        Returns:
+            Current balance
+        """
+        # Get user
+        result = await session.execute(
+            select(User).where(User.id == user_id)
+        )
+        user = result.scalar_one_or_none()
+        
+        if not user:
+            raise UserNotFoundError(f"User {user_id} not found")
+        
+        if not verify:
+            return user.credits
+        
+        # Calculate balance from transactions
+        tx_result = await session.execute(
+            select(func.sum(CreditTransaction.amount)).where(
+                CreditTransaction.user_id == user_id
+            )
+        )
+        calculated_balance = tx_result.scalar() or 0
+        
+        if calculated_balance != user.credits:
+            logger.warning(
+                f"Balance mismatch for user {user_id}: "
+                f"stored={user.credits}, calculated={calculated_balance}"
+            )
+        
+        return user.credits
+    
+    @staticmethod
+    async def get_transaction_history(
+        session: AsyncSession,
+        user_id: int,
+        transaction_type: Optional[str] = None,
+        limit: int = 50,
+        offset: int = 0
+    ) -> List[CreditTransaction]:
+        """
+        Get transaction history with filters.
+        
+        Args:
+            session: Database session
+            user_id: User ID
+            transaction_type: Filter by transaction type
+            limit: Maximum number of results
+            offset: Offset for pagination
+            
+        Returns:
+            List of CreditTransaction records
+        """
+        query = select(CreditTransaction).where(
+            CreditTransaction.user_id == user_id
+        )
+        
+        if transaction_type:
+            query = query.where(CreditTransaction.transaction_type == transaction_type)
+        
+        query = query.order_by(desc(CreditTransaction.created_at)).offset(offset).limit(limit)
+        
+        result = await session.execute(query)
+        return list(result.scalars().all())
+
+
+__all__ = [
+    'CreditTransactionManager',
+    'InsufficientCreditsError',
+    'TransactionNotFoundError',
+    'UserNotFoundError'
+]