felhom-controller/controller/internal/web/netstorage_job.go

package web

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-controller/internal/agentapi"
	"gitea.dooplex.hu/admin/felhom-controller/internal/settings"
)

// NAS add orchestration (verify-before-commit). POST /api/storage/netstorage/add no longer
// registers blind: it validates sync, then a DETACHED job (migrate.go shape — sync-validate-then-
// goroutine, single-flight, deep-copied status) drives:
//
//	agent_add  → the agent installs units + starts ITS detached verify (auto-rollback on fail)
//	verifying  → poll the agent's verify slot every 2 s; "none" after install = agent restarted
//	             mid-verify ⇒ WE roll back (Scenario F)
//	probing    → in-guest uid-1000 write probe (the squash trap — a mountable-but-unwritable
//	             export must never register)
//	registering→ AddStoragePath — the LAST step, so the worst crash outcome is an agent-side
//	             orphan (surfaced by the list), never a registered-but-broken path
//	done | failed
//
// The job runs on context.Background() with an overall deadline — a closed browser tab can never
// abort a half-done add mid-rollback. Rollback (agent RemoveNetStorage) is idempotent; a
// double-remove after the agent's own auto-rollback is harmless.

// netAgent is the agent surface the orchestrator needs (seam — tests inject a fake; production is
// the shared *agentapi.Client).
type netAgent interface {
	AddNetStorage(ctx context.Context, req agentapi.AddNetStorageRequest) (agentapi.NetStorageAddResult, error)
	NetVerifyStatus(ctx context.Context) (agentapi.NetVerifyStatus, error)
	RemoveNetStorage(ctx context.Context, name string) error
}

// netAddJob is the poll-visible orchestration state (GET /api/storage/netstorage/add/status).
type netAddJob struct {
	Name      string    `json:"name"`
	Phase     string    `json:"phase"`              // agent_add | verifying | probing | registering | done | failed
	Category  string    `json:"category,omitempty"` // failure category (agent classifier vocabulary + not_writable/probe_io)
	Message   string    `json:"message,omitempty"`  // the category's Hungarian customer message (§3.2)
	Detail    string    `json:"detail,omitempty"`   // raw English/journal detail (the collapsible)
	Warn      string    `json:"warn,omitempty"`     // non-fatal note on a successful add (e.g. probe cleanup)
	Path      string    `json:"path,omitempty"`     // the registered in-guest path (done only)
	StartedAt time.Time `json:"started_at"`
	UpdatedAt time.Time `json:"updated_at"`
}

const (
	netAddPhaseAgentAdd    = "agent_add"
	netAddPhaseVerifying   = "verifying"
	netAddPhaseProbing     = "probing"
	netAddPhaseRegistering = "registering"
	netAddPhaseDone        = "done"
	netAddPhaseFailed      = "failed"
)

const (
	// netAddDeadline bounds the WHOLE orchestration (§8: detached ctx + ~150 s). The dominant term
	// is the agent verify's own 95 s worst case (black-holed-but-routed server).
	netAddDeadline = 150 * time.Second
	// netVerifyPollEvery is the agent verify-slot poll cadence.
	netVerifyPollEvery = 2 * time.Second
	// netRollbackBudget bounds a rollback call on a possibly-expired job context.
	netRollbackBudget = 30 * time.Second
)

// netAddState is the single-flight slot (migrate.go's acquire/release + deep-copy status shape).
type netAddState struct {
	mu      sync.Mutex
	running bool
	cur     *netAddJob
}

func (s *netAddState) acquire(job *netAddJob) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.running {
		return false
	}
	s.running = true
	cp := *job
	s.cur = &cp
	return true
}

func (s *netAddState) release() {
	s.mu.Lock()
	s.running = false
	s.mu.Unlock()
}

func (s *netAddState) set(job *netAddJob) {
	s.mu.Lock()
	cp := *job
	s.cur = &cp
	s.mu.Unlock()
}

// snapshot returns a copy of the last / in-flight job (nil = never ran).
func (s *netAddState) snapshot() *netAddJob {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.cur == nil {
		return nil
	}
	cp := *s.cur
	return &cp
}

// netAgentForAdd resolves the orchestrator's agent surface (test seam first, then the shared client).
func (s *Server) netAgentForAdd() (netAgent, error) {
	if s.netAgentFn != nil {
		return s.netAgentFn()
	}
	c, err := s.agentClient()
	if err != nil {
		return nil, err
	}
	return c, nil
}

