M1(1-2/8): migrations + seed + cmd/seed runner

- migrations/001_init.up.sql: companies, individuals, fcm_tokens
  (minimum viable for M1 round-trip; full SPEC §4 schema lands
  in M2 via additional migrations)
- migrations/001_init.down.sql: reverse
- migrations/002_deliveries.up.sql: deliveries table for per-attempt
  audit + M8 replay (status: pending|sent|failed|dlq)
- migrations/002_deliveries.down.sql: reverse
- migrations/seed.sql: idempotent. 1 company, 1 individual, 1 fcm
  token. ON CONFLICT DO NOTHING.
- internal/postgres: pgxpool wrapper with retry-on-startup so
  the seed runner and services survive DB not-yet-up.
- cmd/seed: applies *.up.sql in lexical order, then seed.sql.
  Acquire()s a single connection so multi-statement files work.

.gitignore

@@ -26,3 +26,5 @@ docker-compose.override.yml
 /routerd
 /deliverd
 /admind
+/seed
+/fakefcmd

cmd/seed/main.go

@@ -0,0 +1,100 @@
+// Command seed applies migrations and the seed file to a fresh
+// Postgres. Idempotent — safe to re-run.
+//
+// Usage:
+//
+//	BA_POSTGRES_DSN=postgres://... BA_MIGRATIONS_DIR=./migrations \
+//	  go run ./cmd/seed
+//
+// In docker-compose this is a one-shot sidecar that runs before
+// the app services start.
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"git3.techno-world.net/lrosales/broad-announce/internal/postgres"
+)
+
+func main() {
+	dsn := os.Getenv("BA_POSTGRES_DSN")
+	if dsn == "" {
+		die("BA_POSTGRES_DSN is required")
+	}
+	dir := os.Getenv("BA_MIGRATIONS_DIR")
+	if dir == "" {
+		die("BA_MIGRATIONS_DIR is required")
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	pool, err := postgres.Connect(ctx, dsn)
+	if err != nil {
+		die("connect: " + err.Error())
+	}
+	defer pool.Close()
+
+	if err := applyDir(ctx, pool, dir, "*.up.sql"); err != nil {
+		die("apply migrations: " + err.Error())
+	}
+	seed := filepath.Join(dir, "seed.sql")
+	if _, err := os.Stat(seed); err == nil {
+		if err := execFile(ctx, pool, seed); err != nil {
+			die("seed: " + err.Error())
+		}
+		fmt.Fprintln(os.Stderr, "seed applied:", seed)
+	} else {
+		fmt.Fprintln(os.Stderr, "no seed.sql in", dir)
+	}
+	fmt.Fprintln(os.Stderr, "ok")
+}
+
+func applyDir(ctx context.Context, pool *postgres.Pool, dir, pattern string) error {
+	ups, err := filepath.Glob(filepath.Join(dir, pattern))
+	if err != nil {
+		return err
+	}
+	sort.Strings(ups)
+	for _, p := range ups {
+		if err := execFile(ctx, pool, p); err != nil {
+			return fmt.Errorf("%s: %w", filepath.Base(p), err)
+		}
+		fmt.Fprintln(os.Stderr, "applied:", filepath.Base(p))
+	}
+	return nil
+}
+
+func execFile(ctx context.Context, pool *postgres.Pool, path string) error {
+	body, err := os.ReadFile(path)
+	if err != nil {
+		return err
+	}
+	// pgx's Pool.Exec supports multi-statement SQL when the
+	// underlying connection's protocol-mode supports it. To be
+	// safe we acquire a single connection and exec the whole file
+	// as one batch. If the file contains a CREATE EXTENSION that
+	// requires superuser, the migration container runs as superuser
+	// so this is fine.
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		return fmt.Errorf("acquire conn: %w", err)
+	}
+	defer conn.Release()
+	_, err = conn.Exec(ctx, string(body))
+	return err
+}
+
+func die(msg string) {
+	fmt.Fprintln(os.Stderr, "seed:", msg)
+	os.Exit(1)
+}
+
+// _ silences unused import for build tags.
+var _ = strings.TrimSpace

