diff --git a/app/logging_utils.py b/app/logging_utils.py
new file mode 100644
index 0000000..e81628e
--- /dev/null
+++ b/app/logging_utils.py
@@ -0,0 +1,29 @@
+"""Structured logging utility — eliminates boilerplate across domain services."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+
+def structured_log(
+    logger: logging.Logger,
+    level: str,
+    event: str,
+    /,
+    **fields: Any,
+) -> None:
+    """Emit a structured log entry.
+
+    The event name is passed as the log message. The JsonLogFormatter in
+    logging_config.py extracts it via record.getMessage() when no explicit
+    'event' key exists in extra — so we do NOT duplicate it.
+
+    Usage:
+        structured_log(logger, "info", "ingestion.run_started", user_id=1, scholar_count=5)
+    """
+    fields.pop("metric_name", None)
+    fields.pop("metric_value", None)
+
+    log_method = getattr(logger, level.lower())
+    log_method(event, extra=fields)
diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py
index 84cf882..2a08d4c 100644
--- a/tests/unit/test_logging.py
+++ b/tests/unit/test_logging.py
@@ -7,7 +7,8 @@ from unittest.mock import AsyncMock
 
 from fastapi.testclient import TestClient
 
-from app.logging_config import JsonLogFormatter, parse_redact_fields
+from app.logging_config import ConsoleLogFormatter, JsonLogFormatter, parse_redact_fields
+from app.logging_utils import structured_log
 from app.main import app
 from app.http.middleware import REQUEST_ID_HEADER, parse_skip_paths
 
@@ -53,3 +54,67 @@ def test_parse_skip_paths_trims_and_discards_empty_segments() -> None:
         "/healthz",
         "/api/v1/metrics",
     )
+
+
+# --- structured_log tests ---
+
+
+def _capture_structured_log(caplog, level, event, **fields):
+    """Helper: call structured_log and return the captured LogRecord."""
+    logger = logging.getLogger("tests.structured")
+    with caplog.at_level(logging.DEBUG, logger="tests.structured"):
+        structured_log(logger, level, event, **fields)
+    return caplog.records[-1]
+
+
+def test_structured_log_json_formatter_uses_event_as_message(caplog) -> None:
+    record = _capture_structured_log(caplog, "info", "ingestion.run_started", user_id=42)
+    formatter = JsonLogFormatter(redact_fields=set())
+    payload = json.loads(formatter.format(record))
+
+    assert payload["event"] == "ingestion.run_started"
+    assert payload["user_id"] == 42
+
+
+def test_structured_log_console_formatter(caplog) -> None:
+    record = _capture_structured_log(caplog, "warning", "export.failed", scholar_id=7)
+    formatter = ConsoleLogFormatter(redact_fields=set())
+    output = formatter.format(record)
+
+    assert "export.failed" in output
+    assert "scholar_id=7" in output
+
+
+def test_structured_log_strips_metric_fields(caplog) -> None:
+    record = _capture_structured_log(
+        caplog,
+        "info",
+        "scrape.complete",
+        metric_name="articles_scraped",
+        metric_value=15,
+        scholar_id=3,
+    )
+    formatter = JsonLogFormatter(redact_fields=set())
+    payload = json.loads(formatter.format(record))
+
+    assert "metric_name" not in payload
+    assert "metric_value" not in payload
+    assert payload["scholar_id"] == 3
+
+
+def test_structured_log_extra_fields_in_output(caplog) -> None:
+    record = _capture_structured_log(
+        caplog,
+        "info",
+        "scholar.created",
+        user_id=1,
+        scholar_id=99,
+        scholar_name="Ada Lovelace",
+    )
+    formatter = JsonLogFormatter(redact_fields=set())
+    payload = json.loads(formatter.format(record))
+
+    assert payload["event"] == "scholar.created"
+    assert payload["user_id"] == 1
+    assert payload["scholar_id"] == 99
+    assert payload["scholar_name"] == "Ada Lovelace"