feat(logging): implement production-grade structured logging with JSON and rotating files

- Add JSONFormatter for structured JSON output compatible with log aggregators (ELK, CloudWatch, Datadog)
- Add ConsoleFormatter with color-coded output for local development
- Implement rotating file handlers with configurable size and backup limits
- Add separate error.log file for ERROR and CRITICAL level logs
- Create LoggerAdapter for injecting worker/entity context into log records
- Add structured startup/shutdown events with service metadata
- Extend .env.example with LOG_FORMAT, LOG_DIR, LOG_MAX_BYTES, and LOG_BACKUP_COUNT configuration options
- Update worker_manager.py to use new logging context for worker lifecycle events
- Update base_service.py to emit structured logs for sync operations
- Update main.py to initialize logging with startup event
- Enables production-ready log aggregation while maintaining human-readable console output for local development

.env.example

@@ -24,3 +24,7 @@ UOM_WORKERS=2
 
 # Logging
 LOG_LEVEL=INFO
+# LOG_FORMAT=json     # json (default, for production/log aggregators) | console (human-readable, for local dev)
+# LOG_DIR=logs        # directory for rotating log files (default: logs/)
+# LOG_MAX_BYTES=52428800   # max size per log file in bytes (default: 50 MB)
+# LOG_BACKUP_COUNT=10      # number of rotated backups to keep

app/core/logging.py

@@ -1,19 +1,326 @@
 """
-Logging configuration.
+Production-grade logging configuration for Sync Worker Microservice.
+
+Features:
+- Structured JSON output for log aggregation (ELK, CloudWatch, Datadog, etc.)
+- Rotating file handlers with size/backup limits
+- Separate error log file
+- Console handler with human-readable format for local dev
+- Worker/entity context injection via LoggerAdapter
+- Startup/shutdown structured events
 """
 import logging
+import logging.handlers
 import sys
+import os
+import json
+import traceback
+from datetime import datetime, timezone
+from typing import Optional, Any, Dict
+from pathlib import Path
+
+
+# ── Constants ────────────────────────────────────────────────────────────────
+
+SERVICE_NAME = "sync-worker-ms"
+LOG_DIR = Path(os.getenv("LOG_DIR", "logs"))
+LOG_MAX_BYTES = int(os.getenv("LOG_MAX_BYTES", 50 * 1024 * 1024))   # 50 MB
+LOG_BACKUP_COUNT = int(os.getenv("LOG_BACKUP_COUNT", "10"))
+
+
+# ── JSON Formatter ────────────────────────────────────────────────────────────
+
+class JSONFormatter(logging.Formatter):
+    """
+    Emit log records as single-line JSON objects.
+
+    Every record includes:
+        timestamp   – ISO-8601 UTC
+        level       – DEBUG / INFO / WARNING / ERROR / CRITICAL
+        logger      – dotted module name
+        message     – formatted log message
+        service     – SERVICE_NAME constant
+        pid         – process id
 
+    Any key/value pairs passed via ``extra={}`` are merged at the top level.
+    Exception info is serialised into ``exception.type``, ``exception.message``
+    and ``exception.stacktrace``.
+    """
 
