refactor(notification-ms): enhance structured logging across all channels

- Migrate from f-string logging to structured logging with extra context fields
- Add event identifiers to all log statements for better observability
- Include contextual metadata (merchant_id, template_name, recipient, etc.) in logs
- Replace exception string interpolation with exc_info=True for proper stack traces
- Update cache service logging to use structured format with operation-specific events
- Enhance email channel logging with recipient and template context
- Improve push notification logging with provider message IDs and merchant tracking
- Update SMS channel logging with provider request IDs and error details
- Add WhatsApp channel structured logging with message IDs and status tracking
- Update worker and dispatcher service logging for consistency
- Add logging configuration options to .env.example (LOG_FORMAT, LOG_DIR, LOG_MAX_BYTES, LOG_BACKUP_COUNT)
- Improves log parsing, filtering, and correlation for production monitoring

.env.example

@@ -3,6 +3,10 @@ APP_NAME=Notification Microservice
 APP_VERSION=1.0.0
 DEBUG=false
 LOG_LEVEL=INFO
+# LOG_FORMAT=json
+# LOG_DIR=logs
+# LOG_MAX_BYTES=52428800
+# LOG_BACKUP_COUNT=10
 ROOT_PATH=
 
 # MongoDB (shared DB — merchant settings are read from scm_merchant_settings collection)

app/cache.py

@@ -31,7 +31,15 @@ def get_redis() -> Optional[redis.Redis]:
             pool_params["password"] = settings.REDIS_PASSWORD
         pool = redis.ConnectionPool(**pool_params)
         _redis_client = redis.Redis(connection_pool=pool)
-        logger.info(f"Redis connected ({settings.REDIS_HOST}:{settings.REDIS_PORT})")
+        logger.info(
+            "Redis connected",
+            extra={
+                "event": "redis_connected",
+                "redis_host": settings.REDIS_HOST,
+                "redis_port": settings.REDIS_PORT,
+                "redis_db": settings.REDIS_DB,
+            },
+        )
     return _redis_client
 
 
@@ -49,8 +57,8 @@ class CacheService:
         try:
             v = json.dumps(value) if not isinstance(value, str) else value
             return self.client.setex(key, ttl, v)
-        except Exception as e:
-            logger.error(f"Cache set error: {e}")
+        except Exception:
+            logger.error("Cache set failed", exc_info=True, extra={"event": "cache_set_failure", "key": key})
             return False
 
     async def get(self, key: str) -> Optional[Any]:
@@ -62,15 +70,19 @@ class CacheService:
                 except json.JSONDecodeError:
                     return value
             return None
-        except Exception as e:
-            logger.error(f"Cache get error: {e}")
+        except Exception:
+            logger.error("Cache get failed", exc_info=True, extra={"event": "cache_get_failure", "key": key})
             return None
 
     async def delete(self, key: str) -> bool:
         try:
             return self.client.delete(key) > 0
-        except Exception as e:
-            logger.error(f"Cache delete error: {e}")
+        except Exception:
+            logger.error(
+                "Cache delete failed",
+                exc_info=True,
+                extra={"event": "cache_delete_failure", "key": key},
+            )
             return False
 
 

app/channels/email.py

@@ -81,11 +81,35 @@ class EmailChannel:
                 start_tls=use_tls,
                 timeout=30,
             )
-            logger.info(f"Email sent to {recipient}, subject={subject}")
+            logger.info(
+                "Email sent",
+                extra={
+                    "event": "email_send_success",
+                    "recipient": recipient,
+                    "template_name": template_name,
+                    "subject": subject,
+                },
+            )
             return True, "Email sent", None
         except aiosmtplib.SMTPException as e:
-            logger.error(f"SMTP error sending to {recipient}: {e}")
+            logger.error(
+                "Email SMTP error",
+                extra={
+                    "event": "email_send_failure",
+                    "recipient": recipient,
+                    "template_name": template_name,
+                    "error": str(e),
+                },
+            )
             return False, f"SMTP error: {e}", None
         except Exception as e:
-            logger.error(f"Email error: {e}", exc_info=True)
+            logger.error(
+                "Email send failed",
+                exc_info=True,
+                extra={
+                    "event": "email_send_failure",
+                    "recipient": recipient,
+                    "template_name": template_name,
+                },
+            )
             return False, f"Email error: {e}", None

app/channels/push.py

@@ -26,10 +26,17 @@ def _get_firebase_app(merchant_id: str, service_account_json: str):
         cred = credentials.Certificate(cred_dict)
         app = firebase_admin.initialize_app(cred, name=f"merchant_{merchant_id}")
         _firebase_apps[merchant_id] = app
-        logger.info(f"Firebase app initialized for merchant {merchant_id}")
+        logger.info(
+            "Firebase app initialized",
+            extra={"event": "firebase_init_success", "merchant_id": merchant_id},
+        )
         return app
-    except Exception as e:
-        logger.error(f"Firebase init error for merchant {merchant_id}: {e}", exc_info=True)
+    except Exception:
+        logger.error(
+            "Firebase init failed",
+            exc_info=True,
+            extra={"event": "firebase_init_failure", "merchant_id": merchant_id},
+        )
         return None
 
 
@@ -82,9 +89,26 @@ class PushChannel:
             response = await loop.run_in_executor(
                 None, lambda: messaging.send(message, app=app)
             )
-            logger.info(f"Push sent to token {recipient[:20]}..., response={response}")
+            logger.info(
+                "Push sent",
+                extra={
+                    "event": "push_send_success",
+                    "recipient_prefix": recipient[:20],
+                    "template_name": template_name,
+                    "provider_message_id": response,
+                    "merchant_id": merchant_id,
+                },
+            )
             return True, "Push notification sent", response
 
         except Exception as e:
-            logger.error(f"Push error: {e}", exc_info=True)
+            logger.error(
+                "Push send failed",
+                exc_info=True,
+                extra={
+                    "event": "push_send_failure",
+                    "template_name": template_name,
+                    "merchant_id": merchant_id,
+                },
+            )
             return False, f"Push error: {e}", None

app/channels/sms.py

@@ -85,12 +85,26 @@ class SMSChannel:
             if resp.status_code == 200 and data.get("return"):
                 request_id = data.get("request_id", "")
                 logger.info(
-                    f"SMS sent to {numbers}, request_id={request_id}"
+                    "SMS sent",
+                    extra={
+                        "event": "sms_send_success",
+                        "recipient": numbers,
+                        "provider_request_id": request_id,
+                        "template_name": template_name,
+                    },
                 )
                 return True, "SMS sent via Fast2SMS", request_id
 
             error_msg = data.get("message", "Unknown Fast2SMS error")
-            logger.error(f"Fast2SMS error for {numbers}: {error_msg}")
+            logger.error(
+                "SMS provider returned error",
+                extra={
+                    "event": "sms_send_failure",
+                    "recipient": numbers,
+                    "template_name": template_name,
+                    "provider_error": error_msg,
+                },
+            )
             return False, f"Fast2SMS error: {error_msg}", None
 
         except httpx.TimeoutException:
@@ -98,5 +112,13 @@ class SMSChannel:
         except httpx.RequestError as e:
             return False, f"Fast2SMS network error: {e}", None
         except Exception as e:
-            logger.error(f"SMS error to {numbers}: {e}", exc_info=True)
+            logger.error(
+                "SMS send failed",
+                exc_info=True,
+                extra={
+                    "event": "sms_send_failure",
+                    "recipient": numbers,
+                    "template_name": template_name,
+                },
+            )
             return False, f"SMS error: {e}", None

app/channels/whatsapp.py

@@ -83,14 +83,31 @@ class WhatsAppChannel:
                         r = receivers[0]
                         mid = r.get("localMessageId")
                         if r.get("isValidWhatsAppNumber") and not r.get("errors"):
-                            logger.info(f"WhatsApp sent to {whatsapp_number}, mid={mid}")
+                            logger.info(
+                                "WhatsApp sent",
+                                extra={
+                                    "event": "whatsapp_send_success",
+                                    "recipient": whatsapp_number,
+                                    "template_name": template_name,
+                                    "provider_message_id": mid,
+                                },
+                            )
                             return True, "Sent via WhatsApp", mid
                         err = (r.get("errors") or ["Invalid WhatsApp number"])[0]
                         return False, f"WhatsApp error: {err}", None
                     return False, "No receivers in WATI response", None
                 return False, f"WATI error: {data.get('error', 'Unknown')}", None
 
