runtime.go

package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/codefly-dev/core/agents/helpers/code"

	"github.com/codefly-dev/core/agents/services"
	basev0 "github.com/codefly-dev/core/generated/go/codefly/base/v0"
	"github.com/codefly-dev/core/wool"

	runtimev0 "github.com/codefly-dev/core/generated/go/codefly/services/runtime/v0"
	"github.com/codefly-dev/core/resources"
	dockerrun "github.com/codefly-dev/core/runners/dockerrun"
	_ "github.com/golang-migrate/migrate/v4/source/file"
	_ "github.com/lib/pq"
)

const (
	postgresDataCacheKey  = "postgres-data"
	postgresDataDirectory = "/var/lib/postgresql/data"

	// readinessProbeInterval paces the readiness loop. The number of attempts
	// is whatever the readiness budget affords, not a separate knob.
	readinessProbeInterval = 3 * time.Second
)

type persistentCacheMounter interface {
	WithPersistentCacheMount(context.Context, string, string) (string, error)
}

// mountPersistentPostgresData layers the Codefly-owned persistent directory
// over the image's data directory and returns that host path, which is where
// the database survives every lifecycle operation.
func mountPersistentPostgresData(ctx context.Context, runner persistentCacheMounter) (string, error) {
	return runner.WithPersistentCacheMount(ctx, postgresDataCacheKey, postgresDataDirectory)
}

type Runtime struct {
	services.RuntimeServer
	*Service

	// internal
	runnerEnvironment *dockerrun.DockerEnvironment

	// nixRuntime is set instead of runnerEnvironment when the caller requests
	// RuntimeContextNix: this service's attachment to its run's one native
	// postgres server (nixpg.go), serving the same connection string and
	// database.
	nixRuntime *nixPostgres

	// extensionAdmin is the connection extensions are created with. On the
	// run's shared server that is the server's administrator in this service's
	// database — the migration owner is not a superuser there, and pgvector is
	// not a trusted extension — the way a managed server's provider creates
	// them. Empty elsewhere: the owner connection creates them.
	extensionAdmin string

	postgresPort uint16

	// lifecycleMu serializes Stop and Destroy. Both act on runnerEnvironment,
	// and the CLI can overlap them: it bounds each Stop at ten seconds and then
	// fans out Destroy as soon as that deadline passes, while the abandoned Stop
	// handler is still inside the docker client. DockerEnvironment.instance is
	// written by Shutdown and read by Stop with no lock of its own, so without
	// this the two race on the same handle.
	lifecycleMu sync.Mutex

	// released records that Stop already gave up this invocation's execution
	// resources. Start probes the database for ninety seconds before giving up,
	// which is a long way to discover that the server it is waiting for was
	// deliberately stopped.
	released bool

	// retainedDataPath is where this invocation's database state lives on disk —
	// the Codefly-owned cache directory bind-mounted into the container, or the
	// nix cluster's data directory. Reported in Stop/Destroy results so a caller
	// sees what was kept without having to inspect the backend.
	retainedDataPath string

	// migrationReload serializes hot-reload migration work against this
	// database. Deciding what to apply reads the ledger and applies it in
	// separate statements, so two reloads at once would both plan from the same
	// version.
	migrationReload sync.Mutex

	// controlPlane serializes this agent's database control-plane mutations:
	// extension creation, migration application, hot-reload changes, and
	// runtime-access reconciliation. The hot-reload watcher drives its work from
	// its own goroutine while Init and Start drive the rest from RPC handlers,
	// and Postgres does not serialize them for us: REVOKE/GRANT ON ALL TABLES
	// rewrites the pg_class row of every table in the schema — each lineage's
	// golang-migrate tracking table included — while taking no lock on the table
	// itself, so it collides with the TRUNCATE golang-migrate uses to record a
	// version and one side aborts with "tuple concurrently updated".
	//
	// This lock covers ONE process. The cross-process half of the same collision
	// is held by the runtime-access advisory lock — see runtimeAccessLockID,
	// which every control-plane mutation takes inside the database itself.
	controlPlane controlPlaneLock
}