-def setup_logging(level: str = "INFO"):
-    """Setup logging configuration."""
-    logging.basicConfig(
-        level=getattr(logging, level.upper()),
-        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-        handlers=[logging.StreamHandler(sys.stdout)]
+    RESERVED = frozenset(
+        {
+            "args", "created", "exc_info", "exc_text", "filename",
+            "funcName", "levelname", "levelno", "lineno", "message",
+            "module", "msecs", "msg", "name", "pathname", "process",
+            "processName", "relativeCreated", "stack_info", "thread",
+            "threadName",
+        }
     )
 
+    def format(self, record: logging.LogRecord) -> str:
+        # Base fields
+        payload: Dict[str, Any] = {
+            "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+            "service": SERVICE_NAME,
+            "pid": record.process,
+        }
+
+        # Merge caller location for WARNING and above
+        if record.levelno >= logging.WARNING:
+            payload["caller"] = f"{record.pathname}:{record.lineno}"
+
+        # Merge extra fields (skip stdlib internals)
+        for key, val in record.__dict__.items():
+            if key not in self.RESERVED and not key.startswith("_"):
+                payload[key] = val
+
+        # Serialise exception
+        if record.exc_info and record.exc_info[0] is not None:
+            exc_type, exc_value, exc_tb = record.exc_info
+            payload["exception"] = {
+                "type": exc_type.__name__,
+                "message": str(exc_value),
+                "stacktrace": traceback.format_exception(exc_type, exc_value, exc_tb),
+            }
+
+        try:
+            return json.dumps(payload, default=str)
+        except Exception:
+            # Fallback – never let the formatter crash the app
+            payload["message"] = str(record.getMessage())
+            return json.dumps(payload, default=str)
+
+
+# ── Human-readable formatter (console / local dev) ───────────────────────────
+
+class ConsoleFormatter(logging.Formatter):
+    """Coloured, human-readable formatter for local development."""
+
+    GREY    = "\x1b[38;5;240m"
+    CYAN    = "\x1b[36m"
+    YELLOW  = "\x1b[33m"
+    RED     = "\x1b[31m"
+    BOLD_RED = "\x1b[1;31m"
+    RESET   = "\x1b[0m"
+
+    LEVEL_COLOURS = {
+        logging.DEBUG:    GREY,
+        logging.INFO:     CYAN,
+        logging.WARNING:  YELLOW,
+        logging.ERROR:    RED,
+        logging.CRITICAL: BOLD_RED,
+    }
+
+    FMT = "%(asctime)s %(levelname)-8s %(name)s — %(message)s"
+
+    def format(self, record: logging.LogRecord) -> str:
+        colour = self.LEVEL_COLOURS.get(record.levelno, self.RESET)
+        formatter = logging.Formatter(
+            f"{colour}{self.FMT}{self.RESET}",
+            datefmt="%Y-%m-%d %H:%M:%S",
+        )
+        result = formatter.format(record)
+
+        # Append extra context fields inline
+        extras = {
+            k: v for k, v in record.__dict__.items()
+            if k not in JSONFormatter.RESERVED and not k.startswith("_")
+            and k not in ("color_message",)
+        }
+        if extras:
+            result += f"  {self.GREY}{extras}{self.RESET}"
+
+        return result
+
+
+# ── Setup ─────────────────────────────────────────────────────────────────────
+
+def setup_logging(level: str = "INFO") -> None:
+    """
+    Configure root logger with production-grade handlers.
+
+    Handlers created:
+    - stdout  – JSON (production) or coloured console (LOG_FORMAT=console)
+    - app.log – rotating JSON, all levels ≥ level
+    - error.log – rotating JSON, ERROR and above only
+
+    Args:
+        level: Minimum log level string (DEBUG / INFO / WARNING / ERROR)
+    """
+    numeric_level = getattr(logging, level.upper(), logging.INFO)
+    log_format = os.getenv("LOG_FORMAT", "json").lower()
+
+    # Ensure log directory exists
+    LOG_DIR.mkdir(parents=True, exist_ok=True)
+
+    root = logging.getLogger()
+    root.setLevel(numeric_level)
+
+    # Remove any handlers added by earlier basicConfig calls
+    root.handlers.clear()
+
+    # ── stdout handler ────────────────────────────────────────────────────────
+    stdout_handler = logging.StreamHandler(sys.stdout)
+    stdout_handler.setLevel(numeric_level)
+    if log_format == "console":
+        stdout_handler.setFormatter(ConsoleFormatter())
+    else:
+        stdout_handler.setFormatter(JSONFormatter())
+    root.addHandler(stdout_handler)
+
+    # ── Rotating app log (all levels) ─────────────────────────────────────────
+    app_log_path = LOG_DIR / "app.log"
+    file_handler = logging.handlers.RotatingFileHandler(
+        filename=app_log_path,
+        maxBytes=LOG_MAX_BYTES,
+        backupCount=LOG_BACKUP_COUNT,
+        encoding="utf-8",
+    )
+    file_handler.setLevel(numeric_level)
+    file_handler.setFormatter(JSONFormatter())
+    root.addHandler(file_handler)
+
+    # ── Rotating error log (ERROR+) ───────────────────────────────────────────
+    error_log_path = LOG_DIR / "error.log"
+    error_handler = logging.handlers.RotatingFileHandler(
+        filename=error_log_path,
+        maxBytes=LOG_MAX_BYTES,
+        backupCount=LOG_BACKUP_COUNT,
+        encoding="utf-8",
+    )
+    error_handler.setLevel(logging.ERROR)
+    error_handler.setFormatter(JSONFormatter())
+    root.addHandler(error_handler)
+
+    # Silence noisy third-party loggers
+    for noisy in ("motor", "pymongo", "asyncpg", "sqlalchemy.engine", "aioredis"):
+        logging.getLogger(noisy).setLevel(logging.WARNING)
+
+    # Emit a structured startup event so log pipelines can detect restarts
+    startup_logger = logging.getLogger(SERVICE_NAME)
+    startup_logger.info(
+        "Logging initialised",
+        extra={
+            "event": "logging_init",
+            "log_level": level.upper(),
+            "log_format": log_format,
+            "log_dir": str(LOG_DIR.resolve()),
+            "app_log": str(app_log_path.resolve()),
+            "error_log": str(error_log_path.resolve()),
+        },
+    )
+
+
+# ── Logger factory ────────────────────────────────────────────────────────────
 
 def get_logger(name: str) -> logging.Logger:
-    """Get logger instance."""
+    """Return a standard Logger for the given module name."""
     return logging.getLogger(name)
+
+
+def get_worker_logger(name: str, entity_type: str, worker_id: int) -> logging.LoggerAdapter:
+    """
+    Return a LoggerAdapter that automatically injects worker context into every
+    log record, so you never have to repeat entity_type / worker_id in extra={}.
+
+    Usage::
+
+        logger = get_worker_logger(__name__, "catalogue", 3)
+        logger.info("Processing item", extra={"entity_id": "abc-123"})
+        # → {"entity_type": "catalogue", "worker_id": 3, "entity_id": "abc-123", ...}
+    """
+    base = logging.getLogger(name)
+    return logging.LoggerAdapter(base, {"entity_type": entity_type, "worker_id": worker_id})
+
+
+# ── Structured event helpers ──────────────────────────────────────────────────
+
+def log_sync_start(logger: logging.Logger, entity_type: str, entity_id: str, operation: str, worker_id: int) -> None:
+    logger.info(
+        "Sync started",
+        extra={
+            "event": "sync_start",
+            "entity_type": entity_type,
+            "entity_id": entity_id,
+            "operation": operation,
+            "worker_id": worker_id,
+        },
+    )
+
+
+def log_sync_success(
+    logger: logging.Logger,
+    entity_type: str,
+    entity_id: str,
+    operation: str,
+    worker_id: int,
+    duration_ms: float,
+) -> None:
+    logger.info(
+        "Sync succeeded",
+        extra={
+            "event": "sync_success",
+            "entity_type": entity_type,
+            "entity_id": entity_id,
+            "operation": operation,
+            "worker_id": worker_id,
+            "duration_ms": round(duration_ms, 2),
+        },
+    )
+
+
+def log_sync_failure(
+    logger: logging.Logger,
+    entity_type: str,
+    entity_id: str,
+    operation: str,
+    worker_id: int,
+    duration_ms: float,
+    error: str,
+    attempt: int,
+    max_attempts: int,
+) -> None:
+    logger.error(
+        "Sync failed",
+        extra={
+            "event": "sync_failure",
+            "entity_type": entity_type,
+            "entity_id": entity_id,
+            "operation": operation,
+            "worker_id": worker_id,
+            "duration_ms": round(duration_ms, 2),
+            "error": error,
+            "attempt": attempt,
+            "max_attempts": max_attempts,
+        },
+    )
+
+
+def log_worker_heartbeat(
+    logger: logging.Logger,
+    entity_type: str,
+    worker_id: int,
+    iteration: int,
+    queue_size: int,
+) -> None:
+    logger.info(
+        "Worker heartbeat",
+        extra={
+            "event": "worker_heartbeat",
+            "entity_type": entity_type,
+            "worker_id": worker_id,
+            "iteration": iteration,
+            "queue_size": queue_size,
+        },
+    )
+
+
+def log_service_lifecycle(logger: logging.Logger, event: str, **kwargs: Any) -> None:
+    """Emit a structured lifecycle event (startup, shutdown, connection, etc.)."""
+    logger.info(
+        event.replace("_", " ").capitalize(),
+        extra={"event": event, **kwargs},
+    )

app/main.py

@@ -7,7 +7,7 @@ No HTTP server, just pure queue processing workers.
 import asyncio
 import signal
 from app.core.config import settings
-from app.core.logging import setup_logging, get_logger
+from app.core.logging import setup_logging, get_logger, log_service_lifecycle
 from app.sync.worker_manager import WorkerManager
 
 # Setup logging
@@ -17,66 +17,62 @@ logger = get_logger(__name__)
 
 class SyncWorkerService:
     """Main sync worker service."""
-    
+
     def __init__(self):
         self.worker_manager = None
         self.running = False
-    
+
     async def start(self):
         """Start the sync worker service."""
-        logger.info("=" * 70)
-        logger.info("Starting Sync Worker Microservice")
-        logger.info("=" * 70)
-        
+        log_service_lifecycle(logger, "service_starting", service="sync-worker-ms")
+
         try:
-            # Initialize worker manager
             self.worker_manager = WorkerManager()
             await self.worker_manager.initialize()
-            
-            # Start all workers
             await self.worker_manager.start_all_workers()
-            
+
             self.running = True
-            logger.info("✅ Sync Worker Service started successfully")
-            logger.info("=" * 70)
-            
-            # Keep running
+            log_service_lifecycle(logger, "service_ready", service="sync-worker-ms")
+
             while self.running:
                 await asyncio.sleep(1)
-                
+
         except Exception as e:
-            logger.error(f"Failed to start sync worker service: {e}", exc_info=True)
+            logger.critical(
+                "Service failed to start",
+                exc_info=True,
+                extra={"event": "service_start_error", "error": str(e)},
+            )
             raise
-    
+
     async def stop(self):
         """Stop the sync worker service."""
-        logger.info("Shutting down Sync Worker Service...")
+        log_service_lifecycle(logger, "service_stopping", service="sync-worker-ms")
         self.running = False
-        
+
         if self.worker_manager:
             await self.worker_manager.stop_all_workers()
-        
-        logger.info("✅ Sync Worker Service stopped")
+
+        log_service_lifecycle(logger, "service_stopped", service="sync-worker-ms")
 
 
 async def main():
     """Main entry point."""
     service = SyncWorkerService()
-    
-    # Setup signal handlers for graceful shutdown
+
     loop = asyncio.get_running_loop()
-    
+
     def signal_handler():
-        logger.info("Received shutdown signal")
+        logger.info("Received shutdown signal", extra={"event": "signal_received"})
         asyncio.create_task(service.stop())
-    
+
     for sig in (signal.SIGTERM, signal.SIGINT):
         loop.add_signal_handler(sig, signal_handler)
-    
+
     try:
         await service.start()
     except KeyboardInterrupt:
-        logger.info("Keyboard interrupt received")
+        logger.info("Keyboard interrupt received", extra={"event": "keyboard_interrupt"})
     finally:
         await service.stop()
 

app/sync/common/base_service.py

@@ -7,7 +7,13 @@ from typing import Optional
 from motor.motor_asyncio import AsyncIOMotorDatabase
 from sqlalchemy.ext.asyncio import AsyncEngine
 import redis.asyncio as redis
-from app.core.logging import get_logger
+from app.core.logging import (
+    get_logger,
+    log_sync_start,
+    log_sync_success,
+    log_sync_failure,
+    log_worker_heartbeat,
+)
 import time
 import traceback
 
@@ -182,56 +188,53 @@ class BaseSyncService:
     
     async def _worker(self, worker_id: int) -> None:
         """Background worker that processes Redis sync queue."""
-        logger.info(f"{self.entity_type} sync worker {worker_id} started (is_running={self.is_running})")
-        
-        # Log immediately to confirm we enter the loop
-        logger.info(f"{self.entity_type} worker {worker_id} entering main loop")
-        
+        logger.info(
+            "Worker started",
+            extra={
+                "event": "worker_start",
+                "entity_type": self.entity_type,
+                "worker_id": worker_id,
+            },
+        )
+
         iteration = 0
         while self.is_running:
             try:
                 iteration += 1
-                if iteration % 180 == 0:  # Log every ~3 minutes (1s poll interval)
+                if iteration % 180 == 0:  # heartbeat every ~3 minutes
                     queue_size = await self.sync_queue.size()
-                    logger.info(f"{self.entity_type} worker {worker_id} alive (iteration {iteration}, queue_size={queue_size})")
-                
-                # Get next operation from Redis queue (blocking with timeout)
+                    log_worker_heartbeat(logger, self.entity_type, worker_id, iteration, queue_size)
+
                 sync_op = await self.sync_queue.dequeue(timeout=1)
-                
                 if not sync_op:
-                    # Timeout, no operations available
                     continue
-                
-                logger.info(
-                    f"{self.entity_type} worker {worker_id} processing {sync_op.entity_id}",
-                    extra={
-                        "worker_id": worker_id,
-                        "entity_type": self.entity_type,
-                        "entity_id": sync_op.entity_id,
-                        "operation": sync_op.operation
-                    }
-                )
-                
-                # Process the sync operation
+
+                log_sync_start(logger, self.entity_type, sync_op.entity_id, sync_op.operation, worker_id)
                 await self._process_sync_operation(sync_op, worker_id)
-                
+
             except asyncio.CancelledError:
-                logger.info(f"{self.entity_type} sync worker {worker_id} cancelled")
+                logger.info(
+                    "Worker cancelled",
+                    extra={"event": "worker_cancel", "entity_type": self.entity_type, "worker_id": worker_id},
+                )
                 break
             except Exception as e:
                 logger.error(
-                    f"Error in {self.entity_type} sync worker {worker_id}",
-                    exc_info=e,
+                    "Unhandled error in worker loop",
+                    exc_info=True,
                     extra={
+                        "event": "worker_loop_error",
                         "entity_type": self.entity_type,
                         "worker_id": worker_id,
-                        "error": str(e)
-                    }
+                        "error": str(e),
+                    },
                 )
