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

package postgres

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

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"

	"knowledge-governance/internal/governance"
)

// Errors for R1 release ChangeSets.
var (
	ErrInvalidReleaseChangeSet       = errors.New("invalid release change set request")
	ErrReleaseChangeSetNotFound      = errors.New("release change set not found")
	ErrReleaseChangeSetStateConflict = errors.New("release change set state conflict")
	ErrReleaseChangeSetEmpty         = errors.New("release plan has no releasable items")
)

const releaseChangeSetColumns = `id,number,name,state,source_snapshot_ids,source_change_set_ids,acceptance_cutoff,production_baseline_ref,item_count,add_count,update_count,delete_count,blocked_count,manifest_sha256,created_by,created_at,confirmed_by,COALESCE(confirmed_at::text,'')`

func scanReleaseChangeSet(row pgx.Row) (governance.ReleaseChangeSet, error) {
	var item governance.ReleaseChangeSet
	var snapshots, changeSets []byte
	var confirmedAt string
	err := row.Scan(&item.ID, &item.Number, &item.Name, &item.State, &snapshots, &changeSets, &item.AcceptanceCutoff, &item.ProductionBaselineRef,
		&item.ItemCount, &item.AddCount, &item.UpdateCount, &item.DeleteCount, &item.BlockedCount, &item.ManifestSHA256, &item.CreatedBy, &item.CreatedAt, &item.ConfirmedBy, &confirmedAt)
	if err != nil {
		return governance.ReleaseChangeSet{}, err
	}
	_ = json.Unmarshal(snapshots, &item.SourceSnapshotIDs)
	_ = json.Unmarshal(changeSets, &item.SourceChangeSetIDs)
	if item.SourceSnapshotIDs == nil {
		item.SourceSnapshotIDs = []string{}
	}
	if item.SourceChangeSetIDs == nil {
		item.SourceChangeSetIDs = []string{}
	}
	if confirmedAt != "" {
		if parsed, parseErr := time.Parse(time.RFC3339Nano, confirmedAt); parseErr == nil {
			item.ConfirmedAt = &parsed
		}
	}
	return item, nil
}

// ComputeReleaseChangeSetManifest is exported for tests: the manifest seals
// every frozen decision in seq order with an exact delimiter.
func ComputeReleaseChangeSetManifest(items []governance.ReleaseChangeSetItem) string {
	lines := make([]string, 0, len(items))
	for _, item := range items {
		lines = append(lines, strings.Join([]string{
			strconv.Itoa(item.Seq), item.PublicationID, item.TargetScope, item.Action,
			item.TestState, item.ProductionState, strconv.FormatBool(item.CanRelease), item.Blocker,
		}, "|"))
	}
	digest := sha256.Sum256([]byte(strings.Join(lines, "\n")))
	return hex.EncodeToString(digest[:])
}

// GenerateReleaseChangeSet freezes the current live plan into an immutable
// versioned set. Repeated generation with an identical plan reuses the latest
// PLANNED set (dedup), so concurrent generation cannot create duplicates.
func (s *Store) GenerateReleaseChangeSet(ctx context.Context, request governance.GenerateReleaseChangeSetRequest) (governance.ReleaseChangeSet, error) {
	if request.Name == "" || request.Name != strings.TrimSpace(request.Name) || len(request.Name) > 200 ||
		strings.ContainsAny(request.Name, "\x00\r\n") || !validReviewerRef(request.CreatedBy) || request.Now.IsZero() {
		return governance.ReleaseChangeSet{}, ErrInvalidReleaseChangeSet
	}
	plan, err := s.ListReleasePlan(ctx)
	if err != nil {
		return governance.ReleaseChangeSet{}, err
	}
	items := make([]governance.ReleaseChangeSetItem, 0, len(plan.Items))
	for index, planItem := range plan.Items {
		items = append(items, governance.ReleaseChangeSetItem{Seq: index + 1, PublicationID: planItem.PublicationID, Title: planItem.Title, TargetScope: planItem.TargetScope, Action: planItem.Action, TestState: planItem.TestState, ProductionState: planItem.ProductionState, CanRelease: planItem.CanRelease, Blocker: planItem.Blocker})
	}
	if len(items) == 0 {
		return governance.ReleaseChangeSet{}, ErrReleaseChangeSetEmpty
	}
	manifest := ComputeReleaseChangeSetManifest(items)
	counts := map[string]int{}
	for _, item := range items {
		counts[item.Action]++
	}
	blocked := 0
	for _, item := range items {
		if item.Action == "blocked" {
			blocked++
		}
	}

	// Same plan twice (sequentially) reuses the latest PLANNED set instead of
	// piling up identical versions. Concurrent racers fall through to the
	// unique-number conflict path below.
	if existing, lookupErr := s.findPlannedByManifest(ctx, manifest); lookupErr == nil && existing.ManifestSHA256 == manifest {
		return existing, nil
	} else if lookupErr != nil && !errors.Is(lookupErr, pgx.ErrNoRows) {
		return governance.ReleaseChangeSet{}, lookupErr
	}

	for attempt := 0; attempt < 3; attempt++ {
		result, retry, err := s.tryInsertReleaseChangeSet(ctx, request, items, manifest, counts["add"], counts["update"], counts["delete"], blocked)
		if err == nil {
			return result, nil
		}
		if !retry {
			return governance.ReleaseChangeSet{}, err
		}
		// A concurrent identical generation won; return its set (dedup) if the
		// manifest matches, otherwise loop to take the next version number.
		existing, lookupErr := s.findPlannedByManifest(ctx, manifest)
		if lookupErr == nil && existing.ManifestSHA256 == manifest {
			return existing, nil
		}
		if lookupErr != nil && !errors.Is(lookupErr, ErrReleaseChangeSetNotFound) {
			return governance.ReleaseChangeSet{}, lookupErr
		}
	}
	return governance.ReleaseChangeSet{}, fmt.Errorf("release change set generation kept conflicting")
}

