// Package handler — HTTP handlers wired to the OpenTofu-based provisioner.
//
// Per docs/INVIOLABLE-PRINCIPLES.md principle #3 + docs/ARCHITECTURE.md §10:
// Phase 0 cloud provisioning is OpenTofu's job, NOT bespoke Go code. This
// handler invokes `tofu apply` against the canonical infra/hetzner/ module
// and streams the output to the wizard via SSE.
//
// Phase 1 — bootstrap-kit installation (Cilium → cert-manager → Flux →
// Crossplane → ... → bp-catalyst-platform) — runs INSIDE the new
// Sovereign cluster via Flux reconciling clusters/<sovereign-fqdn>/ in
// the public OpenOva monorepo. catalyst-api does NOT orchestrate that
// (per Lesson #24, never call helm/kubectl), but it DOES OBSERVE Phase
// 1's HelmRelease state via a read-only client-go dynamic informer
// against the new Sovereign's kubeconfig (internal/helmwatch). The
// observed state flows back through the same SSE buffer Phase 0 used,
// surfacing per-component pills ("cilium installing → installed") for
// the Sovereign Admin's app cards.
package handler

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/go-chi/chi/v5"

	"github.com/openova-io/openova/products/catalyst/bootstrap/api/internal/auth"
	"github.com/openova-io/openova/products/catalyst/bootstrap/api/internal/helmwatch"
	"github.com/openova-io/openova/products/catalyst/bootstrap/api/internal/jobs"
	"github.com/openova-io/openova/products/catalyst/bootstrap/api/internal/pdm"
	"github.com/openova-io/openova/products/catalyst/bootstrap/api/internal/provisioner"
	"github.com/openova-io/openova/products/catalyst/bootstrap/api/internal/store"
)

// EventBufferCap bounds the per-deployment in-memory event slice. A long-
// running multi-region `tofu apply` can emit thousands of stdout lines; if
// the buffer ever reaches this cap we drop the oldest entry (FIFO) so a
// runaway producer cannot OOM the catalyst-api Pod. 10,000 events ≈ 1MB at
// typical event size — well within the Pod's memory budget.
const EventBufferCap = 10000

// Deployment captures provisioning state for a single Sovereign run.
//
// Events flow lives in two parallel structures:
//
//   - eventsCh — the live SSE channel. runProvisioning closes this when
//     BOTH the Phase-0 OpenTofu provisioning AND (when launched) the
//     Phase-1 HelmRelease watch have finished. StreamLogs ranges over
//     it; when it closes, the SSE stream emits `event: done`.
//   - eventsBuf — a bounded, mutex-guarded slice of every event ever
//     emitted for this deployment. StreamLogs reads this on first
//     connection so a browser that lands on the page AFTER
//     provisioning finished still renders the full history. GET
//     /events surfaces the same slice as JSON for any client that
//     wants a one-shot snapshot.
//
// done is closed once runProvisioning has finished AND the Phase-1
// watch has either terminated or was never launched (Phase-0 failure).
// StreamLogs uses it to know when a deployment is fully complete
// (replay-then-emit-done) versus still running (replay-then-tail-
// channel).
type Deployment struct {
	ID         string
	Status     string // pending | provisioning | tofu-applying | flux-bootstrapping | phase1-watching | ready | failed
	Request    provisioner.Request
	Result     *provisioner.Result
	Error      string
	StartedAt  time.Time
	FinishedAt time.Time

	// eventsCh carries live events to the active SSE consumer.
	// runProvisioning + the Phase-1 watch goroutine both emit through
	// it; closed once both have finished.
	eventsCh chan provisioner.Event

	// eventsBuf is the durable history every emitted event lands in. Mutex
	// guarded by mu. Bounded at EventBufferCap with FIFO eviction.
	eventsBuf []provisioner.Event

	// done is closed when runProvisioning's full lifecycle (Phase 0 +
	// optional Phase 1 watch) has finished and the terminal fields
	// (Status, Result, Error, FinishedAt, ComponentStates,
	// Phase1FinishedAt) are committed under mu.
	done chan struct{}

	mu sync.Mutex

	// PDM reservation captured before `tofu apply` for managed-pool
	// deployments. The reservationToken is held until `tofu apply`
	// returns the LB IP, at which point we POST it to PDM /commit. On
	// `tofu destroy` (or a phase-0 retry that decides to abandon) we
	// DELETE /release.
	//
	// Empty for BYO deployments — those keep their own DNS off-platform.
	pdmReservationToken string
	pdmPoolDomain       string
	pdmSubdomain        string

	// AdoptedAt — set by the handover finalisation flow (issue #317) once
	// the customer has accepted operational ownership of their Sovereign.
	// Until set, console.openova.io/sovereign/<id> renders the normal
	// Catalyst-Zero admin shell. Once set, the UI's beforeLoad redirect
	// (issue #319) sends the operator to https://console.<sovereign-fqdn>
	// instead — the customer's Sovereign is now self-sufficient and
	// Catalyst-Zero retains only the minimum-retention surface (PDM
	// allocation row + parent-zone NS delegation for pool subdomains).
	//
	// This field is the contract between #317 (writer) and #319
	// (reader). #317's handover handler stamps it; #319's redirect +
	// decommission UI consume it.
	AdoptedAt *time.Time `json:"adoptedAt,omitempty"`

	// kubeconfigBearerHash — hex-encoded SHA-256 of the 32-byte bearer
	// token templated into the new Sovereign's cloud-init (issue #183).
	// Persisted on the on-disk record so a Pod restart can still
	// verify a delayed cloud-init PUT after the original Pod died.
	// The plaintext bearer NEVER lives on this struct — it is
	// generated in CreateDeployment, stamped onto the
	// provisioner.Request just long enough for writeTfvars to render
	// it, and then GC'd. The only durable copies are the cloud-init
	// template inside the Sovereign's user_data and the SHA-256 hash
	// here.
	kubeconfigBearerHash string

	// OwnerEmail — the email address of the operator who created this
	// deployment (issue #689). Stamped at CreateDeployment time from the
	// authenticated session (X-User-Email injected by RequireSession).
	// Read by checkOwnership() on every /deployments/{id}* handler so a
	// signed-in operator cannot read or mutate another operator's
	// deployment. Empty for legacy records persisted before #689 — the
	// check skips with a log warning when empty so an in-place upgrade
	// doesn't lock operators out of pre-existing rows.
	OwnerEmail string

	// phase1Started gates the at-most-once launch of the Phase-1
	// helmwatch goroutine. Two callers race to start it:
	//   - runProvisioning, after `tofu apply` finishes
	//   - PutKubeconfig, after cloud-init posts back the kubeconfig
	// In the typical happy path cloud-init reaches healthz before
	// `tofu apply` returns (the LB reconcile is the slowest part of
	// Phase 0), so PutKubeconfig wins. A guard ensures the loser is
	// a no-op rather than racing two informers against the same
	// HelmRelease list.
	phase1Started bool

	// jobsBridge — per-deployment helmwatch → Job/Execution/LogLine
	// bridge (issue #205, sub of epic #204). Allocated on first
	// component event in emitWatchEvent. Nil-tolerant: the emit path
	// no-ops the forward when bridge is nil OR the Handler's jobs
	// store is nil (tests without persistence).
	jobsBridge *jobs.Bridge

	// liveWatcher — pointer to the helmwatch.Watcher currently
	// driving Phase-1 events for this deployment, populated by
	// runPhase1Watch / resumePhase1Watch / refreshWatch. The
	// /components/state endpoint reads SnapshotComponents() off it
	// to return the in-memory informer cache as JSON; the
	// /refresh-watch endpoint short-circuits to "already running"
	// when this is non-nil. Cleared after Watch() returns so a
	// subsequent /refresh-watch can spin up a fresh informer (the
	// previous Watcher's GVR informer is single-shot).
	//
	// Holding a pointer here is safe — Watcher is goroutine-safe
	// for SnapshotComponents() and the GC reclaims the old one
	// once the field is overwritten.
	liveWatcher *helmwatch.Watcher
}

// SlimForHandover returns a copy of the receiver retaining ONLY the
// fields a post-handover record needs:
//
//   - id, sovereignFQDN, createdAt (StartedAt), createdBy (Request.OrgEmail)
//     for audit;
//   - AdoptedAt (= adoptedAt) for the redirect contract consumed by
//     UI/router.tsx::maybeRedirectToCustomerConsole (issue #319).
//
// All operational/runtime state (Result, TofuState, kubeconfig path on
// Result.KubeconfigPath, PDM bearer hash, pdm reservation token,
// component states, error string, finishedAt) is zeroed. The slim
// shape is what is persisted by FinaliseHandover (issue #453) instead
// of deleting the record entirely — deletion would 404 the redirect
// fetch and the customer console would never be reached.
//
// Per docs/INVIOLABLE-PRINCIPLES.md §0 minimum-retention: Catalyst-Zero
// keeps ONLY the structural breadcrumb required for the redirect to
// fire; everything else lives on the Sovereign side post-handover.
//
// The returned Deployment carries closed eventsCh + done channels so
// any leftover SSE consumer attempts on the slim record terminate
// immediately (matching the fromRecord rehydration shape).
func (d *Deployment) SlimForHandover(adoptedAt time.Time) *Deployment {
	closedCh := make(chan provisioner.Event)
	closedDone := make(chan struct{})
	close(closedCh)
	close(closedDone)

	stamped := adoptedAt.UTC()

	d.mu.Lock()
	id := d.ID
	startedAt := d.StartedAt
	orgName := d.Request.OrgName
	orgEmail := d.Request.OrgEmail
	sovereignFQDN := d.Request.SovereignFQDN
	d.mu.Unlock()

	return &Deployment{
		ID:     id,
		Status: "adopted",
		// Request retains ONLY the audit-minimum context (orgName,
		// orgEmail, sovereignFQDN). All credentials + sizing + region
		// detail are intentionally zeroed — they belong to the
		// Sovereign side now.
		Request: provisioner.Request{
			OrgName:       orgName,
			OrgEmail:      orgEmail,
			SovereignFQDN: sovereignFQDN,
		},
		StartedAt: startedAt,
		// FinishedAt is also "adoptedAt" semantically — set to the same
		// stamp so the wizard's terminal-state surfaces show a coherent
		// timeline.
		FinishedAt: stamped,
		AdoptedAt:  &stamped,
		eventsCh:   closedCh,
		done:       closedDone,
	}
}

