package issuetrackerconformance

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"sort"
	"strings"

	cp "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
)

type issueWorkflowOpaquePlan struct {
	Scope       cp.ComponentManagedScope `json:"scope"`
	PackName    string                   `json:"pack_name"`
	BusinessKey string                   `json:"business_key,omitempty"`
	Change      bool                     `json:"change,omitempty"`
	Stale       []string                 `json:"stale,omitempty"`
}

type issueWorkflowSetProjection struct {
	BusinessKeys []string `json:"business_keys"`
}

type issueWorkflowAggregate struct {
	Present    bool
	Projection IssueWorkflowDeclaration
	RowCount   int
}

type issueWorkflowOwnerInventory struct {
	Parents        map[string]string
	Owned          map[string]string
	Targets        []cp.ComponentManagedTarget
	ParentRows     int
	StatusRows     int
	TransitionRows int
}

type issueWorkflowReadRows func(context.Context, cp.ComponentManagedRowQuery) ([]cp.ComponentRow, error)

func issueWorkflowLifecyclePlan(ctx context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
	if store == nil {
		return cp.ComponentManagedPlan{}, fmt.Errorf("issue workflow managed read store is required")
	}
	if err := request.Validate(scope); err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	switch scope {
	case cp.ComponentManagedScopeItem:
		return issueWorkflowLifecyclePlanItem(ctx, request, store)
	case cp.ComponentManagedScopeSet:
		return issueWorkflowLifecyclePlanSet(ctx, request, store)
	default:
		return cp.ComponentManagedPlan{}, fmt.Errorf("issue workflow managed scope %q is unsupported", scope)
	}
}

func issueWorkflowLifecyclePlanItem(ctx context.Context, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
	desired, err := decodeIssueWorkflowManagedRequest(*request.Item)
	if err != nil {
		return cp.ComponentManagedPlan{}, fmt.Errorf("issue workflow managed plan: %w", err)
	}
	if err := validateIssueWorkflowDesiredCardinality(desired); err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	if err := requireIssueWorkflowOwnerGuard(ctx, store.ReadRows, request.Item.PackName); err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	current, err := readIssueWorkflowAggregate(ctx, store.ReadRows, request.Item.PackName, desired.ProjectKey, desired.WorkflowKey)
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	live, err := marshalIssueWorkflowAggregate(current)
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	change := !bytes.Equal(request.Item.ExpectedProjection, live)
	currentItems := 1 + current.RowCount
	deltaRows, mutationStatements := 0, 0
	if change {
		currentItems += current.RowCount
		deltaRows = 1 + current.RowCount - boolInt(current.Present) + len(desired.Projection.Statuses) + len(desired.Projection.Transitions)
		mutationStatements = 4
		if len(desired.Projection.Transitions) != 0 {
			mutationStatements++
		}
	}
	opaque, err := json.Marshal(issueWorkflowOpaquePlan{
		Scope: cp.ComponentManagedScopeItem, PackName: request.Item.PackName,
		BusinessKey: request.Item.Component.BusinessKey, Change: change,
	})
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	return cp.NewComponentManagedPlan(cp.ComponentManagedPlanSpec{
		Targets: []cp.ComponentManagedTarget{
			issueWorkflowItemTarget(desired.ProjectKey, desired.WorkflowKey), issueWorkflowSetTarget(),
		},
		RequiredCapabilities: issueWorkflowRequiredCapabilities(false),
		Effects:              issueWorkflowEffects(),
		Estimate: cp.ComponentManagedUsage{
			CurrentItems: currentItems,
			DesiredItems: 1 + len(desired.Projection.Statuses) + len(desired.Projection.Transitions),
			DeltaRows:    deltaRows, MutationStatements: mutationStatements, BatchRows: issueWorkflowMaximumRows,
			EvidenceWrites: 1, EffectCalls: boolInt(change),
		},
		ProjectionIdentity: issueWorkflowItemProjectionIdentity(*request.Item),
		OpaquePlan:         opaque,
	})
}

