feat(dev): add protected recent logs endpoint

Author: 1mpossible-code

server/app.py

@@ -1,5 +1,6 @@
 import logging
 import os
+from collections import deque
 from datetime import datetime, timezone
 from http import HTTPStatus
 from time import monotonic
@@ -16,6 +17,24 @@
 DEFAULT_PORT = 8000
 DEFAULT_CORS_ORIGINS = "http://localhost:3000,http://127.0.0.1:3000"
 APP_START_TIME = monotonic()
+LOG_BUFFER = deque(maxlen=200)
+
+
+class InMemoryLogHandler(logging.Handler):
+    """Keep recent log records available for developer diagnostics."""
+
+    def emit(self, record):
+        LOG_BUFFER.append(
+            {
+                "timestamp": datetime.fromtimestamp(record.created, timezone.utc)
+                .replace(microsecond=0)
+                .isoformat()
+                .replace("+00:00", "Z"),
+                "level": record.levelname,
+                "logger": record.name,
+                "message": record.getMessage(),
+            }
+        )
 
 
 def get_runtime_version() -> str:
@@ -54,6 +73,10 @@ def get_cache_enabled() -> bool:
     }
 
 
+def get_recent_logs(limit: int = 100) -> list[dict[str, str]]:
+    return list(LOG_BUFFER)[-max(1, min(limit, 200)):]
+
+
 def _database_dependency_status() -> str:
     try:
         client = db_connect.connect_db()
@@ -120,6 +143,9 @@ def create_app():
         format="%(asctime)s [%(levelname)s] %(message)s",
         datefmt="%Y-%m-%d %H:%M:%S",
     )
+    root_logger = logging.getLogger()
+    if not any(isinstance(handler, InMemoryLogHandler) for handler in root_logger.handlers):
+        root_logger.addHandler(InMemoryLogHandler())
 
     cors_origins = get_runtime_cors_origins()
     CORS(app, resources={r"/*": {"origins": cors_origins}})

server/endpoints.py

@@ -3,6 +3,7 @@
 import os
 from http import HTTPStatus
 
+from flask import request
 from flask_restx import Resource, Namespace
 
 import data.countries as countries_db
@@ -12,6 +13,7 @@
     APP_NAME,
     get_cache_enabled,
     get_health_payload,
+    get_recent_logs,
     get_runtime_environment,
     get_runtime_log_level,
     get_runtime_port,
@@ -67,6 +69,12 @@ def _get_safe_database_config() -> dict[str, str | bool]:
     }
 
 
+def _has_dev_logs_access() -> bool:
+    expected_token = os.getenv("DEV_LOGS_TOKEN", "")
+    provided_token = request.headers.get("X-Dev-Token", "")
+    return bool(expected_token) and provided_token == expected_token
+
+
 @general_ns.route("/hello")
 class HelloWorld(Resource):
     """
@@ -154,3 +162,15 @@ class Health(Resource):
     def get(self):
         payload, status = get_health_payload()
         return payload, status
+
+
+@general_ns.route("/dev/logs")
+class DevLogs(Resource):
+    """Return recent application logs for developers only."""
+
+    def get(self):
+        if not _has_dev_logs_access():
+            return {"message": "forbidden"}, HTTPStatus.FORBIDDEN
+
+        limit = request.args.get("limit", default=100, type=int)
+        return {"logs": get_recent_logs(limit)}, HTTPStatus.OK

server/tests/test_dev_logs.py

@@ -0,0 +1,29 @@
+import logging
+
+from server.app import create_app
+
+
+def test_dev_logs_requires_token(monkeypatch):
+    monkeypatch.setenv('DEV_LOGS_TOKEN', 'secret-token')
+    app = create_app()
+
+    with app.test_client() as client:
+        response = client.get('/dev/logs')
+
+    assert response.status_code == 403
+
+
+def test_dev_logs_returns_recent_entries(monkeypatch):
+    monkeypatch.setenv('DEV_LOGS_TOKEN', 'secret-token')
+    app = create_app()
+    logging.getLogger('server.tests.dev_logs').warning('developer log endpoint smoke test')
+
+    with app.test_client() as client:
+        response = client.get('/dev/logs?limit=5', headers={'X-Dev-Token': 'secret-token'})
+
+    payload = response.get_json()
+
+    assert response.status_code == 200
+    assert payload['logs']
+    assert any(entry['message'] == 'developer log endpoint smoke test' for entry in payload['logs'])
+    assert len(payload['logs']) <= 5