// recordEvent appends ev to the durable history under the mutex, evicting
// the oldest entry when the buffer is at cap. Returns the event back so
// callers can fluently send it down the live channel.
func (d *Deployment) recordEvent(ev provisioner.Event) provisioner.Event {
	d.mu.Lock()
	if len(d.eventsBuf) >= EventBufferCap {
		// FIFO eviction — drop oldest, keep newest. Allocate a fresh slice
		// so the underlying array doesn't keep growing without bound when
		// the cap is hit repeatedly (a copy is O(n) but the cap is bounded
		// so the amortised cost is fine for the single emit-per-line rate).
		copy(d.eventsBuf, d.eventsBuf[1:])
		d.eventsBuf = d.eventsBuf[:len(d.eventsBuf)-1]
	}
	d.eventsBuf = append(d.eventsBuf, ev)
	d.mu.Unlock()
	return ev
}

// snapshotEvents returns a copy of the durable history for safe iteration
// outside the mutex. Used by StreamLogs (replay-on-connect) and the
// /events endpoint.
func (d *Deployment) snapshotEvents() []provisioner.Event {
	d.mu.Lock()
	defer d.mu.Unlock()
	out := make([]provisioner.Event, len(d.eventsBuf))
	copy(out, d.eventsBuf)
	return out
}

// isDone reports whether the runProvisioning goroutine has finished. Used
// by StreamLogs to distinguish "completed deployment, replay everything
// then send done" from "in-flight, replay then tail channel".
func (d *Deployment) isDone() bool {
	select {
	case <-d.done:
		return true
	default:
		return false
	}
}

// toRecord serializes the deployment for the on-disk store. Caller MUST
// hold dep.mu — every persistence path (recordEventAndPersist on append,
// runProvisioning on terminal state, CreateDeployment on row creation,
// retry handlers on retry-init) calls this under the lock so the
// snapshot is internally consistent.
//
// The credential fields are dropped via store.Redact; see
// internal/store/store.go for the redaction list. The PDM reservation
// token is preserved (per-deployment opaque identifier) so a Pod
// restart in the gap between `tofu apply` returning and PDM /commit
// can still complete the commit on rehydration.
func (d *Deployment) toRecord() store.Record {
	return store.Record{
		ID:                   d.ID,
		Status:               d.Status,
		Request:              store.Redact(d.Request),
		Result:               d.Result,
		Error:                d.Error,
		StartedAt:            d.StartedAt,
		FinishedAt:           d.FinishedAt,
		Events:               append([]provisioner.Event(nil), d.eventsBuf...),
		PDMReservationToken:  d.pdmReservationToken,
		PDMPoolDomain:        d.pdmPoolDomain,
		PDMSubdomain:         d.pdmSubdomain,
		KubeconfigBearerHash: d.kubeconfigBearerHash,
		AdoptedAt:            d.AdoptedAt,
		OwnerEmail:           d.OwnerEmail,
	}
}

// fromRecord rehydrates a Deployment from an on-disk record. Used by
// Handler.restoreFromStore at startup.
//
// The eventsCh and done channels are created closed: the runProvisioning
// goroutine no longer exists for a deployment loaded from disk (the
// catalyst-api Pod that ran it died), so the SSE replay path must
// behave as if the deployment is finished. StreamLogs.isDone() is the
// branch that distinguishes "in-flight, tail the channel" from
// "completed, replay then emit done"; loaded deployments take the
// completed branch unconditionally.
//
// If the on-disk Status is in-flight ("provisioning" / "tofu-applying"
// / "pending"), it is rewritten to "failed" with an explanatory error
// per the architectural requirement: a Pod restart during `tofu apply`
// orphans real Hetzner resources, and the wizard's FailureCard MUST
// surface that to the operator instead of showing a stuck progress
// bar forever. The orphaned resources are listed in the error message
// so the operator knows where to clean up.
func fromRecord(rec store.Record) *Deployment {
	closedCh := make(chan provisioner.Event)
	closedDone := make(chan struct{})
	close(closedCh)
	close(closedDone)

	dep := &Deployment{
		ID:                   rec.ID,
		Status:               rec.Status,
		Request:              rec.Request.ToProvisionerRequest(),
		Result:               rec.Result,
		Error:                rec.Error,
		StartedAt:            rec.StartedAt,
		FinishedAt:           rec.FinishedAt,
		eventsCh:             closedCh,
		eventsBuf:            append([]provisioner.Event(nil), rec.Events...),
		done:                 closedDone,
		pdmReservationToken:  rec.PDMReservationToken,
		pdmPoolDomain:        rec.PDMPoolDomain,
		pdmSubdomain:         rec.PDMSubdomain,
		kubeconfigBearerHash: rec.KubeconfigBearerHash,
		AdoptedAt:            rec.AdoptedAt,
		OwnerEmail:           rec.OwnerEmail,
	}

	// Kubeconfig file lost across restart → mark as failed. If the
	// record claims a path but the file is missing, the helmwatch
	// goroutine has nothing to watch with. Per the spec for #183 we
	// surface this distinctly so the operator can investigate (PVC
	// unmount, accidental delete, fs corruption) instead of seeing a
	// stuck progress bar. The deployment record's Status was already
	// "ready" or "failed" pre-restart; we leave that alone, only
	// stamping a stricter Error message when KubeconfigPath has
	// drifted from the file system.
	if rec.Result != nil && rec.Result.KubeconfigPath != "" {
		if _, err := os.Stat(rec.Result.KubeconfigPath); err != nil {
			// Don't downgrade a healthy "ready" to "failed" silently;
			// only flag this when the deployment is otherwise
			// observable. The error message gives the operator
			// enough to grep the PVC.
			dep.Error = "kubeconfig file lost across catalyst-api restart: " + rec.Result.KubeconfigPath + " — " + err.Error()
		}
	}

	// Phase-0 in-flight at restart time → failed. The wizard's
	// FailureCard is the right surface for this state — operator must
	// purge the orphaned cloud resources by hand because catalyst-api
	// can't resume an OpenTofu run mid-apply (state-lock + the workdir
	// on /tmp emptyDir died with the previous Pod).
	//
	// "phase1-watching" is deliberately EXCLUDED from this rewrite (issue
	// #830 Bug 3). Phase 1 is purely observational — Phase 0 has already
	// committed its tofu state, the Hetzner resources exist and are
	// reachable, the kubeconfig is on the PVC, and the bootstrap-kit
	// HelmReleases keep reconciling on the Sovereign cluster regardless
	// of whether catalyst-api's in-memory watcher is alive. The new Pod
	// can re-attach the helmwatch informer and continue streaming
	// per-component state. The on-disk record retains Status=
	// "phase1-watching" so /deployments/{id} keeps reporting the truth
	// (the watch is in-flight, not failed) until the resumed watcher
	// terminates and markPhase1Done flips Status to ready/failed.
	//
	// Without this carve-out, a transient catalyst-api roll mid-Phase-1
	// (image bump, OOM, node maintenance) latches the deployment record
	// to status=failed even though the Sovereign cluster reaches Ready=
	// True for every HelmRelease seconds later — the auto-fire handover
	// then never triggers and the operator is stranded on the wizard
	// page. otech102 incident, 2026-05-04. The downstream resume happens
	// in restoreFromStore via shouldResumePhase1 + resumePhase1Watch.
	if isPhase0InFlightStatus(rec.Status) {
		dep.Status = "failed"
		dep.Error = "catalyst-api restarted during provisioning — this deployment was abandoned mid-apply. Hetzner resources tagged with `catalyst-deployment-id=" + rec.ID + "` are orphans and must be purged manually (hcloud server, lb, network, firewall, ssh-key) before retrying. The wizard cannot resume — start a new deployment."
		dep.FinishedAt = time.Now()
	}
	return dep
}

// isInFlightStatus reports whether a deployment is operationally
// still running. Used by the orphan-PDM-release decision logic to
// avoid releasing a slot that's still active.
func isInFlightStatus(s string) bool {
	switch s {
	case "pending", "provisioning", "tofu-applying", "flux-bootstrapping", "phase1-watching":
		return true
	}
	return false
}