func issueWorkflowLifecyclePlanSet(ctx context.Context, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
	set := *request.Set
	if err := validateIssueWorkflowSetRequest(set); err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	if len(request.DesiredBusinessKeys) > issueWorkflowMaximumRows {
		return cp.ComponentManagedPlan{}, fmt.Errorf("issue workflow set has %d desired workflows; maximum is %d", len(request.DesiredBusinessKeys), issueWorkflowMaximumRows)
	}
	desired, _, desiredTargets, err := normalizeIssueWorkflowBusinessKeys(request.DesiredBusinessKeys)
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	if err := requireIssueWorkflowOwnerGuard(ctx, store.ReadRows, set.PackName); err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	inventory, err := readIssueWorkflowOwnerInventory(ctx, store.ReadRows, set.PackName)
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	targets := []cp.ComponentManagedTarget{issueWorkflowSetTarget()}
	for _, target := range desiredTargets {
		targets = append(targets, target)
	}
	targets = append(targets, inventory.Targets...)
	stale := make([]string, 0)
	for identity, businessKey := range inventory.Owned {
		_, hasParent := inventory.Parents[identity]
		_, retained := desired[identity]
		if !hasParent || !retained {
			stale = append(stale, businessKey)
		}
	}
	sort.Strings(stale)
	opaque, err := json.Marshal(issueWorkflowOpaquePlan{Scope: cp.ComponentManagedScopeSet, PackName: set.PackName, Change: len(stale) != 0, Stale: stale})
	if err != nil {
		return cp.ComponentManagedPlan{}, err
	}
	return cp.NewComponentManagedPlan(cp.ComponentManagedPlanSpec{
		Targets: targets, RequiredCapabilities: issueWorkflowRequiredCapabilities(true), Effects: issueWorkflowEffects(),
		Estimate: cp.ComponentManagedUsage{
			CurrentItems: 1 + inventory.ParentRows + inventory.StatusRows + inventory.TransitionRows + len(stale)*(1+2*issueWorkflowMaximumRows),
			DesiredItems: len(request.DesiredBusinessKeys),
			DeltaRows:    len(stale) * (1 + 2*issueWorkflowMaximumRows), MutationStatements: len(stale) * 3, BatchRows: issueWorkflowMaximumChildRows,
			EvidenceWrites: 1, EffectCalls: boolInt(len(stale) != 0),
		},
		ProjectionIdentity: issueWorkflowSetProjectionIdentity(set), OpaquePlan: opaque,
	})
}

func issueWorkflowLifecycleMutate(ctx context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, plan cp.ComponentManagedPlan, store cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
	if store == nil {
		return cp.ComponentManagedMutationResult{}, fmt.Errorf("issue workflow managed mutation store is required")
	}
	if err := request.Validate(scope); err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	if err := plan.Validate(); err != nil {
		return cp.ComponentManagedMutationResult{}, fmt.Errorf("issue workflow managed plan: %w", err)
	}
	var opaque issueWorkflowOpaquePlan
	if err := json.Unmarshal(plan.OpaquePlan(), &opaque); err != nil {
		return cp.ComponentManagedMutationResult{}, fmt.Errorf("decode issue workflow opaque plan: %w", err)
	}
	if opaque.Scope != scope {
		return cp.ComponentManagedMutationResult{}, fmt.Errorf("issue workflow opaque scope %q does not match %q", opaque.Scope, scope)
	}
	switch scope {
	case cp.ComponentManagedScopeItem:
		return issueWorkflowLifecycleMutateItem(ctx, request, opaque, store)
	case cp.ComponentManagedScopeSet:
		return issueWorkflowLifecycleMutateSet(ctx, request, opaque, store)
	default:
		return cp.ComponentManagedMutationResult{}, fmt.Errorf("issue workflow managed scope %q is unsupported", scope)
	}
}

func issueWorkflowLifecycleMutateItem(ctx context.Context, request cp.ComponentManagedLifecycleRequest, opaque issueWorkflowOpaquePlan, store cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
	if opaque.PackName != request.Item.PackName || opaque.BusinessKey != request.Item.Component.BusinessKey {
		return cp.ComponentManagedMutationResult{}, fmt.Errorf("issue workflow item opaque identity does not match request")
	}
	if !opaque.Change {
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
	}
	desired, err := decodeIssueWorkflowManagedRequest(*request.Item)
	if err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	if err := validateIssueWorkflowDesiredCardinality(desired); err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	current, err := readIssueWorkflowAggregate(ctx, store.LockRows, request.Item.PackName, desired.ProjectKey, desired.WorkflowKey)
	if err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	parentValues := issueWorkflowParentValues(request.Item.PackName, desired)
	if current.Present {
		affected, updateErr := store.UpdateRows(ctx, cp.ComponentManagedUpdate{
			Table:  issueWorkflowTable,
			Match:  cp.ComponentRow{"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey},
			Values: cp.ComponentRow{"Name": desired.Projection.Name, "OwnerPack": request.Item.PackName}, MaximumAffectedRows: 1,
		})
		if updateErr != nil {
			return cp.ComponentManagedMutationResult{}, updateErr
		}
		if affected != 1 {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf("issue workflow parent update affected %d rows, want 1", affected)
		}
	} else if err := insertIssueWorkflowRows(ctx, store, issueWorkflowTable, []cp.ComponentManagedInsertRow{{
		LogicalKey: cp.ComponentRow{"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey}, Values: parentValues,
	}}); err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	for _, table := range []string{issueWorkflowTransitionTable, issueWorkflowStatusTable} {
		if _, err := store.DeleteRows(ctx, cp.ComponentManagedDelete{
			Table: table, Match: cp.ComponentRow{"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey, "OwnerPack": request.Item.PackName}, MaximumAffectedRows: issueWorkflowMaximumRows,
		}); err != nil {
			return cp.ComponentManagedMutationResult{}, err
		}
	}
	if err := insertIssueWorkflowRows(ctx, store, issueWorkflowStatusTable, issueWorkflowStatusInsertRows(request.Item.PackName, desired)); err != nil {
		return cp.ComponentManagedMutationResult{}, err
	}
	if rows := issueWorkflowTransitionInsertRows(request.Item.PackName, desired); len(rows) != 0 {
		if err := insertIssueWorkflowRows(ctx, store, issueWorkflowTransitionTable, rows); err != nil {
			return cp.ComponentManagedMutationResult{}, err
		}
	}
	return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationDesiredProjectionStaged}, nil
}

func issueWorkflowLifecycleMutateSet(ctx context.Context, request cp.ComponentManagedLifecycleRequest, opaque issueWorkflowOpaquePlan, store cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
	if opaque.PackName != request.Set.PackName {
		return cp.ComponentManagedMutationResult{}, fmt.Errorf("issue workflow set opaque identity does not match request")
	}
	if !opaque.Change {
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
	}
	for _, businessKey := range opaque.Stale {
		projectKey, workflowKey, err := parseIssueWorkflowBusinessKey(businessKey)
		if err != nil {
			return cp.ComponentManagedMutationResult{}, err
		}
		current, err := readIssueWorkflowAggregate(ctx, store.LockRows, request.Set.PackName, projectKey, workflowKey)
		if err != nil {
			return cp.ComponentManagedMutationResult{}, err
		}
		if current.RowCount == 0 {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf("stale issue workflow %q disappeared after locked planning", businessKey)
		}
		var deleted int64
		for _, table := range []string{issueWorkflowTransitionTable, issueWorkflowStatusTable} {
			affected, deleteErr := store.DeleteRows(ctx, cp.ComponentManagedDelete{
				Table: table, Match: cp.ComponentRow{"ProjectKey": projectKey, "WorkflowKey": workflowKey, "OwnerPack": request.Set.PackName}, MaximumAffectedRows: issueWorkflowMaximumRows,
			})
			if deleteErr != nil {
				return cp.ComponentManagedMutationResult{}, deleteErr
			}
			deleted += affected
		}
		affected, err := store.DeleteRows(ctx, cp.ComponentManagedDelete{
			Table:               issueWorkflowTable,
			Match:               cp.ComponentRow{"ProjectKey": projectKey, "WorkflowKey": workflowKey, "OwnerPack": request.Set.PackName},
			MaximumAffectedRows: 1,
		})
		if err != nil {
			return cp.ComponentManagedMutationResult{}, err
		}
		if (current.Present && affected != 1) || (!current.Present && affected != 0) {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf("stale issue workflow %q parent delete affected %d rows, present = %t", businessKey, affected, current.Present)
		}
		deleted += affected
		if deleted != int64(current.RowCount) {
			return cp.ComponentManagedMutationResult{}, fmt.Errorf("stale issue workflow %q delete affected %d rows, want %d", businessKey, deleted, current.RowCount)
		}
	}
	return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationDesiredProjectionStaged}, nil
}

func issueWorkflowLifecycleProject(ctx context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
	if store == nil {
		return cp.ComponentManagedProjection{}, fmt.Errorf("issue workflow managed projection store is required")
	}
	if err := request.Validate(scope); err != nil {
		return cp.ComponentManagedProjection{}, err
	}
	switch scope {
	case cp.ComponentManagedScopeItem:
		desired, err := decodeIssueWorkflowManagedRequest(*request.Item)
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		if err := requireIssueWorkflowOwnerGuard(ctx, store.ReadRows, request.Item.PackName); err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		current, err := readIssueWorkflowAggregate(ctx, store.ReadRows, request.Item.PackName, desired.ProjectKey, desired.WorkflowKey)
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		live, err := marshalIssueWorkflowAggregate(current)
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		return cp.ComponentManagedProjection{Identity: issueWorkflowItemProjectionIdentity(*request.Item), Expected: append([]byte(nil), request.Item.ExpectedProjection...), Live: live}, nil
	case cp.ComponentManagedScopeSet:
		if err := validateIssueWorkflowSetRequest(*request.Set); err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		if len(request.DesiredBusinessKeys) > issueWorkflowMaximumRows {
			return cp.ComponentManagedProjection{}, fmt.Errorf("issue workflow set has %d desired workflows; maximum is %d", len(request.DesiredBusinessKeys), issueWorkflowMaximumRows)
		}
		_, desiredKeys, _, err := normalizeIssueWorkflowBusinessKeys(request.DesiredBusinessKeys)
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		if err := requireIssueWorkflowOwnerGuard(ctx, store.ReadRows, request.Set.PackName); err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		inventory, err := readIssueWorkflowOwnerInventory(ctx, store.ReadRows, request.Set.PackName)
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		for identity, businessKey := range inventory.Owned {
			if _, hasParent := inventory.Parents[identity]; !hasParent {
				return cp.ComponentManagedProjection{}, fmt.Errorf("issue workflow owner set contains orphaned child rows for %q", businessKey)
			}
		}
		liveKeys := make([]string, 0, len(inventory.Parents))
		for _, businessKey := range inventory.Parents {
			liveKeys = append(liveKeys, businessKey)
		}
		sort.Strings(liveKeys)
		expected, err := json.Marshal(issueWorkflowSetProjection{BusinessKeys: desiredKeys})
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		live, err := json.Marshal(issueWorkflowSetProjection{BusinessKeys: liveKeys})
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		return cp.ComponentManagedProjection{Identity: issueWorkflowSetProjectionIdentity(*request.Set), Expected: expected, Live: live}, nil
	default:
		return cp.ComponentManagedProjection{}, fmt.Errorf("issue workflow managed scope %q is unsupported", scope)
	}
}

func validateIssueWorkflowDesiredCardinality(desired issueWorkflowDesired) error {
	if len(desired.Projection.Statuses) > issueWorkflowMaximumRows {
		return fmt.Errorf("issue workflow has %d desired status rows; maximum is %d", len(desired.Projection.Statuses), issueWorkflowMaximumRows)
	}
	if len(desired.Projection.Transitions) > issueWorkflowMaximumRows {
		return fmt.Errorf("issue workflow has %d desired transition rows; maximum is %d", len(desired.Projection.Transitions), issueWorkflowMaximumRows)
	}
	return nil
}

func validateIssueWorkflowSetRequest(request cp.ComponentManagedSetRequest) error {
	if request.PackName == "" || strings.TrimSpace(request.PackName) != request.PackName || len(request.PackName) > issueWorkflowMaximumText {
		return fmt.Errorf("issue workflow set pack name %q is not canonical", request.PackName)
	}
	if request.SourceOwner != "" {
		return fmt.Errorf("pack-owned issue workflow set must not declare source owner %q", request.SourceOwner)
	}
	if request.Kind != IssueWorkflowComponentKind {
		return fmt.Errorf("issue workflow set kind = %q, want %q", request.Kind, IssueWorkflowComponentKind)
	}
	return nil
}

func requireIssueWorkflowOwnerGuard(ctx context.Context, read issueWorkflowReadRows, packName string) error {
	rows, err := readIssueWorkflowRows(ctx, read, cp.ComponentManagedRowQuery{
		Table: issueWorkflowOwnerGuardTable, Columns: []string{"OwnerPack", "Enabled"}, Equal: cp.ComponentRow{"OwnerPack": packName},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "OwnerPack", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"OwnerPack"}, MaximumRows: 1,
	})
	if err != nil {
		return fmt.Errorf("read issue workflow owner guard %q: %w", packName, err)
	}
	if len(rows) != 1 {
		return fmt.Errorf("issue workflow owner guard %q resolves to %d rows, want 1", packName, len(rows))
	}
	owner, err := issueWorkflowRowString(issueWorkflowOwnerGuardTable, rows[0], "OwnerPack")
	if err != nil || owner != packName {
		return fmt.Errorf("issue workflow owner guard escaped owner %q", packName)
	}
	enabled, ok := rows[0]["Enabled"].(bool)
	if !ok || !enabled {
		return fmt.Errorf("issue workflow owner guard %q is not enabled", packName)
	}
	return nil
}