// controlPlaneLock is context-aware mutual exclusion over one Runtime's
// database control-plane mutations. sync.Mutex would serialize just as well,
// but Lock() cannot be cancelled: a caller whose ctx already expired would
// still queue behind an in-flight migration and then run DDL on behalf of an RPC
// that returned long ago. Acquiring through a channel lets a cancelled caller
// return ctx.Err() instead of executing late. The zero value is ready to use.
type controlPlaneLock struct {
	once sync.Once
	ch   chan struct{}
}

// acquire blocks until the control plane is free or ctx is done. The returned
// release must be called exactly once, and only when err is nil.
func (l *controlPlaneLock) acquire(ctx context.Context) (func(), error) {
	l.once.Do(func() { l.ch = make(chan struct{}, 1) })
	// Checked BEFORE the select: when the lock is free and ctx is already done,
	// both select cases are ready and Go picks between them at random, so an
	// expired caller would sometimes proceed to run DDL anyway. Failing up front
	// makes "cancelled callers never mutate" deterministic.
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	select {
	case l.ch <- struct{}{}:
		return func() { <-l.ch }, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func NewRuntime() *Runtime {
	return &Runtime{
		Service: NewService(),
	}
}

func (s *Runtime) Load(ctx context.Context, req *runtimev0.LoadRequest) (*runtimev0.LoadResponse, error) {
	defer s.Wool.Catch()

	response, err := s.Runtime.LoadService(ctx, req, services.RuntimeLoad{
		Settings:     s.Settings,
		Requirements: requirements,
		ResolveEndpoints: func(ctx context.Context, endpoints []*basev0.Endpoint) error {
			s.Wool.Debug("endpoints", wool.Field("endpoints", resources.MakeManyEndpointSummary(endpoints)))
			endpoint, err := resources.FindTCPEndpoint(ctx, endpoints)
			if err != nil {
				return s.Wool.Wrapf(err, "cannot find TCP endpoint")
			}
			s.TcpEndpoint = endpoint
			return nil
		},
	})
	if err != nil {
		return response, err
	}
	if err = s.Settings.Timeouts.validate(); err != nil {
		return s.Runtime.LoadError(err)
	}
	return response, nil
}

func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtimev0.InitResponse, error) {
	defer s.Wool.Catch()
	ctx = s.Wool.Inject(ctx)

	s.Runtime.LogInitRequest(req)
	s.Runtime.WithContext(req.GetRuntimeContext())

	w := s.Wool.In("runtime::init")

	s.NetworkMappings = req.ProposedNetworkMappings

	// On the native runtime the service's database lives on its run's shared
	// server, where roles are server-wide: the migration owner is this
	// database's own role there, not the configured POSTGRES_USER every local
	// service ships with.
	native := req.GetRuntimeContext().GetKind() == resources.RuntimeContextNix
	s.nativeOwner = ""
	if native {
		s.nativeOwner = nativeOwnerRole(s.DatabaseName)
	}

	configuration := req.GetConfiguration()

	net, err := resources.FindNetworkMapping(ctx, s.NetworkMappings, s.TcpEndpoint)
	if err != nil {
		return s.Runtime.InitError(err)
	}

	if net == nil {
		return s.Runtime.InitError(w.NewError("network mapping is nil"))
	}

	// ARCHITECTURE: the Postgres container publishes a port to the agent host,
	// and migrations/runtime-role reconciliation execute in this host agent
	// process. Always select the native mapping for those control-plane calls.
	// A container mapping such as host.docker.internal is for a *consumer*
	// running in another container; it is not a portable hostname on the host
	// itself (notably on Linux and several macOS Docker backends).
	instance, err := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.TcpEndpoint, resources.NewNativeNetworkAccess())
	if err != nil {
		return s.Runtime.InitError(err)
	}

	if instance == nil {
		return s.Runtime.InitError(w.NewError("network instance is nil"))
	}

	w.Debug("tcp network instance", wool.Field("instance", instance))

	s.Infof("will run on %s", instance.Host)
	s.postgresPort = 5432

	// Create connection string resources for the network instance
	for _, inst := range net.Instances {
		conf, errConn := s.CreateConnectionConfiguration(ctx, configuration, inst, false)
		if errConn != nil {
			return s.Runtime.InitError(errConn)
		}
		w.Debug("adding configuration", wool.Field("config", resources.MakeConfigurationSummary(conf)), wool.Field("instance", inst))
		s.Runtime.RuntimeConfigurations = append(s.Runtime.RuntimeConfigurations, conf)
	}
	s.Wool.Debug("sending runtime configuration", wool.Field("conf", resources.MakeManyConfigurationSummary(s.Runtime.RuntimeConfigurations)))

	w.Debug("setting up connection string for migrations")
	// Setup a connection string for migration
	hostInstance, err := resources.FindNetworkInstanceInNetworkMappings(ctx, s.NetworkMappings, s.TcpEndpoint, resources.NewNativeNetworkAccess())
	if err != nil {
		return s.Runtime.InitError(err)

	}

	connection, err := s.createOwnerConnectionString(ctx, configuration, hostInstance.Address, false)
	if err != nil {
		return s.Runtime.InitError(err)
	}
	s.connection, err = withConnectTimeout(connection, s.Settings.Timeouts.connect())
	if err != nil {
		return s.Runtime.InitError(err)
	}

	// Configuration (postgres user/password) is needed by both runtimes.
	err = s.LoadConfiguration(ctx, configuration)
	if err != nil {
		return s.Runtime.InitError(err)
	}

	// Native runtime: this service's database on its run's one postgres
	// server (nixpg.go), published at this service's own endpoint
	// (forward.go). Same connection strings and database as the Docker path,
	// so the rest of the agent is unchanged.
	if native {
		w.Debug("using the run's native postgres server", wool.Field("port", instance.Port))
		// An interrupted native run leaves System V resources the postmaster
		// never got to remove, and a leaked semaphore run poisons every later
		// start on this host. Collect them before a server is started, while
		// nothing this invocation owns is attached to them yet.
		s.reapHostResources(ctx)
		if errNative := s.initNative(ctx, instance); errNative != nil {
			return s.Runtime.InitError(errNative)
		}
		return s.Runtime.InitResponse()
	}

	// Docker
	runner, err := dockerrun.NewDockerHeadlessEnvironment(ctx, s.dockerImage(), s.UniqueWithWorkspace())
	if err != nil {
		return s.Runtime.InitError(err)
	}
	dataPath, err := mountPersistentPostgresData(ctx, runner)
	if err != nil {
		return s.Runtime.InitError(s.Wool.Wrapf(err, "cannot configure persistent postgres data"))
	}
	s.retainedDataPath = dataPath

	runner.WithOutput(newPGLogWriter(s.Wool))
	runner.WithPortMapping(ctx, uint16(instance.Port), s.postgresPort)

	runner.WithEnvironmentVariables(
		ctx,
		resources.Env("POSTGRES_USER", s.postgresUser),
		resources.Env("POSTGRES_PASSWORD", s.postgresPassword),
		resources.Env("POSTGRES_DB", s.DatabaseName))

	// Quieten the server when a log level is configured. The official
	// postgres image's ENTRYPOINT is docker-entrypoint.sh and its
	// default CMD is `postgres`. WithCommand only overrides CMD, so
	// we provide just `postgres <flags>` — the entrypoint still runs
	// initdb on first boot, then execs our postgres + flags. The
	// `-c` args are passed straight to the server and override the
	// equivalent postgresql.conf entries.
	if lvl := strings.ToLower(strings.TrimSpace(s.LogLevel)); lvl != "" {
		runner.WithCommand(
			"postgres",
			"-c", "log_min_messages="+lvl,
			"-c", "log_statement=none",
			"-c", "log_connections=off",
			"-c", "log_disconnections=off",
		)
	}

	s.runnerEnvironment = runner

	w.Debug("init for runner environment: will start container")
	err = s.runnerEnvironment.Init(ctx)
	if err != nil {
		return s.Runtime.InitError(err)
	}

	s.released = false

	s.Wool.Debug("init successful")
	if err := s.migrateOnInit(ctx); err != nil {
		return s.Runtime.InitError(err)
	}
	return s.Runtime.InitResponse()
}

