db services

app.py

@@ -11,9 +11,11 @@ from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
-from core.database import init_db, DB_FILENAME
+from core.database import engine, DB_FILENAME
 from routers import auth, blink, contact, credits, general, gemini, payments, schema
 from services.drive_service import DriveService
+from services.db_service import init_database, reset_database
+from services.db_service.register_config import register_db_service_config
 
 # Configure logging
 logging.basicConfig(
@@ -33,12 +35,20 @@ async def lifespan(app: FastAPI):
     """
     logger.info("Starting up - initializing database...")
     
+    # Register DB Service configuration
+    register_db_service_config()
+    logger.info("✅ DB Service configured")
+    
     # Check for RESET_DB environment variable
     if os.getenv("RESET_DB", "").lower() == "true":
         logger.warning(f"RESET_DB is set to true. Skipping download and clearing local database ({DB_FILENAME}).")
         if os.path.exists(DB_FILENAME):
             os.remove(DB_FILENAME)
             logger.info("Local database deleted.")
+        
+        # Reset database (drop + create all tables)
+        await reset_database(engine)
+        logger.info("✅ Database reset complete")
     else:
         # Startup: Download DB from Drive ONLY if local file doesn't exist
         if not os.path.exists(DB_FILENAME):
@@ -46,9 +56,10 @@ async def lifespan(app: FastAPI):
             drive_service.download_db()
         else:
             logger.info("Startup: Local DB found. Skipping download to preserve data.")
-    
-    await init_db()
-    logger.info("Database initialized successfully")
+        
+        # Initialize database (create tables if not exist)
+        await init_database(engine)
+        logger.info("✅ Database initialized")
     
     # Start background job worker
     from services.gemini_job_worker import start_worker, stop_worker

core/models.py

@@ -47,6 +47,7 @@ class User(Base):
     updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
     last_used_at = Column(DateTime(timezone=True), nullable=True)
     is_active = Column(Boolean, default=True)
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete timestamp
 
     # Relationships
     client_users = relationship("ClientUser", back_populates="user", lazy="dynamic")
@@ -86,6 +87,7 @@ class ClientUser(Base):
     # Timestamps
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     last_seen_at = Column(DateTime(timezone=True), nullable=True)
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete timestamp
 
     # Relationship
     user = relationship("User", back_populates="client_users")
@@ -131,6 +133,7 @@ class AuditLog(Base):
     
     # Timestamp
     timestamp = Column(DateTime(timezone=True), server_default=func.now(), index=True)
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete timestamp
 
     # Relationship
     user = relationship("User", back_populates="audit_logs")
@@ -174,6 +177,7 @@ class GeminiJob(Base):
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     started_at = Column(DateTime(timezone=True), nullable=True)
     completed_at = Column(DateTime(timezone=True), nullable=True)
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete timestamp
     
     # Credit tracking for reservation pattern
     credits_reserved = Column(Integer, default=0)  # Credits reserved for this job
@@ -219,6 +223,7 @@ class PaymentTransaction(Base):
     # Timestamps
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     paid_at = Column(DateTime(timezone=True), nullable=True)
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete timestamp
     
     # Metadata
     razorpay_signature = Column(String(255), nullable=True)  # For verification audit
@@ -250,6 +255,7 @@ class Contact(Base):
     message = Column(Text, nullable=False)
     ip_address = Column(String(45), nullable=True)  # IPv6 can be up to 45 chars
     created_at = Column(DateTime(timezone=True), server_default=func.now())
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete timestamp
 
     # Relationship
     user = relationship("User", back_populates="contacts")
@@ -270,6 +276,7 @@ class RateLimit(Base):
     attempts = Column(Integer, default=0)
     window_start = Column(DateTime(timezone=True), nullable=False)
     expires_at = Column(DateTime(timezone=True), nullable=False)
+    deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)  # Soft delete timestamp
 
 
 class ApiKeyUsage(Base):

routers/blink.py

@@ -11,7 +11,7 @@ import logging
 from core.database import get_db
 from core.models import User, AuditLog, GeminiJob, Contact, ClientUser
 from services.encryption_service import decrypt_multiple_blocks
-from dependencies import get_geolocation, get_optional_user
+from dependencies import get_geolocation, get_optional_user, get_current_user
 
 logger = logging.getLogger(__name__)
 
@@ -30,37 +30,36 @@ async def get_data(
     page: int = Query(1, ge=1, description="Page number"),
     limit: int = Query(100, ge=1, le=500, description="Items per page"),
     log_type: str = Query(None, description="Filter by log type: client, server"),
+    user: User = Depends(get_current_user),  # Auth required
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Get paginated audit log data.
-    Replaces old BlinkData endpoint - now uses unified AuditLog.
+    Get paginated audit log data for the authenticated user.
+    Admins see all logs from all users.
     """
+    from services.db_service import QueryService
+    
     try:
+        qs = QueryService(user, db)
         offset = (page - 1) * limit
         
         # Build query with optional filter
-        base_query = select(AuditLog)
-        count_query = select(func.count(AuditLog.id))
-        
+        query = select(AuditLog)
         if log_type:
-            base_query = base_query.where(AuditLog.log_type == log_type)
-            count_query = count_query.where(AuditLog.log_type == log_type)
+            query = query.where(AuditLog.log_type == log_type)
         
-        # Get total count
-        total_result = await db.execute(count_query)
-        total = total_result.scalar() or 0
+        # Automatically filtered by user (unless admin)
+        query = query.order_by(AuditLog.timestamp.desc()).offset(offset).limit(limit)
+        items = await qs.select().execute(query)
         
-        # Get unique users count
-        unique_result = await db.execute(
-            select(func.count(func.distinct(AuditLog.user_id)))
-        )
-        unique_users = unique_result.scalar() or 0
+        # Count also automatically filtered
+        count_query = select(AuditLog)
+        if log_type:
+            count_query = count_query.where(AuditLog.log_type == log_type)
+        total = await qs.select().count(count_query)
         
-        # Get paginated items
-        query = base_query.order_by(AuditLog.timestamp.desc()).offset(offset).limit(limit)
-        result = await db.execute(query)
-        items = result.scalars().all()
+        # Unique users: 1 for regular users, multiple for admins
+        unique_users = 1 if not qs.select().is_admin else total
         
         return {
             "items": [
@@ -96,22 +95,29 @@ async def get_data(
 async def get_users(
     page: int = Query(1, ge=1, description="Page number"),
     limit: int = Query(50, ge=1, le=500, description="Items per page"),
+    user: User = Depends(get_current_user),  # Auth required
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Get paginated users data.
+    Get current user's profile data.
+    Admins see paginated list of all users.
     """
+    from services.db_service import QueryService
+    
     try:
-        offset = (page - 1) * limit
-        
-        # Get total count
-        total_result = await db.execute(select(func.count(User.id)))
-        total = total_result.scalar() or 0
-        
-        # Get paginated items
-        query = select(User).order_by(User.id.desc()).offset(offset).limit(limit)
-        result = await db.execute(query)
-        items = result.scalars().all()
+        qs = QueryService(user, db)
+        
+        if qs.select().is_admin:
+            # Admins get paginated list of all users
+            offset = (page - 1) * limit
+            query = select(User).order_by(User.id.desc()).offset(offset).limit(limit)
+            items = await qs.select().execute(query)  # No filter for admin
+            total = await qs.select().count(select(User))
+        else:
+            # Regular users only see their own profile
+            query = select(User)
+            items = await qs.select().execute(query)  # Filtered to current user
+            total = 1
         
         return {
             "items": [
@@ -145,31 +151,23 @@ async def get_client_users(
     page: int = Query(1, ge=1, description="Page number"),
     limit: int = Query(50, ge=1, le=500, description="Items per page"),
     user_id: str = Query(None, description="Filter by server user_id"),
+    user: User = Depends(get_current_user),  # Auth required
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Get paginated client user mappings.
-    Shows how server users map to multiple client identifiers.
+    Get current user's client mappings.
+    Admins see all client mappings from all users.
     """
+    from services.db_service import QueryService
+    
     try:
+        qs = QueryService(user, db)
         offset = (page - 1) * limit
         
-        # Build query with optional filter
-        base_query = select(ClientUser)
-        count_query = select(func.count(ClientUser.id))
-        
-        if user_id:
-            base_query = base_query.where(ClientUser.user_id == user_id)
-            count_query = count_query.where(ClientUser.user_id == user_id)
-        
-        # Get total count
-        total_result = await db.execute(count_query)
-        total = total_result.scalar() or 0
-        
-        # Get paginated items
-        query = base_query.order_by(ClientUser.id.desc()).offset(offset).limit(limit)
-        result = await db.execute(query)
-        items = result.scalars().all()
+        # Automatically filtered by user (unless admin)
+        query = select(ClientUser).order_by(ClientUser.id.desc()).offset(offset).limit(limit)
+        items = await qs.select().execute(query)
+        total = await qs.select().count(select(ClientUser))
         
         return {
             "items": [
@@ -203,34 +201,37 @@ async def get_audit_logs(
     limit: int = Query(50, ge=1, le=500, description="Items per page"),
     log_type: str = Query(None, description="Filter by log type: client, server"),
     action: str = Query(None, description="Filter by action"),
+    user: User = Depends(get_current_user),  # Auth required
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Get paginated audit logs with optional filters.
+    Get current user's audit logs with optional filters.
+    Admins see all logs from all users.
     """
+    from services.db_service import QueryService
+    
     try:
+        qs = QueryService(user, db)
         offset = (page - 1) * limit
         
         # Build query with filters
-        base_query = select(AuditLog)
-        count_query = select(func.count(AuditLog.id))
+        query = select(AuditLog)
+        if log_type:
+            query = query.where(AuditLog.log_type == log_type)
+        if action:
+            query = query.where(AuditLog.action == action)
         
+        # Automatically filtered by user (unless admin)
+        query = query.order_by(AuditLog.timestamp.desc()).offset(offset).limit(limit)
+        items = await qs.select().execute(query)
+        
+        # Count with same filters
+        count_query = select(AuditLog)
         if log_type:
-            base_query = base_query.where(AuditLog.log_type == log_type)
             count_query = count_query.where(AuditLog.log_type == log_type)
-        
         if action:
-            base_query = base_query.where(AuditLog.action == action)
             count_query = count_query.where(AuditLog.action == action)
-        
-        # Get total count
-        total_result = await db.execute(count_query)
-        total = total_result.scalar() or 0
-        
-        # Get paginated items
-        query = base_query.order_by(AuditLog.timestamp.desc()).offset(offset).limit(limit)
-        result = await db.execute(query)
-        items = result.scalars().all()
+        total = await qs.select().count(count_query)
         
         return {
             "items": [
@@ -264,22 +265,25 @@ async def get_audit_logs(
 async def get_gemini_jobs(
     page: int = Query(1, ge=1, description="Page number"),
     limit: int = Query(50, ge=1, le=500, description="Items per page"),
+    user: User = Depends(get_current_user),  # Auth required
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Get paginated Gemini jobs.
+    Get current user's Gemini jobs.
+    Admins see all jobs from all users.
     """
+    from services.db_service import QueryService
+    
     try:
+        qs = QueryService(user, db)
         offset = (page - 1) * limit
         
-        # Get total count
-        total_result = await db.execute(select(func.count(GeminiJob.id)))
-        total = total_result.scalar() or 0
-        
-        # Get paginated items
+        # Automatically filtered by user (unless admin)
         query = select(GeminiJob).order_by(GeminiJob.id.desc()).offset(offset).limit(limit)
-        result = await db.execute(query)
-        items = result.scalars().all()
+        items = await qs.select().execute(query)
+        
+        # Count also automatically filtered
+        total = await qs.select().count(select(GeminiJob))
         
         return {
             "items": [
@@ -311,32 +315,35 @@ async def get_gemini_jobs(
 async def get_payment_transactions(
     page: int = Query(1, ge=1, description="Page number"),
     limit: int = Query(50, ge=1, le=500, description="Items per page"),
+    user: User = Depends(get_current_user),  # Auth required
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Get paginated payment transactions.
+    Get current user's payment transactions.
+    Admins see all transactions from all users.
     """
     from core.models import PaymentTransaction
+    from services.db_service import QueryService
     
     try:
+        qs = QueryService(user, db)
         offset = (page - 1) * limit
         
-        # Get total count
-        total_result = await db.execute(select(func.count(PaymentTransaction.id)))
-        total = total_result.scalar() or 0
+        # Automatically filtered by user (unless admin)
+        total = await qs.select().count(select(PaymentTransaction))
         
-        # Get total revenue (paid transactions only)
-        revenue_result = await db.execute(
-            select(func.sum(PaymentTransaction.amount_paise)).where(
-                PaymentTransaction.status == "paid"
-            )
+        # Get paid transactions revenue - automatically filtered
+        revenue_query = select(func.sum(PaymentTransaction.amount_paise)).where(
+            PaymentTransaction.status == "paid"
         )
-        total_revenue_paise = revenue_result.scalar() or 0
+        revenue_result = await qs.execute_one(revenue_query)
+        total_revenue_paise = revenue_result or 0
         
-        # Get paginated items
-        query = select(PaymentTransaction).order_by(PaymentTransaction.id.desc()).offset(offset).limit(limit)
-        result = await db.execute(query)
-        items = result.scalars().all()
+        # Get paginated items - automatically filtered
+        query = select(PaymentTransaction).order_by(
+            PaymentTransaction.id.desc()
+        ).offset(offset).limit(limit)
+        items = await qs.select().execute(query)
         
         return {
             "items": [
@@ -378,22 +385,23 @@ async def get_payment_transactions(
 async def get_contacts(
     page: int = Query(1, ge=1, description="Page number"),
     limit: int = Query(50, ge=1, le=500, description="Items per page"),
+    user: User = Depends(get_current_user),  # Auth required
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Get paginated contact form submissions.
+    Get current user's contact form submissions.
+    Admins see all contact submissions from all users.
     """
+    from services.db_service import QueryService
+    
     try:
+        qs = QueryService(user, db)
         offset = (page - 1) * limit
         
-        # Get total count
-        total_result = await db.execute(select(func.count(Contact.id)))
-        total = total_result.scalar() or 0
-        
-        # Get paginated items
+        # Automatically filtered by user (unless admin)
         query = select(Contact).order_by(Contact.id.desc()).offset(offset).limit(limit)
-        result = await db.execute(query)
-        items = result.scalars().all()
+        items = await qs.select().execute(query)
+        total = await qs.select().count(select(Contact))
         
         return {
             "items": [

routers/gemini.py

@@ -516,16 +516,20 @@ async def delete_job(
     db: AsyncSession = Depends(get_db)
 ):
     """
-    Delete a job.
-    - If queued: Refund 8 credits (10 cost - 2 penalty), delete job.
-    - If processing/completed/failed: Delete job (no refund).
+    Soft delete a job with conditional credit refund.
+    
+    Refund policy:
+    - If queued: Refund 8 credits (10 cost - 2 penalty), soft delete job.
+    - If processing/completed/failed: Soft delete job (no refund).
     """
-    query = select(GeminiJob).where(
-        GeminiJob.job_id == job_id,
-        GeminiJob.user_id == user.id  # Integer FK comparison
+    from services.db_service import QueryService
+    
+    qs = QueryService(user, db)
+    
+    # Get job (automatically filtered to user's job)
+    job = await qs.select().execute_one(
+        select(GeminiJob).where(GeminiJob.job_id == job_id)
     )
-    result = await db.execute(query)
-    job = result.scalar_one_or_none()
     
     if not job:
         raise HTTPException(
@@ -548,23 +552,37 @@ async def delete_job(
         # Job has third_party_id but is queued? (Unlikely for video, but maybe for others?)
         # Or maybe it was reset to queued?
         # Use existing logic: Refund 8 credits (10 - 2) for video
-        if job.credits_reserved >= 10 and not job.credits_refunded:
-            refund_amount = 8
+        penalty = 2  # Fixed penalty
+        refund_amount = max(0, job.credits_reserved - penalty)
+        
+        if refund_amount > 0 and not job.credits_refunded:
             user.credits += refund_amount
-            # Don't mark as fully refunded, as it's partial? 
-            # Actually credits_refunded is boolean. Maybe we shouldn't set it if partial?
-            # But we gave back credits. Let's just update user credits.
-            message = f"Job deleted. {refund_amount} credits refunded."
-             
-    await db.delete(job)
-    await db.commit()
+            job.credits_refunded = True
+            message = f"Job deleted. {refund_amount} credits refunded (queued status)."
+        else:
+            message = "Job deleted (no refund - already refunded or 0 credit job)."
+    elif job.status in ["processing", "completed", "failed", "expired"]:
+        # No refund for jobs that started or completed
+        message = f"Job deleted (no refund for {job.status} jobs)."
+    
+    # Commit credit refund if any
+    if refund_amount > 0:
+        await db.commit()
+    
+    # Soft delete the job using DeleteQuery
+    deleted = await qs.delete().soft_delete_one(job)
+    
+    if not deleted:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to delete job"
+        )
     
     return {
         "success": True,
-        "job_id": job_id,
         "message": message,
-        "credits_refunded": refund_amount,
-        "credits_remaining": user.credits
+        "refund_amount": refund_amount,
+        "new_credit_balance": user.credits
     }
 
 

services/db_service/__init__.py

@@ -0,0 +1,59 @@
+"""
+DB Service - Database access layer with automatic filtering and access control.
+
+Provides:
+- DBServiceConfig: Configuration registration (call at startup)
+- QueryService: Main entry point
+- SelectQuery: Read operations
+- UpdateQuery: Update operations
+- DeleteQuery: Soft delete operations
+- BaseQuery: Shared filtering logic
+
+Usage:
+    # 1. Register configuration at startup
+    from services.db_service import DBServiceConfig
+    from core.models import User, GeminiJob, ...
+    
+    DBServiceConfig.register(
+        user_filter_column="user_id",
+        soft_delete_column="deleted_at",
+        user_read_scoped=[User, GeminiJob, ...],
+        ...
+    )
+    
+    # 2. Use in endpoints
+    from services.db_service import QueryService
+    
+    qs = QueryService(user, db)
+    jobs = await qs.select().execute(select(GeminiJob))
+"""
+
+from services.db_service.config import DBServiceConfig
+from services.db_service.query_service import QueryService, get_query_service
+from services.db_service.select_query import SelectQuery
+from services.db_service.update_query import UpdateQuery
+from services.db_service.delete_query import DeleteQuery
+from services.db_service.base_query import BaseQuery
+from services.db_service.db_init import (
+    init_database,
+    drop_database,
+    reset_database,
+    get_registered_models,
+    get_model_by_name,
+)
+
+__all__ = [
+    'DBServiceConfig',
+    'QueryService',
+    'get_query_service',
+    'SelectQuery',
+    'UpdateQuery',
+    'DeleteQuery',
+    'BaseQuery',
+    # Database initialization
+    'init_database',
+    'drop_database',
+    'reset_database',
+    'get_registered_models',
+    'get_model_by_name',
+]

services/db_service/base_query.py

@@ -0,0 +1,233 @@
+"""
+BaseQuery - Base class for all query types with shared filtering logic.
+
+Uses DBServiceConfig for plug-and-play configuration of models and columns.
+"""
+
+import os
+import logging
+from typing import Type
+from fastapi import HTTPException, status as http_status
+from sqlalchemy import Select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.models import User
+from services.db_service.config import DBServiceConfig
+
+logger = logging.getLogger(__name__)
+
+
+class BaseQuery:
+    """
+    Base class for all query operations.
+    
+    Uses DBServiceConfig for model scopes and column names.
+    Configuration must be registered at application startup.
+    """
+    
+    def __init__(self, user: User, db: AsyncSession, is_system: bool = False):
+        """Initialize base query with user and database session."""
+        # Ensure config is registered
+        DBServiceConfig.assert_registered()
+        
+        self.user = user
+        self.db = db
+        self._is_system = is_system
+        self._config = DBServiceConfig
+    
+    @property
+    def is_admin(self) -> bool:
+        """Check if current user is an admin."""
+        admin_emails_str = os.getenv("ADMIN_EMAILS", "")
+        if not admin_emails_str:
+            return False
+        
+        admin_emails = [email.strip() for email in admin_emails_str.split(",")]
+        is_admin = self.user.email in admin_emails
+        
+        if is_admin:
+            logger.info(f"Admin access granted for {self.user.email}")
+        
+        return is_admin
+    
+    def _verify_operation_access(self, model_class: Type, operation: str) -> None:
+        """
+        Check if user has permission for this operation on this model.
+        
+        Permission hierarchy: SYSTEM > ADMIN > USER
+        Uses DBServiceConfig for scope checking.
+        """
+        # System operations have highest priority
+        if self._is_system:
+            logger.info(f"System operation: {operation} on {model_class.__name__}")
+            return
+        
+        # Admins can do anything
+        if self.is_admin:
+            return
+        
+        # Map operation to config scope sets
+        admin_only_sets = {
+            'read': self._config.admin_read_only,
+            'create': self._config.admin_create_only,
+            'update': self._config.admin_update_only,
+            'delete': self._config.admin_delete_only,
+        }
+        
+        user_allowed_sets = {
+            'read': self._config.user_read_scoped,
+            'create': self._config.user_create_scoped,
+            'update': self._config.user_update_scoped,
+            'delete': self._config.user_delete_scoped,
+        }
+        
+        system_only_sets = {
+            'read': self._config.system_read_scoped,
+            'create': self._config.system_create_scoped,
+            'update': self._config.system_update_scoped,
+            'delete': self._config.system_delete_scoped,
+        }
+        
+        admin_only = admin_only_sets.get(operation, set())
+        user_allowed = user_allowed_sets.get(operation, set())
+        system_only = system_only_sets.get(operation, set())
+        
+        # Check if this is system-only operation
+        if model_class in system_only and model_class not in user_allowed and model_class not in admin_only:
+            raise HTTPException(
+                status_code=http_status.HTTP_403_FORBIDDEN,
+                detail=f"Only system processes can {operation} {model_class.__name__}"
+            )
+        
+        # Check if admin-only
+        if model_class in admin_only:
+            raise HTTPException(
+                status_code=http_status.HTTP_403_FORBIDDEN,
+                detail=f"Only administrators can {operation} {model_class.__name__}"
+            )
+        
+        # Check if user is allowed
+        if model_class not in user_allowed:
+            raise HTTPException(
+                status_code=http_status.HTTP_403_FORBIDDEN,
+                detail=f"You do not have permission to {operation} {model_class.__name__}"
+            )
+    
+    def _apply_ownership_filter(self, stmt, model_class: Type, operation: str):
+        """
+        Shared method to apply ownership filter for UPDATE/DELETE operations.
+        Uses DBServiceConfig.user_filter_column for filtering.
+        """
+        # Admins can modify all records
+        if self.is_admin:
+            logger.info(f"Admin {self.user.email} {operation}ing {model_class.__name__} records")
+            return stmt
+        
+        # Non-admins can only modify their own records
+        filter_column = self._config.user_filter_column
+        if hasattr(model_class, filter_column):
+            user_id_col = getattr(model_class, filter_column)
+            stmt = stmt.where(user_id_col == self.user.id)
+            logger.info(f"User {self.user.email} {operation}ing own {model_class.__name__} records")
+        
+        return stmt
+    
+    def _verify_admin_access(self, query: Select) -> None:
+        """
+        Check if query is for admin-only models (READ operation).
+        Uses DBServiceConfig.admin_read_only.
+        """
+        if self.is_admin:
+            return
+        
+        # Check if querying admin-only models
+        if hasattr(query, 'column_descriptions'):
+            for description in query.column_descriptions:
+                entity = description.get('entity') or description.get('type')
+                if entity in self._config.admin_read_only:
+                    raise HTTPException(
+                        status_code=http_status.HTTP_403_FORBIDDEN,
+                        detail=f"Only administrators can read {entity.__name__}"
+                    )
+        
+        # Check froms for admin-only models
+        for from_clause in query.froms:
+            table_class = from_clause.entity_namespace.get('__class__')
+            if table_class in self._config.admin_read_only:
+                raise HTTPException(
+                    status_code=http_status.HTTP_403_FORBIDDEN,
+                    detail=f"Only administrators can read {table_class.__name__}"
+                )
+    
+    def _apply_user_filter(self, query: Select) -> Select:
+        """
+        Automatically apply user_id filter to READ queries.
+        Uses DBServiceConfig for model scopes and filter column.
+        """
+        # First check if this is an admin-only query
+        self._verify_admin_access(query)
+        
+        # Admins see all data
+        if self.is_admin:
+            logger.debug(f"Admin query - no user filter applied")
+            return query
+        
+        # Get filter column from config
+        filter_column = self._config.user_filter_column
+        special_user_model = self._config.special_user_model
+        user_id_column = self._config.user_id_column
+        
+        # Detect which model is being queried
+        if hasattr(query, 'column_descriptions'):
+            for description in query.column_descriptions:
+                entity = description.get('entity') or description.get('type')
+                if entity in self._config.user_read_scoped:
+                    logger.debug(f"Applying user filter to {entity.__name__} query")
+                    # Special handling for User model (uses id instead of user_id)
+                    if entity == special_user_model:
+                        user_col = getattr(entity, user_id_column)
+                        return query.where(user_col == self.user.id)
+                    # Standard user_id filtering
+                    if hasattr(entity, filter_column):
+                        user_col = getattr(entity, filter_column)
+                        return query.where(user_col == self.user.id)
+        
+        # Check froms
+        for from_clause in query.froms:
+            table_class = from_clause.entity_namespace.get('__class__')
+            if table_class in self._config.user_read_scoped:
+                logger.debug(f"Applying user filter to {table_class.__name__} query")
+                # Special handling for User model
+                if table_class == special_user_model:
+                    user_col = getattr(table_class, user_id_column)
+                    return query.where(user_col == self.user.id)
+                # Standard user_id filtering
+                if hasattr(table_class, filter_column):
+                    user_col = getattr(table_class, filter_column)
+                    return query.where(user_col == self.user.id)
+        
+        return query
+    
+    def _filter_deleted(self, query: Select) -> Select:
+        """
+        Add filter to exclude soft-deleted records.
+        Uses DBServiceConfig.soft_delete_column.
+        """
+        delete_column = self._config.soft_delete_column
+        
+        # Detect which model is being queried
+        if hasattr(query, 'column_descriptions'):
+            for description in query.column_descriptions:
+                entity = description.get('entity') or description.get('type')
+                if entity and hasattr(entity, delete_column):
+                    deleted_at_col = getattr(entity, delete_column)
+                    return query.where(deleted_at_col == None)
+        
+        # Check froms
+        for from_clause in query.froms:
+            table_class = from_clause.entity_namespace.get('__class__')
+            if table_class and hasattr(table_class, delete_column):
+                deleted_at_col = getattr(table_class, delete_column)
+                return query.where(deleted_at_col == None)
+        
+        return query

services/db_service/config.py

@@ -0,0 +1,135 @@
+"""
+DBServiceConfig - Configuration for DB Service permissions and filtering.
+
+Allows application layer to register model scopes and column names,
+making the DB service completely generic and plug-and-play.
+"""
+
+import logging
+from typing import Type, Set, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class DBServiceConfig:
+    """
+    Centralized configuration for DB Service.
+    
+    Register your application's models and permissions at startup.
+    """
+    
+    # Configuration state
+    _registered = False
+    
+    # Database metadata
+    db_base = None  # SQLAlchemy declarative base
+    all_models: list = []  # All model classes
+    
+    # Column names (generic)
+    user_filter_column: str = "user_id"
+    user_id_column: str = "id"
+    soft_delete_column: str = "deleted_at"
+    
+    # Special models
+    special_user_model: Optional[Type] = None
+    
+    # USER scopes
+    user_read_scoped: Set[Type] = set()
+    user_create_scoped: Set[Type] = set()
+    user_update_scoped: Set[Type] = set()
+    user_delete_scoped: Set[Type] = set()
+    
+    # ADMIN scopes
+    admin_read_only: Set[Type] = set()
+    admin_create_only: Set[Type] = set()
+    admin_update_only: Set[Type] = set()
+    admin_delete_only: Set[Type] = set()
+    
+    # SYSTEM scopes
+    system_read_scoped: Set[Type] = set()
+    system_create_scoped: Set[Type] = set()
+    system_update_scoped: Set[Type] = set()
+    system_delete_scoped: Set[Type] = set()
+    
+    @classmethod
+    def register(
+        cls,
+        # Database metadata
+        db_base=None,          # SQLAlchemy Base (for table creation)
+        all_models: list = None,  # All model classes
+        # Column names
+        user_filter_column: str = "user_id",
+        user_id_column: str = "id",
+        soft_delete_column: str = "deleted_at",
+        # Special models
+        special_user_model: Optional[Type] = None,
+        # USER scopes
+        user_read_scoped: list = None,
+        user_create_scoped: list = None,
+        user_update_scoped: list = None,
+        user_delete_scoped: list = None,
+        # ADMIN scopes
+        admin_read_only: list = None,
+        admin_create_only: list = None,
+        admin_update_only: list = None,
+        admin_delete_only: list = None,
+        # SYSTEM scopes
+        system_read_scoped: list = None,
+        system_create_scoped: list = None,
+        system_update_scoped: list = None,
+        system_delete_scoped: list = None,
+    ) -> None:
+        """Register DB Service configuration at application startup."""
+        # Database metadata
+        cls.db_base = db_base
+        cls.all_models = all_models or []
+        
+        # Column names
+        cls.user_filter_column = user_filter_column
+        cls.user_id_column = user_id_column
+        cls.soft_delete_column = soft_delete_column
+        
+        # Special models
+        cls.special_user_model = special_user_model
+        
+        # USER scopes
+        cls.user_read_scoped = set(user_read_scoped or [])
+        cls.user_create_scoped = set(user_create_scoped or [])
+        cls.user_update_scoped = set(user_update_scoped or [])
+        cls.user_delete_scoped = set(user_delete_scoped or [])
+        
+        # ADMIN scopes
+        cls.admin_read_only = set(admin_read_only or [])
+        cls.admin_create_only = set(admin_create_only or [])
+        cls.admin_update_only = set(admin_update_only or [])
+        cls.admin_delete_only = set(admin_delete_only or [])
+        
+        # SYSTEM scopes
+        cls.system_read_scoped = set(system_read_scoped or [])
+        cls.system_create_scoped = set(system_create_scoped or [])
+        cls.system_update_scoped = set(system_update_scoped or [])
+        cls.system_delete_scoped = set(system_delete_scoped or [])
+        
+        cls._registered = True
+        
+        logger.info("✅ DBServiceConfig registered successfully")
+        logger.info(f"   Models registered: {len(cls.all_models)}")
+        logger.info(f"   User filter column: {cls.user_filter_column}")
+        logger.info(f"   Soft delete column: {cls.soft_delete_column}")
+        logger.info(f"   USER scopes: {len(cls.user_read_scoped)} read, {len(cls.user_create_scoped)} create")
+        logger.info(f"   ADMIN scopes: {len(cls.admin_read_only)} read, {len(cls.admin_create_only)} create")
+        logger.info(f"   SYSTEM scopes: {len(cls.system_read_scoped)} read, {len(cls.system_create_scoped)} create")
+    
+    @classmethod
+    def is_registered(cls) -> bool:
+        """Check if configuration has been registered."""
+        return cls._registered
+    
+    @classmethod
+    def assert_registered(cls) -> None:
+        """Assert that configuration has been registered, raise if not."""
+        if not cls._registered:
+            raise RuntimeError(
+                "DBServiceConfig not registered! "
+                "Call DBServiceConfig.register() at application startup."
+            )

services/db_service/db_init.py

@@ -0,0 +1,74 @@
+"""
+DBService Database Initialization
+
+Provides utilities for creating and managing database tables.
+"""
+
+import logging
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from services.db_service.config import DBServiceConfig
+
+logger = logging.getLogger(__name__)
+
+
+async def init_database(engine: AsyncEngine) -> None:
+    """
+    Initialize database tables based on registered models.
+    
+    Creates all tables defined in DBServiceConfig.all_models.
+    """
+    DBServiceConfig.assert_registered()
+    
+    if not DBServiceConfig.db_base:
+        raise RuntimeError(
+            "No database base registered! "
+            "Pass db_base parameter to DBServiceConfig.register()"
+        )
+    
+    logger.info("Creating database tables...")
+    
+    async with engine.begin() as conn:
+        await conn.run_sync(DBServiceConfig.db_base.metadata.create_all)
+    
+    model_count = len(DBServiceConfig.all_models)
+    logger.info(f"✅ Database initialized with {model_count} models")
+
+
+async def drop_database(engine: AsyncEngine) -> None:
+    """Drop all database tables. WARNING: Deletes all data!"""
+    DBServiceConfig.assert_registered()
+    
+    if not DBServiceConfig.db_base:
+        raise RuntimeError("No database base registered!")
+    
+    logger.warning("⚠️  Dropping all database tables...")
+    
+    async with engine.begin() as conn:
+        await conn.run_sync(DBServiceConfig.db_base.metadata.drop_all)
+    
+    logger.info("✅ All tables dropped")
+
+
+async def reset_database(engine: AsyncEngine) -> None:
+    """Reset database (drop + create). WARNING: Deletes all data!"""
+    await drop_database(engine)
+    await init_database(engine)
+    logger.info("✅ Database reset complete")
+
+
+def get_registered_models() -> list:
+    """Get list of all registered models."""
+    DBServiceConfig.assert_registered()
+    return DBServiceConfig.all_models
+
+
+def get_model_by_name(model_name: str):
+    """Get model class by name."""
+    DBServiceConfig.assert_registered()
+    
+    for model in DBServiceConfig.all_models:
+        if model.__name__ == model_name:
+            return model
+    
+    return None

services/db_service/delete_query.py

@@ -0,0 +1,160 @@
+"""
+DeleteQuery - Soft delete operations with access control.
+
+Inherits from BaseQuery for shared filtering logic.
+"""
+
+import logging
+from typing import Type, TypeVar, List
+from fastapi import HTTPException, status as http_status
+from sqlalchemy import update, select
+from sqlalchemy.sql import func
+
+from services.db_service.base_query import BaseQuery
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar('T')
+
+
+class DeleteQuery(BaseQuery):
+    """
+    Handles soft DELETE operations.
+    
+    Inherits filtering logic from BaseQuery:
+    - User ownership checks (_is_admin)
+    - Records are marked as deleted (deleted_at = NOW()) instead of being physically removed
+    """
+    
+    async def soft_delete(self, model_class: Type[T], **filters) -> int:
+        """
+        Soft delete records matching filters.
+        
+        Sets deleted_at = NOW() instead of physically deleting.
+        """
+        # Verify user has permission to delete this model
+        self._verify_operation_access(model_class, 'delete')
+        
+        # Build update statement to set deleted_at
+        delete_col = getattr(model_class, self._config.soft_delete_column)
+        stmt = update(model_class).where(
+            delete_col == None  # Only soft-delete non-deleted records
+        )
+        
+        # Apply ownership filter (shared method from BaseQuery)
+        stmt = self._apply_ownership_filter(stmt, model_class, 'delet')
+        
+        # Apply user's filters
+        for key, value in filters.items():
+            if not hasattr(model_class, key):
+                raise ValueError(f"{model_class.__name__} has no attribute '{key}'")
+            stmt = stmt.where(getattr(model_class, key) == value)
+        
+        # Set deleted_at timestamp
+        stmt = stmt.values({self._config.soft_delete_column: func.now()})
+        
+        result = await self.db.execute(stmt)
+        await self.db.commit()
+        
+        count = result.rowcount
+        logger.info(f"Soft-deleted {count} {model_class.__name__} record(s)")
+        return count
+    
+    async def soft_delete_one(self, instance: T) -> bool:
+        """Soft delete a single model instance."""
+        # Check if already deleted
+        delete_column = self._config.soft_delete_column
+        if hasattr(instance, delete_column) and getattr(instance, delete_column) is not None:
+            logger.warning(f"Attempted to delete already-deleted {instance.__class__.__name__}")
+            return False
+        
+        # Verify user has permission
+        self._verify_operation_access(instance.__class__, 'delete')
+        
+        # Check ownership for non-admins
+        filter_column = self._config.user_filter_column
+        if not self.is_admin and hasattr(instance, filter_column):
+            if getattr(instance, filter_column) != self.user.id:
+                raise HTTPException(
+                    status_code=http_status.HTTP_403_FORBIDDEN,
+                    detail="You do not have permission to delete this record"
+                )
+        
+        # Set deleted_at
+        setattr(instance, self._config.soft_delete_column, func.now())
+        await self.db.commit()
+        await self.db.refresh(instance)
+        
+        logger.info(f"Soft-deleted {instance.__class__.__name__} instance")
+        return True
+    
+    async def restore(self, model_class: Type[T], **filters) -> int:
+        """
+        Restore soft-deleted records.
+        Only admins can restore.
+        """
+        if not self.is_admin:
+            raise HTTPException(
+                status_code=http_status.HTTP_403_FORBIDDEN,
+                detail="Only administrators can restore deleted records"
+            )
+        
+        # Build update to clear deleted_at
+        delete_col = getattr(model_class, self._config.soft_delete_column)
+        stmt = update(model_class).where(
+            delete_col != None  # Only restore deleted records
+        )
+        
+        # Apply filters
+        for key, value in filters.items():
+            if not hasattr(model_class, key):
+                raise ValueError(f"{model_class.__name__} has no attribute '{key}'")
+            stmt = stmt.where(getattr(model_class, key) == value)
+        
+        # Clear deleted_at
+        stmt = stmt.values({self._config.soft_delete_column: None})
+        
+        result = await self.db.execute(stmt)
+        await self.db.commit()
+        
+        count = result.rowcount
+        logger.info(f"Admin {self.user.email} restored {count} {model_class.__name__} record(s)")
+        return count
+    
+    async def restore_one(self, instance: T) -> bool:
+        """Restore a single soft-deleted model instance. Admin only."""
+        if not self.is_admin:
+            raise HTTPException(
+                status_code=http_status.HTTP_403_FORBIDDEN,
+                detail="Only administrators can restore deleted records"
+            )
+        
+        # Check if actually deleted
+        delete_column = self._config.soft_delete_column
+        if not hasattr(instance, delete_column) or getattr(instance, delete_column) is None:
+            logger.warning(f"Attempted to restore non-deleted {instance.__class__.__name__}")
+            return False
+        
+        # Clear deleted_at
+        setattr(instance, self._config.soft_delete_column, None)
+        await self.db.commit()
+        await self.db.refresh(instance)
+        
+        logger.info(f"Admin restored {instance.__class__.__name__} instance")
+        return True
+    
+    async def list_deleted(self, model_class: Type[T], limit: int = 100) -> List[T]:
+        """List soft-deleted records. Admin only."""
+        if not self.is_admin:
+            raise HTTPException(
+                status_code=http_status.HTTP_403_FORBIDDEN,
+                detail="Only administrators can view deleted records"
+            )
+        
+        delete_col = getattr(model_class, self._config.soft_delete_column)
+        query = select(model_class).where(
+            delete_col != None
+        ).order_by(delete_col.desc()).limit(limit)
+        
+        result = await self.db.execute(query)
+        return result.scalars().all()

services/db_service/query_service.py

@@ -0,0 +1,78 @@
+"""
+QueryService - Main entry point for database operations.
+
+Factory class that provides access to SelectQuery, UpdateQuery, and DeleteQuery.
+All query types inherit from BaseQuery for shared filtering logic.
+"""
+
+import os
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.models import User
+from services.db_service.select_query import SelectQuery
+from services.db_service.update_query import UpdateQuery
+from services.db_service.delete_query import DeleteQuery
+
+
+class QueryService:
+    """
+    Main query service factory.
+    
+    Provides access to:
+    - SelectQuery: Read operations (SELECT)
+    - UpdateQuery: Update operations (UPDATE)
+    - DeleteQuery: Soft delete operations (soft DELETE)
+    
+    Usage:
+        qs = QueryService(user, db)
+        
+        # SELECT
+        jobs = await qs.select().execute(select(GeminiJob))
+        
+        # UPDATE
+        await qs.update().update(Job, {...}, id=123)
+        
+        # DELETE
+        await qs.delete().soft_delete(Job, id=123)
+        
+        # Check admin status
+        if qs.is_admin():
+            # Admin-only logic
+    """
+    
+    def __init__(self, user: User, db: AsyncSession, is_system: bool = False):
+        """
+        Initialize QueryService.
+        
+        Args:  
+            user: Current authenticated user (or system user for background ops)
+            db: Database session
+            is_system: True if this is a system/background operation (OAuth, webhooks, workers)
+        """
+        self.user = user
+        self.db = db
+        self.is_system = is_system
+    
+    def select(self) -> SelectQuery:
+        """Get SelectQuery instance for read operations."""
+        return SelectQuery(self.user, self.db, self.is_system)
+    
+    def update(self) -> UpdateQuery:
+        """Get UpdateQuery instance for update operations."""
+        return UpdateQuery(self.user, self.db, self.is_system)
+    
+    def delete(self) -> DeleteQuery:
+        """Get DeleteQuery instance for soft delete operations."""
+        return DeleteQuery(self.user, self.db, self.is_system)
+
+
+def get_query_service(user: User, db: AsyncSession) -> QueryService:
+    """
+    Factory function to create QueryService.
+    
+    Can be used as a FastAPI dependency:
+        from services.db_service import get_query_service
+        
+        qs: QueryService = Depends(lambda user=Depends(get_current_user), db=Depends(get_db): get_query_service(user, db))
+    """
+    return QueryService(user, db)

services/db_service/register_config.py

@@ -0,0 +1,121 @@
+"""
+DB Service Configuration Registration
+
+Add this to your main.py or app initialization:
+"""
+
+from services.db_service import DBServiceConfig
+from core.models import (
+    Base,  # SQLAlchemy declarative base
+    User, GeminiJob, PaymentTransaction, Contact,
+    RateLimit, ApiKeyUsage, ClientUser, AuditLog
+)
+
+
+def register_db_service_config():
+    """Register DB Service configuration at application startup."""
+    DBServiceConfig.register(
+        # Database metadata (for table creation)
+        db_base=Base,
+        all_models=[
+            User, GeminiJob, PaymentTransaction, Contact,
+            RateLimit, ApiKeyUsage, ClientUser, AuditLog
+        ],
+        
+        # Column names
+        user_filter_column="user_id",      # Column name for user ownership
+        user_id_column="id",               # Column name for user ID
+        soft_delete_column="deleted_at",   # Column name for soft delete tracking
+        
+        # Special models
+        special_user_model=User,  # Model that uses 'id' instead of 'user_id'
+        
+        # ================================================================
+        # USER SCOPES (Regular authenticated users)
+        # ================================================================
+        user_read_scoped=[
+            User,               # Users can read own profile
+            GeminiJob,          # Users can read own jobs
+            PaymentTransaction, # Users can read own payments
+            Contact,            # Users can read own contacts
+        ],
+        
+        user_create_scoped=[
+            GeminiJob,          # Users can create jobs
+            PaymentTransaction, # Users can create payments (via API)
+            Contact,            # Users can submit contact forms
+        ],
+        
+        user_update_scoped=[
+            User,               # Users can update own profile
+            GeminiJob,          # Users can update own jobs
+        ],
+        
+        user_delete_scoped=[
+            GeminiJob,          # Users can delete own jobs
+            Contact,            # Users can delete own contacts
+        ],
+        
+        # ================================================================
+        # ADMIN SCOPES (Administrators only - via ADMIN_EMAILS env var)
+        # ================================================================
+        admin_read_only=[
+            RateLimit,    # Only admins view rate limits
+            ApiKeyUsage,  # Only admins view API usage
+            ClientUser,   # Only admins view client  mappings
+            AuditLog,     # Only admins view audit logs
+        ],
+        
+        admin_create_only=[
+            RateLimit,    # Only admins create rate limits
+            ApiKeyUsage,  # Only admins create API usage entries
+            ClientUser,   # Only admins create client mappings
+            AuditLog,     # Only admins create audit entries
+        ],
+        
+        admin_update_only=[
+            RateLimit,         # Only admins update rate limits
+            ApiKeyUsage,       # Only admins update API settings
+            ClientUser,        # Only admins modify client mappings
+            PaymentTransaction,# Only admins refund/adjust payments
+        ],
+        
+        admin_delete_only=[
+            RateLimit,    # Only admins delete rate limits
+            ApiKeyUsage,  # Only admins remove API tracking
+            User,         # Only admins delete user accounts
+        ],
+        
+        # ================================================================
+        # SYSTEM SCOPES (Background processes - OAuth, webhooks, workers)
+        # ================================================================
+        system_read_scoped=[
+            User, GeminiJob, PaymentTransaction, RateLimit,
+            ApiKeyUsage, ClientUser, AuditLog,
+        ],
+        
+        system_create_scoped=[
+            User,              # OAuth creates users
+            ClientUser,        # Middleware creates client mappings
+            AuditLog,          # System creates audit entries
+            PaymentTransaction,# Webhooks create transactions
+            ApiKeyUsage,       # Middleware tracks API usage
+            GeminiJob,         # System creates jobs
+            RateLimit,         # Middleware creates rate limits
+        ],
+        
+        system_update_scoped=[
+            User,              # OAuth updates profiles
+            GeminiJob,         # Workers update job status
+            PaymentTransaction,# Webhooks update verification
+            ApiKeyUsage,       # System updates API stats
+            RateLimit,         # System updates rate counters
+            ClientUser,        # System updates mappings
+        ],
+        
+        system_delete_scoped=[
+            GeminiJob,         # Cleanup deletes expired jobs
+            RateLimit,         # Cleanup deletes old limits
+            ApiKeyUsage,       # Cleanup deletes old usage
+        ],
+    )

services/db_service/select_query.py

@@ -0,0 +1,98 @@
+"""
+SelectQuery - Read operations with access control.
+
+Inherits from BaseQuery for shared filtering logic.
+"""
+
+import logging
+from typing import TypeVar, Optional, List, Type
+from fastapi import HTTPException, status as http_status
+from sqlalchemy import Select, select, func
+
+from services.db_service.base_query import BaseQuery
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar('T')
+
+
+class SelectQuery(BaseQuery):
+    """
+    Handles SELECT operations with automatic filtering.
+    
+    Inherits filtering logic from BaseQuery:
+    - User filtering (_apply_user_filter)
+    - Deleted record filtering (_filter_deleted)
+    - Admin checks (_check_admin, _is_admin)
+    """
+    
+    async def execute(self, query: Select) -> List[T]:
+        """
+        Execute a query with automatic filtering.
+        
+        Filtering is applied automatically based on:
+        - Model type (USER_SCOPED, ADMIN_ONLY)
+        - User's admin status
+        - Deleted records (always excluded)
+        
+        Returns:
+            List of results
+        """
+        query = self._apply_user_filter(query)
+        query = self._filter_deleted(query)
+        
+        result = await self.db.execute(query)
+        return result.scalars().all()
+    
+    async def execute_one(self, query: Select) -> Optional[T]:
+        """
+        Execute a query expecting a single result.
+        Automatic filtering applied.
+        
+        Returns:
+            Single result or None
+        """
+        query = self._apply_user_filter(query)
+        query = self._filter_deleted(query)
+        
+        result = await self.db.execute(query)
+        return result.scalar_one_or_none()
+    
+    async def count(self, query: Select) -> int:
+        """
+        Count query results with automatic filtering.
+        
+        Returns:
+            Count of results
+        """
+        query = self._apply_user_filter(query)
+        query = self._filter_deleted(query)
+        
+        # Convert to count query
+        count_query = select(func.count()).select_from(query.alias())
+        result = await self.db.execute(count_query)
+        return result.scalar() or 0
+    
+    async def count_deleted(self, model_class: Type[T]) -> int:
+        """
+        Count soft-deleted records for a model.
+        Only admins can access this.
+        
+        Returns:
+            Count of deleted records
+            
+        Raises:
+            HTTPException: 403 if non-admin tries to access
+        """
+        if not self.is_admin:
+            raise HTTPException(
+                status_code=http_status.HTTP_403_FORBIDDEN,
+                detail="Only administrators can view deleted records"
+            )
+        
+        delete_col = getattr(model_class, self._config.soft_delete_column)
+        query = select(func.count()).select_from(model_class).where(
+            delete_col != None
+        )
+        result = await self.db.execute(query)
+        return result.scalar() or 0

services/db_service/update_query.py

@@ -0,0 +1,188 @@
+"""
+UpdateQuery - Update operations with access control.
+
+Inherits from BaseQuery for shared filtering logic.
+"""
+
+import logging
+from typing import Type, TypeVar, Dict, Any
+from fastapi import HTTPException, status as http_status
+from sqlalchemy import update
+
+from services.db_service.base_query import BaseQuery
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar('T')
+
+
+class UpdateQuery(BaseQuery):
+    """
+    Handles UPDATE operations with automatic access control.
+    
+    Inherits filtering logic from BaseQuery:
+    - User ownership checks (_is_admin)
+    - Deleted record protection
+    """
+    
+    async def update(self, model_class: Type[T], values: Dict[str, Any], **filters) -> int:
+        """
+        Update records matching filters with new values.
+        
+        Automatically applies ownership filter for non-admins.
+        
+        Args:
+            model_class: Model class to update
+            values: Dictionary of field=value pairs to update
+            **filters: Field=value filters to match records
+            
+        Returns:
+            Number of records updated
+            
+        Raises:
+            HTTPException: 403 if user doesn't have permission
+            ValueError: If invalid field names provided
+        """
+        # Validate that fields exist
+        for key in values.keys():
+            if not hasattr(model_class, key):
+                raise ValueError(f"{model_class.__name__} has no attribute '{key}'")
+        
+        # Verify user has permission to update this model
+        self._verify_operation_access(model_class, 'update')
+        
+        # Build update statement
+        delete_col = getattr(model_class, self._config.soft_delete_column)
+        stmt = update(model_class).where(
+            delete_col == None  # Don't update deleted records
+        )
+        
+        # Apply ownership filter (shared method from BaseQuery)
+        stmt = self._apply_ownership_filter(stmt, model_class, 'updat')
+        
+        # Apply user's filters
+        for key, value in filters.items():
+            if not hasattr(model_class, key):
+                raise ValueError(f"{model_class.__name__} has no attribute '{key}'")
+            stmt = stmt.where(getattr(model_class, key) == value)
+        
+        # Apply values
+        stmt = stmt.values(**values)
+        
+        # Execute
+        result = await self.db.execute(stmt)
+        await self.db.commit()
+        
+        count = result.rowcount
+        logger.info(f"Updated {count} {model_class.__name__} record(s)")
+        return count
+    
+    async def update_one(self, instance: T, values: Dict[str, Any]) -> T:
+        """
+        Update a single model instance with validation.
+        
+        Args:
+            instance: Model instance to update
+            values: Dictionary of field=value pairs to update
+            
+        Returns:
+            Updated model instance
+            
+        Raises:
+            HTTPException: 403 if user doesn't have permission
+            ValueError: If invalid field names provided
+        """
+        # Check if deleted
+        delete_column = self._config.soft_delete_column
+        if hasattr(instance, delete_column) and getattr(instance, delete_column) is not None:
+            raise HTTPException(
+                status_code=http_status.HTTP_400_BAD_REQUEST,
+                detail="Cannot update deleted record"
+            )
+        
+        # Verify user has permission
+        self._verify_operation_access(instance.__class__, 'update')
+        
+        # Check ownership for non-admins
+        filter_column = self._config.user_filter_column
+        if not self.is_admin and hasattr(instance, filter_column):
+            if getattr(instance, filter_column) != self.user.id:
+                raise HTTPException(
+                    status_code=http_status.HTTP_403_FORBIDDEN,
+                    detail="You do not have permission to update this record"
+                )
+        
+        # Validate fields
+        for key in values.keys():
+            if not hasattr(instance, key):
+                raise ValueError(f"{instance.__class__.__name__} has no attribute '{key}'")
+        
+        # Apply updates
+        for key, value in values.items():
+            setattr(instance, key, value)
+        
+        # Commit
+        await self.db.commit()
+        await self.db.refresh(instance)
+        
+        logger.info(f"Updated {instance.__class__.__name__} instance")
+        return instance
+    
+    async def increment(self, model_class: Type[T], field: str, amount: int = 1, **filters) -> int:
+        """
+        Increment a numeric field by a specified amount.
+        """
+        if not hasattr(model_class, field):
+            raise ValueError(f"{model_class.__name__} has no attribute '{field}'")
+        
+        self._verify_operation_access(model_class, 'update')
+        
+        field_obj = getattr(model_class, field)
+        
+        delete_col = getattr(model_class, self._config.soft_delete_column)
+        stmt = update(model_class).where(delete_col == None)
+        stmt = self._apply_ownership_filter(stmt, model_class, 'updat')
+        
+        for key, value in filters.items():
+            if not hasattr(model_class, key):
+                raise ValueError(f"{model_class.__name__} has no attribute '{key}'")
+            stmt = stmt.where(getattr(model_class, key) == value)
+        
+        stmt = stmt.values({field: field_obj + amount})
+        
+        result = await self.db.execute(stmt)
+        await self.db.commit()
+        
+        count = result.rowcount
+        logger.info(f"Incremented {field} by {amount} for {count} {model_class.__name__} record(s)")
+        return count
+    
+    async def decrement(self, model_class: Type[T], field: str, amount: int = 1, **filters) -> int:
+        """Decrement a numeric field by a specified amount."""
+        return await self.increment(model_class, field, -amount, **filters)
+    
+    async def toggle_boolean(self, model_class: Type[T], field: str, **filters) -> int:
+        """Toggle a boolean field (True -> False, False -> True)."""
+        if not hasattr(model_class, field):
+            raise ValueError(f"{model_class.__name__} has no attribute '{field}'")
+        
+        self._verify_operation_access(model_class, 'update')
+        
+        field_obj = getattr(model_class, field)
+        
+        stmt = update(model_class).where(model_class.deleted_at == None)
+        stmt = self._apply_ownership_filter(stmt, model_class, 'updat')
+        
+        for key, value in filters.items():
+            if not hasattr(model_class, key):
+                raise ValueError(f"{model_class.__name__} has no attribute '{key}'")
+            stmt = stmt.where(getattr(model_class, key) == value)
+        
+        stmt = stmt.values({field: ~field_obj})
+        
+        result = await self.db.execute(stmt)
+        await self.db.commit()
+        
+        count = result.rowcount
+        logger.info(f"Toggled {field} for {count} {model_class.__name__} record(s)")
+        return count

tests/README.md

@@ -0,0 +1,78 @@
+# DB Service - Quick Start
+
+## Run Tests
+
+```bash
+# Install test dependencies
+pip install pytest pytest-asyncio httpx
+
+# Run all tests
+ADMIN_EMAILS="admin@example.com" pytest tests/test_db_service.py -v
+
+# Run specific test
+pytest tests/test_db_service.py::TestPermissions::test_user_can_read_own_data -v
+
+# Run with coverage
+pytest tests/test_db_service.py --cov=services/db_service --cov-report=html -v
+```
+
+## Start Application
+
+```bash
+# Development (reset DB on startup)
+RESET_DB=true ADMIN_EMAILS="your@email.com" python app.py
+
+# Production (preserve DB)
+ADMIN_EMAILS="admin@example.com" python app.py
+```
+
+## Verify Integration
+
+Check startup logs for:
+```
+✅ DB Service configured
+✅ Database initialized
+✅ Database reset complete (if RESET_DB=true)
+```
+
+## Test Endpoints
+
+```bash
+# As regular user
+curl -X GET http://localhost:8000/gemini/jobs \
+  -H "Authorization: Bearer <user_token>"
+
+# As admin
+curl -X GET http://localhost:8000/blink/audit-logs \
+  -H "Authorization: Bearer <admin_token>"
+```
+
+## Troubleshooting
+
+**Config not registered:**
+```
+RuntimeError: DBServiceConfig not registered!
+```
+→ Ensure `register_db_service_config()` is called in app startup
+
+**Module not found:**
+```
+ModuleNotFoundError: No module named 'services.db_service'
+```
+→ Verify `__init__.py` exists in `services/db_service/`
+
+**Permission denied:**
+```
+403 Forbidden: Only administrators can...
+```
+→ Check `ADMIN_EMAILS` environment variable
+
+## Success Indicators
+
+✅ All files compile without errors
+✅ Application starts successfully
+✅ Database tables created
+✅ Tests pass
+✅ Endpoints return data correctly
+✅ User isolation working (users see only own data)
+✅ Admin access working (admins see all data)

tests/requirements-test.txt

@@ -0,0 +1,3 @@
+pytest>=7.4.0
+pytest-asyncio>=0.21.0
+httpx>=0.24.0

tests/test_db_service.py

@@ -0,0 +1,338 @@
+"""
+Test Suite for DB Service
+
+Comprehensive tests for the plug-and-play DB Service including:
+- Configuration
+- Permissions (USER/ADMIN/SYSTEM)
+- Filtering (user ownership, soft deletes)
+- CRUD operations
+- Database initialization
+"""
+
+import pytest
+import os
+from datetime import datetime
+from sqlalchemy import create_async_engine, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from sqlalchemy.orm import declarative_base
+
+from services.db_service import (
+    DBServiceConfig,
+    QueryService,
+    init_database,
+    reset_database,
+    get_registered_models,
+)
+from core.models import (
+    Base, User, GeminiJob, PaymentTransaction, Contact,
+    RateLimit, ApiKeyUsage, ClientUser, AuditLog
+)
+
+
+# Test database URL
+TEST_DB_URL = "sqlite+aiosqlite:///:memory:"
+
+
+@pytest.fixture
+async def engine():
+    """Create test database engine."""
+    engine = create_async_engine(TEST_DB_URL, echo=False)
+    yield engine
+    await engine.dispose()
+
+
+@pytest.fixture
+async def session(engine):
+    """Create test database session."""
+    async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    
+    async with async_session() as session:
+        yield session
+
+
+@pytest.fixture(autouse=True)
+async def setup_db(engine):
+    """Setup test database with configuration."""
+    # Register configuration
+    DBServiceConfig.register(
+        db_base=Base,
+        all_models=[User, GeminiJob, PaymentTransaction, Contact,
+                    RateLimit, ApiKeyUsage, ClientUser, AuditLog],
+        user_filter_column="user_id",
+        user_id_column="id",
+        soft_delete_column="deleted_at",
+        special_user_model=User,
+        user_read_scoped=[User, GeminiJob, PaymentTransaction, Contact],
+        user_create_scoped=[GeminiJob, PaymentTransaction, Contact],
+        user_update_scoped=[User, GeminiJob],
+        user_delete_scoped=[GeminiJob, Contact],
+        admin_read_only=[RateLimit, ApiKeyUsage, ClientUser, AuditLog],
+        admin_create_only=[RateLimit, ApiKeyUsage, ClientUser, AuditLog],
+        admin_update_only=[RateLimit, ApiKeyUsage, ClientUser, PaymentTransaction],
+        admin_delete_only=[RateLimit, ApiKeyUsage, User],
+        system_read_scoped=[User, GeminiJob, PaymentTransaction, RateLimit,
+                            ApiKeyUsage, ClientUser, AuditLog],
+        system_create_scoped=[User, ClientUser, AuditLog, PaymentTransaction,
+                              ApiKeyUsage, GeminiJob, RateLimit],
+        system_update_scoped=[User, GeminiJob, PaymentTransaction, ApiKeyUsage,
+                              RateLimit, ClientUser],
+        system_delete_scoped=[GeminiJob, RateLimit, ApiKeyUsage],
+    )
+    
+    # Initialize database
+    await init_database(engine)
+    
+    yield
+    
+    # Cleanup
+    await reset_database(engine)
+
+
+@pytest.fixture
+async def regular_user(session):
+    """Create a regular test user."""
+    user = User(
+        email="user@example.com",
+        name="Test User",
+        credits=100
+    )
+    session.add(user)
+    await session.commit()
+    await session.refresh(user)
+    return user
+
+
+@pytest.fixture
+async def admin_user(session):
+    """Create an admin test user."""
+    user = User(
+        email=os.getenv("ADMIN_EMAILS", "admin@example.com").split(",")[0],
+        name="Admin User",
+        credits=1000
+    )
+    session.add(user)
+    await session.commit()
+    await session.refresh(user)
+    return user
+
+
+@pytest.fixture
+async def other_user(session):
+    """Create another test user."""
+    user = User(
+        email="other@example.com",
+        name="Other User",
+        credits=50
+    )
+    session.add(user)
+    await session.commit()
+    await session.refresh(user)
+    return user
+
+
+# ============================================================================
+# Configuration Tests
+# ============================================================================
+
+class TestConfiguration:
+    """Test DB Service configuration."""
+    
+    def test_config_registered(self):
+        """Test that configuration is registered."""
+        assert DBServiceConfig.is_registered()
+        assert DBServiceConfig.db_base == Base
+        assert len(DBServiceConfig.all_models) == 8
+    
+    def test_get_registered_models(self):
+        """Test getting registered models."""
+        models = get_registered_models()
+        assert len(models) == 8
+        assert User in models
+        assert GeminiJob in models
+    
+    def test_column_names(self):
+        """Test configured column names."""
+        assert DBServiceConfig.user_filter_column == "user_id"
+        assert DBServiceConfig.soft_delete_column == "deleted_at"
+        assert DBServiceConfig.special_user_model == User
+
+
+# ============================================================================
+# Permission Tests
+# ============================================================================
+
+class TestPermissions:
+    """Test USER/ADMIN/SYSTEM permission hierarchy."""
+    
+    async def test_user_can_read_own_data(self, session, regular_user):
+        """Test that users can read their own data."""
+        job = GeminiJob(user_id=regular_user.id, prompt="Test", status="queued")
+        session.add(job)
+        await session.commit()
+        
+        qs = QueryService(regular_user, session)
+        jobs = await qs.select().execute(select(GeminiJob))
+        
+        assert len(jobs) == 1
+        assert jobs[0].id == job.id
+    
+    async def test_user_cannot_read_others_data(self, session, regular_user, other_user):
+        """Test that users cannot read other users' data."""
+        # Create job for other user
+        job = GeminiJob(user_id=other_user.id, prompt="Other", status="queued")
+        session.add(job)
+        await session.commit()
+        
+        # Regular user tries to read
+        qs = QueryService(regular_user, session)
+        jobs = await qs.select().execute(select(GeminiJob))
+        
+        assert len(jobs) == 0  # Should not see other user's jobs
+    
+    async def test_admin_can_read_all_data(self, session, admin_user, regular_user):
+        """Test that admins can read all users' data."""
+        # Create jobs for different users
+        job1 = GeminiJob(user_id=regular_user.id, prompt="User Job", status="queued")
+        job2 = GeminiJob(user_id=admin_user.id, prompt="Admin Job", status="queued")
+        session.add_all([job1, job2])
+        await session.commit()
+        
+        qs = QueryService(admin_user, session)
+        jobs = await qs.select().execute(select(GeminiJob))
+        
+        assert len(jobs) == 2  # Admin sees all jobs
+    
+    async def test_user_cannot_access_admin_only_models(self, session, regular_user):
+        """Test that regular users cannot access admin-only models."""
+        qs = QueryService(regular_user, session)
+        
+        with pytest.raises(Exception) as exc_info:
+            await qs.select().execute(select(RateLimit))
+        
+        assert "403" in str(exc_info.value) or "administrator" in str(exc_info.value).lower()
+    
+    async def test_admin_can_access_admin_only_models(self, session, admin_user):
+        """Test that admins can access admin-only models."""
+        rate_limit = RateLimit(identifier="test", endpoint="/api/test", request_count=10)
+        session.add(rate_limit)
+        await session.commit()
+        
+        qs = QueryService(admin_user, session)
+        limits = await qs.select().execute(select(RateLimit))
+        
+        assert len(limits) == 1
+    
+    async def test_system_can_create_user(self, session, regular_user):
+        """Test that system operations can create users."""
+        qs = QueryService(regular_user, session, is_system=True)
+        
+        # System should be able to bypass permissions
+        # (actual create would use direct SQLAlchemy, but permission check passes)
+        assert qs._is_system is True
+
+
+#  ============================================================================
+# Soft Delete Tests
+# ============================================================================
+
+class TestSoftDeletes:
+    """Test soft delete functionality."""
+    
+    async def test_soft_delete_marks_record(self, session, regular_user):
+        """Test that soft delete sets deleted_at."""
+        job = GeminiJob(user_id=regular_user.id, prompt="Delete Me", status="queued")
+        session.add(job)
+        await session.commit()
+        
+        qs = QueryService(regular_user, session)
+        await qs.delete().soft_delete_one(job)
+        
+        assert job.deleted_at is not None
+    
+    async def test_soft_deleted_not_in_query(self, session, regular_user):
+        """Test that soft-deleted records don't appear in queries."""
+        job = GeminiJob(user_id=regular_user.id, prompt="Delete Me", status="queued")
+        session.add(job)
+        await session.commit()
+        
+        qs = QueryService(regular_user, session)
+        
+        # Before delete
+        jobs = await qs.select().execute(select(GeminiJob))
+        assert len(jobs) == 1
+        
+        # After delete
+        await qs.delete().soft_delete_one(job)
+        jobs = await qs.select().execute(select(GeminiJob))
+        assert len(jobs) == 0  # Should not appear
+    
+    async def test_admin_can_restore(self, session, admin_user, regular_user):
+        """Test that admins can restore deleted records."""
+        job = GeminiJob(user_id=regular_user.id, prompt="Restore Me", status="queued")
+        session.add(job)
+        await session.commit()
+        job_id = job.id
+        
+        qs = QueryService(admin_user, session)
+        
+        # Delete
+        await qs.delete().soft_delete_one(job)
+        assert job.deleted_at is not None
+        
+        # Restore
+        await qs.delete().restore_one(job)
+        assert job.deleted_at is None
+    
+    async def test_user_cannot_restore(self, session, regular_user):
+        """Test that regular users cannot restore records."""
+        job = GeminiJob(user_id=regular_user.id, prompt="Deleted", status="queued")
+        session.add(job)
+        await session.commit()
+        
+        qs = QueryService(regular_user, session)
+        await qs.delete().soft_delete_one(job)
+        
+        with pytest.raises(Exception) as exc_info:
+            await qs.delete().restore_one(job)
+        
+        assert "403" in str(exc_info.value) or "administrator" in str(exc_info.value).lower()
+
+
+# ============================================================================
+# Database Initialization Tests
+# ============================================================================
+
+class TestDatabaseInitialization:
+    """Test database initialization utilities."""
+    
+    async def test_init_database_creates_tables(self, engine):
+        """Test that init_database creates all tables."""
+        await init_database(engine)
+        
+        # Verify tables exist by querying
+        async with AsyncSession(engine) as session:
+            result = await session.execute(select(User))
+            assert result.scalars().all() == []  # Empty but table exists
+    
+    async def test_reset_database_clears_data(self, engine, session, regular_user):
+        """Test that reset_database clears all data."""
+        # Add some data
+        user = User(email="test@example.com", name="Test", credits=10)
+        session.add(user)
+        await session.commit()
+        
+        # Reset
+        await reset_database(engine)
+        
+        # Verify data cleared
+        async with AsyncSession(engine) as new_session:
+            result = await new_session.execute(select(User))
+            assert len(result.scalars().all()) == 0
+
+
+# ============================================================================
+# Run Tests
+# ============================================================================
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v", "--tb=short"])