func readIssueWorkflowAggregate(ctx context.Context, read issueWorkflowReadRows, packName string, projectKey, workflowKey []byte) (issueWorkflowAggregate, error) {
	parentRows, err := readIssueWorkflowRows(ctx, read, issueWorkflowParentQuery(projectKey, workflowKey))
	if err != nil {
		return issueWorkflowAggregate{}, fmt.Errorf("read issue workflow parent: %w", err)
	}
	statusRows, err := readIssueWorkflowRows(ctx, read, issueWorkflowStatusQuery(projectKey, workflowKey))
	if err != nil {
		return issueWorkflowAggregate{}, fmt.Errorf("read issue workflow statuses: %w", err)
	}
	transitionRows, err := readIssueWorkflowRows(ctx, read, issueWorkflowTransitionQuery(projectKey, workflowKey))
	if err != nil {
		return issueWorkflowAggregate{}, fmt.Errorf("read issue workflow transitions: %w", err)
	}
	aggregate := issueWorkflowAggregate{RowCount: len(parentRows) + len(statusRows) + len(transitionRows)}
	if len(parentRows) == 0 {
		for _, group := range []struct {
			table string
			rows  []cp.ComponentRow
		}{
			{table: issueWorkflowStatusTable, rows: statusRows},
			{table: issueWorkflowTransitionTable, rows: transitionRows},
		} {
			for _, row := range group.rows {
				if err := validateIssueWorkflowRowTarget(group.table, row, projectKey, workflowKey); err != nil {
					return issueWorkflowAggregate{}, err
				}
				if err := validateIssueWorkflowRowOwner(group.table, row, packName); err != nil {
					return issueWorkflowAggregate{}, err
				}
			}
		}
		return aggregate, nil
	}
	owner, err := issueWorkflowRowString(issueWorkflowTable, parentRows[0], "OwnerPack")
	if err != nil {
		return issueWorkflowAggregate{}, err
	}
	if owner != packName {
		return issueWorkflowAggregate{}, fmt.Errorf("issue workflow binary target is owned by %q, not %q", owner, packName)
	}
	if err := validateIssueWorkflowRowTarget(issueWorkflowTable, parentRows[0], projectKey, workflowKey); err != nil {
		return issueWorkflowAggregate{}, err
	}
	name, err := issueWorkflowRowString(issueWorkflowTable, parentRows[0], "Name")
	if err != nil {
		return issueWorkflowAggregate{}, err
	}
	projection := IssueWorkflowDeclaration{
		ProjectKey: base64.RawURLEncoding.EncodeToString(projectKey), WorkflowKey: base64.RawURLEncoding.EncodeToString(workflowKey), Name: name,
		Statuses: make([]IssueWorkflowStatus, 0, len(statusRows)), Transitions: make([]IssueWorkflowTransition, 0, len(transitionRows)),
	}
	seenStatuses := make(map[string]struct{}, len(statusRows))
	for _, row := range statusRows {
		if err := validateIssueWorkflowRowTarget(issueWorkflowStatusTable, row, projectKey, workflowKey); err != nil {
			return issueWorkflowAggregate{}, err
		}
		if err := validateIssueWorkflowRowOwner(issueWorkflowStatusTable, row, packName); err != nil {
			return issueWorkflowAggregate{}, err
		}
		key, rowErr := issueWorkflowRowString(issueWorkflowStatusTable, row, "StatusKey")
		if rowErr != nil {
			return issueWorkflowAggregate{}, rowErr
		}
		label, rowErr := issueWorkflowRowString(issueWorkflowStatusTable, row, "Label")
		if rowErr != nil {
			return issueWorkflowAggregate{}, rowErr
		}
		if !issueWorkflowKeyPattern.MatchString(key) || label == "" || strings.TrimSpace(label) != label || len(label) > issueWorkflowMaximumText {
			return issueWorkflowAggregate{}, fmt.Errorf("issue workflow status row %q is not canonical", key)
		}
		if _, duplicate := seenStatuses[key]; duplicate {
			return issueWorkflowAggregate{}, fmt.Errorf("issue workflow status row key %q is duplicated", key)
		}
		seenStatuses[key] = struct{}{}
		projection.Statuses = append(projection.Statuses, IssueWorkflowStatus{Key: key, Label: label})
	}
	seenTransitions := make(map[string]struct{}, len(transitionRows))
	for _, row := range transitionRows {
		if err := validateIssueWorkflowRowTarget(issueWorkflowTransitionTable, row, projectKey, workflowKey); err != nil {
			return issueWorkflowAggregate{}, err
		}
		if err := validateIssueWorkflowRowOwner(issueWorkflowTransitionTable, row, packName); err != nil {
			return issueWorkflowAggregate{}, err
		}
		key, rowErr := issueWorkflowRowString(issueWorkflowTransitionTable, row, "TransitionKey")
		if rowErr != nil {
			return issueWorkflowAggregate{}, rowErr
		}
		from, rowErr := issueWorkflowRowString(issueWorkflowTransitionTable, row, "FromStatusKey")
		if rowErr != nil {
			return issueWorkflowAggregate{}, rowErr
		}
		to, rowErr := issueWorkflowRowString(issueWorkflowTransitionTable, row, "ToStatusKey")
		if rowErr != nil {
			return issueWorkflowAggregate{}, rowErr
		}
		if !issueWorkflowKeyPattern.MatchString(key) {
			return issueWorkflowAggregate{}, fmt.Errorf("issue workflow transition row key %q is not canonical", key)
		}
		if _, duplicate := seenTransitions[key]; duplicate {
			return issueWorkflowAggregate{}, fmt.Errorf("issue workflow transition row key %q is duplicated", key)
		}
		seenTransitions[key] = struct{}{}
		if _, ok := seenStatuses[from]; !ok {
			return issueWorkflowAggregate{}, fmt.Errorf("issue workflow transition %q has missing from status %q", key, from)
		}
		if _, ok := seenStatuses[to]; !ok {
			return issueWorkflowAggregate{}, fmt.Errorf("issue workflow transition %q has missing to status %q", key, to)
		}
		projection.Transitions = append(projection.Transitions, IssueWorkflowTransition{Key: key, From: from, To: to})
	}
	sort.Slice(projection.Statuses, func(i, j int) bool { return projection.Statuses[i].Key < projection.Statuses[j].Key })
	sort.Slice(projection.Transitions, func(i, j int) bool { return projection.Transitions[i].Key < projection.Transitions[j].Key })
	return issueWorkflowAggregate{Present: true, Projection: projection, RowCount: aggregate.RowCount}, nil
}

