diff --git a/alembic/env.py b/alembic/env.py index f169e14..4da8bc3 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -20,25 +20,22 @@ target_metadata = Base.metadata def get_database_url() -> str: - """Get database URL from environment or config.""" + """Get the database URL using the same resolution as the app. + + Delegates to CommonSettings.effective_database_url so DATABASE_BACKEND=postgres + (+ DATABASE_* components) and an explicit DATABASE_URL are honoured identically to + the running services — otherwise migrations would silently target SQLite. + """ from pathlib import Path - # First try explicit DATABASE_URL environment variable - url = os.environ.get("DATABASE_URL") - if url: - # Ensure directory exists for sqlite URLs - if url.startswith("sqlite:///"): - db_path = Path(url.replace("sqlite:///", "")) - db_path.parent.mkdir(parents=True, exist_ok=True) - return url - # Try DATA_HOME environment variable - data_home = os.environ.get("DATA_HOME") - if data_home: - db_path = Path(data_home) / "collector" / "meshcore.db" + from meshcore_hub.common.config import CommonSettings + + url = CommonSettings().effective_database_url + # Ensure the parent directory exists for SQLite file URLs. + if url.startswith("sqlite:///"): + db_path = Path(url.replace("sqlite:///", "")) db_path.parent.mkdir(parents=True, exist_ok=True) - return f"sqlite:///{db_path}" - # Fall back to alembic.ini - return config.get_main_option("sqlalchemy.url", "sqlite:///./meshcore.db") + return url def get_schema(url: str) -> str | None: diff --git a/alembic/versions/20260421_0001_normalize_public_key_case.py b/alembic/versions/20260421_0001_normalize_public_key_case.py index e2f8441..a980e98 100644 --- a/alembic/versions/20260421_0001_normalize_public_key_case.py +++ b/alembic/versions/20260421_0001_normalize_public_key_case.py @@ -32,14 +32,22 @@ depends_on = None def upgrade() -> None: conn = op.get_bind() + # Aggregate the duplicate ids per lowercased key. SQLite uses GROUP_CONCAT; + # Postgres uses STRING_AGG. HAVING references COUNT(*) directly (not the alias) + # since Postgres does not allow SELECT aliases in HAVING. + if conn.dialect.name == "postgresql": + id_agg = "STRING_AGG(id, ',')" + else: + id_agg = "GROUP_CONCAT(id)" + # Find groups of duplicate nodes (same lowercase public_key, different actual case) - duplicates = conn.execute(text(""" + duplicates = conn.execute(text(f""" SELECT LOWER(public_key) AS lower_pk, - GROUP_CONCAT(id) AS ids, + {id_agg} AS ids, COUNT(*) AS cnt FROM nodes GROUP BY LOWER(public_key) - HAVING cnt > 1 + HAVING COUNT(*) > 1 """)).fetchall() for row in duplicates: diff --git a/alembic/versions/20260612_1947_e9f0c4079540_add_raw_packets_table.py b/alembic/versions/20260612_1947_e9f0c4079540_add_raw_packets_table.py index d74188f..ddd379e 100644 --- a/alembic/versions/20260612_1947_e9f0c4079540_add_raw_packets_table.py +++ b/alembic/versions/20260612_1947_e9f0c4079540_add_raw_packets_table.py @@ -10,7 +10,6 @@ from typing import Sequence, Union from alembic import op import sqlalchemy as sa -from sqlalchemy.dialects import sqlite # revision identifiers, used by Alembic. revision: str = "e9f0c4079540" @@ -34,7 +33,7 @@ def upgrade() -> None: sa.Column("route_type", sa.String(length=20), nullable=True), sa.Column("path_len", sa.Integer(), nullable=True), sa.Column("snr", sa.Float(), nullable=True), - sa.Column("decoded", sqlite.JSON(), nullable=True), + sa.Column("decoded", sa.JSON(), nullable=True), sa.Column("received_at", sa.DateTime(timezone=True), nullable=False), sa.Column("id", sa.String(), nullable=False), sa.Column( diff --git a/docker-compose.yml b/docker-compose.yml index 998127c..6fa0a5d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -147,6 +147,36 @@ services: timeout: 5s retries: 3 + # ========================================================================== + # PostgreSQL - optional database backend (alternative to default SQLite) + # Enable with: DATABASE_BACKEND=postgres (+ DATABASE_* vars) AND the + # 'postgres' profile, e.g. `docker compose --profile core --profile postgres up`. + # POSTGRES_* init vars derive from the same DATABASE_* values (single source + # of truth); the schema itself is created by `db upgrade` (alembic). + # ========================================================================== + postgres: + image: postgres:17-alpine + container_name: ${COMPOSE_PROJECT_NAME:-hub}-postgres + profiles: + - postgres + restart: unless-stopped + environment: + - POSTGRES_USER=${DATABASE_USER:-meshcorehub} + - POSTGRES_PASSWORD=${DATABASE_PASSWORD:-} + - POSTGRES_DB=${DATABASE_NAME:-meshcorehub} + volumes: + - postgres_data:/var/lib/postgresql/data + healthcheck: + test: + [ + "CMD-SHELL", + "pg_isready -U ${DATABASE_USER:-meshcorehub} -d ${DATABASE_NAME:-meshcorehub}", + ] + interval: 10s + timeout: 5s + retries: 5 + start_period: 10s + # ========================================================================== # Collector - MQTT subscriber and database storage # ========================================================================== @@ -179,6 +209,14 @@ services: - CHANNEL_REFRESH_INTERVAL_SECONDS=${CHANNEL_REFRESH_INTERVAL_SECONDS:-300} - DATA_HOME=/data - SEED_HOME=/seed + # Database backend (SQLite default; set DATABASE_BACKEND=postgres to switch) + - DATABASE_BACKEND=${DATABASE_BACKEND:-sqlite} + - DATABASE_HOST=${DATABASE_HOST:-postgres} + - DATABASE_PORT=${DATABASE_PORT:-5432} + - DATABASE_NAME=${DATABASE_NAME:-meshcorehub} + - DATABASE_SCHEMA=${DATABASE_SCHEMA:-meshcorehub} + - DATABASE_USER=${DATABASE_USER:-meshcorehub} + - DATABASE_PASSWORD=${DATABASE_PASSWORD:-} # Webhook configuration - WEBHOOK_ADVERTISEMENT_URL=${WEBHOOK_ADVERTISEMENT_URL:-} - WEBHOOK_ADVERTISEMENT_SECRET=${WEBHOOK_ADVERTISEMENT_SECRET:-} @@ -240,6 +278,14 @@ services: - MQTT_TRANSPORT=${MQTT_TRANSPORT:-websockets} - MQTT_WS_PATH=${MQTT_WS_PATH:-/} - DATA_HOME=/data + # Database backend (SQLite default; set DATABASE_BACKEND=postgres to switch) + - DATABASE_BACKEND=${DATABASE_BACKEND:-sqlite} + - DATABASE_HOST=${DATABASE_HOST:-postgres} + - DATABASE_PORT=${DATABASE_PORT:-5432} + - DATABASE_NAME=${DATABASE_NAME:-meshcorehub} + - DATABASE_SCHEMA=${DATABASE_SCHEMA:-meshcorehub} + - DATABASE_USER=${DATABASE_USER:-meshcorehub} + - DATABASE_PASSWORD=${DATABASE_PASSWORD:-} - API_HOST=0.0.0.0 - API_PORT=8000 # Worker processes for multi-core concurrency (1 = single process). @@ -374,10 +420,23 @@ services: - core - migrate restart: "no" + # Wait for postgres only when it's running (postgres profile active); + # required:false means SQLite deployments don't pull it in or block on it. + depends_on: + postgres: + condition: service_healthy + required: false volumes: - data:/data environment: - DATA_HOME=/data + - DATABASE_BACKEND=${DATABASE_BACKEND:-sqlite} + - DATABASE_HOST=${DATABASE_HOST:-postgres} + - DATABASE_PORT=${DATABASE_PORT:-5432} + - DATABASE_NAME=${DATABASE_NAME:-meshcorehub} + - DATABASE_SCHEMA=${DATABASE_SCHEMA:-meshcorehub} + - DATABASE_USER=${DATABASE_USER:-meshcorehub} + - DATABASE_PASSWORD=${DATABASE_PASSWORD:-} command: ["db", "upgrade"] # ========================================================================== @@ -417,3 +476,5 @@ volumes: name: ${COMPOSE_PROJECT_NAME:-hub}_observer_data redis_data: name: ${COMPOSE_PROJECT_NAME:-hub}_redis_data + postgres_data: + name: ${COMPOSE_PROJECT_NAME:-hub}_postgres_data diff --git a/src/meshcore_hub/common/database.py b/src/meshcore_hub/common/database.py index 9ad9dbc..0ad7dff 100644 --- a/src/meshcore_hub/common/database.py +++ b/src/meshcore_hub/common/database.py @@ -1,5 +1,6 @@ """Database connection and session management.""" +import os from contextlib import asynccontextmanager, contextmanager from typing import Any, AsyncGenerator, Generator @@ -11,19 +12,37 @@ from sqlalchemy.orm import Session, sessionmaker from meshcore_hub.common.models.base import Base +def _resolve_pg_schema(database_url: str, schema: str | None) -> str | None: + """Resolve the Postgres schema to scope a connection to (search_path). + + Returns None for SQLite (no schema concept). For Postgres, an explicit ``schema`` + wins; otherwise it falls back to the ``DATABASE_SCHEMA`` env var. The CLI's + ``load_dotenv()`` and Docker both populate that var, so runtime entrypoints don't + need to thread the schema through every constructor. + """ + if not database_url.startswith(("postgresql", "postgres")): + return None + if schema: + return schema + return os.environ.get("DATABASE_SCHEMA") + + def _to_async_url(database_url: str) -> str: """Map a sync database URL to its async-driver equivalent. - Leaves an already driver-qualified URL (``dialect+driver://``) untouched so an - explicit driver choice is respected. + Postgres always maps to asyncpg for the async engine, even when the sync URL + names a sync driver (e.g. ``postgresql+psycopg2://``, which is what the config + assembler produces) — otherwise async sessions would try to use the sync driver. + SQLite maps to aiosqlite unless a driver is already specified. """ scheme = database_url.split("://", 1)[0] - if "+" in scheme: - return database_url - if scheme == "sqlite": - return database_url.replace("sqlite://", "sqlite+aiosqlite://", 1) - if scheme in ("postgresql", "postgres"): + dialect = scheme.split("+", 1)[0] + if dialect in ("postgresql", "postgres"): return database_url.replace(f"{scheme}://", "postgresql+asyncpg://", 1) + if dialect == "sqlite": + if "+" in scheme: + return database_url + return database_url.replace("sqlite://", "sqlite+aiosqlite://", 1) return database_url @@ -37,8 +56,8 @@ def create_database_engine( Args: database_url: SQLAlchemy database URL echo: Enable SQL query logging - schema: Postgres schema to scope connections to via search_path. Ignored for - SQLite (which has no schema concept). + schema: Postgres schema to scope connections to via search_path. Defaults to + the DATABASE_SCHEMA env var when not given. Ignored for SQLite. Returns: SQLAlchemy Engine instance @@ -53,8 +72,9 @@ def create_database_engine( # Scope Postgres connections to the configured schema via search_path. This keeps # the models schema-agnostic (no hardcoded schema=) so the same code serves SQLite, # single-instance Postgres, and multiple schema-isolated instances on one cluster. - if schema and database_url.startswith(("postgresql", "postgres")): - connect_args["options"] = f"-csearch_path={schema}" + resolved_schema = _resolve_pg_schema(database_url, schema) + if resolved_schema: + connect_args["options"] = f"-csearch_path={resolved_schema}" # Size the pool above the default Starlette threadpool (~40 threads) so # concurrent request handlers don't block waiting for a connection. Applies @@ -143,12 +163,12 @@ class DatabaseManager: Args: database_url: SQLAlchemy database URL echo: Enable SQL query logging - schema: Postgres schema to scope connections to (search_path); ignored for - SQLite + schema: Postgres schema to scope connections to (search_path). Defaults to + the DATABASE_SCHEMA env var when not given; ignored for SQLite. """ self.database_url = database_url self._echo = echo - self._schema = schema + self._schema = _resolve_pg_schema(database_url, schema) # Ensure parent directory exists for SQLite databases if database_url.startswith("sqlite:///"): @@ -175,8 +195,9 @@ class DatabaseManager: async_url = _to_async_url(self.database_url) async_connect_args: dict[str, Any] = {} # asyncpg sets search_path via server_settings (not the libpq -c options - # string the sync psycopg2 engine uses). - if self._schema and self.database_url.startswith(("postgresql", "postgres")): + # string the sync psycopg2 engine uses). self._schema is already resolved + # (explicit arg or DATABASE_SCHEMA env) and None for SQLite. + if self._schema: async_connect_args["server_settings"] = {"search_path": self._schema} self._async_engine = create_async_engine( async_url, echo=self._echo, connect_args=async_connect_args diff --git a/tests/test_common/test_database.py b/tests/test_common/test_database.py index 6cc6284..51dc9a0 100644 --- a/tests/test_common/test_database.py +++ b/tests/test_common/test_database.py @@ -2,9 +2,14 @@ from pathlib import Path +import pytest from sqlalchemy import text -from meshcore_hub.common.database import create_database_engine +from meshcore_hub.common.database import ( + _resolve_pg_schema, + _to_async_url, + create_database_engine, +) class TestSqlitePragmas: @@ -34,3 +39,42 @@ class TestSqlitePragmas: assert conn.execute(text("SELECT 1")).scalar() == 1 finally: engine.dispose() + + +class TestAsyncUrlMapping: + """Map sync URLs to their async-driver equivalents for the async engine.""" + + @pytest.mark.parametrize( + "sync_url,expected", + [ + ("sqlite:///x.db", "sqlite+aiosqlite:///x.db"), + ("sqlite+aiosqlite:///x.db", "sqlite+aiosqlite:///x.db"), + ("postgresql://u:p@h/db", "postgresql+asyncpg://u:p@h/db"), + ("postgres://u:p@h/db", "postgresql+asyncpg://u:p@h/db"), + # config assembles +psycopg2; the async engine must still use asyncpg + ("postgresql+psycopg2://u:p@h/db", "postgresql+asyncpg://u:p@h/db"), + ("postgresql+asyncpg://u:p@h/db", "postgresql+asyncpg://u:p@h/db"), + ], + ) + def test_to_async_url(self, sync_url: str, expected: str) -> None: + assert _to_async_url(sync_url) == expected + + +class TestSchemaResolution: + """search_path schema resolution (explicit arg vs DATABASE_SCHEMA env).""" + + def test_sqlite_never_has_schema(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DATABASE_SCHEMA", "ignored") + assert _resolve_pg_schema("sqlite:///x.db", None) is None + assert _resolve_pg_schema("sqlite:///x.db", "explicit") is None + + def test_explicit_schema_wins(self) -> None: + assert _resolve_pg_schema("postgresql://u@h/db", "prod") == "prod" + + def test_falls_back_to_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DATABASE_SCHEMA", "stg") + assert _resolve_pg_schema("postgresql://u@h/db", None) == "stg" + + def test_none_when_no_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("DATABASE_SCHEMA", raising=False) + assert _resolve_pg_schema("postgresql://u@h/db", None) is None