// netProbe runs the in-guest uid-1000 write probe (test seam first, then the re-exec runner).
func (s *Server) netProbe(ctx context.Context, dir string) probeOutcome {
	if s.netProbeFn != nil {
		return s.netProbeFn(ctx, dir)
	}
	return runNetProbe(ctx, dir)
}

// startNetAdd claims the single-flight slot and launches the detached orchestration. false = an add
// is already in flight (Scenario G).
func (s *Server) startNetAdd(agent netAgent, req agentapi.AddNetStorageRequest, label string) bool {
	job := &netAddJob{
		Name:      req.Name,
		Phase:     netAddPhaseAgentAdd,
		StartedAt: time.Now().UTC(),
		UpdatedAt: time.Now().UTC(),
	}
	if !s.netAdd.acquire(job) {
		return false
	}
	go s.runNetAdd(agent, req, label, job)
	return true
}

// runNetAdd drives the phase machine on a DETACHED context (a closed tab can't abort a rollback).
func (s *Server) runNetAdd(agent netAgent, req agentapi.AddNetStorageRequest, label string, job *netAddJob) {
	defer s.netAdd.release()
	ctx, cancel := context.WithTimeout(context.Background(), netAddDeadline)
	defer cancel()

	setPhase := func(p string) {
		job.Phase = p
		job.UpdatedAt = time.Now().UTC()
		s.netAdd.set(job)
	}
	fail := func(category, detail string) {
		job.Phase = netAddPhaseFailed
		job.Category = category
		job.Message = netAddMessage(category, req.Server, req.MappedUID)
		job.Detail = detail
		job.UpdatedAt = time.Now().UTC()
		s.netAdd.set(job)
		s.logger.Printf("[WARN] [web] netstorage add %q failed: category=%s detail=%s", req.Name, category, detail)
	}
	rollback := func(why string) {
		rctx, rcancel := context.WithTimeout(context.Background(), netRollbackBudget)
		defer rcancel()
		if err := agent.RemoveNetStorage(rctx, req.Name); err != nil {
			// Best-effort by design: the agent may have auto-rolled-back already (double-remove is
			// harmless) — but log it, a REAL leftover shows up as an orphan row in the list.
			s.logger.Printf("[WARN] [web] netstorage add %q rollback (%s): remove: %v", req.Name, why, err)
		}
	}

	// Phase 1 — agent add (install units + start the agent-side verify).
	res, err := agent.AddNetStorage(ctx, req)
	if err != nil {
		var refused *agentapi.NetAddRefusedError
		if errors.As(err, &refused) {
			fail(refused.Code, refused.Msg) // categorized sync refusal — NOTHING was installed
			return
		}
		fail("agent_error", err.Error())
		return
	}

	// Phase 2 — poll the agent's verify slot. On a pre-verify agent (no job started) skip straight
	// to the probe: the mount-trigger check then happens implicitly through the probe's write.
	if res.Verify == "started" {
		setPhase(netAddPhaseVerifying)
		verdict, verr := s.pollAgentVerify(ctx, agent, res.JobID)
		switch {
		case verr != nil:
			rollback("verify poll failed")
			fail("agent_error", "verify status unavailable: "+verr.Error())
			return
		case verdict.Phase == "failed":
			// The agent has ALREADY auto-rolled-back (units + creds) — no controller rollback needed.
			fail(verdict.Code, verdict.Detail)
			return
		case verdict.Phase == "done":
			// verified — proceed
		default:
			// "none" (agent restarted mid-verify — Scenario F) or an alien job id: the verify is
			// LOST, the units may linger — roll back and fail loud.
			rollback("verify lost (agent restart?)")
			fail("mount_failed", fmt.Sprintf("agent verify job lost after install (phase=%q) — rolled back", verdict.Phase))
			return
		}
	}

	// Phase 3 — the in-guest uid-1000 write probe (the squash trap).
	setPhase(netAddPhaseProbing)
	outcome := s.netProbe(ctx, res.GuestPath)
	if !outcome.OK {
		rollback("probe failed")
		fail(outcome.Category, outcome.Detail)
		return
	}

	// Phase 4 — register. THE LAST STEP (worst crash outcome = an agent-side orphan, never a
	// registered-but-broken path).
	setPhase(netAddPhaseRegistering)
	sp := settings.StoragePath{
		Path:        res.GuestPath,
		Label:       label,
		Schedulable: true,
		AddedAt:     time.Now().UTC().Format(time.RFC3339),
		Kind:        settings.StorageKindNetwork,
		Protocol:    req.Protocol,
		Server:      req.Server,
		Export:      req.Export,
		MappedUID:   req.MappedUID,
		MappedGID:   req.MappedGID,
	}
	if err := s.settings.AddStoragePath(sp); err != nil {
		rollback("register failed")
		fail("register_failed", err.Error())
		return
	}

	job.Phase = netAddPhaseDone
	job.Path = res.GuestPath
	job.Warn = outcome.Warn
	job.UpdatedAt = time.Now().UTC()
	s.netAdd.set(job)
	s.logger.Printf("[INFO] [web] network storage added + verified: %s (%s %s:%s) → %s (warn=%q)",
		req.Name, req.Protocol, req.Server, req.Export, res.GuestPath, outcome.Warn)
}