// isPhase0InFlightStatus reports whether a deployment was abandoned
// mid-Phase-0 (a state catalyst-api cannot resume — tofu workdir lives
// on /tmp emptyDir which dies with the Pod). These statuses get
// rewritten to "failed" on restart so the wizard renders FailureCard.
//
// "phase1-watching" is deliberately EXCLUDED — Phase 1 is observational
// and resumable across Pod restarts. See fromRecord's comment for the
// full rationale (issue #830 Bug 3).
func isPhase0InFlightStatus(s string) bool {
	switch s {
	case "pending", "provisioning", "tofu-applying", "flux-bootstrapping":
		return true
	}
	return false
}

// recordEventAndPersist appends ev to the durable history, persists the
// deployment to disk under the lock, and returns the event. This is the
// hot path — every emit goes through here. Persistence is best-effort:
// if disk write fails (full PVC, unmount), we log and continue. The
// in-memory state remains authoritative for the live SSE consumer; the
// on-disk gap is reported through h.log so an operator can spot it.
//
// We persist under d.mu so concurrent emits can't tear the on-disk
// record. The store's own mutex serializes the temp-file rename
// against any concurrent Save (e.g. the terminal-state Save in
// runProvisioning racing the last emitted event).
func (h *Handler) recordEventAndPersist(d *Deployment, ev provisioner.Event) provisioner.Event {
	d.mu.Lock()
	if len(d.eventsBuf) >= EventBufferCap {
		copy(d.eventsBuf, d.eventsBuf[1:])
		d.eventsBuf = d.eventsBuf[:len(d.eventsBuf)-1]
	}
	d.eventsBuf = append(d.eventsBuf, ev)
	rec := d.toRecord()
	d.mu.Unlock()

	if h.store != nil {
		if err := h.store.Save(rec); err != nil {
			h.log.Warn("persist deployment after event failed",
				"id", d.ID,
				"err", err,
			)
		}
	}
	return ev
}

// persistDeployment serializes the current state under the lock and
// writes it. Used at terminal state (runProvisioning end) and at
// row creation (CreateDeployment). Same best-effort policy as
// recordEventAndPersist.
func (h *Handler) persistDeployment(d *Deployment) {
	if h.store == nil {
		return
	}
	d.mu.Lock()
	rec := d.toRecord()
	d.mu.Unlock()
	if err := h.store.Save(rec); err != nil {
		h.log.Warn("persist deployment failed",
			"id", d.ID,
			"err", err,
		)
	}
}

// restoreFromStore reads every record from h.store and registers each
// in h.deployments. Called from New() after the store wires up so the
// catalyst-api process restored from a Pod restart still answers
// /api/v1/deployments/<id> for every deployment the previous Pod knew
// about. Per-file decode failures are logged but do not abort the
// load — the user-facing requirement is "as many deployments as
// possible recovered" not "all-or-nothing".
func (h *Handler) restoreFromStore() {
	if h.store == nil {
		return
	}
	records, err := h.store.LoadAll(func(path string, e error) {
		h.log.Warn("skipping unreadable deployment record",
			"path", path,
			"err", e,
		)
	})
	if err != nil {
		h.log.Error("restoreFromStore: walk failed; running with empty in-memory state",
			"err", err,
		)
		return
	}

	resumed := 0
	for _, rec := range records {
		dep := fromRecord(rec)
		h.deployments.Store(dep.ID, dep)
		// If the load step rewrote a stuck in-flight status to failed,
		// re-persist so the on-disk state matches what we serve. This
		// is the architectural promise: the wizard reading
		// /deployments/<id> after a Pod restart sees a coherent
		// terminal state, not a ghost still labelled "provisioning".
		if dep.Status != rec.Status && h.store != nil {
			if err := h.store.Save(dep.toRecord()); err != nil {
				h.log.Warn("re-persist rewritten in-flight record failed",
					"id", dep.ID,
					"err", err,
				)
			}
			// Pod-restart-orphan PDM release (issue #489). The previous
			// catalyst-api Pod owned the runProvisioning goroutine that
			// would have called pdm.Release on Phase-0 failure. The
			// goroutine died with the Pod; the reservation is now
			// orphaned in PDM, locking the subdomain forever. fromRecord
			// rewrites the status to "failed" precisely BECAUSE the
			// previous Pod was killed mid-apply — the same condition
			// means we MUST release the PDM slot here, otherwise a
			// retry under the same subdomain returns 409 conflict and
			// the franchise customer is forced to pick `acmeN+1` for
			// what should be a routine retry.
			//
			// Best-effort + asynchronous: a slow PDM at startup must
			// NOT block the rest of the rehydration loop (other
			// deployments need to land in sync.Map quickly so /events
			// poll works).
			if dep.Status == "failed" && rec.PDMReservationToken != "" &&
				rec.PDMPoolDomain != "" && rec.PDMSubdomain != "" &&
				dep.AdoptedAt == nil && h.pdm != nil {
				go h.releaseOrphanedReservation(dep.ID, rec.PDMPoolDomain, rec.PDMSubdomain)
			}
		}

		// Resume the Phase-1 helmwatch goroutine after a Pod restart
		// when the rehydrated deployment carries a KubeconfigPath
		// pointing at an existing file (issue #183 spec gate #6).
		// The previous Pod died mid-watch — the kubeconfig file
		// survived on the PVC, so the new Pod can re-attach the
		// informer and continue streaming per-component events to
		// any wizard that polls /events or reopens the SSE stream.
		//
		// We skip resume for deployments whose Phase-1 already
		// terminated (Result.Phase1FinishedAt != nil) — re-running
		// the watcher on a finished deployment would just emit
		// duplicate events. We also skip when the file is missing
		// (fromRecord already stamped a clear Error on dep) and
		// when Status is the rewritten-to-failed phase1-watching
		// case — those are operator-actionable failures, not
		// resumable runs.
		if h.shouldResumePhase1(dep, rec) {
			resumed++
			h.resumePhase1Watch(dep)
		}
	}
	h.log.Info("restored deployments from PVC",
		"count", len(records),
		"resumed", resumed,
		"dir", h.store.Dir(),
	)
}

