story/internal/diagnostics/integration_issues.go

package diagnostics

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/concurrency"
	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/idempotency"
	"data-terminal-cloud/internal/identity"

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

var integrationIssueStatuses = map[string]bool{
	"OPEN": true, "IN_PROGRESS": true, "RESOLVED": true, "VERIFIED": true, "CLOSED": true,
}

var integrationIssueSeverities = map[string]bool{
	"LOW": true, "MEDIUM": true, "HIGH": true, "CRITICAL": true,
}

type IntegrationIssue struct {
	ID                   string    `json:"id"`
	TenantID             string    `json:"tenant_id"`
	Phenomenon           string    `json:"phenomenon"`
	Severity             string    `json:"severity"`
	DeviceID             string    `json:"device_id,omitempty"`
	ConversationID       string    `json:"conversation_id,omitempty"`
	OccurredAt           time.Time `json:"occurred_at"`
	RequestID            string    `json:"request_id,omitempty"`
	ResponsibleParty     string    `json:"responsible_party"`
	Status               string    `json:"status"`
	ResolutionResult     string    `json:"resolution_result,omitempty"`
	RootCause            string    `json:"root_cause,omitempty"`
	FixedVersion         string    `json:"fixed_version,omitempty"`
	RegressionConclusion string    `json:"regression_conclusion,omitempty"`
	RegressionEvidence   string    `json:"regression_evidence,omitempty"`
	CreatedBy            string    `json:"created_by"`
	UpdatedBy            string    `json:"updated_by"`
	Version              int64     `json:"version"`
	CreatedAt            time.Time `json:"created_at"`
	UpdatedAt            time.Time `json:"updated_at"`
}

type CreateIntegrationIssueInput struct {
	Phenomenon       string    `json:"phenomenon"`
	Severity         string    `json:"severity"`
	DeviceID         string    `json:"device_id,omitempty"`
	ConversationID   string    `json:"conversation_id,omitempty"`
	OccurredAt       time.Time `json:"occurred_at"`
	RequestID        string    `json:"request_id,omitempty"`
	ResponsibleParty string    `json:"responsible_party"`
}

type UpdateIntegrationIssueInput struct {
	Status               *string `json:"status,omitempty"`
	Severity             *string `json:"severity,omitempty"`
	ResponsibleParty     *string `json:"responsible_party,omitempty"`
	ResolutionResult     *string `json:"resolution_result,omitempty"`
	RootCause            *string `json:"root_cause,omitempty"`
	FixedVersion         *string `json:"fixed_version,omitempty"`
	RegressionConclusion *string `json:"regression_conclusion,omitempty"`
	RegressionEvidence   *string `json:"regression_evidence,omitempty"`
}

type IntegrationIssueSummary struct {
	Total                int64 `json:"total"`
	Open                 int64 `json:"open"`
	CriticalOpen         int64 `json:"critical_open"`
	HighOpen             int64 `json:"high_open"`
	ReleaseBlocking      int64 `json:"release_blocking"`
	ReleaseGateSatisfied bool  `json:"release_gate_satisfied"`
}

func (s *Service) CreateIntegrationIssue(ctx context.Context, principal identity.Principal, tenantPublicID, key, requestID string, input CreateIntegrationIssueInput) (IntegrationIssue, bool, error) {
	normalizeIntegrationIssueCreate(&input)
	if !validIntegrationIssueCreate(input) {
		return IntegrationIssue{}, false, ErrInvalid
	}
	currentActor, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return IntegrationIssue{}, false, err
	}
	requestBody, _ := json.Marshal(input)
	result, err := s.idem.Execute(ctx, idempotency.Request{
		TenantID: &currentActor.tenantID, ActorKey: "ops:" + currentActor.userPublicID,
		Scope: "POST /api/v1/ops/integration-issues", Key: key,
		RequestHash: idempotency.Hash(requestBody), ExpiresAfter: 24 * time.Hour,
	}, func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		deviceID, conversationID, resolveErr := resolveIntegrationIssueReferences(operationContext, tx, currentActor.tenantID, input.DeviceID, input.ConversationID)
		if resolveErr != nil {
			return idempotency.Result{}, resolveErr
		}
		publicID := uuid.NewString()
		var internalID string
		if queryErr := tx.QueryRow(operationContext, `
			INSERT INTO integration_issue(public_id,tenant_id,phenomenon,severity,device_id,conversation_id,occurred_at,request_id,responsible_party,created_by,updated_by)
			VALUES($1::uuid,$2::uuid,$3,$4,$5::uuid,$6::uuid,$7,NULLIF($8,''),$9,$10::uuid,$10::uuid)
			RETURNING id::text`, publicID, currentActor.tenantID, input.Phenomenon, input.Severity, deviceID, conversationID, input.OccurredAt, input.RequestID, input.ResponsibleParty, currentActor.userID).Scan(&internalID); queryErr != nil {
			return idempotency.Result{}, queryErr
		}
		resourceID := internalID
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{
			TenantID: &currentActor.tenantID, ActorType: "user", ActorPublicID: &currentActor.userPublicID,
			Action: "ops.integration_issue.create", ResourceType: "integration_issue", ResourceID: &resourceID,
			RequestID: requestID, Purpose: "记录联调问题", Result: "success",
			Detail: map[string]any{"issue_id": publicID, "severity": input.Severity, "device_id": input.DeviceID, "conversation_id": input.ConversationID, "responsible_party": input.ResponsibleParty},
		}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		item, queryErr := getIntegrationIssue(operationContext, tx, currentActor.tenantID, publicID, false)
		if queryErr != nil {
			return idempotency.Result{}, queryErr
		}
		body, _ := json.Marshal(item)
		return idempotency.Result{StatusCode: 201, Body: body, ResourceID: &internalID}, nil
	})
	if err != nil {
		return IntegrationIssue{}, false, err
	}
	var item IntegrationIssue
	if err := json.Unmarshal(result.Body, &item); err != nil {
		return IntegrationIssue{}, false, err
	}
	return item, !result.Replayed, nil
}