// migrateOnInit applies schema migrations DURING Init — after the database is
// up but BEFORE Init returns. This closes a readiness race: the codefly
// --exclude-root readiness gate every WithDependencies consumer uses is a plain
// TCP dial on the postgres port (cli/pkg/orchestration/flow.go networkMapping
// TCPReachable), and that port opens in Init. When migrations ran only in Start
// (after Init exposed the port), a fast consumer could be told "ready" and
// connect mid-migration — reading an incomplete schema, or tearing the stack
// down and leaving golang-migrate "dirty" at a random version. Running them in
// Init makes "port reachable" imply "schema migrated". Start still calls
// applyMigration; with the schema already current it is an idempotent no-op
// (migrate.ErrNoChange).
func (s *Runtime) migrateOnInit(ctx context.Context) error {
	// Resolve the declared prerequisites BEFORE waiting on the database: a typo
	// in a source path, a duplicate lineage, or an unsafe name must fail without
	// touching a schema.
	prerequisites, err := s.resolveSchemaPrerequisites()
	if err != nil {
		return err
	}
	if err := s.WaitForReady(ctx); err != nil {
		return err
	}
	// Extensions, migrations and grants are ONE control-plane transition, so
	// they are taken under ONE acquisition. Releasing between them lets a
	// hot-reload change apply a forward migration in the gap, so the grants
	// this transition computes describe a schema that has already moved on: the
	// tables that migration created fall outside them, and no runtime role can
	// read them until something reconciles again.
	release, acquireErr := s.controlPlane.acquire(ctx)
	if acquireErr != nil {
		return acquireErr
	}
	defer release()

	if err := s.applySchema(ctx, prerequisites); err != nil {
		return err
	}
	return s.ensureRuntimeAccessLocked(ctx)
}