// releaseOrphanedReservation calls pdm.Release for a deployment whose
// status was rewritten to "failed" because the catalyst-api Pod died
// mid-provisioning (issue #489). Best-effort: any failure is logged
// at warn but does not block other rehydration work. Idempotent — a
// pdm.ErrNotFound response (the reservation already expired or was
// previously released) is treated as success. The dep entry's PDM
// pointers are cleared on success so a follow-up Cancel & Wipe doesn't
// double-release.
//
// Run as a goroutine from restoreFromStore: the rehydration loop must
// not block on PDM. A 30s timeout caps each call.
func (h *Handler) releaseOrphanedReservation(deploymentID, poolDomain, subdomain string) {
	if h.pdm == nil || poolDomain == "" || subdomain == "" {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	err := h.pdm.Release(ctx, poolDomain, subdomain)
	if err != nil && !errors.Is(err, pdm.ErrNotFound) {
		h.log.Warn("orphan-release: pdm release failed; subdomain may stay locked until TTL",
			"id", deploymentID,
			"poolDomain", poolDomain,
			"subdomain", subdomain,
			"err", err,
		)
		return
	}
	h.log.Info("orphan-release: released PDM allocation for pod-restart-orphaned deployment",
		"id", deploymentID,
		"poolDomain", poolDomain,
		"subdomain", subdomain,
	)
	if val, ok := h.deployments.Load(deploymentID); ok {
		dep := val.(*Deployment)
		dep.mu.Lock()
		dep.pdmPoolDomain = ""
		dep.pdmSubdomain = ""
		dep.pdmReservationToken = ""
		dep.mu.Unlock()
		h.persistDeployment(dep)
	}
}

// shouldResumePhase1 returns true when a rehydrated deployment is a
// candidate for re-attaching the Phase-1 helmwatch goroutine. The
// criteria are:
//   - Result.KubeconfigPath is non-empty AND points at a readable file
//   - Phase 1 has NOT already terminated (Phase1FinishedAt == nil)
//   - Original on-disk status was NOT a Phase-0 in-flight state (those
//     are unrecoverable — fromRecord rewrites them to "failed")
//
// Resume IS triggered for the following original statuses:
//   - "phase1-watching" — the canonical durable-watcher case (issue
//     #830 Bug 3). Phase 0 already committed its tofu state, the
//     Sovereign cluster is healthy, and the new Pod re-attaches the
//     helmwatch informer to continue streaming per-component events.
//   - Terminal states ("ready", "failed", "adopted") with
//     Phase1FinishedAt == nil — contract-violation residual; resume is
//     the safer-default action because the helmwatch is idempotent
//     (it just observes HelmRelease.status — no patches/applies).
//
// Why we resume even when rec.Status is "ready" but Phase1FinishedAt
// is nil: that combination is a contract violation — markPhase1Done
// would have set Phase1FinishedAt before flipping Status to ready. If
// we ever observe it, it's a residual bug or hand-edited record;
// resume is the safer-default action because the helmwatch is
// idempotent (it just observes HelmRelease.status).
func (h *Handler) shouldResumePhase1(dep *Deployment, rec store.Record) bool {
	if dep.Result == nil || dep.Result.KubeconfigPath == "" {
		return false
	}
	if dep.Result.Phase1FinishedAt != nil {
		return false
	}
	// Phase-0 in-flight statuses are rewritten to "failed" by
	// fromRecord — those are not resumable (tofu workdir died with
	// the previous Pod, Hetzner resources are orphaned).
	// "phase1-watching" is NOT in this set per issue #830 Bug 3 — the
	// watcher IS resumable, and resuming is the whole point of the
	// durable-watcher fix.
	if isPhase0InFlightStatus(rec.Status) {
		return false
	}
	if _, err := os.Stat(dep.Result.KubeconfigPath); err != nil {
		return false
	}
	return true
}

// resumePhase1Watch re-attaches a Phase-1 helmwatch goroutine to a
// rehydrated deployment. The fromRecord path constructs the
// Deployment with closed eventsCh + done channels (because the
// previous Pod's runProvisioning goroutine no longer exists); we
// allocate fresh ones here so emitWatchEvent + StreamLogs see a
// live channel pair. A goroutine then runs runPhase1Watch which
// closes them when the watch terminates.
func (h *Handler) resumePhase1Watch(dep *Deployment) {
	dep.mu.Lock()
	dep.eventsCh = make(chan provisioner.Event, 256)
	dep.done = make(chan struct{})
	// The watcher will set phase1Started=true inside runPhase1Watch
	// under the same lock; we leave it false here so the launch is
	// correctly gated.
	dep.phase1Started = false
	// Re-flag the deployment as in-flight so isDone()/StreamLogs
	// behave correctly while the resumed watch runs. Status flips
	// back to "ready"/"failed" inside markPhase1Done when the
	// watch terminates.
	dep.Status = "phase1-watching"
	dep.mu.Unlock()

	h.log.Info("resuming phase 1 watch after pod restart",
		"id", dep.ID,
		"kubeconfigPath", dep.Result.KubeconfigPath,
	)

	go func() {
		h.runPhase1Watch(dep)
		// runPhase1Watch -> markPhase1Done flips Status terminal,
		// but does NOT close the channels (the original
		// runProvisioning closes them on the first-launch path).
		// On resume we own them, so close here to release any
		// SSE consumers waiting on `event: done`.
		dep.mu.Lock()
		select {
		case <-dep.done:
			// Already closed (defensive).
		default:
			close(dep.eventsCh)
			close(dep.done)
		}
		dep.mu.Unlock()
	}()
}

// State returns a JSON-safe snapshot for the GET endpoint.
//
// numEvents surfaces the buffer size so callers polling /deployments/{id}
// can confirm the catalyst-api is recording progress even before they open
// the SSE stream. ProvisionPage uses this in its diagnostic readout.
//
// componentStates + phase1FinishedAt surface the Phase-1 HelmRelease
// watch outcome to the Sovereign Admin shell so its top-level pill
// can render "X of Y components installed" without having to walk
// the full event buffer.
func (d *Deployment) State() map[string]any {
	d.mu.Lock()
	defer d.mu.Unlock()
	out := map[string]any{
		"id":            d.ID,
		"status":        d.Status,
		"startedAt":     d.StartedAt.Format(time.RFC3339),
		"finishedAt":    nil,
		"sovereignFQDN": d.Request.SovereignFQDN,
		"region":        d.Request.Region,
		"numEvents":     len(d.eventsBuf),
		// ownerEmail — the operator who created this deployment (issue #689).
		// Always emitted so the UI can compare against the current session;
		// only operators authorised by checkOwnership() ever reach this path,
		// so the email shown is necessarily either the current user's or
		// blank (legacy record).
		"ownerEmail": d.OwnerEmail,
	}
	if !d.FinishedAt.IsZero() {
		out["finishedAt"] = d.FinishedAt.Format(time.RFC3339)
	}
	if d.Error != "" {
		out["error"] = d.Error
	}
	if d.Result != nil {
		out["result"] = d.Result
		// Lift the Phase-1 fields to the top level too — the
		// Sovereign Admin polls /deployments/<id> and reads them
		// without unwrapping result.
		if len(d.Result.ComponentStates) > 0 {
			out["componentStates"] = d.Result.ComponentStates
		}
		if d.Result.Phase1FinishedAt != nil {
			out["phase1FinishedAt"] = d.Result.Phase1FinishedAt.UTC().Format(time.RFC3339)
		}
		// Issue #519 — lift phase1Outcome too so the wizard's
		// `markFailedTerminal` reducer can read it without unwrapping
		// result. The presence of any non-empty value is the durable
		// proof that runPhase1Watch reached its terminal classification,
		// which in turn proves Phase 0 finished — even if the
		// `tofu-output` event was lost in producer-channel overflow.
		if d.Result.Phase1Outcome != "" {
			out["phase1Outcome"] = d.Result.Phase1Outcome
		}
		// Issue #923 — lift the live Phase-1 substate to the top
		// level so the Sovereign Admin's wizard banner can render
		// "reconnecting…" / "watching…" while Status itself stays
		// "phase1-watching". Empty string is omitted so a UI client
		// that pre-dates the substate field never sees a stray key.
		if d.Result.Phase1Substate != "" {
			out["phase1Substate"] = d.Result.Phase1Substate
		}
		// Issues #764 + #768 — lift the handover-fire surface to the
		// top level so the wizard's provision-page reducer can drive
		// the "Open your Sovereign console →" button + auto-redirect
		// without unwrapping result. handoverURL is the canonical
		// post-mint redirect; handoverFiredAt is the proof-of-fire
		// timestamp the UI uses to render the toast exactly once
		// even on a poll-after-SSE-reconnect.
		if d.Result.HandoverURL != "" {
			out["handoverURL"] = d.Result.HandoverURL
		}
		if d.Result.HandoverFiredAt != nil {
			out["handoverFiredAt"] = d.Result.HandoverFiredAt.UTC().Format(time.RFC3339)
		}
	}
	// adoptedAt — handover-finalisation flag (issue #317) lifted to the
	// top level so the UI's beforeLoad redirect (issue #319) reads it
	// without walking nested result fields. Nil until #317's handover
	// handler stamps it; once set, the Sovereign is operationally
	// self-sufficient and console.openova.io/sovereign/<id> redirects
	// to console.<sovereign-fqdn>.
	if d.AdoptedAt != nil {
		out["adoptedAt"] = d.AdoptedAt.UTC().Format(time.RFC3339)
	}
	return out
}

func (h *Handler) CreateDeployment(w http.ResponseWriter, r *http.Request) {
	var req provisioner.Request
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "invalid request body: "+err.Error(), http.StatusBadRequest)
		return
	}

	// Issue #748 — orgEmail MUST equal the authenticated session's email.
	// Background: until this fix landed, a signed-in user could POST a
	// deployment whose req.OrgEmail belonged to someone ELSE — the catalyst-
	// api accepted the body verbatim and stamped the wrong identity onto
	// the Sovereign-admin / Catalyst-Organization owner. Any subsequent
	// session-scoped list (GET /deployments — issue #747) would then hide
	// the deployment from its real creator and leak it to the typo'd
	// recipient. The session JWT is the load-bearing identity claim; the
	// request body's OrgEmail is no more trustworthy than any other
	// client-provided field.
	//
	// Per docs/INVIOLABLE-PRINCIPLES.md #1 (never trust the client) the
	// server-side check is the load-bearing fix. The wizard's read-only
	// orgEmail input is defense-in-depth only.
	//
	// We use ClaimsFromContext (populated by auth.RequireSession) as the
	// canonical source — when it's nil we fall back to the X-User-Email
	// header so existing tests that build a Handler{} directly without
	// the middleware in the chain still exercise the orgEmail-match
	// check. When BOTH are absent (off-prod / Sovereign-side bootstrap
	// with no auth wired), the check is skipped — the deployment row is
	// stamped with empty OwnerEmail and the legacy passthrough applies.
	claims := auth.ClaimsFromContext(r.Context())
	sessionEmail := ""
	if claims != nil {
		sessionEmail = strings.TrimSpace(claims.Email)
	} else {
		sessionEmail = strings.TrimSpace(r.Header.Get("X-User-Email"))
	}
	if sessionEmail != "" {
		if !strings.EqualFold(strings.TrimSpace(req.OrgEmail), sessionEmail) {
			writeJSON(w, http.StatusForbidden, map[string]string{
				"error": "orgEmail must match the authenticated session's email",
			})
			return
		}
	}

	// Inject Dynadot credentials when the customer chose a pool domain so the
	// OpenTofu module can write DNS records via the dynadot variables.
	// Credentials come from environment variables mounted from the
	// dynadot-api-credentials K8s secret in the openova-system namespace.
	if req.SovereignDomainMode == "pool" {
		req.DynadotAPIKey = h.dynadotAPIKey
		req.DynadotAPISecret = h.dynadotAPISecret
	}

	// Stamp the GHCR pull token from CATALYST_GHCR_PULL_TOKEN onto the
	// Request BEFORE Validate() so a missing-secret misconfiguration
	// surfaces here as 400 with a clear pointer to docs/SECRET-ROTATION.md
	// rather than 5 minutes into the runProvisioning goroutine. The
	// provisioner.New() inside runProvisioning re-stamps the same env
	// var as a defense-in-depth: if the env was missing here but the
	// Pod was rolled in between, the late stamp picks it up.
	//
	// The wizard payload NEVER carries this field — Request.GHCRPullToken
	// is `json:"-"` precisely so the wire format cannot inject it.
	if tok := os.Getenv("CATALYST_GHCR_PULL_TOKEN"); tok != "" {
		req.GHCRPullToken = tok
	}
	// Same fail-fast pattern for the Harbor proxy-cache robot token
	// (issue #557). Empty token here = Validate() rejects the deployment
	// at /api/v1/deployments POST, which is the right time to surface
	// the missing CATALYST_HARBOR_ROBOT_TOKEN env var on the catalyst-
	// api Pod. Falling through to docker.io is forbidden architecturally;
	// the alternative is 5+ min of `tofu apply` followed by a stuck CP
	// at Init:0/6 (caught live during otech24).
	if tok := os.Getenv("CATALYST_HARBOR_ROBOT_TOKEN"); tok != "" {
		req.HarborRobotToken = tok
	}

	// Derive the per-Sovereign Object Storage bucket name (issue #371).
	// Hetzner Object Storage bucket names share a global namespace across
	// every tenant, so a deterministic per-FQDN slug minimises collision
	// risk while keeping the resource name predictable for operators
	// inspecting the Hetzner Console. The wizard never surfaces this as
	// a free-form input — it's plumbing, not user-facing.
	//
	// Pattern: catalyst-<sovereign-fqdn-with-dots-replaced-by-dashes>.
	// `catalyst-omantel-omani-works` for the reference deployment. The
	// downstream Validate() re-checks the resulting string against the
	// S3 bucket-naming RFC; if a future FQDN format breaks the rule the
	// validator emits a clear error rather than letting tofu apply fail
	// 5 minutes in.
	if strings.TrimSpace(req.ObjectStorageBucket) == "" && strings.TrimSpace(req.SovereignFQDN) != "" {
		req.ObjectStorageBucket = "catalyst-" + strings.ReplaceAll(req.SovereignFQDN, ".", "-")
	}

	if err := req.Validate(); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
		return
	}

	id := newID()

	// Mint the cloud-init kubeconfig postback bearer token (issue
	// #183, Option D) BEFORE kicking off the provisioner so
	// writeTfvars renders the plaintext into the Sovereign's
	// cloud-init. The plaintext lives on the provisioner.Request
	// only — never on the Deployment struct, never in the on-disk
	// JSON record. The hash is what we persist; constant-time
	// compared on PUT to verify the inbound bearer.
	bearerToken, bearerHash, err := newBearerToken()
	if err != nil {
		// crypto/rand failure is exceptional — the standard library
		// only fails this when the OS RNG is unavailable. Surface as
		// 500 so the wizard sees a coherent failure instead of a
		// silent crash.
		writeJSON(w, http.StatusInternalServerError, map[string]string{
			"error":  "bearer-token-generation-failed",
			"detail": err.Error(),
		})
		return
	}
	req.DeploymentID = id
	req.KubeconfigBearerToken = bearerToken

	// Stamp the Catalyst-Zero handover-JWT public JWK so cloud-init can
	// write it to /var/lib/catalyst/handover-jwt-public.jwk on the new
	// Sovereign (issue #605, Phase-8b). When the signer is unavailable
	// (CATALYST_HANDOVER_KEY_PATH unset) we leave the field empty — the
	// OpenTofu module's variables.tf default ("") preserves backwards
	// compatibility, the file gets written empty, and the handover flow
	// is simply not yet wired on that Sovereign.
	if h.handoverSigner != nil {
		if jwk, jerr := h.handoverSigner.PublicJWK(); jerr == nil {
			req.HandoverJWTPublicKey = string(jwk)
		} else {
			h.log.Warn("handover signer present but PublicJWK failed; provisioning without JWK on new Sovereign", "err", jerr)
		}
	}

	// Stamp the owner email from the authenticated session (issue #689).
	// auth.RequireSession sets X-User-Email after validating the session
	// cookie AND injects Claims into the context; if both are missing,
	// the deployment was created on a catalyst-api instance running
	// without the auth middleware (legacy CI / Sovereign-side bootstrap)
	// and we leave the field empty. The ownership-check helper treats
	// empty as "legacy — skip".
	//
	// Issue #748: the orgEmail-match check above already guaranteed
	// req.OrgEmail equals the session email when a session is present,
	// so OwnerEmail is necessarily the same identity. We persist the
	// session-derived value (not req.OrgEmail) so a future client-side
	// bug that bypasses the check still cannot poison the durable
	// owner field.
	ownerEmail := sessionEmail

	dep := &Deployment{
		ID:                   id,
		Status:               "provisioning",
		Request:              req,
		StartedAt:            time.Now(),
		eventsCh:             make(chan provisioner.Event, 256),
		done:                 make(chan struct{}),
		kubeconfigBearerHash: bearerHash,
		OwnerEmail:           ownerEmail,
	}

	// Reserve the pool subdomain via PDM BEFORE we kick off `tofu apply`.
	// PDM holds the name with a TTL — if `tofu apply` fails or this catalyst-
	// api Pod crashes, the TTL expires and the name is freed automatically.
	// On the success path the runProvisioning goroutine calls /commit with
	// the LB IP, which flips the reservation to ACTIVE and writes the
	// Dynadot DNS records.
	//
	// For BYO deployments (the customer owns the DNS zone) we skip PDM
	// entirely — the customer points their own CNAME at the LB IP shown
	// on the success screen.
	if req.SovereignDomainMode == "pool" && pdm.IsManagedDomain(req.SovereignPoolDomain) {
		if h.pdm == nil {
			writeJSON(w, http.StatusInternalServerError, map[string]string{
				"error": "pool-domain-manager client is not configured (POOL_DOMAIN_MANAGER_URL)",
			})
			return
		}
		reserveCtx, reserveCancel := context.WithTimeout(r.Context(), 10*time.Second)
		reservation, reserveErr := h.pdm.Reserve(reserveCtx, req.SovereignPoolDomain, req.SovereignSubdomain, "catalyst-api/deployment-"+id)
		reserveCancel()
		if reserveErr != nil {
			if errors.Is(reserveErr, pdm.ErrConflict) {
				writeJSON(w, http.StatusConflict, map[string]string{
					"error":  "subdomain-conflict",
					"detail": "this subdomain has been reserved or activated for the chosen pool — pick a different name",
				})
				return
			}
			h.log.Error("pdm reserve failed", "id", id, "err", reserveErr)
			writeJSON(w, http.StatusServiceUnavailable, map[string]string{
				"error":  "pdm-unavailable",
				"detail": "pool-domain-manager is temporarily unreachable: " + reserveErr.Error(),
			})
			return
		}
		dep.pdmReservationToken = reservation.ReservationToken
		dep.pdmPoolDomain = reservation.PoolDomain
		dep.pdmSubdomain = reservation.Subdomain
	}

	h.deployments.Store(id, dep)

	// Persist the freshly-created deployment row before kicking off
	// the goroutine. If the catalyst-api Pod is killed in the gap
	// between Store and the first emit, the next Pod's restoreFromStore
	// rehydrates it as "failed: catalyst-api restarted during
	// provisioning" so the wizard's FailureCard renders. Without this
	// initial Save, the row wouldn't exist on disk yet and the wizard
	// would see a 404.
	h.persistDeployment(dep)

	// Capture status before launching the goroutine — runProvisioning races
	// with this read otherwise (the goroutine takes dep.mu before mutating
	// Status, but the response writer here reads it without the lock).
	initialStatus := dep.Status
	go h.runProvisioning(dep)

	writeJSON(w, http.StatusCreated, map[string]string{
		"id":        id,
		"status":    initialStatus,
		"streamURL": fmt.Sprintf("/api/v1/deployments/%s/logs", id),
	})
}