-                # Continue processing despite error
                 await asyncio.sleep(1.0)
-        
-        logger.info(f"{self.entity_type} sync worker {worker_id} stopped")
+
+        logger.info(
+            "Worker stopped",
+            extra={"event": "worker_stop", "entity_type": self.entity_type, "worker_id": worker_id},
+        )
     
     async def _process_sync_operation(
         self,
@@ -240,19 +243,8 @@ class BaseSyncService:
     ) -> None:
         """Process a single sync operation with retry logic."""
         start_time = time.time()
-        
+
         try:
-            logger.debug(
-                f"Worker {worker_id} processing {self.entity_type} 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_with_connection,
                 sync_op.entity_id,
@@ -260,38 +252,36 @@ class BaseSyncService:
                 entity_type=self.entity_type,
                 entity_id=sync_op.entity_id
             )
-            
-            # Record success
+
             duration_ms = (time.time() - start_time) * 1000
             self.monitoring.record_sync_success(
                 entity_type=self.entity_type,
                 entity_id=sync_op.entity_id,
-                duration_ms=duration_ms
+                duration_ms=duration_ms,
             )
             await self.sync_queue.record_processed()
-            
+            log_sync_success(
+                logger, self.entity_type, sync_op.entity_id,
+                sync_op.operation, worker_id, duration_ms,
+            )
+
         except Exception as e:
-            # Record failure
+            duration_ms = (time.time() - start_time) * 1000
             error_msg = str(e)
-            stack_trace = traceback.format_exc()
-            
+
             self.monitoring.record_sync_failure(
                 entity_type=self.entity_type,
                 entity_id=sync_op.entity_id,
                 error=error_msg,
-                stack_trace=stack_trace
+                stack_trace=traceback.format_exc(),
             )
             await self.sync_queue.record_failed()
-            
-            logger.error(
-                f"Worker {worker_id} failed to sync {self.entity_type} 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
-                }
+            log_sync_failure(
+                logger, self.entity_type, sync_op.entity_id,
+                sync_op.operation, worker_id, duration_ms,
+                error=error_msg,
+                attempt=self.retry_manager.max_retries + 1,
+                max_attempts=self.retry_manager.max_retries + 1,
             )
     
     async def _sync_with_connection(self, entity_id: str, operation: str = "update") -> bool:

app/sync/worker_manager.py

@@ -8,7 +8,7 @@ from sqlalchemy import text
 import redis.asyncio as redis
 
 from app.core.config import settings
-from app.core.logging import get_logger
+from app.core.logging import get_logger, log_service_lifecycle
 
 logger = get_logger(__name__)
 
@@ -25,42 +25,41 @@ class WorkerManager:
     
     async def initialize(self):
         """Initialize database connections."""
