Dockerfile

@@ -7,7 +7,7 @@ WORKDIR /app
 COPY src/ /app/
 
 # Runtime image
-FROM nvcr.io/nvidia/distroless/python:3.12-v3.5.1
+FROM nvcr.io/nvidia/distroless/python:3.12-v4.1.6
 COPY --from=builder /install /usr/local
 COPY --from=builder /app /app
 WORKDIR /app

pyproject.toml

@@ -0,0 +1,28 @@
+[tool.ruff]
+line-length = 120
+target-version = "py312"
+[tool.ruff.lint]
+select = [
+	"E", "F", # pycodestyle errors, pyflakes (unused imports/variables, undefined names)
+	"I", # import sorting (replaces isort)
+	"B", # bugbear: likely bugs
+	"ASYNC", # async mistakes
+	"RUF006", # un-referenced asyncio.create_task (tasks can be garbage-collected)
+]
+ignore = ["E501"]
+
+[tool.ruff.lint.isort]
+known-first-party = ["db_module", "metrics_server", "rabbitmq_handler", "secret_manager",
+					"simple_logger_handler", "uvicorn_logger_config", "validator"]
+
+[tool.ruff.format]
+indent-style = "tab"
+
+[tool.ruff.lint.flake8-bugbear]
+extend-immutable-calls = [
+	"fastapi.Depends",
+	"fastapi.Query",
+	"fastapi.Header",
+	"fastapi.Body",
+	"fastapi.Security",
+]

requirements-dev.txt

@@ -0,0 +1 @@
+ruff==0.17.0

requirements.txt

@@ -24,3 +24,6 @@ starlette==0.48.0
 typing-inspection==0.4.2
 typing_extensions==4.15.0
 uvicorn==0.37.0
+certifi==2026.7.22
+httpcore==1.0.9
+httpx==0.28.1

src/main.py

@@ -1,26 +1,33 @@
-from fastapi import FastAPI, Depends, HTTPException, Request
+import asyncio
+import os
+import uuid
+from contextlib import asynccontextmanager
+from hashlib import sha256
+from typing import List, Optional
+
+import httpx
+import uvicorn
+from fastapi import Depends, FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse
 from fastapi.security.api_key import APIKeyHeader
-from starlette.exceptions import HTTPException as StarletteHTTPException
-from typing import Optional,List
 from pydantic import BaseModel
-from validator import is_valid_platform,is_valid_token,verify_api_key
+from starlette.exceptions import HTTPException as StarletteHTTPException
-from secret_manager.fernet_encryption import encrypt_token
+
-from secret_manager import cleanup_secret_files
+from db_module import close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread
-from db_module import get_db, create_connection_pool, close_connection_pool, start_healthcheck_thread
-from simple_logger_handler import setup_logger, LOG_LEVEL
-import uuid
-from hashlib import sha256
-import uvicorn
-from contextlib import asynccontextmanager
 from metrics_server import REQUEST_COUNTER
-import asyncio
+from secret_manager import cleanup_secret_files, return_credentials
+from secret_manager.fernet_encryption import encrypt_token
+from simple_logger_handler import LOG_LEVEL, setup_logger
 from uvicorn_logger_config import LOGGING_CONFIG
+from validator import is_valid_platform, is_valid_token, verify_api_key
 
 logger = setup_logger(__name__)
-
 api_key_header = APIKeyHeader(name="X-API-Key")
 
+class PullResponse(BaseModel):
+	messages: list[dict]
+	count: int
+	has_more: bool
 
 def hash_token(token: str) -> str:
 	logger.debug(f"Hashing token: {token}")
@@ -40,22 +47,42 @@ class TokenRequest(BaseModel):
 async def lifespan(app: FastAPI):
 	logger.info("Starting application...")
 
-	logger.info("Creating MySQL connection pool...")
+	backend_api_internal_url = os.getenv(
-	create_connection_pool("ApiPool")
+	"BACKEND_API_INTERNAL_URL",
+	"http://backend-api-internal.app-notifications.svc.cluster.local:8101",
+	)
+	api_key_backend_api = return_credentials("/etc/secrets/api_key")
 