func readIssueWorkflowRows(ctx context.Context, read issueWorkflowReadRows, query cp.ComponentManagedRowQuery) ([]cp.ComponentRow, error) {
	rows, err := read(ctx, query)
	if err != nil {
		return nil, err
	}
	if len(rows) > query.MaximumRows {
		return nil, fmt.Errorf("issue workflow managed read returned %d rows; maximum is %d", len(rows), query.MaximumRows)
	}
	return rows, nil
}

func issueWorkflowParentQuery(projectKey, workflowKey []byte) cp.ComponentManagedRowQuery {
	return cp.ComponentManagedRowQuery{
		Table: issueWorkflowTable, Columns: []string{"ProjectKey", "WorkflowKey", "OwnerPack", "Name"},
		Equal:             cp.ComponentRow{"ProjectKey": projectKey, "WorkflowKey": workflowKey},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "ProjectKey", Direction: cp.ComponentManagedOrderAscending}, {Column: "WorkflowKey", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"ProjectKey", "WorkflowKey"}, MaximumRows: 1,
	}
}

func issueWorkflowStatusQuery(projectKey, workflowKey []byte) cp.ComponentManagedRowQuery {
	return cp.ComponentManagedRowQuery{
		Table: issueWorkflowStatusTable, Columns: []string{"ProjectKey", "WorkflowKey", "OwnerPack", "StatusKey", "Label"},
		Equal:             cp.ComponentRow{"ProjectKey": projectKey, "WorkflowKey": workflowKey},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "StatusKey", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"ProjectKey", "WorkflowKey", "StatusKey"}, MaximumRows: issueWorkflowMaximumRows,
	}
}

