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

package postgres

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

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

	"knowledge-governance/internal/governance"
)

const releaseGatePolicyVersion = "release-gate-v1"

var (
	ErrInvalidReleaseBatch         = errors.New("invalid release batch")
	ErrInvalidReleaseBatchFilter   = errors.New("invalid release batch filter")
	ErrReleaseBatchNotEligible     = errors.New("release batch contains ineligible publications")
	ErrReleaseBatchGateConflict    = errors.New("release batch gate conflict")
	ErrReleaseBatchNotFound        = errors.New("release batch not found")
	ErrReleaseChangeSetUnconfirmed = errors.New("release change set must be confirmed before batching")
)

const releaseGateCTEs = `
acceptance_run_summary AS (
    SELECT publication_id, run_id, MAX(observed_at) AS observed_at,
           COALESCE(MAX(state) FILTER (WHERE check_type = 'knowledge_exists'), '') AS knowledge_exists,
           COALESCE(MAX(state) FILTER (WHERE check_type = 'chunks_present'), '') AS chunks_present,
           COALESCE(MAX(state) FILTER (WHERE check_type = 'retrieval_hit'), '') AS retrieval_hit
    FROM acceptance_results
    GROUP BY publication_id, run_id
), latest_acceptance AS (
    SELECT DISTINCT ON (publication_id)
           publication_id, run_id, knowledge_exists, chunks_present, retrieval_hit, observed_at
    FROM acceptance_run_summary
    ORDER BY publication_id, observed_at DESC, run_id DESC
), latest_review AS (
    SELECT DISTINCT ON (publication_id)
           id, publication_id, decision, decided_visibility, acceptance_run_id,
           publication_updated_at, created_at
    FROM publication_reviews
    ORDER BY publication_id, created_at DESC, id DESC
), release_ready AS (
    SELECT publication.id AS publication_id,
           publication.source_item_version_id,
           publication.updated_at AS publication_updated_at,
           review.id AS review_id,
           review.acceptance_run_id,
           review.decided_visibility
    FROM test_publications AS publication
    JOIN source_item_versions AS version ON version.id = publication.source_item_version_id
    JOIN source_items AS source ON source.id = version.source_item_id
    JOIN latest_acceptance AS acceptance ON acceptance.publication_id = publication.id
    JOIN publication_reviews AS review ON review.id = (
        SELECT latest_review.id FROM latest_review WHERE latest_review.publication_id = publication.id
    )
    WHERE review.decision = 'approved'
      AND review.acceptance_run_id = acceptance.run_id
      AND review.publication_updated_at = publication.updated_at
      AND publication.ingest_state = 'uploaded'
      AND publication.processing_state = 'completed'
      AND publication.retrieval_state = 'unpublished'
      AND version.revision = source.revision
      AND source.source_change NOT IN ('deleted', 'unavailable')
      AND acceptance.knowledge_exists = 'passed'
      AND acceptance.chunks_present = 'passed'
      AND acceptance.retrieval_hit IN ('passed', 'manual_required')
)
`

