dsh-web/internal/api/upgrade.go

package api

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"syscall"
	"time"

	"git.misaka.ren/m1saka/dsh-web/internal/backup"
	"git.misaka.ren/m1saka/dsh-web/internal/store"
)

const upgradeSnapshotLimit int64 = 1 << 30

func upgradeWorkspace(config, host map[string]json.RawMessage, recipe, home string) error {
	var mounts []map[string]json.RawMessage
	if err := json.Unmarshal(host["Mounts"], &mounts); err != nil {
		return errors.New("unsupported runtime mounts")
	}
	get := func(m map[string]json.RawMessage, key string) string {
		var value string
		_ = json.Unmarshal(m[key], &value)
		return value
	}
	if !supportedLayout(recipe) {
		return errors.New("unsupported template layout; template 7 required")
	}
	if len(mounts) != 1 || get(mounts[0], "Type") != "bind" || get(mounts[0], "Source") != home || get(mounts[0], "Target") != "/workspace" {
		return errors.New("independent instance home at /workspace required")
	}
	for _, key := range []string{"Binds", "VolumesFrom"} {
		var values []string
		if raw := host[key]; len(raw) != 0 {
			if json.Unmarshal(raw, &values) != nil || len(values) != 0 {
				return errors.New("unsupported additional mounts")
			}
		}
	}
	var env []string
	if raw := config["Env"]; len(raw) != 0 {
		if err := json.Unmarshal(raw, &env); err != nil {
			return err
		}
	}
	kept := make([]string, 0, len(env)+4)
	for _, entry := range env {
		key, _, _ := strings.Cut(entry, "=")
		switch key {
		case "DSH_HOME", "HOME", "XDG_DATA_HOME", "XDG_CONFIG_HOME":
		default:
			kept = append(kept, entry)
		}
	}
	kept = append(kept, "DSH_HOME=/workspace", "HOME=/workspace", "XDG_DATA_HOME=/workspace/.local/share", "XDG_CONFIG_HOME=/workspace/.config")
	config["Env"], _ = json.Marshal(kept)
	return nil
}

func upgradeRoot(root string, i store.Instance) (*os.Root, error) {
	r, err := os.OpenRoot(root)
	if err != nil {
		return nil, err
	}
	for _, part := range []string{"workspaces", i.UID, "instances", i.ID} {
		if part == "" || part == "." || part == ".." || strings.ContainsAny(part, "/\\") {
			r.Close()
			return nil, errors.New("invalid storage path")
		}
		st, e := r.Lstat(part)
		if e != nil || !st.IsDir() {
			r.Close()
			return nil, errors.New("unsafe storage path")
		}
		next, e := r.OpenRoot(part)
		r.Close()
		if e != nil {
			return nil, e
		}
		r = next
	}
	return r, nil
}

func upgradeSync(r *os.Root) error {
	f, err := r.Open(".")
	if err != nil {
		return err
	}
	defer f.Close()
	return f.Sync()
}

