Adopt golib/pg for migrations and pool construction

forgebot applied its schema by executing an inline DDL string on every boot,
with no version tracking and no lock, so two API replicas starting together
raced and the SQL had nowhere to grow. golib/pg already owns that mechanic for
the estate; take it and keep owning the SQL.

- Move the schema into migrations/0001_init.sql, embedded via migrations.FS.
  The DDL is verbatim.
- The four legacy-status UPDATEs move into 0001 unchanged. Each reads only
  retired statuses (pending/failed/running/succeeded/cancelled) and writes only
  current ones, and no current status is a source, so replaying 0001 once
  against the live database is a no-op. A test pins that property.
- Build the pool with pg.NewMigrated, LockName "forgebot-migrations": the
  advisory lock serializes replicas, schema_migrations records what ran, and a
  migration failure fails startup instead of half-migrating. database.New and
  apiserver.New take a context and logger for it.
- Render the DSN with pg.DSN, which percent-escapes the credentials the
  fmt.Sprintf builder pasted in raw. LoadConfig still reads the environment
  itself: pg.DSNFromEnv has no defaults for user and database name, where
  forgebot defaults both to "forgebot", and would newly honour DATABASE_URL and
  PG*. The deployed DBHOST/DBPORT/DBUSER/DBPASS/DBNAME/DBSSL contract and its
  defaults are unchanged, and pinned by a test.
- Plumb GOPRIVATE=git.unkin.net for the first cross-repo Go dependency:
  exported by the Makefile, set in both Dockerfiles and the woodpecker Go
  steps, documented in the README.
- gofmt the four files that were already unformatted on main, so the
  pre-commit step can pass.

internal/apiserver/config.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"os"
 	"strconv"