// ListDeployments returns the slim state of every deployment owned by the
// authenticated session, optionally filtered by ?owner=<email> (issue #747).
//
// Why a list endpoint:
//
//   The wizard's index route (/sovereign/wizard, /sovereign/) needs to
//   know whether the signed-in operator already has an in-flight
//   deployment so it can auto-redirect to /sovereign/provision/<id>
//   instead of presenting Step 1 of an empty wizard. Without this
//   redirect, an operator who refreshes the tab during a 15-minute
//   provisioning run loses the progress page and lands on the org
//   form — which is the exact bug the founder hit live with otech90
//   on 2026-05-04.
//
// Filtering rules:
//
//   1. ?owner=<email> — restrict to deployments whose OwnerEmail
//      matches (case-insensitive).
//   2. When auth.RequireSession is wired, X-User-Email is always set
//      and we ALSO restrict to that session's email regardless of
//      the ?owner= value, so an operator can't list someone else's
//      deployments by passing a different ?owner=. The server-side
//      check is the security boundary; ?owner= is effectively a
//      hint that mirrors the session.
//   3. When the middleware is bypassed (CI / tests / catalyst-api
//      running without CATALYST_KC_ADDR), the ?owner= filter is the
//      only filter and an empty ?owner= returns every deployment —
//      same passthrough policy as checkOwnership() to keep existing
//      tests working unchanged.
//
// Adopted deployments (AdoptedAt != nil) are EXCLUDED from the list —
// post-handover the customer's Sovereign is operationally self-
// sufficient and the Catalyst-Zero record is a minimum-retention
// breadcrumb, not an active session. The wizard-redirect logic only
// cares about live or recently-failed deployments.
//
// The response is a slim shape (id, status, sovereignFQDN, region,
// startedAt, finishedAt, ownerEmail, adoptedAt, error) so a tab full
// of deployments stays cheap to render — the full event buffer is
// fetched per-deployment via /deployments/{id}/events when the
// operator opens the progress page.
func (h *Handler) ListDeployments(w http.ResponseWriter, r *http.Request) {
	ownerFilter := strings.TrimSpace(strings.ToLower(r.URL.Query().Get("owner")))
	// Issue #748 — read the canonical session identity from the context
	// claims first (auth.RequireSession injects them), falling back to
	// X-User-Email so existing tests that build a Handler{} directly
	// continue to exercise the session-bound filter without rewiring.
	sessionEmail := ""
	if c := auth.ClaimsFromContext(r.Context()); c != nil {
		sessionEmail = strings.TrimSpace(strings.ToLower(c.Email))
	}
	if sessionEmail == "" {
		sessionEmail = strings.TrimSpace(strings.ToLower(r.Header.Get("X-User-Email")))
	}

	// Issue #748 — when a session is present AND a ?owner= query param
	// is also supplied AND ?owner != session.email, return an empty list
	// rather than silently collapsing to session-only rows. Returning
	// rows for a query that asked for a different identity is the kind
	// of subtle UX bug that hides the security boundary; "200 + empty"
	// makes the boundary explicit without leaking existence (we don't
	// emit 403 — the response shape MUST NOT differentiate "exists but
	// not yours" from "doesn't exist", same posture as the issue #689
	// 404-not-403 rule on /deployments/{id}).
	if sessionEmail != "" && ownerFilter != "" && ownerFilter != sessionEmail {
		writeJSON(w, http.StatusOK, map[string]any{
			"deployments": []any{},
		})
		return
	}

	// Effective owner filter — when the session is set (production behind
	// RequireSession), it always defines the boundary so the ?owner=
	// query is at most a redundant assertion. In tests / CI without the
	// middleware, fall back to whatever the caller passed.
	effectiveOwner := sessionEmail
	if effectiveOwner == "" {
		effectiveOwner = ownerFilter
	}

	type listEntry struct {
		ID            string  `json:"id"`
		Status        string  `json:"status"`
		SovereignFQDN string  `json:"sovereignFQDN,omitempty"`
		Region        string  `json:"region,omitempty"`
		StartedAt     string  `json:"startedAt,omitempty"`
		FinishedAt    string  `json:"finishedAt,omitempty"`
		OwnerEmail    string  `json:"ownerEmail,omitempty"`
		AdoptedAt     *string `json:"adoptedAt,omitempty"`
		Error         string  `json:"error,omitempty"`
	}

	out := make([]listEntry, 0)
	h.deployments.Range(func(_, val any) bool {
		dep, ok := val.(*Deployment)
		if !ok || dep == nil {
			return true
		}
		dep.mu.Lock()
		entry := listEntry{
			ID:            dep.ID,
			Status:        dep.Status,
			SovereignFQDN: dep.Request.SovereignFQDN,
			Region:        dep.Request.Region,
			OwnerEmail:    dep.OwnerEmail,
			Error:         dep.Error,
		}
		if !dep.StartedAt.IsZero() {
			entry.StartedAt = dep.StartedAt.UTC().Format(time.RFC3339)
		}
		if !dep.FinishedAt.IsZero() {
			entry.FinishedAt = dep.FinishedAt.UTC().Format(time.RFC3339)
		}
		if dep.AdoptedAt != nil {
			s := dep.AdoptedAt.UTC().Format(time.RFC3339)
			entry.AdoptedAt = &s
		}
		owner := strings.TrimSpace(strings.ToLower(dep.OwnerEmail))
		dep.mu.Unlock()

		// Owner filter — when an effective owner is set, only emit
		// deployments whose OwnerEmail matches case-insensitively.
		// Legacy records with empty OwnerEmail are treated the same way
		// checkOwnership treats them on per-id reads: passthrough when
		// the request has no session, but skipped when the session does
		// not match (so a signed-in operator does not see legacy rows
		// owned by someone else).
		if effectiveOwner != "" {
			if owner == "" {
				// Legacy record — only include it when the caller has
				// no session AND no ?owner= filter, which collapses to
				// "no effective owner". Since effectiveOwner != "" here,
				// we skip the legacy row.
				return true
			}
			if owner != effectiveOwner {
				return true
			}
		}

		// Skip adopted deployments — the customer's Sovereign owns
		// these post-handover, so the wizard redirect should not pull
		// the operator back into the Catalyst-Zero shell.
		if entry.AdoptedAt != nil {
			return true
		}

		out = append(out, entry)
		return true
	})

	writeJSON(w, http.StatusOK, map[string]any{
		"deployments": out,
	})
}

