From 21dcbbc56fc3942471d64e46c4e54f5b7fa546cb Mon Sep 17 00:00:00 2001 From: Louis King Date: Sat, 13 Jun 2026 21:13:46 +0100 Subject: [PATCH 01/13] Add PostgreSQL migration plan Planning doc for adding Postgres support: code compatibility fixes, a postgres container, component-based connection config, and a SQLAlchemy ORM-based migration command for existing SQLite databases. Co-Authored-By: Claude Opus 4.8 --- .../20260613-2111-postgres-migration/plan.md | 198 ++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 docs/plans/20260613-2111-postgres-migration/plan.md diff --git a/docs/plans/20260613-2111-postgres-migration/plan.md b/docs/plans/20260613-2111-postgres-migration/plan.md new file mode 100644 index 0000000..b6e5832 --- /dev/null +++ b/docs/plans/20260613-2111-postgres-migration/plan.md @@ -0,0 +1,198 @@ +# Plan: Add PostgreSQL support and migrate existing SQLite databases + +## Context + +`meshcore-hub` currently runs on SQLite (`sqlite:///{DATA_HOME}/collector/meshcore.db`). +SQLite WAL does not work over network filesystems and limits concurrent writers, so it +caps the project at a single host — the README already flags switching to Postgres for +multi-host scaling. The goal is to (1) make the codebase genuinely Postgres-compatible, +(2) add a Postgres container and component-based connection config, and (3) give existing +community operators a one-command path to migrate their live SQLite data into Postgres +(downtime is acceptable). + +The stack is already mostly ready: SQLAlchemy 2.0 + Alembic, `asyncpg`/`psycopg2-binary` +declared as the `[postgres]` optional dependency in `pyproject.toml`, and `DATABASE_URL` +threaded through `config.py` and `alembic/env.py`. The work is closing the SQLite-specific +gaps and adding the container + migration tooling. + +**Decisions made:** data migration uses a **SQLAlchemy ORM copy script** (type-safe, no +extra system dependency for operators), and connection config uses **component env vars +assembled into a URL** (with explicit `DATABASE_URL` still taking precedence). pgloader is +*not* used — see "Why not pgloader" below. + +### Why not pgloader +pgloader would infer the target schema from SQLite's *dynamic* typing and produce wrong +Postgres types: `is_observer` (stored `0/1`) → `bigint` not `boolean`; `decoded` JSON +(stored as `TEXT`) → `text` not `json`; `DateTime(timezone=True)` values (stored as text) +→ no `timestamptz`; `String(64)` length constraints lost; and no `alembic_version` +consistent with our migration history. The ORM copy script reuses the existing models, so +SQLAlchemy performs every type conversion correctly and the schema is created by +`alembic upgrade head`. + +--- + +## Part A — Make the code Postgres-compatible (required regardless of migration tool) + +These are real runtime bugs on Postgres, not cosmetics. + +1. **Dialect-aware upsert** — `src/meshcore_hub/common/models/event_observer.py:17,125-139` + `add_event_observer()` is live collector code and currently uses + `from sqlalchemy.dialects.sqlite import insert as sqlite_insert` + + `.on_conflict_do_nothing(...)`. On Postgres this emits invalid SQL. + Fix: pick the insert construct by bind dialect, e.g. + ```python + if session.bind.dialect.name == "postgresql": + from sqlalchemy.dialects.postgresql import insert as pg_insert + stmt = pg_insert(EventObserver).values(...).on_conflict_do_nothing( + index_elements=["event_hash", "observer_node_id"]) + else: + from sqlalchemy.dialects.sqlite import insert as sqlite_insert + stmt = sqlite_insert(EventObserver).values(...).on_conflict_do_nothing(...) + ``` + Both dialects expose the same `.on_conflict_do_nothing(index_elements=...)` API, so only + the import/constructor differs. Grep for other `dialects.sqlite import insert` usages. + +2. **Async driver mapping** — `src/meshcore_hub/common/database.py:145` + `_ensure_async_engine()` only rewrites `sqlite://` → `sqlite+aiosqlite://`. A + `postgresql://` URL keeps the sync `psycopg2` driver and async API sessions fail. + Fix: map `postgresql://` / `postgres://` → `postgresql+asyncpg://` (leave an already + `+driver`-qualified URL untouched). Add a small helper (e.g. `_to_async_url(url)`) used + here. + +3. **Generic JSON type** — 4 models import `from sqlalchemy.dialects.sqlite import JSON`: + `models/raw_packet.py:7`, `models/telemetry.py`, `models/trace_path.py`, + `models/event_log.py`. Switch to generic `from sqlalchemy import JSON`. Generic `JSON` + maps to SQLite JSON and Postgres `JSON` automatically. (Optional: use + `postgresql.JSONB` via `.with_variant()` for indexability — not required for parity.) + +4. **Conditional batch migrations** — `alembic/env.py:61,87` + `render_as_batch=True` is unconditional (it's a SQLite ALTER-TABLE workaround). Make it + `render_as_batch = get_database_url().startswith("sqlite")` in both + `run_migrations_offline()` and `run_migrations_online()`. Existing migrations that call + `op.batch_alter_table(...)` still run correctly on Postgres (Alembic emits direct + `ALTER` there), and a fresh Postgres DB runs the whole history from scratch. + +> The SQLite `PRAGMA` block in `database.py:52-65,150-161` is already guarded by +> `startswith("sqlite")` — no change needed. + +**Verification for Part A:** run the existing test suite against Postgres (see Part E). + +--- + +## Part B — Component-based connection config + +Centralize config in `src/meshcore_hub/common/config.py`. `CollectorSettings` and +`APISettings` currently each carry `database_url` + a duplicated `effective_database_url` +property (`config.py:72-75,174-182` and the matching block in `APISettings`). + +- Add component fields to **`CommonSettings`** (so both inherit): `database_host`, + `database_port` (default `5432`), `database_name`, `database_user`, `database_password`. + (`DATABASE_NAME` is the user's "DATABASE_SCHEMA".) +- Add a shared resolution helper (method on `CommonSettings`, or module function) used by + both `effective_database_url` properties, with this precedence: + 1. explicit `database_url` if set → use verbatim (keeps existing SQLite/`DATABASE_URL` + deployments working unchanged); + 2. else if `database_host` is set → assemble + `postgresql+psycopg2://{user}:{password}@{host}:{port}/{name}` + (URL-encode the password); + 3. else → existing SQLite default under `DATA_HOME`. +- Collapse the duplicated `effective_database_url` into the shared helper. + +Update `.env.example` with the new `DATABASE_HOST` / `DATABASE_PORT` / `DATABASE_NAME` / +`DATABASE_USER` / `DATABASE_PASSWORD` block, documented as "set these for Postgres; leave +unset for default SQLite." + +--- + +## Part C — Postgres container + +In `docker-compose.yml` (mirror the existing `redis` service style, named volume pattern +at lines ~411-419): + +- Add a `postgres` service (`postgres:17-alpine`), env from `POSTGRES_USER`/ + `POSTGRES_PASSWORD`/`POSTGRES_DB` (sourced from the same `.env` values), a named + `postgres_data:/var/lib/postgresql/data` volume, and a `pg_isready` healthcheck. + Put it behind a `postgres` profile (consistent with optional services) or core depending + on whether Postgres becomes the default — recommend keeping SQLite the zero-config + default and Postgres opt-in via profile + env. +- Add `DATABASE_*` to the env passed to `migrate`, `collector`, and `api` services. +- Make `migrate` (and therefore `collector`/`api`) `depends_on` postgres `service_healthy` + when the Postgres profile is active. +- Add the `postgres_data` named volume to the `volumes:` block. +- Mirror into `docker-compose.prod.yml` networking if Postgres should sit on `proxy-net` + (usually it should stay internal-only — confirm during implementation). + +--- + +## Part D — Data migration command (`meshcore-hub db migrate-to-postgres`) + +Add a new Click command in the `db` group in `src/meshcore_hub/__main__.py` (after +`db_upgrade`, ~line 86), backed by a helper module (e.g. +`src/meshcore_hub/common/db_migrate.py`). + +Operator flow (downtime acceptable): +1. Stop `collector`/`api` (writers). +2. Bring up the `postgres` container. +3. Run `meshcore-hub db upgrade` with the Postgres `DATABASE_URL` → creates schema + + stamps `alembic_version`. +4. Run `meshcore-hub db migrate-to-postgres --source sqlite:///...meshcore.db --target `. + +The command: +- Opens a source `DatabaseManager` (SQLite) and target `DatabaseManager` (Postgres) using + the existing `create_database_engine` (`database.py:14`). +- Verifies the target schema is at `head` and tables are empty (refuse otherwise unless + `--truncate`). +- Copies every table in **FK-dependency order** (`nodes` → `node_tags`, + `user_profiles` → `user_profile_nodes`, `event_observers`, then event tables + `messages`/`advertisements`/`telemetry`/`trace_paths`/`raw_packets`, `channels`, + `events_log`). Derive order from `Base.metadata.sorted_tables` to avoid hardcoding. +- Reads source rows via the ORM models / `select()` and bulk-inserts into the target in + batches (e.g. 1–5k rows), reusing the model classes so SQLAlchemy converts + bool/JSON/`timestamptz`/UUID-strings correctly. Use a single target transaction per + table (or per batch) and report per-table counts. +- After load, reconcile Postgres — no sequences to fix (all PKs are app-generated UUID + strings), but log a row-count comparison source-vs-target per table as a built-in check. +- `--dry-run` prints the per-table row counts without writing. + +> Reuse `Base.metadata.sorted_tables` and the existing models in +> `src/meshcore_hub/common/models/` — do not redefine schema in the script. + +--- + +## Part E — Verification + +1. **Unit/integration tests on Postgres.** Spin up a throwaway Postgres + (`docker run --rm -e POSTGRES_PASSWORD=test -p 55432:5432 postgres:17-alpine`), export + the matching `DATABASE_*`, run `meshcore-hub db upgrade`, then the existing test suite + pointed at Postgres. Confirms Part A (esp. the `event_observer` upsert and async API + sessions) and the migration chain build cleanly on Postgres. +2. **Round-trip data migration.** Use the real dev DB at + `data/collector/meshcore.db` (or `backup/meshcore.db`) as source. Run `db upgrade` + + `db migrate-to-postgres`, then assert per-table row counts match (the command's built-in + reconciliation), and spot-check a `raw_packets.decoded` JSON value, an `is_observer` + boolean, and a `received_at` timestamp survived with correct types. +3. **End-to-end app run.** Bring up the stack with the `postgres` profile + `DATABASE_*` + set, confirm `migrate` completes, `collector` ingests an event (exercises + `add_event_observer` upsert on Postgres), and the `api`/`web` serve data + (`meshcore-hub api health`). + +--- + +## Critical files + +| File | Change | +|------|--------| +| `src/meshcore_hub/common/models/event_observer.py` | Dialect-aware upsert (A1) | +| `src/meshcore_hub/common/database.py` | Async driver mapping for Postgres (A2) | +| `src/meshcore_hub/common/models/{raw_packet,telemetry,trace_path,event_log}.py` | Generic `JSON` import (A3) | +| `alembic/env.py` | Conditional `render_as_batch` (A4) | +| `src/meshcore_hub/common/config.py` | Component DATABASE_* vars + shared URL assembly (B) | +| `.env.example` | Document new DATABASE_* vars (B) | +| `docker-compose.yml` (+ `.prod.yml`) | `postgres` service, volume, depends_on, env (C) | +| `src/meshcore_hub/__main__.py` + new `common/db_migrate.py` | `db migrate-to-postgres` command (D) | +| `README.md` / `docs/upgrading.md` | Document Postgres setup + migration procedure | + +## Out of scope / notes +- Keep SQLite as the zero-config default; Postgres is opt-in. No forced migration. +- No schema redesign — all column types already map cleanly once A1–A4 land. +- Consider gating CI to run the suite against both SQLite and Postgres (follow-up). From 57488239c9b6e3a569234356e3c5acb3105e9cc9 Mon Sep 17 00:00:00 2001 From: Louis King Date: Sat, 13 Jun 2026 21:48:25 +0100 Subject: [PATCH 02/13] Expand Postgres plan: backend switch, schema isolation, runbook, phasing Records the DATABASE_BACKEND explicit switch, schema-per-instance isolation via search_path for shared-cluster prod/stg, the no-admin-credentials provisioning decision, the migrate-to-postgres command internals, the operator runbook, and the phased implementation order with SQLite gates. Co-Authored-By: Claude Opus 4.8 --- .../20260613-2111-postgres-migration/plan.md | 285 +++++++++++++++--- 1 file changed, 246 insertions(+), 39 deletions(-) diff --git a/docs/plans/20260613-2111-postgres-migration/plan.md b/docs/plans/20260613-2111-postgres-migration/plan.md index b6e5832..1c184f3 100644 --- a/docs/plans/20260613-2111-postgres-migration/plan.md +++ b/docs/plans/20260613-2111-postgres-migration/plan.md @@ -16,10 +16,29 @@ threaded through `config.py` and `alembic/env.py`. The work is closing the SQLit gaps and adding the container + migration tooling. **Decisions made:** data migration uses a **SQLAlchemy ORM copy script** (type-safe, no -extra system dependency for operators), and connection config uses **component env vars -assembled into a URL** (with explicit `DATABASE_URL` still taking precedence). pgloader is +extra system dependency for operators); connection config uses **component env vars +assembled into a URL**; and the backend is selected by an **explicit `DATABASE_BACKEND` +switch** (`sqlite` default | `postgres`) that fails fast when set to `postgres` without the +required component vars — rather than silently falling back to SQLite. An explicit +`DATABASE_URL`, if set, still overrides everything (managed/external PG, tests). pgloader is *not* used — see "Why not pgloader" below. +SQLite remains the **zero-config default**: a deployment with no DB env vars behaves exactly +as today. Postgres is opt-in and requires agreement across **two layers** — the app +(`DATABASE_BACKEND=postgres` + component vars) and the infra (the compose `postgres` +profile must be activated). The explicit switch makes a half-configured state fail loudly at +startup instead of silently using the wrong database. + +Two further decisions (detailed in Parts B and C): +- **Multi-instance isolation via Postgres schemas.** Production runs several Hub instances + (prod, stg, …) against one shared Postgres; each is scoped to its own schema + (`DATABASE_SCHEMA`, default `meshcorehub`) via `search_path`, with its own + `alembic_version` for independent migration state. Defaults for database/schema/role are + all `meshcorehub`. +- **No admin/bootstrap credentials.** The app never holds cluster-admin creds; provisioning + is automatic on the bundled container (image entrypoint) and out-of-band for managed + Postgres (IaC/console/DBA). No `POSTGRES_ADMIN_*`. + ### Why not pgloader pgloader would infer the target schema from SQLite's *dynamic* typing and produce wrong Postgres types: `is_observer` (stored `0/1`) → `bigint` not `boolean`; `decoded` JSON @@ -31,6 +50,37 @@ SQLAlchemy performs every type conversion correctly and the schema is created by --- +## Implementation order + +Parts map to phases. The throughline: **only Phases 1–2 touch shared code and can regress +SQLite, so each ends in a SQLite gate; Phases 3–5 are additive and Postgres-only.** Reach +"Postgres-ready code + verified SQLite" at the end of Phase 2 before committing to the +heavier container/migration work. + +- **Phase 1 — Code compatibility (Part A).** The four dialect-neutral fixes (upsert, async + URL mapping, generic `JSON`, conditional `render_as_batch`). + - **Gate 1 (SQLite):** full existing test suite green on SQLite + fresh `db upgrade` from + scratch on a new SQLite file. These changes are behaviour-neutral, so any breakage is + isolated to these four edits — easy to bisect. +- **Phase 2 — Backend switch + config (Part B).** `DATABASE_BACKEND`, the `DATABASE_*` + vars, `search_path` wiring, `version_table_schema`. + - **Gate 2 (SQLite — the key checkpoint):** re-run full suite; add a test asserting the + **default no-env path resolves to the exact same SQLite URL/behaviour as before** (no + schema/search_path logic engaged when `DATABASE_BACKEND` unset). *Milestone: code is + Postgres-compatible and SQLite is proven untouched.* +- **Phase 3 — Postgres container (Part C).** Service, `DATABASE_* → POSTGRES_*` derivation, + `initdb.d` schema script, profile, healthcheck. Pure infra; cannot regress SQLite. + - **Gate 3 (Postgres):** bring container up, `db upgrade` against it (tables land in the + schema, `alembic_version` stamped), run the existing suite against Postgres. Wire a + **SQLite + Postgres test matrix** here so both run going forward. +- **Phase 4 — Data migration command (Part D).** New Postgres-only code path; zero SQLite risk. + - **Gate 4:** round-trip the real dev DB (`data/collector/meshcore.db`) → Postgres, + reconcile per-table row counts, spot-check a JSON/bool/timestamp value. +- **Phase 5 — Verification + docs (Part E).** End-to-end stack run on the `postgres` + profile; `docs/upgrading.md` runbook + `search_path`/provisioning docs. + +--- + ## Part A — Make the code Postgres-compatible (required regardless of migration tool) These are real runtime bugs on Postgres, not cosmetics. @@ -85,22 +135,55 @@ Centralize config in `src/meshcore_hub/common/config.py`. `CollectorSettings` an `APISettings` currently each carry `database_url` + a duplicated `effective_database_url` property (`config.py:72-75,174-182` and the matching block in `APISettings`). -- Add component fields to **`CommonSettings`** (so both inherit): `database_host`, - `database_port` (default `5432`), `database_name`, `database_user`, `database_password`. - (`DATABASE_NAME` is the user's "DATABASE_SCHEMA".) +- Add a `database_backend` field to **`CommonSettings`** (env `DATABASE_BACKEND`, + default `"sqlite"`, validated choice `sqlite | postgres`). +- Add component fields to **`CommonSettings`** (so both inherit), all defaulting to + `"meshcorehub"` where named: `database_host`, `database_port` (default `5432`), + `database_name` (the **database**, default `"meshcorehub"`), `database_schema` (the + **Postgres schema/namespace**, default `"meshcorehub"`), `database_user` + (default `"meshcorehub"`), `database_password`. + - `DATABASE_NAME` and `DATABASE_SCHEMA` are **distinct**: the former is the database, the + latter is the namespace within it. They differ in the shared-cluster case (see + "Multi-instance isolation" below); for a single instance both are `meshcorehub`. - Add a shared resolution helper (method on `CommonSettings`, or module function) used by both `effective_database_url` properties, with this precedence: - 1. explicit `database_url` if set → use verbatim (keeps existing SQLite/`DATABASE_URL` - deployments working unchanged); - 2. else if `database_host` is set → assemble - `postgresql+psycopg2://{user}:{password}@{host}:{port}/{name}` - (URL-encode the password); - 3. else → existing SQLite default under `DATA_HOME`. + 1. explicit `database_url` if set → use verbatim (escape hatch: managed/external PG, tests); + 2. else if `database_backend == "postgres"` → require `database_host`, `database_name`, + `database_user`, `database_password` (raise a clear startup error naming any missing + var — do **not** fall back to SQLite), then assemble + `postgresql+psycopg2://{user}:{password}@{host}:{port}/{name}` (URL-encode the password); + 3. else (`database_backend == "sqlite"`) → existing SQLite default under `DATA_HOME`. - Collapse the duplicated `effective_database_url` into the shared helper. -Update `.env.example` with the new `DATABASE_HOST` / `DATABASE_PORT` / `DATABASE_NAME` / -`DATABASE_USER` / `DATABASE_PASSWORD` block, documented as "set these for Postgres; leave -unset for default SQLite." +### Multi-instance isolation via Postgres schemas +Production will run **multiple Hub instances (prod, stg, …) against one shared Postgres**. +Isolate them with **schema-per-instance**, not database-per-instance: a shared database with +each instance scoped to its own schema (`DATABASE_SCHEMA`). This reuses the schema var and is +lighter to provision on a shared/managed cluster than many databases. (Database-per-instance +is the heavier alternative — stronger isolation, but more provisioning; not chosen.) + +Mechanics (Postgres only — SQLite ignores all of this): +- **Scope every connection to the schema via `search_path`** rather than hardcoding + `schema=` on the models/metadata (hardcoding would pin the ORM to one schema and break + SQLite). Set it in `create_database_engine` for Postgres, e.g. psycopg2 + `connect_args={"options": f"-csearch_path={schema}"}` (and the asyncpg equivalent for the + async engine — a `SET search_path` on connect via an event listener). Models stay + schema-agnostic; the active schema is chosen entirely by config. +- **Per-instance migration state**: Alembic must place its bookkeeping in the instance's + schema — pass `version_table_schema=` in `alembic/env.py` (Postgres only) so each + instance has its own `alembic_version` and prod/stg can sit on **different revisions + independently**. `include_schemas=True` for autogenerate. The `search_path` set on the + connection means `upgrade` creates the tables in the right schema automatically. +- **Schema must exist first**: the app role needs its schema present. Either pre-provisioned + out-of-band (managed; see Part C) or `CREATE SCHEMA IF NOT EXISTS` on the bundled + container. The `db migrate-to-postgres` command inherits the same `search_path`, so it + loads into the instance's schema with no extra flags. + +Update `.env.example` with a `DATABASE_BACKEND` line (default `sqlite`) plus the +`DATABASE_HOST` / `DATABASE_PORT` / `DATABASE_NAME` / `DATABASE_SCHEMA` / `DATABASE_USER` / +`DATABASE_PASSWORD` block (defaults `meshcorehub`), documented as "set +`DATABASE_BACKEND=postgres` and fill these in to use Postgres; override `DATABASE_SCHEMA` +per instance (e.g. `prod`, `stg`) when sharing one cluster; leave defaults for SQLite." --- @@ -109,19 +192,66 @@ unset for default SQLite." In `docker-compose.yml` (mirror the existing `redis` service style, named volume pattern at lines ~411-419): -- Add a `postgres` service (`postgres:17-alpine`), env from `POSTGRES_USER`/ - `POSTGRES_PASSWORD`/`POSTGRES_DB` (sourced from the same `.env` values), a named +- Add a `postgres` service (`postgres:17-alpine`), a named `postgres_data:/var/lib/postgresql/data` volume, and a `pg_isready` healthcheck. - Put it behind a `postgres` profile (consistent with optional services) or core depending - on whether Postgres becomes the default — recommend keeping SQLite the zero-config - default and Postgres opt-in via profile + env. -- Add `DATABASE_*` to the env passed to `migrate`, `collector`, and `api` services. + Put it behind a `postgres` compose profile (SQLite stays the zero-config default; the + container only runs when the profile is activated, e.g. `docker compose --profile postgres up`). +- **Single source of truth for credentials** — derive the container's init vars from the + app's `DATABASE_*` rather than maintaining a duplicate set: + ```yaml + environment: + POSTGRES_USER: ${DATABASE_USER:-meshcorehub} + POSTGRES_PASSWORD: ${DATABASE_PASSWORD} + POSTGRES_DB: ${DATABASE_NAME:-meshcorehub} + ``` + The image entrypoint auto-creates the role + database from these on first init (empty + volume only). Add a tiny `/docker-entrypoint-initdb.d/` script to also + `CREATE SCHEMA IF NOT EXISTS "${DATABASE_SCHEMA}" AUTHORIZATION "${DATABASE_USER}"` so the + bundled path needs no manual SQL. +- Add `DATABASE_*` (including `DATABASE_BACKEND`) to the env passed to `migrate`, + `collector`, and `api` services. +- **Document the two-layer requirement explicitly**: enabling Postgres = set + `DATABASE_BACKEND=postgres` + component vars **and** activate the `postgres` profile. + Compose can't read the app env var to auto-activate its profile, so these stay two + switches — but the app's fail-fast validation (Part B) turns a half-configured state into + a clear startup error rather than a silent SQLite fallback. - Make `migrate` (and therefore `collector`/`api`) `depends_on` postgres `service_healthy` when the Postgres profile is active. - Add the `postgres_data` named volume to the `volumes:` block. - Mirror into `docker-compose.prod.yml` networking if Postgres should sit on `proxy-net` (usually it should stay internal-only — confirm during implementation). +### Provisioning decision: no admin/bootstrap credentials +The app and tooling **never hold cluster-admin credentials**. The only privileged +operations are the one-time `CREATE ROLE` / `CREATE DATABASE` / `CREATE SCHEMA` / `GRANT`; +everything afterwards (Alembic `db upgrade` DDL, runtime DML, the data-migration command) +runs as the least-privileged app role, which only needs to **own its schema/database**, not +superuser. Provisioning is therefore split by environment, and we do **not** add +`POSTGRES_ADMIN_USER/PASSWORD` (handing the app permanent admin creds for a one-time task is +a security regression and not the production pattern): + +| Environment | Provisioning of role/db/schema/grants | App holds | +|-------------|----------------------------------------|-----------| +| **Bundled container** (dev / small prod) | Automatic: image entrypoint creates role+db from `POSTGRES_*` (= `DATABASE_*`), init script creates the schema. No admin creds exist or are needed. | `DATABASE_*` | +| **Managed / shared Postgres** (prod, stg) | **Out-of-band** — Terraform/IaC, cloud console, or a DBA runs the SQL once. The platform's master user stays in the infra layer, never in the app. | `DATABASE_*` only | + +Managed/shared-cluster provisioning SQL to document in `docs/` (one schema per instance in a +shared database; role must own its schema so `db upgrade` can create tables; **need not be +superuser** — which is why the migration command's `session_replication_role` step has a +documented fallback, Part D): +```sql +-- once per cluster +CREATE ROLE meshcorehub LOGIN PASSWORD '…'; +CREATE DATABASE meshcorehub OWNER meshcorehub; +-- connected to the meshcorehub database, once per Hub instance: +CREATE SCHEMA IF NOT EXISTS prod AUTHORIZATION meshcorehub; -- and stg, etc. +GRANT ALL ON SCHEMA prod TO meshcorehub; +``` +Each instance then sets `DATABASE_SCHEMA=prod` / `stg`. A Terraform example is a nice-to-have +follow-up, not required for v1. (Optional future convenience: a `meshcore-hub db bootstrap` +command that accepts admin creds **only as one-shot CLI flags**, never persisted env — out of +scope for v1.) + --- ## Part D — Data migration command (`meshcore-hub db migrate-to-postgres`) @@ -130,33 +260,110 @@ Add a new Click command in the `db` group in `src/meshcore_hub/__main__.py` (aft `db_upgrade`, ~line 86), backed by a helper module (e.g. `src/meshcore_hub/common/db_migrate.py`). -Operator flow (downtime acceptable): -1. Stop `collector`/`api` (writers). -2. Bring up the `postgres` container. -3. Run `meshcore-hub db upgrade` with the Postgres `DATABASE_URL` → creates schema + - stamps `alembic_version`. -4. Run `meshcore-hub db migrate-to-postgres --source sqlite:///...meshcore.db --target `. - -The command: +### Command behaviour - Opens a source `DatabaseManager` (SQLite) and target `DatabaseManager` (Postgres) using the existing `create_database_engine` (`database.py:14`). +- **Sensible defaults make the common case zero-flag:** `--source` defaults to the legacy + SQLite path under `DATA_HOME` (`sqlite:////data/collector/meshcore.db`) regardless of the + configured backend; `--target` defaults to the configured `effective_database_url` + (Postgres). Both can be overridden explicitly. - Verifies the target schema is at `head` and tables are empty (refuse otherwise unless - `--truncate`). -- Copies every table in **FK-dependency order** (`nodes` → `node_tags`, - `user_profiles` → `user_profile_nodes`, `event_observers`, then event tables - `messages`/`advertisements`/`telemetry`/`trace_paths`/`raw_packets`, `channels`, - `events_log`). Derive order from `Base.metadata.sorted_tables` to avoid hardcoding. -- Reads source rows via the ORM models / `select()` and bulk-inserts into the target in - batches (e.g. 1–5k rows), reusing the model classes so SQLAlchemy converts - bool/JSON/`timestamptz`/UUID-strings correctly. Use a single target transaction per - table (or per batch) and report per-table counts. -- After load, reconcile Postgres — no sequences to fix (all PKs are app-generated UUID - strings), but log a row-count comparison source-vs-target per table as a built-in check. + `--truncate`) — guards against accidental double-runs. **Non-destructive to the source.** +- **Copies at the SQLAlchemy Core level, table-by-table** — *not* through ORM objects. + Iterate `Base.metadata.sorted_tables` and for each `Table`, `select(table)` from SQLite + and `insert(table)` into Postgres reusing the same `Table` object. Type conversion is a + free side-effect of the dialects' type processors: reading applies the SQLite *result + processor* per column (`0/1`→`bool`, JSON `TEXT`→`dict`, ISO string→`datetime`), writing + applies the Postgres *bind processor* (`bool`→`boolean`, `dict`→`json`, + `datetime`→`timestamptz`, UUID-string→`varchar`). No per-model code, works generically + for every table. (Core, not ORM, because instances can't be shared across two sessions.) +- `Base.metadata.sorted_tables` gives a **parent-first FK order** (`nodes` → `node_tags`, + `user_profiles` → `user_profile_nodes`, `event_observers`, event tables, `channels`, + `events_log`) and naturally **excludes `alembic_version`** (not a mapped model — it was + already created/stamped by `db upgrade`, so we must not touch it). +- **Streams large tables**: `select(...).execution_options(stream_results=True)` + + `result.partitions(batch_size)` (~2k) so `raw_packets` doesn't materialize in memory; + inserts go out via the executemany path. +- **Timezone normalization (the one explicit conversion):** SQLite doesn't persist tz, so + `DateTime(timezone=True)` values read back as *naive* datetimes; inserting naive values + into Postgres `timestamptz` would assume the session tz. Since the app always writes UTC + (`utc_now()`), a small normalize step attaches `tzinfo=UTC` to naive datetimes for + tz-aware columns before insert. +- **Single target transaction** wrapping the whole copy → all-or-nothing; any failure + leaves Postgres empty and the operator re-runs. Acceptable given the downtime window. +- **FK enforcement during load** — `sorted_tables` order is normally sufficient (the schema + has no FK cycles). For robustness the load issues `SET session_replication_role = replica` + to disable FK triggers, then restores `DEFAULT`. **Caveat: this requires DB-superuser + privilege**, which the bundled compose `postgres` container has but a *managed* Postgres + (RDS / Cloud SQL) may not grant. Fallback for managed targets: skip the + `session_replication_role` toggle and rely solely on `sorted_tables` order (safe here as + there are no FK cycles). The command should detect the missing privilege and fall back + gracefully (or expose a `--no-replication-role` flag), and the docs should call this out. +- After load, reconcile — no sequences to fix (all PKs are app-generated UUID strings), but + log a **per-table source-vs-target row-count comparison** as a built-in check. - `--dry-run` prints the per-table row counts without writing. > Reuse `Base.metadata.sorted_tables` and the existing models in > `src/meshcore_hub/common/models/` — do not redefine schema in the script. +> **Why run it inside the `migrate` service:** the command needs both databases reachable +> at once — the `migrate` service mounts the `data` volume (SQLite source) *and* sits on the +> compose network with `postgres` (target). Hence `docker compose run --rm migrate ...` +> rather than a bare host invocation. + +### Operator runbook (docker-compose deployment, downtime acceptable) + +Starting state: running on SQLite, data in the `data` volume at `/data/collector/meshcore.db`. + +1. **Back up the SQLite database** — this is the rollback path: + ```bash + docker compose cp collector:/data/collector/meshcore.db ./meshcore-backup-$(date +%F).db + ``` +2. **Stop the writers** (downtime starts; quiesces the SQLite file): + ```bash + docker compose stop collector api web + ``` +3. **Configure Postgres in `.env`** — only the `DATABASE_*` block (the bundled container's + `POSTGRES_*` init vars derive from these, see Part C): + ```ini + DATABASE_BACKEND=postgres + DATABASE_HOST=postgres + DATABASE_PORT=5432 + DATABASE_NAME=meshcorehub + DATABASE_SCHEMA=meshcorehub # override per instance (prod/stg) on a shared cluster + DATABASE_USER=meshcorehub + DATABASE_PASSWORD= # e.g. openssl rand -base64 32 + ``` +4. **Start the Postgres container** and wait for its `pg_isready` healthcheck: + ```bash + docker compose --profile postgres up -d postgres + ``` +5. **Create the schema in Postgres** (builds correctly-typed tables + stamps `alembic_version`): + ```bash + docker compose --profile postgres run --rm migrate meshcore-hub db upgrade + ``` +6. **Copy the data across** (zero-flag common case; optionally `--dry-run` first): + ```bash + docker compose --profile postgres run --rm migrate \ + meshcore-hub db migrate-to-postgres + ``` + Confirm the per-table reconciliation counts all match. +7. **Bring the stack up against Postgres and verify**: + ```bash + docker compose --profile postgres up -d + docker compose run --rm api meshcore-hub api health + ``` + Spot-check the web dashboard shows nodes/events. Downtime ends here. +8. **Decommission SQLite (later)** — once confident (a few days), remove the old + `meshcore.db` from the `data` volume; keep the step-1 backup archived. + +**Rollback:** stop the stack, set `DATABASE_BACKEND=sqlite` (or remove it) in `.env`, and +`docker compose up -d` without the `postgres` profile. You're back on the untouched SQLite +file — the migration never mutates the source. + +> The implementation should mirror this runbook into `docs/upgrading.md` (Part D +> deliverable). + --- ## Part E — Verification From 9eab07d244f9f08d3a82cef57f26811266eb15ec Mon Sep 17 00:00:00 2001 From: Louis King Date: Sat, 13 Jun 2026 21:48:49 +0100 Subject: [PATCH 03/13] Phase 1: make ORM/migrations Postgres-compatible Dialect-neutral fixes that keep SQLite behaviour unchanged: - event_observer: choose INSERT construct by bind dialect so the on_conflict upsert works on both SQLite and Postgres - database: map postgresql:// -> postgresql+asyncpg:// for async sessions (was sqlite-only), via a _to_async_url helper - models: import JSON from sqlalchemy (generic) not the sqlite dialect - alembic/env: render_as_batch only for SQLite (its ALTER TABLE workaround) Full SQLite test suite green (1045 passed) and fresh db upgrade builds all tables; black/flake8/mypy clean. Co-Authored-By: Claude Opus 4.8 --- alembic/env.py | 4 +- src/meshcore_hub/common/database.py | 18 ++++++- src/meshcore_hub/common/models/event_log.py | 3 +- .../common/models/event_observer.py | 49 +++++++++++++------ src/meshcore_hub/common/models/raw_packet.py | 3 +- src/meshcore_hub/common/models/telemetry.py | 3 +- src/meshcore_hub/common/models/trace_path.py | 3 +- 7 files changed, 58 insertions(+), 25 deletions(-) diff --git a/alembic/env.py b/alembic/env.py index e37432c..ab12c86 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -58,7 +58,9 @@ def run_migrations_offline() -> None: target_metadata=target_metadata, literal_binds=True, dialect_opts={"paramstyle": "named"}, - render_as_batch=True, # SQLite batch mode for ALTER TABLE + # Batch mode is a SQLite-only workaround for its limited ALTER TABLE; + # Postgres performs ALTERs directly. + render_as_batch=url.startswith("sqlite"), ) with context.begin_transaction(): diff --git a/src/meshcore_hub/common/database.py b/src/meshcore_hub/common/database.py index 4b0427b..90c1b31 100644 --- a/src/meshcore_hub/common/database.py +++ b/src/meshcore_hub/common/database.py @@ -11,6 +11,22 @@ from sqlalchemy.orm import Session, sessionmaker from meshcore_hub.common.models.base import Base +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. + """ + 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"): + return database_url.replace(f"{scheme}://", "postgresql+asyncpg://", 1) + return database_url + + def create_database_engine( database_url: str, echo: bool = False, @@ -142,7 +158,7 @@ class DatabaseManager: from sqlalchemy.ext.asyncio import async_sessionmaker - async_url = self.database_url.replace("sqlite://", "sqlite+aiosqlite://") + async_url = _to_async_url(self.database_url) self._async_engine = create_async_engine(async_url, echo=self._echo) # Apply the same SQLite pragmas as the sync engine (see diff --git a/src/meshcore_hub/common/models/event_log.py b/src/meshcore_hub/common/models/event_log.py index 13e0a11..ccab483 100644 --- a/src/meshcore_hub/common/models/event_log.py +++ b/src/meshcore_hub/common/models/event_log.py @@ -3,8 +3,7 @@ from datetime import datetime from typing import Any, Optional -from sqlalchemy import DateTime, ForeignKey, Index, String -from sqlalchemy.dialects.sqlite import JSON +from sqlalchemy import DateTime, ForeignKey, Index, JSON, String from sqlalchemy.orm import Mapped, mapped_column from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now diff --git a/src/meshcore_hub/common/models/event_observer.py b/src/meshcore_hub/common/models/event_observer.py index a3478cd..40166d3 100644 --- a/src/meshcore_hub/common/models/event_observer.py +++ b/src/meshcore_hub/common/models/event_observer.py @@ -8,13 +8,13 @@ from sqlalchemy import ( DateTime, Float, ForeignKey, + Insert, Integer, Index, String, UniqueConstraint, update, ) -from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.orm import Mapped, Session, mapped_column, relationship from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now @@ -122,21 +122,40 @@ def add_event_observer( now = observed_at or datetime.now(timezone.utc) - stmt = ( - sqlite_insert(EventObserver) - .values( - id=str(uuid4()), - event_type=event_type, - event_hash=event_hash, - observer_node_id=observer_node_id, - snr=snr, - path_len=path_len, - observed_at=now, - created_at=now, - updated_at=now, + # Both SQLite and Postgres expose on_conflict_do_nothing() with the same signature, + # but the INSERT construct must come from the matching dialect or it emits SQL for + # the wrong backend. Build the statement in each branch (rather than aliasing the + # insert() function) so the two dialect-specific Insert types stay distinct. + values = { + "id": str(uuid4()), + "event_type": event_type, + "event_hash": event_hash, + "observer_node_id": observer_node_id, + "snr": snr, + "path_len": path_len, + "observed_at": now, + "created_at": now, + "updated_at": now, + } + conflict_cols = ["event_hash", "observer_node_id"] + stmt: Insert + + if session.get_bind().dialect.name == "postgresql": + from sqlalchemy.dialects.postgresql import insert as pg_insert + + stmt = ( + pg_insert(EventObserver) + .values(**values) + .on_conflict_do_nothing(index_elements=conflict_cols) + ) + else: + from sqlalchemy.dialects.sqlite import insert as sqlite_insert + + stmt = ( + sqlite_insert(EventObserver) + .values(**values) + .on_conflict_do_nothing(index_elements=conflict_cols) ) - .on_conflict_do_nothing(index_elements=["event_hash", "observer_node_id"]) - ) result = session.execute(stmt) rowcount = getattr(result, "rowcount", 0) diff --git a/src/meshcore_hub/common/models/raw_packet.py b/src/meshcore_hub/common/models/raw_packet.py index 66576ab..835c9f7 100644 --- a/src/meshcore_hub/common/models/raw_packet.py +++ b/src/meshcore_hub/common/models/raw_packet.py @@ -3,8 +3,7 @@ from datetime import datetime from typing import Any, Optional -from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String, Text -from sqlalchemy.dialects.sqlite import JSON +from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, JSON, String, Text from sqlalchemy.orm import Mapped, mapped_column from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now diff --git a/src/meshcore_hub/common/models/telemetry.py b/src/meshcore_hub/common/models/telemetry.py index abca8ac..db95332 100644 --- a/src/meshcore_hub/common/models/telemetry.py +++ b/src/meshcore_hub/common/models/telemetry.py @@ -3,8 +3,7 @@ from datetime import datetime from typing import Any, Optional -from sqlalchemy import DateTime, ForeignKey, Index, LargeBinary, String -from sqlalchemy.dialects.sqlite import JSON +from sqlalchemy import DateTime, ForeignKey, Index, JSON, LargeBinary, String from sqlalchemy.orm import Mapped, mapped_column from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now diff --git a/src/meshcore_hub/common/models/trace_path.py b/src/meshcore_hub/common/models/trace_path.py index df0d48b..31a65da 100644 --- a/src/meshcore_hub/common/models/trace_path.py +++ b/src/meshcore_hub/common/models/trace_path.py @@ -3,8 +3,7 @@ from datetime import datetime from typing import Optional -from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, String -from sqlalchemy.dialects.sqlite import JSON +from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, JSON, String from sqlalchemy.orm import Mapped, mapped_column from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now From f342f5bc7041334b5d1529b366b6fad21c76718e Mon Sep 17 00:00:00 2001 From: Louis King Date: Sat, 13 Jun 2026 22:01:00 +0100 Subject: [PATCH 04/13] Phase 2: DATABASE_BACKEND switch, component config, schema scoping - config: add DatabaseBackend enum + DATABASE_* component vars on CommonSettings (host/port/name/schema/user/password, defaults 'meshcorehub'); single effective_database_url resolver with precedence DATABASE_URL > postgres (assembled, fail-fast on missing vars) > SQLite default. effective_database_schema returns the schema only for Postgres. Collapses the duplicated resolver/field out of Collector/API settings. - database: create_database_engine/DatabaseManager accept a schema arg and scope Postgres connections via search_path (psycopg2 -c options; asyncpg server_settings). No-op for SQLite. - alembic/env: migrate into the instance schema (version_table_schema + CREATE SCHEMA + search_path) for Postgres; fix online render_as_batch to be SQLite-only (matching the offline path). - .env.example: document the DATABASE_* block. SQLite behaviour unchanged: default no-env path resolves to the same URL (new regression tests), full suite green (1051 passed), fresh db upgrade OK. Co-Authored-By: Claude Opus 4.8 --- .env.example | 23 +++++ alembic/env.py | 33 ++++++- src/meshcore_hub/common/config.py | 137 +++++++++++++++++++--------- src/meshcore_hub/common/database.py | 29 +++++- tests/test_common/test_config.py | 70 ++++++++++++++ 5 files changed, 240 insertions(+), 52 deletions(-) diff --git a/.env.example b/.env.example index 194e6f7..dfaab4a 100644 --- a/.env.example +++ b/.env.example @@ -66,6 +66,29 @@ LOG_LEVEL=INFO # └── meshcore.db # SQLite database DATA_HOME=./data +# ----------------------------------------------------------------------------- +# Database +# ----------------------------------------------------------------------------- +# SQLite is the zero-config default and needs nothing here — it lives at +# ${DATA_HOME}/collector/meshcore.db. +# +# To use PostgreSQL instead, set DATABASE_BACKEND=postgres and fill in the +# DATABASE_* values below (the bundled postgres container derives its +# POSTGRES_USER/PASSWORD/DB from DATABASE_USER/PASSWORD/NAME). You must also +# activate the compose 'postgres' profile, e.g. `docker compose --profile postgres up`. +# +# DATABASE_BACKEND=postgres +# DATABASE_HOST=postgres +# DATABASE_PORT=5432 +# DATABASE_NAME=meshcorehub +# DATABASE_SCHEMA=meshcorehub # override per instance (e.g. prod, stg) on a shared cluster +# DATABASE_USER=meshcorehub +# DATABASE_PASSWORD= # required for postgres; e.g. `openssl rand -base64 32` +# +# Advanced: set DATABASE_URL to a full SQLAlchemy URL to override all of the above +# (e.g. a managed/external Postgres). Takes precedence over DATABASE_BACKEND. +# DATABASE_URL=postgresql+psycopg2://user:pass@host:5432/dbname + # Directory containing seed data files for import # Default: ./seed (relative to docker-compose.yml location) # Inside containers this is mapped to /seed diff --git a/alembic/env.py b/alembic/env.py index ab12c86..f169e14 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -4,7 +4,7 @@ import os from logging.config import fileConfig from alembic import context -from sqlalchemy import engine_from_config, pool +from sqlalchemy import engine_from_config, pool, text from meshcore_hub.common.models import Base @@ -41,6 +41,18 @@ def get_database_url() -> str: return config.get_main_option("sqlalchemy.url", "sqlite:///./meshcore.db") +def get_schema(url: str) -> str | None: + """Postgres schema to migrate into, or None for SQLite. + + Each Hub instance keeps its tables and alembic_version in its own schema so + multiple instances (prod, stg, ...) can share one Postgres database with + independent migration state. + """ + if url.startswith(("postgresql", "postgres")): + return os.environ.get("DATABASE_SCHEMA", "meshcorehub") + return None + + def run_migrations_offline() -> None: """Run migrations in 'offline' mode. @@ -53,6 +65,7 @@ def run_migrations_offline() -> None: script output. """ url = get_database_url() + schema = get_schema(url) context.configure( url=url, target_metadata=target_metadata, @@ -61,6 +74,8 @@ def run_migrations_offline() -> None: # Batch mode is a SQLite-only workaround for its limited ALTER TABLE; # Postgres performs ALTERs directly. render_as_batch=url.startswith("sqlite"), + version_table_schema=schema, + include_schemas=schema is not None, ) with context.begin_transaction(): @@ -74,7 +89,9 @@ def run_migrations_online() -> None: and associate a connection with the context. """ configuration = config.get_section(config.config_ini_section, {}) - configuration["sqlalchemy.url"] = get_database_url() + url = get_database_url() + configuration["sqlalchemy.url"] = url + schema = get_schema(url) connectable = engine_from_config( configuration, @@ -83,10 +100,20 @@ def run_migrations_online() -> None: ) with connectable.connect() as connection: + # Ensure the instance's schema exists and scope this connection to it so + # tables (and alembic_version) are created there. No-op for SQLite. + if schema is not None: + connection.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{schema}"')) + connection.execute(text(f'SET search_path TO "{schema}"')) + connection.commit() + context.configure( connection=connection, target_metadata=target_metadata, - render_as_batch=True, # SQLite batch mode for ALTER TABLE + # Batch mode is a SQLite-only workaround for its limited ALTER TABLE. + render_as_batch=url.startswith("sqlite"), + version_table_schema=schema, + include_schemas=schema is not None, ) with context.begin_transaction(): diff --git a/src/meshcore_hub/common/config.py b/src/meshcore_hub/common/config.py index 5efb880..416901a 100644 --- a/src/meshcore_hub/common/config.py +++ b/src/meshcore_hub/common/config.py @@ -3,7 +3,7 @@ from enum import Enum from typing import Optional -from pydantic import Field, field_validator +from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict @@ -24,6 +24,13 @@ class MQTTTransport(str, Enum): WEBSOCKETS = "websockets" +class DatabaseBackend(str, Enum): + """Database backend selector.""" + + SQLITE = "sqlite" + POSTGRES = "postgres" + + class CommonSettings(BaseSettings): """Common settings shared by all components.""" @@ -39,6 +46,88 @@ class CommonSettings(BaseSettings): description="Base directory for service data (e.g., ./data or /data)", ) + # Database backend selection and connection components. + # SQLite is the zero-config default; set DATABASE_BACKEND=postgres (plus the + # DATABASE_* component vars) to use Postgres. An explicit DATABASE_URL overrides + # everything (managed/external Postgres, tests). + database_backend: DatabaseBackend = Field( + default=DatabaseBackend.SQLITE, + description="Database backend: 'sqlite' (default) or 'postgres'", + ) + database_url: Optional[str] = Field( + default=None, + description=( + "Explicit SQLAlchemy database URL; overrides DATABASE_BACKEND/component vars. " + "Default: sqlite:///{data_home}/collector/meshcore.db" + ), + ) + database_host: Optional[str] = Field( + default=None, + description="Postgres host (required when DATABASE_BACKEND=postgres)", + ) + database_port: int = Field(default=5432, description="Postgres port") + database_name: str = Field( + default="meshcorehub", description="Postgres database name" + ) + database_schema: str = Field( + default="meshcorehub", + description="Postgres schema (namespace); override per instance on a shared cluster", + ) + database_user: str = Field(default="meshcorehub", description="Postgres role/user") + database_password: Optional[str] = Field( + default=None, + description="Postgres password (required when DATABASE_BACKEND=postgres)", + ) + + @property + def effective_database_url(self) -> str: + """Resolve the SQLAlchemy database URL. + + Precedence: explicit DATABASE_URL > postgres (assembled from components) > + SQLite default under DATA_HOME. Fails fast for a misconfigured postgres backend + rather than silently falling back to SQLite. + """ + if self.database_url: + return self.database_url + if self.database_backend == DatabaseBackend.POSTGRES: + missing = [ + name + for name, value in ( + ("DATABASE_HOST", self.database_host), + ("DATABASE_NAME", self.database_name), + ("DATABASE_USER", self.database_user), + ("DATABASE_PASSWORD", self.database_password), + ) + if not value + ] + if missing: + raise ValueError( + "DATABASE_BACKEND=postgres requires: " + ", ".join(missing) + ) + from urllib.parse import quote_plus + + user = quote_plus(self.database_user) + password = quote_plus(self.database_password or "") + return ( + f"postgresql+psycopg2://{user}:{password}" + f"@{self.database_host}:{self.database_port}/{self.database_name}" + ) + from pathlib import Path + + db_path = Path(self.data_home) / "collector" / "meshcore.db" + return f"sqlite:///{db_path}" + + @property + def effective_database_schema(self) -> Optional[str]: + """Postgres schema to scope connections to, or None for SQLite. + + Returns the schema only when the effective URL is Postgres; SQLite has no + schema concept, so callers leave search_path untouched. + """ + if self.effective_database_url.startswith(("postgresql", "postgres")): + return self.database_schema + return None + # Logging log_level: LogLevel = Field(default=LogLevel.INFO, description="Logging level") @@ -68,11 +157,7 @@ class CommonSettings(BaseSettings): class CollectorSettings(CommonSettings): """Settings for the Collector component.""" - # Database - default uses data_home/collector/meshcore.db - database_url: Optional[str] = Field( - default=None, - description="SQLAlchemy database URL (default: sqlite:///{data_home}/collector/meshcore.db)", - ) + # Database config (backend selector + connection) is inherited from CommonSettings. # Seed home directory - contains initial data files (node_tags.yaml) seed_home: str = Field( @@ -171,16 +256,6 @@ class CollectorSettings(CommonSettings): return str(Path(self.data_home) / "collector") - @property - def effective_database_url(self) -> str: - """Get the effective database URL, using default if not set.""" - if self.database_url: - return self.database_url - from pathlib import Path - - db_path = Path(self.data_home) / "collector" / "meshcore.db" - return f"sqlite:///{db_path}" - @property def effective_seed_home(self) -> str: """Get the effective seed home directory.""" @@ -202,13 +277,6 @@ class CollectorSettings(CommonSettings): return str(Path(self.effective_seed_home) / "channels.yaml") - @field_validator("database_url") - @classmethod - def validate_database_url(cls, v: Optional[str]) -> Optional[str]: - """Validate database URL format.""" - # None is allowed - will use default - return v - class APISettings(CommonSettings): """Settings for the API component.""" @@ -217,11 +285,7 @@ class APISettings(CommonSettings): api_host: str = Field(default="0.0.0.0", description="API server host") api_port: int = Field(default=8000, description="API server port") - # Database - default uses data_home/collector/meshcore.db (same as collector) - database_url: Optional[str] = Field( - default=None, - description="SQLAlchemy database URL (default: sqlite:///{data_home}/collector/meshcore.db)", - ) + # Database config (backend selector + connection) is inherited from CommonSettings. # Authentication api_read_key: Optional[str] = Field(default=None, description="Read-only API key") @@ -252,23 +316,6 @@ class APISettings(CommonSettings): description="Cache TTL for dashboard endpoints (seconds)", ) - @property - def effective_database_url(self) -> str: - """Get the effective database URL, using default if not set.""" - if self.database_url: - return self.database_url - from pathlib import Path - - db_path = Path(self.data_home) / "collector" / "meshcore.db" - return f"sqlite:///{db_path}" - - @field_validator("database_url") - @classmethod - def validate_database_url(cls, v: Optional[str]) -> Optional[str]: - """Validate database URL format.""" - # None is allowed - will use default - return v - class WebSettings(CommonSettings): """Settings for the Web Dashboard component.""" diff --git a/src/meshcore_hub/common/database.py b/src/meshcore_hub/common/database.py index 90c1b31..9ad9dbc 100644 --- a/src/meshcore_hub/common/database.py +++ b/src/meshcore_hub/common/database.py @@ -30,23 +30,32 @@ def _to_async_url(database_url: str) -> str: def create_database_engine( database_url: str, echo: bool = False, + schema: str | None = None, ) -> Engine: """Create a SQLAlchemy 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). Returns: SQLAlchemy Engine instance """ - connect_args = {} + connect_args: dict[str, Any] = {} engine_kwargs: dict[str, Any] = {} # SQLite-specific configuration if database_url.startswith("sqlite"): connect_args["check_same_thread"] = False + # 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}" + # Size the pool above the default Starlette threadpool (~40 threads) so # concurrent request handlers don't block waiting for a connection. Applies # to file-based SQLite and networked backends (e.g. a future Postgres). @@ -126,15 +135,20 @@ class DatabaseManager: to avoid leaking connections when only sync operations are needed. """ - def __init__(self, database_url: str, echo: bool = False): + def __init__( + self, database_url: str, echo: bool = False, schema: str | None = None + ): """Initialize the database manager. Args: database_url: SQLAlchemy database URL echo: Enable SQL query logging + schema: Postgres schema to scope connections to (search_path); ignored for + SQLite """ self.database_url = database_url self._echo = echo + self._schema = schema # Ensure parent directory exists for SQLite databases if database_url.startswith("sqlite:///"): @@ -144,7 +158,7 @@ class DatabaseManager: db_path = Path(database_url.replace("sqlite:///", "")) db_path.parent.mkdir(parents=True, exist_ok=True) - self.engine = create_database_engine(database_url, echo=echo) + self.engine = create_database_engine(database_url, echo=echo, schema=schema) self.session_factory = create_session_factory(self.engine) # Lazy-initialized async engine (created on first async_session call) @@ -159,7 +173,14 @@ class DatabaseManager: from sqlalchemy.ext.asyncio import async_sessionmaker async_url = _to_async_url(self.database_url) - self._async_engine = create_async_engine(async_url, echo=self._echo) + 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")): + 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 + ) # Apply the same SQLite pragmas as the sync engine (see # create_database_engine) for the async engine's connections. diff --git a/tests/test_common/test_config.py b/tests/test_common/test_config.py index d5bb5d4..856ef13 100644 --- a/tests/test_common/test_config.py +++ b/tests/test_common/test_config.py @@ -1,5 +1,7 @@ """Tests for configuration settings.""" +import pytest + from meshcore_hub.common.config import ( CommonSettings, CollectorSettings, @@ -246,3 +248,71 @@ class TestWebSettings: assert settings.feature_radio_config is False assert settings.features["radio_config"] is False + + +class TestDatabaseBackendResolution: + """Tests for DATABASE_BACKEND selection and URL/schema resolution.""" + + def test_default_backend_is_sqlite_unchanged(self) -> None: + """No DB env vars -> the same SQLite path and no schema as before.""" + settings = CollectorSettings(_env_file=None, data_home="/data") + + assert settings.database_backend.value == "sqlite" + assert ( + settings.effective_database_url == "sqlite:////data/collector/meshcore.db" + ) + assert settings.effective_database_schema is None + + def test_postgres_backend_assembles_url_and_schema(self) -> None: + """Postgres backend assembles a URL from components and exposes the schema.""" + settings = APISettings( + _env_file=None, + database_backend="postgres", + database_host="pg", + database_password="pw", + ) + + assert settings.effective_database_url == ( + "postgresql+psycopg2://meshcorehub:pw@pg:5432/meshcorehub" + ) + assert settings.effective_database_schema == "meshcorehub" + + def test_postgres_password_is_url_encoded(self) -> None: + """Special characters in the password are percent-encoded.""" + settings = APISettings( + _env_file=None, + database_backend="postgres", + database_host="pg", + database_password="s3cr3t/p@ss", + ) + + assert "s3cr3t%2Fp%40ss" in settings.effective_database_url + + def test_postgres_schema_override_per_instance(self) -> None: + """DATABASE_SCHEMA isolates instances sharing one database.""" + settings = APISettings( + _env_file=None, + database_backend="postgres", + database_host="pg", + database_password="pw", + database_schema="stg", + ) + + assert settings.effective_database_schema == "stg" + + def test_postgres_missing_required_vars_fails_fast(self) -> None: + """Misconfigured postgres backend raises rather than silently using SQLite.""" + settings = APISettings(_env_file=None, database_backend="postgres") + + with pytest.raises(ValueError, match="DATABASE_BACKEND=postgres requires"): + _ = settings.effective_database_url + + def test_explicit_url_overrides_backend(self) -> None: + """An explicit DATABASE_URL wins even when a backend is selected.""" + settings = CollectorSettings( + _env_file=None, + database_backend="postgres", + database_url="postgresql+psycopg2://u:p@h/db", + ) + + assert settings.effective_database_url == "postgresql+psycopg2://u:p@h/db" From caef666c02c53b34913b67a540518cba696eac0c Mon Sep 17 00:00:00 2001 From: Louis King Date: Sat, 13 Jun 2026 22:26:45 +0100 Subject: [PATCH 05/13] Phase 3: Postgres container + make migrations Postgres-clean - compose: add optional postgres service (postgres:17-alpine, profile 'postgres', healthcheck, postgres_data volume); POSTGRES_* derive from DATABASE_* (single source of truth). DATABASE_* env added to migrate/ collector/api; migrate depends_on postgres with required:false so SQLite deployments are unaffected. - alembic/env: resolve the URL via CommonSettings.effective_database_url so DATABASE_BACKEND=postgres is honoured (previously DATABASE_URL/DATA_HOME only -> would silently migrate SQLite). - migrations: normalize_public_key uses STRING_AGG + HAVING COUNT(*) on Postgres (was SQLite GROUP_CONCAT + alias); raw_packets uses sa.JSON() not the sqlite dialect type. - database: fix _to_async_url to map postgresql+psycopg2:// (what the config assembles) to asyncpg, so API async sessions work on Postgres; resolve the search_path schema from DATABASE_SCHEMA env when not passed explicitly. Validated against a live postgres:17: db upgrade builds all 13 tables in the meshcorehub schema with correct native types (is_observer boolean, decoded json) and alembic_version stamped; the upsert, JSON/timestamptz round-trip, and asyncpg async sessions all work. SQLite suite still green (1061 passed). Co-Authored-By: Claude Opus 4.8 --- alembic/env.py | 29 ++++----- ...20260421_0001_normalize_public_key_case.py | 14 ++++- ...1947_e9f0c4079540_add_raw_packets_table.py | 3 +- docker-compose.yml | 61 +++++++++++++++++++ src/meshcore_hub/common/database.py | 53 +++++++++++----- tests/test_common/test_database.py | 46 +++++++++++++- 6 files changed, 168 insertions(+), 38 deletions(-) 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 From 9a5c6e91172d06a76e26e27bb5cfb9aee5d46e2b Mon Sep 17 00:00:00 2001 From: Louis King Date: Sat, 13 Jun 2026 22:45:29 +0100 Subject: [PATCH 06/13] Phase 4: add 'db migrate-to-postgres' data migration command Copies an existing SQLite database into Postgres at the SQLAlchemy Core level, iterating Base.metadata.sorted_tables (parent-first; excludes alembic_version) and round-tripping each row through the typed columns so booleans, JSON, and timestamptz convert correctly with no per-model code. - streams large tables (stream_results + partitions) in batches - stamps UTC on naive datetimes for tz-aware columns before insert - single target transaction (all-or-nothing); refuses a non-empty target unless --truncate; --dry-run previews per-table counts - disables FK triggers via session_replication_role only when the target role is a superuser, else relies on parent-first order (--no-replication-role to force; managed Postgres). Defaults: source = SQLite under DATA_HOME, target = configured DATABASE_* (schema-scoped). - prints a per-table source->target reconciliation and fails on mismatch Validated end-to-end against live postgres:17 (nodes/observers/raw_packets/ channels): counts reconcile, dedup preserved, is_observer->boolean, decoded->json, received_at->timestamptz (UTC). SQLite suite green (1064). Co-Authored-By: Claude Opus 4.8 --- src/meshcore_hub/__main__.py | 95 ++++++++++++ src/meshcore_hub/common/db_migrate.py | 213 ++++++++++++++++++++++++++ tests/test_common/test_db_migrate.py | 81 ++++++++++ 3 files changed, 389 insertions(+) create mode 100644 src/meshcore_hub/common/db_migrate.py create mode 100644 tests/test_common/test_db_migrate.py diff --git a/src/meshcore_hub/__main__.py b/src/meshcore_hub/__main__.py index 43e8f75..a978c6e 100644 --- a/src/meshcore_hub/__main__.py +++ b/src/meshcore_hub/__main__.py @@ -85,6 +85,101 @@ def db_upgrade(revision: str, database_url: str | None) -> None: click.echo("Database upgrade complete.") +@db.command("migrate-to-postgres") +@click.option( + "--source", + type=str, + default=None, + help="Source SQLite URL (default: sqlite:///{DATA_HOME}/collector/meshcore.db)", +) +@click.option( + "--target", + type=str, + default=None, + help="Target Postgres URL (default: the configured DATABASE_* connection)", +) +@click.option("--batch-size", type=int, default=2000, help="Rows per insert batch") +@click.option( + "--truncate", + is_flag=True, + default=False, + help="Delete existing rows from target tables before loading", +) +@click.option( + "--dry-run", + is_flag=True, + default=False, + help="Report source/target row counts without writing", +) +@click.option( + "--no-replication-role", + is_flag=True, + default=False, + help="Don't disable FK triggers via session_replication_role (managed Postgres)", +) +def db_migrate_to_postgres( + source: str | None, + target: str | None, + batch_size: int, + truncate: bool, + dry_run: bool, + no_replication_role: bool, +) -> None: + """Copy data from an existing SQLite database into PostgreSQL. + + Run 'db upgrade' against the target first to create the schema. This command + only moves data and never modifies the source. + """ + from pathlib import Path + + from meshcore_hub.common.config import CollectorSettings + from meshcore_hub.common.db_migrate import migrate_sqlite_to_postgres + + settings = CollectorSettings() + source_url = ( + source or f"sqlite:///{Path(settings.data_home) / 'collector' / 'meshcore.db'}" + ) + target_url = target or settings.effective_database_url + target_schema = settings.effective_database_schema + + click.echo(f"Source: {source_url}") + click.echo(f"Target: {target_url} (schema: {target_schema})") + if dry_run: + click.echo("Mode: dry-run (no writes)") + + try: + result = migrate_sqlite_to_postgres( + source_url, + target_url, + target_schema=target_schema, + batch_size=batch_size, + truncate=truncate, + dry_run=dry_run, + disable_replication_role=no_replication_role, + ) + except (RuntimeError, ValueError) as exc: + raise click.ClickException(str(exc)) from exc + + click.echo("") + if dry_run: + # Preview only: the target is expected to be empty, so no OK/MISMATCH judgement. + click.echo("table (source rows -> current target rows)") + for t in result.tables: + click.echo(f" {t.name:28} {t.source_rows:>8} -> {t.target_rows:>8}") + click.echo("") + click.echo("Dry run complete.") + return + + click.echo("table (source -> target)") + for t in result.tables: + status = "OK" if t.ok else "MISMATCH" + click.echo(f" {t.name:28} {t.source_rows:>8} -> {t.target_rows:>8} {status}") + if not result.ok: + raise click.ClickException("Row-count mismatch between source and target") + click.echo("") + click.echo("Migration complete.") + + @db.command("downgrade") @click.option( "--revision", diff --git a/src/meshcore_hub/common/db_migrate.py b/src/meshcore_hub/common/db_migrate.py new file mode 100644 index 0000000..0b6e5b3 --- /dev/null +++ b/src/meshcore_hub/common/db_migrate.py @@ -0,0 +1,213 @@ +"""Copy all data from a source (SQLite) database into a target (Postgres) database. + +This powers ``meshcore-hub db migrate-to-postgres``. It operates at the SQLAlchemy +Core level, iterating ``Base.metadata.sorted_tables`` and copying each table through +the ORM's typed columns. The round-trip ``SQLite value -> Python object -> Postgres +value`` is what makes the conversion correct (e.g. integer ``0/1`` -> ``bool``, JSON +``TEXT`` -> ``dict``, datetime string -> ``timestamptz``) without any per-model code. + +The schema must already exist in the target (created by ``db upgrade``); this module +only moves data and never mutates the source. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import timezone +from typing import Optional + +import sqlalchemy as sa +from sqlalchemy import func, insert, select, text +from sqlalchemy.engine import Engine + +from meshcore_hub.common.database import create_database_engine +from meshcore_hub.common.models.base import Base + +logger = logging.getLogger(__name__) + + +@dataclass +class TableResult: + """Per-table outcome of a migration run.""" + + name: str + source_rows: int + target_rows: int + + @property + def ok(self) -> bool: + return self.source_rows == self.target_rows + + +@dataclass +class MigrationResult: + """Aggregate outcome of a migration run.""" + + tables: list[TableResult] = field(default_factory=list) + dry_run: bool = False + + @property + def ok(self) -> bool: + return all(t.ok for t in self.tables) + + +def _tz_aware_columns(table: sa.Table) -> list[str]: + """Names of timezone-aware DateTime columns on a table. + + SQLite does not persist tzinfo, so these values read back naive and must be + stamped UTC before insert into Postgres ``timestamptz`` (the app always writes + UTC via utc_now()). + """ + return [ + col.name + for col in table.columns + if isinstance(col.type, sa.DateTime) + and bool(getattr(col.type, "timezone", False)) + ] + + +def _count(engine: Engine, table: sa.Table) -> int: + with engine.connect() as conn: + return int(conn.execute(select(func.count()).select_from(table)).scalar() or 0) + + +def _is_superuser(engine: Engine) -> bool: + """Whether the target role can SET session_replication_role (superuser-only).""" + if engine.dialect.name != "postgresql": + return False + try: + with engine.connect() as conn: + value = conn.execute( + text("SELECT current_setting('is_superuser')") + ).scalar() + return str(value).lower() == "on" + except Exception: # pragma: no cover - defensive + return False + + +def _copy_table( + source_engine: Engine, + tgt_conn: sa.engine.Connection, + table: sa.Table, + batch_size: int, +) -> int: + """Stream rows from source and bulk-insert into the target connection.""" + tz_cols = _tz_aware_columns(table) + copied = 0 + with source_engine.connect().execution_options(stream_results=True) as src: + result = src.execute(select(table)) + for partition in result.partitions(batch_size): + rows = [] + for row in partition: + data = dict(row._mapping) + for col in tz_cols: + value = data.get(col) + if value is not None and value.tzinfo is None: + data[col] = value.replace(tzinfo=timezone.utc) + rows.append(data) + if rows: + tgt_conn.execute(insert(table), rows) + copied += len(rows) + return copied + + +def migrate_sqlite_to_postgres( + source_url: str, + target_url: str, + *, + target_schema: Optional[str] = None, + batch_size: int = 2000, + truncate: bool = False, + dry_run: bool = False, + disable_replication_role: bool = False, +) -> MigrationResult: + """Copy every table from ``source_url`` into ``target_url``. + + Args: + source_url: SQLAlchemy URL of the source (SQLite) database. + target_url: SQLAlchemy URL of the target (Postgres) database. + target_schema: Postgres schema to load into (search_path). Defaults to the + DATABASE_SCHEMA env var via the engine. + batch_size: Rows per insert batch. + truncate: Delete existing rows from target tables before loading. + dry_run: Report source/target counts without writing. + disable_replication_role: Skip the session_replication_role trick even if the + target role is a superuser (e.g. to mirror managed-Postgres behaviour). + + Returns: + MigrationResult with per-table source/target row counts. + """ + if not target_url.startswith(("postgresql", "postgres")): + raise ValueError("Target must be a PostgreSQL database URL") + + source_engine = create_database_engine(source_url) + target_engine = create_database_engine(target_url, schema=target_schema) + tables = list(Base.metadata.sorted_tables) # parent-first; excludes alembic_version + + try: + # Pre-flight: schema present + (unless truncating) target empty. + for table in tables: + try: + existing = _count(target_engine, table) + except Exception as exc: # table missing -> schema not initialised + raise RuntimeError( + f"Target table {table.name!r} not found. Run 'meshcore-hub db " + f"upgrade' against the target first." + ) from exc + if existing and not (truncate or dry_run): + raise RuntimeError( + f"Target table {table.name!r} is not empty ({existing} rows). " + f"Refusing to load; pass --truncate to overwrite." + ) + + if dry_run: + result = MigrationResult(dry_run=True) + for table in tables: + result.tables.append( + TableResult( + table.name, + _count(source_engine, table), + _count(target_engine, table), + ) + ) + return result + + use_replica = ( + target_engine.dialect.name == "postgresql" + and not disable_replication_role + and _is_superuser(target_engine) + ) + if not use_replica: + logger.info( + "Not disabling FK triggers (session_replication_role); relying on " + "parent-first table order." + ) + + # Single transaction: all-or-nothing. A failure leaves the target empty. + with target_engine.begin() as tgt: + if use_replica: + tgt.execute(text("SET session_replication_role = replica")) + if truncate: + for table in reversed(tables): # children first + tgt.execute(table.delete()) + for table in tables: + copied = _copy_table(source_engine, tgt, table, batch_size) + logger.info("Copied %s rows into %s", copied, table.name) + if use_replica: + tgt.execute(text("SET session_replication_role = DEFAULT")) + + # Reconcile. + result = MigrationResult() + for table in tables: + result.tables.append( + TableResult( + table.name, + _count(source_engine, table), + _count(target_engine, table), + ) + ) + return result + finally: + source_engine.dispose() + target_engine.dispose() diff --git a/tests/test_common/test_db_migrate.py b/tests/test_common/test_db_migrate.py new file mode 100644 index 0000000..3ad0a6c --- /dev/null +++ b/tests/test_common/test_db_migrate.py @@ -0,0 +1,81 @@ +"""Tests for the SQLite -> Postgres data migration helper. + +The full round-trip is validated against a live Postgres; here we cover the +dialect-agnostic pieces (the Postgres-target guard, tz-aware column detection, and +the Core copy/stream/normalize logic exercised SQLite -> SQLite) so they run in CI. +""" + +from datetime import datetime + +import pytest +from sqlalchemy import create_engine, func, select + +from meshcore_hub.common.db_migrate import ( + _copy_table, + _tz_aware_columns, + migrate_sqlite_to_postgres, +) +from meshcore_hub.common.models import Base + +# Import models so their tables register on Base.metadata. +import meshcore_hub.common.models.node # noqa: F401 +import meshcore_hub.common.models.raw_packet # noqa: F401 + + +def test_target_must_be_postgres() -> None: + """The command refuses a non-Postgres target.""" + with pytest.raises(ValueError, match="PostgreSQL"): + migrate_sqlite_to_postgres("sqlite:///a.db", "sqlite:///b.db") + + +def test_tz_aware_columns_detects_timestamptz() -> None: + """Timezone-aware DateTime columns are identified for UTC normalization.""" + cols = _tz_aware_columns(Base.metadata.tables["raw_packets"]) + + assert "received_at" in cols + assert "created_at" in cols + assert "packet_hash" not in cols # not a datetime + + +def test_copy_table_roundtrips_rows_and_boolean() -> None: + """_copy_table streams rows across engines, preserving the boolean value. + + (UTC normalization of naive datetimes is validated against Postgres, where + timestamptz round-trips reliably; SQLite does not retain tzinfo.) + """ + nodes = Base.metadata.tables["nodes"] + src = create_engine("sqlite:///:memory:") + dst = create_engine("sqlite:///:memory:") + Base.metadata.create_all(src) + Base.metadata.create_all(dst) + + now = datetime(2026, 6, 13, 10, 0) + with src.begin() as conn: + conn.execute( + nodes.insert(), + [ + { + "id": f"n{i}", + "public_key": f"key{i}", + "is_observer": i == 0, + "first_seen": now, + "created_at": now, + "updated_at": now, + } + for i in range(3) + ], + ) + + with dst.begin() as conn: + copied = _copy_table(src, conn, nodes, batch_size=2) # forces >1 batch + + assert copied == 3 + with dst.connect() as conn: + assert conn.execute(select(func.count()).select_from(nodes)).scalar() == 3 + observers = conn.execute( + select(func.count()).select_from(nodes).where(nodes.c.is_observer.is_(True)) + ).scalar() + assert observers == 1 # boolean preserved across the copy + + src.dispose() + dst.dispose() From da012afd51855b92d761bd5bf30ae650c698065e Mon Sep 17 00:00:00 2001 From: Louis King Date: Sat, 13 Jun 2026 22:55:37 +0100 Subject: [PATCH 07/13] Install [postgres] extra in the image The runtime image installed the package without extras, so psycopg2/asyncpg were missing and the migrate/collector/api services failed with ModuleNotFoundError when DATABASE_BACKEND=postgres. Install ".[postgres]" so the one image serves both SQLite and Postgres. Co-Authored-By: Claude Opus 4.8 --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index a314db2..6e5fe3b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -55,7 +55,7 @@ ARG BUILD_VERSION=dev # Set version in _version.py and install the package RUN sed -i "s|__version__ = \"dev\"|__version__ = \"${BUILD_VERSION}\"|" src/meshcore_hub/_version.py && \ pip install --upgrade pip && \ - pip install . + pip install ".[postgres]" # ============================================================================= # Stage 3: Runtime - Final production image From afda05403fcd0f4fc2cf73bd527ab8bee6552454 Mon Sep 17 00:00:00 2001 From: Louis King Date: Sun, 14 Jun 2026 08:37:26 +0100 Subject: [PATCH 08/13] Phase 5: document optional Postgres backend + migration runbook Add a Database Backend section to the README (config vars, compose profile, schema-per-instance) and an "Optional PostgreSQL Backend" section to docs/upgrading.md covering enablement, search_path isolation, role/db provisioning, and the SQLite -> Postgres data-migration runbook. Co-Authored-By: Claude Opus 4.8 --- README.md | 28 ++++++++++++++++++- docs/upgrading.md | 69 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index cc4b137..4caa776 100644 --- a/README.md +++ b/README.md @@ -244,7 +244,7 @@ Each worker is an independent process sharing one listening socket, so the kerne Pick a worker count around the number of CPU cores available to the container; start with `2`–`4` and measure under realistic load. -**SQLite caveat:** all workers share the same SQLite file on the same host. WAL mode (enabled automatically) allows concurrent readers alongside the single writer (the collector), so reads scale — but **writes do not**, and this does not extend across multiple hosts (a network filesystem breaks SQLite locking). To scale the API across hosts, switch `DATABASE_URL` to PostgreSQL; the API requires no code changes for this. +**SQLite caveat:** all workers share the same SQLite file on the same host. WAL mode (enabled automatically) allows concurrent readers alongside the single writer (the collector), so reads scale — but **writes do not**, and this does not extend across multiple hosts (a network filesystem breaks SQLite locking). To scale the API across hosts, switch to PostgreSQL (`DATABASE_BACKEND=postgres`); the API requires no code changes for this. See [Database Backend](#database-backend). > Prefer `API_WORKERS` over running multiple `api` containers (`--scale api=N`): the `api` service uses a fixed `container_name`, and one process-managed container per stack keeps logs, health checks, and monitoring simple. @@ -346,6 +346,32 @@ All components are configured via environment variables. Create a `.env` file or > **Note:** `MQTT_PREFIX` also accepts the legacy alias `MQTT_TOPIC_PREFIX` for backward compatibility. +### Database Backend + +MeshCore Hub defaults to **SQLite** (zero-config, single host). Set `DATABASE_BACKEND=postgres` to switch to **PostgreSQL** for write scaling and multi-host deployments. Postgres is opt-in — leave these unset to keep using SQLite. + +| Variable | Default | Description | +| ------------------- | ------------- | --------------------------------------------------------------------------------------- | +| `DATABASE_BACKEND` | `sqlite` | `sqlite` or `postgres`. Explicit switch — Postgres is never selected implicitly. | +| `DATABASE_HOST` | `postgres` | Postgres hostname (`postgres` = bundled container service name) | +| `DATABASE_PORT` | `5432` | Postgres port | +| `DATABASE_NAME` | `meshcorehub` | Database name | +| `DATABASE_SCHEMA` | `meshcorehub` | Schema (search_path). Set a distinct value per instance on a shared cluster | +| `DATABASE_USER` | `meshcorehub` | Role name | +| `DATABASE_PASSWORD` | _(none)_ | **Required** for Postgres | +| `DATABASE_URL` | _(none)_ | Advanced: full SQLAlchemy URL; overrides all of the above | + +**Docker:** Postgres is bundled behind the `postgres` profile. The container's credentials/name are derived from the `DATABASE_*` values (single source of truth). + +```bash +docker compose --profile postgres --profile core up # Start on Postgres +docker compose --profile core up # Start on SQLite (default) +``` + +**Schema-per-instance:** several instances (e.g. `prod`, `stg`) can share one Postgres cluster, each isolated to its own schema via `search_path` — give each a distinct `DATABASE_SCHEMA`. The schema is created automatically on `db upgrade`. + +See [docs/upgrading.md](docs/upgrading.md#optional-postgresql-backend) for the setup reference and the SQLite → Postgres data-migration runbook. + ### Collector Settings | Variable | Default | Description | diff --git a/docs/upgrading.md b/docs/upgrading.md index 08021f9..8752c51 100644 --- a/docs/upgrading.md +++ b/docs/upgrading.md @@ -4,6 +4,75 @@ This guide covers upgrading from a previous MeshCore Hub release to the current ## v0.13.0 +### Optional PostgreSQL Backend + +MeshCore Hub can now run on **PostgreSQL** as an alternative to the default SQLite database. SQLite remains the zero-config default — Postgres is entirely opt-in and **no action is required** to keep using SQLite. Switch to Postgres to scale writes and run the stack across multiple hosts (SQLite's file locking does not work over network filesystems and caps you at a single host). Existing operators can migrate their live SQLite data into Postgres with a single command (downtime required while writers are stopped). + +#### Enabling Postgres + +Set `DATABASE_BACKEND=postgres` and the `DATABASE_*` connection variables, then activate the compose `postgres` profile: + +| Variable | Default | Description | +| ------------------- | ------------- | -------------------------------------------------------------------------------------------- | +| `DATABASE_BACKEND` | `sqlite` | `sqlite` (default) or `postgres`. Explicit switch — Postgres is never used implicitly. | +| `DATABASE_HOST` | `postgres` | Postgres hostname (`postgres` is the bundled container's service name). | +| `DATABASE_PORT` | `5432` | Postgres port. | +| `DATABASE_NAME` | `meshcorehub` | Database name. The bundled container is initialised with this name. | +| `DATABASE_SCHEMA` | `meshcorehub` | Postgres schema (search_path). **Set a distinct value per instance** on a shared cluster. | +| `DATABASE_USER` | `meshcorehub` | Role name. The bundled container is initialised with this user. | +| `DATABASE_PASSWORD` | _(none)_ | **Required** for Postgres. Generate one, e.g. `openssl rand -base64 32`. | + +```bash +# Start the stack on Postgres (bundled container) +docker compose -f docker-compose.yml -f docker-compose.dev.yml \ + --profile postgres --profile core up -d +``` + +The bundled `postgres` container derives its `POSTGRES_USER` / `POSTGRES_PASSWORD` / `POSTGRES_DB` from the same `DATABASE_USER` / `DATABASE_PASSWORD` / `DATABASE_NAME` values — one source of truth. For a **managed/external** Postgres, point `DATABASE_HOST` at it (and skip the `postgres` profile). Advanced users can instead set a full `DATABASE_URL` (e.g. `postgresql+psycopg2://user:pass@host:5432/db`), which takes precedence over the component variables. + +#### Schema-per-instance (`search_path`) + +Each Hub instance is isolated to its own Postgres **schema** via the connection's `search_path`, not its own database. This lets several instances (e.g. `prod`, `stg`) share **one** Postgres cluster without colliding — each gets its own tables and its own `alembic_version`. Give every instance a distinct `DATABASE_SCHEMA` (e.g. `meshcorehub_prod`, `meshcorehub_stg`). The schema is created automatically on `db upgrade` if it does not exist. + +#### Provisioning the role and database + +The bundled container provisions the role and database for you on first start. For a managed/external Postgres, create them once before pointing Hub at it: + +```sql +CREATE ROLE meshcorehub LOGIN PASSWORD 'your-password'; +CREATE DATABASE meshcorehub OWNER meshcorehub; +-- The schema is created by `db upgrade`; the role just needs CREATE on the database. +``` + +No admin/bootstrap credentials are needed at runtime — Hub only ever connects as `DATABASE_USER`. + +#### Migrating an existing SQLite database to Postgres + +Downtime is required while writers are stopped; the source SQLite file is never modified. + +1. **Back up first.** Copy your `meshcore.db` (or back up the `hub_data` volume — see [Backup & Restore](../README.md#backup--restore)). +2. **Stop the writers** (collector and api): + ```bash + docker compose -f docker-compose.yml -f docker-compose.dev.yml stop collector api + ``` +3. **Bring up Postgres** and create the schema: + ```bash + docker compose -f docker-compose.yml -f docker-compose.dev.yml --profile postgres up -d postgres + docker compose -f docker-compose.yml -f docker-compose.dev.yml --profile postgres run --rm migrate + ``` + `migrate` runs `db upgrade` against Postgres, creating the schema, all tables (with correct native types — `boolean`, `json`, `timestamptz`), and stamping `alembic_version`. +4. **Copy the data** with the built-in command: + ```bash + docker compose -f docker-compose.yml -f docker-compose.dev.yml --profile postgres \ + run --rm migrate meshcore-hub db migrate-to-postgres + ``` + It defaults the source to `sqlite:///{DATA_HOME}/collector/meshcore.db` and the target to your configured `DATABASE_*` connection. It copies every table in foreign-key order through the ORM (so SQLite's dynamically typed values are converted correctly — `0/1` → `boolean`, JSON text → `json`, naive datetimes → UTC `timestamptz`), then prints a per-table source-vs-target row-count reconciliation and fails on any mismatch. Use `--dry-run` to preview counts first, and `--truncate` to overwrite a non-empty target. +5. **Start the stack on Postgres** with `DATABASE_BACKEND=postgres` set (see *Enabling Postgres* above). + +> **Why not pgloader?** pgloader infers the target schema from SQLite's *dynamic* typing and produces wrong Postgres types (e.g. `is_observer` as `bigint` not `boolean`, JSON columns as `text`, no `timestamptz`), and no `alembic_version` consistent with the migration history. The built-in command reuses the ORM models, so types convert correctly and the schema is created by `db upgrade`. + +> **Managed Postgres / non-superuser roles:** the migration disables foreign-key triggers during the copy via `session_replication_role = replica`, which requires a superuser. When the target role is not a superuser (typical for managed Postgres), the command automatically falls back to copying in parent-first order instead. Pass `--no-replication-role` to force the fallback explicitly. + ### Raw Packets (capture, browse, and search wire packets) A new **Raw Packets** feature captures every inbound MeshCore packet exactly as it arrives over the LetsMesh `packets` feed into a dedicated `raw_packets` table, independent of how the collector later classifies it. A new `/packets` API and a SPA **Packets** page (table on desktop, cards on mobile) let operators browse, filter, and search the raw traffic. From ae700c45fa977844361287daea02da9c7c9e6129 Mon Sep 17 00:00:00 2001 From: Louis King Date: Sun, 14 Jun 2026 09:03:01 +0100 Subject: [PATCH 09/13] Isolate tests from a local .env Settings classes use env_file=".env" for deployments, which also caused tests to pick up a developer's local .env (e.g. DATABASE_BACKEND=postgres without DATABASE_HOST), failing unrelated API/CLI tests. Add an autouse fixture that sets env_file=None on CommonSettings and every subclass, so tests depend only on defaults and explicit monkeypatch.setenv overrides. Co-Authored-By: Claude Opus 4.8 --- tests/conftest.py | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index 206605a..ed0a8ab 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,9 +4,40 @@ import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker +from meshcore_hub.common import config as config_module from meshcore_hub.common.models import Base +def _settings_classes(): + """CommonSettings and every subclass (recursively).""" + seen: set[type] = set() + stack = [config_module.CommonSettings] + while stack: + cls = stack.pop() + if cls in seen: + continue + seen.add(cls) + stack.extend(cls.__subclasses__()) + return seen + + +@pytest.fixture(autouse=True) +def _ignore_dotenv(monkeypatch): + """Stop pydantic-settings from reading the repo-root ``.env`` during tests. + + Settings classes set ``env_file=".env"`` so deployments can drop a file in + place. Tests must depend only on defaults and explicit env overrides + (``monkeypatch.setenv``), never on whatever a developer happens to have in + their local ``.env`` — otherwise e.g. ``DATABASE_BACKEND=postgres`` there + leaks into unrelated tests. Each settings subclass carries its own merged + ``model_config`` dict, so patch every one. + """ + for cls in _settings_classes(): + cfg = dict(cls.model_config) + cfg["env_file"] = None + monkeypatch.setattr(cls, "model_config", cfg) + + @pytest.fixture def db_engine(): """Create an in-memory SQLite database engine for testing.""" From 1c04baec7de514e321cad20c41c45329ec1c3678 Mon Sep 17 00:00:00 2001 From: Louis King Date: Sun, 14 Jun 2026 09:17:53 +0100 Subject: [PATCH 10/13] Widen messages.signature to String(32) for Postgres The column was declared String(8) but stores 16-char hex signatures (and can hold the up-to-32-char packet_hash fallback from the LetsMesh normalizer). SQLite never enforced the length, so the undersized definition went unnoticed until db migrate-to-postgres rejected the data with StringDataRightTruncation (varchar(8)). Widen the model column and add an Alembic migration (batch_alter_table, so it applies on both SQLite and Postgres). Verified end-to-end migrating a 1.2GB live SQLite DB into Postgres (all 12 tables reconcile). Co-Authored-By: Claude Opus 4.8 --- ...15_d4e5f6a7b8c9_widen_message_signature.py | 42 +++++++++++++++++++ src/meshcore_hub/common/models/message.py | 4 +- 2 files changed, 44 insertions(+), 2 deletions(-) create mode 100644 alembic/versions/20260614_0915_d4e5f6a7b8c9_widen_message_signature.py diff --git a/alembic/versions/20260614_0915_d4e5f6a7b8c9_widen_message_signature.py b/alembic/versions/20260614_0915_d4e5f6a7b8c9_widen_message_signature.py new file mode 100644 index 0000000..ea6f563 --- /dev/null +++ b/alembic/versions/20260614_0915_d4e5f6a7b8c9_widen_message_signature.py @@ -0,0 +1,42 @@ +"""widen messages.signature to 32 chars + +Revision ID: d4e5f6a7b8c9 +Revises: c3d4e5f6a7b8 +Create Date: 2026-06-14 09:15:00.000000+00:00 + +The column was declared String(8) but actually stores 16-char hex signatures +(and can hold the up-to-32-char packet_hash fallback). SQLite never enforced the +length, so the undersized definition went unnoticed until a Postgres migration +rejected the data (varchar(8)). Widen it to String(32). + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "d4e5f6a7b8c9" +down_revision: Union[str, None] = "c3d4e5f6a7b8" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("messages", schema=None) as batch_op: + batch_op.alter_column( + "signature", + existing_type=sa.String(length=8), + type_=sa.String(length=32), + existing_nullable=True, + ) + + +def downgrade() -> None: + with op.batch_alter_table("messages", schema=None) as batch_op: + batch_op.alter_column( + "signature", + existing_type=sa.String(length=32), + type_=sa.String(length=8), + existing_nullable=True, + ) diff --git a/src/meshcore_hub/common/models/message.py b/src/meshcore_hub/common/models/message.py index 5b0813b..938bee6 100644 --- a/src/meshcore_hub/common/models/message.py +++ b/src/meshcore_hub/common/models/message.py @@ -21,7 +21,7 @@ class Message(Base, UUIDMixin, TimestampMixin): text: Message content path_len: Number of hops txt_type: Message type indicator - signature: Message signature (8 hex chars) + signature: Message signature (hex), or the packet_hash fallback snr: Signal-to-noise ratio sender_timestamp: Sender's timestamp received_at: When received by interface @@ -60,7 +60,7 @@ class Message(Base, UUIDMixin, TimestampMixin): nullable=True, ) signature: Mapped[Optional[str]] = mapped_column( - String(8), + String(32), nullable=True, ) snr: Mapped[Optional[float]] = mapped_column( From 7a372dc9862118f1b1abcfdf4cc0702baf5d78b5 Mon Sep 17 00:00:00 2001 From: Louis King Date: Sun, 14 Jun 2026 10:30:23 +0100 Subject: [PATCH 11/13] Move Postgres upgrade notes to a new v0.14.0 section PostgreSQL support ships in v0.14.0; v0.13.0 covers the packet-handling enhancements. Reparent the "Optional PostgreSQL Backend" notes accordingly. Co-Authored-By: Claude Opus 4.8 --- docs/upgrading.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/upgrading.md b/docs/upgrading.md index 8752c51..dd633b8 100644 --- a/docs/upgrading.md +++ b/docs/upgrading.md @@ -2,7 +2,7 @@ This guide covers upgrading from a previous MeshCore Hub release to the current version. Check the relevant version section below before upgrading. -## v0.13.0 +## v0.14.0 ### Optional PostgreSQL Backend @@ -73,6 +73,8 @@ Downtime is required while writers are stopped; the source SQLite file is never > **Managed Postgres / non-superuser roles:** the migration disables foreign-key triggers during the copy via `session_replication_role = replica`, which requires a superuser. When the target role is not a superuser (typical for managed Postgres), the command automatically falls back to copying in parent-first order instead. Pass `--no-replication-role` to force the fallback explicitly. +## v0.13.0 + ### Raw Packets (capture, browse, and search wire packets) A new **Raw Packets** feature captures every inbound MeshCore packet exactly as it arrives over the LetsMesh `packets` feed into a dedicated `raw_packets` table, independent of how the collector later classifies it. A new `/packets` API and a SPA **Packets** page (table on desktop, cards on mobile) let operators browse, filter, and search the raw traffic. From 34b6e6b3281f1912953c03355786160d4df9da6d Mon Sep 17 00:00:00 2001 From: Louis King Date: Sun, 14 Jun 2026 21:11:02 +0100 Subject: [PATCH 12/13] test: cover Postgres migration helper and CLI command Raise patch coverage on the v0.14.0 Postgres backend: - test_db_migrate: _is_superuser plus the full migrate_sqlite_to_postgres flow (dry-run, copy, non-empty-target refusal, --truncate, missing schema), routing create_database_engine to SQLite files so a postgresql:// target satisfies the guard while the run executes SQLite -> SQLite in CI. - test_main: the `db migrate-to-postgres` CLI command via CliRunner (success, dry-run, row-count mismatch, ValueError -> ClickException). - conftest: neutralise dotenv.load_dotenv before collection. Importing the CLI entrypoint runs load_dotenv() at import time, which leaked a local .env into os.environ and broke unrelated config/redis tests. Co-Authored-By: Claude Opus 4.8 --- tests/conftest.py | 9 ++ tests/test_common/test_db_migrate.py | 126 ++++++++++++++++++++++++++ tests/test_main.py | 127 +++++++++++++++++++++++++++ 3 files changed, 262 insertions(+) create mode 100644 tests/test_main.py diff --git a/tests/conftest.py b/tests/conftest.py index ed0a8ab..bb1b26e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,6 @@ """Shared pytest fixtures for all tests.""" +import dotenv import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker @@ -7,6 +8,14 @@ from sqlalchemy.orm import sessionmaker from meshcore_hub.common import config as config_module from meshcore_hub.common.models import Base +# The CLI entrypoint (meshcore_hub.__main__) calls load_dotenv() at import time so +# deployments can drop a .env in place. Importing it during collection (e.g. from +# test_main.py) would otherwise leak a developer's repo-root .env straight into +# os.environ for the whole session — bypassing _ignore_dotenv, which only stops +# pydantic-settings from reading the file. conftest.py is imported before any test +# module is collected, so neutralising load_dotenv here binds first. +dotenv.load_dotenv = lambda *args, **kwargs: False + def _settings_classes(): """CommonSettings and every subclass (recursively).""" diff --git a/tests/test_common/test_db_migrate.py b/tests/test_common/test_db_migrate.py index 3ad0a6c..d886679 100644 --- a/tests/test_common/test_db_migrate.py +++ b/tests/test_common/test_db_migrate.py @@ -10,8 +10,10 @@ from datetime import datetime import pytest from sqlalchemy import create_engine, func, select +from meshcore_hub.common import db_migrate from meshcore_hub.common.db_migrate import ( _copy_table, + _is_superuser, _tz_aware_columns, migrate_sqlite_to_postgres, ) @@ -79,3 +81,127 @@ def test_copy_table_roundtrips_rows_and_boolean() -> None: src.dispose() dst.dispose() + + +def test_is_superuser_false_for_non_postgres() -> None: + """session_replication_role is Postgres-only; SQLite is never a superuser target.""" + engine = create_engine("sqlite:///:memory:") + try: + assert _is_superuser(engine) is False + finally: + engine.dispose() + + +def _seed_nodes(engine, count: int) -> None: + nodes = Base.metadata.tables["nodes"] + now = datetime(2026, 6, 13, 10, 0) + with engine.begin() as conn: + conn.execute( + nodes.insert(), + [ + { + "id": f"n{i}", + "public_key": f"key{i}", + "is_observer": i == 0, + "first_seen": now, + "created_at": now, + "updated_at": now, + } + for i in range(count) + ], + ) + + +@pytest.fixture +def _patch_engines(monkeypatch, tmp_path): + """Route create_database_engine to SQLite files keyed by URL. + + Lets the Postgres-targeting migration flow run end-to-end SQLite -> SQLite in CI: + a ``postgresql://`` target URL satisfies the guard while the real engine is a + local SQLite file, so the schema/empty/truncate/copy logic is exercised without + a live Postgres. + """ + src_url = f"sqlite:///{tmp_path / 'src.db'}" + target_url = "postgresql://fake/target" # satisfies the Postgres guard + src_engine = create_engine(src_url) + tgt_engine = create_engine(f"sqlite:///{tmp_path / 'tgt.db'}") + Base.metadata.create_all(src_engine) + + def fake_create_engine(url, echo=False, schema=None): + if url == src_url: + return src_engine + if url == target_url: + return tgt_engine + raise AssertionError(f"unexpected url {url!r}") + + monkeypatch.setattr(db_migrate, "create_database_engine", fake_create_engine) + return src_url, target_url, src_engine, tgt_engine + + +def _count_rows(engine) -> int: + nodes = Base.metadata.tables["nodes"] + with engine.connect() as conn: + return int(conn.execute(select(func.count()).select_from(nodes)).scalar() or 0) + + +def test_migrate_dry_run_reports_counts_without_writing(_patch_engines) -> None: + """Dry run reports source/target counts and leaves the target untouched.""" + src_url, target_url, src_engine, tgt_engine = _patch_engines + Base.metadata.create_all(tgt_engine) + _seed_nodes(src_engine, 3) + + result = migrate_sqlite_to_postgres(src_url, target_url, dry_run=True) + + assert result.dry_run is True + nodes_result = next(t for t in result.tables if t.name == "nodes") + assert nodes_result.source_rows == 3 + assert nodes_result.target_rows == 0 + assert _count_rows(tgt_engine) == 0 # nothing written + + +def test_migrate_copies_all_rows(_patch_engines) -> None: + """A full run copies rows and reconciles source/target counts as OK.""" + src_url, target_url, src_engine, tgt_engine = _patch_engines + Base.metadata.create_all(tgt_engine) + _seed_nodes(src_engine, 5) + + result = migrate_sqlite_to_postgres(src_url, target_url, batch_size=2) + + assert result.ok is True + assert _count_rows(tgt_engine) == 5 + nodes_result = next(t for t in result.tables if t.name == "nodes") + assert nodes_result.source_rows == nodes_result.target_rows == 5 + + +def test_migrate_refuses_non_empty_target(_patch_engines) -> None: + """Without --truncate, a non-empty target is refused before any write.""" + src_url, target_url, src_engine, tgt_engine = _patch_engines + Base.metadata.create_all(tgt_engine) + _seed_nodes(src_engine, 2) + _seed_nodes(tgt_engine, 1) + + with pytest.raises(RuntimeError, match="not empty"): + migrate_sqlite_to_postgres(src_url, target_url) + + +def test_migrate_truncate_overwrites_target(_patch_engines) -> None: + """--truncate clears existing target rows before loading from source.""" + src_url, target_url, src_engine, tgt_engine = _patch_engines + Base.metadata.create_all(tgt_engine) + _seed_nodes(src_engine, 2) + _seed_nodes(tgt_engine, 4) + + result = migrate_sqlite_to_postgres(src_url, target_url, truncate=True) + + assert result.ok is True + assert _count_rows(tgt_engine) == 2 + + +def test_migrate_errors_when_target_schema_missing(_patch_engines) -> None: + """A target without the schema (no create_all) fails with a clear message.""" + src_url, target_url, src_engine, tgt_engine = _patch_engines + # Note: target schema intentionally not created. + _seed_nodes(src_engine, 1) + + with pytest.raises(RuntimeError, match="Run 'meshcore-hub db upgrade'"): + migrate_sqlite_to_postgres(src_url, target_url) diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..8a1618e --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,127 @@ +"""Tests for the top-level CLI, focused on ``db migrate-to-postgres``. + +The migration engine itself is covered in test_common/test_db_migrate.py; here we +verify the command's wiring: option plumbing, dry-run vs. real output, and how the +MigrationResult (or an error) maps onto exit codes and messages. +""" + +from unittest.mock import patch + +from click.testing import CliRunner + +from meshcore_hub.__main__ import cli +from meshcore_hub.common.db_migrate import MigrationResult, TableResult + + +def _result(*tables: TableResult, dry_run: bool = False) -> MigrationResult: + return MigrationResult(tables=list(tables), dry_run=dry_run) + + +def test_migrate_to_postgres_success() -> None: + """A matching run reports OK per table and exits 0.""" + runner = CliRunner() + fake = _result(TableResult("nodes", 5, 5)) + + with patch( + "meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres", + return_value=fake, + ) as mock_migrate: + result = runner.invoke( + cli, + [ + "db", + "migrate-to-postgres", + "--source", + "sqlite:///src.db", + "--target", + "postgresql://u@h/db", + ], + ) + + assert result.exit_code == 0 + assert "nodes" in result.output + assert "OK" in result.output + assert "Migration complete." in result.output + # Flags thread through to the engine call. + _, kwargs = mock_migrate.call_args + assert kwargs["dry_run"] is False + assert kwargs["truncate"] is False + + +def test_migrate_to_postgres_dry_run() -> None: + """Dry run prints a preview and never renders OK/MISMATCH judgements.""" + runner = CliRunner() + fake = _result(TableResult("nodes", 3, 0), dry_run=True) + + with patch( + "meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres", + return_value=fake, + ) as mock_migrate: + result = runner.invoke( + cli, + [ + "db", + "migrate-to-postgres", + "--source", + "sqlite:///src.db", + "--target", + "postgresql://u@h/db", + "--dry-run", + ], + ) + + assert result.exit_code == 0 + assert "dry-run" in result.output + assert "Dry run complete." in result.output + assert "OK" not in result.output + assert mock_migrate.call_args.kwargs["dry_run"] is True + + +def test_migrate_to_postgres_mismatch_exits_nonzero() -> None: + """A row-count mismatch surfaces as a ClickException (non-zero exit).""" + runner = CliRunner() + fake = _result(TableResult("nodes", 5, 4)) # ok == False + + with patch( + "meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres", + return_value=fake, + ): + result = runner.invoke( + cli, + [ + "db", + "migrate-to-postgres", + "--source", + "sqlite:///src.db", + "--target", + "postgresql://u@h/db", + ], + ) + + assert result.exit_code != 0 + assert "MISMATCH" in result.output + assert "mismatch" in result.output.lower() + + +def test_migrate_to_postgres_value_error_becomes_click_exception() -> None: + """A ValueError from the engine (e.g. bad target) maps to a clean CLI error.""" + runner = CliRunner() + + with patch( + "meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres", + side_effect=ValueError("Target must be a PostgreSQL database URL"), + ): + result = runner.invoke( + cli, + [ + "db", + "migrate-to-postgres", + "--source", + "sqlite:///src.db", + "--target", + "sqlite:///bad.db", + ], + ) + + assert result.exit_code != 0 + assert "PostgreSQL" in result.output From 7603b16092ba28bda8e5434a2b0fa58a503959ff Mon Sep 17 00:00:00 2001 From: Louis King Date: Sun, 14 Jun 2026 22:58:52 +0100 Subject: [PATCH 13/13] Fixed pre-commit issues --- tests/conftest.py | 30 ------------------------------ 1 file changed, 30 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index e973d8d..34fbc9a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,36 +17,6 @@ from meshcore_hub.common.models import Base dotenv.load_dotenv = lambda *args, **kwargs: False -def _settings_classes(): - """CommonSettings and every subclass (recursively).""" - seen: set[type] = set() - stack = [config_module.CommonSettings] - while stack: - cls = stack.pop() - if cls in seen: - continue - seen.add(cls) - stack.extend(cls.__subclasses__()) - return seen - - -@pytest.fixture(autouse=True) -def _ignore_dotenv(monkeypatch): - """Stop pydantic-settings from reading the repo-root ``.env`` during tests. - - Settings classes set ``env_file=".env"`` so deployments can drop a file in - place. Tests must depend only on defaults and explicit env overrides - (``monkeypatch.setenv``), never on whatever a developer happens to have in - their local ``.env`` — otherwise e.g. ``DATABASE_BACKEND=postgres`` there - leaks into unrelated tests. Each settings subclass carries its own merged - ``model_config`` dict, so patch every one. - """ - for cls in _settings_classes(): - cfg = dict(cls.model_config) - cfg["env_file"] = None - monkeypatch.setattr(cls, "model_config", cfg) - - def _settings_classes(): """CommonSettings and every subclass (recursively).""" seen: set[type] = set()