const releaseDeleteReadyCTEs = `
source_truth AS (
    SELECT source.id, source.source_change,
           COALESCE(item_rule.visibility, directory_rule.visibility, source.visibility) AS desired_policy
    FROM source_items AS source
    LEFT JOIN source_item_scope_overrides AS item_rule ON item_rule.source_item_id = source.id
    LEFT JOIN LATERAL (
        SELECT rule.visibility
        FROM directory_scope_rules AS rule
        WHERE rule.source_connection_id = source.source_connection_id
          AND (rule.directory_path = '' OR source.normalized_path = rule.directory_path OR source.normalized_path LIKE rule.directory_path || '/%')
        ORDER BY LENGTH(rule.directory_path) DESC, rule.directory_path DESC
        LIMIT 1
    ) AS directory_rule ON TRUE
), latest_production AS (
    SELECT DISTINCT ON (version.source_item_id, publication.target_scope)
           publication.id AS publication_id, version.source_item_id, publication.target_scope,
           execution_item.production_revision,
           batch_item.review_id, batch_item.source_item_version_id,
           batch_item.acceptance_run_id, batch_item.publication_updated_at,
           batch_item.decided_visibility
    FROM release_item_executions AS execution_item
    JOIN release_executions AS execution ON execution.id = execution_item.release_execution_id
    JOIN test_publications AS publication ON publication.id = execution_item.publication_id
    JOIN source_item_versions AS version ON version.id = publication.source_item_version_id
    JOIN release_batch_items AS batch_item
      ON batch_item.release_batch_id = execution.release_batch_id
     AND batch_item.publication_id = publication.id
    WHERE execution.state IN ('completed', 'rolled_back')
      AND execution_item.state IN ('verified', 'rolled_back')
    ORDER BY version.source_item_id, publication.target_scope,
             execution.completed_at DESC, execution_item.completed_at DESC, execution_item.id DESC
), release_delete_ready AS (
    SELECT production.publication_id, production.source_item_version_id,
           production.publication_updated_at, production.review_id,
           production.acceptance_run_id, production.decided_visibility,
           'delete' AS action
    FROM latest_production AS production
    JOIN source_truth AS truth ON truth.id = production.source_item_id
    WHERE production.production_revision <> 'absent'
      AND truth.source_change <> 'unavailable'
      AND (
        truth.source_change = 'deleted'
        OR truth.desired_policy = 'blocked'
        OR (production.target_scope = 'internal' AND truth.desired_policy NOT IN ('internal', 'both'))
        OR (production.target_scope = 'external' AND truth.desired_policy NOT IN ('external', 'both'))
      )
), release_upsert_ready AS (
    SELECT ready.*
    FROM release_ready AS ready
    JOIN test_publications AS publication ON publication.id = ready.publication_id
    JOIN source_item_versions AS version ON version.id = publication.source_item_version_id
    JOIN source_truth AS truth ON truth.id = version.source_item_id
    LEFT JOIN latest_production AS production
      ON production.source_item_id = version.source_item_id
     AND production.target_scope = publication.target_scope
    WHERE truth.source_change NOT IN ('deleted', 'unavailable')
      AND truth.desired_policy <> 'blocked'
      AND publication.target_scope = ready.decided_visibility
      AND (
        (publication.target_scope = 'internal' AND truth.desired_policy IN ('internal', 'both'))
        OR (publication.target_scope = 'external' AND truth.desired_policy IN ('external', 'both'))
      )
      AND (production.production_revision IS NULL OR production.production_revision = 'absent' OR production.production_revision <> ready.review_id)
)
`

const createReleaseBatchSQL = `
WITH requested AS (
    SELECT value AS publication_id, 'upsert' AS action
    FROM jsonb_array_elements_text($5::jsonb)
    UNION ALL
    SELECT value AS publication_id, 'delete' AS action
    FROM jsonb_array_elements_text($6::jsonb)
), ` + releaseGateCTEs + `, ` + releaseDeleteReadyCTEs + `,
request_stats AS (
    SELECT COUNT(*)::INTEGER AS requested_count,
           COUNT(DISTINCT publication_id)::INTEGER AS distinct_count
    FROM requested
), binding AS (
    -- R1: when a batch cites a release change set, the set must be CONFIRMED
    -- and every requested publication must exist in its frozen manifest as a
    -- releasable item. Source refreshes can never move this goalpost.
    SELECT 1
    WHERE $7::text = '' OR EXISTS (
        SELECT 1 FROM release_change_sets rcs
        WHERE rcs.id = $7 AND rcs.state = 'confirmed'
          AND NOT EXISTS (
              SELECT 1 FROM requested q
              WHERE NOT EXISTS (
                  SELECT 1 FROM release_change_set_items frozen
                  WHERE frozen.change_set_id = rcs.id
                    AND frozen.publication_id = q.publication_id
                    AND frozen.can_release
              )
          )
    )
), eligible AS (
    SELECT ready.publication_id, ready.source_item_version_id, ready.publication_updated_at,
           ready.review_id, ready.acceptance_run_id, ready.decided_visibility,
           requested.action
    FROM release_upsert_ready AS ready
    JOIN requested ON requested.publication_id = ready.publication_id AND requested.action = 'upsert'
    UNION ALL
    SELECT ready.publication_id, ready.source_item_version_id, ready.publication_updated_at,
           ready.review_id, ready.acceptance_run_id, ready.decided_visibility,
           ready.action
    FROM release_delete_ready AS ready
    JOIN requested ON requested.publication_id = ready.publication_id AND requested.action = 'delete'
), created_batch AS (
    INSERT INTO release_batches (
        id, name, state, policy_version, created_by, item_count, created_at, release_change_set_id
    )
    SELECT $1, $2, 'draft', 'release-gate-v1', $3, request_stats.requested_count, $4, NULLIF($7::text, '')
    FROM request_stats
    WHERE EXISTS (SELECT 1 FROM binding)
      AND request_stats.requested_count BETWEEN 1 AND 100
      AND request_stats.requested_count = request_stats.distinct_count
      AND request_stats.requested_count = (SELECT COUNT(*) FROM eligible)
    RETURNING id, name, state, policy_version, created_by, sealed_by,
              item_count, created_at, sealed_at, release_change_set_id
), inserted_items AS (
    INSERT INTO release_batch_items (
        release_batch_id, publication_id, review_id, source_item_version_id,
        acceptance_run_id, publication_updated_at, decided_visibility, action, created_at
    )
    SELECT created_batch.id, eligible.publication_id, eligible.review_id,
           eligible.source_item_version_id, eligible.acceptance_run_id,
           eligible.publication_updated_at, eligible.decided_visibility, eligible.action, $4
    FROM created_batch CROSS JOIN eligible
    RETURNING release_batch_id
)
SELECT id, name, state, policy_version, created_by, sealed_by,
       item_count, created_at, sealed_at, COALESCE(release_change_set_id::text,'')
FROM created_batch
WHERE item_count = (SELECT COUNT(*) FROM inserted_items)
`