func upgradeCopy(ctx context.Context, r *os.Root, source, target string, limit int64) error {
	remaining, entries := limit, 100000
	var copyTree func(string, string, int) error
	copyTree = func(src, dst string, depth int) (copyErr error) {
		defer func() {
			if copyErr != nil {
				if pathErr, ok := copyErr.(*os.PathError); ok {
					copyErr = fmt.Errorf("snapshot copy: %w", &os.PathError{Op: pathErr.Op, Path: src, Err: pathErr.Err})
					return
				}
				copyErr = fmt.Errorf("%s: %w", src, copyErr)
			}
		}()
		if err := ctx.Err(); err != nil {
			return err
		}
		entries--
		if entries < 0 || depth > 128 {
			return errors.New("snapshot entry limit")
		}
		st, err := r.Lstat(src)
		if err != nil {
			return err
		}
		if st.IsDir() {
			if err = r.Mkdir(dst, 0700); err != nil {
				return err
			}
			f, err := r.OpenFile(src, os.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_DIRECTORY, 0)
			if err != nil {
				return err
			}
			defer f.Close()
			for {
				children, err := f.ReadDir(128)
				if err != nil && err != io.EOF {
					return err
				}
				for _, child := range children {
					if err := copyTree(filepath.Join(src, child.Name()), filepath.Join(dst, child.Name()), depth+1); err != nil {
						return err
					}
				}
				if err == io.EOF {
					break
				}
			}
			if err = r.Chown(dst, 1000, 1000); err != nil {
				return err
			}
			if err = r.Chmod(dst, st.Mode().Perm()); err != nil {
				return err
			}
			d, err := r.Open(dst)
			if err != nil {
				return err
			}
			defer d.Close()
			return d.Sync()
		}
		if st.Mode()&os.ModeSymlink != 0 && depth > 0 {
			link, err := r.Readlink(src)
			if err != nil {
				return err
			}
			// Preserve the link itself, including dangling and container-absolute
			// targets. Never resolve it against the panel's filesystem.
			if err = r.Symlink(link, dst); err != nil {
				return err
			}
			return r.Lchown(dst, 1000, 1000)
		}
		if !st.Mode().IsRegular() || st.Size() > remaining {
			return errors.New("unsafe or oversized snapshot")
		}
		in, err := r.OpenFile(src, os.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, 0)
		if err != nil {
			return err
		}
		defer in.Close()
		actual, err := in.Stat()
		if err != nil || !actual.Mode().IsRegular() || !os.SameFile(st, actual) {
			return errors.New("snapshot source changed")
		}
		out, err := r.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_EXCL|syscall.O_NOFOLLOW, 0600)
		if err != nil {
			return err
		}
		defer out.Close()
		n, err := io.Copy(out, io.LimitReader(in, remaining+1))
		if err != nil {
			return err
		}
		remaining -= n
		if remaining < 0 {
			return errors.New("snapshot size limit")
		}
		if err = out.Chown(1000, 1000); err != nil {
			return err
		}
		if err = out.Chmod(st.Mode().Perm()); err != nil {
			return err
		}
		return out.Sync()
	}
	if err := copyTree(source, target, 0); err != nil {
		return err
	}
	return upgradeSync(r)
}

// Call while holding the instance lock before any operation that can start or mutate a runtime.
func (a *API) upgradeRecoveryRequired(i store.Instance) bool {
	var exists int
	if err := a.s.DB.QueryRow("SELECT count(*) FROM sqlite_master WHERE type='table' AND name='upgrade_journal'").Scan(&exists); err != nil {
		return true
	}
	if exists == 0 {
		return false
	}
	var pending int
	if err := a.s.DB.QueryRow("SELECT count(*) FROM upgrade_journal WHERE instance_id=? AND phase NOT IN ('complete','rolled_back')", i.ID).Scan(&pending); err != nil {
		return true
	}
	if pending != 0 {
		a.s.DB.Exec("UPDATE instances SET status='error' WHERE id=?", i.ID)
	}
	return pending != 0
}

func (a *API) upgrade(w http.ResponseWriter, r *http.Request, i store.Instance) {
	if r.Method != http.MethodPost {
		fail(w, 405, "method not allowed")
		return
	}
	var in struct {
		ImageID  string `json:"image_id"`
		Password string `json:"password"`
	}
	dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10))
	dec.DisallowUnknownFields()
	if dec.Decode(&in) != nil || !runtimeImageID.MatchString(in.ImageID) || dec.Decode(new(any)) != io.EOF {
		fail(w, 400, "invalid image_id")
		return
	}
	inst := a.instanceLocks.Acquire(i.ID)
	defer inst.Release()
	a.readersMu.Lock()
	readers := a.workspaceReaders[i.ID]
	a.readersMu.Unlock()
	if readers > 0 {
		fail(w, 409, "instance files busy; retry later")
		return
	}
	user := a.userLocks.Acquire(i.UID)
	defer user.Release()
	admissionMu.RLock()
	defer admissionMu.RUnlock()
	current, err := a.s.GetInstance(i.ID)
	if err != nil || current.UID != i.UID {
		fail(w, 404, "not found")
		return
	}
	i = current
	if len(in.Password) == 0 || len(in.Password) > 72 {
		fail(w, 400, "login password required for pre-upgrade backup")
		return
	}
	if !a.reauthenticate(w, i.UID, in.Password) {
		return
	}
	if a.upgradeRecoveryRequired(i) || i.Status == "error" {
		fail(w, 409, "instance requires recovery")
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 3*time.Minute)
	defer cancel()
	var old struct {
		Config     map[string]json.RawMessage
		HostConfig map[string]json.RawMessage
		State      struct {
			Status                      string
			Running, Paused, Restarting bool
		}
		NetworkSettings struct{ Networks map[string]json.RawMessage }
	}
	if i.ContainerID == "" || a.docker.Do(ctx, "GET", "/containers/"+url.PathEscape(i.ContainerID)+"/json", nil, &old) != nil {
		fail(w, 502, "container inspection failed")
		return
	}
	if old.State.Paused || old.State.Restarting || (old.State.Status != "running" && old.State.Status != "exited" && old.State.Status != "created") {
		fail(w, 409, "unsupported container state")
		return
	}
	if old.Config == nil || old.HostConfig == nil {
		fail(w, 502, "container configuration unavailable")
		return
	}
	verified, err := a.verifiedImages(ctx)
	if err != nil {
		fail(w, 500, "database error")
		return
	}
	var image struct {
		ID     string `json:"Id"`
		Config struct{ Labels map[string]string }
	}
	if a.docker.Do(ctx, "GET", "/images/"+url.PathEscape(in.ImageID)+"/json", nil, &image) != nil || image.ID != in.ImageID || image.Config.Labels["dsh-mgr.role"] != "runtime" || !supportedLayout(image.Config.Labels["dsh-mgr.recipe"]) || verified[image.ID] == "" || verified[image.ID] != image.Config.Labels["dsh-mgr.version"] {
		fail(w, 400, "local runtime image required")
		return
	}
	root, err := upgradeRoot(a.root, i)
	if err != nil {
		fail(w, 500, "storage unavailable")
		return
	}
	defer root.Close()
	hostRoot := os.Getenv("HOST_DATA_ROOT")
	if hostRoot == "" {
		hostRoot = a.root
	}
	homeSource, homeErr := toHostPath(a.root, hostRoot, filepath.Join(root.Name(), "home"))
	if homeErr != nil {
		fail(w, 500, "host storage unavailable")
		return
	}
	if err = upgradeWorkspace(old.Config, old.HostConfig, image.Config.Labels["dsh-mgr.recipe"], homeSource); err != nil {
		fail(w, 409, "incompatible runtime workspace layout")
		return
	}
	if _, err = a.quotaSnapshot(i.UID, root, 0); err != nil {
		fail(w, 413, "quota exceeded or snapshot size unavailable")
		return
	}
	op := store.ID()
	snapshot, failedHome := "upgrade-"+op+"-snapshot", "failed-home-"+op
	if a.recordSnapshot(i, snapshot, i.Version, i.ImageID) != nil || a.recordSnapshot(i, failedHome, image.Config.Labels["dsh-mgr.version"], image.ID) != nil {
		fail(w, 500, "snapshot metadata unavailable")
		return
	}
	newName := runtimeNetwork(i.ID) + "-upgrade-" + op
	_, err = a.s.DB.Exec(`CREATE TABLE IF NOT EXISTS upgrade_journal (id TEXT PRIMARY KEY, instance_id TEXT NOT NULL, old_container TEXT NOT NULL, new_container TEXT NOT NULL DEFAULT '', new_name TEXT NOT NULL, image_id TEXT NOT NULL, snapshot TEXT NOT NULL, failed_home TEXT NOT NULL, phase TEXT NOT NULL)`)
	if err != nil {
		fail(w, 500, "upgrade journal unavailable")
		return
	}
	_, err = a.s.DB.Exec("CREATE TABLE IF NOT EXISTS upgrade_state (id TEXT PRIMARY KEY, old_status TEXT NOT NULL, was_running INTEGER NOT NULL)")
	if err != nil {
		fail(w, 500, "upgrade journal unavailable")
		return
	}
	tx, err := a.s.DB.Begin()
	if err != nil {
		fail(w, 500, "database error")
		return
	}
	_, err = tx.Exec("INSERT INTO upgrade_journal(id,instance_id,old_container,new_name,image_id,snapshot,failed_home,phase) VALUES(?,?,?,?,?,?,?,'snapshotting')", op, i.ID, i.ContainerID, newName, image.ID, snapshot, failedHome)
	if err == nil {
		_, err = tx.Exec("INSERT INTO upgrade_state VALUES(?,?,?)", op, old.State.Status, old.State.Running)
	}
	if err == nil {
		_, err = tx.Exec("UPDATE instances SET status='error' WHERE id=?", i.ID)
	}
	if err != nil {
		tx.Rollback()
		fail(w, 500, "database error")
		return
	}
	if tx.Commit() != nil {
		fail(w, 500, "database error")
		return
	}
	phase := func(p string) error {
		_, err := a.s.DB.Exec("UPDATE upgrade_journal SET phase=? WHERE id=?", p, op)
		return err
	}
	newID, complete, snapReady, createAttempted := "", false, false, false
	stopAttempted := false
	defer func() {
		if complete {
			return
		}
		cleanup, done := context.WithTimeout(context.Background(), 30*time.Second)
		defer done()
		if phase("rolling_back") != nil {
			return
		}
		if createAttempted {
			id := newID
			if id == "" {
				id = newName
			}
			// A timed-out create can still have succeeded; the journaled unique name is authoritative.
			_ = a.docker.Do(cleanup, "POST", "/containers/"+url.PathEscape(id)+"/stop?t=10", nil, nil)
			if a.removeRuntimeContainer(cleanup, id) != nil {
				return
			}
		}
		if snapReady && createAttempted {
			if phase("saving_failed_home") != nil {
				return
			}
			if root.Rename("home", failedHome) != nil || upgradeSync(root) != nil {
				return
			}
			if phase("restoring_home") != nil {
				return
			}
			if root.Rename(snapshot, "home") != nil || upgradeSync(root) != nil {
				return
			}
		}
		if stopAttempted && old.State.Running {
			if phase("restoring_runtime") != nil || a.docker.Do(cleanup, "POST", "/containers/"+url.PathEscape(i.ContainerID)+"/start", nil, nil) != nil {
				return
			}
		}
		runtimeAuthCache.Delete(runtimeAuthKey{a, i.ID})
		tx, err := a.s.DB.Begin()
		if err != nil {
			return
		}
		defer tx.Rollback()
		if _, err = tx.Exec("UPDATE upgrade_journal SET phase='rolled_back' WHERE id=?", op); err != nil {
			return
		}
		if _, err = tx.Exec("UPDATE instances SET status=? WHERE id=?", old.State.Status, i.ID); err != nil {
			return
		}
		_ = tx.Commit()
	}()
	if old.State.Running {
		if phase("stopping") != nil {
			fail(w, 500, "upgrade journal unavailable")
			return
		}
		stopAttempted = true
		if a.docker.Do(ctx, "POST", "/containers/"+url.PathEscape(i.ContainerID)+"/stop?t=10", nil, nil) != nil {
			fail(w, 502, "old runtime stop failed")
			return
		}
	}
	stopped, inspectErr := a.docker.Inspect(ctx, i.ContainerID)
	if inspectErr != nil || stopped.State.Running || stopped.State.Paused || stopped.State.Restarting {
		fail(w, 502, "old runtime did not stop")
		return
	}
	if phase("snapshotting") != nil {
		fail(w, 500, "upgrade journal unavailable")
		return
	}
	snapshotBytes, quotaErr := a.quotaSnapshot(i.UID, root, 0)
	if quotaErr != nil {
		fail(w, 413, "quota exceeded or snapshot size unavailable")
		return
	}
	// Only publish a completed copy under a name the snapshot API recognizes.
	staging := "." + snapshot + "-partial"
	if _, err := a.s.DB.Exec("CREATE TABLE IF NOT EXISTS snapshot_copy_state (instance_id TEXT NOT NULL, id TEXT NOT NULL, ready INTEGER NOT NULL, PRIMARY KEY(instance_id,id))"); err != nil {
		fail(w, 500, "snapshot completion metadata unavailable")
		return
	}
	if _, err := a.s.DB.Exec("INSERT INTO snapshot_copy_state VALUES(?,?,0)", i.ID, snapshot); err != nil {
		fail(w, 500, "snapshot completion metadata unavailable")
		return
	}
	if copyErr := upgradeCopy(ctx, root, "home", staging, snapshotBytes); copyErr != nil {
		if cleanupErr := root.RemoveAll(staging); cleanupErr != nil {
			a.s.AddAudit(i.UID, "upgrade.snapshot.cleanup_failed", i.ID, staging)
			fail(w, 500, "home snapshot failed: "+copyErr.Error()+"; partial snapshot cleanup required")
			return
		}
		fail(w, 500, "home snapshot failed: "+copyErr.Error())
		return
	}
	if err := root.Rename(staging, snapshot); err != nil {
		if root.RemoveAll(staging) != nil {
			a.s.AddAudit(i.UID, "upgrade.snapshot.cleanup_failed", i.ID, staging)
		}
		fail(w, 500, "home snapshot publication failed")
		return
	}
	if upgradeSync(root) != nil {
		fail(w, 500, "home snapshot publication sync failed")
		return
	}
	if _, err := a.s.DB.Exec("UPDATE snapshot_copy_state SET ready=1 WHERE instance_id=? AND id=?", i.ID, snapshot); err != nil {
		fail(w, 500, "snapshot completion metadata unavailable")
		return
	}
	snapReady = true
	backupDir := filepath.Join(a.root, "backups", i.UID, i.ID)
	storage, err := os.OpenRoot(a.root)
	if err != nil {
		fail(w, 500, "backup storage unavailable")
		return
	}
	err = configCheckPath(storage, filepath.ToSlash(filepath.Join("backups", i.UID, i.ID, "archive")))
	storage.Close()
	if err != nil || os.MkdirAll(backupDir, 0700) != nil {
		fail(w, 500, "unsafe backup storage")
		return
	}
	entries, err := os.ReadDir(backupDir)
	if err != nil || len(entries) >= 20 {
		fail(w, 409, "backup retention limit reached")
		return
	}
	backupID := store.ID()
	backupPath := filepath.Join(backupDir, backupID+".dshbak")
	if err = backup.CreateWithPassword(filepath.Join(root.Name(), "home"), backupPath, in.Password, backup.Metadata{DSHVersion: i.Version, CreatedAt: time.Now().UTC()}); err != nil {
		fail(w, 500, "pre-upgrade backup failed")
		return
	}
	backupStat, err := os.Stat(backupPath)
	if err != nil {
		fail(w, 500, "pre-upgrade backup unavailable")
		return
	}
	if err = a.indexBackup(i, backupID, "pre-upgrade", 2, backupStat.Size()); err != nil {
		os.Remove(backupPath)
		fail(w, 500, "backup index unavailable")
		return
	}
	delete(old.Config, "Hostname")
	old.Config["Image"], _ = json.Marshal(image.ID)
	old.Config["Cmd"], _ = json.Marshal([]string{"web", "--no-open", "--port", "3080", "--trusted-host", a.runtimeHost(r)})
	old.Config["HostConfig"], _ = json.Marshal(old.HostConfig)
	endpoints := map[string]any{}
	for network := range old.NetworkSettings.Networks {
		endpoints[network] = map[string]any{}
	}
	old.Config["NetworkingConfig"], _ = json.Marshal(map[string]any{"EndpointsConfig": endpoints})
	if phase("creating") != nil {
		fail(w, 500, "upgrade journal unavailable")
		return
	}
	createAttempted = true
	var created struct {
		ID string `json:"Id"`
	}
	if a.docker.Do(ctx, "POST", "/containers/create?name="+url.QueryEscape(newName), old.Config, &created) != nil || created.ID == "" {
		fail(w, 502, "upgrade container creation failed")
		return
	}
	newID = created.ID
	if _, err = a.s.DB.Exec("UPDATE upgrade_journal SET new_container=?, phase='starting' WHERE id=?", newID, op); err != nil {
		fail(w, 500, "upgrade journal unavailable")
		return
	}
	if a.docker.Do(ctx, "POST", "/containers/"+url.PathEscape(newID)+"/start", nil, nil) != nil {
		fail(w, 502, "upgrade start failed")
		return
	}
	if phase("readiness") != nil {
		fail(w, 500, "upgrade journal unavailable")
		return
	}
	if a.upgradeReady(ctx, newID, i.ID, a.runtimeHost(r)) != nil {
		fail(w, 502, "upgrade readiness failed")
		return
	}
	tx, err = a.s.DB.Begin()
	if err != nil {
		fail(w, 500, "database error")
		return
	}
	defer tx.Rollback()
	_, err = tx.Exec("UPDATE instances SET container_id=?,image_id=?,version=?,status='running' WHERE id=?", newID, image.ID, image.Config.Labels["dsh-mgr.version"], i.ID)
	if err == nil {
		_, err = tx.Exec("UPDATE upgrade_journal SET phase='committed' WHERE id=?", op)
	}
	if err != nil {
		fail(w, 500, "database error")
		return
	}
	if tx.Commit() != nil {
		fail(w, 500, "database error")
		return
	}
	complete = true
	if a.removeRuntimeContainer(ctx, i.ContainerID) != nil || phase("complete") != nil {
		a.s.SetInstance(i.ID, "error")
		fail(w, 502, "upgrade committed; old container cleanup requires recovery")
		return
	}
	runtimeAuthCache.Delete(runtimeAuthKey{a, i.ID})
	i.ContainerID, i.ImageID, i.Version, i.Status = newID, image.ID, image.Config.Labels["dsh-mgr.version"], "running"
	a.s.AddAudit(i.UID, "instance.upgrade", i.ID, image.ID)
	writeJSON(w, i)
}

func (a *API) upgradeReady(ctx context.Context, id, instanceID, host string) error {
	ctx, cancel := context.WithTimeout(ctx, 150*time.Second)
	defer cancel()
	ticker := time.NewTicker(250 * time.Millisecond)
	defer ticker.Stop()
	for {
		c, err := a.docker.Inspect(ctx, id)
		if err != nil {
			return errors.New("readiness inspection failed")
		}
		if !c.State.Running {
			return errors.New("runtime stopped")
		}
		logs, err := a.docker.Logs(ctx, id)
		if err == nil && latestRuntimeToken(logs) != "" && (c.State.Health == nil || c.State.Health.Status == "healthy") {
			{
				ip := c.NetworkSettings.Networks[runtimeNetwork(instanceID)].IPAddress
				if net.ParseIP(ip) != nil {
					target := &url.URL{Scheme: "http", Host: net.JoinHostPort(ip, "8080")}
					if cookies, e := exchangeRuntimeCookie(ctx, target, host, latestRuntimeToken(logs)); e == nil {
						req, _ := http.NewRequestWithContext(ctx, "GET", target.String()+"/", nil)
						req.Host = host
						req.Header.Set("Cookie", cookies)
						client := &http.Client{Transport: runtimeTransport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
						if resp, e := client.Do(req); e == nil {
							resp.Body.Close()
							if resp.StatusCode == http.StatusOK {
								return nil
							}
						}
					}
				}
			}
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}