// pollAgentVerify polls the agent's verify slot until it leaves `running` (or ctx expires).
// Transient poll errors are tolerated up to a small budget — the agent may be busy mounting.
func (s *Server) pollAgentVerify(ctx context.Context, agent netAgent, jobID string) (agentapi.NetVerifyStatus, error) {
	var last agentapi.NetVerifyStatus
	consecutiveErrs := 0
	t := time.NewTicker(netVerifyPollEvery)
	defer t.Stop()
	for {
		st, err := agent.NetVerifyStatus(ctx)
		if err != nil {
			consecutiveErrs++
			if consecutiveErrs >= 5 {
				return last, err
			}
		} else {
			consecutiveErrs = 0
			last = st
			if st.Phase != "running" {
				// A different job id in the slot means OUR job is gone (restart + a newer add) —
				// report it as the lost-verify shape, not a false done/failed.
				if st.Phase != "none" && jobID != "" && st.JobID != jobID {
					last.Phase = "none"
				}
				return last, nil
			}
		}
		select {
		case <-ctx.Done():
			return last, ctx.Err()
		case <-t.C:
		}
	}
}

// netAddMessage maps a failure category to the EXACT Hungarian customer message (§3.2). server and
// mappedUID parameterize the unreachable/not_writable texts (host id = mapped uid + 100000).
func netAddMessage(category, server string, mappedUID int) string {
	switch category {
	case "unreachable":
		return "A szerver nem érhető el (" + server + "). Ellenőrizze az IP-címet, és hogy a NAS be van-e kapcsolva."
	case "nfs_export":
		return "A megosztás nem található, vagy a NAS nem engedélyezi ennek a gépnek a hozzáférését. Ellenőrizze az export útvonalát, és hogy a NAS engedélyezi-e a Felhom gép IP-címét."
	case "smb_auth":
		return "Hibás SMB felhasználónév vagy jelszó."
	case "smb_share":
		return "A megadott SMB-megosztás nem található a szerveren. Ellenőrizze a megosztás nevét."
	case "timeout":
		return "Időtúllépés a csatolás közben — a szerver elérhető a hálózaton, de a megosztás nem csatolható. Ellenőrizze a NAS NFS/SMB szolgáltatását."
	case "not_writable":
		return "A megosztás csatolható, de az alkalmazások nem tudnak rá írni. NFS esetén kapcsolja be a NAS-on a „minden felhasználó leképezése” (map all users / all squash) beállítást a megosztáson — vagy állítsa a fájlok tulajdonosát a(z) " + fmt.Sprint(mappedUID+100000) + " azonosítóra. SMB esetén ellenőrizze, hogy a felhasználónak írási joga van a megosztáson."
	case "probe_io":
		return "Írási hiba a megosztáson (az adat nem olvasható vissza hibátlanul). Ellenőrizze a megosztást és a hálózatot."
	case "busy":
		return "Már folyamatban van egy csatlakoztatás. Várja meg, amíg befejeződik."
	default: // mount_failed + agent_error + register_failed + any future agent code
		return "A csatolás sikertelen. Részletek alább."
	}
}

// trimNetDetail bounds the raw detail shown in the UI collapsible.
func trimNetDetail(d string) string {
	d = strings.TrimSpace(d)
	if len(d) > 800 {
		return d[:800] + "…"
	}
	return d
}