Merge pull request 'feat(cli): manage configured databases' (#5) from issue-3-cli-database-management into main

Reviewed-on: #5
Reviewed-by: Warky <2+warkanum@noreply@warky.dev>

README.md

@@ -369,6 +369,27 @@ The `databases` array can contain multiple database configurations. Each entry s
 | `conn_max_idle_time` | Connection max idle time | `10m` |
 | `queue_count` | Number of queues for this database | `4` |
 | `tenant_id` | Tenant id set for this connection (RLS) | `default` |
+| `auto_migrate` | Apply pending migrations during `start` | `false` |
+| `disabled` | Exclude this database from `start` | `false` |
+
+### Database Management Commands
+
+The configuration can be managed without editing YAML by hand:
+
+```bash
+pgsql-broker db list --config broker.yaml
+pgsql-broker db add replica --from db1 --host db.example --database jobs_replica --user broker --password secret --yes --non-interactive --config broker.yaml
+pgsql-broker db disable replica --config broker.yaml
+pgsql-broker db enable replica --config broker.yaml
+pgsql-broker db remove replica --yes --config broker.yaml
+```
+
+`db add` accepts flags for every database setting. If a setting is omitted in
+interactive mode, it is pre-filled from the last database in the file, or
+from the instance named by `--from`; press Enter to keep that value. Use
+`--non-interactive` to reject missing required values instead of prompting.
+The add command asks for confirmation unless `--yes` is supplied. Remove also
+requires `--yes` when stdin is not a terminal.
 
 ### Broker Settings
 
@@ -383,8 +404,12 @@ Global settings applied to all database instances:
 | `worker_idle_timeout_sec` | Worker idle timeout | `10` |
 | `notify_retry_seconds` | NOTIFY retry interval | `30s` |
 | `enable_debug` | Enable debug logging | `false` |
-| `lease_seconds` | Job lease duration before reclaimable | - |
-| `stale_job_recovery_sec` | Interval for reclaiming expired leases | - |
+| `lease_seconds` | Job lease duration before reclaimable | `60` |
+| `stale_job_recovery_sec` | Interval for reclaiming expired leases | `30` |
+| `metrics_enabled` | Enable Prometheus and the embedded dashboard | `true` |
+| `metrics_host` | Metrics HTTP bind host | `0.0.0.0` |
+| `metrics_port` | Metrics HTTP bind port | `9469` |
+| `queue_depth_poll_sec` | Queue depth collection interval | `15` |
 
 ## Development
 

broker.example.yaml

@@ -41,6 +41,10 @@ broker:
   worker_idle_timeout_sec: 10  # Worker idle timeout
   notify_retry_seconds: 30s  # LISTEN/NOTIFY retry interval
   enable_debug: false  # Enable debug logging
+  metrics_enabled: true # Expose Prometheus and the embedded dashboard
+  metrics_host: 0.0.0.0
+  metrics_port: 9469
+  queue_depth_poll_sec: 15
 
 # Logging settings
 logging:

cmd/broker/db.go

@@ -0,0 +1,513 @@
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+	"golang.org/x/term"
+
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/config"
+)
+
+// Flags shared by the `db` subcommands.
+var (
+	dbConfigPath string
+
+	dbAddFrom            string
+	dbAddName            string
+	dbAddHost            string
+	dbAddPort            int
+	dbAddDatabase        string
+	dbAddUser            string
+	dbAddPassword        string
+	dbAddSSLMode         string
+	dbAddMaxOpenConns    int
+	dbAddMaxIdleConns    int
+	dbAddConnMaxLifetime string
+	dbAddConnMaxIdleTime string
+	dbAddQueueCount      int
+	dbAddTenantID        string
+	dbAddAutoMigrate     bool
+	dbAddDisabled        bool
+	dbAddNonInteractive  bool
+	dbYes                bool
+)
+
+var dbCmd = &cobra.Command{
+	Use:   "db",
+	Short: "Manage databases in the broker config file",
+	Long:  `Add, remove, enable, disable, or list the databases configured in the broker config file.`,
+}
+
+var dbAddCmd = &cobra.Command{
+	Use:   "add [name]",
+	Short: "Add a database to the config file",
+	Long: `Add a database entry to the config file.
+
+Every field can be supplied via flags for fully non-interactive use. Any
+field left unset falls back to an interactive prompt (when stdin is a
+terminal) that is pre-filled with the value from --from (a named existing
+instance) or, if --from is not given, the last database currently in the
+file. Press Enter at a prompt to accept the shown value.`,
+	Args: cobra.MaximumNArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		if len(args) == 1 && dbAddName == "" {
+			dbAddName = args[0]
+		}
+		return runDBAdd(cmd)
+	},
+}
+
+var dbRemoveCmd = &cobra.Command{
+	Use:   "remove <name>",
+	Short: "Remove a database from the config file",
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runDBRemove(args[0])
+	},
+}
+
+var dbEnableCmd = &cobra.Command{
+	Use:   "enable <name>",
+	Short: "Enable a database (removes the disabled flag)",
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runDBSetDisabled(args[0], false)
+	},
+}
+
+var dbDisableCmd = &cobra.Command{
+	Use:   "disable <name>",
+	Short: "Disable a database (excluded from `start` until re-enabled)",
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runDBSetDisabled(args[0], true)
+	},
+}
+
+var dbListCmd = &cobra.Command{
+	Use:   "list",
+	Short: "List databases configured in the config file",
+	Args:  cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runDBList()
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(dbCmd)
+	dbCmd.AddCommand(dbAddCmd, dbRemoveCmd, dbEnableCmd, dbDisableCmd, dbListCmd)
+
+	dbCmd.PersistentFlags().StringVar(&dbConfigPath, "config", "broker.yaml", "config file to edit")
+
+	dbAddCmd.Flags().StringVar(&dbAddFrom, "from", "", "prime defaults from this existing database name (default: last database in the file)")
+	dbAddCmd.Flags().StringVar(&dbAddName, "name", "", "unique name for the new database (required)")
+	dbAddCmd.Flags().StringVar(&dbAddHost, "host", "", "PostgreSQL host")
+	dbAddCmd.Flags().IntVar(&dbAddPort, "port", 0, "PostgreSQL port (default 5432)")
+	dbAddCmd.Flags().StringVar(&dbAddDatabase, "database", "", "database name")
+	dbAddCmd.Flags().StringVar(&dbAddUser, "user", "", "database user")
+	dbAddCmd.Flags().StringVar(&dbAddPassword, "password", "", "database password")
+	dbAddCmd.Flags().StringVar(&dbAddSSLMode, "sslmode", "", "SSL mode (disable, require, verify-ca, verify-full)")
+	dbAddCmd.Flags().IntVar(&dbAddMaxOpenConns, "max-open-conns", 0, "max open connections (default 25)")
+	dbAddCmd.Flags().IntVar(&dbAddMaxIdleConns, "max-idle-conns", 0, "max idle connections (default 5)")
+	dbAddCmd.Flags().StringVar(&dbAddConnMaxLifetime, "conn-max-lifetime", "", "connection max lifetime (e.g. 5m, default 5m)")
+	dbAddCmd.Flags().StringVar(&dbAddConnMaxIdleTime, "conn-max-idle-time", "", "connection max idle time (e.g. 10m, default 10m)")
+	dbAddCmd.Flags().IntVar(&dbAddQueueCount, "queue-count", 0, "number of concurrent queues (default 4)")
+	dbAddCmd.Flags().StringVar(&dbAddTenantID, "tenant-id", "", "RLS tenant id (default \"default\")")
+	dbAddCmd.Flags().BoolVar(&dbAddAutoMigrate, "auto-migrate", false, "apply pending migrations automatically on connect")
+	dbAddCmd.Flags().BoolVar(&dbAddDisabled, "disabled", false, "add the database in a disabled state")
+	dbAddCmd.Flags().BoolVar(&dbAddNonInteractive, "non-interactive", false, "fail instead of prompting for missing fields")
+	dbAddCmd.Flags().BoolVarP(&dbYes, "yes", "y", false, "skip the confirmation prompt")
+
+	dbRemoveCmd.Flags().BoolVarP(&dbYes, "yes", "y", false, "skip the confirmation prompt")
+}
+
+// runDBAdd builds a new DatabaseConfig by layering (in priority order)
+// explicit flags, then interactive prompts primed from --from / the last
+// database in the file, then falls back to leaving optional fields unset
+// so config.applyDatabaseDefaults keeps handling defaults at load time.
+func runDBAdd(cmd *cobra.Command) error {
+	doc, err := config.LoadFileDoc(dbConfigPath)
+	if err != nil {
+		return err
+	}
+
+	var prime config.DatabaseConfig
+	if dbAddFrom != "" {
+		found, ok, err := doc.FindDatabase(dbAddFrom)
+		if err != nil {
+			return err
+		}
+		if !ok {
+			return fmt.Errorf("--from instance %q not found in %s", dbAddFrom, dbConfigPath)
+		}
+		prime = found
+	} else if last, ok, err := doc.LastDatabase(); err != nil {
+		return err
+	} else if ok {
+		prime = last
+	}
+	// Never prime the new entry's name from an existing one.
+	prime.Name = ""
+
+	interactive := !dbAddNonInteractive && term.IsTerminal(int(os.Stdin.Fd()))
+	reader := bufio.NewReader(os.Stdin)
+
+	if interactive {
+		fmt.Fprintln(os.Stderr, "Adding a new database. Press Enter to accept the shown value.")
+	}
+
+	db := config.DatabaseConfig{}
+
+	if dbAddName != "" {
+		db.Name = dbAddName
+	} else {
+		db.Name, err = resolveStringField(cmd, reader, interactive, "name", "", true, dbAddName)
+		if err != nil {
+			return err
+		}
+	}
+	if db.Name == "" {
+		return fmt.Errorf("name is required (--name, a positional argument, or run interactively)")
+	}
+	db.Host, err = resolveStringField(cmd, reader, interactive, "host", prime.Host, true, dbAddHost)
+	if err != nil {
+		return err
+	}
+	db.Database, err = resolveStringField(cmd, reader, interactive, "database", prime.Database, true, dbAddDatabase)
+	if err != nil {
+		return err
+	}
+	db.User, err = resolveStringField(cmd, reader, interactive, "user", prime.User, true, dbAddUser)
+	if err != nil {
+		return err
+	}
+	db.Password, err = resolveStringField(cmd, reader, interactive, "password", prime.Password, false, dbAddPassword)
+	if err != nil {
+		return err
+	}
+	db.SSLMode, err = resolveStringField(cmd, reader, interactive, "sslmode", prime.SSLMode, false, dbAddSSLMode)
+	if err != nil {
+		return err
+	}
+	db.TenantID, err = resolveStringField(cmd, reader, interactive, "tenant_id", prime.TenantID, false, dbAddTenantID)
+	if err != nil {
+		return err
+	}
+
+	port, err := resolveIntField(cmd, reader, interactive, "port", prime.Port, "port", dbAddPort)
+	if err != nil {
+		return err
+	}
+	db.Port = port
+
+	maxOpen, err := resolveIntField(cmd, reader, interactive, "max_open_conns", prime.MaxOpenConns, "max-open-conns", dbAddMaxOpenConns)
+	if err != nil {
+		return err
+	}
+	db.MaxOpenConns = maxOpen
+
+	maxIdle, err := resolveIntField(cmd, reader, interactive, "max_idle_conns", prime.MaxIdleConns, "max-idle-conns", dbAddMaxIdleConns)
+	if err != nil {
+		return err
+	}
+	db.MaxIdleConns = maxIdle
+
+	queueCount, err := resolveIntField(cmd, reader, interactive, "queue_count", prime.QueueCount, "queue-count", dbAddQueueCount)
+	if err != nil {
+		return err
+	}
+	db.QueueCount = queueCount
+
+	connMaxLifetime, err := resolveDurationField(cmd, reader, interactive, "conn_max_lifetime", prime.ConnMaxLifetime, "conn-max-lifetime", dbAddConnMaxLifetime)
+	if err != nil {
+		return err
+	}
+	db.ConnMaxLifetime = connMaxLifetime
+
+	connMaxIdleTime, err := resolveDurationField(cmd, reader, interactive, "conn_max_idle_time", prime.ConnMaxIdleTime, "conn-max-idle-time", dbAddConnMaxIdleTime)
+	if err != nil {
+		return err
+	}
+	db.ConnMaxIdleTime = connMaxIdleTime
+
+	if cmd.Flags().Changed("auto-migrate") {
+		db.AutoMigrate = dbAddAutoMigrate
+	} else {
+		db.AutoMigrate = prime.AutoMigrate
+	}
+	if cmd.Flags().Changed("disabled") {
+		db.Disabled = dbAddDisabled
+	} else {
+		db.Disabled = prime.Disabled
+	}
+
+	for field, value := range map[string]string{
+		"host":     db.Host,
+		"database": db.Database,
+		"user":     db.User,
+	} {
+		if strings.TrimSpace(value) == "" {
+			return fmt.Errorf("%s is required (--%s or an interactive value)", field, strings.ReplaceAll(field, "_", "-"))
+		}
+	}
+
+	if err := doc.AddDatabase(db); err != nil {
+		return err
+	}
+
+	if !dbYes && interactive {
+		fmt.Fprintf(os.Stderr, "\nAbout to add database %q to %s:\n", db.Name, dbConfigPath)
+		printDBSummary(os.Stderr, db)
+		fmt.Fprint(os.Stderr, "Proceed? [Y/n]: ")
+		line, _ := reader.ReadString('\n')
+		line = strings.TrimSpace(strings.ToLower(line))
+		if line == "n" || line == "no" {
+			fmt.Println("Aborted; config file not modified.")
+			return nil
+		}
+	}
+
+	if err := doc.Save(); err != nil {
+		return err
+	}
+	fmt.Printf("Added database %q to %s\n", db.Name, dbConfigPath)
+	return nil
+}
+
+func runDBRemove(name string) error {
+	doc, err := config.LoadFileDoc(dbConfigPath)
+	if err != nil {
+		return err
+	}
+	db, ok, err := doc.FindDatabase(name)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return fmt.Errorf("database %q not found in %s", name, dbConfigPath)
+	}
+
+	if !dbYes {
+		if !term.IsTerminal(int(os.Stdin.Fd())) {
+			return fmt.Errorf("refusing to remove %q non-interactively without --yes", name)
+		}
+		fmt.Fprintf(os.Stderr, "Remove database %q (%s@%s:%d/%s) from %s? [y/N]: ",
+			name, db.User, db.Host, db.Port, db.Database, dbConfigPath)
+		reader := bufio.NewReader(os.Stdin)
+		line, _ := reader.ReadString('\n')
+		line = strings.TrimSpace(strings.ToLower(line))
+		if line != "y" && line != "yes" {
+			fmt.Println("Aborted; config file not modified.")
+			return nil
+		}
+	}
+
+	if _, err := doc.RemoveDatabase(name); err != nil {
+		return err
+	}
+	if err := doc.Save(); err != nil {
+		return err
+	}
+	fmt.Printf("Removed database %q from %s\n", name, dbConfigPath)
+	return nil
+}
+
+func runDBSetDisabled(name string, disabled bool) error {
+	doc, err := config.LoadFileDoc(dbConfigPath)
+	if err != nil {
+		return err
+	}
+	found, err := doc.SetDisabled(name, disabled)
+	if err != nil {
+		return err
+	}
+	if !found {
+		return fmt.Errorf("database %q not found in %s", name, dbConfigPath)
+	}
+	if err := doc.Save(); err != nil {
+		return err
+	}
+	verb := "Enabled"
+	if disabled {
+		verb = "Disabled"
+	}
+	fmt.Printf("%s database %q in %s\n", verb, name, dbConfigPath)
+	return nil
+}
+
+func runDBList() error {
+	doc, err := config.LoadFileDoc(dbConfigPath)
+	if err != nil {
+		return err
+	}
+	dbs, err := doc.ListDatabases()
+	if err != nil {
+		return err
+	}
+	if len(dbs) == 0 {
+		fmt.Printf("No databases configured in %s\n", dbConfigPath)
+		return nil
+	}
+	fmt.Printf("%-20s %-20s %-6s %-16s %-16s %-10s\n", "NAME", "HOST", "PORT", "DATABASE", "USER", "STATUS")
+	for i := range dbs {
+		db := &dbs[i]
+		status := "enabled"
+		if db.Disabled {
+			status = "disabled"
+		}
+		port := db.Port
+		if port == 0 {
+			port = 5432
+		}
+		fmt.Printf("%-20s %-20s %-6d %-16s %-16s %-10s\n", db.Name, db.Host, port, db.Database, db.User, status)
+	}
+	return nil
+}
+
+func printDBSummary(w *os.File, db config.DatabaseConfig) {
+	password := ""
+	if db.Password != "" {
+		password = "(set)"
+	}
+	fmt.Fprintf(w, "  host:              %s\n", db.Host)
+	fmt.Fprintf(w, "  port:              %s\n", intOrDefault(db.Port, 5432))
+	fmt.Fprintf(w, "  database:          %s\n", db.Database)
+	fmt.Fprintf(w, "  user:              %s\n", db.User)
+	fmt.Fprintf(w, "  password:          %s\n", password)
+	fmt.Fprintf(w, "  sslmode:           %s\n", stringOrDefault(db.SSLMode, "disable"))
+	fmt.Fprintf(w, "  max_open_conns:    %s\n", intOrDefault(db.MaxOpenConns, 25))
+	fmt.Fprintf(w, "  max_idle_conns:    %s\n", intOrDefault(db.MaxIdleConns, 5))
+	fmt.Fprintf(w, "  conn_max_lifetime: %s\n", durationOrDefault(db.ConnMaxLifetime, 5*time.Minute))
+	fmt.Fprintf(w, "  conn_max_idle_time:%s\n", durationOrDefault(db.ConnMaxIdleTime, 10*time.Minute))
+	fmt.Fprintf(w, "  queue_count:       %s\n", intOrDefault(db.QueueCount, 4))
+	fmt.Fprintf(w, "  tenant_id:         %s\n", stringOrDefault(db.TenantID, "default"))
+	fmt.Fprintf(w, "  auto_migrate:      %t\n", db.AutoMigrate)
+	fmt.Fprintf(w, "  disabled:          %t\n", db.Disabled)
+}
+
+func intOrDefault(v, def int) string {
+	if v == 0 {
+		return fmt.Sprintf("%d (default)", def)
+	}
+	return strconv.Itoa(v)
+}
+
+func stringOrDefault(v, def string) string {
+	if v == "" {
+		return fmt.Sprintf("%s (default)", def)
+	}
+	return v
+}
+
+func durationOrDefault(v, def time.Duration) string {
+	if v == 0 {
+		return fmt.Sprintf("%s (default)", def)
+	}
+	return v.String()
+}
+
+// resolveStringField returns, in priority order: the flag value if the
+// flag was explicitly set, otherwise an interactive prompt (pre-filled
+// with defaultVal) when interactive, otherwise defaultVal. When required
+// and still empty after all of that, an error is returned by the caller.
+func resolveStringField(cmd *cobra.Command, reader *bufio.Reader, interactive bool, label, defaultVal string, required bool, flagVal string) (string, error) {
+	flagName := flagNameFor(label)
+	if cmd.Flags().Changed(flagName) {
+		return flagVal, nil
+	}
+	if !interactive {
+		return defaultVal, nil
+	}
+	for {
+		fmt.Fprintf(os.Stderr, "%s [%s]: ", label, defaultVal)
+		line, _ := reader.ReadString('\n')
+		line = strings.TrimSpace(line)
+		if line == "" {
+			line = defaultVal
+		}
+		if line == "" && required {
+			fmt.Fprintln(os.Stderr, "value is required")
+			continue
+		}
+		return line, nil
+	}
+}
+
+func resolveIntField(cmd *cobra.Command, reader *bufio.Reader, interactive bool, label string, defaultVal int, flagName string, flagVal int) (int, error) {
+	if cmd.Flags().Changed(flagName) {
+		return flagVal, nil
+	}
+	if !interactive {
+		return defaultVal, nil
+	}
+	for {
+		fmt.Fprintf(os.Stderr, "%s [%d]: ", label, defaultVal)
+		line, _ := reader.ReadString('\n')
+		line = strings.TrimSpace(line)
+		if line == "" {
+			return defaultVal, nil
+		}
+		n, err := strconv.Atoi(line)
+		if err != nil {
+			fmt.Fprintln(os.Stderr, "must be a whole number")
+			continue
+		}
+		return n, nil
+	}
+}
+
+func resolveDurationField(cmd *cobra.Command, reader *bufio.Reader, interactive bool, label string, defaultVal time.Duration, flagName, flagVal string) (time.Duration, error) {
+	if cmd.Flags().Changed(flagName) {
+		d, err := time.ParseDuration(flagVal)
+		if err != nil {
+			return 0, fmt.Errorf("--%s: %w", flagName, err)
+		}
+		return d, nil
+	}
+	if !interactive {
+		return defaultVal, nil
+	}
+	for {
+		fmt.Fprintf(os.Stderr, "%s [%s]: ", label, defaultVal)
+		line, _ := reader.ReadString('\n')
+		line = strings.TrimSpace(line)
+		if line == "" {
+			return defaultVal, nil
+		}
+		d, err := time.ParseDuration(line)
+		if err != nil {
+			fmt.Fprintln(os.Stderr, "must be a duration like 5m, 30s")
+			continue
+		}
+		return d, nil
+	}
+}
+
+// flagNameFor maps a prompt label to its cobra flag name for the string
+// fields prompted in runDBAdd.
+func flagNameFor(label string) string {
+	switch label {
+	case "name":
+		return "name"
+	case "host":
+		return "host"
+	case "database":
+		return "database"
+	case "user":
+		return "user"
+	case "password":
+		return "password"
+	case "sslmode":
+		return "sslmode"
+	case "tenant_id":
+		return "tenant-id"
+	default:
+		return label
+	}
+}

