fix: adopt rotated database leases

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02b22-9638-76d2-bbff-b7ea1770b118

src/sbom_nexus/api.py

@@ -207,9 +207,17 @@ def create_app(database_path: str | Path | None = None) -> FastAPI:
         version="0.1.0",
         description="SBOM capture, history, evaluation, and bounded catch-up service",
     )
-    application.state.store = Store(
+    configured_target = (
         database_path if database_path is not None else database_target("sbom-nexus.db")
     )
+    application.state.store = Store(
+        configured_target,
+        database_url_file=(
+            os.getenv("SBOM_NEXUS_DATABASE_URL_FILE")
+            if database_path is None
+            else None
+        ),
+    )
     if _auto_create(application.state.store):
         application.state.store.init_schema()
 
@@ -219,6 +227,10 @@ def create_app(database_path: str | Path | None = None) -> FastAPI:
         store.health()
         return {"status": "ok", "store": "connected", "dialect": store.dialect}
 
+    @application.get("/state/live")
+    def live() -> dict[str, str]:
+        return {"status": "ok"}
+
     @application.put("/repositories/{repo_slug}")
     def upsert_repository(
         repo_slug: str, body: RepositoryUpsert, request: Request

src/sbom_nexus/database.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import os
 from pathlib import Path
 
+import psycopg
 from sqlalchemy import (
     JSON,
     Boolean,
@@ -128,9 +130,49 @@ def database_url(value: str | Path) -> str:
     return text
 
 
-def create_database_engine(value: str | Path) -> Engine:
+def _dynamic_connection_factory(url_file: Path):
+    def connect():
+        try:
+            value = url_file.read_text(encoding="utf-8").strip()
+        except OSError as exc:
+            raise RuntimeError("Unable to read mounted database URL") from exc
+        if not value:
+            raise RuntimeError("Mounted database URL is empty")
+        url = database_url(value)
+        if not url.startswith("postgresql+psycopg://"):
+            raise RuntimeError("Mounted database URL must use PostgreSQL")
+        return psycopg.connect(
+            url.replace("postgresql+psycopg://", "postgresql://", 1)
+        )
+
+    return connect
+
+
+def create_database_engine(
+    value: str | Path, *, database_url_file: str | Path | None = None
+) -> Engine:
     url = database_url(value)
     options: dict[str, object] = {"pool_pre_ping": True}
+    if database_url_file:
+        if not url.startswith("postgresql+psycopg://"):
+            raise RuntimeError("Dynamic database URL files require PostgreSQL")
+        try:
+            recycle_seconds = int(
+                os.getenv("SBOM_NEXUS_DATABASE_POOL_RECYCLE_SECONDS", "300")
+            )
+        except ValueError as exc:
+            raise RuntimeError(
+                "SBOM_NEXUS_DATABASE_POOL_RECYCLE_SECONDS must be an integer"
+            ) from exc
+        if recycle_seconds < 1:
+            raise RuntimeError(
+                "SBOM_NEXUS_DATABASE_POOL_RECYCLE_SECONDS must be positive"
+            )
+        options["creator"] = _dynamic_connection_factory(Path(database_url_file))
+        options["pool_recycle"] = recycle_seconds
+        # Keep only the dialect in SQLAlchemy's URL; the creator rereads the
+        # mounted value without retaining credentials in engine repr/logging.
+        url = "postgresql+psycopg://"
     if url.startswith("sqlite:"):
         options["connect_args"] = {"check_same_thread": False}
     engine = create_engine(url, **options)

src/sbom_nexus/storage.py

@@ -50,8 +50,15 @@ def datetime_text(value: datetime | str | None) -> str | None:
 
 
 class Store:
-    def __init__(self, database_target: str | Path) -> None:
-        self.engine = create_database_engine(database_target)
+    def __init__(
+        self,
+        database_target: str | Path,
+        *,
+        database_url_file: str | Path | None = None,
+    ) -> None:
+        self.engine = create_database_engine(
+            database_target, database_url_file=database_url_file
+        )
 
     @property
     def dialect(self) -> str: