yms-daemon/internal/backendstatus/reconcile.go

package backendstatus

import (
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"

	"yms-daemon/internal/containerengine"
	"yms-daemon/internal/daemonapi"
	"yms-daemon/internal/deploymentconfig"
	"yms-daemon/internal/transaction"
)

// reconcileSource 表示对账修复事务的来源标识。
const reconcileSource = "local-cli"

// reconcileContainer 执行 container 后端的对账：只读时返回修复计划，apply 时执行自动修复。
func (d *Diagnoser) reconcileContainer(ctx context.Context) (daemonapi.Diagnosis, error) {
	findings, err := d.containerGather(ctx)
	if err != nil {
		return daemonapi.Diagnosis{}, err
	}
	return d.applyFixes(ctx, deploymentconfig.BackendTypeContainer, findings)
}

// reconcileNative 执行 native 后端的对账。当前 native 没有可自动修复项，仅返回只读诊断。
func (d *Diagnoser) reconcileNative(ctx context.Context) (daemonapi.Diagnosis, error) {
	findings, err := d.nativeGather(ctx)
	if err != nil {
		return daemonapi.Diagnosis{}, err
	}
	return d.applyFixes(ctx, deploymentconfig.BackendTypeNative, findings)
}

// applyFixes 对诊断结果中的可自动修复项执行修复动作。
// 若没有可修复项则直接返回只读诊断；否则创建对账事务，通过 coordinator 对每个修复动作
// 先记录意图、再执行、再核对，全部成功后把事务推进到 COMMITTED，最后重新诊断返回修复后状态。
func (d *Diagnoser) applyFixes(ctx context.Context, runtimeType string, findings []finding) (daemonapi.Diagnosis, error) {
	var fixable []finding
	for _, f := range findings {
		if f.level == daemonapi.DiagnosisLevelFixable && f.fix != nil {
			fixable = append(fixable, f)
		}
	}
	if len(fixable) == 0 {
		return toDiagnosis("backend", runtimeType, findings), nil
	}

	transactionID := rand.Text()
	codes := make([]string, 0, len(fixable))
	for _, f := range fixable {
		codes = append(codes, f.code)
	}
	requestJSON, err := json.Marshal(struct {
		Service string   `json:"service"`
		Type    string   `json:"type"`
		Fixes   []string `json:"fixes"`
	}{"backend", runtimeType, codes})
	if err != nil {
		return daemonapi.Diagnosis{}, fmt.Errorf("encode reconcile request: %w", err)
	}
	if _, _, err := d.store.CreateTransaction(ctx, transaction.CreateRequest{
		ID:             transactionID,
		IdempotencyKey: "backend:reconcile:" + rand.Text(),
		Source:         reconcileSource,
		Service:        "backend",
		Request:        requestJSON,
	}); err != nil {
		return daemonapi.Diagnosis{}, fmt.Errorf("create reconcile transaction: %w", err)
	}

	for _, f := range fixable {
		intent, operation, err := d.fixOperation(f)
		if err != nil {
			_, _ = d.store.Transition(ctx, transactionID, transaction.StateFailed, err.Error())
			return daemonapi.Diagnosis{}, err
		}
		if _, err := d.coordinator.ExecuteStep(ctx, transactionID, intent, operation); err != nil {
			_, _ = d.store.Transition(ctx, transactionID, transaction.StateFailed, err.Error())
			return daemonapi.Diagnosis{}, fmt.Errorf("apply reconcile fix %s: %w", f.code, err)
		}
	}

	for _, next := range []transaction.State{
		transaction.StateValidating,
		transaction.StatePrepared,
		transaction.StateStarting,
		transaction.StateSwitching,
		transaction.StateVerifying,
		transaction.StateDraining,
		transaction.StateCommitted,
	} {
		if _, err := d.store.Transition(ctx, transactionID, next, "reconcile"); err != nil {
			return daemonapi.Diagnosis{}, fmt.Errorf("commit reconcile transaction: %w", err)
		}
	}

	diagnosis, err := d.Diagnose(ctx)
	if err != nil {
		return daemonapi.Diagnosis{}, err
	}
	diagnosis.RepairApplied = true
	diagnosis.RepairTransactionID = transactionID
	return diagnosis, nil
}

// fixOperation 把一条可修复诊断项转换为对账事务中可执行的步骤意图与操作。
func (d *Diagnoser) fixOperation(f finding) (transaction.StepIntent, transaction.Operation, error) {
	switch f.fix.kind {
	case fixSwitchPort:
		intent := reconcileIntent("reconcile.nginx.switch."+strconv.Itoa(f.fix.port), "switch host Nginx to committed backend port", struct {
			Port int `json:"port"`
		}{f.fix.port})
		return intent, &switchNginxOperation{gateway: d.gateway, port: f.fix.port}, nil
	case fixRemoveContainer:
		intent := reconcileIntent("reconcile.container.remove."+f.fix.container, "remove residual stopped backend container", struct {
			ContainerName string `json:"containerName"`
		}{f.fix.container})
		return intent, &removeContainerOperation{engine: d.engine, name: f.fix.container}, nil
	default:
		return transaction.StepIntent{}, nil, fmt.Errorf("unsupported reconcile fix kind %q", f.fix.kind)
	}
}

// reconcileIntent 构造对账修复步骤的持久化意图。
func reconcileIntent(key, name string, value any) transaction.StepIntent {
	payload, err := json.Marshal(value)
	if err != nil {
		panic(fmt.Sprintf("marshal reconcile intent: %v", err))
	}
	return transaction.StepIntent{Key: key, Name: name, Intent: payload}
}

// switchNginxOperation 把宿主 Nginx 切流到指定端口，实现 transaction.Operation。
type switchNginxOperation struct {
	gateway gateway
	port    int
}

// Apply 执行宿主 Nginx 切流，切到当前已指向的端口时为空操作。
func (o *switchNginxOperation) Apply(ctx context.Context) error {
	_, err := o.gateway.Switch(ctx, o.port)
	return err
}

// Inspect 核对宿主 Nginx 当前活动端口是否等于目标端口。
func (o *switchNginxOperation) Inspect(ctx context.Context) (transaction.Inspection, error) {
	snapshot, err := o.gateway.Read()
	if err != nil {
		return transaction.Inspection{}, err
	}
	if snapshot.ActivePort == o.port {
		return transaction.Inspection{Status: transaction.InspectionApplied}, nil
	}
	return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil
}

// removeContainerOperation 移除一个残留容器，实现 transaction.Operation。
type removeContainerOperation struct {
	engine containerengine.Engine
	name   string
}

// Apply 强制移除目标容器，容器不存在时视为成功。
func (o *removeContainerOperation) Apply(ctx context.Context) error {
	err := o.engine.RemoveContainer(ctx, o.name, true)
	if errors.Is(err, containerengine.ErrNotFound) {
		return nil
	}
	return err
}

// Inspect 核对目标容器是否已经不存在。
func (o *removeContainerOperation) Inspect(ctx context.Context) (transaction.Inspection, error) {
	if _, err := o.engine.InspectContainer(ctx, o.name); errors.Is(err, containerengine.ErrNotFound) {
		return transaction.Inspection{Status: transaction.InspectionApplied}, nil
	} else if err != nil {
		return transaction.Inspection{}, err
	}
	return transaction.Inspection{Status: transaction.InspectionNotApplied}, nil
}

var _ transaction.Operation = (*switchNginxOperation)(nil)
var _ transaction.Operation = (*removeContainerOperation)(nil)