feat(sync): Implement warehouse synchronization from MongoDB to PostgreSQL

- Added WarehouseSyncService for managing async sync operations.
- Created WarehouseSyncHandler for handling warehouse data fetching and validation.
- Introduced models for field mapping and required fields for warehouse data.
- Developed service methods for upserting warehouse data into PostgreSQL.
- Implemented async task queue and background workers for processing sync operations.
- Updated warehouse service to trigger async sync on create, update, and delete operations.
- Created SQL script for the new scm_warehouse_ref table in PostgreSQL.
- Added migration scripts for updating existing warehouse documents in MongoDB.
- Implemented tests for verifying warehouse sync functionality.
- Updated documentation to reflect changes in MongoDB collection naming and sync operations.

app/sync/warehouses/__init__.py

@@ -0,0 +1,4 @@
+"""Warehouse sync package."""
+from .service import WarehouseSyncService
+
+__all__ = ["WarehouseSyncService"]

app/sync/warehouses/handler.py

@@ -0,0 +1,253 @@
+"""
+Warehouse sync handler for MongoDB to PostgreSQL synchronization.
+"""
+from typing import Dict, Any, Optional
+from sqlalchemy.ext.asyncio import AsyncConnection
+from sqlalchemy import text
+from motor.motor_asyncio import AsyncIOMotorDatabase
+from datetime import datetime
+from app.core.logging import get_logger
+
+from app.sync.common.handler import SyncHandler
+from app.sync.warehouses.models import WAREHOUSE_FIELD_MAPPING, WAREHOUSE_REQUIRED_FIELDS
+from app.warehouses.constants import SCM_WAREHOUSES_COLLECTION
+
+logger = get_logger(__name__)
+
+
+class WarehouseSyncHandler(SyncHandler):
+    """
+    Handler for syncing warehouse data from MongoDB to PostgreSQL.
+    
+    Implements entity-specific logic for warehouse synchronization including
+    field mapping, validation, and upsert operations.
+    """
+    
+    def __init__(self):
+        super().__init__(entity_type="warehouse")
+    
+    async def fetch_from_mongodb(
+        self,
+        entity_id: str,
+        mongo_db: AsyncIOMotorDatabase
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Fetch warehouse from MongoDB by warehouse_id.
+        
+        Args:
+            entity_id: warehouse_id to fetch
+            mongo_db: MongoDB database instance
+            
+        Returns:
+            Warehouse document or None if not found
+        """
+        try:
+            collection = mongo_db[SCM_WAREHOUSES_COLLECTION]
+            warehouse = await collection.find_one({"warehouse_id": entity_id})
+            return warehouse
+        except Exception as e:
+            logger.error(
+                "Error fetching warehouse from MongoDB",
+                exc_info=e,
+                extra={
+                    "entity_type": self.entity_type,
+                    "entity_id": entity_id
+                }
+            )
+            raise
+    
+    def get_field_mapping(self) -> Dict[str, str]:
+        """
+        Get field mapping from MongoDB to PostgreSQL.
+        
+        Returns:
+            Dictionary mapping MongoDB field names to PostgreSQL column names
+        """
+        return WAREHOUSE_FIELD_MAPPING
+    
+    def validate_required_fields(self, entity: Dict[str, Any]) -> bool:
+        """
+        Validate that all required fields are present in warehouse document.
+        
+        Required fields: warehouse_id, warehouse_code, warehouse_name, 
+                        warehouse_type, merchant_id, merchant_type, status
+        
+        Args:
+            entity: Warehouse document from MongoDB
+            
+        Returns:
+            True if all required fields present, False otherwise
+        """
+        missing_fields = []
+        
+        for field in WAREHOUSE_REQUIRED_FIELDS:
+            if field not in entity or entity[field] is None:
+                missing_fields.append(field)
+        
+        if missing_fields:
+            logger.error(
+                "Warehouse missing required fields",
+                extra={
+                    "entity_type": self.entity_type,
+                    "entity_id": entity.get("warehouse_id"),
+                    "missing_fields": missing_fields
+                }
+            )
+            return False
+        
+        return True
+    
+    def transform_field_value(self, field_name: str, value: Any) -> Any:
+        """
+        Transform field value for PostgreSQL.
+        
+        Handles type conversions and nested field extraction.
+        
+        Args:
+            field_name: Name of the field
+            value: Value from MongoDB
+            
+        Returns:
+            Transformed value for PostgreSQL
+        """
+        # Convert warehouse_type enum to string if needed
+        if field_name == "warehouse_type" and hasattr(value, 'value'):
+            return value.value
+        
+        # Convert status enum to string if needed
+        if field_name == "status" and hasattr(value, 'value'):
+            return value.value
+        
+        # Handle datetime objects
+        if isinstance(value, datetime):
+            return value
+        
+        return value
+    
+    def extract_nested_fields(self, entity: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Extract nested fields from warehouse document.
+        
+        Extracts:
+        - pincode from address
+        - contact_name, contact_phone, contact_email from contact
+        
+        Args:
+            entity: Warehouse document from MongoDB
+            
+        Returns:
+            Dictionary with flattened fields
+        """
+        flattened = {}
+        
+        # Extract address fields (only pincode)
+        if "address" in entity and entity["address"]:
+            address = entity["address"]
+            flattened["pincode"] = address.get("pincode")
+        else:
+            flattened["pincode"] = None
+        
+        # Extract contact fields
+        if "contact" in entity and entity["contact"]:
+            contact = entity["contact"]
+            flattened["contact_name"] = contact.get("name")
+            flattened["contact_phone"] = contact.get("phone")
+            flattened["contact_email"] = contact.get("email")
+        else:
+            flattened["contact_name"] = None
+            flattened["contact_phone"] = None
+            flattened["contact_email"] = None
+        
+        return flattened
+    
+    async def upsert_to_postgres(
+        self,
+        entity: Dict[str, Any],
+        pg_conn: AsyncConnection
+    ) -> bool:
+        """
+        Upsert warehouse to PostgreSQL trans.scm_warehouse_ref table.
+        
+        Performs timestamp-based conflict resolution:
+        - If record doesn't exist, insert
+        - If MongoDB updated_at >= PostgreSQL updated_at, update
+        - Otherwise, skip update
+        
+        Args:
+            entity: Warehouse document from MongoDB
+            pg_conn: PostgreSQL connection
+            
+        Returns:
+            True if upsert successful, False otherwise
+        """
+        try:
+            warehouse_id = entity["warehouse_id"]
+            updated_at = entity.get("updated_at") or entity.get("created_at")
+            
+            # Check timestamp conflict
+            should_update = await self.check_timestamp_conflict(
+                entity_id=warehouse_id,
+                mongo_updated_at=updated_at,
+                pg_conn=pg_conn,
+                table_name="trans.scm_warehouse_ref",
+                id_column="warehouse_id"
+            )
+            
+            if not should_update:
+                logger.debug(
+                    "Skipping warehouse sync due to timestamp conflict",
+                    extra={
+                        "entity_type": self.entity_type,
+                        "entity_id": warehouse_id,
+                        "mongo_updated_at": updated_at
+                    }
+                )
+                return True  # Not an error, just skipped
+            
+            # Extract nested fields
+            nested_fields = self.extract_nested_fields(entity)
+            
+            # Merge nested fields into entity for mapping
+            entity_with_nested = {**entity, **nested_fields}
+            
+            # Map fields
+            mapped_entity = self.map_fields(entity_with_nested)
+            
+            # Build UPSERT query
+            columns = list(mapped_entity.keys())
+            placeholders = [f":{col}" for col in columns]
+            
+            # Build UPDATE clause (exclude primary key)
+            update_columns = [col for col in columns if col != "warehouse_id"]
+            update_clause = ", ".join([f"{col} = EXCLUDED.{col}" for col in update_columns])
+            
+            query = text(f"""
+                INSERT INTO trans.scm_warehouse_ref ({', '.join(columns)})
+                VALUES ({', '.join(placeholders)})
+                ON CONFLICT (warehouse_id)
+                DO UPDATE SET {update_clause}
+            """)
+            
+            await pg_conn.execute(query, mapped_entity)
+            
+            logger.debug(
+                "Warehouse upserted to PostgreSQL",
+                extra={
+                    "entity_type": self.entity_type,
+                    "entity_id": warehouse_id
+                }
+            )
+            
+            return True
+            
+        except Exception as e:
+            logger.error(
+                "Error upserting warehouse to PostgreSQL",
+                exc_info=e,
+                extra={
+                    "entity_type": self.entity_type,
+                    "entity_id": entity.get("warehouse_id"),
+                    "error": str(e)
+                }
+            )
+            raise

app/sync/warehouses/models.py

@@ -0,0 +1,31 @@
+"""
+Data models for warehouse sync operations.
+"""
+
+# Field mapping from MongoDB to PostgreSQL
+WAREHOUSE_FIELD_MAPPING = {
+    "warehouse_id": "warehouse_id",           # TEXT PRIMARY KEY
+    "warehouse_code": "warehouse_code",       # TEXT NOT NULL
+    "warehouse_name": "warehouse_name",       # TEXT NOT NULL
+    "warehouse_type": "warehouse_type",       # TEXT NOT NULL
+    "merchant_id": "merchant_id",             # TEXT NOT NULL
+    "merchant_type": "merchant_type",         # TEXT NOT NULL
+    "status": "status",                       # TEXT NOT NULL
+    "pincode": "pincode",                     # TEXT (from address.pincode)
+    "contact_name": "contact_name",           # TEXT (from contact.name)
+    "contact_phone": "contact_phone",         # TEXT (from contact.phone)
+    "contact_email": "contact_email",         # TEXT (from contact.email)
+    "created_at": "created_at",               # TIMESTAMP NOT NULL
+    "updated_at": "updated_at"                # TIMESTAMP
+}
+
+# Required fields that must be present
+WAREHOUSE_REQUIRED_FIELDS = [
+    "warehouse_id",
+    "warehouse_code",
+    "warehouse_name",
+    "warehouse_type",
+    "merchant_id",
+    "merchant_type",
+    "status"
+]

app/sync/warehouses/service.py

@@ -0,0 +1,341 @@
+"""
+Warehouse sync service with async task queue and background workers.
+"""
+import asyncio
+from typing import Optional
+from motor.motor_asyncio import AsyncIOMotorDatabase
+from app.core.logging import get_logger
+import time
+import traceback
+
+from app.sync.warehouses.handler import WarehouseSyncHandler
+from app.sync.common.retry import RetryManager
+from app.sync.common.monitoring import SyncMonitoringService
+from app.sync.common.models import SyncOperation
+from app.sql import async_engine
+
+logger = get_logger(__name__)
+
+
+class WarehouseSyncService:
+    """
+    Service for asynchronously syncing warehouses from MongoDB to PostgreSQL.
+    
+    Features:
+    - Async task queue for non-blocking sync operations
+    - Background worker for processing sync queue
+    - Retry logic with exponential backoff
+    - Monitoring and metrics tracking
+    """
+    
+    def __init__(
+        self,
+        mongo_db: AsyncIOMotorDatabase,
+        max_queue_size: int = 1000,
+        worker_count: int = 5,
+        max_retries: int = 3
+    ):
+        """
+        Initialize warehouse sync service.
+        
+        Args:
+            mongo_db: MongoDB database instance
+            max_queue_size: Maximum size of sync queue
+            worker_count: Number of background workers
+            max_retries: Maximum retry attempts for failed syncs
+        """
+        self.mongo_db = mongo_db
+        self.max_queue_size = max_queue_size
+        self.worker_count = worker_count
+        
+        # Initialize components
+        self.handler = WarehouseSyncHandler()
+        self.retry_manager = RetryManager(max_retries=max_retries)
+        self.monitoring = SyncMonitoringService()
+        
+        # Async queue for sync operations
+        self.sync_queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue_size)
+        
+        # Background workers
+        self.workers: list[asyncio.Task] = []
+        self.is_running = False
+    
+    async def start_workers(self) -> None:
+        """
+        Start background workers for processing sync queue.
+        
+        Creates worker_count background tasks that continuously process
+        the sync queue.
+        """
+        if self.is_running:
+            logger.warning("Warehouse sync workers already running")
+            return
+        
+        self.is_running = True
+        
+        logger.info(
+            "Starting warehouse sync workers",
+            extra={
+                "worker_count": self.worker_count,
+                "max_queue_size": self.max_queue_size
+            }
+        )
+        
+        for i in range(self.worker_count):
+            worker = asyncio.create_task(self._worker(worker_id=i))
+            self.workers.append(worker)
+        
+        logger.info(f"Started {self.worker_count} warehouse sync workers")
+    
+    async def stop_workers(self) -> None:
+        """
+        Stop all background workers gracefully.
+        
+        Waits for workers to finish processing current operations
+        before shutting down.
+        """
+        if not self.is_running:
+            logger.warning("Warehouse sync workers not running")
+            return
+        
+        logger.info("Stopping warehouse sync workers")
+        
+        self.is_running = False
+        
+        # Cancel all workers
+        for worker in self.workers:
+            worker.cancel()
+        
+        # Wait for workers to finish
+        await asyncio.gather(*self.workers, return_exceptions=True)
+        
+        self.workers.clear()
+        
+        logger.info("Warehouse sync workers stopped")
+    
+    async def sync_warehouse(
+        self,
+        warehouse_id: str,
+        operation: str = "update"
+    ) -> None:
+        """
+        Queue a warehouse for synchronization.
+        
+        This is a fire-and-forget operation that adds the warehouse to the
+        sync queue for async processing. Does not wait for sync completion.
+        
+        Args:
+            warehouse_id: ID of warehouse to sync
+            operation: Type of operation ("create" | "update")
+        """
+        try:
+            sync_op = SyncOperation(
+                entity_type="warehouse",
+                entity_id=warehouse_id,
+                operation=operation
+            )
+            
+            # Try to add to queue without blocking
+            try:
+                self.sync_queue.put_nowait(sync_op)
+                logger.debug(
+                    "Warehouse queued for sync",
+                    extra={
+                        "entity_type": "warehouse",
+                        "entity_id": warehouse_id,
+                        "operation": operation,
+                        "queue_size": self.sync_queue.qsize()
+                    }
+                )
+            except asyncio.QueueFull:
+                # Queue is full, drop oldest and add new
+                logger.warning(
+                    "Sync queue full, dropping oldest operation",
+                    extra={
+                        "entity_type": "warehouse",
+                        "entity_id": warehouse_id,
+                        "queue_size": self.sync_queue.qsize()
+                    }
+                )
+                
+                # Drop oldest (get without waiting)
+                try:
+                    dropped_op = self.sync_queue.get_nowait()
+                    logger.warning(
+                        "Dropped sync operation",
+                        extra={
+                            "entity_type": dropped_op.entity_type,
+                            "entity_id": dropped_op.entity_id
+                        }
+                    )
+                except asyncio.QueueEmpty:
+                    pass
+                
+                # Add new operation
+                self.sync_queue.put_nowait(sync_op)
+        
+        except Exception as e:
+            logger.error(
+                "Error queuing warehouse for sync",
+                exc_info=e,
+                extra={
+                    "entity_type": "warehouse",
+                    "entity_id": warehouse_id,
+                    "error": str(e)
+                }
+            )
+    
+    async def _worker(self, worker_id: int) -> None:
+        """
+        Background worker that processes sync queue.
+        
+        Continuously pulls operations from the queue and processes them
+        with retry logic and monitoring.
+        
+        Args:
+            worker_id: Unique identifier for this worker
+        """
+        logger.info(f"Warehouse sync worker {worker_id} started")
+        
+        while self.is_running:
+            try:
+                # Get next operation from queue (with timeout)
+                try:
+                    sync_op = await asyncio.wait_for(
+                        self.sync_queue.get(),
+                        timeout=1.0
+                    )
+                except asyncio.TimeoutError:
+                    # No operations in queue, continue loop
+                    continue
+                
+                # Process the sync operation
+                await self._process_sync_operation(sync_op, worker_id)
+                
+                # Mark task as done
+                self.sync_queue.task_done()
+                
+            except asyncio.CancelledError:
+                logger.info(f"Warehouse sync worker {worker_id} cancelled")
+                break
+            except Exception as e:
+                logger.error(
+                    f"Error in warehouse sync worker {worker_id}",
+                    exc_info=e,
+                    extra={
+                        "worker_id": worker_id,
+                        "error": str(e)
+                    }
+                )
+                # Continue processing despite error
+                await asyncio.sleep(1.0)
+        
+        logger.info(f"Warehouse sync worker {worker_id} stopped")
+    
+    async def _process_sync_operation(
+        self,
+        sync_op: SyncOperation,
+        worker_id: int
+    ) -> None:
+        """
+        Process a single sync operation with retry logic.
+        
+        Args:
+            sync_op: Sync operation to process
+            worker_id: ID of worker processing this operation
+        """
+        start_time = time.time()
+        
+        try:
+            logger.debug(
+                f"Worker {worker_id} processing warehouse sync",
+                extra={
+                    "worker_id": worker_id,
+                    "entity_type": sync_op.entity_type,
+                    "entity_id": sync_op.entity_id,
+                    "operation": sync_op.operation
+                }
+            )
+            
+            # Execute sync with retry logic
+            await self.retry_manager.execute_with_retry(
+                self._sync_warehouse_with_connection,
+                sync_op.entity_id,
+                entity_type="warehouse",
+                entity_id=sync_op.entity_id
+            )
+            
+            # Record success
+            duration_ms = (time.time() - start_time) * 1000
+            self.monitoring.record_sync_success(
+                entity_type="warehouse",
+                entity_id=sync_op.entity_id,
+                duration_ms=duration_ms
+            )
+            
+        except Exception as e:
+            # Record failure
+            error_msg = str(e)
+            stack_trace = traceback.format_exc()
+            
+            self.monitoring.record_sync_failure(
+                entity_type="warehouse",
+                entity_id=sync_op.entity_id,
+                error=error_msg,
+                stack_trace=stack_trace
+            )
+            
+            logger.error(
+                f"Worker {worker_id} failed to sync warehouse after retries",
+                exc_info=e,
+                extra={
+                    "worker_id": worker_id,
+                    "entity_type": sync_op.entity_type,
+                    "entity_id": sync_op.entity_id,
+                    "error": error_msg
+                }
+            )
+    
+    async def _sync_warehouse_with_connection(self, warehouse_id: str) -> bool:
+        """
+        Sync a single warehouse with connection management.
+        
+        Acquires a PostgreSQL connection, performs the sync, and releases
+        the connection.
+        
+        Args:
+            warehouse_id: ID of warehouse to sync
+            
+        Returns:
+            True if sync successful
+            
+        Raises:
+            Exception: If sync fails
+        """
+        async with async_engine.begin() as pg_conn:
+            # Perform sync
+            success = await self.handler.sync(
+                entity_id=warehouse_id,
+                mongo_db=self.mongo_db,
+                pg_conn=pg_conn
+            )
+            
+            return success
+    
+    def get_queue_size(self) -> int:
+        """
+        Get current size of sync queue.
+        
+        Returns:
+            Number of operations in queue
+        """
+        return self.sync_queue.qsize()
+    
+    def get_metrics(self) -> dict:
+        """
+        Get sync metrics for warehouse operations.
+        
+        Returns:
+            Dictionary containing sync metrics
+        """
+        return self.monitoring.get_entity_metrics("warehouse")