go.mod

@@ -4,19 +4,28 @@ go 1.26.0
 
 require (
 	github.com/lib/pq v1.10.9
+	github.com/prometheus/client_golang v1.20.5
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/viper v1.21.0
 	github.com/stretchr/testify v1.11.1
 	golang.org/x/term v0.46.0
+	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.6.1 // indirect
+	github.com/prometheus/common v0.55.0 // indirect
+	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/sagikazarmark/locafero v0.11.0 // indirect
 	github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect
 	github.com/spf13/afero v1.15.0 // indirect
@@ -26,5 +35,5 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/sys v0.48.0 // indirect
 	golang.org/x/text v0.28.0 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
 )

go.sum

@@ -1,3 +1,7 @@
+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/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -11,18 +15,32 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
-github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
+github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
+github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
+github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
+github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc=
+github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
+github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
+github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
 github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik=
@@ -51,8 +69,10 @@ golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
 golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc=
 golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng=
 golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU=
+google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
+google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/broker/broker.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/config"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 )
 
 // Broker manages multiple database instances
@@ -17,6 +19,8 @@ type Broker struct {
 	instances []*DatabaseInstance
 	ctx       context.Context
 	cancel    context.CancelFunc
+	metrics   *metrics.Metrics
+	server    *metrics.Server
 	shutdown  bool
 	mu        sync.RWMutex
 }
