knowledge-governance/internal/platform/postgres/operation_requests.go

package postgres

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"

	"knowledge-governance/internal/governance"
	"knowledge-governance/internal/jobs"
)

var ErrOperationRequestConflict = errors.New("operation request conflicts with an existing request")

type operationExecutor func(pgx.Tx) (json.RawMessage, error)

// ExecuteOperation lets bounded modules share the durable operation transaction
// instead of caching responses after their business writes have committed.
func (s *Store) ExecuteOperation(ctx context.Context, operationType, operationID, fingerprint string, now time.Time, execute func(pgx.Tx) (json.RawMessage, error)) (json.RawMessage, bool, error) {
	return s.executeOperation(ctx, operationType, operationID, fingerprint, now, execute)
}

func (s *Store) RouteCatalogFilesWithOperation(ctx context.Context, request governance.CatalogRouteRequest, operationID string) (governance.CatalogRouteResult, bool, error) {
	if err := validateCatalogRouteRequest(request); err != nil {
		return governance.CatalogRouteResult{}, false, err
	}
	fingerprint, err := catalogRouteFingerprint(request)
	if err != nil {
		return governance.CatalogRouteResult{}, false, err
	}
	resultJSON, replayed, err := s.executeOperation(ctx, "catalog.route", operationID, fingerprint, request.Now, func(tx pgx.Tx) (json.RawMessage, error) {
		result, err := routeCatalogFilesTx(ctx, tx, request)
		if err != nil {
			return nil, err
		}
		encoded, err := json.Marshal(result)
		if err != nil {
			return nil, fmt.Errorf("encode catalog route result: %w", err)
		}
		return encoded, nil
	})
	if err != nil {
		return governance.CatalogRouteResult{}, false, err
	}
	var result governance.CatalogRouteResult
	if err := json.Unmarshal(resultJSON, &result); err != nil {
		return governance.CatalogRouteResult{}, false, fmt.Errorf("decode stored catalog route result: %w", err)
	}
	return result, replayed, nil
}

func catalogRouteFingerprint(request governance.CatalogRouteRequest) (string, error) {
	targets := request.TargetKnowledgeBaseIDs
	if len(targets) == 0 && request.TargetKnowledgeBaseID != "" {
		targets = []string{request.TargetKnowledgeBaseID}
	}
	payload, err := json.Marshal(struct {
		ConnectionID string                `json:"connection_id"`
		NodeIDs      []string              `json:"node_ids"`
		Visibility   governance.Visibility `json:"visibility"`
		Targets      []string              `json:"target_knowledge_base_ids"`
		OperatorRef  string                `json:"operator_ref"`
		Reason       string                `json:"reason"`
	}{
		ConnectionID: request.ConnectionID,
		NodeIDs:      request.NodeIDs,
		Visibility:   request.Visibility,
		Targets:      targets,
		OperatorRef:  strings.TrimSpace(request.OperatorRef),
		Reason:       strings.TrimSpace(request.Reason),
	})
	if err != nil {
		return "", fmt.Errorf("encode catalog route fingerprint: %w", err)
	}
	digest := sha256.Sum256(payload)
	return hex.EncodeToString(digest[:]), nil
}

