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

package postgres

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

	"github.com/jackc/pgx/v5"

	"knowledge-governance/internal/governance"
)

// ConfirmCatalogChangeSet executes ONLY the human-selected items of one planned
// ChangeSet: it creates publications + confirmations + test.upload jobs, or
// disables retrieval and enqueues test.remove for deletes. NO_CHANGE never
// enqueues. Re-confirmation replays the first stored outcome.
func (s *Store) ConfirmCatalogChangeSet(ctx context.Context, request governance.ConfirmCatalogChangeSetRequest) (governance.ConfirmCatalogChangeSetResult, error) {
	operator := strings.TrimSpace(request.OperatorRef)
	reason := strings.TrimSpace(request.Reason)
	if !validRouteID(request.ChangeSetID) || operator == "" || len(operator) > 128 || reason == "" || len(reason) > 500 || request.Now.IsZero() || len(request.ItemSeqs) == 0 || len(request.ItemSeqs) > 200 {
		return governance.ConfirmCatalogChangeSetResult{}, errors.New("invalid change set confirmation")
	}
	seen := make(map[int64]bool, len(request.ItemSeqs))
	for _, seq := range request.ItemSeqs {
		if seq < 1 || seen[seq] {
			return governance.ConfirmCatalogChangeSetResult{}, errors.New("invalid change set item selections")
		}
		seen[seq] = true
	}
	var tx pgx.Tx
	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.ConfirmCatalogChangeSetResult{}, errors.New("change set confirmation transactions are unavailable")
		}
		var err error
		tx, err = beginner.Begin(ctx)
		if err != nil {
			return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("begin change set confirmation: %w", err)
		}
		ownsTransaction = true
		defer tx.Rollback(ctx)
	}
	result, err := confirmCatalogChangeSetTx(ctx, tx, request)
	if err != nil {
		return governance.ConfirmCatalogChangeSetResult{}, err
	}
	if ownsTransaction {
		if err := tx.Commit(ctx); err != nil {
			return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("commit change set confirmation: %w", err)
		}
	}
	return result, nil
}