@@ -33,6 +37,16 @@ func New(cfg *config.Config, logger adapter.Logger, version string) (*Broker, er
 		ctx:       ctx,
 		cancel:    cancel,
 	}
+	if cfg.Broker.MetricsEnabled {
+		broker.metrics = metrics.New()
+		broker.metrics.SetDatabaseCount(len(cfg.Databases))
+		addr := fmt.Sprintf("%s:%d", cfg.Broker.MetricsHost, cfg.Broker.MetricsPort)
+		server, err := metrics.NewServer(broker.metrics, addr, broker.logger)
+		if err != nil {
+			return nil, err
+		}
+		broker.server = server
+	}
 
 	return broker, nil
 }
@@ -41,16 +55,20 @@ func New(cfg *config.Config, logger adapter.Logger, version string) (*Broker, er
 func (b *Broker) Start() error {
 	b.logger.Info("starting broker", "database_count", len(b.config.Databases))
 
-	// Create and start an instance for each database
+	// Create and start an instance for each enabled database
 	for i := range b.config.Databases {
 		dbCfg := &b.config.Databases[i]
+		if dbCfg.Disabled {
+			b.logger.Info("skipping disabled database", "name", dbCfg.Name)
+			continue
+		}
 		b.logger.Info("starting database instance", "name", dbCfg.Name, "host", dbCfg.Host, "database", dbCfg.Database)
 
 		// Create database adapter
 		dbAdapter := adapter.NewPostgresAdapter(dbCfg.ToPostgresConfig(), b.logger)
 
 		// Create database instance
-		instance, err := NewDatabaseInstance(b.config, dbCfg, dbAdapter, b.logger, b.version, b.ctx)
+		instance, err := NewDatabaseInstance(b.config, dbCfg, dbAdapter, b.logger, b.version, b.ctx, b.metrics)
 		if err != nil {
 			// Stop any already-started instances
 			b.stopInstances()
@@ -68,7 +86,14 @@ func (b *Broker) Start() error {
 		b.logger.Info("database instance started", "name", dbCfg.Name, "instance_id", instance.ID)
 	}
 
+	if len(b.instances) == 0 {
+		return fmt.Errorf("no enabled databases configured (all %d database(s) are disabled)", len(b.config.Databases))
+	}
+
 	b.logger.Info("broker started successfully", "database_instances", len(b.instances))
+	if b.server != nil {
+		b.server.Start()
+	}
 	return nil
 }
 
@@ -89,6 +114,13 @@ func (b *Broker) Stop() error {
 
 	// Stop all instances
 	b.stopInstances()
+	if b.server != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := b.server.Stop(ctx); err != nil {
+			b.logger.Error("failed to stop metrics server", "error", err)
+		}
+	}
 
 	b.logger.Info("broker stopped")
 	return nil

pkg/broker/config/config.go

@@ -39,6 +39,10 @@ type DatabaseConfig struct {
 	// fast if the schema is behind, naming the missing migrations and
 	// pointing at `pgsql-broker install`.
 	AutoMigrate bool `mapstructure:"auto_migrate"`
+	// Disabled, when true, excludes this database from `start` (no
+	// instance is created for it). Defaults to false so existing config
+	// files without this field are unaffected.
+	Disabled bool `mapstructure:"disabled"`
 }
 
 // BrokerConfig holds broker-specific settings
@@ -55,6 +59,16 @@ type BrokerConfig struct {
 	LeaseSeconds int `mapstructure:"lease_seconds"`
 	// StaleJobRecoverySec is the interval between broker_recover_stale_jobs sweeps.
 	StaleJobRecoverySec int `mapstructure:"stale_job_recovery_sec"`
+	// MetricsEnabled controls whether the embedded Prometheus metrics HTTP
+	// server (exposition endpoint + HTML dashboard) is started.
+	MetricsEnabled bool `mapstructure:"metrics_enabled"`
+	// MetricsHost is the bind address for the metrics server.
+	MetricsHost string `mapstructure:"metrics_host"`
+	// MetricsPort is the bind port for the metrics server.
+	MetricsPort int `mapstructure:"metrics_port"`
+	// QueueDepthPollSec is the interval between polls of pending job counts
+	// used to populate the broker_jobs_queued gauge.
+	QueueDepthPollSec int `mapstructure:"queue_depth_poll_sec"`
 }
 
 // LoggingConfig holds logging settings
@@ -120,6 +134,10 @@ func setDefaults(v *viper.Viper) {
 	v.SetDefault("broker.enable_debug", false)
 	v.SetDefault("broker.lease_seconds", 60)
 	v.SetDefault("broker.stale_job_recovery_sec", 30)
+	v.SetDefault("broker.metrics_enabled", true)
+	v.SetDefault("broker.metrics_host", "0.0.0.0")
+	v.SetDefault("broker.metrics_port", 9469)
+	v.SetDefault("broker.queue_depth_poll_sec", 15)
 
 	// Logging defaults
 	v.SetDefault("logging.level", "info")

pkg/broker/config/dbedit.go

@@ -0,0 +1,337 @@
+package config
+
+import (
+	"fmt"
+	"os"
+	"time"
+
+	"gopkg.in/yaml.v3"
+)
+
+// FileDoc wraps the parsed YAML node tree of a broker config file so that
+// database entries can be added, removed, or toggled while leaving the
+// rest of the file (formatting, comments, unrelated keys) intact.
+type FileDoc struct {
+	path string
+	root *yaml.Node
+}
+
+// databaseFileConfig mirrors DatabaseConfig with duration fields represented
+// as strings, which is the form used by the human-edited YAML config. YAML
+// unmarshalling does not parse strings into time.Duration automatically.
+type databaseFileConfig struct {
+	Name            string `yaml:"name"`
+	Host            string `yaml:"host"`
+	Port            int    `yaml:"port"`
+	Database        string `yaml:"database"`
+	User            string `yaml:"user"`
+	Password        string `yaml:"password"`
+	SSLMode         string `yaml:"sslmode"`
+	MaxOpenConns    int    `yaml:"max_open_conns"`
+	MaxIdleConns    int    `yaml:"max_idle_conns"`
+	ConnMaxLifetime string `yaml:"conn_max_lifetime"`
+	ConnMaxIdleTime string `yaml:"conn_max_idle_time"`
+	QueueCount      int    `yaml:"queue_count"`
+	TenantID        string `yaml:"tenant_id"`
+	AutoMigrate     bool   `yaml:"auto_migrate"`
+	Disabled        bool   `yaml:"disabled"`
+}
+
+func decodeDatabase(node *yaml.Node) (DatabaseConfig, error) {
+	var raw databaseFileConfig
+	if err := node.Decode(&raw); err != nil {
+		return DatabaseConfig{}, err
+	}
+	db := DatabaseConfig{
+		Name: raw.Name, Host: raw.Host, Port: raw.Port, Database: raw.Database,
+		User: raw.User, Password: raw.Password, SSLMode: raw.SSLMode,
+		MaxOpenConns: raw.MaxOpenConns, MaxIdleConns: raw.MaxIdleConns,
+		QueueCount: raw.QueueCount, TenantID: raw.TenantID,
+		AutoMigrate: raw.AutoMigrate, Disabled: raw.Disabled,
+	}
+	var err error
+	if raw.ConnMaxLifetime != "" {
+		db.ConnMaxLifetime, err = time.ParseDuration(raw.ConnMaxLifetime)
+		if err != nil {
+			return DatabaseConfig{}, fmt.Errorf("conn_max_lifetime: %w", err)
+		}
+	}
+	if raw.ConnMaxIdleTime != "" {
+		db.ConnMaxIdleTime, err = time.ParseDuration(raw.ConnMaxIdleTime)
+		if err != nil {
+			return DatabaseConfig{}, fmt.Errorf("conn_max_idle_time: %w", err)
+		}
+	}
+	return db, nil
+}
+
+// LoadFileDoc reads and parses the YAML config file at path for editing.
+// A missing file is treated as an empty document so `db add` can be used
+// to create a new config file from scratch.
+func LoadFileDoc(path string) (*FileDoc, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			return nil, fmt.Errorf("failed to read config file %s: %w", path, err)
+		}
+		data = nil
+	}
+
+	var root yaml.Node
+	if len(data) > 0 {
+		if err := yaml.Unmarshal(data, &root); err != nil {
+			return nil, fmt.Errorf("failed to parse config file %s: %w", path, err)
+		}
+	}
+	if root.Kind == 0 {
+		root.Kind = yaml.DocumentNode
+		root.Content = []*yaml.Node{{Kind: yaml.MappingNode, Tag: "!!map"}}
+	}
+
+	return &FileDoc{path: path, root: &root}, nil
+}
+
+// Save writes the document back to its original path.
+func (f *FileDoc) Save() error {
+	data, err := yaml.Marshal(f.root)
+	if err != nil {
+		return fmt.Errorf("failed to marshal config: %w", err)
+	}
+	if err := os.WriteFile(f.path, data, 0o644); err != nil {
+		return fmt.Errorf("failed to write config file %s: %w", f.path, err)
+	}
+	return nil
+}
+
+func (f *FileDoc) mappingRoot() (*yaml.Node, error) {
+	if len(f.root.Content) == 0 {
+		return nil, fmt.Errorf("config file is empty")
+	}
+	m := f.root.Content[0]
+	if m.Kind != yaml.MappingNode {
+		return nil, fmt.Errorf("config file root is not a mapping")
+	}
+	return m, nil
+}
+
+// databasesSeq returns the "databases" sequence node, creating it (and the
+// key) if it isn't already present.
+func (f *FileDoc) databasesSeq() (*yaml.Node, error) {
+	m, err := f.mappingRoot()
+	if err != nil {
+		return nil, err
+	}
+	for i := 0; i+1 < len(m.Content); i += 2 {
+		if m.Content[i].Value == "databases" {
+			seq := m.Content[i+1]
+			if seq.Kind != yaml.SequenceNode {
+				return nil, fmt.Errorf("'databases' key in config is not a list")
+			}
+			return seq, nil
+		}
+	}
+	keyNode := &yaml.Node{Kind: yaml.ScalarNode, Value: "databases"}
+	seqNode := &yaml.Node{Kind: yaml.SequenceNode, Tag: "!!seq"}
+	m.Content = append(m.Content, keyNode, seqNode)
+	return seqNode, nil
+}
+
+// ListDatabases decodes all database entries currently in the document, in
+// file order.
+func (f *FileDoc) ListDatabases() ([]DatabaseConfig, error) {
+	seq, err := f.databasesSeq()
+	if err != nil {
+		return nil, err
+	}
+	dbs := make([]DatabaseConfig, 0, len(seq.Content))
+	for _, item := range seq.Content {
+		db, err := decodeDatabase(item)
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode database entry: %w", err)
+		}
+		dbs = append(dbs, db)
+	}
+	return dbs, nil
+}
+
+// FindDatabase returns the decoded entry with the given name.
+func (f *FileDoc) FindDatabase(name string) (DatabaseConfig, bool, error) {
+	dbs, err := f.ListDatabases()
+	if err != nil {
+		return DatabaseConfig{}, false, err
+	}
+	for i := range dbs {
+		db := &dbs[i]
+		if db.Name == name {
+			return *db, true, nil
+		}
+	}
+	return DatabaseConfig{}, false, nil
+}
+
+// LastDatabase returns the last database entry in the file, used to prime
+// defaults for a new `db add` when no --from instance is given.
+func (f *FileDoc) LastDatabase() (DatabaseConfig, bool, error) {
+	dbs, err := f.ListDatabases()
+	if err != nil {
+		return DatabaseConfig{}, false, err
+	}
+	if len(dbs) == 0 {
+		return DatabaseConfig{}, false, nil
+	}
+	return dbs[len(dbs)-1], true, nil
+}
+
+// AddDatabase appends a new database entry. It fails if an entry with the
+// same name already exists.
+func (f *FileDoc) AddDatabase(db DatabaseConfig) error {
+	seq, err := f.databasesSeq()
+	if err != nil {
+		return err
+	}
+	for _, item := range seq.Content {
+		existing, err := decodeDatabase(item)
+		if err != nil {
+			return fmt.Errorf("failed to decode database entry: %w", err)
+		}
+		if existing.Name == db.Name {
+			return fmt.Errorf("a database named %q already exists", db.Name)
+		}
+	}
+	seq.Content = append(seq.Content, dbToNode(db))
+	return nil
+}
+
+// RemoveDatabase deletes the entry with the given name, reporting whether
+// it was found.
+func (f *FileDoc) RemoveDatabase(name string) (bool, error) {
+	seq, err := f.databasesSeq()
+	if err != nil {
+		return false, err
+	}
+	for i, item := range seq.Content {
+		existing, err := decodeDatabase(item)
+		if err != nil {
+			return false, fmt.Errorf("failed to decode database entry: %w", err)
+		}
+		if existing.Name == name {
+			seq.Content = append(seq.Content[:i], seq.Content[i+1:]...)
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// SetDisabled sets (or clears) the disabled flag on the named entry,
+// reporting whether the entry was found.
+func (f *FileDoc) SetDisabled(name string, disabled bool) (bool, error) {
+	seq, err := f.databasesSeq()
+	if err != nil {
+		return false, err
+	}
+	for _, item := range seq.Content {
+		existing, err := decodeDatabase(item)
+		if err != nil {
+			return false, fmt.Errorf("failed to decode database entry: %w", err)
+		}
+		if existing.Name != name {
+			continue
+		}
+		if disabled {
+			setMappingKey(item, "disabled", &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!bool", Value: "true"})
+		} else {
+			removeMappingKey(item, "disabled")
+		}
+		return true, nil
+	}
+	return false, nil
+}
+
+// setMappingKey sets key to value within a mapping node, appending the
+// pair if the key isn't already present.
+func setMappingKey(m *yaml.Node, key string, value *yaml.Node) {
+	for i := 0; i+1 < len(m.Content); i += 2 {
+		if m.Content[i].Value == key {
+			m.Content[i+1] = value
+			return
+		}
+	}
+	m.Content = append(m.Content, &yaml.Node{Kind: yaml.ScalarNode, Value: key}, value)
+}
+
+// removeMappingKey removes key from a mapping node if present.
+func removeMappingKey(m *yaml.Node, key string) {
+	for i := 0; i+1 < len(m.Content); i += 2 {
+		if m.Content[i].Value == key {
+			m.Content = append(m.Content[:i], m.Content[i+2:]...)
+			return
+		}
+	}
+}
+
+// dbToNode builds a YAML mapping node for a database entry, omitting
+// fields left at their zero value so the on-disk defaults from
+// applyDatabaseDefaults keep applying (matching broker.example.yaml
+// style). Field order mirrors DatabaseConfig / broker.example.yaml.
+func dbToNode(db DatabaseConfig) *yaml.Node {
+	m := &yaml.Node{Kind: yaml.MappingNode, Tag: "!!map"}
+
+	add := func(key, value string) {
+		m.Content = append(m.Content,
+			&yaml.Node{Kind: yaml.ScalarNode, Value: key},
+			&yaml.Node{Kind: yaml.ScalarNode, Value: value},
+		)
+	}
+	addBool := func(key string, value bool) {
+		m.Content = append(m.Content,
+			&yaml.Node{Kind: yaml.ScalarNode, Value: key},
+			&yaml.Node{Kind: yaml.ScalarNode, Tag: "!!bool", Value: fmt.Sprintf("%t", value)},
+		)
+	}
+	addInt := func(key string, value int) {
+		m.Content = append(m.Content,
+			&yaml.Node{Kind: yaml.ScalarNode, Value: key},
+			&yaml.Node{Kind: yaml.ScalarNode, Tag: "!!int", Value: fmt.Sprintf("%d", value)},
+		)
+	}
+
+	add("name", db.Name)
+	add("host", db.Host)
+	if db.Port != 0 {
+		addInt("port", db.Port)
+	}
+	add("database", db.Database)
+	add("user", db.User)
+	if db.Password != "" {
+		add("password", db.Password)
+	}
+	if db.SSLMode != "" {
+		add("sslmode", db.SSLMode)
+	}
+	if db.MaxOpenConns != 0 {
+		addInt("max_open_conns", db.MaxOpenConns)
+	}
+	if db.MaxIdleConns != 0 {
+		addInt("max_idle_conns", db.MaxIdleConns)
+	}
+	if db.ConnMaxLifetime != 0 {
+		add("conn_max_lifetime", db.ConnMaxLifetime.String())
+	}
+	if db.ConnMaxIdleTime != 0 {
+		add("conn_max_idle_time", db.ConnMaxIdleTime.String())
+	}
+	if db.QueueCount != 0 {
+		addInt("queue_count", db.QueueCount)
+	}
+	if db.TenantID != "" {
+		add("tenant_id", db.TenantID)
+	}
+	if db.AutoMigrate {
+		addBool("auto_migrate", db.AutoMigrate)
+	}
+	if db.Disabled {
+		addBool("disabled", db.Disabled)
+	}
+
+	return m
+}

pkg/broker/config/dbedit_test.go

@@ -0,0 +1,102 @@
+package config
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFileDocDatabaseManagement(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "broker.yaml")
+	require.NoError(t, os.WriteFile(path, []byte(`broker:
+  name: test
+# keep this comment
+databases:
+  - name: primary
+    host: db.example
+    port: 5433
+    database: jobs
+    user: broker
+    password: secret
+    sslmode: verify-full
+    max_open_conns: 12
+    max_idle_conns: 4
+    conn_max_lifetime: 2m
+    conn_max_idle_time: 30s
+    queue_count: 3
+    tenant_id: tenant-a
+    auto_migrate: true
+`), 0o600))
+
+	doc, err := LoadFileDoc(path)
+	require.NoError(t, err)
+	primary, found, err := doc.FindDatabase("primary")
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Equal(t, 2*time.Minute, primary.ConnMaxLifetime)
+	require.Equal(t, 30*time.Second, primary.ConnMaxIdleTime)
+	require.True(t, primary.AutoMigrate)
+
+	copy := primary
+	copy.Name = "replica"
+	require.NoError(t, doc.AddDatabase(copy))
+	found, err = doc.SetDisabled("replica", true)
+	require.NoError(t, err)
+	require.True(t, found)
+	require.NoError(t, doc.Save())
+
+	doc, err = LoadFileDoc(path)
+	require.NoError(t, err)
+	replica, found, err := doc.FindDatabase("replica")
+	require.NoError(t, err)
+	require.True(t, found)
+	require.True(t, replica.Disabled)
+	require.NoError(t, func() error {
+		found, err := doc.SetDisabled("replica", false)
+		require.True(t, found)
+		return err
+	}())
+	require.NoError(t, doc.Save())
+
+	doc, err = LoadFileDoc(path)
+	require.NoError(t, err)
+	replica, found, err = doc.FindDatabase("replica")
+	require.NoError(t, err)
+	require.True(t, found)
+	require.False(t, replica.Disabled)
+
+	removed, err := doc.RemoveDatabase("primary")
+	require.NoError(t, err)
+	require.True(t, removed)
+	require.NoError(t, doc.Save())
+	contents, err := os.ReadFile(path)
+	require.NoError(t, err)
+	require.Contains(t, string(contents), "# keep this comment")
+
+	_, found, err = doc.FindDatabase("primary")
+	require.NoError(t, err)
+	require.False(t, found)
+}
+
+func TestFileDocMissingFileCanAddDatabase(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "new.yaml")
+	doc, err := LoadFileDoc(path)
+	require.NoError(t, err)
+	require.NoError(t, doc.AddDatabase(DatabaseConfig{
+		Name:     "new",
+		Host:     "localhost",
+		Database: "jobs",
+		User:     "broker",
+	}))
+	require.NoError(t, doc.Save())
+
+	reloaded, err := LoadFileDoc(path)
+	require.NoError(t, err)
+	dbs, err := reloaded.ListDatabases()
+	require.NoError(t, err)
+	require.Len(t, dbs, 1)
+	require.Equal(t, "new", dbs[0].Name)
+}

pkg/broker/database_instance.go

@@ -13,6 +13,7 @@ import (
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/config"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/install"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/models"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/queue"
 )
@@ -36,6 +37,7 @@ type DatabaseInstance struct {
 	shutdownMu   sync.RWMutex
 	jobsHandled  int64
 	startTime    time.Time
+	metrics      *metrics.Metrics
 
 	// sessionConn holds the pg_try_advisory_lock acquired by
 	// registerInstance. The lock is scoped to this one physical connection,
@@ -45,12 +47,16 @@ type DatabaseInstance struct {
 }
 
 // NewDatabaseInstance creates a new database instance
-func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db adapter.DBAdapter, logger adapter.Logger, version string, parentCtx context.Context) (*DatabaseInstance, error) {
+func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db adapter.DBAdapter, logger adapter.Logger, version string, parentCtx context.Context, brokerMetrics ...*metrics.Metrics) (*DatabaseInstance, error) {
 	hostname, err := os.Hostname()
 	if err != nil {
 		hostname = "unknown"
 	}
 
+	var instanceMetrics *metrics.Metrics
+	if len(brokerMetrics) > 0 {
+		instanceMetrics = brokerMetrics[0]
+	}
 	instance := &DatabaseInstance{
 		Name:         fmt.Sprintf("%s-%s", cfg.Broker.Name, dbCfg.Name),
 		DatabaseName: dbCfg.Name,
@@ -64,6 +70,7 @@ func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db ad
 		queues:       make(map[int]*queue.Queue),
 		ctx:          parentCtx,
 		startTime:    time.Now(),
+		metrics:      instanceMetrics,
 	}
 
 	return instance, nil
@@ -109,9 +116,54 @@ func (i *DatabaseInstance) Start() error {
 	adapter.SupervisedGo(i.logger, "stale-job-recovery-routine", i.staleJobRecoveryRoutine)
 
 	i.logger.Info("database instance started successfully")
+	if i.metrics != nil {
+		adapter.SupervisedGo(i.logger, "metrics-queue-depth-routine", i.queueDepthRoutine)
+	}
 	return nil
 }
 
+// queueDepthRoutine periodically exports pending jobs grouped by queue.
+func (i *DatabaseInstance) queueDepthRoutine() {
+	interval := time.Duration(i.config.Broker.QueueDepthPollSec) * time.Second
+	if interval <= 0 {
+		interval = 15 * time.Second
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	i.updateQueueDepthMetrics()
+	for {
+		select {
+		case <-ticker.C:
+			i.updateQueueDepthMetrics()
+		case <-i.ctx.Done():
+			return
+		}
+	}
+}
+
+func (i *DatabaseInstance) updateQueueDepthMetrics() {
+	rows, err := i.db.Query(i.ctx, "SELECT job_queue, COUNT(*) FROM broker.broker_jobs WHERE complete_status = 0 GROUP BY job_queue")
+	if err != nil {
+		i.logger.Warn("failed to collect queue depth metrics", "error", err)
+		return
+	}
+	defer rows.Close()
+	for queueNumber := 1; queueNumber <= i.dbConfig.QueueCount; queueNumber++ {
+		i.metrics.SetJobsQueued(i.DatabaseName, queueNumber, 0)
+	}
+	for rows.Next() {
+		var queueNumber, count int
+		if err := rows.Scan(&queueNumber, &count); err != nil {
+			i.logger.Warn("failed to scan queue depth metric", "error", err)
+			return
+		}
+		i.metrics.SetJobsQueued(i.DatabaseName, queueNumber, float64(count))
+	}
+	if err := rows.Err(); err != nil {
+		i.logger.Warn("failed to read queue depth metrics", "error", err)
+	}
+}
+
 // ensureSchema checks the embedded migration set against the database and,
 // depending on dbConfig.AutoMigrate, either applies pending migrations or
 // fails startup fast rather than running against a stale/missing schema.
@@ -261,6 +313,8 @@ func (i *DatabaseInstance) startQueues() error {
 			FetchSize:    i.config.Broker.FetchQueryQueSize,
 			TenantID:     i.dbConfig.TenantID,
 			LeaseSeconds: leaseSeconds,
+			Metrics:      i.metrics,
+			DatabaseName: i.DatabaseName,
 		}
 
 		q := queue.New(queueCfg)
@@ -271,6 +325,7 @@ func (i *DatabaseInstance) startQueues() error {
 		i.queues[queueNum] = q
 		i.logger.Info("queue started", "number", queueNum)
 	}
+	i.metrics.SetQueueCount(i.DatabaseName, len(i.queues))
 
 	return nil
 }

pkg/broker/metrics/metrics.go

@@ -0,0 +1,130 @@
+// Package metrics defines the broker's Prometheus collectors and a small
+// embedded HTTP server that exposes them, both as the standard /metrics
+// exposition endpoint and as a single-page HTML dashboard.
+package metrics
+
+import (
+	"strconv"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Metrics holds every Prometheus collector the broker exposes. It is safe
+// for concurrent use, and a nil *Metrics is safe to call methods on (all
+// recording methods become no-ops), so callers that construct a Broker
+// without metrics enabled don't need to special-case it.
+type Metrics struct {
+	Registry *prometheus.Registry
+
+	jobsCompleted *prometheus.CounterVec
+	jobsFailed    *prometheus.CounterVec
+	jobsRequeued  *prometheus.CounterVec
+	jobDuration   *prometheus.HistogramVec
+	jobsQueued    *prometheus.GaugeVec
+	databaseCount prometheus.Gauge
+	queueCount    *prometheus.GaugeVec
+}
+
+// New creates a Metrics instance with a fresh (non-global) registry, so
+// multiple brokers can coexist in the same process -- e.g. in tests --
+// without colliding on prometheus.DefaultRegisterer.
+func New() *Metrics {
+	registry := prometheus.NewRegistry()
+
+	m := &Metrics{
+		Registry: registry,
+		jobsCompleted: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "broker_jobs_completed_total",
+			Help: "Total number of jobs that completed successfully.",
+		}, []string{"database", "job_group", "job_name"}),
+		jobsFailed: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "broker_jobs_failed_total",
+			Help: "Total number of jobs that were dead-lettered after exhausting retries.",
+		}, []string{"database", "job_group", "job_name"}),
+		jobsRequeued: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "broker_jobs_requeued_total",
+			Help: "Total number of job attempts that failed and were requeued for retry.",
+		}, []string{"database", "job_group", "job_name"}),
+		jobDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "broker_job_duration_seconds",
+			Help:    "Job execution duration in seconds, by group and name.",
+			Buckets: prometheus.DefBuckets,
+		}, []string{"database", "job_group", "job_name"}),
+		jobsQueued: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "broker_jobs_queued",
+			Help: "Current number of pending (not yet claimed) jobs, by database and queue.",
+		}, []string{"database", "queue"}),
+		databaseCount: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "broker_databases",
+			Help: "Number of database instances managed by this broker process.",
+		}),
+		queueCount: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "broker_queues",
+			Help: "Number of queues configured for a database instance.",
+		}, []string{"database"}),
+	}
+
+	registry.MustRegister(
+		m.jobsCompleted,
+		m.jobsFailed,
+		m.jobsRequeued,
+		m.jobDuration,
+		m.jobsQueued,
+		m.databaseCount,
+		m.queueCount,
+	)
+
+	return m
+}
+
+// RecordJobCompleted records a successfully completed job attempt.
+func (m *Metrics) RecordJobCompleted(database, group, name string, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	m.jobsCompleted.WithLabelValues(database, group, name).Inc()
+	m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
+}
+
+// RecordJobFailed records a job attempt that was dead-lettered (attempts exhausted).
+func (m *Metrics) RecordJobFailed(database, group, name string, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	m.jobsFailed.WithLabelValues(database, group, name).Inc()
+	m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
+}
+
+// RecordJobRequeued records a job attempt that failed but was requeued for retry.
+func (m *Metrics) RecordJobRequeued(database, group, name string, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	m.jobsRequeued.WithLabelValues(database, group, name).Inc()
+	m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
+}
+
+// SetJobsQueued sets the current pending job count for a database/queue pair.
+func (m *Metrics) SetJobsQueued(database string, queue int, count float64) {
+	if m == nil {
+		return
+	}
+	m.jobsQueued.WithLabelValues(database, strconv.Itoa(queue)).Set(count)
+}
+
+// SetDatabaseCount sets the number of database instances managed by this process.
+func (m *Metrics) SetDatabaseCount(n int) {
+	if m == nil {
+		return
+	}
+	m.databaseCount.Set(float64(n))
+}
+
+// SetQueueCount sets the number of queues configured for a database instance.
+func (m *Metrics) SetQueueCount(database string, n int) {
+	if m == nil {
+		return
+	}
+	m.queueCount.WithLabelValues(database).Set(float64(n))
+}