-	start_healthcheck_thread()
+	try: 
-	logger.info("MySQL healthcheck thread started.")
+		logger.info("[FastAPI] Creating MySQL connection pool...")
+		create_connection_pool("ApiPool")
 
-	SECRET_PATHS = frozenset({
+		start_healthcheck_thread()
-		"/etc/secrets/encryption_key",
+		logger.info("[FastAPI] MySQL healthcheck thread started.")
-		"/etc/secrets/db_username",
-		"/etc/secrets/db_password"
-	})
-	cleanup_secret_files(SECRET_PATHS)
 
-	yield
+		app.state.http_client = httpx.AsyncClient (
-	logger.info("Closing MySQL connection pool...")
+			base_url=backend_api_internal_url,
-	close_connection_pool()
+			headers={"X-API-Key-Internal": api_key_backend_api},
+			timeout=httpx.Timeout(15.0,connect=3.0)
+		)
+		logger.info("[FastAPI] HTTP client initialized.")
+
+		SECRET_PATHS = frozenset({
+			"/etc/secrets/api_key",
+			"/etc/secrets/encryption_key",
+			"/etc/secrets/db_username",
+			"/etc/secrets/db_password"
+		})
+		cleanup_secret_files(SECRET_PATHS)
+
+		yield
+
+	finally:
+		logger.info("[FastAPI] Closing MySQL connection pool...")
+		close_connection_pool()
+
+		await app.state.http_client.aclose()
+		logger.info("[FastAPI] Closing Http client")
 
 
 api = FastAPI(
@@ -69,30 +96,30 @@ api = FastAPI(
 @api.middleware("http")
 async def prometheus_middleware(request: Request, call_next):
 	status = 500
-	logger.debug(f"Incoming request: {request.method} {request.url.path}")
+	logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}")
 	try:
 		response = await call_next(request)
 		status = response.status_code
-		logger.debug(f"Request processed with status {status}")
+		logger.debug(f"[FastAPI] Request processed with status {status}")
 	except Exception as e:
-		logger.exception(f"Exception during request processing: {e}")
+		logger.exception(f"[FastAPI] Exception during request processing: {e}")
 		raise
 	finally:
 		REQUEST_COUNTER.labels(request.method, request.url.path, status).inc()
-		logger.debug(f"Prometheus counter incremented for {request.method} {request.url.path} {status}")
+		logger.debug(f"[FastAPI] Prometheus counter incremented for {request.method} {request.url.path} {status}")
 	return response
 
 @api.get("/health", tags=["Health"])
 def return_health(db=Depends(get_db)):
-	logger.debug("Health check initiated")
+	logger.debug("[FastAPI] Health check initiated")
 	try:
 		cursor = db.cursor()
 		cursor.execute("SELECT 1")
 		cursor.fetchone()
 		db_status = "ok"
-		logger.debug("Database health check passed")
+		logger.debug("[FastAPI] Database health check passed")
 	except Exception as e:
-		logger.error(f"Health check DB failed: {e}")
+		logger.error(f"[FastAPI] Health check DB failed: {e}")
 		db_status = "error"
 
 	return JSONResponse(
@@ -102,19 +129,18 @@ def return_health(db=Depends(get_db)):
 
 
 def verify_api_key_dependency(db=Depends(get_db), api_key: str = Depends(api_key_header)) -> int:
-	logger.debug(f"Verifying API key: {api_key}")
 	cursor = db.cursor()
 	cursor.execute("SELECT user_id, api_key FROM users WHERE status = 'active'")
 	for user_id, hashed_key in cursor.fetchall():
 		if verify_api_key(api_key=api_key, hashed=hashed_key):
-			logger.debug(f"API key verified for user_id={user_id}")
+			logger.debug(f"[FastAPI] API key verified for user_id={user_id}")
 			return user_id
-	logger.warning("API key verification failed")
+	logger.warning("[FastAPI] API key verification failed")
-	raise HTTPException(status_code=403, detail="Unauthorized here")
+	raise HTTPException(status_code=403, detail="Unauthorized")
 
 @api.exception_handler(StarletteHTTPException)
 async def custom_http_exception_handler(request,exc):
-	logger.debug(f"Handling HTTP exception: {exc}")
+	logger.debug(f"[FastAPI] Handling HTTP exception: {exc}")
 	if exc.status_code == 404:
 		return JSONResponse(
 			status_code=401,
@@ -131,32 +157,31 @@ def register_token(
 	db = Depends(get_db),
 	user_id: int = Depends(verify_api_key_dependency)
 ):
-	logger.info(f"Registering token for user_id={user_id}, platform={request_data.platform}")
+	logger.info(f"[FastAPI] Registering token for user_id={user_id}, platform={request_data.platform}")
-	logger.debug(f"Request data: {request_data}")
+	logger.debug(f"[FastAPI] Request data: {request_data}")
 	if not is_valid_platform(request_data.platform) or not is_valid_token(request_data.token):
-		logger.warning(f"Invalid platform or token for user_id={user_id}")
+		logger.warning(f"[FastAPI] Invalid platform or token for user_id={user_id}")
 		raise HTTPException(status_code=403,detail="Unathorized")
 	
 	secure_token = encrypt_token(request_data.token)
 	hashed_token = hash_token(request_data.token)
-	logger.debug(f"Secure token: {secure_token}, Hashed token: {hashed_token}")
 
 	try: 
 		cursor = db.cursor()
 		cursor.execute(
 			"SELECT * FROM device_tokens WHERE user_id=%s AND hashed_token=%s",
-			  (user_id,hashed_token))
+			(user_id,hashed_token))
 		existing = cursor.fetchone()
-		logger.debug(f"Existing token fetched: {existing}")
+		logger.debug(f"[FastAPI] Existing token fetched: {existing}")
 
 		if existing:
-			logger.debug(f"Updating existing token for user_id={user_id}")
+			logger.debug(f"[FastAPI] Updating existing token for user_id={user_id}")
 			cursor.execute("""
 				UPDATE device_tokens
 				SET platform=%s, app_ver=%s, locale=%s, topics=%s, last_seen_at=NOW()
 				WHERE user_id=%s AND hashed_token=%s
-			""", (request_data.platform,
+			""",(request_data.platform,
-		 		request_data.app_ver,
+				request_data.app_ver,
 				request_data.locale,
 				request_data.topics,
 				user_id,
@@ -164,13 +189,13 @@ def register_token(
 				))
 		else:
 			token_id = str(uuid.uuid4())
-			logger.debug(f"Creating new entry user_id={user_id}, token_id={token_id}")
+			logger.debug(f"[FastAPI] Creating new entry user_id={user_id}, token_id={token_id}")
 			cursor.execute("""
 				INSERT INTO device_tokens
 				(token_id, user_id, platform, token, hashed_token, status, app_ver, locale, topics, created_at)
 				VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW())
 			""", (token_id,
-		 		user_id,
+				user_id,
 				request_data.platform,
 				secure_token,
 				hashed_token,
@@ -181,10 +206,10 @@ def register_token(
 				))
 
 		db.commit()
-		logger.info(f"Success: Registering token for user_id={user_id}, platform={request_data.platform}")
+		logger.info(f"[FastAPI] Success: Registering token for user_id={user_id}, platform={request_data.platform}")
 	except Exception as e:
-		logger.exception(f"Failed to register token for user_id={user_id}: {e}")
+		logger.exception(f"[FastAPI] Failed to register token for user_id={user_id}: {e}")
-		raise HTTPException(status_code=500, detail=str(e))
+		raise HTTPException(status_code=500, detail=str(e)) from e
 
 	return {"status":"registered"}	
 
@@ -195,9 +220,9 @@ def unregister_token(
 	db = Depends(get_db),
 	user_id: int = Depends(verify_api_key_dependency)
 ):
-	logger.info(f"Unregistering token for user_id={user_id}, platform={request_data.platform}")
+	logger.info(f"[FastAPI] Unregistering token for user_id={user_id}, platform={request_data.platform}")
 	hashed_token = hash_token(request_data.token)
-	logger.debug(f"Hashed token for removal: {hashed_token}")
+	logger.debug(f"[FastAPI] Hashed token for removal: {hashed_token}")
 	try:
 		cursor = db.cursor()
 		cursor.execute("""
@@ -206,13 +231,35 @@ def unregister_token(
 				WHERE hashed_token=%s
 			""", ('expired', hashed_token))
 	except Exception as e:
-		logger.exception(f"Failed to unregister token for user_id={user_id}: {e}")
+		logger.exception(f"[FastAPI] Failed to unregister token for user_id={user_id}: {e}")
-		raise HTTPException(status_code=500, detail=str(e))
+		raise HTTPException(status_code=500, detail=str(e)) from e
-	logger.info(f"Success: Unregistering token for user_id={user_id}, platform={request_data.platform}")
+	logger.info(f"[FastAPI] Success: Unregistering token for user_id={user_id}, platform={request_data.platform}")
 	return {"status":"unregistered"}
 
+@api.get("/internal/pull-notifications", response_model=PullResponse)
+async def pull_notifications(
+	request: Request,
+	user_id: int = Depends(verify_api_key_dependency)
+):
+	logger.info(f"[FastAPI] User id {user_id} is requesting messages")
+	client: httpx.AsyncClient = request.app.state.http_client
+
+	try:
+		resp = await client.post("/internal/pull-notifications",json={"user_id": user_id})
+		resp.raise_for_status()
+		data = PullResponse.model_validate(resp.json())
+	except httpx.HTTPStatusError as e:
+		logger.error(f"[FastAPI] Upstream returned {e.response.status_code} for user_id={user_id}")
+		raise HTTPException(status_code=502, detail="Backend-API-Internal unavailable") from e
+	except ValueError as e:
+		logger.error(f"[FastAPI] Backend-API-Internal call failed with ValueError for user_id={user_id}: {e!r}")
+		raise HTTPException(status_code=502, detail="Backend-API-Internal unavailable") from e
+
+	logger.info(f"[FastAPI] Returning {len(data.messages)} message(s) for user_id={user_id}, has_more={data.has_more}")
+	return data
+
 async def start_servers():
-	logger.debug("Starting servers...")
+	logger.debug("[FastAPI] Starting servers...")
 	config_main = uvicorn.Config("main:api", host="0.0.0.0", port=8100, log_level=LOG_LEVEL.lower(), log_config=LOGGING_CONFIG)
 	config_metrics = uvicorn.Config("metrics_server:metrics_api", host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(),log_config=LOGGING_CONFIG)
 
@@ -220,7 +267,7 @@ async def start_servers():
 	server_metrics = uvicorn.Server(config_metrics)
 
 	await asyncio.gather(server_main.serve(), server_metrics.serve())
-	logger.debug("Servers started")
+	logger.debug("[FastAPI] Servers started")
 
 if __name__ == "__main__":
 	asyncio.run(start_servers())

src/metrics_server.py

@@ -1,5 +1,5 @@
 from fastapi import FastAPI, Response
-from prometheus_client import generate_latest, CONTENT_TYPE_LATEST, Counter
+from prometheus_client import CONTENT_TYPE_LATEST, Counter, generate_latest
 
 metrics_api = FastAPI(title="Metrics Server", description="Prometheus metrics endpoint")
 

src/validator.py

@@ -1,5 +1,7 @@
-from argon2 import PasswordHasher
 import re
+
+from argon2 import PasswordHasher
+
 from simple_logger_handler import setup_logger
 
 logger = setup_logger(__name__)