// applySchema brings the database up to the resolved prerequisites: extensions
// first, so migration files can rely on them, then every migration lineage.
// Extensions are not migrations — they are ensured even when NoMigration is set,
// so "port reachable" also implies "declared extensions available".
func (s *Runtime) applySchema(ctx context.Context, prerequisites *schemaPrerequisites) error {
	// Deferred so a skipped prerequisite is reported even when a later step
	// fails: a migration that needs an extension reports only the missing
	// function, and the skip is the reason it is missing.
	defer s.reportSchemaPrerequisites(prerequisites)
	if err := s.ensureExtensions(ctx, prerequisites); err != nil {
		return err
	}
	if !s.Settings.NoMigration {
		if err := s.applyMigration(ctx, prerequisites.sources); err != nil {
			return err
		}
	}
	return nil
}

// ensureExtensions CREATE EXTENSION IF NOT EXISTS for every resolved request.
// An explicitly declared extension is required: its failure — an absent shared
// library, a migration owner without the privilege, a lost connection — fails
// readiness rather than leaving the service running against a schema it cannot
// use. The convenience defaults and declarations marked optional record a
// structured skip instead.
func (s *Runtime) ensureExtensions(ctx context.Context, prerequisites *schemaPrerequisites) error {
	db, err := sql.Open("postgres", s.extensionConnection())
	if err != nil {
		return s.Wool.Wrapf(err, "cannot open database to create extensions")
	}
	defer db.Close()

	for _, extension := range prerequisites.extensions {
		// Extension names cannot be parameterized; resolveExtensions restricts
		// them to [A-Za-z0-9_-] so the quoted identifier is injection-safe.
		if _, err := db.ExecContext(ctx, `CREATE EXTENSION IF NOT EXISTS "`+extension.name+`"`); err != nil {
			if extension.required {
				return s.Wool.Wrapf(err,
					"cannot create required extension %q: the configured image must ship its library and the migration owner must be allowed to install it",
					extension.name)
			}
			prerequisites.skipped = append(prerequisites.skipped, skippedPrerequisite{
				kind:   prerequisiteExtension,
				name:   extension.name,
				reason: err.Error(),
			})
			continue
		}
		s.Wool.Debug("extension ready", wool.Field("extension", extension.name))
	}
	return nil
}