-            logger.error(f"WATI HTTP {resp.status_code}: {resp.text[:200]}")
+            logger.error(
+                "WhatsApp provider HTTP error",
+                extra={
+                    "event": "whatsapp_send_failure",
+                    "status_code": resp.status_code,
+                    "response_excerpt": resp.text[:200],
+                    "recipient": whatsapp_number,
+                    "template_name": template_name,
+                },
+            )
             if resp.status_code == 401:
                 return False, "WATI credentials invalid or expired", None
             if resp.status_code == 403:
@@ -101,8 +118,16 @@ class WhatsAppChannel:
             return False, "WhatsApp request timeout", None
         except httpx.RequestError as e:
             return False, f"WhatsApp network error: {e}", None
-        except Exception as e:
-            logger.error(f"WhatsApp unexpected error: {e}", exc_info=True)
+        except Exception:
+            logger.error(
+                "WhatsApp send failed",
+                exc_info=True,
+                extra={
+                    "event": "whatsapp_send_failure",
+                    "recipient": whatsapp_number,
+                    "template_name": template_name,
+                },
+            )
             return False, "WhatsApp internal error", None
 
     @staticmethod
@@ -134,6 +159,10 @@ class WhatsAppChannel:
                 d = resp.json()
                 return True, d.get("optedIn", False), d.get("id")
             return False, False, None
-        except Exception as e:
-            logger.error(f"Opt-in check error: {e}")
+        except Exception:
+            logger.error(
+                "WhatsApp opt-in check failed",
+                exc_info=True,
+                extra={"event": "whatsapp_optin_check_failure", "recipient": number},
+            )
             return False, False, None

app/controllers/router.py

@@ -40,7 +40,15 @@ async def send_notification(payload: SendNotificationRequest):
             channels=doc["channels"],
         )
     except Exception as e:
-        logger.error(f"Send error: {e}", exc_info=True)
+        logger.error(
+            "Send notification failed",
+            exc_info=True,
+            extra={
+                "event": "notification_create_failure",
+                "merchant_id": payload.merchant_id,
+                "template_name": payload.template_name,
+            },
+        )
         raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
 
 

app/core/logging.py

@@ -1,95 +1,151 @@
-"""
-Structured logging for Notification Microservice.
-"""
+"""Production-grade logging configuration for Notification Microservice."""
+import json
 import logging
 import logging.handlers
-import json
+import os
 import sys
+import traceback
+from datetime import datetime, timezone
 from pathlib import Path
-from datetime import datetime
+from typing import Any, Dict
+
+
+SERVICE_NAME = "notification-ms"
+LOG_DIR = Path(os.getenv("LOG_DIR", "logs"))
+LOG_MAX_BYTES = int(os.getenv("LOG_MAX_BYTES", 50 * 1024 * 1024))
+LOG_BACKUP_COUNT = int(os.getenv("LOG_BACKUP_COUNT", "10"))
 
 
 class JSONFormatter(logging.Formatter):
+    """Emit log records as single-line JSON objects."""
+
+    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:
-        log_data = {
-            "timestamp": datetime.utcnow().isoformat(),
+        payload: Dict[str, Any] = {
+            "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
             "level": record.levelname,
             "logger": record.name,
             "message": record.getMessage(),
-            "module": record.module,
-            "function": record.funcName,
-            "line": record.lineno,
+            "service": SERVICE_NAME,
+            "pid": record.process,
         }
-        if hasattr(record, "correlation_id") and record.correlation_id:
-            log_data["correlation_id"] = record.correlation_id
-        if record.exc_info:
-            log_data["exception"] = self.formatException(record.exc_info)
-        return json.dumps(log_data, default=str)
 
+        if record.levelno >= logging.WARNING:
+            payload["caller"] = f"{record.pathname}:{record.lineno}"
 
-class StructuredLogger:
-    def __init__(self, name: str):
-        self.logger = logging.getLogger(name)
+        for key, val in record.__dict__.items():
+            if key not in self.RESERVED and not key.startswith("_"):
+                payload[key] = val
 
-    def _log(self, level: int, message: str, extra=None, exc_info=False):
-        if extra:
-            self.logger.log(level, message, extra={"extra": extra}, exc_info=exc_info)
-        else:
-            self.logger.log(level, message, exc_info=exc_info)
+        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),
+            }
 
-    def debug(self, message, extra=None):
-        self._log(logging.DEBUG, message, extra)
+        return json.dumps(payload, default=str)
 
-    def info(self, message, extra=None):
-        self._log(logging.INFO, message, extra)
 
