"""Idempotent schema bootstrap.

The customer's database is empty until they put something in it. The first

thing every teesql-backed Python service does on startup is run the

migration to ensure its schema exists. Re-running is a no-op.

The schema:

events (id BIGSERIAL, topic TEXT, payload JSONB, recorded_at TIMESTAMPTZ)

Plus an `events_notify` trigger that fires `pg_notify('events_channel', ...)`

on every INSERT. The example doesn't currently consume the NOTIFY because

postgres LISTEN/NOTIFY is NOT replicated to streaming standbys — instead

the secondary poller does `WHERE id > last_seen` on a 50ms cadence (see

`poller.py`). The trigger stays anyway so:

1. Forks that prefer LISTEN/NOTIFY against the PRIMARY can flip the

poller into a listener without a schema migration.

2. The cluster's WAL records the trigger fire — operationally useful

to confirm that writes actually committed.

Migrations are forward-only and idempotent: `CREATE TABLE/TRIGGER IF NOT

EXISTS`, `CREATE OR REPLACE FUNCTION`. No `DROP` paths.

"""

from __future__ import annotations

import logging

import psycopg

log = logging.getLogger(__name__)

_MIGRATIONS: list[tuple[str, str]] = [

(

"0001_events",

"""

CREATE TABLE IF NOT EXISTS events (

id BIGSERIAL PRIMARY KEY,

topic TEXT NOT NULL,

payload JSONB NOT NULL,

recorded_at TIMESTAMPTZ NOT NULL DEFAULT now()

);

CREATE INDEX IF NOT EXISTS events_id_idx

ON events (id);

CREATE INDEX IF NOT EXISTS events_recorded_at_idx

ON events (recorded_at DESC);

""",

),

(

"0002_events_notify_trigger",

"""

CREATE OR REPLACE FUNCTION events_notify() RETURNS trigger AS $func$

BEGIN

PERFORM pg_notify(

'events_channel',

json_build_object(

'id', NEW.id,

'topic', NEW.topic,

'payload', NEW.payload,

'recorded_at', NEW.recorded_at

)::text

);

RETURN NEW;

END;

$func$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS events_notify_after_insert ON events;

CREATE TRIGGER events_notify_after_insert

AFTER INSERT ON events

FOR EACH ROW EXECUTE FUNCTION events_notify();

""",

),

]

def apply_migrations(conn: psycopg.Connection) -> None:

"""Apply every migration in `_MIGRATIONS`. Idempotent.

No bookkeeping table — every DDL is `IF NOT EXISTS` or `OR REPLACE`.

Forks that grow beyond a couple tables should swap this for alembic

and a proper schema_version table.

"""

for name, ddl in _MIGRATIONS:

log.info("applying migration %s", name)

with conn.cursor() as cur:

cur.execute(ddl)

conn.commit()

log.info("migrations complete (%d total)", len(_MIGRATIONS))