+
+	"git.unkin.net/unkin/golib/pg"
 )
 
 type Config struct {
@@ -19,9 +21,16 @@ type Config struct {
 	GiteaToken    string
 }
 
+// DatabaseDSN renders the connection string with golib's builder, which
+// percent-escapes the credentials — byte-identical to the fmt.Sprintf form it
+// replaces for values without reserved characters.
+//
+// The environment is still read by LoadConfig rather than pg.DSNFromEnv: the
+// library has no defaults for the user and database name, where forgebot
+// defaults both to "forgebot", and it would newly honour DATABASE_URL and the
+// PG* variables. Deployments set the DB* names below and nothing else.
 func (c *Config) DatabaseDSN() string {
-	return fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=%s",
-		c.DBUser, c.DBPass, c.DBHost, c.DBPort, c.DBName, c.DBSSL)
+	return pg.DSN(c.DBHost, c.DBPort, c.DBUser, c.DBPass, c.DBName, c.DBSSL)
 }
 
 func LoadConfig() (*Config, error) {

internal/apiserver/config_test.go

@@ -0,0 +1,55 @@
+package apiserver
+
+import (
+	"testing"
+)
+
+// The deployed contract is these six variables and these defaults; nothing else
+// is consulted for the database connection.
+func TestLoadConfig_DatabaseEnvContract(t *testing.T) {
+	// Set by a deployment that would confuse a DATABASE_URL/PG*-aware loader.
+	t.Setenv("DATABASE_URL", "postgres://someone@elsewhere:5432/other")
+	t.Setenv("PGHOST", "elsewhere")
+
+	cfg, err := LoadConfig()
+	if err != nil {
+		t.Fatalf("LoadConfig: %v", err)
+	}
+	if got, want := cfg.DatabaseDSN(), "postgres://forgebot:@localhost:5432/forgebot?sslmode=disable"; got != want {
+		t.Fatalf("default DSN = %q, want %q", got, want)
+	}
+
+	t.Setenv("DBHOST", "db.example")
+	t.Setenv("DBPORT", "6432")
+	t.Setenv("DBUSER", "bot")
+	t.Setenv("DBPASS", "hunter2")
+	t.Setenv("DBNAME", "tasks")
+	t.Setenv("DBSSL", "require")
+
+	cfg, err = LoadConfig()
+	if err != nil {
+		t.Fatalf("LoadConfig: %v", err)
+	}
+	if got, want := cfg.DatabaseDSN(), "postgres://bot:hunter2@db.example:6432/tasks?sslmode=require"; got != want {
+		t.Fatalf("DSN = %q, want %q", got, want)
+	}
+}
+
+func TestLoadConfig_RejectsBadPort(t *testing.T) {
+	t.Setenv("DBPORT", "not-a-port")
+	if _, err := LoadConfig(); err == nil {
+		t.Fatal("expected an error for a non-numeric DBPORT")
+	}
+}
+
+// A password with reserved characters used to truncate the DSN; the builder
+// percent-escapes the credentials so it round-trips through pgx intact.
+func TestDatabaseDSN_EscapesCredentials(t *testing.T) {
+	cfg := &Config{
+		DBHost: "db.example", DBPort: 5432, DBUser: "bo/t",
+		DBPass: "p@ss/word", DBName: "tasks", DBSSL: "disable",
+	}
+	if got, want := cfg.DatabaseDSN(), "postgres://bo%2Ft:p%40ss%2Fword@db.example:5432/tasks?sslmode=disable"; got != want {
+		t.Fatalf("DSN = %q, want %q", got, want)
+	}
+}

internal/apiserver/server.go

@@ -22,8 +22,10 @@ type Server struct {
 	provider *gitea.Client
 }
 
-func New(cfg *Config) (*Server, error) {
-	db, err := database.New(cfg.DatabaseDSN())
+// New connects to Postgres and migrates the schema before wiring the routes,
+// so a failed migration is a failed startup rather than a broken server.
+func New(ctx context.Context, cfg *Config) (*Server, error) {
+	db, err := database.New(ctx, cfg.DatabaseDSN(), slog.Default())
 	if err != nil {
 		return nil, err
 	}

internal/database/migrations.go

@@ -1,41 +0,0 @@
-package database
-
-import "context"
-
-func (db *DB) migrate() error {
-	_, err := db.Pool.Exec(context.Background(), `
-		CREATE TABLE IF NOT EXISTS tasks (
-			id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
-			parent_task_id  UUID REFERENCES tasks(id),
-			command         TEXT NOT NULL,
-			skill           TEXT NOT NULL DEFAULT '',
-			repository      TEXT NOT NULL,
-			ref             TEXT NOT NULL,
-			issue_number    INTEGER NOT NULL DEFAULT 0,
-			pr_number       INTEGER NOT NULL DEFAULT 0,
-			comment_id      BIGINT NOT NULL DEFAULT 0,
-			body            TEXT NOT NULL DEFAULT '',
-			author          TEXT NOT NULL,
-			extra_tools     TEXT[] NOT NULL DEFAULT '{}',
-			status          TEXT NOT NULL DEFAULT 'todo',
-			pool_ref        TEXT NOT NULL DEFAULT '',
-			job_name        TEXT NOT NULL DEFAULT '',
-			result          TEXT NOT NULL DEFAULT '',
-			error_message   TEXT NOT NULL DEFAULT '',
-			created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-			started_at      TIMESTAMPTZ,
-			completed_at    TIMESTAMPTZ
-		);
-
-		CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
-		CREATE INDEX IF NOT EXISTS idx_tasks_repository ON tasks(repository);
-		CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_task_id);
-
-		-- migrate legacy statuses
-		UPDATE tasks SET status = 'todo' WHERE status IN ('pending', 'failed');
-		UPDATE tasks SET status = 'in_progress' WHERE status = 'running';
-		UPDATE tasks SET status = 'done' WHERE status = 'succeeded';
-		UPDATE tasks SET status = 'wontdo' WHERE status = 'cancelled';
-	`)
-	return err
-}

internal/database/migrations_test.go

@@ -0,0 +1,195 @@
+package database
+
+import (
+	"io/fs"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"testing"
+
+	"git.unkin.net/unkin/golib/pg"
+
+	"git.unkin.net/unkin/forgebot/migrations"
+	"git.unkin.net/unkin/forgebot/pkg/models"
+)
+
+// migrationsDir is the repo's migrations/ directory, relative to this package.
+const migrationsDir = "../../migrations"
+
+func readMigrations(t *testing.T) map[string]string {
+	t.Helper()
+	entries, err := os.ReadDir(migrationsDir)
+	if err != nil {
+		t.Fatalf("read migrations dir: %v", err)
+	}
+	out := map[string]string{}
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") {
+			continue
+		}
+		b, err := os.ReadFile(filepath.Join(migrationsDir, e.Name()))
+		if err != nil {
+			t.Fatalf("read %s: %v", e.Name(), err)
+		}
+		out[e.Name()] = string(b)
+	}
+	if len(out) == 0 {
+		t.Fatal("no migrations found")
+	}
+	return out
+}
+
+// The embedded set is the shipped schema, so it must match the migrations/
+// directory exactly — a file added on disk but not embedded would never run.
+func TestEmbeddedMigrationsMatchDirectory(t *testing.T) {
+	onDisk := readMigrations(t)
+	entries, err := fs.ReadDir(migrations.FS, ".")
+	if err != nil {
+		t.Fatalf("read embedded migrations: %v", err)
+	}
+	embedded := map[string]string{}
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") {
+			continue
+		}
+		b, err := migrations.FS.ReadFile(e.Name())
+		if err != nil {
+			t.Fatalf("read embedded %s: %v", e.Name(), err)
+		}
+		embedded[e.Name()] = string(b)
+	}
+	if len(embedded) != len(onDisk) {
+		t.Fatalf("embedded %d files, migrations/ has %d", len(embedded), len(onDisk))
+	}
+	for name, body := range embedded {
+		want, ok := onDisk[name]
+		if !ok {
+			t.Errorf("%s is embedded but not in migrations/", name)
+			continue
+		}
+		if body != want {
+			t.Errorf("%s: embedded body differs from migrations/%s", name, name)
+		}
+	}
+}
+
+// The advisory lock key is derived from migrationLockName by golib. Every
+// forgebot-api replica must contend on the same key, so renaming the lock would
+// silently let two versions migrate at once during a rolling deploy.
+func TestMigrationLockKeyIsStable(t *testing.T) {
+	const deployedKey int64 = -570492662391362667
+	if got := pg.LockKey(migrationLockName); got != deployedKey {
+		t.Fatalf("LockKey(%q) = %d, want %d", migrationLockName, got, deployedKey)
+	}
+}
+
+// Migrations run against a live database with existing rows, and a file absent
+// from schema_migrations is re-run even where the schema is already present, so
+// every statement must be guarded and re-runnable.
+func TestMigrations_AreAdditiveAndIdempotent(t *testing.T) {
+	for name, body := range readMigrations(t) {
+		upper := strings.ToUpper(body)
+		for _, forbidden := range []string{"DROP TABLE", "DROP COLUMN", "ALTER COLUMN", "TRUNCATE", "DELETE FROM"} {
+			if strings.Contains(upper, forbidden) {
+				t.Errorf("%s contains destructive statement %q", name, forbidden)
+			}
+		}
+		for _, stmt := range strings.Split(upper, ";") {
+			stmt = strings.TrimSpace(stmt)
+			switch {
+			case strings.HasPrefix(stmt, "CREATE TABLE"), strings.HasPrefix(stmt, "CREATE INDEX"):
+				if !strings.Contains(stmt, "IF NOT EXISTS") {
+					t.Errorf("%s: %q is not guarded with IF NOT EXISTS", name, firstLine(stmt))
+				}
+			case strings.HasPrefix(stmt, "ALTER TABLE"):
+				if !strings.Contains(stmt, "ADD COLUMN IF NOT EXISTS") {
+					t.Errorf("%s: %q is not an idempotent ADD COLUMN IF NOT EXISTS", name, firstLine(stmt))
+				}
+			}
+		}
+	}
+}
+
+// The legacy-status rewrites are data fixups carried over from the pre-golib
+// runner, which re-ran them on every boot. Under versioned migrations 0001 is
+// replayed once against the live database, so each rewrite must read only
+// retired statuses and write only current ones: a rewrite whose target is also
+// one of its sources would cascade rows on that replay.
+func TestMigration0001_StatusRewritesAreIdempotent(t *testing.T) {
+	body, ok := readMigrations(t)["0001_init.sql"]
+	if !ok {
+		t.Fatal("0001_init.sql missing")
+	}
+
+	current := map[string]bool{}
+	for _, s := range []models.TaskStatus{
+		models.StatusTodo, models.StatusInProgress, models.StatusInReview,
+		models.StatusDone, models.StatusWontdo,
+	} {
+		current[string(s)] = true
+	}
+
+	rewrite := regexp.MustCompile(`(?i)UPDATE tasks SET status = '(\w+)' WHERE status (?:=|IN) \(?([^)\n;]+)\)?`)
+	matches := rewrite.FindAllStringSubmatch(body, -1)
+	if len(matches) == 0 {
+		t.Fatal("no status rewrites found in 0001_init.sql")
+	}
+	for _, m := range matches {
+		target := m[1]
+		if !current[target] {
+			t.Errorf("rewrite targets %q, which is not a current status", target)
+		}
+		for _, raw := range strings.Split(m[2], ",") {
+			source := strings.Trim(strings.TrimSpace(raw), "'")
+			if current[source] {
+				t.Errorf("rewrite reads current status %q as a legacy source, so a replay would cascade", source)
+			}
+		}
+	}
+}
+
+// Every column 0001 creates must be read back by the queries in tasks.go, so a
+// schema change can never silently stop being scanned.
+func TestMigrations_ColumnsAreSelected(t *testing.T) {
+	body := readMigrations(t)["0001_init.sql"]
+	create := regexp.MustCompile(`(?s)CREATE TABLE IF NOT EXISTS tasks \((.*?)\n\);`)
+	m := create.FindStringSubmatch(body)
+	if m == nil {
+		t.Fatal("could not parse the tasks CREATE TABLE in 0001_init.sql")
+	}
+	cols := map[string]bool{}
+	for _, line := range strings.Split(m[1], "\n") {
+		if f := strings.Fields(strings.TrimSpace(line)); len(f) > 0 {
+			cols[f[0]] = true
+		}
+	}
+
+	src, err := os.ReadFile("tasks.go")
+	if err != nil {
+		t.Fatalf("read tasks.go: %v", err)
+	}
+	// The task queries all select the full column list; find it and check it
+	// covers the table.
+	selected := map[string]bool{}
+	for _, sel := range regexp.MustCompile(`(?s)SELECT (id, parent_task_id.*?)\n\s*FROM tasks`).FindAllStringSubmatch(string(src), -1) {
+		for _, c := range strings.Split(sel[1], ",") {
+			selected[strings.TrimSpace(c)] = true
+		}
+	}
+	if len(selected) == 0 {
+		t.Fatal("no task SELECT found in tasks.go")
+	}
+	for c := range cols {
+		if !selected[c] {
+			t.Errorf("column %q is never read back by the task queries", c)
+		}
+	}
+}
+
+func firstLine(s string) string {
+	if i := strings.IndexByte(s, '\n'); i >= 0 {
+		return s[:i]
+	}
+	return s
+}

