pangolin/server/internal/scheduler/wiring_lifecycle.go

package scheduler

import (
	"context"
	"database/sql"
	"encoding/json"
	"strconv"
	"strings"
	"time"

	dbx "github.com/wangjia/pangolin/server/internal/db"
	"github.com/wangjia/pangolin/server/internal/nodes"
	"github.com/wangjia/pangolin/server/internal/scheduler/detect"
	"github.com/wangjia/pangolin/server/internal/scheduler/orchestrate"
	"github.com/wangjia/pangolin/server/internal/store"
)

// SQLLifecycle is a MySQL-backed lifecycle store. The detect and orchestrate
// LifecycleService interfaces are satisfied by the two thin adapter wrappers at
// the bottom of this file (detectLifecycle / orchestrateLifecycle), both of
// which delegate to this shared core.
//
// TransitionStatus is the documented optimistic-lock contract:
//
//	UPDATE nodes SET status=to WHERE id=nodeID AND status=from
//
// returning rows affected (1 = applied, 0 = lock conflict / no-op). On a 1-row
// transition it best-effort writes a node_events row and bumps the global
// directory_version so clients re-fetch.
//
// NOTE: cert revocation on a destroyed transition is intentionally NOT done here
// — provision.DestroyNode tears down the VM, and the agent's mTLS cert can be
// revoked separately. Wire CRL revocation when the real fleet exists (TODO).
type SQLLifecycle struct {
	db      *sql.DB
	dialect dbx.Dialect
	loads   *nodes.LoadCache
}

// NewSQLLifecycle wires the store. loads may be nil (load reads then return zero).
func NewSQLLifecycle(db *sql.DB, loads *nodes.LoadCache) *SQLLifecycle {
	return &SQLLifecycle{db: db, dialect: dbx.DialectForDB(db), loads: loads}
}

const nodeSelectCols = `id, uuid, status, region, tier, role, reality_sni, reality_pbk, hy2_port, provider_id, weight, name_zh, name_en`

type lcRow struct {
	id         int64
	uuid       string
	status     string
	region     string
	tier       string
	role       string
	sni        string
	pbk        string
	hy2Port    int
	providerID int64
	weight     int
	nameZH     string
	nameEn     string
}

func scanNode(rows interface{ Scan(...any) error }) (lcRow, error) {
	var r lcRow
	var hy2 sql.NullInt32
	err := rows.Scan(&r.id, &r.uuid, &r.status, &r.region, &r.tier, &r.role,
		&r.sni, &r.pbk, &hy2, &r.providerID, &r.weight, &r.nameZH, &r.nameEn)
	if hy2.Valid {
		r.hy2Port = int(hy2.Int32)
	}
	return r, err
}

func (l *SQLLifecycle) list(ctx context.Context, statuses []string) ([]lcRow, int64, error) {
	q := `SELECT ` + nodeSelectCols + ` FROM nodes`
	var args []any
	if len(statuses) > 0 {
		ph := make([]string, len(statuses))
		for i, s := range statuses {
			ph[i] = "?"
			args = append(args, s)
		}
		q += ` WHERE status IN (` + strings.Join(ph, ",") + `)`
	}
	rows, err := l.db.QueryContext(ctx, q, args...)
	if err != nil {
		return nil, 0, err
	}
	defer rows.Close()
	var out []lcRow
	for rows.Next() {
		r, err := scanNode(rows)
		if err != nil {
			return nil, 0, err
		}
		out = append(out, r)
	}
	return out, l.version(ctx), rows.Err()
}

func (l *SQLLifecycle) get(ctx context.Context, nodeID string) (*lcRow, error) {
	row := l.db.QueryRowContext(ctx, `SELECT `+nodeSelectCols+` FROM nodes WHERE id = ?`, nodeID)
	r, err := scanNode(row)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &r, nil
}