func confirmCatalogChangeSetTx(ctx context.Context, tx pgx.Tx, request governance.ConfirmCatalogChangeSetRequest) (governance.ConfirmCatalogChangeSetResult, error) {
	result := governance.ConfirmCatalogChangeSetResult{ChangeSetID: request.ChangeSetID}
	selections := append([]int64(nil), request.ItemSeqs...)
	var state string
	err := tx.QueryRow(ctx, `SELECT state FROM source_catalog_change_sets WHERE id=$1 FOR UPDATE`, request.ChangeSetID).Scan(&state)
	if errors.Is(err, pgx.ErrNoRows) {
		return governance.ConfirmCatalogChangeSetResult{}, ErrCatalogChangeSetNotFound
	}
	if err != nil {
		return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("lock change set for confirmation: %w", err)
	}
	if state != "planned" {
		// Even on replay the requested locators must belong to this change set.
		var owned int
		if err := tx.QueryRow(ctx, `SELECT COUNT(*)::integer FROM source_catalog_change_items WHERE change_set_id=$1 AND item_seq=ANY($2::bigint[])`, request.ChangeSetID, selections).Scan(&owned); err != nil {
			return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("replay change set item check: %w", err)
		}
		if owned != len(selections) {
			return governance.ConfirmCatalogChangeSetResult{}, errors.New("change set confirmation references unknown items")
		}
		// Replay the first stored outcome from persisted evidence.
		if err := tx.QueryRow(ctx, `SELECT COUNT(*)::integer FROM processing_confirmations WHERE change_set_id=$1`, request.ChangeSetID).Scan(&result.ConfirmedItems); err != nil {
			return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("replay change set confirmation: %w", err)
		}
		if err := tx.QueryRow(ctx, `SELECT COUNT(*)::integer FROM jobs WHERE type='test.upload' AND payload->>'change_set_id'=$1`, request.ChangeSetID).Scan(&result.UploadsCreated); err != nil {
			return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("replay change set uploads: %w", err)
		}
		if err := tx.QueryRow(ctx, `SELECT COUNT(*)::integer FROM jobs WHERE type='test.remove' AND payload->>'change_set_id'=$1`, request.ChangeSetID).Scan(&result.RemovalsRequested); err != nil {
			return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("replay change set removals: %w", err)
		}
		return result, nil
	}

	rows, err := tx.Query(ctx, `
SELECT item.item_seq,item.action,COALESCE(item.new_scope,''),COALESCE(item.old_scope,''),item.title,
       COALESCE(item.new_source_item_id,item.old_source_item_id) AS source_item_id,
       version.id AS version_id,
       version.content_state='verified' AND content.content_state='verified' AND content.local_spool_ref<>'' AS artifact_ready
FROM source_catalog_change_items item
LEFT JOIN source_items source ON source.id=COALESCE(item.new_source_item_id,item.old_source_item_id)
LEFT JOIN source_item_versions version ON version.source_item_id=source.id AND version.revision=source.revision
LEFT JOIN content_objects content ON content.id=version.content_object_id
WHERE item.change_set_id=$1 AND item.item_seq=ANY($2::bigint[])
ORDER BY item.item_seq`, request.ChangeSetID, selections)
	if err != nil {
		return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("select change set items: %w", err)
	}
	type changeRow struct {
		seq           int64
		action        string
		newScope      string
		oldScope      string
		sourceItemID  *string
		versionID     *string
		artifactReady bool
	}
	changes := make([]changeRow, 0, len(selections))
	for rows.Next() {
		var item changeRow
		var title string
		var versionID *string
		var artifactReady *bool
		if err := rows.Scan(&item.seq, &item.action, &item.newScope, &item.oldScope, &title, &item.sourceItemID, &versionID, &artifactReady); err != nil {
			rows.Close()
			return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("scan change set item: %w", err)
		}
		if versionID != nil {
			item.versionID = versionID
		}
		if artifactReady != nil {
			item.artifactReady = *artifactReady
		}
		changes = append(changes, item)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("iterate change set items: %w", err)
	}
	rows.Close()
	if len(changes) != len(selections) {
		return governance.ConfirmCatalogChangeSetResult{}, errors.New("change set confirmation references unknown items")
	}

	for _, item := range changes {
		if item.action == "no_change" {
			result.NoChangeIgnored++
			continue
		}
		if item.action == "delete" || item.newScope == string(governance.VisibilityBlocked) {
			if item.sourceItemID == nil {
				result.SkippedNoArtifact++
				continue
			}
			removals, err := retireItemPublicationsForChange(ctx, tx, *item.sourceItemID, request)
			if err != nil {
				return governance.ConfirmCatalogChangeSetResult{}, err
			}
			result.RemovalsRequested += removals
			result.ConfirmedItems++
			continue
		}
		targets := make([]string, 0, 2)
		switch item.newScope {
		case string(governance.VisibilityInternal):
			if request.InternalTargetKBID == "" {
				result.SkippedNoTarget++
				continue
			}
			targets = append(targets, request.InternalTargetKBID)
		case string(governance.VisibilityExternal):
			if request.ExternalTargetKBID == "" {
				result.SkippedNoTarget++
				continue
			}
			targets = append(targets, request.ExternalTargetKBID)
		case string(governance.VisibilityBoth):
			if request.InternalTargetKBID != "" {
				targets = append(targets, request.InternalTargetKBID)
			}
			if request.ExternalTargetKBID != "" {
				targets = append(targets, request.ExternalTargetKBID)
			}
			if len(targets) < 2 {
				result.SkippedNoTarget++
				continue
			}
		default:
			result.SkippedNoTarget++
			continue
		}
		if item.versionID == nil || !item.artifactReady {
			result.SkippedNoArtifact++
			continue
		}
		result.ConfirmedItems++
		for _, target := range targets {
			scope := item.newScope
			if scope == string(governance.VisibilityBoth) {
				scope = string(governance.VisibilityInternal)
				if target == request.ExternalTargetKBID {
					scope = string(governance.VisibilityExternal)
				}
			}
			var existingPublication string
			err := tx.QueryRow(ctx, `SELECT id FROM test_publications WHERE source_item_version_id=$1 AND test_kb_id=$2`, *item.versionID, target).Scan(&existingPublication)
			if err == nil {
				if err := ensureChangeSetConfirmation(ctx, tx, existingPublication, request); err != nil {
					return governance.ConfirmCatalogChangeSetResult{}, err
				}
				result.UploadsExisting++
				continue
			}
			if !errors.Is(err, pgx.ErrNoRows) {
				return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("inspect change set publication: %w", err)
			}
			publicationID, err := newOpaqueID("tp-")
			if err != nil {
				return governance.ConfirmCatalogChangeSetResult{}, err
			}
			tag, err := tx.Exec(ctx, `INSERT INTO test_publications(id,source_item_version_id,test_kb_id,target_scope,change_set_id,created_at,updated_at) VALUES($1,$2,$3,$4,$5,$6,$6)
ON CONFLICT (source_item_version_id,test_kb_id) DO NOTHING`,
				publicationID, *item.versionID, target, scope, request.ChangeSetID, request.Now)
			if err != nil {
				return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("create change set publication: %w", err)
			}
			if tag.RowsAffected() == 0 {
				// A concurrent confirmation already owns this version+target.
				var existing string
				if err := tx.QueryRow(ctx, `SELECT id FROM test_publications WHERE source_item_version_id=$1 AND test_kb_id=$2`, *item.versionID, target).Scan(&existing); err != nil {
					return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("resolve concurrent publication: %w", err)
				}
				if err := ensureChangeSetConfirmation(ctx, tx, existing, request); err != nil {
					return governance.ConfirmCatalogChangeSetResult{}, err
				}
				result.UploadsExisting++
				continue
			}
			jobID, err := newOpaqueID("job-")
			if err != nil {
				return governance.ConfirmCatalogChangeSetResult{}, err
			}
			payload, err := json.Marshal(map[string]string{"publication_id": publicationID, "change_set_id": request.ChangeSetID, "item_seq": strconv.FormatInt(item.seq, 10), "target_scope": scope, "target_kb_id": target})
			if err != nil {
				return governance.ConfirmCatalogChangeSetResult{}, errors.New("encode change set upload job")
			}
			if _, err := tx.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,priority,available_at,max_attempts,idempotency_key,created_at,updated_at,business_change_set_id,business_version_id,business_target_kb_id,business_action)
VALUES($1,'test.upload',$2::jsonb,'pending',10,$3,5,$4,$3,$3,$5,$6,$7,'upload')
ON CONFLICT DO NOTHING`, jobID, payload, request.Now, "test-start:"+publicationID, request.ChangeSetID, *item.versionID, target); err != nil {
				return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("enqueue change set upload: %w", err)
			}
			if err := ensureChangeSetConfirmation(ctx, tx, publicationID, request); err != nil {
				return governance.ConfirmCatalogChangeSetResult{}, err
			}
			result.UploadsCreated++
		}
	}
	if _, err := tx.Exec(ctx, `UPDATE source_catalog_change_sets SET state='confirmed',confirmed_at=$2 WHERE id=$1 AND state='planned'`, request.ChangeSetID, request.Now); err != nil {
		return governance.ConfirmCatalogChangeSetResult{}, fmt.Errorf("confirm change set: %w", err)
	}
	return result, nil
}

func ensureChangeSetConfirmation(ctx context.Context, tx pgx.Tx, publicationID string, request governance.ConfirmCatalogChangeSetRequest) error {
	confirmationID, err := newOpaqueID("pc-")
	if err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `INSERT INTO processing_confirmations(id,publication_id,operator_ref,reason,confirmed_at,change_set_id) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT (publication_id) DO NOTHING`,
		confirmationID, publicationID, request.OperatorRef, request.Reason, request.Now, request.ChangeSetID); err != nil {
		return fmt.Errorf("record change set confirmation: %w", err)
	}
	return nil
}

func retireItemPublicationsForChange(ctx context.Context, tx pgx.Tx, itemID string, request governance.ConfirmCatalogChangeSetRequest) (int, error) {
	rows, err := tx.Query(ctx, `SELECT publication.id,publication.knowledge_id,publication.retrieval_state
FROM test_publications publication
JOIN source_item_versions version ON version.id=publication.source_item_version_id
WHERE version.source_item_id=$1 AND publication.retrieval_state NOT IN ('disabling','disabled','unpublished')
FOR UPDATE OF publication`, itemID)
	if err != nil {
		return 0, fmt.Errorf("inspect change set removals: %w", err)
	}
	type target struct{ id, knowledgeID string }
	targets := make([]target, 0)
	for rows.Next() {
		var item target
		if err := rows.Scan(&item.id, &item.knowledgeID, new(string)); err != nil {
			rows.Close()
			return 0, fmt.Errorf("scan change set removal: %w", err)
		}
		targets = append(targets, item)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return 0, fmt.Errorf("iterate change set removals: %w", err)
	}
	rows.Close()
	removals := 0
	for _, item := range targets {
		if item.knowledgeID == "" {
			if _, err := tx.Exec(ctx, `UPDATE test_publications SET ingest_state='failed',processing_state='failed',retrieval_state='disabled',last_error='blocked_by_changeset',updated_at=$2 WHERE id=$1 AND knowledge_id=''`, item.id, request.Now); err != nil {
				return 0, fmt.Errorf("stop pre-upload change set removal: %w", err)
			}
			continue
		}
		if _, err := tx.Exec(ctx, `UPDATE test_publications SET retrieval_state='disabling',last_error='',updated_at=$2 WHERE id=$1`, item.id, request.Now); err != nil {
			return 0, fmt.Errorf("mark change set removal: %w", err)
		}
		jobID, err := newOpaqueID("job-")
		if err != nil {
			return 0, err
		}
		payload, err := json.Marshal(map[string]string{"publication_id": item.id, "change_set_id": request.ChangeSetID})
		if err != nil {
			return 0, errors.New("encode change set remove job")
		}
		// Removals key on the item (version-independent): a confirmed delete must
		// not enqueue twice even if version rows rotate between deliveries.
		if _, err := tx.Exec(ctx, `INSERT INTO jobs(id,type,payload,state,priority,available_at,max_attempts,idempotency_key,created_at,updated_at,business_change_set_id,business_version_id,business_target_kb_id,business_action)
VALUES($1,'test.remove',$2,'pending',20,$3,10,$4,$3,$3,$5,$6,'','remove')
ON CONFLICT DO NOTHING`, jobID, payload, request.Now, "changeset-remove:"+item.id+":"+request.ChangeSetID, request.ChangeSetID, item.id); err != nil {
			return 0, fmt.Errorf("enqueue change set removal: %w", err)
		}
		removals++
	}
	return removals, nil
}

func (s *Store) ConfirmCatalogChangeSetWithOperation(ctx context.Context, request governance.ConfirmCatalogChangeSetRequest, operationID string) (governance.ConfirmCatalogChangeSetResult, bool, error) {
	fingerprint, err := operationFingerprint(struct {
		ChangeSetID string  `json:"change_set_id"`
		ItemSeqs    []int64 `json:"item_seqs"`
		OperatorRef string  `json:"operator_ref"`
		Reason      string  `json:"reason"`
	}{request.ChangeSetID, request.ItemSeqs, request.OperatorRef, request.Reason})
	if err != nil {
		return governance.ConfirmCatalogChangeSetResult{}, false, err
	}
	raw, replayed, err := s.executeOperation(ctx, "changeset.confirm", operationID, fingerprint, request.Now, func(tx pgx.Tx) (json.RawMessage, error) {
		result, err := NewStore(tx).ConfirmCatalogChangeSet(ctx, request)
		if err != nil {
			return nil, err
		}
		return json.Marshal(result)
	})
	if err != nil {
		return governance.ConfirmCatalogChangeSetResult{}, false, err
	}
	var result governance.ConfirmCatalogChangeSetResult
	if err := json.Unmarshal(raw, &result); err != nil {
		return governance.ConfirmCatalogChangeSetResult{}, false, fmt.Errorf("decode stored change set result: %w", err)
	}
	return result, replayed, nil
}