const sealReleaseBatchSQL = `
WITH ` + releaseGateCTEs + `, ` + releaseDeleteReadyCTEs + `,
item_stats AS (
    SELECT batch.id AS batch_id,
           COUNT(item.publication_id)::INTEGER AS total_count,
           COUNT(*) FILTER (WHERE
             (item.action = 'upsert' AND ready.publication_id IS NOT NULL)
             OR (item.action = 'delete' AND delete_ready.publication_id IS NOT NULL)
           )::INTEGER AS valid_count
    FROM release_batches AS batch
    LEFT JOIN release_batch_items AS item ON item.release_batch_id = batch.id
    LEFT JOIN release_upsert_ready AS ready
      ON ready.publication_id = item.publication_id
     AND item.action = 'upsert'
     AND ready.review_id = item.review_id
     AND ready.acceptance_run_id = item.acceptance_run_id
     AND ready.publication_updated_at = item.publication_updated_at
     AND ready.decided_visibility = item.decided_visibility
    LEFT JOIN release_delete_ready AS delete_ready
      ON delete_ready.publication_id = item.publication_id
     AND item.action = 'delete'
     AND delete_ready.review_id = item.review_id
     AND delete_ready.acceptance_run_id = item.acceptance_run_id
     AND delete_ready.publication_updated_at = item.publication_updated_at
     AND delete_ready.decided_visibility = item.decided_visibility
    WHERE batch.id = $1 AND batch.state = 'draft'
    GROUP BY batch.id
), sealed AS (
    UPDATE release_batches AS batch
    SET state = 'sealed', sealed_by = $2, sealed_at = $3
    FROM item_stats
    WHERE batch.id = item_stats.batch_id
      AND batch.policy_version = 'release-gate-v1'
      AND item_stats.total_count > 0
      AND item_stats.total_count = item_stats.valid_count
      AND batch.item_count = item_stats.total_count
    RETURNING batch.id, batch.name, batch.state, batch.policy_version,
              batch.created_by, batch.sealed_by, batch.item_count,
              batch.created_at, batch.sealed_at, COALESCE(batch.release_change_set_id::text,'')
), already_sealed AS (
    SELECT id, name, state, policy_version, created_by, sealed_by,
           item_count, created_at, sealed_at, COALESCE(release_change_set_id::text,'')
    FROM release_batches
    WHERE id = $1 AND state = 'sealed'
)
SELECT * FROM sealed
UNION ALL
SELECT * FROM already_sealed
LIMIT 1
`

const countReleaseBatchesSQL = `
SELECT COUNT(*)::INTEGER
FROM release_batches
WHERE $1 = '' OR state = $1
`

const listReleaseBatchesSQL = `
SELECT id, name, state, policy_version, created_by, sealed_by,
       item_count, created_at, sealed_at, COALESCE(release_change_set_id::text,'')
FROM release_batches
WHERE $1 = '' OR state = $1
ORDER BY created_at DESC, id DESC
LIMIT $2 OFFSET $3
`

