From 6ce567646b971bf9fc2d42f6cca31af4b8aa9231 Mon Sep 17 00:00:00 2001
From: Amirhossein Khalili <amirhosseinkhalili901@gmail.com>
Date: Sun, 20 Sep 2026 10:40:13 +0330
Subject: [PATCH] fix(worker): return structured internal endpoint errors

---
 app/worker/internal_server.py | 32 ++++++++++++++++++++++------
 tests/test_worker.py          | 40 +++++++++++++++++++++++++++++++++++
 2 files changed, 66 insertions(+), 6 deletions(-)

diff --git a/app/worker/internal_server.py b/app/worker/internal_server.py
index 2d5e51a..b55f05a 100644
--- a/app/worker/internal_server.py
+++ b/app/worker/internal_server.py
@@ -14,11 +14,36 @@ class WorkerHTTPRequestHandler(BaseHTTPRequestHandler):
     """
     Lightweight HTTP request handler for worker internal metrics and stats.
     """
+    error_content_type = "application/json"
 
     def log_message(self, format: str, *args) -> None:
         # Suppress default noisy stderr logging
         pass
 
+    def send_error(self, code: int, message: Optional[str] = None, explain: Optional[str] = None) -> None:
+        from http import HTTPStatus
+        try:
+            if hasattr(self, "headers") and self.headers:
+                content_length = int(self.headers.get("Content-Length", 0))
+                if content_length > 0:
+                    self.rfile.read(min(content_length, 65536))
+        except Exception:
+            pass
+        try:
+            status = HTTPStatus(code)
+            short_msg = status.phrase
+        except ValueError:
+            short_msg = "error"
+        err_msg = message or short_msg
+        detail = explain or err_msg
+        body = json.dumps({"error": err_msg, "detail": detail}).encode("utf-8")
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Connection", "close")
+        self.end_headers()
+        self.wfile.write(body)
+
     def do_GET(self) -> None:
         if self.path == "/internal/stats":
             stats = worker_metrics.get_internal_stats()
@@ -51,12 +76,7 @@ class WorkerHTTPRequestHandler(BaseHTTPRequestHandler):
             self.wfile.write(body)
             return
 
-        self.send_response(404)
-        self.send_header("Content-Type", "application/json")
-        body = b'{"error": "not_found"}'
-        self.send_header("Content-Length", str(len(body)))
-        self.end_headers()
-        self.wfile.write(body)
+        self.send_error(404, "not_found", f"Cannot GET {self.path}")
 
 
 class WorkerInternalServer:
diff --git a/tests/test_worker.py b/tests/test_worker.py
index fbfdf36..b354c7e 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -282,6 +282,46 @@ class TestProcessingWorker(unittest.TestCase):
             self.assertIn("cdr_processed_total", text)
             self.assertIn("cdr_dlq_total", text)
 
+        # Verify unsupported HTTP method returns 501 application/json with structured error
+        req = urllib.request.Request("http://127.0.0.1:18001/metrics", data=b"{}", method="POST")
+        try:
+            urllib.request.urlopen(req)
+            self.fail("Expected HTTPError 501")
+        except urllib.error.HTTPError as exc:
+            self.assertEqual(exc.code, 501)
+            self.assertIn("application/json", exc.headers.get("Content-Type", ""))
+            error_body = json.loads(exc.read().decode("utf-8"))
+            self.assertIn("error", error_body)
+            self.assertIn("detail", error_body)
+
+        # Every error endpoint must use the same structured error contract.
+        try:
+            urllib.request.urlopen("http://127.0.0.1:18001/not-a-route")
+            self.fail("Expected HTTPError 404")
+        except urllib.error.HTTPError as exc:
+            self.assertEqual(exc.code, 404)
+            self.assertIn("application/json", exc.headers.get("Content-Type", ""))
+            error_body = json.loads(exc.read().decode("utf-8"))
+            self.assertEqual(error_body["error"], "not_found")
+            self.assertIn("detail", error_body)
+
+    def test_invalid_unicode_raw_id_routes_to_dlq_safely(self):
+        worker = ProcessingWorker(
+            consumer=self.consumer,
+            dlq_producer=self.dlq_producer,
+            clickhouse=self.clickhouse,
+            batch_size=10,
+            internal_server=self.internal_server,
+        )
+        payload_bytes = b'{"raw_id": "\\ud800", "device_id": "1", "caller_num": "+19875551234", "svc": "4G_DATA", "up_bytes": 10, "down_bytes": 20, "rat": 6, "lac": "1", "ci": "2", "epoch_start": 100, "epoch_end": 200}'
+        msg = MockKafkaMessage(payload_bytes)
+        worker.process_single_message(msg)
+
+        self.assertEqual(len(self.dlq_producer.messages), 1)
+        self.assertIsNone(self.dlq_producer.messages[0]["key"])
+        dlq_data = json.loads(self.dlq_producer.messages[0]["value"].decode("utf-8"))
+        self.assertEqual(dlq_data["failure_reason"], "normalization_error")
+
 
 if __name__ == "__main__":
     unittest.main()