func (s *Store) tryInsertReleaseChangeSet(ctx context.Context, request governance.GenerateReleaseChangeSetRequest, items []governance.ReleaseChangeSetItem, manifest string, add, update, del, blocked int) (governance.ReleaseChangeSet, bool, error) {
	tx, err := s.db.Begin(ctx)
	if err != nil {
		return governance.ReleaseChangeSet{}, false, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	id, err := newOpaqueID("rcs-")
	if err != nil {
		return governance.ReleaseChangeSet{}, false, err
	}
	set, err := scanReleaseChangeSet(tx.QueryRow(ctx, `
INSERT INTO release_change_sets(id,name,number,state,source_snapshot_ids,source_change_set_ids,acceptance_cutoff,production_baseline_ref,item_count,add_count,update_count,delete_count,blocked_count,manifest_sha256,created_by,created_at)
SELECT $1,$2,(SELECT COALESCE(MAX(number),0)+1 FROM release_change_sets),'planned',
       COALESCE((SELECT jsonb_agg(snapshot_id ORDER BY snapshot_id) FROM (SELECT DISTINCT snapshot_id FROM source_catalog_heads) s),'[]'::jsonb),
       COALESCE((SELECT jsonb_agg(change_set_id ORDER BY change_set_id) FROM (SELECT DISTINCT id AS change_set_id FROM source_catalog_change_sets WHERE state IN ('confirmed','applied')) c),'[]'::jsonb),
       COALESCE((SELECT MAX(updated_at) FROM test_publications),(SELECT now())),
       (SELECT COALESCE(MAX(completed_at)::text,'') FROM release_executions WHERE state='completed'),
       $3,$4,$5,$6,$7,$8,$9,$10
RETURNING `+releaseChangeSetColumns,
		id, request.Name, len(items), add, update, del, blocked, manifest, request.CreatedBy, request.Now))
	if err != nil {
		var pgErr *pgconn.PgError
		if errors.As(err, &pgErr) && pgErr.Code == "23505" {
			return governance.ReleaseChangeSet{}, true, err
		}
		return governance.ReleaseChangeSet{}, false, fmt.Errorf("create release change set: %w", err)
	}
	// R6 audit chain: mirror the JSONB binding into real FK rows so the trace
	// survives even if someone later edits denormalized display fields.
	if _, err := tx.Exec(ctx, `
INSERT INTO release_change_set_snapshot_links(change_set_id, snapshot_id)
SELECT $1, elem.value
FROM jsonb_array_elements_text($2::jsonb) AS elem(value)
ON CONFLICT DO NOTHING`, set.ID, mustMarshalJSON(set.SourceSnapshotIDs)); err != nil {
		return governance.ReleaseChangeSet{}, false, fmt.Errorf("link release snapshots: %w", err)
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO release_change_set_catalog_links(change_set_id, catalog_change_set_id)
SELECT $1, elem.value
FROM jsonb_array_elements_text($2::jsonb) AS elem(value)
ON CONFLICT DO NOTHING`, set.ID, mustMarshalJSON(set.SourceChangeSetIDs)); err != nil {
		return governance.ReleaseChangeSet{}, false, fmt.Errorf("link release catalog sets: %w", err)
	}
	for _, item := range items {
		if _, err := tx.Exec(ctx, `INSERT INTO release_change_set_items(change_set_id,seq,publication_id,title,target_scope,action,test_state,production_state,can_release,blocker) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`,
			set.ID, item.Seq, item.PublicationID, item.Title, item.TargetScope, item.Action, item.TestState, item.ProductionState, item.CanRelease, item.Blocker); err != nil {
			return governance.ReleaseChangeSet{}, false, fmt.Errorf("insert release change set item: %w", err)
		}
	}
	if err := tx.Commit(ctx); err != nil {
		var pgErr *pgconn.PgError
		if errors.As(err, &pgErr) && pgErr.Code == "23505" {
			return governance.ReleaseChangeSet{}, true, err
		}
		return governance.ReleaseChangeSet{}, false, err
	}
	return set, false, nil
}

func (s *Store) findPlannedByManifest(ctx context.Context, manifest string) (governance.ReleaseChangeSet, error) {
	return scanReleaseChangeSet(s.db.QueryRow(ctx, `SELECT `+releaseChangeSetColumns+` FROM release_change_sets WHERE manifest_sha256=$1 AND state='planned' ORDER BY number DESC LIMIT 1`, manifest))
}

// ConfirmReleaseChangeSet freezes a planned set for batching. Re-confirming an
// already-confirmed set replays the stored confirmation (immutability).
func (s *Store) ConfirmReleaseChangeSet(ctx context.Context, request governance.ConfirmReleaseChangeSetRequest) (governance.ReleaseChangeSet, error) {
	if !validRouteID(request.ChangeSetID) || !validReviewerRef(request.ConfirmedBy) || request.Now.IsZero() {
		return governance.ReleaseChangeSet{}, ErrInvalidReleaseChangeSet
	}
	set, err := scanReleaseChangeSet(s.db.QueryRow(ctx, `UPDATE release_change_sets SET state='confirmed',confirmed_by=$2,confirmed_at=$3 WHERE id=$1 AND state='planned' RETURNING `+releaseChangeSetColumns, request.ChangeSetID, request.ConfirmedBy, request.Now))
	if errors.Is(err, pgx.ErrNoRows) {
		current, lookupErr := s.GetReleaseChangeSet(ctx, request.ChangeSetID)
		if lookupErr != nil {
			return governance.ReleaseChangeSet{}, lookupErr
		}
		if current.State == "confirmed" || current.State == "consumed" {
			return current.ReleaseChangeSet, nil
		}
		return governance.ReleaseChangeSet{}, ErrReleaseChangeSetStateConflict
	}
	if err != nil {
		return governance.ReleaseChangeSet{}, fmt.Errorf("confirm release change set: %w", err)
	}
	return set, nil
}

func (s *Store) GetReleaseChangeSet(ctx context.Context, id string) (governance.ReleaseChangeSetDetail, error) {
	set, err := scanReleaseChangeSet(s.db.QueryRow(ctx, `SELECT `+releaseChangeSetColumns+` FROM release_change_sets WHERE id=$1`, id))
	if errors.Is(err, pgx.ErrNoRows) {
		return governance.ReleaseChangeSetDetail{}, ErrReleaseChangeSetNotFound
	}
	if err != nil {
		return governance.ReleaseChangeSetDetail{}, err
	}
	detail := governance.ReleaseChangeSetDetail{ReleaseChangeSet: set, Items: []governance.ReleaseChangeSetItem{}}
	rows, err := s.db.Query(ctx, `SELECT seq,publication_id,title,target_scope,action,test_state,production_state,can_release,blocker FROM release_change_set_items WHERE change_set_id=$1 ORDER BY seq`, id)
	if err != nil {
		return governance.ReleaseChangeSetDetail{}, fmt.Errorf("list release change set items: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var item governance.ReleaseChangeSetItem
		if err := rows.Scan(&item.Seq, &item.PublicationID, &item.Title, &item.TargetScope, &item.Action, &item.TestState, &item.ProductionState, &item.CanRelease, &item.Blocker); err != nil {
			return governance.ReleaseChangeSetDetail{}, err
		}
		detail.Items = append(detail.Items, item)
	}
	return detail, rows.Err()
}

func (s *Store) ListReleaseChangeSets(ctx context.Context, filter governance.ReleaseBatchFilter) (governance.ReleaseChangeSetPage, error) {
	limit, offset := filter.Limit, filter.Offset
	if limit <= 0 {
		limit = 20
	}
	if offset < 0 {
		offset = 0
	}
	page := governance.ReleaseChangeSetPage{Items: []governance.ReleaseChangeSet{}, Limit: limit, Offset: offset}
	if limit > 100 {
		return page, errors.New("release change set limit must be between 1 and 100")
	}
	var total int
	if err := s.db.QueryRow(ctx, `SELECT COUNT(*) FROM release_change_sets`).Scan(&total); err != nil {
		return page, err
	}
	page.Total = total
	rows, err := s.db.Query(ctx, `SELECT `+releaseChangeSetColumns+` FROM release_change_sets ORDER BY number DESC LIMIT $1 OFFSET $2`, limit, offset)
	if err != nil {
		return page, err
	}
	defer rows.Close()
	for rows.Next() {
		item, err := scanReleaseChangeSet(rows)
		if err != nil {
			return page, err
		}
		page.Items = append(page.Items, item)
	}
	return page, rows.Err()
}

func (s *Store) LatestReleaseChangeSet(ctx context.Context) (governance.ReleaseChangeSetDetail, error) {
	var id string
	if err := s.db.QueryRow(ctx, `SELECT id FROM release_change_sets ORDER BY number DESC LIMIT 1`).Scan(&id); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return governance.ReleaseChangeSetDetail{}, ErrReleaseChangeSetNotFound
		}
		return governance.ReleaseChangeSetDetail{}, err
	}
	return s.GetReleaseChangeSet(ctx, id)
}

func mustMarshalJSON(values []string) []byte {
	raw, err := json.Marshal(values)
	if err != nil {
		return []byte("[]")
	}
	return raw
}