felhom.eu/hub/internal/store/pbsdr.go

package store

import (
	"database/sql"
	"encoding/json"
)

// PBS DR tier (SLICE 1): the HOST-scoped one-time PBS token secret — the host/agent twin of the
// customer-scoped one_time_secrets pair (SaveOneTimeSecret/ConsumeOneTimeSecret). The hub stores
// the tenantsync-returned token secret here; the agent consumes it EXACTLY ONCE with its per-host
// key. Transient custody: never logged, never in desired-state or any served config.

// SaveHostPBSSecret stores (last-write-wins) the one-time PBS token secret for a host, resetting
// the consumed flag (a re-issue supersedes any prior unconsumed value). Never logged.
func (s *Store) SaveHostPBSSecret(hostID, value string) error {
	_, err := s.db.Exec(`
		INSERT INTO host_pbs_secrets (host_id, value, created_at, consumed_at)
		VALUES (?, ?, datetime('now'), NULL)
		ON CONFLICT(host_id) DO UPDATE SET value = excluded.value, created_at = datetime('now'), consumed_at = NULL`,
		hostID, value)
	return err
}

// ConsumeHostPBSSecret returns the host's one-time PBS token secret and marks it consumed in the
// SAME transaction (single use). A second call — or a call when none is stored — returns
// ("", sql.ErrNoRows). The value is never logged.
func (s *Store) ConsumeHostPBSSecret(hostID string) (string, error) {
	tx, err := s.db.Begin()
	if err != nil {
		return "", err
	}
	defer tx.Rollback()
	var value string
	err = tx.QueryRow(`SELECT value FROM host_pbs_secrets WHERE host_id = ? AND consumed_at IS NULL`, hostID).Scan(&value)
	if err != nil {
		return "", err // sql.ErrNoRows when absent OR already consumed
	}
	if _, err := tx.Exec(`UPDATE host_pbs_secrets SET consumed_at = datetime('now') WHERE host_id = ?`, hostID); err != nil {
		return "", err
	}
	if err := tx.Commit(); err != nil {
		return "", err
	}
	return value, nil
}

// RestageHostPBSSecret re-arms an ALREADY-STORED one-time PBS secret for re-consumption by clearing
// its consumed flag — WITHOUT changing the secret value, WITHOUT inserting a row, and WITHOUT any
// endpoint/token interaction. This is the PBS-DR self-heal primitive
// (SPIKE-pbsdr-selfheal-2026-07-15, e8f8c44): a box that lost its converged marker (re-install /
// snapshot rollback) but whose ep0 token is still valid re-consumes the SAME secret and converges in
// one agent tick (spike SQ-2b′, zero ep0 churn). Returns restaged=true when a row existed (its
// consumed flag is now cleared, so ConsumeHostPBSSecret will serve it once); restaged=false when NO
// secret is stored for the host — the caller must escalate to Re-issue (a fresh mint). The value is
// never read or logged here, and NO host generation is bumped (a re-stage changes no descriptor
// content — a bump would trigger an agent desired-state refetch loop).
func (s *Store) RestageHostPBSSecret(hostID string) (bool, error) {
	res, err := s.db.Exec(`UPDATE host_pbs_secrets SET consumed_at = NULL WHERE host_id = ?`, hostID)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return n > 0, nil
}

// PBSDRHealRow is the per-host self-heal decision input (SPIKE-pbsdr-selfheal-2026-07-15): the host's
// descriptor enable/provision state (parsed from desired_json) joined to the agent's latest reported
// pbs_dr.state (from the newest host_report). Mirrors GetHostOOBStates' latest-report-per-host shape.
type PBSDRHealRow struct {
	HostID                string
	CustomerID            string
	DescriptorEnabled     bool   // desired_json pbs_dr.enabled — the DR-ON reality
	DescriptorProvisioned bool   // desired_json pbs_dr.namespace != "" (was provisioned, not a bare enable)
	ReportedState         string // latest report pbs_dr.state ("" = no report / no stanza)
	ReportID              int64  // id of that latest host_report (0 = none); the debounce distinctness key
}

// PBSDRHealStates returns one row per host: its descriptor enable/provision flags + the agent's
// latest reported pbs_dr.state + that report's id. Malformed JSON degrades to zero values (never an
// error) — the reconciler only acts on positively-parsed stuck states. One query for the whole fleet
// (the reconciler filters to enabled+provisioned hosts).
func (s *Store) PBSDRHealStates() ([]PBSDRHealRow, error) {
	rows, err := s.db.Query(`
		SELECT h.host_id, h.customer_id, h.desired_json, latest.mx, hr.report_json
		FROM hosts h
		LEFT JOIN (SELECT host_id, MAX(id) AS mx FROM host_reports GROUP BY host_id) latest
		       ON latest.host_id = h.host_id
		LEFT JOIN host_reports hr ON hr.id = latest.mx`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []PBSDRHealRow
	for rows.Next() {
		var hostID, customerID, desiredJSON string
		var reportID sql.NullInt64
		var reportJSON sql.NullString
		if err := rows.Scan(&hostID, &customerID, &desiredJSON, &reportID, &reportJSON); err != nil {
			return nil, err
		}
		r := PBSDRHealRow{HostID: hostID, CustomerID: customerID}
		var dd struct {
			PBSDR *struct {
				Enabled   bool   `json:"enabled"`
				Namespace string `json:"namespace"`
			} `json:"pbs_dr"`
		}
		if json.Unmarshal([]byte(desiredJSON), &dd) == nil && dd.PBSDR != nil {
			r.DescriptorEnabled = dd.PBSDR.Enabled
			r.DescriptorProvisioned = dd.PBSDR.Namespace != ""
		}
		if reportID.Valid {
			r.ReportID = reportID.Int64
		}
		if reportJSON.Valid {
			var rr struct {
				PBSDR *struct {
					State string `json:"state"`
				} `json:"pbs_dr"`
			}
			if json.Unmarshal([]byte(reportJSON.String), &rr) == nil && rr.PBSDR != nil {
				r.ReportedState = rr.PBSDR.State
			}
		}
		out = append(out, r)
	}
	return out, rows.Err()
}