go.mod

@@ -5,6 +5,10 @@ go 1.25.0
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/pgx/v5 v5.10.0 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/klauspost/compress v1.18.5 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nats-io/nats.go v1.52.0 // indirect
@@ -18,6 +22,8 @@ require (
 	go.uber.org/atomic v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/crypto v0.49.0 // indirect
+	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
+	golang.org/x/text v0.35.0 // indirect
 	google.golang.org/protobuf v1.36.8 // indirect
 )

go.sum

@@ -2,6 +2,15 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
 github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
@@ -12,6 +21,7 @@ github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
 github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
 github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
 github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
@@ -22,16 +32,24 @@ github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzM
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
 github.com/redis/go-redis/v9 v9.20.1 h1:sfCU6A8P3dXbKyWes02uxA2baehGux9dZHfEKtsTB1w=
 github.com/redis/go-redis/v9 v9.20.1/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
 go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
 golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
 golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
 golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
 golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
+golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
 google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
 google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/postgres/postgres.go

@@ -0,0 +1,56 @@
+// Package postgres wraps pgx for the few queries the services need.
+// M0 doesn't use Postgres; M1 needs it for routing + delivery.
+//
+// We use the pgxpool connection pool so concurrent queries don't
+// step on each other.
+package postgres
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// Pool is a thin alias for *pgxpool.Pool so the call sites don't
+// import pgx directly.
+type Pool = pgxpool.Pool
+
+// Connect dials Postgres with retries. The first migration run will
+// hit this before the DB is up, so we wait.
+func Connect(ctx context.Context, dsn string) (*Pool, error) {
+	cfg, err := pgxpool.ParseConfig(dsn)
+	if err != nil {
+		return nil, fmt.Errorf("parse dsn: %w", err)
+	}
+	cfg.MaxConns = 20
+	cfg.MinConns = 2
+	cfg.MaxConnIdleTime = 5 * time.Minute
+
+	var pool *Pool
+	deadline, hasDeadline := ctx.Deadline()
+	if !hasDeadline {
+		deadline = time.Now().Add(30 * time.Second)
+	}
+	for {
+		pool, err = pgxpool.NewWithConfig(ctx, cfg)
+		if err == nil {
+			pingCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
+			err = pool.Ping(pingCtx)
+			cancel()
+			if err == nil {
+				return pool, nil
+			}
+			pool.Close()
+		}
+		if time.Now().After(deadline) {
+			return nil, fmt.Errorf("postgres connect timeout: %w", err)
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-time.After(1 * time.Second):
+		}
+	}
+}

migrations/001_init.down.sql

@@ -0,0 +1,6 @@
+-- 001_init.down.sql
+DROP INDEX IF EXISTS idx_fcm_tokens_individual;
+DROP TABLE IF EXISTS fcm_tokens;
+DROP INDEX IF EXISTS idx_individuals_company;
+DROP TABLE IF EXISTS individuals;
+DROP TABLE IF EXISTS companies;

migrations/001_init.up.sql