-    def warning(self, message, extra=None):
-        self._log(logging.WARNING, message, extra)
+class ConsoleFormatter(logging.Formatter):
+    """Readable formatter for local development."""
 
-    def error(self, message, extra=None, exc_info=False):
-        self._log(logging.ERROR, message, extra, exc_info)
+    FMT = "%(asctime)s %(levelname)-8s %(name)s — %(message)s"
 
-    def critical(self, message, extra=None, exc_info=False):
-        self._log(logging.CRITICAL, message, extra, exc_info)
+    def format(self, record: logging.LogRecord) -> str:
+        formatter = logging.Formatter(self.FMT, datefmt="%Y-%m-%d %H:%M:%S")
+        result = formatter.format(record)
+        extras = {
+            k: v
+            for k, v in record.__dict__.items()
+            if k not in JSONFormatter.RESERVED and not k.startswith("_") and k != "color_message"
+        }
+        if extras:
+            result += f"  {extras}"
+        return result
 
 
-def setup_logging(log_level: str = "INFO", log_dir: str = "logs"):
-    log_path = Path(log_dir)
-    log_path.mkdir(exist_ok=True)
+def setup_logging(log_level: str = "INFO") -> None:
+    """Configure root logger with JSON/console and rotating file handlers."""
+    numeric_level = getattr(logging, log_level.upper(), logging.INFO)
+    log_format = os.getenv("LOG_FORMAT", "json").lower()
 
-    root_logger = logging.getLogger()
-    root_logger.setLevel(getattr(logging, log_level.upper()))
-    root_logger.handlers = []
+    LOG_DIR.mkdir(parents=True, exist_ok=True)
 
-    console_fmt = logging.Formatter(
-        fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-        datefmt="%Y-%m-%d %H:%M:%S",
-    )
-    json_fmt = JSONFormatter()
+    root = logging.getLogger()
+    root.setLevel(numeric_level)
+    root.handlers.clear()
 
-    console = logging.StreamHandler(sys.stderr)
-    console.setLevel(getattr(logging, log_level.upper()))
-    console.setFormatter(console_fmt)
-    root_logger.addHandler(console)
+    stdout_handler = logging.StreamHandler(sys.stdout)
+    stdout_handler.setLevel(numeric_level)
+    stdout_handler.setFormatter(ConsoleFormatter() if log_format == "console" else JSONFormatter())
+    root.addHandler(stdout_handler)
 
-    file_all = logging.handlers.RotatingFileHandler(
-        filename=log_path / "app.log", maxBytes=10 * 1024 * 1024, backupCount=10
+    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_all.setLevel(logging.DEBUG)
-    file_all.setFormatter(json_fmt)
-    root_logger.addHandler(file_all)
-
-    file_err = logging.handlers.RotatingFileHandler(
-        filename=log_path / "app_errors.log", maxBytes=10 * 1024 * 1024, backupCount=10
+    file_handler.setLevel(numeric_level)
+    file_handler.setFormatter(JSONFormatter())
+    root.addHandler(file_handler)
+
+    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)
+
+    for noisy in ("motor", "pymongo", "httpx", "uvicorn.access", "aioredis"):
+        logging.getLogger(noisy).setLevel(logging.WARNING)
+
+    startup_logger = logging.getLogger(SERVICE_NAME)
+    startup_logger.info(
+        "Logging initialised",
+        extra={
+            "event": "logging_init",
+            "log_level": log_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()),
+        },
     )
-    file_err.setLevel(logging.ERROR)
-    file_err.setFormatter(json_fmt)
-    root_logger.addHandler(file_err)
-
-    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
-    logging.getLogger("motor").setLevel(logging.WARNING)
-    logging.getLogger("httpx").setLevel(logging.WARNING)
 
 
-def get_logger(name: str) -> StructuredLogger:
-    return StructuredLogger(name)
+def get_logger(name: str) -> logging.Logger:
+    """Return a standard logger for the given module name."""
+    return logging.getLogger(name)

app/main.py

@@ -25,15 +25,15 @@ _worker = NotificationWorker()
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    logger.info("Starting Notification Microservice")
+    logger.info("Service starting", extra={"event": "service_starting"})
     await connect_to_mongo()
     await _worker.start()
-    logger.info("Notification Microservice ready")
+    logger.info("Service ready", extra={"event": "service_ready"})
     yield