-        logger.info("Initializing connections...")
-        
+        log_service_lifecycle(logger, "connections_init")
+
         # MongoDB
-        logger.info(f"Connecting to MongoDB: {settings.MONGODB_URI}")
+        log_service_lifecycle(logger, "mongodb_connecting", host=settings.MONGODB_URI.split("@")[-1])
         self.mongo_client = AsyncIOMotorClient(settings.MONGODB_URI)
         self.mongo_db = self.mongo_client[settings.MONGODB_DB_NAME]
         await self.mongo_client.admin.command('ping')
-        logger.info("✅ Connected to MongoDB")
-        
+        log_service_lifecycle(logger, "mongodb_connected", db=settings.MONGODB_DB_NAME)
+
         # PostgreSQL
         pg_url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}"
-        logger.info(f"Connecting to PostgreSQL: {settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}")
+        log_service_lifecycle(logger, "postgres_connecting", host=settings.POSTGRES_HOST, port=settings.POSTGRES_PORT)
         self.pg_engine = create_async_engine(pg_url, echo=False, pool_pre_ping=True)
         async with self.pg_engine.begin() as conn:
             await conn.execute(text("SELECT 1"))
-        logger.info("✅ Connected to PostgreSQL")
-        
-        # Redis with connection pooling
-        logger.info(f"Connecting to Redis: {settings.REDIS_HOST}:{settings.REDIS_PORT}")
+        log_service_lifecycle(logger, "postgres_connected", db=settings.POSTGRES_DB)
+
+        # Redis
+        log_service_lifecycle(logger, "redis_connecting", host=settings.REDIS_HOST, port=settings.REDIS_PORT)
         pool = redis.ConnectionPool(
             host=settings.REDIS_HOST,
             port=settings.REDIS_PORT,
             password=settings.REDIS_PASSWORD,
             db=settings.REDIS_DB,
             decode_responses=True,
-            max_connections=15,  # Slightly higher for sync workers
+            max_connections=15,
             socket_keepalive=True,
             socket_connect_timeout=5,
             retry_on_timeout=True,
-            health_check_interval=30
+            health_check_interval=30,
         )
         self.redis_client = redis.Redis(connection_pool=pool)
         await self.redis_client.ping()