@@ -0,0 +1,45 @@
+-- 001_init.up.sql
+-- M1 minimum-viable schema. Three tables, all tenant-scoped.
+-- Future migrations add: groups, group_members, sources,
+-- subscriptions, routing_rules, telegram_bots.
+--
+-- All tables include company_id and the per-tenant queries MUST
+-- filter on it. We do NOT enable RLS in v1; isolation is enforced
+-- in the app layer. See SPEC §4 + §22.
+
+CREATE EXTENSION IF NOT EXISTS pgcrypto;
+
+CREATE TABLE IF NOT EXISTS companies (
+    id            TEXT PRIMARY KEY,
+    name          TEXT NOT NULL,
+    status        TEXT NOT NULL DEFAULT 'active',   -- active | suspended
+    rate_limit_per_sec INTEGER NOT NULL DEFAULT 10000,
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS individuals (
+    id            TEXT PRIMARY KEY,                 -- individuals are global IDs, not scoped
+    company_id    TEXT NOT NULL REFERENCES companies(id) ON DELETE CASCADE,
+    full_name     TEXT NOT NULL,
+    email         TEXT,
+    phone_e164    TEXT,
+    locale        TEXT DEFAULT 'en',
+    tz            TEXT DEFAULT 'UTC',
+    status        TEXT NOT NULL DEFAULT 'active',   -- active | suspended
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_individuals_company ON individuals(company_id) WHERE status = 'active';
+
+CREATE TABLE IF NOT EXISTS fcm_tokens (
+    id            BIGSERIAL PRIMARY KEY,
+    individual_id TEXT NOT NULL REFERENCES individuals(id) ON DELETE CASCADE,
+    token         TEXT NOT NULL UNIQUE,
+    device_id     TEXT,
+    platform      TEXT NOT NULL DEFAULT 'android',
+    locale        TEXT,
+    app_version   TEXT,
+    last_seen     TIMESTAMPTZ,
+    status        TEXT NOT NULL DEFAULT 'active',   -- active | unregistered
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_fcm_tokens_individual ON fcm_tokens(individual_id) WHERE status = 'active';

migrations/002_deliveries.down.sql

@@ -0,0 +1,4 @@
+-- 002_deliveries.down.sql
+DROP INDEX IF EXISTS idx_deliveries_status;
+DROP INDEX IF EXISTS idx_deliveries_company_alert;
+DROP TABLE IF EXISTS deliveries;

migrations/002_deliveries.up.sql

@@ -0,0 +1,27 @@
+-- 002_deliveries.up.sql
+-- Per-attempt delivery row. In M1 this is written by deliverd-fcm
+-- after each fake-fcm call. In M3+ it gets the full DLQ + retry
+-- semantics from SPEC §9.
+--
+-- For M1 the schema is intentionally minimal:
+--   - alert_id, individual_id, channel, target, status, attempts
+--   - the payload (jsonb) for replay (M8)
+-- Timescale conversion lands in M7.
+
+CREATE TABLE IF NOT EXISTS deliveries (
+    id            BIGSERIAL PRIMARY KEY,
+    alert_id      TEXT NOT NULL,
+    company_id    TEXT NOT NULL,
+    individual_id TEXT NOT NULL,
+    channel       TEXT NOT NULL,                    -- fcm | telegram | sms | email | slack | teams | webhook
+    target        TEXT NOT NULL,                    -- the fcm_token, telegram_chat_id, phone_e164, …
+    status        TEXT NOT NULL DEFAULT 'pending',  -- pending | sent | failed | dlq
+    attempts      INT NOT NULL DEFAULT 0,
+    last_error    TEXT,
+    payload       JSONB,                            -- snapshot at enqueue time, for M8 replay
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
+    sent_at       TIMESTAMPTZ,
+    next_attempt_at TIMESTAMPTZ
+);
+CREATE INDEX IF NOT EXISTS idx_deliveries_company_alert ON deliveries(company_id, alert_id);
+CREATE INDEX IF NOT EXISTS idx_deliveries_status ON deliveries(status) WHERE status IN ('pending','failed');

migrations/seed.sql

@@ -0,0 +1,15 @@
+-- seed.sql
+-- M1 minimum seed: one company, one individual, one FCM token.
+-- Idempotent (uses ON CONFLICT DO NOTHING). Safe to re-run.
+
+INSERT INTO companies (id, name) VALUES
+    ('acme-001', 'Acme Corp')
+ON CONFLICT (id) DO NOTHING;
+
+INSERT INTO individuals (id, company_id, full_name, email, locale) VALUES
+    ('ind-acme-001', 'acme-001', 'Alice Operator', 'alice@acme.example', 'en')
+ON CONFLICT (id) DO NOTHING;
+
+INSERT INTO fcm_tokens (individual_id, token, device_id, locale, app_version) VALUES
+    ('ind-acme-001', 'fake-fcm-token-acme-alice-001', 'pixel-7-alice', 'en', '1.0.0')
+ON CONFLICT (token) DO NOTHING;