func issueWorkflowTransitionQuery(projectKey, workflowKey []byte) cp.ComponentManagedRowQuery {
	return cp.ComponentManagedRowQuery{
		Table: issueWorkflowTransitionTable, Columns: []string{"ProjectKey", "WorkflowKey", "OwnerPack", "TransitionKey", "FromStatusKey", "ToStatusKey"},
		Equal:             cp.ComponentRow{"ProjectKey": projectKey, "WorkflowKey": workflowKey},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "TransitionKey", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"ProjectKey", "WorkflowKey", "TransitionKey"}, MaximumRows: issueWorkflowMaximumRows,
	}
}

func issueWorkflowOwnerParentQuery(packName string) cp.ComponentManagedRowQuery {
	return cp.ComponentManagedRowQuery{
		Table: issueWorkflowTable, Columns: []string{"ProjectKey", "WorkflowKey", "OwnerPack", "Name"}, Equal: cp.ComponentRow{"OwnerPack": packName},
		OrderBy:           []cp.ComponentManagedOrder{{Column: "ProjectKey", Direction: cp.ComponentManagedOrderAscending}, {Column: "WorkflowKey", Direction: cp.ComponentManagedOrderAscending}},
		LogicalKeyColumns: []string{"ProjectKey", "WorkflowKey"}, MaximumRows: issueWorkflowMaximumRows,
	}
}

func issueWorkflowOwnerStatusQuery(packName string) cp.ComponentManagedRowQuery {
	return cp.ComponentManagedRowQuery{
		Table: issueWorkflowStatusTable, Columns: []string{"ProjectKey", "WorkflowKey", "OwnerPack", "StatusKey"}, Equal: cp.ComponentRow{"OwnerPack": packName},
		OrderBy: []cp.ComponentManagedOrder{
			{Column: "ProjectKey", Direction: cp.ComponentManagedOrderAscending},
			{Column: "WorkflowKey", Direction: cp.ComponentManagedOrderAscending},
			{Column: "StatusKey", Direction: cp.ComponentManagedOrderAscending},
		},
		LogicalKeyColumns: []string{"ProjectKey", "WorkflowKey", "StatusKey"}, MaximumRows: issueWorkflowMaximumChildRows,
	}
}

func issueWorkflowOwnerTransitionQuery(packName string) cp.ComponentManagedRowQuery {
	return cp.ComponentManagedRowQuery{
		Table: issueWorkflowTransitionTable, Columns: []string{"ProjectKey", "WorkflowKey", "OwnerPack", "TransitionKey"}, Equal: cp.ComponentRow{"OwnerPack": packName},
		OrderBy: []cp.ComponentManagedOrder{
			{Column: "ProjectKey", Direction: cp.ComponentManagedOrderAscending},
			{Column: "WorkflowKey", Direction: cp.ComponentManagedOrderAscending},
			{Column: "TransitionKey", Direction: cp.ComponentManagedOrderAscending},
		},
		LogicalKeyColumns: []string{"ProjectKey", "WorkflowKey", "TransitionKey"}, MaximumRows: issueWorkflowMaximumChildRows,
	}
}