app/warehouses/README.md

@@ -5,7 +5,7 @@
 Complete CRUD operations for warehouse management in the SCM system.
 
 ## Database Collection
-- MongoDB Collection: `scm_warehouses`
+- MongoDB Collection: `scm_warehouse`
 
 ## Data Structure
 
@@ -328,7 +328,7 @@ All operations are tracked in the audit object:
 - Router: `app/warehouses/controllers/router.py`
 
 ### Collection Name
-- MongoDB: `scm_warehouses`
+- MongoDB: `scm_warehouse`
 
 ### Warehouse ID Generation
 - Format: `wh_{12_char_hex_id}`

app/warehouses/constants.py

@@ -3,4 +3,4 @@ Warehouse-related constants.
 """
 
 # Collection names
-SCM_WAREHOUSES_COLLECTION = "scm_warehouses"
+SCM_WAREHOUSES_COLLECTION = "scm_warehouse"

app/warehouses/services/service.py

@@ -16,6 +16,23 @@ from app.warehouses.schemas.schema import (
     WarehouseResponse,
 )
 
+# Import sync service
+_warehouse_sync_service = None
+
+
+def get_warehouse_sync_service():
+    """Get or initialize the warehouse sync service."""
+    global _warehouse_sync_service
+    if _warehouse_sync_service is None:
+        from app.sync.warehouses.service import WarehouseSyncService
+        _warehouse_sync_service = WarehouseSyncService(
+            mongo_db=get_database(),
+            max_queue_size=1000,
+            worker_count=3,
+            max_retries=3
+        )
+    return _warehouse_sync_service
+
 logger = get_logger(__name__)
 
 
@@ -141,6 +158,16 @@ class WarehouseService:
                 "created_by": payload.created_by
             }
         )
+        
+        # Trigger async sync to PostgreSQL
+        try:
+            sync_service = get_warehouse_sync_service()
+            await sync_service.sync_warehouse(warehouse_id, operation="create")
+            logger.debug(f"Warehouse {warehouse_id} queued for sync")
+        except Exception as e:
+            logger.warning(f"Failed to queue warehouse sync: {e}", exc_info=e)
+            # Don't fail the operation if sync fails
+        
         return WarehouseResponse(**warehouse_dict)
         
 
@@ -210,6 +237,15 @@ class WarehouseService:
                 extra={"warehouse_id": warehouse_id, "fields_updated": list(update_data.keys())}
             )
             
+            # Trigger async sync to PostgreSQL
+            try:
+                sync_service = get_warehouse_sync_service()
+                await sync_service.sync_warehouse(warehouse_id, operation="update")
+                logger.debug(f"Warehouse {warehouse_id} queued for sync")
+            except Exception as e:
+                logger.warning(f"Failed to queue warehouse sync: {e}", exc_info=e)
+                # Don't fail the operation if sync fails
+            
             return WarehouseResponse(**updated_warehouse)
             
         except HTTPException:
@@ -445,6 +481,15 @@ class WarehouseService:
             
             logger.info(f"Deleted (soft) warehouse {warehouse_id}")
             
+            # Trigger async sync to PostgreSQL to update status
+            try:
+                sync_service = get_warehouse_sync_service()
+                await sync_service.sync_warehouse(warehouse_id, operation="update")
+                logger.debug(f"Warehouse {warehouse_id} queued for sync after delete")
+            except Exception as e:
+                logger.warning(f"Failed to queue warehouse sync: {e}", exc_info=e)
+                # Don't fail the operation if sync fails
+            
             return {"message": f"Warehouse {warehouse_id} has been deleted"}
             
         except HTTPException:

create_warehouse_ref_table.sql

@@ -0,0 +1,176 @@
+-- =====================================================
+-- Warehouse Reference Table Creation Script
+-- =====================================================
+-- This script creates the scm_warehouse_ref table in the trans schema
+-- for syncing warehouse data from MongoDB to PostgreSQL
+
+-- Create trans schema if it doesn't exist
+CREATE SCHEMA IF NOT EXISTS trans;
+
+-- =====================================================
+-- SCM Warehouse Reference Table
+-- =====================================================
+-- This table stores warehouse reference data synced from MongoDB
+-- Used for fast lookups and reporting in PostgreSQL
+CREATE TABLE IF NOT EXISTS trans.scm_warehouse_ref (
+    -- Primary identifiers
+    warehouse_id TEXT PRIMARY KEY,
+    warehouse_code TEXT NOT NULL,
+    warehouse_name TEXT NOT NULL,
+    warehouse_type TEXT NOT NULL,
+    
+    -- Merchant information
+    merchant_id TEXT NOT NULL,
+    merchant_type TEXT NOT NULL,
+    
+    -- Status
+    status TEXT NOT NULL,
+    
+    pincode TEXT,
+    
+    -- Contact information (flattened from nested structure)
+    contact_name TEXT,
+    contact_phone TEXT,
+    contact_email TEXT,
+    
+     -- Audit fields
+    created_at TIMESTAMPTZ NOT NULL,
+    updated_at TIMESTAMPTZ,
+    
+    -- Constraints
+    CONSTRAINT chk_warehouse_status CHECK (status IN ('draft', 'active', 'inactive')),
+    CONSTRAINT chk_warehouse_type CHECK (warehouse_type IN ('MAIN', 'SECONDARY', 'STORE', 'TEMP')),
+    CONSTRAINT chk_merchant_type CHECK (merchant_type IN ('ncnf', 'cnf', 'distributor', 'retail', 'company'))
+);
+
+-- =====================================================
+-- Indexes for Performance
+-- =====================================================
+
+-- Index on merchant_id for filtering warehouses by merchant
+CREATE INDEX IF NOT EXISTS idx_scm_warehouse_ref_merchant_id 
+ON trans.scm_warehouse_ref (merchant_id);
+
+-- Index on warehouse_code for unique code lookups
+CREATE INDEX IF NOT EXISTS idx_scm_warehouse_ref_warehouse_code 
+ON trans.scm_warehouse_ref (warehouse_code);
+
+-- Index on status for filtering active/inactive warehouses
+CREATE INDEX IF NOT EXISTS idx_scm_warehouse_ref_status 
+ON trans.scm_warehouse_ref (status);
+
+-- Composite index for merchant + status queries
+CREATE INDEX IF NOT EXISTS idx_scm_warehouse_ref_merchant_status 
+ON trans.scm_warehouse_ref (merchant_id, status);
+
+-- Index on warehouse_type for filtering by type
+CREATE INDEX IF NOT EXISTS idx_scm_warehouse_ref_warehouse_type 
+ON trans.scm_warehouse_ref (warehouse_type);
+
+-- =====================================================
+-- Comments for Documentation
+-- =====================================================
+
+COMMENT ON TABLE trans.scm_warehouse_ref IS 
+'Warehouse reference table synced from MongoDB. Contains flattened warehouse data for fast lookups and reporting.';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.warehouse_id IS 
+'Unique warehouse identifier (synced from MongoDB)';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.warehouse_code IS 
+'Unique warehouse code (e.g., WH-DIST-0001)';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.warehouse_name IS 
+'Display name of the warehouse';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.warehouse_type IS 
+'Type of warehouse: MAIN, SECONDARY, STORE, TEMP';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.merchant_id IS 
+'Merchant who owns this warehouse';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.merchant_type IS 
+'Type of merchant: ncnf, cnf, distributor, retail, company';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.status IS 
+'Warehouse status: draft, active, inactive';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.city IS 
+'City from warehouse address';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.state IS 
+'State from warehouse address';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.pincode IS 
+'Pincode from warehouse address';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.contact_name IS 
+'Contact person name';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.contact_phone IS 
+'Contact phone number';pincode IS 
+'Pincode from warehouse address';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.contact_name IS 
+'Contact person name';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.contact_phone IS 
+'Contact phone number';
+
+COMMENT ON COLUMN trans.scm_warehouse_ref.contact_email IS 
+'Contact email address
+    column_name, 
+    data_type, 
+    is_nullable,
+    column_default
+FROM information_schema.columns 
+WHERE table_schema = 'trans' 
+  AND table_name = 'scm_warehouse_ref'
+ORDER BY ordinal_position;
+
+-- Check indexes
+SELECT 
+    indexname,
+    indexdef
+FROM pg_indexes 
+WHERE schemaname = 'trans' 
+  AND tablename = 'scm_warehouse_ref';
+
+-- Check constraints
+SELECT 
+    conname AS constraint_name,
+    contype AS constraint_type,
+    pg_get_constraintdef(oid) AS constraint_definition
+FROM pg_constraint 
+WHERE conrelid = 'trans.scm_warehouse_ref'::regclass;
+
+-- Check row count (should be 0 after creation)
+SELECT COUNT(*) as warehouse_ref_count 
+FROM trans.scm_warehouse_ref;
+
+-- =====================================================
+-- Sample Query Examples
+-- =====================================================
+
+-- Get all active warehouses for a merchant
+-- SELECT * FROM trans.scm_warehouse_ref 
+-- WHERE merchant_id = 'mch_DIST_0004' AND status = 'active';
+
+-- Get all warehouses in a specific city
+-- SELECT * FROM trans.scm_warehouse_ref 
+-- WHERE city = 'Mumbai' AND status = 'active';
+
+-- Get all warehouses that can receive goods
+-- SELECT * FROM trans.scm_warehouse_ref 
+-- WHERE can_receive = true AND status = 'active';
+
+-- Get warehouses by type
+-- SELECT * FROM trans.scm_warehouse_ref 
+-- WHERE warehouse_type = 'MAIN' AND status = 'active';
+warehouses by type
+-- SELECT * FROM trans.scm_warehouse_ref 
+-- WHERE warehouse_type = 'MAIN' AND status = 'active';
+
+-- Get warehouses by pincode
+-- SELECT * FROM trans.scm_warehouse_ref 
+-- WHERE pincode = '400001

docs/database/migrations/migration_warehouse_fields.sql

@@ -2,7 +2,7 @@
 -- This script should be run in MongoDB shell or via MongoDB Compass
 
 -- Update warehouses missing created_at field
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "created_at": { $exists: false } },
   { 
     $set: { 
@@ -13,7 +13,7 @@ db.scm_warehouses.updateMany(
 );
 
 -- Update warehouses with null created_at
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "created_at": null },
   { 
     $set: { 
@@ -24,31 +24,31 @@ db.scm_warehouses.updateMany(
 );
 
 -- Update warehouses missing created_by field
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "created_by": { $exists: false } },
   { $set: { "created_by": "system" } }
 );
 
 -- Update warehouses with null created_by
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "created_by": null },
   { $set: { "created_by": "system" } }
 );
 
 -- Update warehouses missing updated_at field
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "updated_at": { $exists: false } },
   { $set: { "updated_at": null } }
 );
 
 -- Update warehouses missing updated_by field
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "updated_by": { $exists: false } },
   { $set: { "updated_by": null } }
 );
 
 -- Update warehouses missing capabilities field
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "capabilities": { $exists: false } },
   { 
     $set: { 
@@ -64,7 +64,7 @@ db.scm_warehouses.updateMany(
 );
 
 -- Update warehouses with null capabilities
-db.scm_warehouses.updateMany(
+db.scm_warehouse.updateMany(
   { "capabilities": null },
   { 
     $set: { 
@@ -83,25 +83,25 @@ db.scm_warehouses.updateMany(
 print("=== Migration Verification ===");
 
 print("Warehouses missing created_at:");
-db.scm_warehouses.countDocuments({ "created_at": { $exists: false } });
+db.scm_warehouse.countDocuments({ "created_at": { $exists: false } });
 
 print("Warehouses with null created_at:");
-db.scm_warehouses.countDocuments({ "created_at": null });
+db.scm_warehouse.countDocuments({ "created_at": null });
 
 print("Warehouses missing created_by:");
-db.scm_warehouses.countDocuments({ "created_by": { $exists: false } });
+db.scm_warehouse.countDocuments({ "created_by": { $exists: false } });
 
 print("Warehouses with null created_by:");
-db.scm_warehouses.countDocuments({ "created_by": null });
+db.scm_warehouse.countDocuments({ "created_by": null });
 
 print("Warehouses missing capabilities:");
-db.scm_warehouses.countDocuments({ "capabilities": { $exists: false } });
+db.scm_warehouse.countDocuments({ "capabilities": { $exists: false } });
 
 print("Warehouses with null capabilities:");
-db.scm_warehouses.countDocuments({ "capabilities": null });
+db.scm_warehouse.countDocuments({ "capabilities": null });
 
 print("Total warehouses:");
-db.scm_warehouses.countDocuments({});
+db.scm_warehouse.countDocuments({});
 
 print("=== Sample Documents ===");
-db.scm_warehouses.find({}).limit(2).pretty();
+db.scm_warehouse.find({}).limit(2).pretty();

test_warehouse_sync.py

@@ -0,0 +1,229 @@
+"""
+Test script for warehouse sync functionality.
+
+This script tests the warehouse sync implementation by:
+1. Creating a test warehouse in MongoDB
+2. Verifying sync to PostgreSQL
+3. Updating the warehouse
+4. Verifying sync of updates
+5. Deleting the warehouse
+6. Verifying sync of delete (status change)
+"""
+import asyncio
+from datetime import datetime, timezone
+from sqlalchemy import text
+
+from app.nosql import get_database
+from app.sql import async_engine
+from app.warehouses.services.service import WarehouseService, get_warehouse_sync_service
+from app.warehouses.schemas.schema import WarehouseCreate, WarehouseUpdate
+from app.warehouses.models.model import AddressModel, GeoModel, ContactModel
+
+
+async def test_warehouse_sync():
+    """Test warehouse sync functionality."""
+    
+    print("\n" + "="*60)
+    print("WAREHOUSE SYNC TEST")
+    print("="*60 + "\n")
+    
+    # Test data
+    test_warehouse_code = f"WH-TEST-{datetime.now().strftime('%Y%m%d%H%M%S')}"
+    test_warehouse_id = None
+    
+    try:
+        # ===== TEST 1: Create Warehouse =====
+        print("TEST 1: Creating warehouse...")
+        
+        payload = WarehouseCreate(
+            warehouse_code=test_warehouse_code,
+            warehouse_name="Test Sync Warehouse",
+            warehouse_type="MAIN",
+            address=AddressModel(
+                line1="123 Test Street",
+                line2="Suite 100",
+                city="Mumbai",
+                state="Maharashtra",
+                country="India",
+                pincode="400001"
+            ),
+            geo=GeoModel(
+                lat=19.0760,
+                lng=72.8777
+            ),
+            contact=ContactModel(
+                name="Test Contact",
+                phone="+91-9876543210",
+                email="test@example.com"
+            ),
+            notes="Test warehouse for sync verification",
+            merchant_id="mch_test_001",
+            merchant_type="distributor",
+            created_by="test_user"
+        )
+        
+        warehouse = await WarehouseService.create_warehouse(payload)
+        test_warehouse_id = warehouse.warehouse_id
+        
+        print(f"✓ Warehouse created: {test_warehouse_id}")
+        print(f"  Code: {warehouse.warehouse_code}")
+        print(f"  Name: {warehouse.warehouse_name}")
+        print(f"  Status: {warehouse.status}")
+        
+        # Wait for async sync
+        print("\n  Waiting for sync (3 seconds)...")
+        await asyncio.sleep(3)
+        
+        # Check PostgreSQL
+        print("\n  Checking PostgreSQL...")
+        async with async_engine.begin() as conn:
+            result = await conn.execute(
+                text("SELECT * FROM trans.scm_warehouse_ref WHERE warehouse_id = :id"),
+                {"id": test_warehouse_id}
+            )
+            row = result.fetchone()
+            
+            if row:
+                print("  ✓ Warehouse synced to PostgreSQL")
+                print(f"    - warehouse_code: {row[1]}")
+                print(f"    - warehouse_name: {row[2]}")
+                print(f"    - city: {row[8]}")
+                print(f"    - status: {row[6]}")
+            else:
+                print("  ✗ Warehouse NOT found in PostgreSQL")
+                return False
+        
+        # ===== TEST 2: Update Warehouse =====
+        print("\n" + "-"*60)
+        print("TEST 2: Updating warehouse...")
+        
+        update_payload = WarehouseUpdate(
+            warehouse_name="Test Sync Warehouse Updated",
+            notes="Updated notes for sync test",
+            updated_by="test_user"
+        )
+        
+        updated_warehouse = await WarehouseService.update_warehouse(
+            warehouse_id=test_warehouse_id,
+            payload=update_payload,
+            merchant_id="mch_test_001"
+        )
+        
+        print(f"✓ Warehouse updated: {test_warehouse_id}")
+        print(f"  New name: {updated_warehouse.warehouse_name}")
+        
+        # Wait for async sync
+        print("\n  Waiting for sync (3 seconds)...")
+        await asyncio.sleep(3)
+        
+        # Check PostgreSQL
+        print("\n  Checking PostgreSQL...")
+        async with async_engine.begin() as conn:
+            result = await conn.execute(
+                text("SELECT warehouse_name, updated_at FROM trans.scm_warehouse_ref WHERE warehouse_id = :id"),
+                {"id": test_warehouse_id}
+            )
+            row = result.fetchone()
+            
+            if row and row[0] == "Test Sync Warehouse Updated":
+                print("  ✓ Warehouse update synced to PostgreSQL")
+                print(f"    - warehouse_name: {row[0]}")
+                print(f"    - updated_at: {row[1]}")
+            else:
+                print("  ✗ Warehouse update NOT synced")
+                return False
+        
+        # ===== TEST 3: Delete Warehouse (Soft Delete) =====
+        print("\n" + "-"*60)
+        print("TEST 3: Deleting warehouse (soft delete)...")
+        
+        result = await WarehouseService.delete_warehouse(
+            warehouse_id=test_warehouse_id,
+            merchant_id="mch_test_001"
+        )
+        
+        print(f"✓ Warehouse deleted: {result['message']}")
+        
+        # Wait for async sync
+        print("\n  Waiting for sync (3 seconds)...")
+        await asyncio.sleep(3)
+        
+        # Check PostgreSQL
+        print("\n  Checking PostgreSQL...")
+        async with async_engine.begin() as conn:
+            result = await conn.execute(
+                text("SELECT status FROM trans.scm_warehouse_ref WHERE warehouse_id = :id"),
+                {"id": test_warehouse_id}
+            )
+            row = result.fetchone()
+            
+            if row and row[0] == "inactive":
+                print("  ✓ Warehouse delete (status change) synced to PostgreSQL")
+                print(f"    - status: {row[0]}")
+            else:
+                print("  ✗ Warehouse delete NOT synced")
+                return False
+        
+        # ===== TEST 4: Check Sync Metrics =====
+        print("\n" + "-"*60)
+        print("TEST 4: Checking sync metrics...")
+        
+        sync_service = get_warehouse_sync_service()
+        metrics = sync_service.get_metrics()
+        queue_size = sync_service.get_queue_size()
+        
+        print(f"✓ Sync metrics retrieved:")
+        print(f"  - Total operations: {metrics.get('total', 0)}")
+        print(f"  - Success: {metrics.get('success', 0)}")
+        print(f"  - Failure: {metrics.get('failure', 0)}")
+        print(f"  - Avg duration: {metrics.get('avg_duration_ms', 0):.2f}ms")
+        print(f"  - Queue size: {queue_size}")
+        
+        print("\n" + "="*60)
+        print("ALL TESTS PASSED ✓")
+        print("="*60 + "\n")
+        
+        return True
+        
+    except Exception as e:
+        print(f"\n✗ TEST FAILED: {str(e)}")
+        import traceback
+        traceback.print_exc()
+        return False
+        
+    finally:
+        # Cleanup: Remove test data from MongoDB and PostgreSQL
+        print("\nCleaning up test data...")
+        try:
+            if test_warehouse_id:
+                # Remove from MongoDB
+                await get_database()["scm_warehouses"].delete_one({"warehouse_id": test_warehouse_id})
+                print(f"  ✓ Removed from MongoDB: {test_warehouse_id}")
+                
+                # Remove from PostgreSQL
+                async with async_engine.begin() as conn:
+                    await conn.execute(
+                        text("DELETE FROM trans.scm_warehouse_ref WHERE warehouse_id = :id"),
+                        {"id": test_warehouse_id}
+                    )
+                print(f"  ✓ Removed from PostgreSQL: {test_warehouse_id}")
+        except Exception as e:
+            print(f"  Warning: Cleanup error: {e}")
+
+
+if __name__ == "__main__":
+    print("\nStarting warehouse sync tests...")
+    print("Make sure:")
+    print("1. MongoDB is running and accessible")
+    print("2. PostgreSQL is running and accessible")
+    print("3. trans.scm_warehouse_ref table exists")
+    print()
+    
+    result = asyncio.run(test_warehouse_sync())
+    
+    if result:
+        print("\n✓ All tests passed successfully!")
+        exit(0)
+    else:
+        print("\n✗ Tests failed!")
+        exit(1)