pkg/broker/metrics/page.html

@@ -0,0 +1,215 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>pgsql-broker metrics</title>
+<style>
+  :root {
+    color-scheme: dark;
+    --bg: #0f1115;
+    --panel: #161a22;
+    --border: #262b36;
+    --text: #e6e9ef;
+    --muted: #8b93a7;
+    --accent: #5aa8ff;
+    --good: #4caf7d;
+    --bad: #e5636b;
+  }
+  * { box-sizing: border-box; }
+  body {
+    margin: 0;
+    padding: 2rem;
+    background: var(--bg);
+    color: var(--text);
+    font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
+  }
+  h1 {
+    font-size: 1.25rem;
+    margin: 0 0 0.25rem;
+  }
+  #status {
+    color: var(--muted);
+    font-size: 0.8rem;
+    margin-bottom: 1.5rem;
+  }
+  .grid {
+    display: grid;
+    grid-template-columns: repeat(auto-fill, minmax(360px, 1fr));
+    gap: 1rem;
+  }
+  .card {
+    background: var(--panel);
+    border: 1px solid var(--border);
+    border-radius: 8px;
+    padding: 1rem;
+  }
+  .card h2 {
+    font-size: 0.9rem;
+    margin: 0 0 0.75rem;
+    color: var(--accent);
+    font-weight: 600;
+  }
+  .card .help {
+    color: var(--muted);
+    font-size: 0.75rem;
+    margin: -0.5rem 0 0.75rem;
+  }
+  table {
+    width: 100%;
+    border-collapse: collapse;
+    font-size: 0.8rem;
+  }
+  td, th {
+    text-align: left;
+    padding: 0.2rem 0.4rem 0.2rem 0;
+    border-bottom: 1px solid var(--border);
+  }
+  td.value {
+    text-align: right;
+    font-variant-numeric: tabular-nums;
+    white-space: nowrap;
+  }
+  td.labels {
+    color: var(--muted);
+  }
+  .completed .value { color: var(--good); }
+  .failed .value { color: var(--bad); }
+  .empty {
+    color: var(--muted);
+    font-size: 0.8rem;
+  }
+</style>
+</head>
+<body>
+  <h1>pgsql-broker metrics</h1>
+  <div id="status">loading&hellip;</div>
+  <div class="grid" id="grid"></div>
+
+<script>
+(function () {
+  "use strict";
+
+  // Metric name -> { help, className } for the cards we render, in order.
+  var METRICS = [
+    { name: "broker_jobs_completed_total", title: "Jobs completed", cls: "completed" },
+    { name: "broker_jobs_failed_total", title: "Jobs failed (dead-lettered)", cls: "failed" },
+    { name: "broker_jobs_requeued_total", title: "Jobs requeued (retries)", cls: "" },
+    { name: "broker_jobs_queued", title: "Jobs currently queued", cls: "" },
+    { name: "broker_job_duration_seconds_sum", title: "Job duration, total seconds by group/name", cls: "" },
+    { name: "broker_job_duration_seconds_count", title: "Job duration, sample count by group/name", cls: "" },
+    { name: "broker_databases", title: "Databases managed", cls: "" },
+    { name: "broker_queues", title: "Queues per database", cls: "" }
+  ];
+
+  // Parses Prometheus text exposition format into { name: [{labels, value}] }.
+  function parseMetrics(text) {
+    var byName = {};
+    var lines = text.split("\n");
+    for (var i = 0; i < lines.length; i++) {
+      var line = lines[i];
+      if (!line || line[0] === "#") continue;
+
+      var name, labels = {}, rest;
+      var braceIdx = line.indexOf("{");
+      var spaceIdx;
+      if (braceIdx !== -1) {
+        name = line.slice(0, braceIdx);
+        var closeIdx = line.indexOf("}", braceIdx);
+        if (closeIdx === -1) continue;
+        var labelStr = line.slice(braceIdx + 1, closeIdx);
+        var labelRe = /(\w+)="((?:[^"\\]|\\.)*)"/g;
+        var m;
+        while ((m = labelRe.exec(labelStr)) !== null) {
+          labels[m[1]] = m[2].replace(/\\"/g, '"').replace(/\\\\/g, "\\");
+        }
+        rest = line.slice(closeIdx + 1).trim();
+      } else {
+        spaceIdx = line.indexOf(" ");
+        if (spaceIdx === -1) continue;
+        name = line.slice(0, spaceIdx);
+        rest = line.slice(spaceIdx + 1).trim();
+      }
+
+      var value = parseFloat(rest.split(" ")[0]);
+      if (isNaN(value)) continue;
+
+      if (!byName[name]) byName[name] = [];
+      byName[name].push({ labels: labels, value: value });
+    }
+    return byName;
+  }
+
+  function formatLabels(labels) {
+    var keys = Object.keys(labels).sort();
+    return keys.map(function (k) { return k + "=" + labels[k]; }).join(", ");
+  }
+
+  function formatValue(v) {
+    if (Number.isInteger(v)) return String(v);
+    return v.toFixed(3);
+  }
+
+  function render(byName) {
+    var grid = document.getElementById("grid");
+    grid.innerHTML = "";
+
+    METRICS.forEach(function (spec) {
+      var series = byName[spec.name] || [];
+      var card = document.createElement("div");
+      card.className = "card " + spec.cls;
+
+      var h2 = document.createElement("h2");
+      h2.textContent = spec.title;
+      card.appendChild(h2);
+
+      if (series.length === 0) {
+        var empty = document.createElement("div");
+        empty.className = "empty";
+        empty.textContent = "no data yet";
+        card.appendChild(empty);
+      } else {
+        var table = document.createElement("table");
+        series.sort(function (a, b) { return b.value - a.value; });
+        series.forEach(function (s) {
+          var tr = document.createElement("tr");
+          var tdLabels = document.createElement("td");
+          tdLabels.className = "labels";
+          tdLabels.textContent = formatLabels(s.labels) || "(none)";
+          var tdValue = document.createElement("td");
+          tdValue.className = "value";
+          tdValue.textContent = formatValue(s.value);
+          tr.appendChild(tdLabels);
+          tr.appendChild(tdValue);
+          table.appendChild(tr);
+        });
+        card.appendChild(table);
+      }
+
+      grid.appendChild(card);
+    });
+  }
+
+  function refresh() {
+    fetch("metrics", { cache: "no-store" })
+      .then(function (resp) {
+        if (!resp.ok) throw new Error("HTTP " + resp.status);
+        return resp.text();
+      })
+      .then(function (text) {
+        render(parseMetrics(text));
+        document.getElementById("status").textContent =
+          "last updated " + new Date().toLocaleTimeString();
+      })
+      .catch(function (err) {
+        document.getElementById("status").textContent =
+          "failed to load metrics: " + err.message;
+      });
+  }
+
+  refresh();
+  setInterval(refresh, 5000);
+})();
+</script>
+</body>
+</html>