internal/database/postgres.go

@@ -2,30 +2,35 @@ package database
 
 import (
 	"context"
-	"fmt"
+	"log/slog"
 
 	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.unkin.net/unkin/golib/pg"
+
+	"git.unkin.net/unkin/forgebot/migrations"
 )
 
+// migrationLockName names the cluster-wide advisory lock the migration run
+// contends for. golib derives the key as FNV-1a/64 of this name, so every
+// replica must pass the same string to stay mutually exclusive.
+const migrationLockName = "forgebot-migrations"
+
 type DB struct {
 	Pool *pgxpool.Pool
 }
 
-func New(dsn string) (*DB, error) {
-	pool, err := pgxpool.New(context.Background(), dsn)
+// New opens the pool and brings the schema up to date before returning, so the
+// API never serves against a half-migrated database. log may be nil.
+func New(ctx context.Context, dsn string, log *slog.Logger) (*DB, error) {
+	pool, err := pg.NewMigrated(ctx, dsn, migrations.FS, pg.MigrateOptions{
+		LockName: migrationLockName,
+		Logger:   log,
+	})
 	if err != nil {
-		return nil, fmt.Errorf("connect to postgres: %w", err)
+		return nil, err
 	}
-	if err := pool.Ping(context.Background()); err != nil {
-		pool.Close()
-		return nil, fmt.Errorf("ping postgres: %w", err)
-	}
-	db := &DB{Pool: pool}
-	if err := db.migrate(); err != nil {
-		pool.Close()
-		return nil, fmt.Errorf("run migrations: %w", err)
-	}
-	return db, nil
+	return &DB{Pool: pool}, nil
 }
 
 func (db *DB) Close() {

internal/provider/gitea/webhook.go

@@ -11,11 +11,11 @@ import (
 )
 
 type webhookPayload struct {
-	Action     string `json:"action"`
-	Comment    *commentPayload `json:"comment,omitempty"`
-	Issue      *issuePayload   `json:"issue,omitempty"`
-	Repository *repoPayload    `json:"repository"`
-	PullRequest *prPayload     `json:"pull_request,omitempty"`
+	Action      string          `json:"action"`
+	Comment     *commentPayload `json:"comment,omitempty"`
+	Issue       *issuePayload   `json:"issue,omitempty"`
+	Repository  *repoPayload    `json:"repository"`
+	PullRequest *prPayload      `json:"pull_request,omitempty"`
 }
 
 type commentPayload struct {
@@ -27,7 +27,7 @@ type commentPayload struct {
 }
 
 type issuePayload struct {
-	Number      int  `json:"number"`
+	Number      int       `json:"number"`
 	PullRequest *struct{} `json:"pull_request,omitempty"`
 }
 
@@ -39,7 +39,7 @@ type prPayload struct {
 }
 
 type repoPayload struct {
-	FullName     string `json:"full_name"`
+	FullName      string `json:"full_name"`
 	DefaultBranch string `json:"default_branch"`
 }
 

internal/tui/app.go

@@ -267,4 +267,3 @@ func tickCmd() tea.Cmd {
 		return tickMsg(t)
 	})
 }
-