func (l *SQLLifecycle) transition(ctx context.Context, nodeID, from, to string, detail map[string]any) (int, error) {
	tx, err := l.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer func() { _ = tx.Rollback() }()

	res, err := tx.ExecContext(ctx,
		`UPDATE nodes SET status = ? WHERE id = ? AND status = ?`, to, nodeID, from)
	if err != nil {
		return 0, err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return 0, nil // optimistic-lock conflict: node already moved → no-op
	}

	if ev := eventForStatus(to); ev != "" {
		var dj any
		if detail != nil {
			if b, err := json.Marshal(detail); err == nil {
				dj = string(b)
			}
		}
		_, _ = tx.ExecContext(ctx,
			`INSERT INTO node_events (node_id, event, detail) VALUES (?, ?, ?)`, nodeID, ev, dj)
	}
	if err := store.BumpDirectoryVersion(ctx, tx, l.dialect); err != nil {
		return 0, err
	}
	if err := tx.Commit(); err != nil {
		return 0, err
	}
	return int(n), nil
}

func (l *SQLLifecycle) setWeight(ctx context.Context, nodeID string, weight int) error {
	_, err := l.db.ExecContext(ctx, `UPDATE nodes SET weight = ? WHERE id = ?`, weight, nodeID)
	return err
}

func (l *SQLLifecycle) bumpVersion(ctx context.Context) error {
	return store.BumpDirectoryVersion(ctx, l.db, l.dialect)
}

func (l *SQLLifecycle) version(ctx context.Context) int64 {
	var v int64
	_ = l.db.QueryRowContext(ctx, `SELECT version FROM directory_version WHERE id = 1`).Scan(&v)
	return v
}

// loadFor resolves a node id → uuid → its current load sample (online, Mbps).
func (l *SQLLifecycle) loadFor(ctx context.Context, nodeID string) (int, float64) {
	if l.loads == nil {
		return 0, 0
	}
	var uuid string
	if err := l.db.QueryRowContext(ctx, `SELECT uuid FROM nodes WHERE id = ?`, nodeID).Scan(&uuid); err != nil {
		return 0, 0
	}
	nl, ok, err := l.loads.Get(ctx, uuid)
	if err != nil || !ok || nl == nil {
		return 0, 0
	}
	mbps := float64(nl.BandwidthUpBps+nl.BandwidthDownBps) / 1_000_000.0
	return int(nl.OnlineCount), mbps
}

func (l *SQLLifecycle) writeAudit(ctx context.Context, actor, action, target, meta string) error {
	var metaArg any
	if meta != "" {
		metaArg = meta
	}
	_, err := l.db.ExecContext(ctx,
		`INSERT INTO audit_log (actor, action, target, meta) VALUES (?, ?, ?, ?)`,
		actor, action, target, metaArg)
	return err
}

