feat(agent): agentdb bootstrap, migration runner, v1 schema

agent/src/agent/db.py

@@ -0,0 +1,70 @@
+"""agentdb access: bootstrap, migrations, connections."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import psycopg
+
+from agent.config import AgentConfig
+
+_MIGRATIONS_DIR = Path(__file__).parent / "migrations"
+
+
+def _dsn(cfg: AgentConfig, dbname: str) -> str:
+    return (
+        f"host={cfg.db_host} port={cfg.db_port} dbname={dbname} "
+        f"user={cfg.db_user} password={cfg.db_password}"
+    )
+
+
+def admin_dsn(cfg: AgentConfig) -> str:
+    """DSN for the maintenance database that always exists."""
+    return _dsn(cfg, "postgres")
+
+
+def agent_dsn(cfg: AgentConfig) -> str:
+    return _dsn(cfg, cfg.agent_db_name)
+
+
+def connect(cfg: AgentConfig) -> psycopg.Connection:
+    return psycopg.connect(agent_dsn(cfg))
+
+
+def ensure_database(cfg: AgentConfig) -> None:
+    """Create the agent database if it does not exist. Idempotent."""
+    with psycopg.connect(admin_dsn(cfg), autocommit=True) as conn:
+        exists = conn.execute(
+            "SELECT 1 FROM pg_database WHERE datname = %s", (cfg.agent_db_name,)
+        ).fetchone()
+        if not exists:
+            # CREATE DATABASE cannot be parameterized; the name comes from
+            # our own config, quoted defensively via psycopg's identifier API.
+            from psycopg import sql
+
+            conn.execute(
+                sql.SQL("CREATE DATABASE {}").format(sql.Identifier(cfg.agent_db_name))
+            )
+
+
+def run_migrations(cfg: AgentConfig) -> list[str]:
+    """Apply pending migrations in filename order. Returns names applied."""
+    applied: list[str] = []
+    with connect(cfg) as conn:
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS schema_migrations ("
+            " name TEXT PRIMARY KEY,"
+            " applied_at TIMESTAMPTZ NOT NULL DEFAULT now())"
+        )
+        done = {
+            row[0] for row in conn.execute("SELECT name FROM schema_migrations")
+        }
+        for path in sorted(_MIGRATIONS_DIR.glob("*.sql")):
+            if path.name in done:
+                continue
+            conn.execute(path.read_text())
+            conn.execute(
+                "INSERT INTO schema_migrations (name) VALUES (%s)", (path.name,)
+            )
+            applied.append(path.name)
+    return applied

agent/src/agent/migrations/0001_initial.sql

@@ -0,0 +1,47 @@
+-- agentdb schema v1 (see spec: agentdb schema section)
+
+CREATE TABLE sync_runs (
+    id             BIGSERIAL PRIMARY KEY,
+    connector      TEXT        NOT NULL,
+    started_at     TIMESTAMPTZ NOT NULL,
+    finished_at    TIMESTAMPTZ,
+    status         TEXT        NOT NULL CHECK (status IN ('running', 'ok', 'error')),
+    items_upserted INTEGER     NOT NULL DEFAULT 0,
+    detail         TEXT
+);
+
+CREATE INDEX sync_runs_connector_id_idx ON sync_runs (connector, id DESC);
+
+CREATE TABLE documents (
+    id          BIGSERIAL PRIMARY KEY,
+    path        TEXT        NOT NULL UNIQUE,
+    sha256      CHAR(64)    NOT NULL UNIQUE,
+    institution TEXT        NOT NULL,
+    account     TEXT,
+    doc_type    TEXT        NOT NULL,
+    period      TEXT,
+    source      TEXT        NOT NULL,
+    received_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE bills (
+    id              BIGSERIAL PRIMARY KEY,
+    payee           TEXT          NOT NULL,
+    expected_amount NUMERIC(12,2),
+    due_date        DATE          NOT NULL,
+    funding_account TEXT,
+    autopay         BOOLEAN       NOT NULL DEFAULT false,
+    status          TEXT          NOT NULL DEFAULT 'upcoming'
+                    CHECK (status IN ('upcoming', 'approved', 'paid', 'skipped', 'overdue'))
+);
+
+CREATE TABLE approvals (
+    id                  BIGSERIAL PRIMARY KEY,
+    bill_id             BIGINT        NOT NULL REFERENCES bills (id),
+    amount              NUMERIC(12,2) NOT NULL,
+    requested_at        TIMESTAMPTZ   NOT NULL DEFAULT now(),
+    decided_at          TIMESTAMPTZ,
+    decision            TEXT          CHECK (decision IN ('approved', 'denied')),
+    telegram_message_id TEXT,
+    executed_at         TIMESTAMPTZ
+);

agent/tests/conftest.py

@@ -0,0 +1,29 @@
+import itertools
+
+import pytest
+from testcontainers.postgres import PostgresContainer
+
+from agent.config import AgentConfig
+
+_db_counter = itertools.count()
+
+
+@pytest.fixture(scope="session")
+def pg():
+    with PostgresContainer("postgres:16-alpine") as container:
+        yield container
+
+
+@pytest.fixture
+def cfg(pg) -> AgentConfig:
+    """AgentConfig pointing at a fresh, uniquely named database per test."""
+    return AgentConfig(
+        db_host=pg.get_container_host_ip(),
+        db_port=int(pg.get_exposed_port(5432)),
+        db_user=pg.username,
+        db_password=pg.password,
+        agent_db_name=f"agentdb_test_{next(_db_counter)}",
+        firefly_url="http://firefly.test",
+        firefly_token="test-token",
+        heartbeat_interval_minutes=60,
+    )

agent/tests/test_db.py

@@ -0,0 +1,39 @@
+from agent.db import connect, ensure_database, run_migrations
+
+EXPECTED_TABLES = {"schema_migrations", "sync_runs", "documents", "bills", "approvals"}
+
+
+def test_ensure_database_creates_and_is_idempotent(cfg):
+    ensure_database(cfg)
+    ensure_database(cfg)  # second call must not raise
+    with connect(cfg) as conn:
+        assert conn.execute("SELECT 1").fetchone()[0] == 1
+
+
+def test_run_migrations_applies_each_migration_exactly_once(cfg):
+    ensure_database(cfg)
+
+    first = run_migrations(cfg)
+    assert first == ["0001_initial.sql"]
+
+    second = run_migrations(cfg)
+    assert second == []
+
+    with connect(cfg) as conn:
+        rows = conn.execute(
+            "SELECT tablename FROM pg_tables WHERE schemaname = 'public'"
+        ).fetchall()
+    assert EXPECTED_TABLES <= {r[0] for r in rows}
+
+
+def test_sync_runs_rejects_bad_status(cfg):
+    import psycopg
+    import pytest
+
+    ensure_database(cfg)
+    run_migrations(cfg)
+    with connect(cfg) as conn, pytest.raises(psycopg.errors.CheckViolation):
+        conn.execute(
+            "INSERT INTO sync_runs (connector, started_at, status)"
+            " VALUES ('x', now(), 'bogus')"
+        )