-    logger.info("Shutting down Notification Microservice")
+    logger.info("Service stopping", extra={"event": "service_stopping"})
     await _worker.stop()
     await close_mongo_connection()
-    logger.info("Notification Microservice stopped")
+    logger.info("Service stopped", extra={"event": "service_stopped"})
 
 
 app = FastAPI(
@@ -53,18 +53,33 @@ app.add_middleware(CorrelationIdMiddleware)
 async def log_requests(request: Request, call_next):
     correlation_id = get_correlation_id()
     start = time.time()
-    logger.info(f"→ {request.method} {request.url.path}", extra={"correlation_id": correlation_id})
     try:
         response = await call_next(request)
-        elapsed = time.time() - start
+        duration_ms = round((time.time() - start) * 1000, 2)
         logger.info(
-            f"← {request.method} {request.url.path} {response.status_code} ({elapsed:.3f}s)",
-            extra={"correlation_id": correlation_id},
+            "HTTP request",
+            extra={
+                "event": "http_request",
+                "correlation_id": correlation_id,
+                "method": request.method,
+                "path": request.url.path,
+                "status_code": response.status_code,
+                "duration_ms": duration_ms,
+            },
         )
-        response.headers["X-Process-Time"] = f"{elapsed:.3f}"
+        response.headers["X-Process-Time"] = str(round(duration_ms / 1000, 3))
         return response
-    except Exception as e:
-        logger.error(f"Request failed: {e}", exc_info=True)
+    except Exception:
+        logger.error(
+            "HTTP request failed",
+            exc_info=True,
+            extra={
+                "event": "http_request_failure",
+                "correlation_id": correlation_id,
+                "method": request.method,
+                "path": request.url.path,
+            },
+        )
         raise
 
 
@@ -92,7 +107,16 @@ async def validation_handler(request: Request, exc: RequestValidationError):
 @app.exception_handler(Exception)
 async def general_handler(request: Request, exc: Exception):
     cid = get_correlation_id()
-    logger.error(f"Unhandled: {exc}", exc_info=True, extra={"correlation_id": cid})
+    logger.error(
+        "Unhandled exception",
+        exc_info=True,
+        extra={
+            "event": "unhandled_exception",
+            "correlation_id": cid,
+            "method": request.method,
+            "path": request.url.path,
+        },
+    )
     return JSONResponse(
         status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
         content={"success": False, "error": "Internal Server Error", "correlation_id": cid},

app/nosql.py

@@ -13,19 +13,32 @@ _database: AsyncIOMotorDatabase = None
 
 async def connect_to_mongo():
     global _client, _database
-    logger.info(f"Connecting to MongoDB: {settings.MONGODB_DB_NAME}")
-    _client = AsyncIOMotorClient(settings.MONGODB_URI)
-    _database = _client[settings.MONGODB_DB_NAME]
-    # Verify connection
-    await _client.admin.command("ping")
-    logger.info("MongoDB connected successfully")
+    logger.info(
+        "Connecting to MongoDB",
+        extra={"event": "mongodb_connect_start", "db_name": settings.MONGODB_DB_NAME},
+    )
+    try:
+        _client = AsyncIOMotorClient(settings.MONGODB_URI)
+        _database = _client[settings.MONGODB_DB_NAME]
+        await _client.admin.command("ping")
+        logger.info(
+            "MongoDB connected",
+            extra={"event": "mongodb_connected", "db_name": settings.MONGODB_DB_NAME},
+        )
+    except Exception:
+        logger.error(
+            "MongoDB connection failed",
+            exc_info=True,
+            extra={"event": "mongodb_connect_failure", "db_name": settings.MONGODB_DB_NAME},
+        )
+        raise
 
 
 async def close_mongo_connection():
     global _client
     if _client:
         _client.close()
-        logger.info("MongoDB connection closed")
+        logger.info("MongoDB connection closed", extra={"event": "mongodb_disconnected"})
 
 
 def get_database() -> AsyncIOMotorDatabase:

app/services/dispatcher.py

@@ -117,7 +117,10 @@ class NotificationDispatcher:
                     }
                 },
             )
-            logger.error(f"Notification {nid} failed: missing merchant_id")
+            logger.error(
+                "Dispatch failed: missing merchant_id",
+                extra={"event": "notification_dispatch_failure", "notification_id": nid},
+            )
             doc["status"] = NotificationStatus.FAILED.value
             return doc
 