pkg/broker/metrics/server.go

@@ -0,0 +1,74 @@
+package metrics
+
+import (
+	"context"
+	_ "embed"
+	"fmt"
+	"net"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
+)
+
+//go:embed page.html
+var dashboardHTML []byte
+
+// Server is the embedded HTTP server exposing /metrics (standard Prometheus
+// exposition format) and / (a single-page HTML dashboard that polls
+// /metrics).
+type Server struct {
+	httpServer *http.Server
+	listener   net.Listener
+	logger     adapter.Logger
+}
+
+// NewServer builds a Server bound to addr (e.g. "127.0.0.1:9469"). Binding
+// happens immediately so a port conflict is reported to the caller rather
+// than surfacing later in a background goroutine.
+func NewServer(m *Metrics, addr string, logger adapter.Logger) (*Server, error) {
+	listener, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to bind metrics server to %s: %w", addr, err)
+	}
+
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.HandlerFor(m.Registry, promhttp.HandlerOpts{}))
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "text/html; charset=utf-8")
+		_, _ = w.Write(dashboardHTML)
+	})
+
+	return &Server{
+		httpServer: &http.Server{Handler: mux},
+		listener:   listener,
+		logger:     logger.With("component", "metrics-server"),
+	}, nil
+}
+
+// Addr returns the actual bound address (useful when addr was given with a
+// ":0" port).
+func (s *Server) Addr() string {
+	return s.listener.Addr().String()
+}
+
+// Start serves in the background. It returns immediately; Serve errors
+// (other than a clean Shutdown) are logged.
+func (s *Server) Start() {
+	s.logger.Info("metrics server listening", "addr", s.Addr())
+	go func() {
+		if err := s.httpServer.Serve(s.listener); err != nil && err != http.ErrServerClosed {
+			s.logger.Error("metrics server stopped unexpectedly", "error", err)
+		}
+	}()
+}
+
+// Stop gracefully shuts down the server.
+func (s *Server) Stop(ctx context.Context) error {
+	return s.httpServer.Shutdown(ctx)
+}