// validExtName reports whether name is a safe postgres extension identifier.
func validExtName(name string) bool {
	if name == "" {
		return false
	}
	for _, r := range name {
		ok := r == '_' || r == '-' ||
			(r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9')
		if !ok {
			return false
		}
	}
	return true
}

func (s *Runtime) WaitForReady(ctx context.Context) error {
	defer s.Wool.Catch()
	ctx = s.Wool.Inject(ctx)

	budget := s.Settings.Timeouts.readiness()
	s.Wool.Debug("waiting for database readiness", wool.Field("budget", budget.String()))

	// One pool, opened once and reused for every probe. sql.Open is lazy
	// (it doesn't dial until Ping), so a single *sql.DB pinged in a loop is
	// the idiomatic readiness check. The old code opened a NEW *sql.DB every
	// iteration and never closed any of them — up to 30 leaked connection
	// pools per Init, which alone can exhaust Postgres' default 100-conn limit.
	db, err := sql.Open("postgres", s.connection)
	if err != nil {
		return s.Wool.Wrapf(err, "cannot open database")
	}
	defer db.Close()

	deadline, cancel := context.WithTimeout(ctx, budget)
	defer cancel()

	retry := time.NewTicker(readinessProbeInterval)
	defer retry.Stop()

	// The last probe failure that came from the database rather than from our
	// own budget running out: that is what tells the user why postgres never
	// answered, and it is lost if a cancellation overwrites it.
	var lastProbeErr error
	for {
		err = probeReady(deadline, db)
		if err == nil {
			s.Wool.Debug("database ready!")
			return nil
		}
		if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
			lastProbeErr = err
		}
		s.Wool.Debug("waiting for database to be ready", wool.ErrField(err))
		select {
		case <-deadline.Done():
			return s.readinessFailure(ctx, budget, deadline.Err(), lastProbeErr)
		case <-retry.C:
		}
	}
}

