feat(db): reset and pull verbs for the shared Postgres (spec 15)

internal/cli/db.go

@@ -17,8 +17,10 @@ import (
 // Postgres database + role/grant verbs on the shared engine. create/user/grant/
 // drop/gc mirror the up-saga provision flow (lock → overlay → provisioner →
 // ledger → event) via internal/orchestrate; list is a lock-free ledger read.
-// snapshot/restore (+ snapshot ls) graduate the spec-15 data-lifecycle verbs;
-// reset/pull stay v2 stubs.
+// snapshot/restore (+ snapshot ls) plus reset/pull graduate the full spec-15
+// data-lifecycle surface (reset = drop + re-provision an empty tenant; pull =
+// fetch-by-name from the LOCAL snapshot store + restore — the remote/team shared
+// store is deferred to spec 21).
 func newDbCmd(g *GlobalOpts) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "db",
@@ -34,9 +36,8 @@ func newDbCmd(g *GlobalOpts) *cobra.Command {
 		// spec-15 data-lifecycle verbs.
 		newDbSnapshotCmd(g),
 		newDbRestoreCmd(g),
-		// remaining spec-15 verbs (reset/pull) reserved as stubs.
-		stub("reset", "Drop and re-provision a project's database", "v2 (spec 15)"),
-		stub("pull", "Pull a database snapshot from a shared store", "v2 (spec 15)"),
+		newDbResetCmd(g),
+		newDbPullCmd(g),
 	)
 	return cmd
 }
@@ -173,6 +174,107 @@ func newDbRestoreCmd(g *GlobalOpts) *cobra.Command {
 	return cmd
 }
 