pkg/broker/queue/queue.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/worker"
 )
 
@@ -34,6 +35,8 @@ type Config struct {
 	FetchSize    int
 	TenantID     string
 	LeaseSeconds int
+	Metrics      *metrics.Metrics
+	DatabaseName string
 }
 
 // New creates a new queue manager
@@ -73,6 +76,8 @@ func (q *Queue) Start(cfg Config) error {
 			FetchSize:    cfg.FetchSize,
 			TenantID:     cfg.TenantID,
 			LeaseSeconds: cfg.LeaseSeconds,
+			Metrics:      cfg.Metrics,
+			DatabaseName: cfg.DatabaseName,
 		})
 
 		if err := w.Start(q.ctx); err != nil {

pkg/broker/worker/worker.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 )
 
 // Worker represents a single job processing worker
@@ -29,6 +30,8 @@ type Worker struct {
 	fetchSize    int
 	tenantID     string
 	leaseSeconds int
+	metrics      *metrics.Metrics
+	databaseName string
 }
 
 // Stats holds worker statistics
@@ -50,6 +53,8 @@ type Config struct {
 	FetchSize    int
 	TenantID     string
 	LeaseSeconds int
+	Metrics      *metrics.Metrics
+	DatabaseName string
 }
 
 // New creates a new worker