// GetDeployment returns the current state of a deployment for polling.
func (h *Handler) GetDeployment(w http.ResponseWriter, r *http.Request) {
	id := chi.URLParam(r, "id")
	val, ok := h.deployments.Load(id)
	if !ok {
		http.Error(w, "deployment not found", http.StatusNotFound)
		return
	}
	dep := val.(*Deployment)
	// Issue #689 — ownership check. Returns 404 (not 403) on mismatch so
	// the existence of someone else's deployment is never leaked.
	if !h.checkOwnership(w, r, dep) {
		return
	}
	writeJSON(w, http.StatusOK, dep.State())
}

func (h *Handler) StreamLogs(w http.ResponseWriter, r *http.Request) {
	id := chi.URLParam(r, "id")
	// chroot fallback: empty until deployment record settles.
	// On the chroot (SOVEREIGN_FQDN env set), the cutover does NOT
	// import the mother's deployment record into the chroot's
	// in-memory map. The first dashboard load after handover races
	// the synthesis path so without this fallback the wizard's SSE
	// reconnect loop sees a transient 404 (TC-229).
	// chrootEnsureDeployment synthesises a record with pre-closed
	// channels, so the "completed deployment, replay buffer + emit
	// done" branch fires below — empty buffer + done frame is the
	// right shape for a freshly-handed-over Sovereign. Once the
	// chroot lazy-seed populates jobs/events the next reconnect
	// returns the real history. On the mother (no SOVEREIGN_FQDN)
	// chrootEnsureDeployment returns nil and the legacy 404 stands.
	val, ok := h.deployments.Load(id)
	var dep *Deployment
	if ok {
		dep = val.(*Deployment)
	} else if dep = h.chrootEnsureDeployment(id); dep == nil {
		http.Error(w, "deployment not found", http.StatusNotFound)
		return
	}
	// Issue #689 — ownership check BEFORE any SSE bytes are written. Once
	// the response is hijacked into text/event-stream mode the helper's
	// JSON 404 path can't fire, so this MUST happen before the SSE headers.
	if !h.checkOwnership(w, r, dep) {
		return
	}

	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming not supported", http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no")

	// Replay-on-connect: emit every event already in the durable buffer
	// before tailing the live channel. This is what makes navigating to
	// `/sovereign/provision/<completed-id>` render the full history instead
	// of an empty shell — a browser that connects after `event: done`
	// arrived at an already-closed channel previously got nothing.
	for _, ev := range dep.snapshotEvents() {
		writeSSEEvent(w, ev)
	}
	flusher.Flush()

	// If the deployment is already complete, the live channel is closed and
	// the buffer above is the authoritative history. Emit `event: done`
	// with the terminal state and return — the browser won't try to
	// reconnect because the EventSource handler closes on `done`.
	if dep.isDone() {
		fmt.Fprintf(w, "event: done\ndata: %s\n\n", mustJSON(dep.State()))
		flusher.Flush()
		return
	}

	// In-flight: tail the live channel. Any event arriving here also got
	// recorded into the buffer (recordEvent is the single emit path), so
	// the next reconnect after a flake will replay the same history plus
	// whatever arrived in between.
	for {
		select {
		case <-r.Context().Done():
			return
		case ev, open := <-dep.eventsCh:
			if !open {
				fmt.Fprintf(w, "event: done\ndata: %s\n\n", mustJSON(dep.State()))
				flusher.Flush()
				return
			}
			writeSSEEvent(w, ev)
			flusher.Flush()
		}
	}
}

// PhaseHandoverReady — Phase value set on a synthesized provisioner.Event
// when catalyst-api auto-fires the handover JWT mint after the Phase-1
// watch terminates with OutcomeReady (issues #764 + #768). The event's
// Message carries a JSON object `{"handoverURL": "...", "expiresAt":
// "..."}` (RFC 3339 expiry, RS256 token contract — see
// internal/handoverjwt/signer.go).
//
// StreamLogs renders this Event as a typed SSE event:
//
//	event: handover-ready
//	data: {"handoverURL": "...", "expiresAt": "..."}
//
// The wizard's useDeploymentEvents hook listens via
// EventSource.addEventListener('handover-ready', ...) so the typed
// channel is independent of the default `message` event reducer.
//
// Per docs/INVIOLABLE-PRINCIPLES.md #4 the URL is built from the
// Sovereign FQDN persisted on the deployment record + the JWT minted
// by handoverjwt.Signer.MintToken — no hardcoded host.
const PhaseHandoverReady = "handover-ready"

// writeSSEEvent renders a single provisioner.Event onto an SSE stream.
//
// Default rendering is the legacy shape:
//
//	data: <json-of-event>
//
// (no `event:` prefix → the browser dispatches it as the default
// `message` event, which the existing reducer consumes).
//
// The single typed-event branch is `Phase == PhaseHandoverReady`,
// which renders:
//
//	event: handover-ready
//	data: <Event.Message verbatim>
//
// Event.Message for the handover-ready event is already a JSON object
// — see Handler.fireHandover in phase1_watch.go. The wizard's
// addEventListener('handover-ready', ...) parses it directly without
// the Event wrapper.
func writeSSEEvent(w http.ResponseWriter, ev provisioner.Event) {
	if ev.Phase == PhaseHandoverReady {
		// The Message field is the canonical JSON payload —
		// emit it verbatim under the typed event channel. No
		// double-wrapping in the Event struct, so the Sovereign UI's
		// addEventListener handler receives the exact shape #768
		// specifies: {"handoverURL": "...", "expiresAt": "..."}.
		fmt.Fprintf(w, "event: %s\ndata: %s\n\n", PhaseHandoverReady, ev.Message)
		return
	}
	fmt.Fprintf(w, "data: %s\n\n", mustJSON(ev))
}