const getReleaseBatchSQL = `
SELECT id, name, state, policy_version, created_by, sealed_by,
       item_count, created_at, sealed_at, COALESCE(release_change_set_id::text,'')
FROM release_batches
WHERE id = $1
`

const listReleaseBatchItemsSQL = `
SELECT item.publication_id, source.title, item.decided_visibility, item.action
FROM release_batch_items AS item
JOIN test_publications AS publication ON publication.id = item.publication_id
JOIN source_item_versions AS version ON version.id = publication.source_item_version_id
JOIN source_items AS source ON source.id = version.source_item_id
WHERE item.release_batch_id = $1
ORDER BY item.publication_id
`

func (s *Store) CreateReleaseBatch(ctx context.Context, request governance.CreateReleaseBatchRequest) (governance.ReleaseBatch, error) {
	if !validCreateReleaseBatch(request) {
		return governance.ReleaseBatch{}, ErrInvalidReleaseBatch
	}
	batchID, err := newOpaqueID("rb-")
	if err != nil {
		return governance.ReleaseBatch{}, err
	}
	publicationIDs, err := json.Marshal(append([]string{}, request.PublicationIDs...))
	if err != nil {
		return governance.ReleaseBatch{}, ErrInvalidReleaseBatch
	}
	deletePublicationIDs, err := json.Marshal(append([]string{}, request.DeletePublicationIDs...))
	if err != nil {
		return governance.ReleaseBatch{}, ErrInvalidReleaseBatch
	}
	batch, err := scanReleaseBatch(s.db.QueryRow(ctx, createReleaseBatchSQL,
		batchID, request.Name, request.CreatedBy, request.Now, publicationIDs, deletePublicationIDs,
		strings.TrimSpace(request.ReleaseChangeSetID),
	))
	if errors.Is(err, pgx.ErrNoRows) {
		// R1: distinguish an unconfirmed/non-covering change set from plain
		// gate ineligibility so operators get the honest reason.
		if request.ReleaseChangeSetID != "" {
			var state string
			if lookup := s.db.QueryRow(ctx, `SELECT state FROM release_change_sets WHERE id=$1`, request.ReleaseChangeSetID).Scan(&state); lookup != nil || state != "confirmed" {
				return governance.ReleaseBatch{}, ErrReleaseChangeSetUnconfirmed
			}
		}
		return governance.ReleaseBatch{}, ErrReleaseBatchNotEligible
	}
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
		return governance.ReleaseBatch{}, ErrReleaseBatchNotEligible
	}
	if err != nil {
		return governance.ReleaseBatch{}, fmt.Errorf("create release batch: %w", err)
	}
	return batch, nil
}

func (s *Store) SealReleaseBatch(ctx context.Context, request governance.SealReleaseBatchRequest) (governance.ReleaseBatch, error) {
	if !validShortID(request.BatchID) || !validReviewerRef(request.SealedBy) || request.Now.IsZero() {
		return governance.ReleaseBatch{}, ErrInvalidReleaseBatch
	}
	batch, err := scanReleaseBatch(s.db.QueryRow(ctx, sealReleaseBatchSQL, request.BatchID, request.SealedBy, request.Now))
	if errors.Is(err, pgx.ErrNoRows) {
		return governance.ReleaseBatch{}, ErrReleaseBatchGateConflict
	}
	if err != nil {
		return governance.ReleaseBatch{}, fmt.Errorf("seal release batch: %w", err)
	}
	return batch, nil
}

