WorkflowOperation.java

package org.hammer.audio.workflow;

import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/** Semantic workflow operation with deterministic apply and optional inverse operation. */
public sealed interface WorkflowOperation
    permits WorkflowOperation.CreateNode,
        WorkflowOperation.DeleteNode,
        WorkflowOperation.MoveNode,
        WorkflowOperation.RenameNode,
        WorkflowOperation.ConnectPorts,
        WorkflowOperation.DisconnectPorts,
        WorkflowOperation.UpdateProperty,
        WorkflowOperation.GroupNodes,
        WorkflowOperation.UngroupNodes,
        WorkflowOperation.RestoreNode {

  String NO_GROUP = "__none__";
  String UNDO_SUFFIX = ":undo";
  String NODE_ID = "nodeId";
  String WORKFLOW_FIELD = "workflow";
  String SOURCE = "source";
  String TARGET = "target";
  String CONNECT_PORTS = "ConnectPorts";

  String operationId();

  Instant timestamp();

  String author();

  List<String> affectedObjectIds();

  Map<String, String> payload();

  Optional<WorkflowOperation> inverseOperation();

  Workflow apply(Workflow workflow);

  static void requireOperationMetadata(
      String operationId, Instant timestamp, String author, String fieldName) {
    StableIds.requireStable(operationId, fieldName + ".operationId");
    Objects.requireNonNull(timestamp, fieldName + ".timestamp");
    if (author == null || author.isBlank()) {
      throw new IllegalArgumentException(fieldName + ".author must not be blank");
    }
  }

  static Node requireNode(Workflow workflow, String nodeId, String fieldName) {
    return workflow.nodes().stream()
        .filter(node -> node.id().equals(nodeId))
        .findFirst()
        .orElseThrow(() -> new IllegalArgumentException(fieldName + " node not found: " + nodeId));
  }

  static Edge requireEdge(Workflow workflow, String edgeId, String fieldName) {
    return workflow.edges().stream()
        .filter(edge -> edge.id().equals(edgeId))
        .findFirst()
        .orElseThrow(() -> new IllegalArgumentException(fieldName + " edge not found: " + edgeId));
  }

  static void requireNodeSnapshot(Workflow workflow, Node expectedNode, String fieldName) {
    Node actualNode = requireNode(workflow, expectedNode.id(), fieldName);
    if (!actualNode.equals(expectedNode)) {
      throw new IllegalStateException(
          fieldName + " expected node snapshot to match workflow: " + expectedNode.id());
    }
  }

  static void requireEdgeSnapshot(Workflow workflow, Edge expectedEdge, String fieldName) {
    Edge actualEdge = requireEdge(workflow, expectedEdge.id(), fieldName);
    if (!actualEdge.equals(expectedEdge)) {
      throw new IllegalStateException(
          fieldName + " expected edge snapshot to match workflow: " + expectedEdge.id());
    }
  }

  static void requirePort(Node node, String portId, boolean inputPort, String fieldName) {
    List<Port> ports = inputPort ? node.inputPorts() : node.outputPorts();
    boolean exists = ports.stream().anyMatch(port -> port.id().equals(portId));
    if (!exists) {
      String endpoint = inputPort ? TARGET : SOURCE;
      throw new IllegalArgumentException(
          fieldName + " missing " + endpoint + " port: " + node.id() + ":" + portId);
    }
  }

  static void requireNodeIds(Workflow workflow, List<String> nodeIds, String fieldName) {
    for (String nodeId : nodeIds) {
      requireNode(workflow, nodeId, fieldName);
    }
  }

  static void requireConnectedEdges(
      Workflow workflow, String nodeId, List<Edge> expectedEdges, String fieldName) {
    Set<String> actualEdgeIds = new LinkedHashSet<>();
    for (Edge edge : workflow.edges()) {
      if (edge.sourceNodeId().equals(nodeId) || edge.targetNodeId().equals(nodeId)) {
        actualEdgeIds.add(edge.id());
      }
    }
    Set<String> expectedEdgeIds = new LinkedHashSet<>();
    for (Edge edge : expectedEdges) {
      expectedEdgeIds.add(edge.id());
    }
    if (!actualEdgeIds.equals(expectedEdgeIds)) {
      throw new IllegalStateException(
          fieldName + " deletedEdges must match connected edges for node " + nodeId);
    }
    for (Edge edge : expectedEdges) {
      requireEdgeSnapshot(workflow, edge, fieldName);
    }
  }

  static Metadata updateMetadata(Metadata metadata, String key, String value) {
    StableIds.requireStable(key, "property key");
    Map<String, String> entries = new ConcurrentHashMap<>(metadata.entries());
    if (value == null) {
      entries.remove(key);
    } else {
      entries.put(key, value);
    }
    return entries.isEmpty() ? Metadata.empty() : new Metadata(entries);
  }

  static Workflow replaceNode(Workflow workflow, Node replacementNode) {
    List<Node> nodes = new ArrayList<>(workflow.nodes().size());
    boolean replaced = false;
    for (Node node : workflow.nodes()) {
      if (node.id().equals(replacementNode.id())) {
        nodes.add(replacementNode);
        replaced = true;
      } else {
        nodes.add(node);
      }
    }
    if (!replaced) {
      throw new IllegalArgumentException("node not found: " + replacementNode.id());
    }
    return new Workflow(
        workflow.id(), workflow.name(), nodes, workflow.edges(), workflow.metadata());
  }

  static Map<String, String> requiredPreviousGroups(
      List<String> nodeIds, Map<String, String> previousGroupIds, String fieldName) {
    Objects.requireNonNull(previousGroupIds, fieldName + ".previousGroupIds");
    Map<String, String> normalizedGroups = new ConcurrentHashMap<>();
    for (String nodeId : nodeIds) {
      if (!previousGroupIds.containsKey(nodeId)) {
        throw new IllegalArgumentException(
            fieldName + ".previousGroupIds must contain node id " + nodeId);
      }
      String previousGroupId = previousGroupIds.get(nodeId);
      normalizedGroups.put(nodeId, previousGroupId == null ? NO_GROUP : previousGroupId);
    }
    return Map.copyOf(normalizedGroups);
  }

  static List<String> affectedIdsForDelete(Node deletedNode, List<Edge> deletedEdges) {
    List<String> affectedIds = new ArrayList<>();
    affectedIds.add(deletedNode.id());
    for (Edge deletedEdge : deletedEdges) {
      affectedIds.add(deletedEdge.id());
    }
    return List.copyOf(affectedIds);
  }

  /**
   * Operation that adds a new node to the workflow.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param node node to add
   */
  record CreateNode(String operationId, Instant timestamp, String author, Node node)
      implements WorkflowOperation {

    public CreateNode {
      requireOperationMetadata(operationId, timestamp, author, "CreateNode");
      Objects.requireNonNull(node, "node");
    }

    @Override
    public List<String> affectedObjectIds() {
      return List.of(node.id());
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(NODE_ID, node.id(), "nodeType", node.type(), "label", node.label());
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new DeleteNode(
              operationId + UNDO_SUFFIX, timestamp, author, node, List.of(), List.of(node.id())));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      Set<String> nodeIds = new LinkedHashSet<>();
      for (Node existingNode : workflow.nodes()) {
        nodeIds.add(existingNode.id());
      }
      if (nodeIds.contains(node.id())) {
        throw new IllegalArgumentException("node already exists: " + node.id());
      }
      List<Node> nodes = new ArrayList<>(workflow.nodes());
      nodes.add(node);
      return new Workflow(
          workflow.id(), workflow.name(), nodes, workflow.edges(), workflow.metadata());
    }
  }

  /**
   * Operation that removes a node and its related edges from the workflow.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param deletedNode node snapshot to remove
   * @param deletedEdges edge snapshots to remove
   * @param affectedObjectIds affected node and edge ids
   */
  record DeleteNode(
      String operationId,
      Instant timestamp,
      String author,
      Node deletedNode,
      List<Edge> deletedEdges,
      List<String> affectedObjectIds)
      implements WorkflowOperation {

    public DeleteNode {
      requireOperationMetadata(operationId, timestamp, author, "DeleteNode");
      Objects.requireNonNull(deletedNode, "deletedNode");
      Objects.requireNonNull(deletedEdges, "deletedEdges");
      deletedEdges = List.copyOf(deletedEdges);
      List<String> derivedAffectedObjectIds = affectedIdsForDelete(deletedNode, deletedEdges);
      if (derivedAffectedObjectIds.equals(affectedObjectIds)) {
        affectedObjectIds = List.copyOf(affectedObjectIds);
      } else {
        affectedObjectIds = derivedAffectedObjectIds;
      }
    }

    @Override
    public List<String> affectedObjectIds() {
      return List.copyOf(affectedObjectIds);
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(
          NODE_ID, deletedNode.id(), "deletedEdgeCount", String.valueOf(deletedEdges.size()));
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new RestoreNode(operationId + UNDO_SUFFIX, timestamp, author, deletedNode, deletedEdges));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      requireNodeSnapshot(workflow, deletedNode, "DeleteNode");
      requireConnectedEdges(workflow, deletedNode.id(), deletedEdges, "DeleteNode");

      List<Node> nodes =
          workflow.nodes().stream().filter(node -> !node.id().equals(deletedNode.id())).toList();
      Set<String> deletedEdgeIds = new LinkedHashSet<>();
      for (Edge edge : deletedEdges) {
        deletedEdgeIds.add(edge.id());
      }
      List<Edge> edges =
          workflow.edges().stream().filter(edge -> !deletedEdgeIds.contains(edge.id())).toList();
      return new Workflow(workflow.id(), workflow.name(), nodes, edges, workflow.metadata());
    }
  }

  /**
   * Operation that updates node layout coordinates in metadata.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param nodeId target node id
   * @param fromX previous x coordinate
   * @param fromY previous y coordinate
   * @param toX new x coordinate
   * @param toY new y coordinate
   */
  record MoveNode(
      String operationId,
      Instant timestamp,
      String author,
      String nodeId,
      double fromX,
      double fromY,
      double toX,
      double toY)
      implements WorkflowOperation {

    private static final String X_KEY = "layout.x";
    private static final String Y_KEY = "layout.y";

    public MoveNode {
      requireOperationMetadata(operationId, timestamp, author, "MoveNode");
      StableIds.requireStable(nodeId, "MoveNode.nodeId");
    }

    @Override
    public List<String> affectedObjectIds() {
      return List.of(nodeId);
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(
          NODE_ID,
          nodeId,
          "fromX",
          String.valueOf(fromX),
          "fromY",
          String.valueOf(fromY),
          "toX",
          String.valueOf(toX),
          "toY",
          String.valueOf(toY));
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new MoveNode(
              operationId + UNDO_SUFFIX, timestamp, author, nodeId, toX, toY, fromX, fromY));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      Node node = requireNode(workflow, nodeId, "MoveNode");
      Metadata movedMetadata =
          updateMetadata(
              updateMetadata(node.metadata(), X_KEY, String.valueOf(toX)),
              Y_KEY,
              String.valueOf(toY));
      return replaceNode(
          workflow,
          new Node(
              node.id(),
              node.type(),
              node.label(),
              node.inputPorts(),
              node.outputPorts(),
              movedMetadata));
    }
  }

  /**
   * Operation that renames a node label.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param nodeId target node id
   * @param previousLabel expected previous label
   * @param newLabel new label
   */
  record RenameNode(
      String operationId,
      Instant timestamp,
      String author,
      String nodeId,
      String previousLabel,
      String newLabel)
      implements WorkflowOperation {

    public RenameNode {
      requireOperationMetadata(operationId, timestamp, author, "RenameNode");
      StableIds.requireStable(nodeId, "RenameNode.nodeId");
      if (previousLabel == null || previousLabel.isBlank()) {
        throw new IllegalArgumentException("RenameNode.previousLabel must not be blank");
      }
      if (newLabel == null || newLabel.isBlank()) {
        throw new IllegalArgumentException("RenameNode.newLabel must not be blank");
      }
    }

    @Override
    public List<String> affectedObjectIds() {
      return List.of(nodeId);
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(NODE_ID, nodeId, "from", previousLabel, "to", newLabel);
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new RenameNode(
              operationId + UNDO_SUFFIX, timestamp, author, nodeId, newLabel, previousLabel));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      Node node = requireNode(workflow, nodeId, "RenameNode");
      if (!node.label().equals(previousLabel)) {
        throw new IllegalStateException(
            "RenameNode expected label " + previousLabel + " but was " + node.label());
      }
      return replaceNode(
          workflow,
          new Node(
              node.id(),
              node.type(),
              newLabel,
              node.inputPorts(),
              node.outputPorts(),
              node.metadata()));
    }
  }

  /**
   * Operation that creates an edge connection between ports.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param edge edge to create
   */
  record ConnectPorts(String operationId, Instant timestamp, String author, Edge edge)
      implements WorkflowOperation {

    public ConnectPorts {
      requireOperationMetadata(operationId, timestamp, author, CONNECT_PORTS);
      Objects.requireNonNull(edge, "edge");
    }

    @Override
    public List<String> affectedObjectIds() {
      return List.of(edge.sourceNodeId(), edge.targetNodeId(), edge.id());
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(
          "edgeId",
          edge.id(),
          SOURCE,
          edge.sourceNodeId() + ":" + edge.sourcePortId(),
          TARGET,
          edge.targetNodeId() + ":" + edge.targetPortId());
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new DisconnectPorts(operationId + UNDO_SUFFIX, timestamp, author, edge.id(), edge));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      Node sourceNode = requireNode(workflow, edge.sourceNodeId(), CONNECT_PORTS);
      Node targetNode = requireNode(workflow, edge.targetNodeId(), CONNECT_PORTS);
      requirePort(sourceNode, edge.sourcePortId(), false, CONNECT_PORTS);
      requirePort(targetNode, edge.targetPortId(), true, CONNECT_PORTS);
      boolean edgeExists =
          workflow.edges().stream().anyMatch(existingEdge -> existingEdge.id().equals(edge.id()));
      if (edgeExists) {
        throw new IllegalArgumentException("edge already exists: " + edge.id());
      }
      List<Edge> edges = new ArrayList<>(workflow.edges());
      edges.add(edge);
      return new Workflow(
          workflow.id(), workflow.name(), workflow.nodes(), edges, workflow.metadata());
    }
  }

  /**
   * Operation that removes an existing edge connection.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param edgeId edge id to remove
   * @param disconnectedEdge edge snapshot used for undo
   */
  record DisconnectPorts(
      String operationId, Instant timestamp, String author, String edgeId, Edge disconnectedEdge)
      implements WorkflowOperation {

    public DisconnectPorts {
      requireOperationMetadata(operationId, timestamp, author, "DisconnectPorts");
      StableIds.requireStable(edgeId, "DisconnectPorts.edgeId");
      Objects.requireNonNull(disconnectedEdge, "disconnectedEdge");
      if (!edgeId.equals(disconnectedEdge.id())) {
        throw new IllegalArgumentException("DisconnectPorts.edgeId must match disconnectedEdge.id");
      }
    }

    @Override
    public List<String> affectedObjectIds() {
      return List.of(disconnectedEdge.sourceNodeId(), disconnectedEdge.targetNodeId(), edgeId);
    }

    @Override
    public Map<String, String> payload() {
      return Map.of("edgeId", edgeId);
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new ConnectPorts(operationId + UNDO_SUFFIX, timestamp, author, disconnectedEdge));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      requireEdgeSnapshot(workflow, disconnectedEdge, "DisconnectPorts");
      List<Edge> edges =
          workflow.edges().stream().filter(edge -> !edge.id().equals(edgeId)).toList();
      return new Workflow(
          workflow.id(), workflow.name(), workflow.nodes(), edges, workflow.metadata());
    }
  }

  /**
   * Operation that updates metadata properties on workflow, node or edge targets.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param target target object type
   * @param targetId target object id
   * @param propertyKey metadata key
   * @param previousValue previous property value
   * @param newValue new property value
   */
  record UpdateProperty(
      String operationId,
      Instant timestamp,
      String author,
      PropertyTarget target,
      String targetId,
      String propertyKey,
      String previousValue,
      String newValue)
      implements WorkflowOperation {

    public UpdateProperty {
      requireOperationMetadata(operationId, timestamp, author, "UpdateProperty");
      Objects.requireNonNull(target, TARGET);
      StableIds.requireStable(targetId, "UpdateProperty.targetId");
      StableIds.requireStable(propertyKey, "UpdateProperty.propertyKey");
    }

    @Override
    public List<String> affectedObjectIds() {
      return List.of(targetId);
    }

    @Override
    public Map<String, String> payload() {
      Map<String, String> payload = new ConcurrentHashMap<>();
      payload.put(TARGET, target.name());
      payload.put("targetId", targetId);
      payload.put("propertyKey", propertyKey);
      payload.put("newValue", String.valueOf(newValue));
      payload.put("previousValue", String.valueOf(previousValue));
      return Map.copyOf(payload);
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new UpdateProperty(
              operationId + UNDO_SUFFIX,
              timestamp,
              author,
              target,
              targetId,
              propertyKey,
              newValue,
              previousValue));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      return switch (target) {
        case WORKFLOW -> {
          if (!workflow.id().equals(targetId)) {
            throw new IllegalArgumentException(
                "workflow id mismatch for UpdateProperty: " + targetId);
          }
          Metadata metadata = updateMetadata(workflow.metadata(), propertyKey, newValue);
          yield new Workflow(
              workflow.id(), workflow.name(), workflow.nodes(), workflow.edges(), metadata);
        }
        case NODE -> {
          Node node = requireNode(workflow, targetId, "UpdateProperty");
          Metadata metadata = updateMetadata(node.metadata(), propertyKey, newValue);
          yield replaceNode(
              workflow,
              new Node(
                  node.id(),
                  node.type(),
                  node.label(),
                  node.inputPorts(),
                  node.outputPorts(),
                  metadata));
        }
        case EDGE -> {
          Edge edge = requireEdge(workflow, targetId, "UpdateProperty");
          Metadata metadata = updateMetadata(edge.metadata(), propertyKey, newValue);
          List<Edge> edges = new ArrayList<>(workflow.edges().size());
          for (Edge existingEdge : workflow.edges()) {
            if (existingEdge.id().equals(edge.id())) {
              edges.add(
                  new Edge(
                      edge.id(),
                      edge.sourceNodeId(),
                      edge.sourcePortId(),
                      edge.targetNodeId(),
                      edge.targetPortId(),
                      metadata));
            } else {
              edges.add(existingEdge);
            }
          }
          yield new Workflow(
              workflow.id(), workflow.name(), workflow.nodes(), edges, workflow.metadata());
        }
      };
    }
  }

  /** Target object type for {@link UpdateProperty}. */
  enum PropertyTarget {
    WORKFLOW,
    NODE,
    EDGE
  }

  /**
   * Operation that assigns nodes to a semantic group.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param groupId target group id
   * @param groupLabel human-readable group label
   * @param nodeIds grouped node ids
   * @param previousGroupIds previous group id per node
   */
  record GroupNodes(
      String operationId,
      Instant timestamp,
      String author,
      String groupId,
      String groupLabel,
      List<String> nodeIds,
      Map<String, String> previousGroupIds)
      implements WorkflowOperation {

    private static final String GROUP_KEY = "group.id";
    private static final String GROUP_LABEL_PREFIX = "group.";

    public GroupNodes {
      requireOperationMetadata(operationId, timestamp, author, "GroupNodes");
      StableIds.requireStable(groupId, "GroupNodes.groupId");
      if (groupLabel == null || groupLabel.isBlank()) {
        throw new IllegalArgumentException("GroupNodes.groupLabel must not be blank");
      }
      Objects.requireNonNull(nodeIds, "nodeIds");
      if (nodeIds.isEmpty()) {
        throw new IllegalArgumentException("GroupNodes.nodeIds must not be empty");
      }
      nodeIds = List.copyOf(nodeIds);
      previousGroupIds = requiredPreviousGroups(nodeIds, previousGroupIds, "GroupNodes");
    }

    @Override
    public Map<String, String> previousGroupIds() {
      return Map.copyOf(previousGroupIds);
    }

    @Override
    public List<String> affectedObjectIds() {
      return nodeIds;
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(
          "groupId",
          groupId,
          "groupLabel",
          groupLabel,
          "nodeCount",
          String.valueOf(nodeIds.size()));
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new UngroupNodes(
              operationId + UNDO_SUFFIX,
              timestamp,
              author,
              groupId,
              groupLabel,
              nodeIds,
              previousGroupIds));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      requireNodeIds(workflow, nodeIds, "GroupNodes");
      Set<String> nodeSet = new LinkedHashSet<>(nodeIds);
      List<Node> nodes = new ArrayList<>(workflow.nodes().size());
      for (Node node : workflow.nodes()) {
        if (nodeSet.contains(node.id())) {
          Metadata metadata = updateMetadata(node.metadata(), GROUP_KEY, groupId);
          nodes.add(
              new Node(
                  node.id(),
                  node.type(),
                  node.label(),
                  node.inputPorts(),
                  node.outputPorts(),
                  metadata));
        } else {
          nodes.add(node);
        }
      }
      String groupLabelKey = GROUP_LABEL_PREFIX + groupId + ".label";
      Metadata workflowMetadata = updateMetadata(workflow.metadata(), groupLabelKey, groupLabel);
      return new Workflow(
          workflow.id(), workflow.name(), nodes, workflow.edges(), workflowMetadata);
    }
  }

  /**
   * Operation that removes nodes from a semantic group.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param groupId group id being removed
   * @param groupLabel group label to preserve for undo
   * @param nodeIds node ids to ungroup
   * @param previousGroupIds previous group id per node
   */
  record UngroupNodes(
      String operationId,
      Instant timestamp,
      String author,
      String groupId,
      String groupLabel,
      List<String> nodeIds,
      Map<String, String> previousGroupIds)
      implements WorkflowOperation {

    private static final String GROUP_KEY = "group.id";
    private static final String GROUP_LABEL_PREFIX = "group.";

    public UngroupNodes {
      requireOperationMetadata(operationId, timestamp, author, "UngroupNodes");
      StableIds.requireStable(groupId, "UngroupNodes.groupId");
      if (groupLabel == null || groupLabel.isBlank()) {
        throw new IllegalArgumentException("UngroupNodes.groupLabel must not be blank");
      }
      Objects.requireNonNull(nodeIds, "nodeIds");
      if (nodeIds.isEmpty()) {
        throw new IllegalArgumentException("UngroupNodes.nodeIds must not be empty");
      }
      nodeIds = List.copyOf(nodeIds);
      previousGroupIds = requiredPreviousGroups(nodeIds, previousGroupIds, "UngroupNodes");
    }

    @Override
    public Map<String, String> previousGroupIds() {
      return Map.copyOf(previousGroupIds);
    }

    @Override
    public List<String> affectedObjectIds() {
      return nodeIds;
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(
          "groupId",
          groupId,
          "groupLabel",
          groupLabel,
          "nodeCount",
          String.valueOf(nodeIds.size()));
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      return Optional.of(
          new GroupNodes(
              operationId + UNDO_SUFFIX,
              timestamp,
              author,
              groupId,
              groupLabel,
              nodeIds,
              previousGroupIds));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      requireNodeIds(workflow, nodeIds, "UngroupNodes");
      Set<String> nodeSet = new LinkedHashSet<>(nodeIds);
      List<Node> nodes = new ArrayList<>(workflow.nodes().size());
      for (Node node : workflow.nodes()) {
        if (nodeSet.contains(node.id())) {
          String previousGroupId = previousGroupIds.get(node.id());
          String restoredGroupId = NO_GROUP.equals(previousGroupId) ? null : previousGroupId;
          Metadata metadata = updateMetadata(node.metadata(), GROUP_KEY, restoredGroupId);
          nodes.add(
              new Node(
                  node.id(),
                  node.type(),
                  node.label(),
                  node.inputPorts(),
                  node.outputPorts(),
                  metadata));
        } else {
          nodes.add(node);
        }
      }
      String groupLabelKey = GROUP_LABEL_PREFIX + groupId + ".label";
      boolean hasRemainingGroupMembers =
          nodes.stream().anyMatch(node -> groupId.equals(node.metadata().entries().get(GROUP_KEY)));
      Metadata workflowMetadata =
          hasRemainingGroupMembers
              ? workflow.metadata()
              : updateMetadata(workflow.metadata(), groupLabelKey, null);
      return new Workflow(
          workflow.id(), workflow.name(), nodes, workflow.edges(), workflowMetadata);
    }
  }

  /**
   * Internal restore operation used as inverse for {@link DeleteNode}.
   *
   * @param operationId unique operation id
   * @param timestamp operation timestamp
   * @param author operation author
   * @param restoredNode node snapshot to restore
   * @param restoredEdges edge snapshots to restore
   */
  record RestoreNode(
      String operationId,
      Instant timestamp,
      String author,
      Node restoredNode,
      List<Edge> restoredEdges)
      implements WorkflowOperation {

    public RestoreNode {
      requireOperationMetadata(operationId, timestamp, author, "RestoreNode");
      Objects.requireNonNull(restoredNode, "restoredNode");
      Objects.requireNonNull(restoredEdges, "restoredEdges");
      restoredEdges = List.copyOf(restoredEdges);
    }

    @Override
    public List<String> affectedObjectIds() {
      List<String> ids = new ArrayList<>();
      ids.add(restoredNode.id());
      for (Edge edge : restoredEdges) {
        ids.add(edge.id());
      }
      return List.copyOf(ids);
    }

    @Override
    public Map<String, String> payload() {
      return Map.of(
          NODE_ID, restoredNode.id(), "restoredEdgeCount", String.valueOf(restoredEdges.size()));
    }

    @Override
    public Optional<WorkflowOperation> inverseOperation() {
      List<String> affectedIds = new ArrayList<>();
      affectedIds.add(restoredNode.id());
      for (Edge edge : restoredEdges) {
        affectedIds.add(edge.id());
      }
      return Optional.of(
          new DeleteNode(
              operationId + UNDO_SUFFIX,
              timestamp,
              author,
              restoredNode,
              restoredEdges,
              affectedIds));
    }

    @Override
    public Workflow apply(Workflow workflow) {
      Objects.requireNonNull(workflow, WORKFLOW_FIELD);
      boolean nodeExists =
          workflow.nodes().stream().anyMatch(node -> node.id().equals(restoredNode.id()));
      if (nodeExists) {
        throw new IllegalArgumentException("node already exists: " + restoredNode.id());
      }
      List<Node> nodes = new ArrayList<>(workflow.nodes());
      nodes.add(restoredNode);

      Set<String> edgeIds = new LinkedHashSet<>();
      for (Edge edge : workflow.edges()) {
        edgeIds.add(edge.id());
      }
      List<Edge> edges = new ArrayList<>(workflow.edges());
      for (Edge edge : restoredEdges) {
        if (!edgeIds.contains(edge.id())) {
          edges.add(edge);
        }
      }
      return new Workflow(workflow.id(), workflow.name(), nodes, edges, workflow.metadata());
    }
  }
}