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

package postgres

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

	"github.com/jackc/pgx/v5"

	"knowledge-governance/internal/governance"
)

func (s *Store) StartTestPublications(ctx context.Context, request governance.StartTestPublicationsRequest) (governance.StartTestPublicationsResult, error) {
	result := governance.StartTestPublicationsResult{Selected: len(request.PublicationIDs)}
	operator := strings.TrimSpace(request.OperatorRef)
	reason := strings.TrimSpace(request.Reason)
	if len(request.PublicationIDs) == 0 || len(request.PublicationIDs) > 100 || operator == "" || len(operator) > 128 || reason == "" || len(reason) > 500 || request.Now.IsZero() {
		return governance.StartTestPublicationsResult{}, errors.New("invalid processing confirmation")
	}
	seen := make(map[string]bool, len(request.PublicationIDs))
	for _, publicationID := range request.PublicationIDs {
		if !validRouteID(publicationID) || seen[publicationID] {
			return governance.StartTestPublicationsResult{}, errors.New("invalid processing publication IDs")
		}
		seen[publicationID] = true
	}

	var tx pgx.Tx
	var err error
	ownsTransaction := false
	if existing, ok := s.db.(pgx.Tx); ok {
		tx = existing
	} else {
		beginner, ok := s.db.(interface {
			Begin(context.Context) (pgx.Tx, error)
		})
		if !ok {
			return governance.StartTestPublicationsResult{}, errors.New("processing confirmation transactions are unavailable")
		}
		var err error
		tx, err = beginner.Begin(ctx)
		if err != nil {
			return governance.StartTestPublicationsResult{}, fmt.Errorf("begin processing confirmation: %w", err)
		}
		ownsTransaction = true
		defer tx.Rollback(ctx)
	}

	for _, publicationID := range request.PublicationIDs {
		var ready, confirmed, uploadJobExists bool
		err = tx.QueryRow(ctx, `
SELECT p.ingest_state='pending'
       AND p.processing_state='pending'
       AND p.retrieval_state='unpublished'
       AND p.knowledge_id=''
       AND version.content_state='verified'
       AND content.content_state='verified'
       AND content.local_spool_ref<>''
       AND content.size_bytes>0 AS ready,
       EXISTS(SELECT 1 FROM processing_confirmations confirmation WHERE confirmation.publication_id=p.id),
       EXISTS(SELECT 1 FROM jobs job WHERE job.type='test.upload' AND job.payload->>'publication_id'=p.id)
FROM test_publications p
JOIN source_item_versions version ON version.id=p.source_item_version_id
JOIN content_objects content ON content.id=version.content_object_id
WHERE p.id=$1
FOR UPDATE OF p`, publicationID).Scan(&ready, &confirmed, &uploadJobExists)
		if errors.Is(err, pgx.ErrNoRows) {
			result.Unavailable++
			continue
		}
		if err != nil {
			return governance.StartTestPublicationsResult{}, fmt.Errorf("inspect processing plan: %w", err)
		}
		if confirmed || uploadJobExists {
			result.AlreadyStarted++
			continue
		}
		if !ready {
			result.Unavailable++
			continue
		}
		confirmationID, err := newOpaqueID("pc-")
		if err != nil {
			return governance.StartTestPublicationsResult{}, err
		}
		jobID, err := newOpaqueID("job-")
		if err != nil {
			return governance.StartTestPublicationsResult{}, err
		}
		payload, err := json.Marshal(map[string]string{"publication_id": publicationID})
		if err != nil {
			return governance.StartTestPublicationsResult{}, errors.New("encode processing upload job")
		}
		if _, err := tx.Exec(ctx, `INSERT INTO processing_confirmations(id,publication_id,operator_ref,reason,confirmed_at,created_at) VALUES($1,$2,$3,$4,$5,$5)`, confirmationID, publicationID, operator, reason, request.Now); err != nil {
			return governance.StartTestPublicationsResult{}, fmt.Errorf("record processing confirmation: %w", err)
		}
		if _, err := tx.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,priority,available_at,max_attempts,idempotency_key,created_at,updated_at)
VALUES($1,'test.upload',$2::jsonb,'pending',10,$3,5,$4,$3,$3)`, jobID, payload, request.Now, "test-start:"+publicationID); err != nil {
			return governance.StartTestPublicationsResult{}, fmt.Errorf("enqueue confirmed processing: %w", err)
		}
		result.Started++
	}
	if ownsTransaction {
		if err := tx.Commit(ctx); err != nil {
			return governance.StartTestPublicationsResult{}, fmt.Errorf("commit processing confirmation: %w", err)
		}
	}
	return result, nil
}

func (s *Store) StartTestPublicationsWithOperation(ctx context.Context, request governance.StartTestPublicationsRequest, operationID string) (governance.StartTestPublicationsResult, bool, error) {
	fingerprint, err := operationFingerprint(struct {
		PublicationIDs []string `json:"publication_ids"`
		OperatorRef    string   `json:"operator_ref"`
		Reason         string   `json:"reason"`
	}{request.PublicationIDs, request.OperatorRef, request.Reason})
	if err != nil {
		return governance.StartTestPublicationsResult{}, false, err
	}
	raw, replayed, err := s.executeOperation(ctx, "test.publications.start", operationID, fingerprint, request.Now, func(tx pgx.Tx) (json.RawMessage, error) {
		result, err := NewStore(tx).StartTestPublications(ctx, request)
		if err != nil {
			return nil, err
		}
		return json.Marshal(result)
	})
	if err != nil {
		return governance.StartTestPublicationsResult{}, false, err
	}
	var result governance.StartTestPublicationsResult
	if err := json.Unmarshal(raw, &result); err != nil {
		return governance.StartTestPublicationsResult{}, false, fmt.Errorf("decode stored processing result: %w", err)
	}
	return result, replayed, nil
}