func (s *Store) ListReleaseBatches(ctx context.Context, filter governance.ReleaseBatchFilter) (governance.ReleaseBatchPage, error) {
	if filter.Limit < 1 || filter.Limit > 100 || filter.Offset < 0 || filter.Offset > 1000000 ||
		filter.State != "" && filter.State != string(governance.ReleaseBatchDraft) &&
			filter.State != string(governance.ReleaseBatchSealed) && filter.State != string(governance.ReleaseBatchCancelled) {
		return governance.ReleaseBatchPage{}, ErrInvalidReleaseBatchFilter
	}
	page := governance.ReleaseBatchPage{Items: []governance.ReleaseBatch{}, Limit: filter.Limit, Offset: filter.Offset}
	if err := s.db.QueryRow(ctx, countReleaseBatchesSQL, filter.State).Scan(&page.Total); err != nil {
		return governance.ReleaseBatchPage{}, fmt.Errorf("count release batches: %w", err)
	}
	rows, err := s.db.Query(ctx, listReleaseBatchesSQL, filter.State, filter.Limit, filter.Offset)
	if err != nil {
		return governance.ReleaseBatchPage{}, fmt.Errorf("list release batches: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		batch, err := scanReleaseBatch(rows)
		if err != nil {
			return governance.ReleaseBatchPage{}, fmt.Errorf("scan release batch: %w", err)
		}
		page.Items = append(page.Items, batch)
	}
	if err := rows.Err(); err != nil {
		return governance.ReleaseBatchPage{}, fmt.Errorf("iterate release batches: %w", err)
	}
	return page, nil
}

func (s *Store) GetReleaseBatch(ctx context.Context, batchID string) (governance.ReleaseBatchDetail, error) {
	if !validShortID(batchID) {
		return governance.ReleaseBatchDetail{}, ErrReleaseBatchNotFound
	}
	batch, err := scanReleaseBatch(s.db.QueryRow(ctx, getReleaseBatchSQL, batchID))
	if errors.Is(err, pgx.ErrNoRows) {
		return governance.ReleaseBatchDetail{}, ErrReleaseBatchNotFound
	}
	if err != nil {
		return governance.ReleaseBatchDetail{}, fmt.Errorf("get release batch: %w", err)
	}
	rows, err := s.db.Query(ctx, listReleaseBatchItemsSQL, batchID)
	if err != nil {
		return governance.ReleaseBatchDetail{}, fmt.Errorf("list release batch items: %w", err)
	}
	defer rows.Close()
	detail := governance.ReleaseBatchDetail{ReleaseBatch: batch, Items: []governance.ReleaseBatchItemView{}}
	for rows.Next() {
		var item governance.ReleaseBatchItemView
		var visibility string
		if err := rows.Scan(&item.PublicationID, &item.Title, &visibility, &item.Action); err != nil {
			return governance.ReleaseBatchDetail{}, fmt.Errorf("scan release batch item: %w", err)
		}
		item.DecidedVisibility = governance.Visibility(visibility)
		if (item.DecidedVisibility != governance.VisibilityInternal && item.DecidedVisibility != governance.VisibilityExternal) || (item.Action != "upsert" && item.Action != "delete") {
			return governance.ReleaseBatchDetail{}, errors.New("invalid release batch item visibility")
		}
		detail.Items = append(detail.Items, item)
	}
	if err := rows.Err(); err != nil {
		return governance.ReleaseBatchDetail{}, fmt.Errorf("iterate release batch items: %w", err)
	}
	return detail, nil
}

func validCreateReleaseBatch(request governance.CreateReleaseBatchRequest) bool {
	if request.Name == "" || request.Name != strings.TrimSpace(request.Name) || len(request.Name) > 255 ||
		strings.ContainsAny(request.Name, "\x00\r\n") || !validReviewerRef(request.CreatedBy) ||
		request.Now.IsZero() || len(request.PublicationIDs)+len(request.DeletePublicationIDs) < 1 || len(request.PublicationIDs)+len(request.DeletePublicationIDs) > 100 {
		return false
	}
	if request.ReleaseChangeSetID != "" && !validRouteID(request.ReleaseChangeSetID) {
		return false
	}
	seen := make(map[string]struct{}, len(request.PublicationIDs)+len(request.DeletePublicationIDs))
	for _, publicationID := range append(append([]string{}, request.PublicationIDs...), request.DeletePublicationIDs...) {
		if !validShortID(publicationID) {
			return false
		}
		if _, exists := seen[publicationID]; exists {
			return false
		}
		seen[publicationID] = struct{}{}
	}
	return true
}

func scanReleaseBatch(row rowScanner) (governance.ReleaseBatch, error) {
	var batch governance.ReleaseBatch
	var state string
	if err := row.Scan(
		&batch.ID, &batch.Name, &state, &batch.PolicyVersion,
		&batch.CreatedBy, &batch.SealedBy, &batch.ItemCount,
		&batch.CreatedAt, &batch.SealedAt, &batch.ReleaseChangeSetID,
	); err != nil {
		return governance.ReleaseBatch{}, err
	}
	batch.State = governance.ReleaseBatchState(state)
	if !batch.State.Valid() || batch.PolicyVersion != releaseGatePolicyVersion {
		return governance.ReleaseBatch{}, errors.New("invalid release batch row")
	}
	return batch, nil
}