@@ -72,6 +77,8 @@ func New(cfg Config) *Worker {
 		fetchSize:    cfg.FetchSize,
 		tenantID:     cfg.TenantID,
 		leaseSeconds: leaseSeconds,
+		metrics:      cfg.Metrics,
+		databaseName: cfg.DatabaseName,
 	}
 }
 
@@ -228,8 +235,19 @@ func (w *Worker) processJobs(ctx context.Context) {
 			return // No more jobs
 		}
 
+		jobName, jobGroup, err := w.fetchJobLabelsTx(ctx, tx, jobID)
+		if err != nil {
+			w.logger.Warn("failed to fetch job labels for metrics", "job_id", jobID, "error", err)
+		}
+
 		// Run the job
-		if err := w.runJobTx(ctx, tx, jobID, leaseToken); err != nil {
+		start := time.Now()
+		jobStatus, err := w.runJobTx(ctx, tx, jobID, leaseToken)
+		duration := time.Since(start)
+		if err == nil {
+			w.recordJobMetric(jobStatus, jobGroup, jobName, duration)
+		}
+		if err != nil {
 			// Rollback on genuine infra failure
 			if rbErr := tx.Rollback(); rbErr != nil {
 				w.logger.Error("failed to rollback transaction", "error", rbErr)
@@ -287,7 +305,7 @@ func (w *Worker) fetchNextJobTx(ctx context.Context, tx adapter.DBTransaction) (
 // error (triggering a rollback of the claim) on a genuine infra failure --
 // job outcomes reported via p_job_status (requeued/completed/dead-lettered)
 // are always committed.
-func (w *Worker) runJobTx(ctx context.Context, tx adapter.DBTransaction, jobID int64, leaseToken string) error {
+func (w *Worker) runJobTx(ctx context.Context, tx adapter.DBTransaction, jobID int64, leaseToken string) (int, error) {
 	w.logger.Debug("running job", "job_id", jobID)
 
 	var retval int
@@ -300,15 +318,43 @@ func (w *Worker) runJobTx(ctx context.Context, tx adapter.DBTransaction, jobID i
 	).Scan(&retval, &errmsg, &jobStatus)
 
 	if err != nil {
-		return fmt.Errorf("query error: %w", err)
+		return 0, fmt.Errorf("query error: %w", err)
 	}
 
 	if retval > 0 {
-		return fmt.Errorf("broker_run error: %s", errmsg)
+		return 0, fmt.Errorf("broker_run error: %s", errmsg)
 	}
 
 	w.logger.Debug("job finished", "job_id", jobID, "job_status", jobStatus)
-	return nil
+	return jobStatus, nil
+}
+
+// fetchJobLabelsTx looks up the job_name/job_group of jobID for metric
+// labeling. Best-effort: callers log and continue on error rather than
+// failing the job over a metrics lookup.
+func (w *Worker) fetchJobLabelsTx(ctx context.Context, tx adapter.DBTransaction, jobID int64) (jobName, jobGroup string, err error) {
+	err = tx.QueryRow(ctx,
+		"SELECT job_name, job_group FROM broker.broker_jobs WHERE id_broker_jobs = $1",
+		jobID,
+	).Scan(&jobName, &jobGroup)
+	if err != nil {
+		return "", "", fmt.Errorf("query error: %w", err)
+	}
+	return jobName, jobGroup, nil
+}
+
+// recordJobMetric routes a finished job attempt to the appropriate
+// Prometheus counter/histogram based on the p_job_status broker_run
+// reported (0=requeued, 2=completed, 3=dead-lettered).
+func (w *Worker) recordJobMetric(jobStatus int, jobGroup, jobName string, duration time.Duration) {
+	switch jobStatus {
+	case 2:
+		w.metrics.RecordJobCompleted(w.databaseName, jobGroup, jobName, duration)
+	case 3:
+		w.metrics.RecordJobFailed(w.databaseName, jobGroup, jobName, duration)
+	case 0:
+		w.metrics.RecordJobRequeued(w.databaseName, jobGroup, jobName, duration)
+	}
 }
 
 // updateActivity updates the last activity timestamp