func (s *Service) ListIntegrationIssues(ctx context.Context, principal identity.Principal, tenantPublicID, status string) ([]IntegrationIssue, error) {
	currentActor, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin")
	if err != nil {
		return nil, err
	}
	status = strings.ToUpper(strings.TrimSpace(status))
	if status != "" && !integrationIssueStatuses[status] {
		return nil, ErrInvalid
	}
	rows, err := s.pool.Query(ctx, integrationIssueSelect+` WHERE ii.tenant_id=$1::uuid AND (NULLIF($2,'') IS NULL OR ii.status=$2) ORDER BY ii.updated_at DESC,ii.id DESC LIMIT 100`, currentActor.tenantID, status)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []IntegrationIssue{}
	for rows.Next() {
		item, scanErr := scanIntegrationIssue(rows)
		if scanErr != nil {
			return nil, scanErr
		}
		items = append(items, item)
	}
	return items, rows.Err()
}

func (s *Service) SummarizeIntegrationIssues(ctx context.Context, principal identity.Principal, tenantPublicID string) (IntegrationIssueSummary, error) {
	currentActor, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin")
	if err != nil {
		return IntegrationIssueSummary{}, err
	}
	var summary IntegrationIssueSummary
	err = s.pool.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`, currentActor.tenantID).Scan(
		&summary.Total, &summary.Open, &summary.CriticalOpen, &summary.HighOpen, &summary.ReleaseBlocking,
	)
	if err != nil {
		return IntegrationIssueSummary{}, err
	}
	summary.ReleaseGateSatisfied = summary.ReleaseBlocking == 0
	return summary, nil
}

func (s *Service) UpdateIntegrationIssue(ctx context.Context, principal identity.Principal, tenantPublicID, publicID, requestID string, expectedVersion int64, input UpdateIntegrationIssueInput) (IntegrationIssue, error) {
	if _, err := uuid.Parse(publicID); err != nil {
		return IntegrationIssue{}, ErrInvalid
	}
	normalizeIntegrationIssueUpdate(&input)
	if input.Status == nil && input.Severity == nil && input.ResponsibleParty == nil && input.ResolutionResult == nil && input.RootCause == nil && input.FixedVersion == nil && input.RegressionConclusion == nil && input.RegressionEvidence == nil {
		return IntegrationIssue{}, ErrInvalid
	}
	currentActor, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_operator", "ops_admin")
	if err != nil {
		return IntegrationIssue{}, err
	}
	var updated IntegrationIssue
	err = database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
		current, lockErr := getIntegrationIssue(ctx, tx, currentActor.tenantID, publicID, true)
		if lockErr != nil {
			return lockErr
		}
		if err := concurrency.RequireVersion(expectedVersion, current.Version); err != nil {
			return err
		}
		previousStatus := current.Status
		if input.Status != nil {
			current.Status = *input.Status
		}
		if input.Severity != nil {
			current.Severity = *input.Severity
		}
		if input.ResponsibleParty != nil {
			current.ResponsibleParty = *input.ResponsibleParty
		}
		if input.ResolutionResult != nil {
			current.ResolutionResult = *input.ResolutionResult
		}
		if input.RootCause != nil {
			current.RootCause = *input.RootCause
		}
		if input.FixedVersion != nil {
			current.FixedVersion = *input.FixedVersion
		}
		if input.RegressionConclusion != nil {
			current.RegressionConclusion = *input.RegressionConclusion
		}
		if input.RegressionEvidence != nil {
			current.RegressionEvidence = *input.RegressionEvidence
		}
		if !validIntegrationIssueState(current) {
			return ErrInvalid
		}
		command, commandErr := tx.Exec(ctx, `
			UPDATE integration_issue
			SET status=$3,severity=$4,responsible_party=$5,resolution_result=NULLIF($6,''),root_cause=NULLIF($7,''),fixed_version=NULLIF($8,''),
			    regression_conclusion=NULLIF($9,''),regression_evidence=NULLIF($10,''),updated_by=$11::uuid,version=version+1,updated_at=now()
			WHERE tenant_id=$1::uuid AND public_id=$2::uuid AND version=$12`, currentActor.tenantID, publicID, current.Status, current.Severity, current.ResponsibleParty, current.ResolutionResult, current.RootCause, current.FixedVersion, current.RegressionConclusion, current.RegressionEvidence, currentActor.userID, expectedVersion)
		if commandErr != nil {
			return commandErr
		}
		if command.RowsAffected() != 1 {
			return concurrency.ErrVersionConflict
		}
		var internalID string
		if queryErr := tx.QueryRow(ctx, `SELECT id::text FROM integration_issue WHERE tenant_id=$1::uuid AND public_id=$2::uuid`, currentActor.tenantID, publicID).Scan(&internalID); queryErr != nil {
			return queryErr
		}
		if auditErr := s.audit.Record(ctx, tx, audit.Event{
			TenantID: &currentActor.tenantID, ActorType: "user", ActorPublicID: &currentActor.userPublicID,
			Action: "ops.integration_issue.update", ResourceType: "integration_issue", ResourceID: &internalID,
			RequestID: requestID, Purpose: "维护联调问题闭环", Result: "success",
			Detail: map[string]any{"issue_id": publicID, "from_status": previousStatus, "to_status": current.Status, "severity": current.Severity, "fixed_version": current.FixedVersion, "responsible_party": current.ResponsibleParty},
		}); auditErr != nil {
			return auditErr
		}
		updated, lockErr = getIntegrationIssue(ctx, tx, currentActor.tenantID, publicID, false)
		return lockErr
	})
	return updated, err
}

func normalizeIntegrationIssueCreate(input *CreateIntegrationIssueInput) {
	input.Phenomenon = strings.TrimSpace(input.Phenomenon)
	input.Severity = strings.ToUpper(strings.TrimSpace(input.Severity))
	input.DeviceID = strings.TrimSpace(input.DeviceID)
	input.ConversationID = strings.TrimSpace(input.ConversationID)
	input.RequestID = strings.TrimSpace(input.RequestID)
	input.ResponsibleParty = strings.TrimSpace(input.ResponsibleParty)
}

func validIntegrationIssueCreate(input CreateIntegrationIssueInput) bool {
	if input.Phenomenon == "" || len([]rune(input.Phenomenon)) > 2000 || !integrationIssueSeverities[input.Severity] || input.ResponsibleParty == "" || len([]rune(input.ResponsibleParty)) > 120 || input.OccurredAt.IsZero() {
		return false
	}
	if input.DeviceID != "" {
		if _, err := uuid.Parse(input.DeviceID); err != nil {
			return false
		}
	}
	if input.ConversationID != "" {
		if _, err := uuid.Parse(input.ConversationID); err != nil {
			return false
		}
	}
	return input.RequestID == "" || requestIDPattern.MatchString(input.RequestID)
}

func normalizeIntegrationIssueUpdate(input *UpdateIntegrationIssueInput) {
	if input.Status != nil {
		value := strings.ToUpper(strings.TrimSpace(*input.Status))
		input.Status = &value
	}
	if input.Severity != nil {
		value := strings.ToUpper(strings.TrimSpace(*input.Severity))
		input.Severity = &value
	}
	trimIssueField(&input.ResponsibleParty)
	trimIssueField(&input.ResolutionResult)
	trimIssueField(&input.RootCause)
	trimIssueField(&input.FixedVersion)
	trimIssueField(&input.RegressionConclusion)
	trimIssueField(&input.RegressionEvidence)
}

func trimIssueField(value **string) {
	if *value != nil {
		trimmed := strings.TrimSpace(**value)
		*value = &trimmed
	}
}

func validIntegrationIssueState(item IntegrationIssue) bool {
	if !integrationIssueStatuses[item.Status] || !integrationIssueSeverities[item.Severity] || item.ResponsibleParty == "" || len([]rune(item.ResponsibleParty)) > 120 || len([]rune(item.ResolutionResult)) > 4000 || len([]rune(item.RootCause)) > 4000 || len([]rune(item.FixedVersion)) > 128 || len([]rune(item.RegressionConclusion)) > 4000 || len([]rune(item.RegressionEvidence)) > 4000 {
		return false
	}
	if (item.Status == "RESOLVED" || item.Status == "VERIFIED" || item.Status == "CLOSED") && (item.ResolutionResult == "" || item.RootCause == "" || item.FixedVersion == "") {
		return false
	}
	return (item.Status != "VERIFIED" && item.Status != "CLOSED") || (item.RegressionConclusion != "" && item.RegressionEvidence != "")
}

func resolveIntegrationIssueReferences(ctx context.Context, tx pgx.Tx, tenantID, devicePublicID, conversationPublicID string) (*string, *string, error) {
	var deviceID, conversationID *string
	if devicePublicID != "" {
		var value string
		if err := tx.QueryRow(ctx, `SELECT id::text FROM device WHERE tenant_id=$1::uuid AND public_id=$2::uuid`, tenantID, devicePublicID).Scan(&value); errors.Is(err, pgx.ErrNoRows) {
			return nil, nil, ErrNotFound
		} else if err != nil {
			return nil, nil, err
		}
		deviceID = &value
	}
	if conversationPublicID != "" {
		var value, conversationDeviceID string
		if err := tx.QueryRow(ctx, `SELECT id::text,device_id::text FROM conversation WHERE tenant_id=$1::uuid AND public_id=$2::uuid`, tenantID, conversationPublicID).Scan(&value, &conversationDeviceID); errors.Is(err, pgx.ErrNoRows) {
			return nil, nil, ErrNotFound
		} else if err != nil {
			return nil, nil, err
		}
		if deviceID != nil && *deviceID != conversationDeviceID {
			return nil, nil, ErrInvalid
		}
		conversationID = &value
	}
	return deviceID, conversationID, nil
}

const integrationIssueSelect = `
	SELECT ii.public_id::text,t.public_id::text,ii.phenomenon,ii.severity,COALESCE(d.public_id::text,''),COALESCE(c.public_id::text,''),
	       ii.occurred_at,COALESCE(ii.request_id,''),ii.responsible_party,ii.status,
	       COALESCE(ii.resolution_result,''),COALESCE(ii.root_cause,''),COALESCE(ii.fixed_version,''),COALESCE(ii.regression_conclusion,''),COALESCE(ii.regression_evidence,''),
	       creator.public_id::text,updater.public_id::text,ii.version,ii.created_at,ii.updated_at
	FROM integration_issue ii
	JOIN tenant t ON t.id=ii.tenant_id
	JOIN user_account creator ON creator.id=ii.created_by AND creator.tenant_id=ii.tenant_id
	JOIN user_account updater ON updater.id=ii.updated_by AND updater.tenant_id=ii.tenant_id
	LEFT JOIN device d ON d.id=ii.device_id AND d.tenant_id=ii.tenant_id
	LEFT JOIN conversation c ON c.id=ii.conversation_id AND c.tenant_id=ii.tenant_id`

type integrationIssueScanner interface{ Scan(...any) error }

func scanIntegrationIssue(row integrationIssueScanner) (IntegrationIssue, error) {
	var item IntegrationIssue
	err := row.Scan(&item.ID, &item.TenantID, &item.Phenomenon, &item.Severity, &item.DeviceID, &item.ConversationID, &item.OccurredAt, &item.RequestID, &item.ResponsibleParty, &item.Status, &item.ResolutionResult, &item.RootCause, &item.FixedVersion, &item.RegressionConclusion, &item.RegressionEvidence, &item.CreatedBy, &item.UpdatedBy, &item.Version, &item.CreatedAt, &item.UpdatedAt)
	return item, err
}

func getIntegrationIssue(ctx context.Context, query interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}, tenantID, publicID string, lock bool) (IntegrationIssue, error) {
	queryText := integrationIssueSelect + ` WHERE ii.tenant_id=$1::uuid AND ii.public_id=$2::uuid`
	if lock {
		queryText += ` FOR UPDATE OF ii`
	}
	item, err := scanIntegrationIssue(query.QueryRow(ctx, queryText, tenantID, publicID))
	if errors.Is(err, pgx.ErrNoRows) {
		return IntegrationIssue{}, ErrNotFound
	}
	return item, err
}