story/internal/diagnostics/acceptance_snapshot.go

package diagnostics

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"sort"
	"strings"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/buildinfo"
	"data-terminal-cloud/internal/idempotency"
	"data-terminal-cloud/internal/identity"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

var (
	acceptanceCandidatePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._+-]{0,127}$`)
	acceptanceDigestPattern    = regexp.MustCompile(`^sha256:[a-f0-9]{64}$`)
	acceptanceEnvironments     = map[string]bool{"LAB": true, "STAGING": true, "PRODUCTION": true}
	acceptanceRunTypes         = map[string]bool{"REAL_DEVICE_PROVIDER": true, "PILOT": true, "PRODUCTION_RELEASE": true}
	acceptanceDependencies     = map[string]string{"oidc_jwks": "api", "ark_tls": "worker", "volc_rtc_tls": "worker", "content_safety_tls": "worker"}
)

type AcceptanceSnapshotInput struct {
	CandidateVersion     string   `json:"candidate_version"`
	BuildDigest          string   `json:"build_digest"`
	Environment          string   `json:"environment"`
	RunType              string   `json:"run_type"`
	Scope                string   `json:"scope"`
	RequiredDependencies []string `json:"required_dependencies"`
	Reason               string   `json:"reason"`
}

type AcceptanceDatabaseState struct {
	MigrationVersion int64 `json:"migration_version"`
	Dirty            bool  `json:"dirty"`
}

type AcceptanceDependencyState struct {
	ServiceName         string     `json:"service_name"`
	Dependency          string     `json:"dependency"`
	Status              string     `json:"status"`
	CheckedAt           *time.Time `json:"checked_at,omitempty"`
	LastSuccessAt       *time.Time `json:"last_success_at,omitempty"`
	LastFailureAt       *time.Time `json:"last_failure_at,omitempty"`
	LastErrorCode       string     `json:"last_error_code,omitempty"`
	LatencyMS           int64      `json:"latency_ms"`
	ProbeIntervalMS     int64      `json:"probe_interval_ms"`
	ConsecutiveFailures int        `json:"consecutive_failures"`
	Fresh               bool       `json:"fresh"`
	GateSatisfied       bool       `json:"gate_satisfied"`
}

type AcceptanceOutboxState struct {
	Pending    int64 `json:"pending"`
	Processing int64 `json:"processing"`
	Dead       int64 `json:"dead"`
}

type AcceptanceAlertState struct {
	CriticalOpen           int64                     `json:"critical_open"`
	HighOpen               int64                     `json:"high_open"`
	Blocking               int64                     `json:"blocking"`
	BlockingItems          []AcceptanceBlockingAlert `json:"blocking_items"`
	BlockingItemsTruncated bool                      `json:"blocking_items_truncated"`
}

type AcceptancePlatformHealthState struct {
	State         string     `json:"state"`
	Status        string     `json:"status,omitempty"`
	ObservedAt    *time.Time `json:"observed_at,omitempty"`
	ReceivedAt    *time.Time `json:"received_at,omitempty"`
	CheckCount    int        `json:"check_count"`
	WarningCount  int        `json:"warning_count"`
	CriticalCount int        `json:"critical_count"`
	GateSatisfied bool       `json:"gate_satisfied"`
}

type AcceptanceBlockingAlert struct {
	ID              string    `json:"id"`
	RuleCode        string    `json:"rule_code"`
	Severity        string    `json:"severity"`
	Status          string    `json:"status"`
	ResourceType    string    `json:"resource_type,omitempty"`
	ResourceID      string    `json:"resource_id,omitempty"`
	ResourceState   string    `json:"resource_state"`
	FirstDetectedAt time.Time `json:"first_detected_at"`
	LastDetectedAt  time.Time `json:"last_detected_at"`
}

type AcceptanceGate struct {
	Code      string `json:"code"`
	Satisfied bool   `json:"satisfied"`
	Summary   string `json:"summary"`
}

type AcceptanceSnapshot struct {
	ID                   string                        `json:"id"`
	GeneratedAt          time.Time                     `json:"generated_at"`
	TenantID             string                        `json:"tenant_id"`
	CandidateVersion     string                        `json:"candidate_version"`
	BuildDigest          string                        `json:"build_digest"`
	Environment          string                        `json:"environment"`
	RunType              string                        `json:"run_type"`
	Scope                string                        `json:"scope"`
	RedactionPolicy      string                        `json:"redaction_policy"`
	RuntimeBuild         buildinfo.Info                `json:"runtime_build"`
	Database             AcceptanceDatabaseState       `json:"database"`
	IntegrationIssues    IntegrationIssueSummary       `json:"integration_issues"`
	ExternalDependencies []AcceptanceDependencyState   `json:"external_dependencies"`
	Outbox               AcceptanceOutboxState         `json:"outbox"`
	Alerts               AcceptanceAlertState          `json:"alerts"`
	PlatformHealth       AcceptancePlatformHealthState `json:"platform_health"`
	Gates                []AcceptanceGate              `json:"gates"`
	OverallGateSatisfied bool                          `json:"overall_gate_satisfied"`
	Disclaimer           string                        `json:"disclaimer"`
}

func (s *Service) ExportAcceptanceSnapshot(ctx context.Context, principal identity.Principal, tenantPublicID, key, requestID string, input AcceptanceSnapshotInput) (AcceptanceSnapshot, bool, error) {
	normalizeAcceptanceSnapshotInput(&input)
	if !validAcceptanceSnapshotInput(input) {
		return AcceptanceSnapshot{}, false, ErrInvalid
	}
	a, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return AcceptanceSnapshot{}, false, err
	}
	requestBody, _ := json.Marshal(input)
	result, err := s.idem.Execute(ctx, idempotency.Request{
		TenantID: &a.tenantID, ActorKey: "ops:" + a.userPublicID,
		Scope: "POST /api/v1/ops/diagnostics/acceptance-snapshots", Key: key,
		RequestHash: idempotency.Hash(requestBody), ExpiresAfter: 24 * time.Hour,
	}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		snapshot, snapshotErr := buildAcceptanceSnapshot(operationContext, tx, tenantPublicID, a.tenantID, input)
		if snapshotErr != nil {
			return idempotency.Result{}, snapshotErr
		}
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{
			TenantID: &a.tenantID, ActorType: "user", ActorPublicID: &a.userPublicID,
			Action: "ops.diagnostics.acceptance_snapshot.export", ResourceType: "acceptance_snapshot", ResourceID: &snapshot.ID,
			RequestID: requestID, Purpose: input.Reason, Result: "success",
			Detail: map[string]any{
				"candidate_version": input.CandidateVersion, "build_digest": input.BuildDigest,
				"environment": input.Environment, "run_type": input.RunType, "scope": input.Scope,
				"required_dependencies": input.RequiredDependencies, "overall_gate_satisfied": snapshot.OverallGateSatisfied,
				"redaction_policy": snapshot.RedactionPolicy,
			},
		}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		encoded, _ := json.Marshal(snapshot)
		return idempotency.Result{StatusCode: 201, Body: encoded, ResourceID: &snapshot.ID}, nil
	})
	if err != nil {
		return AcceptanceSnapshot{}, false, err
	}
	var snapshot AcceptanceSnapshot
	if err = json.Unmarshal(result.Body, &snapshot); err != nil {
		return AcceptanceSnapshot{}, false, err
	}
	return snapshot, !result.Replayed, nil
}

func normalizeAcceptanceSnapshotInput(input *AcceptanceSnapshotInput) {
	input.CandidateVersion = strings.TrimSpace(input.CandidateVersion)
	input.BuildDigest = strings.ToLower(strings.TrimSpace(input.BuildDigest))
	input.Environment = strings.ToUpper(strings.TrimSpace(input.Environment))
	input.RunType = strings.ToUpper(strings.TrimSpace(input.RunType))
	input.Scope = strings.TrimSpace(input.Scope)
	input.Reason = strings.TrimSpace(input.Reason)
	for index := range input.RequiredDependencies {
		input.RequiredDependencies[index] = strings.ToLower(strings.TrimSpace(input.RequiredDependencies[index]))
	}
	sort.Strings(input.RequiredDependencies)
}

func validAcceptanceSnapshotInput(input AcceptanceSnapshotInput) bool {
	if !acceptanceCandidatePattern.MatchString(input.CandidateVersion) || !acceptanceDigestPattern.MatchString(input.BuildDigest) ||
		!acceptanceEnvironments[input.Environment] || !acceptanceRunTypes[input.RunType] ||
		input.Scope == "" || len([]rune(input.Scope)) > 500 || input.Reason == "" || len([]rune(input.Reason)) > 255 ||
		len(input.RequiredDependencies) == 0 || len(input.RequiredDependencies) > len(acceptanceDependencies) {
		return false
	}
	seen := map[string]bool{}
	for _, dependency := range input.RequiredDependencies {
		if _, allowed := acceptanceDependencies[dependency]; !allowed || seen[dependency] {
			return false
		}
		seen[dependency] = true
	}
	return true
}

func buildAcceptanceSnapshot(ctx context.Context, tx pgx.Tx, tenantPublicID, tenantInternalID string, input AcceptanceSnapshotInput) (AcceptanceSnapshot, error) {
	now := time.Now().UTC()
	snapshot := AcceptanceSnapshot{
		ID: uuid.NewString(), GeneratedAt: now, TenantID: tenantPublicID,
		CandidateVersion: input.CandidateVersion, BuildDigest: input.BuildDigest, Environment: input.Environment,
		RunType: input.RunType, Scope: input.Scope, RedactionPolicy: "ops-acceptance-minimal-v1",
		RuntimeBuild: buildinfo.Current(), Alerts: AcceptanceAlertState{BlockingItems: []AcceptanceBlockingAlert{}},
		Disclaimer: "本快照仅证明生成时刻的技术预检状态，不替代真实设备、真实供应商、灰度观察或发布审批结论。",
	}
	if err := tx.QueryRow(ctx, `SELECT version,dirty FROM schema_migrations ORDER BY version DESC LIMIT 1`).Scan(&snapshot.Database.MigrationVersion, &snapshot.Database.Dirty); err != nil {
		return AcceptanceSnapshot{}, err
	}
	if err := tx.QueryRow(ctx, `
		SELECT count(*),
		       count(*) FILTER (WHERE status <> 'CLOSED'),
		       count(*) FILTER (WHERE severity='CRITICAL' AND status <> 'CLOSED'),
		       count(*) FILTER (WHERE severity='HIGH' AND status <> 'CLOSED'),
		       count(*) FILTER (WHERE severity IN ('CRITICAL','HIGH') AND status <> 'CLOSED')
		FROM integration_issue WHERE tenant_id=$1::uuid`, tenantInternalID).Scan(
		&snapshot.IntegrationIssues.Total, &snapshot.IntegrationIssues.Open, &snapshot.IntegrationIssues.CriticalOpen,
		&snapshot.IntegrationIssues.HighOpen, &snapshot.IntegrationIssues.ReleaseBlocking,
	); err != nil {
		return AcceptanceSnapshot{}, err
	}
	snapshot.IntegrationIssues.ReleaseGateSatisfied = snapshot.IntegrationIssues.ReleaseBlocking == 0

	for _, dependency := range input.RequiredDependencies {
		state := AcceptanceDependencyState{ServiceName: acceptanceDependencies[dependency], Dependency: dependency, Status: "MISSING"}
		var checkedAt time.Time
		err := tx.QueryRow(ctx, `SELECT status,checked_at,last_success_at,last_failure_at,COALESCE(last_error_code,''),latency_ms,probe_interval_ms,consecutive_failures FROM external_dependency_probe_state WHERE service_name=$1 AND dependency=$2`, state.ServiceName, dependency).Scan(
			&state.Status, &checkedAt, &state.LastSuccessAt, &state.LastFailureAt, &state.LastErrorCode,
			&state.LatencyMS, &state.ProbeIntervalMS, &state.ConsecutiveFailures,
		)
		if err != nil && err != pgx.ErrNoRows {
			return AcceptanceSnapshot{}, err
		}
		if err == nil {
			state.CheckedAt = &checkedAt
			state.Fresh = !checkedAt.After(now) && now.Sub(checkedAt) <= 3*time.Duration(state.ProbeIntervalMS)*time.Millisecond
			state.GateSatisfied = state.Status == "AVAILABLE" && state.Fresh
		}
		snapshot.ExternalDependencies = append(snapshot.ExternalDependencies, state)
	}
	if err := tx.QueryRow(ctx, `SELECT
		count(*) FILTER (WHERE status='pending'),
		count(*) FILTER (WHERE status='processing'),
		count(*) FILTER (WHERE status='dead')
		FROM outbox_event WHERE tenant_id=$1::uuid OR tenant_id IS NULL`, tenantInternalID).Scan(
		&snapshot.Outbox.Pending, &snapshot.Outbox.Processing, &snapshot.Outbox.Dead,
	); err != nil {
		return AcceptanceSnapshot{}, err
	}
	rows, err := tx.Query(ctx, `SELECT public_id::text,rule_code,severity,status,COALESCE(resource_type,''),COALESCE(resource_public_id::text,''),
		CASE
			WHEN resource_public_id IS NULL THEN 'AGGREGATE'
			WHEN resource_type='conversation' THEN CASE WHEN EXISTS(SELECT 1 FROM conversation c WHERE c.public_id=alert_incident.resource_public_id) THEN 'PRESENT' ELSE 'MISSING' END
			WHEN resource_type='privacy_request' THEN CASE WHEN EXISTS(SELECT 1 FROM privacy_request pr WHERE pr.public_id=alert_incident.resource_public_id) THEN 'PRESENT' ELSE 'MISSING' END
			WHEN resource_type='upload_batch' THEN CASE WHEN EXISTS(SELECT 1 FROM upload_batch ub WHERE ub.public_id=alert_incident.resource_public_id) THEN 'PRESENT' ELSE 'MISSING' END
			WHEN resource_type='firmware_rollout' THEN CASE WHEN EXISTS(SELECT 1 FROM firmware_rollout fr WHERE fr.public_id=alert_incident.resource_public_id) THEN 'PRESENT' ELSE 'MISSING' END
			WHEN resource_type='ops_resource_reference' THEN CASE WHEN EXISTS(SELECT 1 FROM ops_resource_reference rr WHERE rr.public_id=alert_incident.resource_public_id) THEN 'PRESENT' ELSE 'MISSING' END
			ELSE 'UNKNOWN'
		END,first_detected_at,last_detected_at
		FROM alert_incident
		WHERE tenant_id=$1::uuid AND status<>'RESOLVED' AND severity IN ('CRITICAL','HIGH')
		ORDER BY CASE severity WHEN 'CRITICAL' THEN 0 ELSE 1 END,first_detected_at,public_id
		LIMIT 100`, tenantInternalID)
	if err != nil {
		return AcceptanceSnapshot{}, err
	}
	for rows.Next() {
		var item AcceptanceBlockingAlert
		if err = rows.Scan(&item.ID, &item.RuleCode, &item.Severity, &item.Status, &item.ResourceType, &item.ResourceID, &item.ResourceState, &item.FirstDetectedAt, &item.LastDetectedAt); err != nil {
			rows.Close()
			return AcceptanceSnapshot{}, err
		}
		snapshot.Alerts.BlockingItems = append(snapshot.Alerts.BlockingItems, item)
	}
	if err = rows.Err(); err != nil {
		rows.Close()
		return AcceptanceSnapshot{}, err
	}
	rows.Close()
	snapshot.Alerts.BlockingItemsTruncated = snapshot.Alerts.Blocking > int64(len(snapshot.Alerts.BlockingItems))
	if err := tx.QueryRow(ctx, `SELECT
		count(*) FILTER (WHERE severity='CRITICAL'),
		count(*) FILTER (WHERE severity='HIGH'),
		count(*)
		FROM alert_incident WHERE tenant_id=$1::uuid AND status<>'RESOLVED' AND severity IN ('CRITICAL','HIGH')`, tenantInternalID).Scan(
		&snapshot.Alerts.CriticalOpen, &snapshot.Alerts.HighOpen, &snapshot.Alerts.Blocking,
	); err != nil {
		return AcceptanceSnapshot{}, err
	}
	platformHealth, err := loadAcceptancePlatformHealth(ctx, tx, now)
	if err != nil {
		return AcceptanceSnapshot{}, err
	}
	snapshot.PlatformHealth = platformHealth

	runtimeSatisfied := input.Environment == "LAB" || (snapshot.RuntimeBuild.Version == input.CandidateVersion && snapshot.RuntimeBuild.Commit != "unknown" && snapshot.RuntimeBuild.BuildTime != "unknown")
	dependenciesSatisfied := true
	for _, dependency := range snapshot.ExternalDependencies {
		dependenciesSatisfied = dependenciesSatisfied && dependency.GateSatisfied
	}
	outboxSatisfied := snapshot.Outbox.Pending == 0 && snapshot.Outbox.Processing == 0 && snapshot.Outbox.Dead == 0
	alertsSatisfied := snapshot.Alerts.Blocking == 0
	snapshot.Gates = []AcceptanceGate{
		{Code: "candidate_build_identified", Satisfied: runtimeSatisfied, Summary: acceptanceRuntimeSummary(input.Environment)},
		{Code: "database_migration_clean", Satisfied: !snapshot.Database.Dirty, Summary: "数据库迁移必须处于非脏状态"},
		{Code: "required_dependencies_available", Satisfied: dependenciesSatisfied, Summary: "所有必需外部依赖探针必须可用且未过期"},
		{Code: "integration_issues_clear", Satisfied: snapshot.IntegrationIssues.ReleaseGateSatisfied, Summary: "不得存在未关闭的高或严重级联调问题"},
		{Code: "outbox_clear", Satisfied: outboxSatisfied, Summary: "租户及平台级待处理、处理中和死信任务必须清零"},
		{Code: "critical_alerts_clear", Satisfied: alertsSatisfied, Summary: "不得存在未解决的高或严重级告警"},
		{Code: "platform_health_current", Satisfied: snapshot.PlatformHealth.GateSatisfied, Summary: acceptancePlatformHealthSummary(snapshot.PlatformHealth)},
	}
	snapshot.OverallGateSatisfied = true
	for _, gate := range snapshot.Gates {
		snapshot.OverallGateSatisfied = snapshot.OverallGateSatisfied && gate.Satisfied
	}
	return snapshot, nil
}

func loadAcceptancePlatformHealth(ctx context.Context, tx pgx.Tx, now time.Time) (AcceptancePlatformHealthState, error) {
	state := AcceptancePlatformHealthState{State: "MISSING"}
	var observedAt, receivedAt time.Time
	var checksJSON []byte
	err := tx.QueryRow(ctx, `
		SELECT status,observed_at,received_at,checks
		FROM platform_health_snapshot ORDER BY observed_at DESC,id DESC LIMIT 1`).Scan(
		&state.Status, &observedAt, &receivedAt, &checksJSON,
	)
	if errors.Is(err, pgx.ErrNoRows) {
		return state, nil
	}
	if err != nil {
		return AcceptancePlatformHealthState{}, err
	}
	var checks []PlatformHealthCheck
	if err = json.Unmarshal(checksJSON, &checks); err != nil {
		return AcceptancePlatformHealthState{}, err
	}
	return evaluateAcceptancePlatformHealth(state.Status, observedAt, receivedAt, checks, now), nil
}

func evaluateAcceptancePlatformHealth(status string, observedAt, receivedAt time.Time, checks []PlatformHealthCheck, now time.Time) AcceptancePlatformHealthState {
	state := AcceptancePlatformHealthState{
		State: "CURRENT", Status: status, ObservedAt: &observedAt, ReceivedAt: &receivedAt, CheckCount: len(checks),
	}
	for _, check := range checks {
		switch check.Status {
		case "warning":
			state.WarningCount++
		case "critical":
			state.CriticalCount++
		}
	}
	if observedAt.Before(now.Add(-platformHealthMaxAge)) || observedAt.After(now.Add(platformHealthFuture)) {
		state.State = "STALE"
	}
	state.GateSatisfied = state.State == "CURRENT" && state.Status == "healthy" && state.WarningCount == 0 && state.CriticalCount == 0
	return state
}

func acceptancePlatformHealthSummary(state AcceptancePlatformHealthState) string {
	switch state.State {
	case "MISSING":
		return "尚未收到云端主机健康快照"
	case "STALE":
		return "云端主机健康快照必须在15分钟内更新"
	}
	if state.Status != "healthy" || state.WarningCount > 0 || state.CriticalCount > 0 {
		return fmt.Sprintf("云端主机存在%d项预警和%d项严重异常", state.WarningCount, state.CriticalCount)
	}
	return fmt.Sprintf("云端主机健康快照当前有效，共%d项检查", state.CheckCount)
}

func acceptanceRuntimeSummary(environment string) string {
	if environment == "LAB" {
		return "实验室环境允许开发构建，但候选版本与镜像摘要必须明确"
	}
	return "预发布和生产环境的运行版本必须等于候选版本，并包含提交与构建时间"
}