-        logger.info("✅ Connected to Redis with connection pool (max_connections=15)")
-        
-        # Initialize sync services
+        log_service_lifecycle(logger, "redis_connected", max_connections=15)
+
         await self._initialize_sync_services()
     
     async def _initialize_sync_services(self):
@@ -70,89 +69,72 @@ class WorkerManager:
         from app.sync.merchants.service import MerchantSyncService
         from app.sync.warehouses.service import WarehouseSyncService
         from app.sync.uom.service import UOMSyncService
-        
-        logger.info("Initializing sync services...")
-        
+
+        log_service_lifecycle(logger, "sync_services_init")
+
         self.sync_services = {
             "catalogues": CatalogueSyncService(
-                mongo_db=self.mongo_db,
-                pg_engine=self.pg_engine,
-                redis_client=self.redis_client,
-                max_queue_size=10000,
-                worker_count=settings.CATALOGUE_WORKERS,
-                max_retries=3
+                mongo_db=self.mongo_db, pg_engine=self.pg_engine, redis_client=self.redis_client,
+                max_queue_size=10000, worker_count=settings.CATALOGUE_WORKERS, max_retries=3,
             ),
             "employees": EmployeeSyncService(
-                mongo_db=self.mongo_db,
-                pg_engine=self.pg_engine,
-                redis_client=self.redis_client,
-                max_queue_size=10000,
-                worker_count=settings.EMPLOYEE_WORKERS,
-                max_retries=3
+                mongo_db=self.mongo_db, pg_engine=self.pg_engine, redis_client=self.redis_client,
+                max_queue_size=10000, worker_count=settings.EMPLOYEE_WORKERS, max_retries=3,
             ),
             "merchants": MerchantSyncService(
-                mongo_db=self.mongo_db,
-                pg_engine=self.pg_engine,
-                redis_client=self.redis_client,
-                max_queue_size=10000,
-                worker_count=settings.MERCHANT_WORKERS,
-                max_retries=3
+                mongo_db=self.mongo_db, pg_engine=self.pg_engine, redis_client=self.redis_client,
+                max_queue_size=10000, worker_count=settings.MERCHANT_WORKERS, max_retries=3,
             ),
             "warehouses": WarehouseSyncService(
-                mongo_db=self.mongo_db,
-                pg_engine=self.pg_engine,
-                redis_client=self.redis_client,
-                max_queue_size=10000,
-                worker_count=settings.WAREHOUSE_WORKERS,
-                max_retries=3
+                mongo_db=self.mongo_db, pg_engine=self.pg_engine, redis_client=self.redis_client,
+                max_queue_size=10000, worker_count=settings.WAREHOUSE_WORKERS, max_retries=3,
             ),
             "uom": UOMSyncService(
-                mongo_db=self.mongo_db,
-                pg_engine=self.pg_engine,
-                redis_client=self.redis_client,
-                max_queue_size=5000,
-                worker_count=settings.UOM_WORKERS,
-                max_retries=3
-            )
+                mongo_db=self.mongo_db, pg_engine=self.pg_engine, redis_client=self.redis_client,
+                max_queue_size=5000, worker_count=settings.UOM_WORKERS, max_retries=3,
+            ),
         }