// probeReady answers whether postgres is accepting work, and returns as soon as
// ctx is done. Establishment is bounded by the DSN's connect_timeout rather
// than by ctx — libpq reads the startup handshake off a raw socket deadline —
// so a peer that accepts TCP and then goes silent would otherwise hold a
// cancelled caller for the whole connect budget.
func probeReady(ctx context.Context, db *sql.DB) error {
	probe := make(chan error, 1)
	go func() {
		if err := db.PingContext(ctx); err != nil {
			probe <- err
			return
		}
		// A backend that accepts connections is not necessarily one that runs
		// queries: postgres answers the handshake during crash recovery and
		// refuses statements until it finishes.
		_, err := db.ExecContext(ctx, "SELECT 1")
		probe <- err
	}()
	select {
	case err := <-probe:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// readinessFailure names the phase, says whether the budget ran out or the
// caller gave up, and carries the last database-side failure. The connection
// string is never included: it holds the migration-owner password.
func (s *Runtime) readinessFailure(ctx context.Context, budget time.Duration, ended, lastProbeErr error) error {
	phase := fmt.Sprintf("database readiness budget of %s expired", budget)
	if errors.Is(ended, context.Canceled) {
		phase = "database readiness wait cancelled"
	}
	if lastProbeErr == nil {
		lastProbeErr = ended
	}
	// Tail container logs so the user sees the real failure (bad CMD,
	// disk full, port collision, ...) instead of a generic timeout.
	tail := ""
	if s.runnerEnvironment != nil {
		tail = s.runnerEnvironment.TailLogs(ctx, 30)
	}
	if tail != "" {
		return s.Wool.NewError("%s (last probe: %v); container logs (tail 30):\n%s", phase, lastProbeErr, tail)
	}
	return s.Wool.NewError("%s (last probe: %v)", phase, lastProbeErr)
}

func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runtimev0.StartResponse, error) {
	defer s.Wool.Catch()
	ctx = s.Wool.Inject(ctx)

	s.Wool.Debug("starting")

	// Stop released this invocation's postmaster or container, and nothing but
	// Init brings one back. Without this, WaitForReady spends ninety seconds
	// probing a database that was deliberately shut down before reporting a
	// generic timeout.
	if s.released {
		return s.Runtime.StartError(s.Wool.NewError("postgres was stopped by this invocation: run Init before Start"))
	}

	s.Wool.Debug("waiting for ready")

	prerequisites, err := s.resolveSchemaPrerequisites()
	if err != nil {
		return s.Runtime.StartError(err)
	}

	err = s.WaitForReady(ctx)
	if err != nil {
		return s.Runtime.StartError(err)
	}

	// One acquisition over the whole transition, for the reason migrateOnInit
	// documents. Arming the watcher inside it is deliberate: a save landing
	// while Start is still migrating queues rather than applying against a
	// half-applied schema.
	release, acquireErr := s.controlPlane.acquire(ctx)
	if acquireErr != nil {
		return s.Runtime.StartError(acquireErr)
	}
	defer release()

	if err := s.applySchema(ctx, prerequisites); err != nil {
		return s.Runtime.StartError(err)
	}

	if !s.Settings.NoMigration && s.Settings.HotReload {
		watch, errWatch := s.migrationWatchRequirements(ctx)
		if errWatch != nil {
			s.Wool.Warn("cannot resolve migration watch roots", wool.ErrField(errWatch))
		} else if errWatch = s.SetupWatcher(ctx, services.NewWatchConfiguration(watch), s.EventHandler); errWatch != nil {
			s.Wool.Warn("error in watcher", wool.ErrField(errWatch))
		}
	}
	if err := s.ensureRuntimeAccessLocked(ctx); err != nil {
		return s.Runtime.StartError(err)
	}
	s.Wool.Debug("start done")
	// Commit StartStatus=STARTED BEFORE arming supervision. StartResponse and
	// MarkRunnerExited both write StartStatus under the same lock, so ordering is
	// by wall-clock, not data race: if Supervise were armed first, a postmaster
	// death already buffered in serverExit (a crash in the Init->Start window)
	// would let the watcher goroutine flip StartStatus to ERROR, and this
	// StartResponse would then clobber it back to STARTED — masking the very
	// death we exist to report (codefly-dev/cli#380). Arming after the commit
	// guarantees any ERROR the watcher writes lands strictly after STARTED.
	resp, err := s.Runtime.StartResponse()
	if err != nil {
		return resp, err
	}
	// Report a mid-run death of the managed postgres process so codefly's Follow
	// loop observes StartStatus ERROR and tears down loudly instead of leaving
	// dependents to spin on connection refused (codefly-dev/cli#380). Docker
	// runtimes are supervised by the container engine and reported via the
	// runner environment; the Nix host process has no such supervisor.
	if s.nixRuntime != nil {
		s.nixRuntime.Supervise(func(err error) {
			if err != nil {
				s.Wool.Error("nix postgres exited unexpectedly", wool.ErrField(err))
			} else {
				s.Wool.Error("nix postgres exited unexpectedly (clean exit, not stopped)")
			}
			s.Runtime.MarkRunnerExited(err)
		})
	}
	return resp, nil
}

// reapHostResources collects host resources a previous native run leaked.
// A failure is reported rather than returned: these resources belong to runs
// this invocation does not own, several agents can reach for the same ones at
// once, and losing that race must not be what fails a start or a teardown that
// otherwise succeeded. The authenticated recovery command returns the failure
// instead, because collecting them is that caller's whole purpose.
//
// On the teardown paths this is opportunistic rather than a guarantee. Proc.Stop
// returns once the SIGTERM grace window elapses, so a postmaster that is still
// exiting still owns its segment and is correctly left alone, and a Destroy
// arriving on a spent context cannot sweep at all. Init is what makes the
// guarantee: it runs unconditionally before every native start, so anything
// missed here is collected before it can affect a cluster.
func (s *Runtime) reapHostResources(ctx context.Context) {
	recovered, err := recoverHostResources(ctx)
	if err != nil {
		s.Wool.Warn("cannot recover orphaned host resources", wool.ErrField(err))
		return
	}
	if !recovered.empty() {
		s.Wool.Info("recovered orphaned host resources", wool.Field("summary", recovered.summary()))
	}
}

func (s *Runtime) Information(ctx context.Context, req *runtimev0.InformationRequest) (*runtimev0.InformationResponse, error) {
	return s.Runtime.InformationResponse(ctx, req)
}

// Postgres lifecycle operations share one state-ownership contract, and it is
// the same contract on both backends:
//
//	operation             execution resources           database state
//	Stop                  released                      retained
//	Stop + keep-running   retained (docker only)        retained
//	Destroy               released, container removed   retained
//
// keep-running is the one row that is not backend-independent: reuse means the
// next invocation reattaches to a live server, and only the container backend
// can do that. See Stop.
//
// Releasing execution resources means terminating the postmaster (nix) or
// stopping the container (docker) so the assigned port is free again and a
// later Init cannot end up running a second postmaster against the same data
// directory.
//
// No operation deletes data. Dropping a database or a data directory requires
// an explicit authorization proving that state is disposable and owned by the
// caller; neither StopRequest nor DestroyRequest carries one, so an agent must
// never infer the right to reset from the fact that it started the process
// (codefly-dev/core#424).
//
// Stop releases only what THIS invocation holds a handle to: a runtime that
// never ran Init owns no postmaster and no container, and stops nothing.
func (s *Runtime) Stop(ctx context.Context, req *runtimev0.StopRequest) (*runtimev0.StopResponse, error) {
	defer s.Wool.Catch()
	ctx = s.Wool.Inject(ctx)

	s.lifecycleMu.Lock()
	defer s.lifecycleMu.Unlock()

	// Tear the hot-reload watcher down on EVERY path, the keep-running early
	// return below included. core's SetupWatcher REPLACES s.Events/s.Watcher/
	// s.watcherCancel without cancelling what was already there, so a Stop that
	// leaves the watcher running makes the next Start leak a second one, and
	// every leaked watcher still holds a live reference to EventHandler. They
	// fire against a database this invocation has already released, and each
	// restart adds another: duplicate apply attempts the forward-only path then
	// reports as already-applied, duplicate grant reconciliations, and an
	// fsnotify handle plus two goroutines leaked per restart. core documents
	// StopWatcher as the mandatory teardown, and it is idempotent when no
	// watcher is running. Under lifecycleMu because it writes s.watcherCancel,
	// which Destroy also clears.
	s.StopWatcher()

	// keep-running asks for a server the NEXT invocation reattaches to, and only
	// the container backend has one per service: GetContainer adopts an
	// existing container by name. The native server belongs to the run, not to
	// this service, and stops when the run's last service releases it, so one
	// service cannot keep it up for itself. Native releases and keeps the half
	// of the contract it can: the data.
	if s.Settings.KeepRunning {
		if s.nixRuntime == nil && !s.Runtime.IsNixRuntime() {
			s.Wool.Debug("keep-running is set: leaving postgres up for reuse")
			return s.stopResponse("kept postgres running for reuse")
		}
		s.Wool.Warn("keep-running is not available on the nix runtime: releasing postgres, its data is retained")
	}

	if s.nixRuntime != nil {
		action, err := s.releaseNative(ctx)
		if err != nil {
			return s.Runtime.StopError(err)
		}
		s.reapHostResources(ctx)
		return s.stopResponse(action)
	}

	if s.runnerEnvironment != nil {
		if err := s.stopContainer(ctx); err != nil {
			return s.Runtime.StopError(err)
		}
		// The docker handle stays usable after its container stops: stopping
		// twice is a no-op, and Destroy still needs it to close the client and
		// log stream this invocation opened.
		s.released = true
		return s.stopResponse("stopped postgres container")
	}

	return s.stopResponse("no postgres owned by this invocation")
}

func (s *Runtime) Destroy(ctx context.Context, req *runtimev0.DestroyRequest) (*runtimev0.DestroyResponse, error) {
	defer s.Wool.Catch()
	ctx = s.Wool.Inject(ctx)

	s.lifecycleMu.Lock()
	defer s.lifecycleMu.Unlock()

	s.Wool.Debug("Destroying")

	// A leaked watcher would keep firing EventHandler at a database that is
	// being removed. Idempotent when Stop already ran.
	s.StopWatcher()

	if s.nixRuntime != nil {
		action, err := s.releaseNative(ctx)
		if err != nil {
			return s.Runtime.DestroyError(err)
		}
		s.reapHostResources(ctx)
		return s.destroyResponse(action)
	}
	// A nix invocation has no container to remove, and reaching for one would
	// demand a docker daemon from a host that was very likely chosen for not
	// having one.
	if s.Runtime.IsNixRuntime() {
		return s.destroyResponse("native postgres already stopped")
	}

	// Prefer the environment this invocation started: shutting that one down
	// also closes the docker client and log stream it opened. Resolving the
	// container by name remains the fallback so destroy keeps working for a
	// caller that only loaded the service.
	runner := s.runnerEnvironment
	if runner == nil {
		var err error
		runner, err = dockerrun.NewDockerHeadlessEnvironment(ctx, s.dockerImage(), s.UniqueWithWorkspace())
		if err != nil {
			return s.Runtime.DestroyError(err)
		}
	}
	if err := runner.Shutdown(ctx); err != nil {
		return s.Runtime.DestroyError(err)
	}
	s.runnerEnvironment = nil
	s.released = true
	return s.destroyResponse("removed postgres container")
}

// stopContainer stops the container, retrying once for a caller that is still
// listening. Core bounds its stop call at ten seconds against the docker
// daemon, and a loaded daemon can blow that deadline after it has already
// accepted the stop: the container is on its way down but the call reports
// failure. A second call is a no-op against a container that has since stopped
// and a real second attempt against one that has not.
//
// The retry is only worth spending when someone will still read the answer.
// codefly's own teardown bounds Stop at ten seconds — the same budget the first
// attempt just consumed — so by now it has abandoned this call and moved on to
// Destroy; retrying would spend another ten seconds producing a status nobody
// receives, while holding the handle Destroy is waiting for. Callers without
// that deadline (the MCP stop tool passes its request context straight through)
// do still get the second attempt.
func (s *Runtime) stopContainer(ctx context.Context) error {
	err := s.runnerEnvironment.Stop(ctx)
	if err == nil {
		return nil
	}
	if ctxErr := ctx.Err(); ctxErr != nil {
		return err
	}
	s.Wool.Warn("postgres container stop did not confirm; retrying", wool.ErrField(err))
	return s.runnerEnvironment.Stop(ctx)
}

// stopResponse and destroyResponse report SUCCESS together with what the
// operation released and where the state it kept now lives, so a caller learns
// the retention outcome from the result instead of by inspecting the backend.
// Only paths are reported, never a credential or a connection string.
func (s *Runtime) stopResponse(action string) (*runtimev0.StopResponse, error) {
	return &runtimev0.StopResponse{Status: &runtimev0.StopStatus{
		State:   runtimev0.StopStatus_SUCCESS,
		Message: s.retentionSummary(action),
	}}, nil
}

func (s *Runtime) destroyResponse(action string) (*runtimev0.DestroyResponse, error) {
	return &runtimev0.DestroyResponse{Status: &runtimev0.DestroyStatus{
		State:   runtimev0.DestroyStatus_SUCCESS,
		Message: s.retentionSummary(action),
	}}, nil
}

func (s *Runtime) retentionSummary(action string) string {
	if s.retainedDataPath == "" {
		return action + "; database state retained"
	}
	return action + "; database state retained at " + s.retainedDataPath
}

func (s *Runtime) Test(ctx context.Context, req *runtimev0.TestRequest) (*runtimev0.TestResponse, error) {
	// A RUN-LEVEL verdict, stated. This agent ships no tests of its own -- it
	// runs a datastore -- so the honest run is zero tests and zero failures,
	// which is what the status-only answer always meant. It has to be said:
	// core's helpers fill the counts but leave Result nil, and the CLI promotes
	// nothing to a pass -- only an explicit TestRunResult_PASSED is one
	// (cli pkg/testrun/verdict.go); UNKNOWN is refused by name.
	response, err := s.Runtime.TestResponseWithResults(0, 0, 0, 0, 0, nil, nil)
	if err != nil {
		return nil, err
	}
	response.Result = &runtimev0.TestRunResult{
		State:   runtimev0.TestRunResult_PASSED,
		Message: "this agent ships no tests of its own: zero run, zero failed",
	}
	return response, nil
}

/* Details

 */

// EventHandler reacts to one watched file change. Errors are reported and
// swallowed: an edit that cannot be applied — an already-applied migration, a
// dirty ledger, bad SQL — must leave hot reload armed so fixing the source or
// adding a new migration gets another attempt.
func (s *Runtime) EventHandler(event code.Change) error {
	ctx := context.Background()

	// The change and the reconciliation that grants its new tables are ONE
	// transition, taken under ONE acquisition, so an Init or Start transition
	// cannot interleave and reconcile grants over a schema this handler is
	// still moving. applyMigrationChange takes migrationReload inside this,
	// never the other way round, so the two locks have one order.
	release, acquireErr := s.controlPlane.acquire(ctx)
	if acquireErr != nil {
		return acquireErr
	}
	defer release()

	applied, err := s.applyMigrationChange(ctx, event.Path)
	if err != nil {
		// Reported, not returned: on the forward-only path an edit to an
		// already-applied migration is a routine condition, not a failure.
		s.Wool.Warn("cannot apply migration change", wool.ErrField(err))
	}
	// Reconcile whenever SQL reached the database: an apply failure reports
	// applied=false, so an error here is a cleanup failure over a schema that
	// already changed and still needs its runtime grants.
	if !applied {
		return nil
	}
	if err := s.ensureRuntimeAccessLocked(ctx); err != nil {
		s.Wool.Warn("cannot reconcile runtime access after migration", wool.ErrField(err))
	}
	return nil
}