+// newDbResetCmd wires `db reset` (spec 15): DROP and re-provision a project's
+// tenant database on the shared Postgres to a clean, empty tenant. Destructive —
+// it refuses without --yes; --force overrides a still-connected database
+// (terminating its live backends). The drop + re-provision DDL runs under the
+// flock; the never-recreate-a-stateful-shared-service guard holds (only the
+// tenant DATABASE is dropped, never the shared container).
+func newDbResetCmd(g *GlobalOpts) *cobra.Command {
+	var project, instance string
+	var yes, force bool
+	cmd := &cobra.Command{
+		Use:   "reset",
+		Short: "Drop and re-provision a project's tenant database to empty (destructive)",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			if g.JSON && !yes {
+				return fmt.Errorf("refusing to reset without --yes for --json/non-interactive use")
+			}
+			d, closeFn, err := buildUpDeps(cmd)
+			if err != nil {
+				return err
+			}
+			defer closeFn()
+			proj := project
+			if proj == "" {
+				proj = defaultProject(d)
+			}
+			if !yes {
+				if !confirm(cmd, fmt.Sprintf("This DROPS and recreates %q's tenant database EMPTY (all data destroyed). Type 'yes' to continue: ", proj)) {
+					fmt.Fprintln(cmd.OutOrStdout(), "aborted")
+					return nil
+				}
+			}
+			res, err := orchestrate.Reset(cmd.Context(), d, orchestrate.ResetOptions{
+				Project: project, Instance: instance, Force: force,
+			})
+			if err != nil {
+				return err
+			}
+			if g.JSON {
+				return writeJSON(cmd, res)
+			}
+			if !g.Quiet {
+				fmt.Fprintf(cmd.OutOrStdout(), "reset %s: dropped and recreated empty database %q owned by %q\n", res.Project, res.Database, res.Role)
+			}
+			return nil
+		},
+	}
+	cmd.Flags().StringVar(&project, "project", "", "owner project (default: the workspace's single/first project)")
+	cmd.Flags().StringVar(&instance, "instance", "", "shared postgres instance (default: the first postgres instance)")
+	cmd.Flags().BoolVar(&yes, "yes", false, "skip the confirmation prompt (required for --json)")
+	cmd.Flags().BoolVar(&force, "force", false, "terminate live connections and reset a still-connected database")
+	return cmd
+}
+
+// newDbPullCmd wires `db pull <name>` (spec 15): fetch a named snapshot from the
+// LOCAL snapshot store and apply it into the project's tenant, seeding it from a
+// real dataset. In this scope the store is local ($DEVSTACK_HOME/snapshots), so a
+// pull is a fetch-by-name + restore that refuses to clobber a non-empty tenant
+// (run `db reset` first to re-seed). The REMOTE/team "shared store" (S3/HTTP fetch
+// + the mandatory sanitize transform) is DEFERRED to spec 21.
+func newDbPullCmd(g *GlobalOpts) *cobra.Command {
+	var project, database, instance string
+	cmd := &cobra.Command{
+		Use:   "pull <name>",
+		Short: "Seed a project's tenant from a named snapshot in the local store",
+		Long: "Fetch a named snapshot from the local snapshot store ($DEVSTACK_HOME/snapshots) " +
+			"and apply it into the project's tenant database, seeding a fresh tenant. " +
+			"Refuses to overwrite a non-empty tenant (run `db reset` first to re-seed). " +
+			"The remote/team shared store (fetch-by-URL + sanitize) is deferred to spec 21.",
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			d, closeFn, err := buildUpDeps(cmd)
+			if err != nil {
+				return err
+			}
+			defer closeFn()
+			dumper := defaultPgDumper()
+			if err := dumper.Preflight(cmd.Context()); err != nil {
+				return err
+			}
+			meta, err := orchestrate.Pull(cmd.Context(), d, dumper, orchestrate.PullOptions{
+				Project: project, Database: database, Instance: instance, Name: args[0],
+			})
+			if err != nil {
+				return err
+			}
+			if g.JSON {
+				return writeJSON(cmd, meta)
+			}
+			if !g.Quiet {
+				fmt.Fprintf(cmd.OutOrStdout(), "pulled snapshot %q into %s (digest %s)\n", meta.Name, meta.Database, shortDigest(meta.Digest))
+			}
+			return nil
+		},
+	}
+	cmd.Flags().StringVar(&project, "project", "", "owner project (default: the workspace's single/first project)")
+	cmd.Flags().StringVar(&database, "db", "", "physical tenant database (default: the project's own database)")
+	cmd.Flags().StringVar(&instance, "instance", "", "shared postgres instance (default: the first postgres instance)")
+	return cmd
+}
+
 // shortDigest is a display helper: the first 12 hex chars of a sha256, or "-".
 func shortDigest(d string) string {
 	if len(d) >= 12 {

internal/cli/db_reset_pull_test.go

@@ -0,0 +1,53 @@
+package cli
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestDbResetPullRegistered(t *testing.T) {
+	root := NewRootCmd(Options{})
+	for _, path := range [][]string{{"db", "reset"}, {"db", "pull"}} {
+		c, _, err := root.Find(path)
+		if err != nil || c.RunE == nil {
+			t.Fatalf("db %v not registered as a real command: %v", path, err)
+		}
+	}
+}
+
+func TestDbResetFlags(t *testing.T) {
+	root := NewRootCmd(Options{})
+	reset, _, err := root.Find([]string{"db", "reset"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, f := range []string{"project", "instance", "yes", "force"} {
+		if reset.Flags().Lookup(f) == nil {
+			t.Errorf("db reset missing --%s", f)
+		}
+	}
+	pull, _, err := root.Find([]string{"db", "pull"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, f := range []string{"project", "db", "instance"} {
+		if pull.Flags().Lookup(f) == nil {
+			t.Errorf("db pull missing --%s", f)
+		}
+	}
+}
+
+// TestDbResetJSONRequiresYes asserts the destructive-verb non-interactive guard:
+// `db reset --json` without --yes errors BEFORE touching any workspace/Docker
+// state (mirrors the `workspace destroy --json` guard).
+func TestDbResetJSONRequiresYes(t *testing.T) {
+	t.Chdir(t.TempDir())
+	var out strings.Builder
+	root := NewRootCmd(Options{})
+	root.SetArgs([]string{"db", "reset", "--json"})
+	root.SetOut(&out)
+	root.SetErr(&out)
+	if err := root.Execute(); err == nil {
+		t.Fatal("db reset --json without --yes must error")
+	}
+}

internal/orchestrate/reset.go

@@ -0,0 +1,130 @@
+package orchestrate
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/open-source-cloud/devstack/internal/generate"
+	"github.com/open-source-cloud/devstack/internal/lock"
+	"github.com/open-source-cloud/devstack/internal/provision"
+)
+
+// This file is the imperative side of spec 15's `db reset`: DROP + re-provision a
+// project's per-project tenant database on the SHARED Postgres, reusing the exact
+// provision-phase host-reachability path (engineTarget → ledger port + loopback
+// overlay via `compose up`) so the pgx admin connection reaches the warm server
+// WITHOUT publishing a permanent host port.
+//
+// Unlike snapshot/restore (whose long-running dump/restore PROCESS runs OUTSIDE
+// the flock), a reset is a handful of quick DDL statements, so the terminate →
+// DROP DATABASE → re-provision SQL runs INSIDE the flock (spec 15: "the drop/
+// recreate SQL happen inside the flock"), exactly like the provision phase. The
+// never-recreate-a-stateful-shared-service guard still holds: we drop the tenant
+// DATABASE only, never the shared container/volume, and terminate only THIS
+// tenant's backends (never other tenants' live connections).
+
+// ResetOptions selects the tenant to drop + re-provision.
+type ResetOptions struct {
+	Project  string // owner project (default: the workspace's single/first project)
+	Database string // physical tenant db (default: the project's own <project> db)
+	Instance string // shared Postgres instance (default: the first postgres instance)
+	Force    bool   // terminate + proceed even if the tenant still has live connections
+}
+
+// ResetResult is the outcome of a reset (the recreated empty tenant).
+type ResetResult struct {
+	Project  string `json:"project"`
+	Instance string `json:"instance"`
+	Database string `json:"database"`
+	Role     string `json:"role"`
+}
+
+// liveBackendsSQL detects any backend (other than our own) attached to the tenant
+// database — the guard that makes reset refuse a still-connected DB unless --force.
+const liveBackendsSQL = `SELECT 1 FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()`
+
+// terminateBackendsSQL terminates only THIS tenant's backends (never other
+// tenants' — the shared-service isolation guard, spec 15).
+const terminateBackendsSQL = `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()`
+
+// Reset drops a project's tenant database and re-runs the idempotent provisioner
+// to recreate an empty tenant (role kept/recreated, database recreated fresh). The
+// terminate → DROP → re-provision DDL runs inside the flock; the loopback overlay
+// is applied outside it (engineTarget self-locks the port allocation).
+func Reset(ctx context.Context, d UpDeps, opt ResetOptions) (ResetResult, error) {
+	proj, dbName, inst, err := resolveTenant(d, opt.Project, opt.Database, opt.Instance)
+	if err != nil {
+		return ResetResult{}, err
+	}
+
+	// Apply the host-reachability overlay (ledger port + `compose up`), same path
+	// the provision phase uses. This runs outside the flock (FreeHostPort self-locks).
+	target, err := engineTarget(ctx, d, "postgres", inst)
+	if err != nil {
+		return ResetResult{}, err
+	}
+	connect := d.PgConnect
+	if connect == nil {
+		connect = defaultPgConnect
+	}
+	user := target.AdminEnv["user"]
+	pass := target.AdminEnv["password"]
+	dsn := provision.DSN(target.Host, target.Port, user, pass, user)
+
+	var result ResetResult
+	if err := lock.WithLock(ctx, d.LockPath, func() error {
+		conn, closeConn, err := connect(ctx, dsn)
+		if err != nil {
+			return fmt.Errorf("connect to shared %s on %s:%d: %w", inst, target.Host, target.Port, err)
+		}
+		defer func() { _ = closeConn() }()
+
+		// Refuse a still-connected tenant unless --force (data-loss guard, spec 15).
+		if !opt.Force {
+			live, err := conn.Exists(ctx, liveBackendsSQL, dbName)
+			if err != nil {
+				return fmt.Errorf("check live connections on %q: %w", dbName, err)
+			}
+			if live {
+				return fmt.Errorf("database %q still has active connections; pass --force to terminate them and reset anyway", dbName)
+			}
+		}
+
+		// Terminate only this tenant's backends, then drop the database. DROP
+		// DATABASE cannot run in a transaction and fails with attached sessions —
+		// so terminate first (spec 15).
+		if err := conn.Exec(ctx, terminateBackendsSQL, dbName); err != nil {
+			return fmt.Errorf("terminate sessions on %q: %w", dbName, err)
+		}
+		if err := conn.Exec(ctx, `DROP DATABASE IF EXISTS `+quotePgIdent(dbName)); err != nil {
+			return fmt.Errorf("drop database %q: %w", dbName, err)
+		}
+
+		// Re-provision an empty tenant via the SAME existence-guarded, idempotent
+		// SQL the provision phase uses (role kept in sync / recreated, fresh db,
+		// predictable dev cred == project name).
+		creds, err := provision.Postgres{}.EnsureProject(ctx, conn, proj, proj)
+		if err != nil {
+			return fmt.Errorf("re-provision %s on %s: %w", proj, inst, err)
+		}
+
+		if err := d.DB.RecordProvisioned(proj, "role", creds.Role); err != nil {
+			return err
+		}
+		if err := d.DB.RecordProvisioned(proj, "database", creds.Database); err != nil {
+			return err
+		}
+		d.DB.LogEvent("db.reset", proj, fmt.Sprintf("dropped+recreated %s on %s", creds.Database, generate.SharedAlias(inst)))
+		result = ResetResult{Project: proj, Instance: inst, Database: creds.Database, Role: creds.Role}
+		return nil
+	}); err != nil {
+		return ResetResult{}, err
+	}
+	return result, nil
+}
+
+// quotePgIdent double-quotes a Postgres identifier, doubling embedded quotes.
+// Local to orchestrate so DROP DATABASE renders a safe identifier (the tenant db
+// is already hyphen-sanitized by pgTenantDB, but quoting is the belt-and-braces).
+func quotePgIdent(s string) string { return `"` + strings.ReplaceAll(s, `"`, `""`) + `"` }