func readIssueWorkflowOwnerInventory(ctx context.Context, read issueWorkflowReadRows, packName string) (issueWorkflowOwnerInventory, error) {
	parentRows, err := readIssueWorkflowRows(ctx, read, issueWorkflowOwnerParentQuery(packName))
	if err != nil {
		return issueWorkflowOwnerInventory{}, fmt.Errorf("read issue workflow owner parents %q: %w", packName, err)
	}
	statusRows, err := readIssueWorkflowRows(ctx, read, issueWorkflowOwnerStatusQuery(packName))
	if err != nil {
		return issueWorkflowOwnerInventory{}, fmt.Errorf("read issue workflow owner statuses %q: %w", packName, err)
	}
	transitionRows, err := readIssueWorkflowRows(ctx, read, issueWorkflowOwnerTransitionQuery(packName))
	if err != nil {
		return issueWorkflowOwnerInventory{}, fmt.Errorf("read issue workflow owner transitions %q: %w", packName, err)
	}
	inventory := issueWorkflowOwnerInventory{
		Parents: make(map[string]string, len(parentRows)), Owned: make(map[string]string, len(parentRows)),
		ParentRows: len(parentRows), StatusRows: len(statusRows), TransitionRows: len(transitionRows),
	}
	add := func(table string, row cp.ComponentRow, parent bool) (string, error) {
		businessKey, projectKey, workflowKey, rowErr := issueWorkflowBusinessKeyFromOwnerRow(table, packName, row)
		if rowErr != nil {
			return "", rowErr
		}
		identity := string(issueWorkflowCompoundTargetKey(projectKey, workflowKey))
		if previous, duplicate := inventory.Owned[identity]; duplicate && previous != businessKey {
			return "", fmt.Errorf("issue workflow owner rows %q and %q share one binary target", previous, businessKey)
		}
		inventory.Owned[identity] = businessKey
		if parent {
			if previous, duplicate := inventory.Parents[identity]; duplicate {
				return "", fmt.Errorf("issue workflow owner set contains duplicate parent target %q and %q", previous, businessKey)
			}
			inventory.Parents[identity] = businessKey
		}
		return identity, nil
	}
	for _, row := range parentRows {
		if _, err := add(issueWorkflowTable, row, true); err != nil {
			return issueWorkflowOwnerInventory{}, err
		}
	}
	statusCounts := make(map[string]int)
	for _, row := range statusRows {
		identity, err := add(issueWorkflowStatusTable, row, false)
		if err != nil {
			return issueWorkflowOwnerInventory{}, err
		}
		statusCounts[identity]++
		if statusCounts[identity] > issueWorkflowMaximumRows {
			return issueWorkflowOwnerInventory{}, fmt.Errorf("issue workflow owner target has %d status rows; maximum is %d", statusCounts[identity], issueWorkflowMaximumRows)
		}
	}
	transitionCounts := make(map[string]int)
	for _, row := range transitionRows {
		identity, err := add(issueWorkflowTransitionTable, row, false)
		if err != nil {
			return issueWorkflowOwnerInventory{}, err
		}
		transitionCounts[identity]++
		if transitionCounts[identity] > issueWorkflowMaximumRows {
			return issueWorkflowOwnerInventory{}, fmt.Errorf("issue workflow owner target has %d transition rows; maximum is %d", transitionCounts[identity], issueWorkflowMaximumRows)
		}
	}
	if len(inventory.Owned) > issueWorkflowMaximumRows {
		return issueWorkflowOwnerInventory{}, fmt.Errorf("issue workflow owner set has %d physical workflows; maximum is %d", len(inventory.Owned), issueWorkflowMaximumRows)
	}
	keys := make([]string, 0, len(inventory.Owned))
	for _, businessKey := range inventory.Owned {
		keys = append(keys, businessKey)
	}
	sort.Strings(keys)
	inventory.Targets = make([]cp.ComponentManagedTarget, 0, len(keys))
	for _, businessKey := range keys {
		projectKey, workflowKey, err := parseIssueWorkflowBusinessKey(businessKey)
		if err != nil {
			return issueWorkflowOwnerInventory{}, err
		}
		inventory.Targets = append(inventory.Targets, issueWorkflowItemTarget(projectKey, workflowKey))
	}
	return inventory, nil
}

func issueWorkflowBusinessKeyFromOwnerRow(table, packName string, row cp.ComponentRow) (string, []byte, []byte, error) {
	projectKey, err := issueWorkflowRowBytes(table, row, "ProjectKey")
	if err != nil {
		return "", nil, nil, err
	}
	workflowKey, err := issueWorkflowRowBytes(table, row, "WorkflowKey")
	if err != nil {
		return "", nil, nil, err
	}
	owner, err := issueWorkflowRowString(table, row, "OwnerPack")
	if err != nil {
		return "", nil, nil, err
	}
	if owner != packName {
		return "", nil, nil, fmt.Errorf("issue workflow owner row %q escaped owner query %q", issueWorkflowBusinessKey(projectKey, workflowKey), packName)
	}
	return issueWorkflowBusinessKey(projectKey, workflowKey), projectKey, workflowKey, nil
}

func normalizeIssueWorkflowBusinessKeys(keys []string) (map[string]string, []string, []cp.ComponentManagedTarget, error) {
	values := make(map[string]string, len(keys))
	canonical := make([]string, 0, len(keys))
	targets := make([]cp.ComponentManagedTarget, 0, len(keys))
	for _, businessKey := range keys {
		projectKey, workflowKey, err := parseIssueWorkflowBusinessKey(businessKey)
		if err != nil {
			return nil, nil, nil, err
		}
		identity := string(issueWorkflowCompoundTargetKey(projectKey, workflowKey))
		if previous, duplicate := values[identity]; duplicate {
			return nil, nil, nil, fmt.Errorf("issue workflow desired keys %q and %q share one binary target", previous, businessKey)
		}
		values[identity] = businessKey
		canonical = append(canonical, businessKey)
		targets = append(targets, issueWorkflowItemTarget(projectKey, workflowKey))
	}
	sort.Strings(canonical)
	return values, canonical, targets, nil
}