// GetDeploymentEvents returns the buffered event history + state JSON for
// a one-shot snapshot. ProvisionPage calls this on mount to seed the
// `applyEventToContext` reducer before opening the SSE stream — the SSE
// stream's replay-on-connect serves the same purpose, but a stateless GET
// is easier to test and gives the wizard a fast-path that doesn't need to
// hold an SSE socket open. Both paths read the same `eventsBuf`, so they
// agree by construction.
func (h *Handler) GetDeploymentEvents(w http.ResponseWriter, r *http.Request) {
	id := chi.URLParam(r, "id")
	// chroot fallback: empty until deployment record settles.
	// On the chroot (SOVEREIGN_FQDN env set), the cutover does NOT
	// import the mother's deployment record. The first dashboard
	// load races the in-memory synthesis path so without this
	// fallback the wizard sees a transient 404 (TC-229) on
	// /api/v1/deployments/<id>/events. Synthesise a minimal record
	// — snapshotEvents() returns an empty slice on the synthetic
	// record so the wizard renders its empty state and reconnects;
	// once Phase-1 watch starts emitting (chroot lazy-seed path),
	// subsequent reads return the populated buffer. On the mother
	// (no SOVEREIGN_FQDN) chrootEnsureDeployment returns nil and
	// the legacy 404 path is preserved.
	val, ok := h.deployments.Load(id)
	var dep *Deployment
	if ok {
		dep = val.(*Deployment)
	} else if dep = h.chrootEnsureDeployment(id); dep == nil {
		http.Error(w, "deployment not found", http.StatusNotFound)
		return
	}
	// Issue #689 — ownership check (404 on mismatch — never leak existence).
	if !h.checkOwnership(w, r, dep) {
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"state":  dep.State(),
		"events": dep.snapshotEvents(),
		"done":   dep.isDone(),
	})
}

func (h *Handler) runProvisioning(dep *Deployment) {
	// Pre-seed the 5 Phase-0 lifecycle Job rows ("provisioner" group)
	// BEFORE prov.Provision starts. This guarantees:
	//
	//   1. Deep-links to /jobs/tofu-init|tofu-plan|tofu-apply|
	//      tofu-output|cluster-bootstrap resolve from the very first
	//      wizard render, not just after the corresponding emit.
	//   2. The JobsTable can never "drop" the Phase-0 rows when
	//      bootstrap-kit children land later (they're durable in
	//      jobs.Store, not derived from the live event stream).
	//
	// The bridge initialiser is the same path emitWatchEvent uses;
	// pre-allocating it here means SeedProvisionerJobs fires once per
	// deployment regardless of whether the first emit beats this
	// goroutine.
	if h.jobs != nil {
		dep.mu.Lock()
		if dep.jobsBridge == nil {
			dep.jobsBridge = jobs.NewBridge(h.jobs, dep.ID)
		}
		bridge := dep.jobsBridge
		dep.mu.Unlock()
		if seedErr := bridge.SeedProvisionerJobs(); seedErr != nil {
			h.log.Warn("jobs bridge: seed Phase-0 lifecycle jobs failed",
				"id", dep.ID, "err", seedErr)
		}
	}

	// Tee — provisioner.Provision writes events into producer; this
	// goroutine records every event in the durable buffer AND forwards
	// it to the live SSE channel. recordEvent is the single emit path,
	// so the buffer and the live stream cannot diverge. The Phase-1
	// watch (when launched) shares the same emit path via
	// h.emitWatchEvent so per-component events flow through identical
	// plumbing.
	producer := make(chan provisioner.Event, 256)
	teeDone := make(chan struct{})
	go func() {
		defer close(teeDone)
		for ev := range producer {
			h.emitWatchEvent(dep, ev)
		}
	}()

	prov := provisioner.New()
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
	defer cancel()

	result, err := prov.Provision(ctx, dep.Request, producer)
	close(producer)
	<-teeDone

	// Stamp the Phase-0 lifecycle Jobs into terminal state so the
	// JobsTable's Started/Finished/Duration columns populate without
	// waiting for Phase-1 events. Failure stamps the in-flight phase
	// as Failed (with the error captured as a final LogLine);
	// success rolls every phase to Succeeded.
	now := time.Now().UTC()
	if h.jobs != nil && dep.jobsBridge != nil {
		if err != nil {
			if mErr := dep.jobsBridge.MarkProvisionerFailed(err.Error(), now); mErr != nil {
				h.log.Warn("jobs bridge: mark Phase-0 failed",
					"id", dep.ID, "err", mErr)
			}
		} else {
			if mErr := dep.jobsBridge.MarkProvisionerComplete(now); mErr != nil {
				h.log.Warn("jobs bridge: mark Phase-0 complete",
					"id", dep.ID, "err", mErr)
			}
		}
	}

	// Capture Phase-0 outcome under the lock.
	dep.mu.Lock()
	if err != nil {
		dep.FinishedAt = time.Now()
		dep.Status = "failed"
		dep.Error = err.Error()
		h.log.Error("provision failed", "id", dep.ID, "err", err)
	} else {
		dep.Status = "phase1-watching"
		dep.Result = result
		h.log.Info("phase 0 complete; phase 1 watch starting",
			"id", dep.ID,
			"sovereignFQDN", result.SovereignFQDN,
			"controlPlaneIP", result.ControlPlaneIP,
			"loadBalancerIP", result.LoadBalancerIP,
		)
	}
	dep.mu.Unlock()
	// Persist the Phase-0 terminal state (ready or failed). This is
	// the line that guarantees a `status: phase1-watching` (or
	// `failed`) deployment on disk before the Phase-1 watch starts,
	// so a Pod kill in the gap between Phase 0 and Phase 1 can be
	// resumed/diagnosed correctly.
	h.persistDeployment(dep)

	// PDM /commit fires HERE — between Phase 0 and Phase 1 — for two
	// architectural reasons:
	//
	//   1. The Phase-0 LB IP is the ONLY data PDM needs to write the
	//      child-zone records (NS delegation + console/api/wildcard
	//      A records). Phase-1 outcome does NOT change DNS routing,
	//      so deferring the commit to AFTER Phase 1 just delays the
	//      moment console.<sub>.<pool> resolves (and previously did
	//      so with the SSE channel already closed — failures were
	//      invisible to the wizard).
	//   2. The PDM reservation TTL is 10 minutes by default; Phase-0
	//      routinely takes 8-12 min on a fresh Hetzner cluster, so
	//      the commit can race the sweeper. CommitWithRetry handles
	//      that by re-Reserving on 404/410/403 — and it does so while
	//      the wizard SSE stream is still open, surfacing each
	//      attempt as an event. See pdm/client.go:CommitWithRetry.
	//
	// On Phase-0 failure (no LB IP), Release is called instead so the
	// reservation TTL doesn't have to expire to free the name.
	if err == nil && result != nil {
		h.commitPDMWithRetry(dep, result)
	} else {
		h.releasePDMReservation(dep)
	}

	// Phase 1 — HelmRelease watch. Only runs on Phase-0 success and
	// only when a kubeconfig is available. The watch emits per-
	// component events into the same SSE buffer + live channel; when
	// it terminates, it writes ComponentStates + Phase1FinishedAt
	// onto dep.Result and flips Status to ready (or leaves failed
	// alone if Phase 0 already failed).
	if err == nil && result != nil {
		h.runPhase1Watch(dep)
	}

	// Close the SSE live channel + done signal AFTER both phases
	// have settled. Existing tests that drive runProvisioning's
	// failure-fast path (no real tofu) still hit close because the
	// Phase-0 error skips the watch.
	close(dep.eventsCh)
	close(dep.done)

	// Final persist — captures Phase 1 terminal state when the watch
	// ran, or is a no-op for the Phase 0 failure path (already
	// persisted above).
	h.persistDeployment(dep)
}