// eventForStatus maps a target node status to the node_events ENUM value, or ""
// when the status has no corresponding event (then no event row is written).
func eventForStatus(status string) string {
	switch status {
	case "provisioning", "probing":
		return "provisioned"
	case "up":
		return "marked_up"
	case "draining":
		return "draining"
	case "blocked_suspect":
		return "blocked_suspect"
	case "blocked_confirmed":
		return "blocked_confirmed"
	case "destroyed":
		return "destroyed"
	default:
		return "" // e.g. "down" has no event enum
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// detect.LifecycleService adapter
// ─────────────────────────────────────────────────────────────────────────────

type detectLifecycle struct{ *SQLLifecycle }

// NewDetectLifecycle adapts SQLLifecycle to detect.LifecycleService.
func NewDetectLifecycle(l *SQLLifecycle) detect.LifecycleService { return detectLifecycle{l} }

func (a detectLifecycle) ListNodes(ctx context.Context, f detect.NodeFilter) ([]detect.NodeInfo, error) {
	statuses := make([]string, len(f.Statuses))
	for i, s := range f.Statuses {
		statuses[i] = string(s)
	}
	rows, ver, err := a.list(ctx, statuses)
	if err != nil {
		return nil, err
	}
	out := make([]detect.NodeInfo, 0, len(rows))
	for _, r := range rows {
		out = append(out, detect.NodeInfo{
			ID:      strconv.FormatInt(r.id, 10),
			UUID:    r.uuid,
			Status:  detect.NodeStatus(r.status),
			Weight:  r.weight,
			Version: ver,
		})
	}
	return out, nil
}

func (a detectLifecycle) TransitionStatus(ctx context.Context, nodeID string, from, to detect.NodeStatus, detail map[string]any) (int, error) {
	return a.transition(ctx, nodeID, string(from), string(to), detail)
}

func (a detectLifecycle) SetWeight(ctx context.Context, nodeID string, weight int) error {
	return a.setWeight(ctx, nodeID, weight)
}

func (a detectLifecycle) BumpVersion(ctx context.Context) error { return a.bumpVersion(ctx) }

func (a detectLifecycle) GetLoad(ctx context.Context, nodeID string) (detect.LoadInfo, error) {
	online, mbps := a.loadFor(ctx, nodeID)
	return detect.LoadInfo{Online: online, BandwidthMbps: mbps, Timestamp: time.Now().Unix()}, nil
}

func (a detectLifecycle) GetLoadHistory(ctx context.Context, nodeID string, _ time.Duration) ([]detect.LoadPoint, error) {
	// No historical load store yet (only the latest sample lives in Redis), so
	// return the current point. The drop-percentage rule degrades safely (no fire).
	online, mbps := a.loadFor(ctx, nodeID)
	if online == 0 && mbps == 0 {
		return nil, nil
	}
	return []detect.LoadPoint{{Timestamp: time.Now().Unix(), Online: online, BandwidthMbps: mbps}}, nil
}

// ─────────────────────────────────────────────────────────────────────────────
// orchestrate.LifecycleService adapter
// ─────────────────────────────────────────────────────────────────────────────

type orchestrateLifecycle struct{ *SQLLifecycle }

// NewOrchestrateLifecycle adapts SQLLifecycle to orchestrate.LifecycleService.
func NewOrchestrateLifecycle(l *SQLLifecycle) orchestrate.LifecycleService {
	return orchestrateLifecycle{l}
}

func (a orchestrateLifecycle) GetNode(ctx context.Context, nodeID string) (*orchestrate.NodeInfo, error) {
	r, err := a.get(ctx, nodeID)
	if err != nil || r == nil {
		return nil, err
	}
	return &orchestrate.NodeInfo{
		ID:         strconv.FormatInt(r.id, 10),
		Tier:       r.tier,
		Region:     r.region,
		Role:       r.role,
		ProviderID: strconv.FormatInt(r.providerID, 10),
		RealitySNI: r.sni,
		RealityPBK: r.pbk,
		HY2Port:    r.hy2Port,
		NameZH:     r.nameZH,
		NameEn:     r.nameEn,
	}, nil
}

func (a orchestrateLifecycle) TransitionStatus(ctx context.Context, nodeID string, from, to string, detail map[string]any) (int, error) {
	return a.transition(ctx, nodeID, from, to, detail)
}

func (a orchestrateLifecycle) SetWeight(ctx context.Context, nodeID string, weight int) error {
	return a.setWeight(ctx, nodeID, weight)
}

func (a orchestrateLifecycle) BumpVersion(ctx context.Context) error { return a.bumpVersion(ctx) }

func (a orchestrateLifecycle) WriteAuditLog(ctx context.Context, actor, action, target, meta string) error {
	return a.writeAudit(ctx, actor, action, target, meta)
}

// assert at compile time that the adapters satisfy their interfaces.
var (
	_ detect.LifecycleService      = detectLifecycle{}
	_ orchestrate.LifecycleService = orchestrateLifecycle{}
)