func issueWorkflowRequiredCapabilities(exactSet bool) []cp.ComponentManagedCapability {
	values := []cp.ComponentManagedCapability{
		cp.ComponentManagedCapabilityTargetSerialization,
		cp.ComponentManagedCapabilityStructuredMutation,
		cp.ComponentManagedCapabilityEffects,
		cp.ComponentManagedCapabilityEvidence,
		cp.ComponentManagedCapabilityFreshProjection,
	}
	if exactSet {
		values = append(values, cp.ComponentManagedCapabilityExactSet)
	}
	return values
}

func issueWorkflowEffects() []cp.ComponentManagedEffectRequirement {
	return []cp.ComponentManagedEffectRequirement{{
		Class: IssueWorkflowRevisionEffectClass,
		Triggers: []cp.ComponentManagedMutationDisposition{
			cp.ComponentManagedMutationDesiredProjectionStaged, cp.ComponentManagedMutationSafeProjectionStaged,
		},
	}}
}

func issueWorkflowParentValues(packName string, desired issueWorkflowDesired) cp.ComponentRow {
	return cp.ComponentRow{"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey, "OwnerPack": packName, "Name": desired.Projection.Name}
}

func issueWorkflowStatusInsertRows(packName string, desired issueWorkflowDesired) []cp.ComponentManagedInsertRow {
	rows := make([]cp.ComponentManagedInsertRow, 0, len(desired.Projection.Statuses))
	for _, status := range desired.Projection.Statuses {
		key := cp.ComponentRow{"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey, "StatusKey": status.Key}
		rows = append(rows, cp.ComponentManagedInsertRow{LogicalKey: key, Values: cp.ComponentRow{
			"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey, "OwnerPack": packName, "StatusKey": status.Key, "Label": status.Label,
		}})
	}
	return rows
}

func issueWorkflowTransitionInsertRows(packName string, desired issueWorkflowDesired) []cp.ComponentManagedInsertRow {
	rows := make([]cp.ComponentManagedInsertRow, 0, len(desired.Projection.Transitions))
	for _, transition := range desired.Projection.Transitions {
		key := cp.ComponentRow{"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey, "TransitionKey": transition.Key}
		rows = append(rows, cp.ComponentManagedInsertRow{LogicalKey: key, Values: cp.ComponentRow{
			"ProjectKey": desired.ProjectKey, "WorkflowKey": desired.WorkflowKey, "OwnerPack": packName, "TransitionKey": transition.Key,
			"FromStatusKey": transition.From, "ToStatusKey": transition.To,
		}})
	}
	return rows
}

func insertIssueWorkflowRows(ctx context.Context, store cp.ComponentManagedMutationStore, table string, rows []cp.ComponentManagedInsertRow) error {
	if len(rows) == 0 {
		return fmt.Errorf("issue workflow insert into %s requires rows", table)
	}
	affected, err := store.InsertRows(ctx, cp.ComponentManagedInsertBatch{Table: table, Rows: rows})
	if err != nil {
		return err
	}
	if affected != int64(len(rows)) {
		return fmt.Errorf("issue workflow insert into %s affected %d rows, want %d", table, affected, len(rows))
	}
	return nil
}

func marshalIssueWorkflowAggregate(aggregate issueWorkflowAggregate) ([]byte, error) {
	if !aggregate.Present {
		return []byte("null"), nil
	}
	return json.Marshal(aggregate.Projection)
}

func issueWorkflowRowString(table string, row cp.ComponentRow, column string) (string, error) {
	value, ok := row[column].(string)
	if !ok || value == "" || strings.TrimSpace(value) != value {
		return "", fmt.Errorf("%s row column %s has non-canonical %T value", table, column, row[column])
	}
	return value, nil
}

func issueWorkflowRowBytes(table string, row cp.ComponentRow, column string) ([]byte, error) {
	value, ok := row[column].([]byte)
	if !ok || len(value) == 0 || len(value) > 255 {
		return nil, fmt.Errorf("%s row column %s has invalid %T value", table, column, row[column])
	}
	return append([]byte(nil), value...), nil
}

func validateIssueWorkflowRowTarget(table string, row cp.ComponentRow, projectKey, workflowKey []byte) error {
	rowProject, err := issueWorkflowRowBytes(table, row, "ProjectKey")
	if err != nil {
		return err
	}
	rowWorkflow, err := issueWorkflowRowBytes(table, row, "WorkflowKey")
	if err != nil {
		return err
	}
	if !bytes.Equal(rowProject, projectKey) || !bytes.Equal(rowWorkflow, workflowKey) {
		return fmt.Errorf("%s row escaped the requested binary target", table)
	}
	return nil
}

func validateIssueWorkflowRowOwner(table string, row cp.ComponentRow, packName string) error {
	owner, err := issueWorkflowRowString(table, row, "OwnerPack")
	if err != nil {
		return err
	}
	if owner != packName {
		return fmt.Errorf("%s row is owned by %q, not %q", table, owner, packName)
	}
	return nil
}

func boolInt(value bool) int {
	if value {
		return 1
	}
	return 0
}