feat: enhance EvoBGP with new command-line options for the evobgp-agent, including a watch command for periodic configuration updates. Update go.mod with additional dependencies and improve Docker Compose setup for new services, including NATS and various worker components.

internal/db/apply.go

@@ -0,0 +1,55 @@
+package db
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io/fs"
+	"path"
+	"sort"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// ApplyPostgresMigrations runs ordered *.up.sql from an embedded subtree (idempotent via schema_migrations).
+func ApplyPostgresMigrations(ctx context.Context, pool *pgxpool.Pool, fsys fs.FS, subdir string) error {
+	if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (version TEXT NOT NULL PRIMARY KEY)`); err != nil {
+		return fmt.Errorf("db: schema_migrations: %w", err)
+	}
+	entries, err := fs.ReadDir(fsys, subdir)
+	if err != nil {
+		return err
+	}
+	var ups []string
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasSuffix(e.Name(), ".up.sql") {
+			continue
+		}
+		ups = append(ups, e.Name())
+	}
+	sort.Strings(ups)
+	for _, name := range ups {
+		ver := strings.TrimSuffix(name, ".up.sql")
+		var dummy int
+		err := pool.QueryRow(ctx, `SELECT 1 FROM schema_migrations WHERE version = $1`, ver).Scan(&dummy)
+		if err == nil {
+			continue
+		}
+		if !errors.Is(err, pgx.ErrNoRows) {
+			return err
+		}
+		body, err := fs.ReadFile(fsys, path.Join(subdir, name))
+		if err != nil {
+			return err
+		}
+		if _, err := pool.Exec(ctx, string(body)); err != nil {
+			return fmt.Errorf("db: migrate %s: %w", name, err)
+		}
+		if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, ver); err != nil {
+			return err
+		}
+	}
+	return nil
+}

internal/db/open.go

@@ -0,0 +1,97 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io/fs"
+	"path"
+	"sort"
+	"strings"
+
+	"evobgp/migrations"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	_ "modernc.org/sqlite"
+)
+
+// OpenPostgresPool connects, applies embedded migrations, returns a pool.
+func OpenPostgresPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
+	cfg, err := pgxpool.ParseConfig(dsn)
+	if err != nil {
+		return nil, err
+	}
+	pool, err := pgxpool.NewWithConfig(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+	if err := ApplyPostgresMigrations(ctx, pool, migrations.Postgres, "postgres"); err != nil {
+		pool.Close()
+		return nil, err
+	}
+	return pool, nil
+}
+
+// OpenSQLite applies sqlite/*.up.sql then opens the DB (foreign keys on).
+func OpenSQLite(ctx context.Context, filePath string) (*sql.DB, error) {
+	dsn := fmt.Sprintf("file:%s?_pragma=foreign_keys(1)", strings.TrimPrefix(filePath, "file:"))
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		return nil, err
+	}
+	if err := db.PingContext(ctx); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	if err := applySQLiteMigrations(ctx, db, migrations.SQLite, "sqlite"); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	return db, nil
+}
+
+func applySQLiteMigrations(ctx context.Context, db *sql.DB, fsys fs.FS, subdir string) error {
+	if _, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (version TEXT NOT NULL PRIMARY KEY)`); err != nil {
+		return fmt.Errorf("db: sqlite schema_migrations: %w", err)
+	}
+	entries, err := fs.ReadDir(fsys, subdir)
+	if err != nil {
+		return err
+	}
+	var ups []string
+	for _, e := range entries {
+		if e.IsDir() || !strings.HasSuffix(e.Name(), ".up.sql") {
+			continue
+		}
+		ups = append(ups, e.Name())
+	}
+	sort.Strings(ups)
+	for _, name := range ups {
+		ver := strings.TrimSuffix(name, ".up.sql")
+		var dummy int
+		err := db.QueryRowContext(ctx, `SELECT 1 FROM schema_migrations WHERE version = ?`, ver).Scan(&dummy)
+		if err == nil {
+			continue
+		}
+		if !errors.Is(err, sql.ErrNoRows) {
+			return err
+		}
+		body, err := fs.ReadFile(fsys, path.Join(subdir, name))
+		if err != nil {
+			return err
+		}
+		if _, err := db.ExecContext(ctx, string(body)); err != nil {
+			// Idempotent ALTER ADD COLUMN for sqlite dev re-runs
+			if strings.Contains(err.Error(), "duplicate column") {
+				_, _ = db.ExecContext(ctx, `INSERT OR IGNORE INTO schema_migrations (version) VALUES (?)`, ver)
+				continue
+			}
+			return fmt.Errorf("db: sqlite migrate %s: %w", name, err)
+		}
+		if _, err := db.ExecContext(ctx, `INSERT INTO schema_migrations (version) VALUES (?)`, ver); err != nil {
+			return err
+		}
+	}
+	return nil
+}