func (s *Store) EnqueueCatalogRefreshWithOperation(ctx context.Context, source governance.CatalogSource, operationID string, now time.Time) (jobs.Job, bool, error) {
	if !validRouteID(source.ConnectionID) || strings.TrimSpace(source.Name) == "" || now.IsZero() {
		return jobs.Job{}, false, errors.New("invalid catalog refresh request")
	}
	fingerprint, err := operationFingerprint(struct {
		ConnectionID string `json:"connection_id"`
	}{ConnectionID: source.ConnectionID})
	if err != nil {
		return jobs.Job{}, false, err
	}
	resultJSON, replayed, err := s.executeOperation(ctx, "catalog.refresh", operationID, fingerprint, now, func(tx pgx.Tx) (json.RawMessage, error) {
		jobID, err := newOpaqueID("job-")
		if err != nil {
			return nil, err
		}
		payload, err := json.Marshal(map[string]any{"connection_id": source.ConnectionID, "source_type": source.SourceType, "source_name": source.Name, "stage": "queued", "progress": 0, "message": "等待后台刷新", "processed": 0, "total": 0})
		if err != nil {
			return nil, fmt.Errorf("encode catalog refresh job: %w", err)
		}
		job, err := scanJob(tx.QueryRow(ctx, enqueueJobSQL, jobID, "source.catalog.refresh", payload, 20, now, 3, "catalog-refresh:"+jobID, now))
		if err != nil {
			return nil, fmt.Errorf("enqueue catalog refresh job: %w", err)
		}
		encoded, err := json.Marshal(job)
		if err != nil {
			return nil, fmt.Errorf("encode catalog refresh result: %w", err)
		}
		return encoded, nil
	})
	if err != nil {
		return jobs.Job{}, false, err
	}
	var result jobs.Job
	if err := json.Unmarshal(resultJSON, &result); err != nil {
		return jobs.Job{}, false, fmt.Errorf("decode stored catalog refresh result: %w", err)
	}
	return result, replayed, nil
}

func operationFingerprint(value any) (string, error) {
	payload, err := json.Marshal(value)
	if err != nil {
		return "", fmt.Errorf("encode operation fingerprint: %w", err)
	}
	digest := sha256.Sum256(payload)
	return hex.EncodeToString(digest[:]), nil
}

func (s *Store) executeOperation(ctx context.Context, operationType, operationID, fingerprint string, now time.Time, execute operationExecutor) (json.RawMessage, bool, error) {
	if strings.TrimSpace(operationType) == "" || len(operationType) > 100 || strings.TrimSpace(operationID) == "" || len(operationID) > 128 || len(fingerprint) != 64 || now.IsZero() || execute == nil {
		return nil, false, errors.New("invalid operation request")
	}
	beginner, ok := s.db.(interface {
		Begin(context.Context) (pgx.Tx, error)
	})
	if !ok {
		return nil, false, errors.New("operation request transactions are unavailable")
	}
	tx, err := beginner.Begin(ctx)
	if err != nil {
		return nil, false, fmt.Errorf("begin operation request: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()
	tag, err := tx.Exec(ctx, `INSERT INTO operation_requests(operation_type,operation_id,request_fingerprint_sha256,state,result,created_at,completed_at)
VALUES($1,$2,$3,'completed','{}'::jsonb,$4,$4)
ON CONFLICT(operation_type,operation_id) DO NOTHING`, operationType, operationID, fingerprint, now)
	if err != nil {
		return nil, false, fmt.Errorf("create operation request: %w", err)
	}
	if tag.RowsAffected() == 0 {
		var storedFingerprint string
		var storedResult json.RawMessage
		if err := tx.QueryRow(ctx, `SELECT request_fingerprint_sha256,result
FROM operation_requests
WHERE operation_type=$1 AND operation_id=$2
FOR UPDATE`, operationType, operationID).Scan(&storedFingerprint, &storedResult); err != nil {
			return nil, false, fmt.Errorf("load operation request: %w", err)
		}
		if storedFingerprint != fingerprint {
			return nil, false, ErrOperationRequestConflict
		}
		if err := tx.Commit(ctx); err != nil {
			return nil, false, fmt.Errorf("commit replay operation request: %w", err)
		}
		return storedResult, true, nil
	}
	result, err := execute(tx)
	if err != nil {
		return nil, false, err
	}
	if !json.Valid(result) {
		return nil, false, errors.New("operation result must be valid JSON")
	}
	if _, err := tx.Exec(ctx, `UPDATE operation_requests SET result=$3::jsonb,completed_at=$4
WHERE operation_type=$1 AND operation_id=$2`, operationType, operationID, result, now); err != nil {
		return nil, false, fmt.Errorf("complete operation request: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, false, fmt.Errorf("commit operation request: %w", err)
	}
	return result, false, nil
}