-        
-        logger.info(f"✅ Initialized {len(self.sync_services)} sync services")
+
+        log_service_lifecycle(logger, "sync_services_ready", count=len(self.sync_services),
+                              services=list(self.sync_services.keys()))
     
     async def start_all_workers(self):
         """Start all sync workers."""
-        logger.info("Starting all sync workers...")
-        
+        log_service_lifecycle(logger, "workers_starting", count=len(self.sync_services))
+
         for entity_type, service in self.sync_services.items():
             try:
                 await service.start_workers()
-                logger.info(f"✅ Started {entity_type} workers")
+                log_service_lifecycle(logger, "workers_started", entity_type=entity_type,
+                                      worker_count=service.worker_count)
             except Exception as e:
-                logger.error(f"Failed to start {entity_type} workers: {e}", exc_info=True)
-        
-        # Log summary
+                logger.error("Failed to start workers", exc_info=True,
+                             extra={"event": "workers_start_error", "entity_type": entity_type, "error": str(e)})
+
         total_workers = sum(s.worker_count for s in self.sync_services.values())
-        logger.info(f"✅ Started {total_workers} workers across {len(self.sync_services)} services")
-    
+        log_service_lifecycle(logger, "all_workers_started", total_workers=total_workers,
+                              service_count=len(self.sync_services))
+
     async def stop_all_workers(self):
         """Stop all sync workers."""
-        logger.info("Stopping all sync workers...")
-        
+        log_service_lifecycle(logger, "workers_stopping")
+
         for entity_type, service in self.sync_services.items():
             try:
                 await service.stop_workers()
-                logger.info(f"✅ Stopped {entity_type} workers")
+                log_service_lifecycle(logger, "workers_stopped", entity_type=entity_type)
             except Exception as e:
-                logger.error(f"Failed to stop {entity_type} workers: {e}", exc_info=True)
-        
-        # Close connections
+                logger.error("Failed to stop workers", exc_info=True,
+                             extra={"event": "workers_stop_error", "entity_type": entity_type, "error": str(e)})
+
         if self.redis_client:
             await self.redis_client.aclose()
-            logger.info("✅ Closed Redis connection")
-        
+            log_service_lifecycle(logger, "redis_disconnected")
+
         if self.pg_engine:
             await self.pg_engine.dispose()
-            logger.info("✅ Closed PostgreSQL connection")
-        
+            log_service_lifecycle(logger, "postgres_disconnected")
+
         if self.mongo_client:
             self.mongo_client.close()
-            logger.info("✅ Closed MongoDB connection")
+            log_service_lifecycle(logger, "mongodb_disconnected")