@@ -136,7 +139,14 @@ class NotificationDispatcher:
                     }
                 },
             )
-            logger.error(f"Notification {nid} failed: no merchant settings for {merchant_id}")
+            logger.error(
+                "Dispatch failed: merchant settings not found",
+                extra={
+                    "event": "notification_dispatch_failure",
+                    "notification_id": nid,
+                    "merchant_id": merchant_id,
+                },
+            )
             doc["status"] = NotificationStatus.FAILED.value
             return doc
 
@@ -206,7 +216,16 @@ class NotificationDispatcher:
             elif channel_enum == NotificationChannel.WHATSAPP:
                 # Use merchant-mapped WATI template name if available, else fall back to doc template_name
                 wa_template_name = tpl_mapping.get("whatsapp_template_name") or doc["template_name"]
-                logger.info(f"WhatsApp template resolved: {doc['template_name']} → {wa_template_name}")
+                logger.info(
+                    "WhatsApp template resolved",
+                    extra={
+                        "event": "template_resolve_success",
+                        "notification_id": nid,
+                        "channel": "whatsapp",
+                        "template_name": doc["template_name"],
+                        "resolved_template_name": wa_template_name,
+                    },
+                )
                 success, message, provider_id = await WhatsAppChannel.send(
                     recipient=doc["recipient"],
                     template_name=wa_template_name,
@@ -227,7 +246,16 @@ class NotificationDispatcher:
             elif channel_enum == NotificationChannel.SMS:
                 # sms_template_id is the name of the template in notification_templates collection
                 sms_tpl_name = tpl_mapping.get("sms_template_id") or doc["template_name"]
-                logger.info(f"SMS template resolved: {doc['template_name']} → {sms_tpl_name}")
+                logger.info(
+                    "SMS template resolved",
+                    extra={
+                        "event": "template_resolve_success",
+                        "notification_id": nid,
+                        "channel": "sms",
+                        "template_name": doc["template_name"],
+                        "resolved_template_name": sms_tpl_name,
+                    },
+                )
                 resolved_data = await NotificationDispatcher._resolve_template(
                     sms_tpl_name, "sms", raw_data
                 )
@@ -267,7 +295,15 @@ class NotificationDispatcher:
 
         doc["status"] = final_status
         doc["channel_results"] = channel_results
-        logger.info(f"Notification {nid} dispatched: status={final_status}")
+        logger.info(
+            "Notification dispatched",
+            extra={
+                "event": "notification_dispatch_success" if any_success else "notification_dispatch_failure",
+                "notification_id": nid,
+                "merchant_id": merchant_id,
+                "status": final_status,
+            },
+        )
         return doc
 
     @staticmethod

app/services/notification_service.py

@@ -35,9 +35,23 @@ class NotificationService:
         r = get_redis()
         if r:
             r.lpush(QUEUE_NAME, json.dumps({"notification_id": doc["notification_id"]}))
-            logger.info(f"Notification {doc['notification_id']} queued")
+            logger.info(
+                "Notification queued",
+                extra={
+                    "event": "enqueued",
+                    "notification_id": doc["notification_id"],
+                    "queue": QUEUE_NAME,
+                },
+            )
         else:
-            logger.warning(f"Redis unavailable — notification {doc['notification_id']} saved but not queued")
+            logger.warning(
+                "Notification saved but queue unavailable",
+                extra={
+                    "event": "enqueue_skipped",
+                    "notification_id": doc["notification_id"],
+                    "queue": QUEUE_NAME,
+                },
+            )
 
         return doc
 
@@ -55,7 +69,7 @@ class NotificationService:
                 results.append(doc)
                 queued += 1
             except Exception as e:
-                logger.error(f"Batch queue error: {e}")
+                logger.error("Batch queue failed", exc_info=True, extra={"event": "batch_enqueue_failure"})
                 failed += 1
                 results.append({"error": str(e), "recipient": req.recipient})
         return queued, failed, results
@@ -122,5 +136,14 @@ class NotificationService:
         r = get_redis()
         if r:
             r.lpush(QUEUE_NAME, json.dumps({"notification_id": notification_id}))
+            logger.info(
+                "Notification re-queued",
+                extra={
+                    "event": "enqueued",
+                    "notification_id": notification_id,
+                    "queue": QUEUE_NAME,
+                    "operation": "retry",
+                },
+            )
 
         return await NotificationService.get_notification(notification_id)

app/services/worker.py

@@ -30,7 +30,7 @@ class NotificationWorker:
         """Start the worker loop as a background task."""
         self._running = True
         self._task = asyncio.create_task(self._run())
-        logger.info("Notification worker started")
+        logger.info("Worker started", extra={"event": "worker_started", "queue": QUEUE_NAME})
 
     async def stop(self):
         """Gracefully stop the worker."""
@@ -41,7 +41,7 @@ class NotificationWorker:
                 await self._task
             except asyncio.CancelledError:
                 pass
-        logger.info("Notification worker stopped")
+        logger.info("Worker stopped", extra={"event": "worker_stopped", "queue": QUEUE_NAME})
 
     async def _run(self):
         """Main worker loop — poll Redis, dispatch each notification."""
@@ -49,22 +49,35 @@ class NotificationWorker:
         batch_size = settings.WORKER_BATCH_SIZE
 
         logger.info(
-            f"Worker loop starting (poll_interval={poll_interval}s, batch_size={batch_size})"
+            "Worker loop starting",
+            extra={
+                "event": "worker_loop_start",
+                "queue": QUEUE_NAME,
+                "poll_interval_seconds": poll_interval,
+                "batch_size": batch_size,
+            },
         )
 
         while self._running:
             try:
                 r = get_redis()
                 if not r:
-                    logger.warning("Redis client not available, retrying in 5s")
+                    logger.warning(
+                        "Redis client unavailable",
+                        extra={"event": "redis_unavailable", "retry_delay_seconds": 5},
+                    )
                     await asyncio.sleep(5)
                     continue
 
                 # Verify Redis connectivity on first iteration
                 try:
                     r.ping()
-                except Exception as ping_err:
-                    logger.error(f"Redis ping failed: {ping_err}")
+                except Exception:
+                    logger.error(
+                        "Redis ping failed",
+                        exc_info=True,
+                        extra={"event": "redis_ping_failure", "retry_delay_seconds": 5},
+                    )
                     await asyncio.sleep(5)
                     continue
 
@@ -79,14 +92,25 @@ class NotificationWorker:
                         payload = json.loads(raw)
                         nid = payload.get("notification_id")
                         if not nid:
-                            logger.warning(f"Queue item missing notification_id: {raw}")
+                            logger.warning(
+                                "Queue item missing notification_id",
+                                extra={"event": "dequeue_invalid_payload", "queue": QUEUE_NAME, "raw": raw},
+                            )
                             continue
 
+                        logger.info(
+                            "Queue item dequeued",
+                            extra={"event": "dequeued", "queue": QUEUE_NAME, "notification_id": nid},
+                        )
+
                         # Fetch full doc from MongoDB
                         db = get_database()
                         doc = await db[COLLECTION].find_one({"notification_id": nid})
                         if not doc:
-                            logger.warning(f"Notification {nid} not found in DB, skipping")
+                            logger.warning(
+                                "Notification not found",
+                                extra={"event": "notification_not_found", "notification_id": nid},
+                            )
                             continue
 
                         # Dispatch through channels
@@ -94,18 +118,24 @@ class NotificationWorker:
                         processed += 1
 
                     except json.JSONDecodeError:
-                        logger.error(f"Invalid JSON in queue: {raw}")
-                    except Exception as e:
-                        logger.error(f"Worker dispatch error: {e}", exc_info=True)
+                        logger.error(
+                            "Invalid JSON in queue",
+                            extra={"event": "dequeue_invalid_json", "queue": QUEUE_NAME, "raw": raw},
+                        )
+                    except Exception:
+                        logger.error("Worker dispatch error", exc_info=True, extra={"event": "dispatch_failure"})
 
                 if processed > 0:
-                    logger.info(f"Processed {processed} notification(s)")
+                    logger.info(
+                        "Worker heartbeat",
+                        extra={"event": "worker_heartbeat", "processed": processed, "queue": QUEUE_NAME},
+                    )
 
                 # Sleep between polls
                 await asyncio.sleep(poll_interval)
 
             except asyncio.CancelledError:
                 raise
-            except Exception as e:
-                logger.error(f"Worker loop error: {e}", exc_info=True)
+            except Exception:
+                logger.error("Worker loop error", exc_info=True, extra={"event": "worker_loop_failure"})
                 await asyncio.sleep(5)