// commitPDMWithRetry promotes the deployment's PDM reservation to
// ACTIVE and triggers the per-Sovereign zone write inside PDM. It runs
// while the wizard SSE stream is still open so each attempt + failure
// surfaces as an event; on final exhaustion the human-readable error
// is appended to dep.Error so the wizard FailureCard renders it.
//
// The retry loop is delegated to pdm.Client.CommitWithRetry (5
// attempts, 1s → 16s backoff). Re-Reserve on 404/410/403 is done
// inline so a TTL-expiry — the actual symptom that broke otech41+ —
// recovers automatically and the persisted deployment row is updated
// to the fresh token (so a Pod restart between re-Reserve and
// re-Commit picks up the new token instead of the stale one).
//
// No-op when the deployment isn't using PDM (BYO domain mode, or
// h.pdm == nil in tests).
func (h *Handler) commitPDMWithRetry(dep *Deployment, result *provisioner.Result) {
	if dep.pdmReservationToken == "" || h.pdm == nil {
		return
	}

	// Up to 5 minutes for the full retry window — 5 attempts at
	// 1s/2s/4s/8s/16s backoff plus per-attempt HTTP timeouts (15s
	// each from Client.HTTP) plus a re-Reserve every other attempt
	// in the worst case = ~3 min ceiling, with a safety margin.
	pdmCtx, pdmCancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer pdmCancel()

	emitInfo := func(msg string) {
		h.emitWatchEvent(dep, provisioner.Event{
			Time:    time.Now().UTC().Format(time.RFC3339),
			Phase:   "pdm-commit",
			Level:   "info",
			Message: msg,
		})
	}
	emitErr := func(msg string) {
		h.emitWatchEvent(dep, provisioner.Event{
			Time:    time.Now().UTC().Format(time.RFC3339),
			Phase:   "pdm-commit",
			Level:   "error",
			Message: msg,
		})
	}

	emitInfo(fmt.Sprintf("Publishing DNS records for %s via pool-domain-manager (LB %s)…",
		result.SovereignFQDN, result.LoadBalancerIP))

	commitErr := h.pdm.CommitWithRetry(
		pdmCtx,
		dep.pdmPoolDomain,
		pdm.CommitInput{
			Subdomain:        dep.pdmSubdomain,
			ReservationToken: dep.pdmReservationToken,
			SovereignFQDN:    result.SovereignFQDN,
			LoadBalancerIP:   result.LoadBalancerIP,
		},
		func(ctx context.Context) (*pdm.Reservation, error) {
			emitInfo(fmt.Sprintf("PDM reservation expired during Phase 0 — re-Reserving %s.%s",
				dep.pdmSubdomain, dep.pdmPoolDomain))
			return h.pdm.Reserve(ctx, dep.pdmPoolDomain, dep.pdmSubdomain, "catalyst-api/deployment-"+dep.ID)
		},
		func(newToken string) {
			// Token rotated — persist the new token onto the row so a
			// Pod restart between this re-Reserve and the next
			// Commit attempt doesn't lose it.
			dep.mu.Lock()
			dep.pdmReservationToken = newToken
			dep.mu.Unlock()
			h.persistDeployment(dep)
		},
		pdm.CommitRetryConfig{}, // production defaults
	)
	if commitErr != nil {
		// Final exhaustion — surface to the wizard. The Sovereign is
		// live (cluster, LB, kubeconfig all good) but DNS is stale,
		// so we annotate dep.Error rather than flipping Status to
		// failed (which would imply the cluster itself is broken).
		// The wizard's FailureCard checks dep.Error before rendering
		// "ready" so the operator sees this regardless.
		h.log.Error("pdm commit failed after retry; sovereign is live but DNS records are stale — operator action required",
			"id", dep.ID,
			"poolDomain", dep.pdmPoolDomain,
			"subdomain", dep.pdmSubdomain,
			"err", commitErr,
		)
		emitErr(fmt.Sprintf("DNS publish FAILED after retry: %v. The Sovereign cluster is live (LB %s) but %s will not resolve until an operator runs PDM /commit manually or re-runs the deployment.",
			commitErr, result.LoadBalancerIP, result.SovereignFQDN))
		dep.mu.Lock()
		errMsg := fmt.Sprintf("DNS commit failed: %v — sovereign cluster is live but %s does not resolve. Re-run `curl -X POST $POOL_DOMAIN_MANAGER_URL/api/v1/pool/%s/commit` with a fresh reservation, or re-trigger the deployment.",
			commitErr, result.SovereignFQDN, dep.pdmPoolDomain)
		if dep.Error == "" {
			dep.Error = errMsg
		} else {
			dep.Error = dep.Error + " | " + errMsg
		}
		dep.mu.Unlock()
		h.persistDeployment(dep)
		return
	}

	h.log.Info("pdm commit complete",
		"id", dep.ID,
		"poolDomain", dep.pdmPoolDomain,
		"subdomain", dep.pdmSubdomain,
		"loadBalancerIP", result.LoadBalancerIP,
	)
	emitInfo(fmt.Sprintf("DNS published — %s now resolves to %s.",
		result.SovereignFQDN, result.LoadBalancerIP))
}

// releasePDMReservation releases the PDM reservation on Phase-0
// failure so the reservation TTL doesn't have to expire to free the
// name. Best-effort: a release failure logs a warning but does not
// block the failure-cleanup path. ErrNotFound is silently ignored
// because the row may have been swept already.
func (h *Handler) releasePDMReservation(dep *Deployment) {
	if dep.pdmReservationToken == "" || h.pdm == nil {
		return
	}
	pdmCtx, pdmCancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer pdmCancel()
	releaseErr := h.pdm.Release(pdmCtx, dep.pdmPoolDomain, dep.pdmSubdomain)
	if releaseErr != nil && !errors.Is(releaseErr, pdm.ErrNotFound) {
		h.log.Error("pdm release failed; reservation will expire on TTL",
			"id", dep.ID,
			"poolDomain", dep.pdmPoolDomain,
			"subdomain", dep.pdmSubdomain,
			"err", releaseErr,
		)
	}
}

// emitWatchEvent — single emit path for Phase 0 + Phase 1 events.
// Records into the durable buffer (which persists every event to
// disk) and forwards onto the live SSE channel. Non-blocking send to
// eventsCh: if no consumer is attached, the buffer (256) absorbs the
// burst; once full we drop on the LIVE side only — the durable
// buffer still has the event so the next /events poll or SSE
// reconnect replays it.
//
// The same call also forwards Phase-1 component events to the per-
// deployment jobs.Bridge, which materialises Jobs / Executions / LogLines
// into the new /api/v1/deployments/{id}/jobs surface (issue #205, sub
// of epic #204). The bridge write is best-effort: a failure does NOT
// abort the SSE feed (the durable buffer is the contract for
// /api/v1/deployments/{id}/events; the jobs surface is a parallel
// projection). Errors are logged at warn so an operator can spot
// drift.
func (h *Handler) emitWatchEvent(dep *Deployment, ev provisioner.Event) {
	recorded := h.recordEventAndPersist(dep, ev)

	// Synchronise channel send with the close() path in
	// resumePhase1Watch (which closes dep.eventsCh under dep.mu after
	// the watch loop returns). Without this guard a helmwatch
	// goroutine still in-flight when resumePhase1Watch closes
	// eventsCh races the close — the race detector flags the read
	// of eventsCh against the close write. Holding dep.mu here makes
	// the close-vs-send linearisation point unambiguous, and the
	// `done` short-circuit prevents a panic-on-closed-channel send.
	dep.mu.Lock()
	closed := false
	select {
	case <-dep.done:
		closed = true
	default:
	}
	if !closed {
		select {
		case dep.eventsCh <- recorded:
		default:
		}
	}
	bridge := dep.jobsBridge
	if h.jobs != nil && bridge == nil {
		bridge = jobs.NewBridge(h.jobs, dep.ID)
		dep.jobsBridge = bridge
	}
	dep.mu.Unlock()

	// Forward Phase-1 component events to the jobs bridge. Phase-0
	// OpenTofu events have no Job analogue and are silently dropped
	// inside the bridge (it filters on Phase=="component" + non-
	// empty Component). The bridge write is best-effort: a failure
	// does NOT abort the SSE feed.
	if bridge == nil {
		return
	}
	if err := bridge.OnProvisionerEvent(recorded); err != nil {
		h.log.Warn("jobs bridge: forward failed",
			"id", dep.ID,
			"phase", recorded.Phase,
			"component", recorded.Component,
			"err", err,
		)
	}
}

func newID() string {
	b := make([]byte, 8)
	_, _ = rand.Read(b)
	return hex.EncodeToString(b)
}

// newBearerToken mints the cloud-init kubeconfig postback bearer
// (issue #183, Option D). Returns (plaintextHex, sha256Hex, error).
//
// 32 bytes from crypto/rand → 64 hex chars. The plaintext NEVER
// lands on disk on the catalyst-api side — it flows out via tfvars
// into the new Sovereign's cloud-init user_data and is consumed
// exactly once when the new control plane PUTs back its kubeconfig.
// The SHA-256 hash is what we persist on the deployment record;
// the PUT handler constant-time compares the inbound bearer's hash
// to that value.
//
// Returns an error only when crypto/rand fails — the standard
// library surfaces that exclusively when the OS RNG is unavailable,
// which is catastrophic for the whole process. The
// CreateDeployment caller translates the error into HTTP 500.
func newBearerToken() (plaintext, hashHex string, err error) {
	raw := make([]byte, 32)
	if _, err := rand.Read(raw); err != nil {
		return "", "", fmt.Errorf("crypto/rand: %w", err)
	}
	plaintext = hex.EncodeToString(raw)
	sum := sha256.Sum256([]byte(plaintext))
	hashHex = hex.EncodeToString(sum[:])
	return plaintext, hashHex, nil
}

// hashBearerToken returns the hex-encoded SHA-256 of the supplied
// bearer plaintext. Used by the PUT /kubeconfig handler to compare
// the inbound bearer to the persisted hash.
func hashBearerToken(plaintext string) string {
	sum := sha256.Sum256([]byte(plaintext))
	return hex.EncodeToString(sum[:])
}

func mustJSON(v any) string {
	b, err := json.Marshal(v)
	if err != nil {
		fallback, _ := json.Marshal(map[string]string{
			"level": "error",
			"msg":   "failed to encode event: " + err.Error(),
		})
		return string(fallback)
	}
	return string(b)
}
