// SPDX-License-Identifier: BUSL 1.1
pragma solidity 0.8.26;

import {ITypeAndVersion} from "../../../shared/interfaces/ITypeAndVersion.sol";

import {Ownable2StepMsgSender} from "../../../shared/access/Ownable2StepMsgSender.sol";

import {ECDSA} from "@openzeppelin/contracts@5.1.0/utils/cryptography/ECDSA.sol";
import {MessageHashUtils} from "@openzeppelin/contracts@5.1.0/utils/cryptography/MessageHashUtils.sol";
import {EnumerableMap} from "@openzeppelin/contracts@5.1.0/utils/structs/EnumerableMap.sol";
import {EnumerableSet} from "@openzeppelin/contracts@5.1.0/utils/structs/EnumerableSet.sol";

// solhint-disable-next-line max-states-count
contract WorkflowRegistry is Ownable2StepMsgSender, ITypeAndVersion {
  using EnumerableSet for EnumerableSet.Bytes32Set;
  using EnumerableSet for EnumerableSet.AddressSet;
  using EnumerableMap for EnumerableMap.AddressToBytes32Map;

  string public constant override typeAndVersion = "WorkflowRegistry 2.0.0-dev";
  /// @dev Default values for contract configuration.
  uint8 private constant DEFAULT_MAX_NAME_LEN = 64;
  uint8 private constant DEFAULT_MAX_TAG_LEN = 32;
  uint8 private constant DEFAULT_MAX_URL_LEN = 200;
  uint16 private constant DEFAULT_MAX_ATTR_LEN = 1024;
  uint32 private constant DEFAULT_MAX_EXPIRY = 604_800; // one week

  /// @dev Configuration struct that keeps config parameters for this contract.
  Config private s_config = Config({
    maxNameLen: DEFAULT_MAX_NAME_LEN,
    maxTagLen: DEFAULT_MAX_TAG_LEN,
    maxUrlLen: DEFAULT_MAX_URL_LEN,
    maxAttrLen: DEFAULT_MAX_ATTR_LEN,
    maxExpiryLen: DEFAULT_MAX_EXPIRY
  });

  /// @dev The set of allowed signers for ownership proofs. These signers are considered to be trusted entities.
  /// If the ownership proof is not signed by one of the allowed signers, signature will be rejected.
  mapping(address signer => bool allowed) private s_allowedSigners;
  /// @dev The set of linked owners. These are the addresses that have successfully linked their ownership proofs.
  /// Ownership proofs are signed messages generated by a trusted entity, with each proof being unique for each owner
  /// address. The owner address is embedded into the proof itself together with other relevant metadata that uniquely
  /// represents an off-chain account.
  /// Fundamental assumption is that only a person (or a group of people) who has access both to the private key of the
  /// owner
  /// address and to the off-chain account, may be able to generate a valid signature signed by the trusted entity. Once
  /// this
  /// valid signature is submitted to this contract, it can be verified and used to link or unlink the owner address.
  EnumerableMap.AddressToBytes32Map private s_linkedOwners;
  /// @dev This is a mapping of ownership proofs indicating whether the proof has been previously used or not. This is
  /// used
  /// to prevent someone from re-using the same proof for linking more than once, and ensures that each proof is unique
  /// per
  /// single linking request, not matter if it originates from the same owner address or not. This allows us to
  /// verifiably
  /// enforce an invariant on proofs.
  mapping(bytes32 proof => bool used) private s_usedProofs;

  /// @dev workflowRid (reference ID) is a hash of (owner ∥ name ∥ label). It functions as the primary index for the
  /// workflow storage.
  mapping(bytes32 workflowRid => WorkflowMetadata workflowMetadata) private s_workflows;
  /// @dev workflowKey is a hash of (owner ∥ name ) and maps to a set of Rids. It is used for filtering workflows
  /// based on just
  /// the name, as there may be multiple workflows with the same name under a different tag.
  mapping(bytes32 workflowKey => EnumerableSet.Bytes32Set) private s_workflowKeyToRids;
  /// @dev workflowId ⇒ workflowRid. This mapping lets us enforce global uniqueness of workflowId while also allows us
  /// to lookup
  /// workflows through the workflowId.
  mapping(bytes32 workflowId => bytes32 workflowRid) private s_idToRid;

  // Secondary indices for iteration / queries
  mapping(address owner => EnumerableSet.Bytes32Set workflowRids) private s_activeOwnerWorkflowRids; // owner ->
    // workflowRid set
  mapping(bytes32 donHash => EnumerableSet.Bytes32Set workflowRids) private s_activeDONWorkflowRids; // donHash ->
    // workflowRid set

  /// @dev Every workflow ever registered under `donFamily` as donHash. Pruned only on delete.
  mapping(bytes32 donHash => EnumerableSet.Bytes32Set workflowRids) private s_allDONRids;
  /// @dev Every workflow ever registered for an owner. Pruned only on delete.
  mapping(address owner => EnumerableSet.Bytes32Set workflowRids) private s_allOwnerRids;
  mapping(bytes32 workflowKey => EnumerableSet.Bytes32Set activeRids) private s_activeRidsByWorkflowKey; // workflowKey
    // → active Rids
  /// @dev Fast counters for limits enforcement
  mapping(address owner => mapping(bytes32 donHash => uint32 workflowCount)) private s_userDONCount; // owner ->
    // (donHash -> #workflows)
  /// @dev The don family (as a hash) that the workflow is assigned to. Only active workflows are assigned don families.
  /// When a workflow is paused it is removed from the don family.
  mapping(bytes32 rid => bytes32 donHash) private s_donByWorkflowRid;

  /// @dev Tracking allowlisted requests for the owner address, required to enable anyone to verify off-chain requests.
  mapping(bytes32 ownerDigestHash => uint32 expiryTimestamp) private s_requestsAllowlist;
  /// @dev Storing allowlisted requests for all owners, enabling fetching all non-expired requests
  OwnerAllowlistedRequest[] private s_requestAllowlistArray;
  /// @dev Map each owner address to their arbitrary config. Can be used to control billing parameters or any other data
  /// per owner
  mapping(address owner => bytes config) private s_ownerConfig;

  /// @dev DON configs storage.
  mapping(bytes32 donHash => DonConfig donCfg) private s_donConfigs;
  /// @dev Tracks every DON hash that has ever been configured via `setDONLimit`.
  EnumerableSet.Bytes32Set private s_donConfigKeys;
  /// Tracks every user that currently has an override for a specific DON.
  mapping(bytes32 donHash => EnumerableSet.AddressSet userOverrides) private s_donOverrideUsers;

  /// @dev Stores the current Capabilities Registry reference used by this contract.
  CapabilitiesRegistryConfig private s_capabilitiesRegistry;
  /// @dev Storage of all capacity and workflow life‑cycle events
  EventRecord[] private s_events;

  // ================================================================
  // |                         Events                               |
  // ================================================================

  event AllowedSignersUpdated(address[] signers, bool allowed);
  event OwnershipLinkUpdated(address indexed owner, bytes32 indexed proof, bool indexed added);
  event DONLimitSet(string donFamily, uint32 limit);
  event UserDONLimitSet(address indexed user, string donFamily, uint32 limit);
  event UserDONLimitUnset(address indexed user, string donFamily);
  event WorkflowRegistered(
    bytes32 indexed workflowId, address indexed owner, string donFamily, WorkflowStatus status, string workflowName
  );
  event WorkflowUpdated(
    bytes32 indexed oldWorkflowId,
    bytes32 indexed newWorkflowId,
    address indexed owner,
    string donFamily,
    string workflowName
  );
  event WorkflowPaused(bytes32 indexed workflowId, address indexed owner, string donFamily, string workflowName);
  event WorkflowActivated(bytes32 indexed workflowId, address indexed owner, string donFamily, string workflowName);
  event WorkflowDeleted(bytes32 indexed workflowId, address indexed owner, string donFamily, string workflowName);
  /// @dev Fired whenever a workflow’s DON family is changed
  event WorkflowDonFamilyUpdated(
    bytes32 indexed workflowId, address indexed owner, string oldDonFamily, string newDonFamily
  );
  event RequestAllowlisted(address indexed owner, bytes32 indexed requestDigest, uint32 expiryTimestamp);
  /// @notice Emitted when metadata length limits are updated
  event ConfigUpdated(uint8 maxNameLen, uint8 maxTagLen, uint8 maxUrlLen, uint16 maxAttrLen, uint32 maxExpiryLen);
  /// @notice Emitted when a workflow owner’s config is updated
  event WorkflowOwnerConfigUpdated(address indexed owner, bytes config);
  /// @notice Emitted whenever the registry reference is changed.
  event CapabilitiesRegistryUpdated(address oldAddr, address newAddr, uint64 oldChainSelector, uint64 newChainSelector);

  // ================================================================
  // |                         Errors                               |
  // ================================================================
  error ZeroAddressNotAllowed();
  error ZeroWorkflowIDNotAllowed();
  error LinkOwnerRequestExpired(address caller, uint256 currentTime, uint256 expiryTimestamp);
  error UnlinkOwnerRequestExpired(address caller, uint256 currentTime, uint256 expiryTimestamp);
  error OwnershipLinkAlreadyExists(address owner);
  error OwnershipLinkDoesNotExist(address owner);
  error InvalidSignature(bytes signature, uint8 recoverErrorId, bytes32 recoverErrorArg);
  error InvalidOwnershipLink(address owner, uint256 validityTimestamp, bytes32 proof, bytes signature);
  error OwnershipProofAlreadyUsed(address caller, bytes32 proof);

  error CallerIsNotWorkflowOwner(address caller);
  error DonLimitNotSet(string donFamily);
  error MaxWorkflowsPerUserDONExceeded(address owner, string donFamily);
  error UserDONOverrideExceedsDONLimit();
  error URLTooLong(uint256 provided, uint8 maxAllowed);
  error WorkflowDoesNotExist();
  error WorkflowIDAlreadyExists(bytes32 workflowId);
  error WorkflowNameRequired();
  error WorkflowNameTooLong(uint256 provided, uint8 maxAllowed);
  error WorkflowTagRequired();
  error WorkflowTagTooLong(uint256 provided, uint8 maxAllowed);
  error AttributesTooLong(uint256 provided, uint256 maxAllowed);
  error EmptyUpdateBatch();
  error BinaryURLRequired();
  error CannotUpdateDONFamilyForPausedWorkflows();
  error InvalidExpiryTimestamp(bytes32 requestDigest, uint32 expiryTimestamp, uint32 maxAllowed);

  // ================================================================
  // |                         Enums                                |
  // ================================================================
  enum WorkflowStatus {
    ACTIVE,
    PAUSED
  }

  enum LinkingRequestType {
    LINK_OWNER, //       Request to link an owner address.
    UNLINK_OWNER //       Request to unlink an owner address.

  }

  enum EventType {
    DONCapacitySet,
    WorkflowAdded,
    WorkflowRemoved
  }

  // ================================================================
  // |                         Structs                              |
  // ================================================================
  /// @dev Struct for Workflow Registry configuration parameters
  struct Config {
    uint8 maxNameLen; // Cap for `workflowName` (0 ➜ unlimited)
    uint8 maxTagLen; // Cap for `tag` (0 ➜ unlimited)
    uint8 maxUrlLen; // Cap for each URL (0 ➜ unlimited)
    uint16 maxAttrLen; // Cap for `attributes` (0 ➜ unlimited)
    uint32 maxExpiryLen; // Cap for every allowlisted request expiration timestamp (0 ➜ unlimited)
  }

  /// @dev Struct for WorkflowMetadata. This is used to store the workflow metadata.
  struct WorkflowMetadata {
    bytes32 workflowId; //       Unique identifier from hash(owner, workflow name, wasm binary, cfg).
    address owner; // ─────────╮ Workflow owner.
    uint64 createdAt; //       │ block.timestamp when the workflow was first registered.
    WorkflowStatus status; // ─╯ Current status of the workflow (active, paused).
    string workflowName; //      Human readable string (64 chars limit).
    string binaryUrl; //         URL to the wasm binary (200 chars limit).
    string configUrl; //         URL to the config (200 chars limit).
    string tag; //               Unique per (owner, workflowName) human readable identifier (32 chars limit)
    bytes attributes; //         Arbitrary bytes for additional workflow details.
  }
  /// @dev View struct for WorkflowMetadata. This is used to return the workflow metadata including the donFamily.

  struct WorkflowMetadataView {
    bytes32 workflowId;
    address owner;
    uint64 createdAt;
    WorkflowStatus status;
    string workflowName;
    string binaryUrl;
    string configUrl;
    string tag;
    bytes attributes;
    string donFamily;
  }

  /// @dev ConfigValue struct to distinguish between unset and explicitly set zero values
  struct ConfigValue {
    uint32 value;
    bool enabled;
  }

  /// @dev Configuration struct for don capacity of active workflows. Paused workflows are ignored.
  ///      One config blob per DON (keyed by _hash(donFamily))
  struct DonConfig {
    /// @dev Human-readable DON label (e.g. “fast-pool”)
    string family;
    /// @dev Global cap for active workflows on this DON
    ConfigValue limitValue;
    /// @dev Optional per-user overrides for this DON
    mapping(address => ConfigValue) userOverride;
  }

  /// @notice Lightweight, return-able view of a DON config.
  /// @dev    Cannot embed the per-user `mapping`, so we expose only the global cap. To get the per-user overrides,
  ///         use `getUserDONOverrides` on the specific DON family.
  struct DonConfigView {
    bytes32 donHash; // keccak256(family).
    string family; // Human-readable DON label.
    uint32 limit; // Global ACTIVE-workflow cap
    bool limitEnabled;
  }

  /// @dev Return-able view of a per-user override for a specific DON family.
  struct UserOverrideView {
    address user;
    uint32 limit;
  }

  /// @notice  CapabilitiesRegistryConfig struct stores the pointer to the Capabilities Registry this Workflow Registry
  /// uses.
  /// @dev     `registry` is the contract address; `chainSelector` identifies the
  ///          chain where the registry lives (Chainlink selector).
  struct CapabilitiesRegistryConfig {
    address registry;
    uint64 chainSelector;
  }

  /// @dev Struct for EventRecord. This is used to store the events that were emited related to capacity and workflow
  /// life-cycle.
  struct EventRecord {
    EventType eventType;
    uint32 timestamp;
    bytes payload; // ABI‑encoded event data
  }

  /// @dev Struct for OwnerAllowlistedRequest. This is used to return the allowlisted request data for each owner.
  struct OwnerAllowlistedRequest {
    bytes32 requestDigest;
    address owner;
    uint32 expiryTimestamp;
  }

  // ================================================================
  // |                   Workflow Metadata Config                   |
  // ================================================================
  /// @notice setConfig function allows the owner to override all the config parameters.
  /// @param nameLen New cap for `workflowName` (0 ➜ unlimited)
  /// @param tagLen  New cap for `tag` (0 ➜ unlimited)
  /// @param urlLen  New cap for each URL (0 ➜ unlimited)
  /// @param attrLen New cap for `attributes` (0 ➜ unlimited)
  /// @param expiryLen New cap for every allowlisted request expiration timestamp (0 ➜ unlimited)
  function setConfig(uint8 nameLen, uint8 tagLen, uint8 urlLen, uint16 attrLen, uint32 expiryLen) external onlyOwner {
    s_config =
      Config({maxNameLen: nameLen, maxTagLen: tagLen, maxUrlLen: urlLen, maxAttrLen: attrLen, maxExpiryLen: expiryLen});

    emit ConfigUpdated(nameLen, tagLen, urlLen, attrLen, expiryLen);
  }

  /// @notice getConfig function returns the current metadata config.
  function getConfig() public view returns (Config memory) {
    return s_config;
  }

  // ================================================================
  // |                   Workflow Owner Config                      |
  // ================================================================
  /// @notice Let each workflow‐owner store an arbitrary “config blob” (e.g. billing params)
  /// @dev    You can put any encoded data here; off‐chain tools will watch the event or call the getter.
  /// @param  config  ABI‐encoded owner‐specific settings
  function setWorkflowOwnerConfig(address owner, bytes calldata config) external onlyOwner {
    s_ownerConfig[owner] = config;
    emit WorkflowOwnerConfigUpdated(owner, config);
  }

  /// @notice Read back an owner’s last‐saved config blob
  /// @param  owner  The address whose config you want
  /// @return The raw `bytes` they most recently set
  function getWorkflowOwnerConfig(
    address owner
  ) external view returns (bytes memory) {
    return s_ownerConfig[owner];
  }

  // ================================================================
  // |                         DON Config                           |
  // ================================================================
  /// @notice Sets or clears a DON-wide limit for the maximum number of
  ///         ACTIVE workflows.
  /// @dev    Only callable by the contract owner.
  ///         When `enabled` is true, a limit is added for the DON family.
  ///         When `enabled` is false, the existing limit is removed.
  ///         When both adding and removing, an event record is created for the event, and the event itself
  ///         is emited in the internal helper.
  /// @notice Sets or clears the DON-wide limit for active workflows.
  /// @param donFamily Human-readable string DON family
  /// @param limit     New cap (ignored if `enabled==false`)
  /// @param enabled   Flag indicating whether to store (`true`) or delete (`false`)
  function setDONLimit(string calldata donFamily, uint32 limit, bool enabled) external onlyOwner {
    bytes32 donHash = _hash(donFamily);
    uint32 newCapacity = enabled ? limit : 0;
    DonConfig storage cfg = s_donConfigs[donHash];

    if (cfg.limitValue.enabled == enabled && cfg.limitValue.value == newCapacity) {
      return;
    }

    // write the human-readable string only once
    if (bytes(cfg.family).length == 0) {
      cfg.family = donFamily;
      s_donConfigKeys.add(donHash); // Tracks every DON hash ever configured (iterable, even if later disabled).
    }
    cfg.limitValue.enabled = enabled;
    cfg.limitValue.value = newCapacity;

    s_events.push(
      EventRecord({
        eventType: EventType.DONCapacitySet,
        timestamp: uint32(block.timestamp),
        payload: abi.encode(donHash, newCapacity)
      })
    );

    emit DONLimitSet(donFamily, newCapacity);
  }

  /// @notice Sets or removes a per-user, per-DON limit for ACTIVE workflows.
  /// @dev    Only the contract owner may call this.
  ///         - When `enabled` is true, stores the override and emits `UserDONLimitSet(user, donFamily, limit)`.
  ///         - When `enabled` is false, deletes any override and emits `UserDONLimitUnset(user, donFamily)`.
  ///         - When `enabled` is true and the same limit already exist, it does not write or emit a new event.
  ///         - When `enabled` is false and there is no limit, it does not remove or emit a new event.
  ///         The per-user override `limit` must not exceed the global DON limit, otherwise it reverts.
  ///         When per-user overrides are added or removed, no event record is added because this does not affect the
  ///         actual capacity of the DON as it is already constrained by the DON capacity.
  /// @notice Sets or clears a per‐user override for the maximum active workflows on a given DON
  /// @param user       The address for which to set or clear the override
  /// @param donFamily  The human‐readable DON family string (must have been configured via `setDONLimit`)
  /// @param limit      New per‐user cap when `enabled == true`
  /// @param enabled    `true` to enable/update the override; `false` to remove it
  function setUserDONOverride(address user, string calldata donFamily, uint32 limit, bool enabled) external onlyOwner {
    bytes32 donHash = _hash(donFamily);
    DonConfig storage cfg = s_donConfigs[donHash];

    // Ensure the DON itself has a global cap configured
    if (!cfg.limitValue.enabled) {
      revert DonLimitNotSet(donFamily);
    }

    ConfigValue storage ov = cfg.userOverride[user];

    if (enabled) {
      // Must not exceed the global DON cap
      if (limit > cfg.limitValue.value) {
        revert UserDONOverrideExceedsDONLimit();
      }

      if (!ov.enabled) {
        // → was OFF, now turning ON with new cap
        ov.enabled = true;
        ov.value = limit;
        s_donOverrideUsers[donHash].add(user);
        emit UserDONLimitSet(user, donFamily, limit);
      } else if (ov.value != limit) {
        // → was ON with a different cap, so update it
        ov.value = limit;
        emit UserDONLimitSet(user, donFamily, limit);
      } else {
        // → already ON at exactly this cap, nothing to do
        return;
      }
    } else {
      if (!ov.enabled) {
        // → already OFF, nothing to do
        return;
      }
      // → was ON, now turning OFF (and clearing the value)
      delete cfg.userOverride[user];
      s_donOverrideUsers[donHash].remove(user);
      emit UserDONLimitUnset(user, donFamily);
    }
  }

  /// @notice Gets the configured maximum number of workflows for a given DON family.
  /// @dev DON familys must first be configured in the Config.donLimit before workflows can be created against them.
  /// @param donFamily The identifier of the DON whose workflow cap is being queried.
  /// @return maxWorkflows The maximum number of workflows allowed for the specified DON, or zero if the DON
  ///                      is not allowlisted or no limit has been explicitly set.
  function getMaxWorkflowsPerDON(
    string calldata donFamily
  ) public view returns (uint32 maxWorkflows) {
    return s_donConfigs[_hash(donFamily)].limitValue.value;
  }

  /// @notice Returns the active-workflow cap that applies to a given DON and user.
  /// @dev    If a DON-specific user override is present and enabled, that override value
  ///         is returned; otherwise, the DON’s default limit is used.
  /// @param  user     Address of the user whose override limit is being queried.
  /// @param  donFamily String identifier of the DON.
  /// @return maxActive Maximum number of ACTIVE workflows allowed for the user on that DON.
  function getMaxWorkflowsPerUserDON(address user, string calldata donFamily) public view returns (uint32) {
    DonConfig storage cfg = s_donConfigs[_hash(donFamily)];

    // If the user has an override, return that
    ConfigValue memory ov = cfg.userOverride[user];
    if (ov.enabled) {
      return ov.value;
    }

    // Otherwise fall back to the global DON cap
    return cfg.limitValue.value;
  }

  /// @notice    Lists every DON configuration ever created.
  /// @param     start  First index to include in the slice.
  /// @param     limit  Maximum number of configs to return.
  /// @return    list   Array of DonConfigView.
  function getDonConfigs(uint256 start, uint256 limit) external view returns (DonConfigView[] memory list) {
    uint256 total = s_donConfigKeys.length();
    uint256 count = _getPageCount(total, start, limit);

    list = new DonConfigView[](count);
    for (uint256 i = 0; i < count; ++i) {
      bytes32 donHash = s_donConfigKeys.at(start + i);
      DonConfig storage cfg = s_donConfigs[donHash];

      list[i] = DonConfigView({
        donHash: donHash,
        family: cfg.family,
        limit: cfg.limitValue.value,
        limitEnabled: cfg.limitValue.enabled
      });
    }
    return list;
  }

  /// @notice  List every per-user override configured for a DON family.
  ///
  /// @dev
  ///  - Relies on the enumerable index `s_donOverrideUsers[donHash]` that’s
  ///    maintained inside `setUserDONOverride`.
  ///  - If `start` is greater than or equal to the number of overrides, the
  ///    function returns an empty array instead of reverting.
  ///  - Each element of the returned array contains:
  ///      • `user`  – the address that has an override, and
  ///      • `limit` – that user’s custom ACTIVE-workflow cap on the DON.
  ///    The global cap set via `setDONLimit` still applies as an upper bound.
  /// @param donFamily  Human-readable DON label (e.g., `"fast-pool"`).
  /// @param start      Zero-based index at which to begin the page.
  /// @param limit      Maximum number of overrides to return.
  /// @return list      Array of `UserOverrideView` structs:
  ///                   – `user`  → address with an override
  ///                   – `limit` → custom cap for that user
  function getUserDONOverrides(
    string calldata donFamily,
    uint256 start,
    uint256 limit
  ) external view returns (UserOverrideView[] memory list) {
    bytes32 donHash = _hash(donFamily);
    EnumerableSet.AddressSet storage set = s_donOverrideUsers[donHash];

    uint256 total = set.length();
    uint256 count = _getPageCount(total, start, limit);

    list = new UserOverrideView[](count);
    for (uint256 i; i < count; ++i) {
      address addr = set.at(start + i);
      ConfigValue memory cv = s_donConfigs[donHash].userOverride[addr];
      list[i] = UserOverrideView({user: addr, limit: cv.value});
    }
    return list;
  }

  // ================================================================
  // |                     Capabilities Registry                    |
  // ================================================================
  /// @notice Sets or replaces the Capabilities Registry that this Workflow Registry points to.
  /// @dev    Owner-only.  Overwrites the previous entry and emits
  ///         {CapabilitiesRegistryUpdated}.
  /// @param  registry       Address of the Capabilities Registry contract.
  /// @param  chainSelector  Chain selector for the registry’s chain.
  function setCapabilitiesRegistry(address registry, uint64 chainSelector) external onlyOwner {
    address oldRegistry = s_capabilitiesRegistry.registry;
    uint64 oldChain = s_capabilitiesRegistry.chainSelector;

    if (registry == oldRegistry && chainSelector == oldChain) {
      return;
    }

    if (registry != oldRegistry) {
      s_capabilitiesRegistry.registry = registry;
    }
    if (chainSelector != oldChain) {
      s_capabilitiesRegistry.chainSelector = chainSelector;
    }

    emit CapabilitiesRegistryUpdated(oldRegistry, registry, oldChain, chainSelector);
  }

  /// @notice Returns the current Capabilities Registry reference and its chain selector.
  /// @return Address of the Capabilities Registry contract.
  /// @return Chain selector for the registry’s chain.
  function getCapabilitiesRegistry() external view returns (address, uint64) {
    return (s_capabilitiesRegistry.registry, s_capabilitiesRegistry.chainSelector);
  }

  // ================================================================
  // |               Linking Admin Functions                        |
  // ================================================================

  /// @notice Sets the allowed signers for ownership proofs. These signers are considered to be trusted entities.
  /// @param signers The addresses of the signers.
  /// @param allowed The boolean value indicating whether the signer is trusted or not.
  /// @dev Ownership proofs can only be signed by approved group of signers.
  /// When submitting signed proof to this contract, if recovered signature doesn't match any of the signers,
  /// it will be rejected.
  function updateAllowedSigners(address[] calldata signers, bool allowed) external onlyOwner {
    for (uint256 i = 0; i < signers.length; ++i) {
      if (signers[i] == address(0)) {
        revert ZeroAddressNotAllowed();
      }
      s_allowedSigners[signers[i]] = allowed;
    }
    emit AllowedSignersUpdated(signers, allowed);
  }

  /// @notice Returns the allowed signer for ownership proofs.
  /// @param signer The address of the signer.
  /// @return The boolean value indicating whether the signer is allowed to sign ownership proofs or not.
  function isAllowedSigner(
    address signer
  ) external view returns (bool) {
    return s_allowedSigners[signer];
  }

  // ================================================================
  // |                Owner linking functions                       |
  // ================================================================
  /// @notice View function to verify if the linkOwner() function can be called successfully.
  /// @param owner The address of the owner to be linked.
  /// @param validityTimestamp Validity of the ownership proof.
  /// @param proof The ownership proof to be submitted.
  /// @param signature The signature of the ownership proof metadata.
  /// @dev This function is used to verify if the ownership proof is valid without actually linking the owner address.
  /// The ownership proof metadata is a combination of the claimed owner address, validity timestamp, and the proof
  /// hash.
  /// Request will be rejected if the validity timestamp has expired, owner addres is already linked, if the proof does
  /// not match the one that was originally submitted, or if the signature is not valid (for different reasons).
  function canLinkOwner(address owner, uint256 validityTimestamp, bytes32 proof, bytes calldata signature) public view {
    if (block.timestamp > validityTimestamp) {
      revert LinkOwnerRequestExpired(owner, block.timestamp, validityTimestamp);
    }

    // Workflow owner address may only be linked once
    if (s_linkedOwners.contains(owner)) {
      revert OwnershipLinkAlreadyExists(owner);
    }

    // Ownership proof must be unique and must not be used for linking more than once
    if (s_usedProofs[proof]) {
      revert OwnershipProofAlreadyUsed(owner, proof);
    }

    address signer = _recoverSigner(uint8(LinkingRequestType.LINK_OWNER), owner, validityTimestamp, proof, signature);
    if (!s_allowedSigners[signer]) {
      revert InvalidOwnershipLink(owner, validityTimestamp, proof, signature);
    }
  }

  /// @notice Transaction sender submits ownership proof for verification and approval. Upon approval, owner is linked.
  /// @param validityTimestamp Validity of the ownership proof.
  /// @param proof The ownership proof to be submitted.
  /// @param signature The signature of the ownership proof metadata.
  /// @dev Run the verification process first by calling canLinkOwner() function. If the verification does not result
  /// in a revert, then the ownership proof is valid and the owner address can be linked. Only the caller can link their
  /// address.
  function linkOwner(uint256 validityTimestamp, bytes32 proof, bytes calldata signature) external {
    canLinkOwner(msg.sender, validityTimestamp, proof, signature);

    s_linkedOwners.set(msg.sender, proof);
    s_usedProofs[proof] = true;
    emit OwnershipLinkUpdated(msg.sender, proof, true);
  }

  /// @notice Validates whether an owner can be unlinked using the provided proof and signature.
  /// @param owner The address of the owner to be unlinked.
  /// @param validityTimestamp Validity of the ownership proof.
  /// @param signature The signature of the ownership proof metadata.
  /// @dev This function is used to verify if the ownership proof is valid without actually unlinking the owner address.
  /// The ownership proof metadata is a combination of the claimed owner address, validity timestamp, and the proof
  /// hash.
  /// Request will be rejected if the validity timestamp has expired, owner address is not linked, if the proof does not
  /// match the one that was originally submitted, or if the signature is not valid (for different reasons).
  /// @dev Important difference between linking and unlinking is that unlinking may be called by any address, as
  /// long as the valid proof is provided. The caller does not have to be the owner of the address being unlinked.
  /// This is done to ensure that unlinking can be done even in cases when access to the private key of the owner
  /// address is lost or compromised, and the owner is not able to submit the unlinking request themselves.
  function canUnlinkOwner(address owner, uint256 validityTimestamp, bytes calldata signature) public view {
    if (block.timestamp > validityTimestamp) {
      revert UnlinkOwnerRequestExpired(owner, block.timestamp, validityTimestamp);
    }

    if (!s_linkedOwners.contains(owner)) {
      revert OwnershipLinkDoesNotExist(owner);
    }

    // The expectation is that the signature must contain the same proof that was originally used for the linking
    bytes32 storedProof = s_linkedOwners.get(owner);

    // Request type prevents replay attacks, since the same proof can be used for both linking and unlinking
    address signer =
      _recoverSigner(uint8(LinkingRequestType.UNLINK_OWNER), owner, validityTimestamp, storedProof, signature);
    if (!s_allowedSigners[signer]) {
      revert InvalidOwnershipLink(owner, validityTimestamp, storedProof, signature);
    }
  }

  /// @notice Transaction sender submits ownership proof for verification and approval. Upon approval, owner is
  /// unlinked.
  ///         This function can be called by anyone with signatures for the owner.
  /// @param owner The address of the owner to be unlinked.
  /// @param validityTimestamp Validity of the ownership proof.
  /// @param signature The signature of the ownership proof metadata.
  /// @dev The function will automatically delete all workflows owned by the owner before unlinking.
  /// Upstream callers are responsible for ensuring this is the intended behavior.
  /// @dev The function validates the ownership proof and signature before proceeding with deletion and unlinking.
  function unlinkOwner(address owner, uint256 validityTimestamp, bytes calldata signature) external {
    // Validate the unlinking request
    if (block.timestamp > validityTimestamp) {
      revert UnlinkOwnerRequestExpired(owner, block.timestamp, validityTimestamp);
    }

    if (!s_linkedOwners.contains(owner)) {
      revert OwnershipLinkDoesNotExist(owner);
    }

    // The expectation is that the signature must contain the same proof that was originally used for the linking
    bytes32 storedProof = s_linkedOwners.get(owner);

    // Request type prevents replay attacks, since the same proof can be used for both linking and unlinking
    address signer =
      _recoverSigner(uint8(LinkingRequestType.UNLINK_OWNER), owner, validityTimestamp, storedProof, signature);
    if (!s_allowedSigners[signer]) {
      revert InvalidOwnershipLink(owner, validityTimestamp, storedProof, signature);
    }

    // Delete all workflows owned by the owner
    EnumerableSet.Bytes32Set storage allRids = s_allOwnerRids[owner];
    // Iterate from the back since EnumerableSet.remove() swaps-and-pops.
    while (allRids.length() > 0) {
      bytes32 rid = allRids.at(allRids.length() - 1);
      WorkflowMetadata storage rec = s_workflows[rid];
      _applyDelete(rid, rec);
    }

    s_linkedOwners.remove(owner);
    emit OwnershipLinkUpdated(owner, storedProof, false);
  }

  /// @notice Returns if the owner is linked to this contract.
  /// @param owner The address of the owner.
  /// @return True if the link exists, false otherwise.
  function isOwnerLinked(
    address owner
  ) external view returns (bool) {
    return s_linkedOwners.contains(owner);
  }

  /// @notice Returns total count of linked owners.
  /// @return The total number of linked owners.
  function totalLinkedOwners() external view returns (uint256) {
    return s_linkedOwners.length();
  }

  /// @notice Retrieves a paginated list of addresses that have linked ownership proofs.
  /// @param start Zero-based index of the first owner to include in the result.
  /// @param limit Maximum number of owners to return (clamped by a sensible internal cap).
  /// @return owners An array of owner addresses in the order they were linked.
  /// @dev    - If `start` ≥ total linked owners, returns an empty array.
  ///         - The list can change between calls; for an immutable snapshot, query at a specific block.
  function getLinkedOwners(uint256 start, uint256 limit) external view returns (address[] memory owners) {
    uint256 total = s_linkedOwners.length();
    uint256 count = _getPageCount(total, start, limit);

    owners = new address[](count);
    for (uint256 i = 0; i < count; ++i) {
      (owners[i],) = s_linkedOwners.at(start + i);
    }

    return owners; // solcov:ignore next
  }

  /// @notice Returns the signer of the recovered signature or revert.
  /// @param requestType The type of the request (LINK_OWNER = 0 or UNLINK_OWNER = 1).
  /// @param owner The address of the owner.
  /// @param validityTimestamp The validity timestamp of the ownership proof.
  /// @param proof The ownership proof.
  /// @param signature The signature of the ownership proof metadata.
  /// @return The signer of the recovered signature.
  /// @dev The function tries to re-generate the message digest based on the provided parameters and by following
  /// EIP-191. The it will try to recover the signer address. The function will revert if the signature is invalid.
  function _recoverSigner(
    uint8 requestType,
    address owner,
    uint256 validityTimestamp,
    bytes32 proof,
    bytes calldata signature
  ) internal view returns (address) {
    // Follow EIP-191 for recoverable signatures
    bytes32 prefixedMessageHash = MessageHashUtils.toEthSignedMessageHash(
      keccak256(abi.encode(requestType, owner, block.chainid, address(this), typeAndVersion, validityTimestamp, proof))
    );

    (address signer, ECDSA.RecoverError err, bytes32 errArg) = ECDSA.tryRecover(prefixedMessageHash, signature);
    if (err != ECDSA.RecoverError.NoError) {
      revert InvalidSignature(signature, uint8(err), errArg);
    }

    return signer;
  }

  // ================================================================
  // |                   Capacity Changing Events                   |
  // ================================================================
  /// @notice Returns a page of events along with the total event count.
  /// @param start Zero-based index of the first event to include in the page.
  /// @param limit  Maximum number of events to return in this page.
  /// @return list  Array of events in the requested window.
  function getEvents(uint256 start, uint256 limit) external view returns (EventRecord[] memory list) {
    uint256 total = s_events.length;
    uint256 count = _getPageCount(total, start, limit);

    list = new EventRecord[](count);
    for (uint256 i = 0; i < count; ++i) {
      list[i] = s_events[start + i];
    }

    return list; // solcov:ignore next
  }

  /// @notice Returns the total number of capacity- and workflow-lifecycle events ever recorded.
  /// @dev    Use this in tandem with `getEvents(start, limit)` to page through the event stream.
  /// @return count The total count of EventRecord entries stored in `s_events`.
  function totalEvents() external view returns (uint256 count) {
    return s_events.length;
  }

  // ================================================================
  // |                       Workflow Management                    |
  // ================================================================
  /// Storage invariant:
  ///   - `s_workflows` RID is the primary key on the workflow storage.
  /// .   It is comprised of a hash of the (workflowName, workflowOwner, workflowTag)
  ///
  ///   - Every **active** workflow adds to:
  ///         • `s_activeDONWorkflowRids[don]` (all active workflows for a don)
  ///         • `s_activeOwnerWorkflowRids[owner]; (all active workflows for an owner)
  ///         • `s_userDONCount[owner][don]` (all active workflows for an owner in a don)
  ///         • `s_activeRidsByWorkflowKey[key]` (all active workflows by (name, owner))
  ///         • `s_donByWorkflowId[rid]` (the don family that this active workflow is assigned to)
  ///   - Similarly, every deactivation removes from the above indices.
  ///   - Status transitions must update **all five** structures.
  ///
  /// @notice Upserts a new workflow based on workflowName + owner + tag. If triplet already exist
  /// as a record, then we will update that existing workflow. Otherwise, a new one with the new
  /// tag is created.
  /// Status and donFamily cannot be updated via upsert. They must use their own separate functions
  /// for any changes to these workflow fields.
  /// @param workflowName  Human‑readable name (≤64 chars)
  /// @param tag           Unique tag for the workflow (if the same workflowName has been used)
  /// @param workflowId    Deterministic hash computed off‑chain (must be unique)
  /// @param donFamily     Family (label) string of the DON
  /// @param status        Initial status (ACTIVE / PAUSED)
  /// @param binaryUrl     URL of the wasm binary (required)
  /// @param configUrl     URL of the config (optional)
  /// @param attributes    Arbitrary bytes for additional workflow details (optional)
  /// @param keepAlive     Boolean flag that determines whether existing workflows with the same
  /// workflowName, workflowOwner combination should be paused or kept active.
  function upsertWorkflow(
    string calldata workflowName,
    string calldata tag,
    bytes32 workflowId,
    WorkflowStatus status,
    string calldata donFamily,
    string calldata binaryUrl,
    string calldata configUrl,
    bytes calldata attributes,
    bool keepAlive
  ) external {
    /* ───────────────────────── 0. VALIDATION ─────────────────────────── */
    // 1) check ownership links
    if (!s_linkedOwners.contains(msg.sender)) {
      revert OwnershipLinkDoesNotExist(msg.sender);
    }
    // 2) check workflowID
    if (workflowId == bytes32(0)) revert ZeroWorkflowIDNotAllowed();
    if (s_idToRid[workflowId] != bytes32(0)) {
      revert WorkflowIDAlreadyExists(workflowId);
    }
    // 3) check URLs (binary url is required. config url is optional; 0 = unlimited)
    uint8 cap = s_config.maxUrlLen;
    uint256 binaryLen = bytes(binaryUrl).length;
    uint256 configLen = bytes(configUrl).length;
    if (binaryLen == 0) {
      revert BinaryURLRequired();
    }
    if (cap != 0) {
      if (binaryLen > cap) {
        revert URLTooLong(binaryLen, cap);
      }
      if (configLen > cap) {
        revert URLTooLong(configLen, cap);
      }
    }

    // 4) check attributes (optional; 0 = unlimited)
    uint16 attrCap = s_config.maxAttrLen;
    if (attrCap != 0 && attributes.length > attrCap) {
      revert AttributesTooLong(attributes.length, attrCap);
    }

    // 5) check tag (required)
    uint256 tagLen = bytes(tag).length;
    if (tagLen == 0) {
      revert WorkflowTagRequired();
    }
    cap = s_config.maxTagLen; // 0  ➜ unlimited
    if (cap != 0 && tagLen > cap) {
      revert WorkflowTagTooLong(tagLen, cap);
    }
    // 6) check workflowName (required)
    uint256 nameLen = bytes(workflowName).length;
    if (nameLen == 0) {
      revert WorkflowNameRequired();
    }
    cap = s_config.maxNameLen; // 0  ➜ unlimited
    if (cap != 0 && nameLen > cap) {
      revert WorkflowNameTooLong(nameLen, cap);
    }

    // using abi.encode here ensures each dynamic field (string) is length-prefixed,
    // so “owner∥name∥tag” can never collide across different triples.
    bytes32 rid = keccak256(abi.encode(msg.sender, workflowName, tag));
    WorkflowMetadata storage rec = s_workflows[rid];
    // Create workflow path
    if (rec.owner == address(0)) {
      bytes32 wKey = _workflowKey(msg.sender, workflowName);
      bytes32 donHash = _hash(donFamily);

      /* ───────────────────────── 1. HOUSEKEEPING ───────────────────────── */
      // we need to do this first, or there may be extra workflows occupying the limit
      if (!keepAlive) {
        EnumerableSet.Bytes32Set storage activeSet = s_activeRidsByWorkflowKey[wKey];
        // Walk from the back since EnumerableSet.remove is a swap and pop.
        while (activeSet.length() > 0) {
          uint256 lastIdx = activeSet.length() - 1;
          bytes32 prevRid = activeSet.at(lastIdx);
          WorkflowMetadata storage prevRec = s_workflows[prevRid];
          // Update workflow state
          _applyPause(prevRid, prevRec);
        }
      }

      /* ───────────────────────── 2. LIMIT CHECKS ───────────────────────── */
      if (status == WorkflowStatus.ACTIVE) {
        _enforceLimits(msg.sender, donHash, donFamily, 1);
        // update indices necessary for active workflows
        _addActiveIndices(rid, msg.sender, donHash, wKey);
      }

      /* ───────────────────────── 3. WRITE PRIMARY RECORD ───────────────── */
      s_workflows[rid] = WorkflowMetadata({
        workflowId: workflowId,
        owner: msg.sender,
        createdAt: uint64(block.timestamp),
        status: status,
        workflowName: workflowName,
        binaryUrl: binaryUrl,
        configUrl: configUrl,
        tag: tag,
        attributes: attributes
      });

      /* ───────────────────────── 4. UPDATE OTHER INDICES ───────────────── */
      s_workflowKeyToRids[wKey].add(rid);
      s_idToRid[workflowId] = rid;
      s_allDONRids[donHash].add(rid);
      s_allOwnerRids[msg.sender].add(rid);

      /* ───────────────────────── 5. EVENT LOG ──────────────────────────── */
      emit WorkflowRegistered(workflowId, msg.sender, donFamily, status, workflowName);
    } else {
      // update workflow path
      // check the workflow belongs to the owner
      if (rec.owner != msg.sender) revert CallerIsNotWorkflowOwner(msg.sender);

      /* ─────── 2. PRIMARY-KEY REMAP ─────── */
      delete s_idToRid[rec.workflowId];
      s_idToRid[workflowId] = rid;

      /* ─────── 3. FIELD PATCHES ─────── */
      bytes32 oldWorkflowId = rec.workflowId;
      rec.workflowId = workflowId;
      if (_hash(rec.binaryUrl) != _hash(binaryUrl)) rec.binaryUrl = binaryUrl;
      if (_hash(rec.configUrl) != _hash(configUrl)) rec.configUrl = configUrl;
      rec.attributes = attributes;

      /* ─────── 4. EVENT ─────── */
      emit WorkflowUpdated(oldWorkflowId, workflowId, msg.sender, donFamily, workflowName);
    }
  }

  function pauseWorkflow(
    bytes32 workflowId
  ) external {
    if (!s_linkedOwners.contains(msg.sender)) {
      revert OwnershipLinkDoesNotExist(msg.sender);
    }

    bytes32 rid = s_idToRid[workflowId];
    WorkflowMetadata storage rec = _getRecord(msg.sender, rid);
    if (rec.status != WorkflowStatus.PAUSED) {
      _applyPause(rid, rec);
    }
  }

  function activateWorkflow(bytes32 workflowId, string calldata donFamily) external {
    if (!s_linkedOwners.contains(msg.sender)) {
      revert OwnershipLinkDoesNotExist(msg.sender);
    }

    bytes32 rid = s_idToRid[workflowId];
    WorkflowMetadata storage rec = _getRecord(msg.sender, rid);
    if (rec.status != WorkflowStatus.ACTIVE) {
      bytes32 donHash = _hash(donFamily);
      _enforceLimits(msg.sender, donHash, donFamily, 1);
      _applyActivate(rid, rec, donHash);
    }
  }

  /// @notice Pauses multiple workflows owned by `msg.sender`.
  /// @dev    There is no enforced batch size limit here.
  ///         **User Risk:** Submitting a very large array of `workflowIds`
  ///         may cause the transaction to run out of gas and revert.
  ///         Clients should cap the array length to a safe value based on the current gas limits.
  /// @param  workflowIds Array of workflow IDs to pause; must not be empty.
  function batchPauseWorkflows(
    bytes32[] calldata workflowIds
  ) external {
    uint256 n = workflowIds.length;
    if (n == 0) revert EmptyUpdateBatch();

    if (!s_linkedOwners.contains(msg.sender)) {
      revert OwnershipLinkDoesNotExist(msg.sender);
    }

    for (uint256 i = 0; i < n; ++i) {
      bytes32 rid = s_idToRid[workflowIds[i]];
      WorkflowMetadata storage rec = _getRecord(msg.sender, rid);
      if (rec.status != WorkflowStatus.PAUSED) {
        _applyPause(rid, rec);
      }
    }
  }

  /// @notice Activate many paused workflows owned by the caller,
  ///        assigning *all* of them to a single DON family.
  ///        If the list contains some workflows that are already ACTIVE on another DON, they are
  ///        silently ignored; the rest are activated on the new DON.
  /// @param workflowIds  Array of workflow IDs to activate (must not be empty).
  /// @param donFamily    Target DON family; must already have a global limit.
  function batchActivateWorkflows(bytes32[] calldata workflowIds, string calldata donFamily) external {
    uint256 n = workflowIds.length;
    if (n == 0) revert EmptyUpdateBatch();
    if (!s_linkedOwners.contains(msg.sender)) {
      revert OwnershipLinkDoesNotExist(msg.sender);
    }

    /* ──────────────────────── 1. PRE‑CHECKS & COUNT ───────────────────── */
    bytes32 donHash = _hash(donFamily);
    uint32 pending = 0; // # workflows that will become ACTIVE

    for (uint256 i; i < n; ++i) {
      bytes32 rid = s_idToRid[workflowIds[i]];
      WorkflowMetadata storage rec = _getRecord(msg.sender, rid);
      if (rec.status == WorkflowStatus.ACTIVE) continue; // already active, ignore
      ++pending;
    }

    if (pending == 0) return; // nothing to do

    /* ───────────────────────── 2. CAP ENFORCEMENT ─────────────────────── */
    _enforceLimits(msg.sender, donHash, donFamily, pending);

    /* ───────────────────────── 3. STATE MUTATIONS ─────────────────────── */
    for (uint256 i; i < n; ++i) {
      bytes32 rid = s_idToRid[workflowIds[i]];
      WorkflowMetadata storage rec = s_workflows[rid];
      if (rec.status == WorkflowStatus.PAUSED) {
        _applyActivate(rid, rec, donHash); // updates indices & emits WorkflowActivated
      }
    }
  }

  /// @dev   Apply the state transition PAUSED ➜ ACTIVE.
  /// @param rid  Registry-internal reference id (owner ∥ name ∥ tag).
  /// @param rec  Storage pointer to the workflow metadata.
  /// @notice *NO CHECKS* – caller must guarantee
  ///         • `rec.status == WorkflowStatus.PAUSED`
  ///         • DON/user caps have been enforced already.
  ///         • Caller can perform action.
  function _applyActivate(bytes32 rid, WorkflowMetadata storage rec, bytes32 donHash) private {
    _addActiveIndices(rid, rec.owner, donHash, _workflowKey(rec.owner, rec.workflowName));
    rec.status = WorkflowStatus.ACTIVE;

    s_events.push(
      EventRecord({
        eventType: EventType.WorkflowAdded,
        timestamp: uint32(block.timestamp),
        payload: abi.encode(donHash, rec.workflowId)
      })
    );

    emit WorkflowActivated(rec.workflowId, rec.owner, s_donConfigs[donHash].family, rec.workflowName);
  }

  /// @dev   Apply the state transition ACTIVE ➜ PAUSED.
  /// @notice No guards – caller must guarantee that:
  ///         • `rec.status == WorkflowStatus.ACTIVE`
  ///         • Any permission or limit logic has already been handled.
  /// @param  rid   Registry-internal reference ID (owner ∥ name ∥ tag hash).
  /// @param  rec   Storage pointer to the workflow metadata struct.
  function _applyPause(bytes32 rid, WorkflowMetadata storage rec) private {
    rec.status = WorkflowStatus.PAUSED;
    bytes32 donHash = s_donByWorkflowRid[rid];
    _removeActiveIndices(rid, rec.owner, donHash, _workflowKey(rec.owner, rec.workflowName));

    s_events.push(
      EventRecord({
        eventType: EventType.WorkflowRemoved,
        timestamp: uint32(block.timestamp),
        payload: abi.encode(donHash, rec.workflowId)
      })
    );

    emit WorkflowPaused(rec.workflowId, rec.owner, s_donConfigs[donHash].family, rec.workflowName);
  }

  /// @notice Permanently delete a workflow owned by the caller.
  /// @dev Sequence:
  ///  1. Verify the caller (owner) is linked to the registry.
  ///  2. Resolve the registry-ID (`rid`) from `workflowId` and verify ownership.
  ///  3. If the workflow is **ACTIVE**, remove it from every
  ///     "active" index and decrement per-DON counters.
  ///  4. Purge the RID from global owner / DON maps and key-based sets.
  ///  5. Clear the ID→RID map, delete the primary record, and emit an event.
  ///
  /// @param workflowId  The globally-unique identifier to remove.
  /// @custom:reverts OwnershipLinkDoesNotExist If the caller is not linked.
  /// @custom:reverts WorkflowDoesNotExist      If the ID is unknown.
  /// @custom:reverts CallerIsNotWorkflowOwner  If `msg.sender` is not the owner.
  function deleteWorkflow(
    bytes32 workflowId
  ) external {
    // Check that the caller (owner) is linked
    if (!s_linkedOwners.contains(msg.sender)) {
      revert OwnershipLinkDoesNotExist(msg.sender);
    }

    bytes32 rid = s_idToRid[workflowId];
    WorkflowMetadata storage rec = _getRecord(msg.sender, rid);
    _applyDelete(rid, rec);
  }

  /// @dev Removes a workflow’s RID from **all “active” indices** and
  ///      decrements the per-user per-DON counter.
  ///
  ///      Caller **must** ensure the workflow is currently
  ///      `WorkflowStatus.ACTIVE`; this helper performs no status
  ///      or validations on its own.
  /// @param rid          Registry-internal reference ID of the workflow.
  /// @param owner        Workflow owner address.
  /// @param donHash      Hash of the DON family of the workflow.
  /// @param workflowKey  keccak256(owner, workflowName) key used for the
  ///                     active-by-name index.
  function _removeActiveIndices(bytes32 rid, address owner, bytes32 donHash, bytes32 workflowKey) private {
    s_activeOwnerWorkflowRids[owner].remove(rid);
    s_activeDONWorkflowRids[donHash].remove(rid);
    s_userDONCount[owner][donHash] -= 1;
    s_activeRidsByWorkflowKey[workflowKey].remove(rid);
    delete s_donByWorkflowRid[rid];
  }

  /// @dev Adds a workflow RID into all “active” indices and increments the per-user, per-DON counter.
  ///      This helper performs no validation on its own.
  /// @param rid          The registry-internal reference ID of the workflow to mark active.
  /// @param owner        The address of the workflow owner.
  /// @param donHash      The keccak256 hash of the DON family under which this workflow is registered.
  /// @param workflowKey  The keccak256(owner, workflowName) key used for name-based active indexing.
  function _addActiveIndices(bytes32 rid, address owner, bytes32 donHash, bytes32 workflowKey) private {
    s_userDONCount[owner][donHash] += 1;
    s_activeDONWorkflowRids[donHash].add(rid);
    s_activeOwnerWorkflowRids[owner].add(rid);
    s_activeRidsByWorkflowKey[workflowKey].add(rid);
    s_donByWorkflowRid[rid] = donHash;
  }

  /// @notice This helper **assumes** all higher-level checks have
  ///         already been performed.
  ///         It also removes the workflow from all active indices.
  /// @param rid  Registry-internal reference ID (hash(owner, name, tag)).
  /// @param rec  Storage pointer to the workflow metadata struct that
  ///             corresponds to `rid`.
  function _applyDelete(bytes32 rid, WorkflowMetadata storage rec) private {
    bytes32 wKey = _workflowKey(rec.owner, rec.workflowName);
    bytes32 donHash = s_donByWorkflowRid[rid];
    string memory donFamily = s_donConfigs[donHash].family;
    if (rec.status == WorkflowStatus.ACTIVE) {
      _removeActiveIndices(rid, rec.owner, donHash, wKey);
    }

    s_allDONRids[donHash].remove(rid);
    s_allOwnerRids[rec.owner].remove(rid);
    s_workflowKeyToRids[wKey].remove(rid);
    delete s_idToRid[rec.workflowId];

    emit WorkflowDeleted(rec.workflowId, rec.owner, donFamily, rec.workflowName);
    delete s_workflows[rid];
  }

  /// @notice Change the DON family for a single workflow, updating all indices. This function only applies
  /// to currently active workflows, as paused workflows do not belong to any DONs and can be set the
  /// DON family upon activation.
  /// @param workflowId   The workflow to reassign
  /// @param newDonFamily The new human‐readable DON family (must be set via setDONLimit)
  function updateWorkflowDONFamily(bytes32 workflowId, string calldata newDonFamily) external {
    if (!s_linkedOwners.contains(msg.sender)) {
      revert OwnershipLinkDoesNotExist(msg.sender);
    }

    bytes32 rid = s_idToRid[workflowId];
    WorkflowMetadata storage rec = _getRecord(msg.sender, rid);
    if (rec.status != WorkflowStatus.ACTIVE) revert CannotUpdateDONFamilyForPausedWorkflows();

    bytes32 oldDonHash = s_donByWorkflowRid[rid];
    string memory oldDonFamily = s_donConfigs[oldDonHash].family;
    bytes32 newDonHash = _hash(newDonFamily);
    if (oldDonHash == newDonHash) return;

    // remove and pause active indices first for the old don family
    _applyPause(rid, rec);
    _enforceLimits(msg.sender, newDonHash, newDonFamily, 1);
    // activate with the new don family
    _applyActivate(rid, rec, newDonHash);
    emit WorkflowDonFamilyUpdated(workflowId, msg.sender, oldDonFamily, newDonFamily);
  }

  // ================================================================
  // |                       Admin Workflow                         |
  // ================================================================
  function adminPauseWorkflow(
    bytes32 workflowId
  ) public onlyOwner {
    bytes32 rid = s_idToRid[workflowId];
    WorkflowMetadata storage rec = s_workflows[rid]; // no msg.sender check when fetched directly from mapping
    if (rec.status == WorkflowStatus.ACTIVE) {
      _applyPause(rid, rec);
    }
  }

  // @notice Pauses a batch of workflows as an admin, bypassing workflow ownership checks.
  /// @dev    - Only the contract owner may call this function (enforced by `onlyOwner`).
  ///         - Reverts if `workflowIds` is empty to prevent no-op transactions.
  ///         - Iterates over each provided ID and calls `adminPauseWorkflow`, which itself
  ///           verifies the workflow is active before pausing.
  ///         - Beware: supplying an excessively large array may exhaust gas and revert.
  /// @param  workflowIds  Array of globally-unique workflow IDs to pause; must contain at least one element.
  function adminBatchPauseWorkflows(
    bytes32[] calldata workflowIds
  ) external {
    uint256 n = workflowIds.length;
    if (n == 0) revert EmptyUpdateBatch();

    for (uint256 i; i < n; ++i) {
      adminPauseWorkflow(workflowIds[i]);
    }
  }

  /// @notice Pauses *all* active workflows for a given workflow owner, as an administrator.
  /// @dev    - Only the contract owner may call this (`onlyOwner`).
  ///         - Iterates from the end of the user’s active-workflow set and directly pauses each workflow.
  /// @param  owner The address whose active workflows should be paused.
  function adminPauseAllByOwner(
    address owner
  ) external onlyOwner {
    EnumerableSet.Bytes32Set storage activeSet = s_activeOwnerWorkflowRids[owner];

    // Loop until the set is empty, always pausing the last element. We also know that all workflows in the list are
    // active.
    while (activeSet.length() > 0) {
      bytes32 rid = activeSet.at(activeSet.length() - 1);
      WorkflowMetadata storage rec = s_workflows[rid]; // no msg.sender check when fetched directly from mapping
      _applyPause(rid, rec);
    }
  }

  /// @notice Pauses *all* active workflows under a specific DON family, as an administrator.
  /// @dev    - Only the contract owner may call this (`onlyOwner`).
  ///         - Iterates from the end of the DON’s active-workflow set and directly pauses each workflow.
  /// @param  donFamily The string identifier of the DON whose active workflows should be paused.
  function adminPauseAllByDON(
    string calldata donFamily
  ) external onlyOwner {
    EnumerableSet.Bytes32Set storage activeSet = s_activeDONWorkflowRids[_hash(donFamily)];

    // Loop until the set is empty, always pausing the last workflow
    while (activeSet.length() > 0) {
      bytes32 rid = activeSet.at(activeSet.length() - 1);
      WorkflowMetadata storage rec = s_workflows[rid]; // no msg.sender check when fetched directly from mapping
      _applyPause(rid, rec);
    }
  }

  // ================================================================
  // |           Allowlisting requests (off-chain)                  |
  // ================================================================
  /// @notice Allowlists request for a specific owner, allowing anyone to verify off-chain requests.
  /// @dev    - This function is used to approve requests that can be executed by the Vault DON.
  ///         - The request is identified by a unique `requestDigest` (hash of the request payload).
  ///         - The `expiryTimestamp` defines until when the request is valid.
  ///         - Only owners that have linked their ownership proof can allowlist requests.
  /// @param requestDigest     Unique identifier for the request (hash of the request payload).
  /// @param expiryTimestamp   Timestamp until which the request is valid (must be in the future).
  /// @custom:revert OwnershipLinkDoesNotExist  If the caller is not a linked owner.
  /// @dev User flow:
  /// - User generates the digest of the request.
  /// - User calls allowlistRequest(requestDigest) on the Workflow Registry.
  /// - Any user can then send the request payload to the Vault DON.
  /// - Vault DON checks if the digest is on-chain for verification purposes.
  function allowlistRequest(bytes32 requestDigest, uint32 expiryTimestamp) external {
    uint32 maxAllowedExpiry = s_config.maxExpiryLen; // 0 -> unlimited
    if (
      expiryTimestamp <= block.timestamp
        || (maxAllowedExpiry != 0 && expiryTimestamp - block.timestamp > maxAllowedExpiry)
    ) {
      revert InvalidExpiryTimestamp(requestDigest, expiryTimestamp, maxAllowedExpiry);
    }
    if (!s_linkedOwners.contains(msg.sender)) revert OwnershipLinkDoesNotExist(msg.sender);

    s_requestsAllowlist[keccak256(abi.encode(msg.sender, requestDigest))] = expiryTimestamp;
    s_requestAllowlistArray.push(
      OwnerAllowlistedRequest({owner: msg.sender, requestDigest: requestDigest, expiryTimestamp: expiryTimestamp})
    );
    emit RequestAllowlisted(msg.sender, requestDigest, expiryTimestamp);
  }

  /// @notice Checks if a request is allowlisted for a specific owner.
  /// @dev    - Returns true if the request is allowlisted and not expired.
  ///         - The request is considered allowlisted if the current block timestamp is less than the expiry timestamp.
  /// @param owner            The address of the linked owner who allowlisted the request.
  /// @param requestDigest    Unique identifier for the request (hash of the request payload).
  /// @return bool            True if the request is allowlisted and not expired, false otherwise.
  function isRequestAllowlisted(address owner, bytes32 requestDigest) external view returns (bool) {
    return s_requestsAllowlist[keccak256(abi.encode(owner, requestDigest))] > block.timestamp;
  }

  function getAllowlistedRequests(
    uint256 start,
    uint256 limit
  ) external view returns (OwnerAllowlistedRequest[] memory allowlistedRequests) {
    uint256 total = s_requestAllowlistArray.length;
    uint256 pageCount = _getPageCount(total, start, limit);

    if (pageCount == 0) return new OwnerAllowlistedRequest[](0);

    allowlistedRequests = new OwnerAllowlistedRequest[](pageCount);
    uint256 addedCount = 0;
    for (uint256 i = 0; i < pageCount; ++i) {
      OwnerAllowlistedRequest storage request = s_requestAllowlistArray[start + i];
      if (request.expiryTimestamp > block.timestamp) {
        allowlistedRequests[addedCount] = request;
        ++addedCount;
      }
    }

    if (addedCount < pageCount) {
      OwnerAllowlistedRequest[] memory shrinkedList = new OwnerAllowlistedRequest[](addedCount);
      for (uint256 i = 0; i < addedCount; ++i) {
        shrinkedList[i] = allowlistedRequests[i];
      }
      allowlistedRequests = shrinkedList;
    }

    return allowlistedRequests;
  }

  function totalAllowlistedRequests() external view returns (uint256) {
    return s_requestAllowlistArray.length;
  }

  // ================================================================
  // |                       Workflow Views                         |
  // ================================================================
  /// @notice Return the full on-chain metadata for a given workflow.
  /// @dev
  /// 1. Looks up the registry-internal reference‐ID (RID) from
  ///    `s_idToRid` using the caller-supplied `workflowId`.
  /// 2. Uses that RID to fetch the packed `WorkflowMetadata` struct.
  /// 3. Reverts with `WorkflowDoesNotExist` if either mapping slot is
  ///    empty (i.e. the workflow was never registered or has been
  ///    deleted).
  ///
  /// @param workflowId Globally-unique, immutable identifier of the
  ///                   workflow (hash of owner, name, binary, cfg, …).
  /// @return workflow  In-memory copy of the workflow’s metadata record
  /// (it also includes the donFamily string for active workflows).
  ///
  /// @custom:revert WorkflowDoesNotExist  If no RID is found for
  ///                                       `workflowId`, or the RID maps
  ///                                       to an empty metadata record.
  function getWorkflowById(
    bytes32 workflowId
  ) external view returns (WorkflowMetadataView memory workflow) {
    return _workflowMetadataView(s_idToRid[workflowId]);
  }

  /// @notice Return the full on-chain metadata for a given workflow.
  /// @dev
  /// 1. Calculates registry-internal reference‐ID (RID) from
  ///    using the caller-supplied parameters.
  /// 2. Uses that RID to fetch the packed `WorkflowMetadata` struct.
  /// 3. Reverts with `WorkflowDoesNotExist` if either mapping slot is
  ///    empty (i.e. the workflow was never registered or has been
  ///    deleted).
  ///
  /// @param   owner         Address that registered the workflows.
  /// @param   workflowName  Case-sensitive name string (≤ 64 chars).
  /// @param   tag           Unique tag for the workflow.
  /// @return workflow  In-memory copy of the workflow’s metadata record
  /// (it also includes the donFamily string for active workflows).
  ///
  /// @custom:revert WorkflowDoesNotExist  If no RID is found for
  ///                                       `workflowId`, or the RID maps
  ///                                       to an empty metadata record.
  function getWorkflow(
    address owner,
    string calldata workflowName,
    string calldata tag
  ) external view returns (WorkflowMetadataView memory workflow) {
    return _workflowMetadataView(keccak256(abi.encode(owner, workflowName, tag)));
  }

  /// @notice  Return a paginated list of all versions (active *and* paused)
  ///          of workflows with the same `workflowName` and `owner`.
  /// @dev     Uses the secondary key ⟨owner, workflowName⟩ → RID-set
  ///          (`s_workflowKeyToRids`).
  ///          Does **not** revert on out-of-range pagination: if `start` is
  ///          beyond the end of the set the function returns an empty array.
  /// @param   owner         Address that registered the workflows.
  /// @param   workflowName  Case-sensitive name string (≤ 64 chars).
  /// @param   start         Zero-based index into the RID set.
  /// @param   limit         Max #records to return
  /// @return  list          Array of `WorkflowMetadataView`.
  function getWorkflowListByOwnerAndName(
    address owner,
    string calldata workflowName,
    uint256 start,
    uint256 limit
  ) external view returns (WorkflowMetadataView[] memory list) {
    bytes32 wKey = _workflowKey(owner, workflowName);
    uint256 total = s_workflowKeyToRids[wKey].length();
    uint256 count = _getPageCount(total, start, limit);

    list = new WorkflowMetadataView[](count);
    for (uint256 i = 0; i < count; ++i) {
      bytes32 rid = s_workflowKeyToRids[wKey].at(start + i);
      list[i] = _workflowMetadataView(rid);
    }

    return list;
  }

  /// @notice Return a paginated slice of all workflows (active + paused)
  ///         owned by a specific address.
  /// @dev
  /// - Reads the RID set `s_allOwnerRids[owner]`, which contains every
  ///   workflow ever registered by `owner`, regardless of status.
  /// - Does not revert on out-of-range pagination; it simply returns the
  ///   largest sub-range that fits inside the set.
  /// @param owner  The address whose workflows are requested.
  /// @param start  Zero-based index into the owner’s RID set.
  /// @param limit  Batch size for the workflows.
  /// @return list Array of `WorkflowMetadataView` with length
  ///              `min(limit, total-start)`.
  function getWorkflowListByOwner(
    address owner,
    uint256 start,
    uint256 limit
  ) external view returns (WorkflowMetadataView[] memory list) {
    uint256 total = s_allOwnerRids[owner].length();
    uint256 count = _getPageCount(total, start, limit);

    list = new WorkflowMetadataView[](count);
    for (uint256 i = 0; i < count; ++i) {
      bytes32 rid = s_allOwnerRids[owner].at(start + i);
      list[i] = _workflowMetadataView(rid);
    }
    return list;
  }

  /// @notice Fetch a paginated slice of **all** workflows (active *and*
  ///         paused) that belong to a given DON.
  /// @dev
  ///  * Reads the RID set `s_allDONRids[donHash]` derived from the donFamily, which tracks every
  ///    workflow ever registered to that DON, regardless of status.
  ///  * Does **not** revert on out-of-range requests; instead it returns
  ///    the largest sub-range that fits inside the set.
  ///
  /// @param donFamily  Human readable string of the DON family.
  /// @param start     Zero-based index into the RID set.
  /// @param limit     Bathc size for the workflows
  /// @return list     Array of `WorkflowMetadataView` structs whose length is
  ///                  `min(limit, total-start)`.
  function getWorkflowListByDON(
    string calldata donFamily,
    uint256 start,
    uint256 limit
  ) external view returns (WorkflowMetadataView[] memory list) {
    bytes32 donHash = _hash(donFamily);
    uint256 total = s_allDONRids[donHash].length();
    uint256 count = _getPageCount(total, start, limit);

    list = new WorkflowMetadataView[](count);
    for (uint256 i = 0; i < count; ++i) {
      bytes32 rid = s_allDONRids[donHash].at(start + i);
      list[i] = _workflowMetadataView(rid);
    }

    return list;
  }

  /// @notice Returns the number of ACTIVE workflows on a given DON.
  /// @param donFamily The human-readable DON label (must have been configured via `setDONLimit`).
  /// @return count The total number of active workflows on that DON.
  function totalActiveWorkflowsOnDON(
    string calldata donFamily
  ) external view returns (uint256 count) {
    return s_activeDONWorkflowRids[_hash(donFamily)].length();
  }

  /// @notice Returns the number of ACTIVE workflows the owner has over all DONs.
  /// @param owner The owner of the workflows
  /// @return count The total number of active workflows owned by the address in the param.
  function totalActiveWorkflowsByOwner(
    address owner
  ) external view returns (uint256 count) {
    return s_activeOwnerWorkflowRids[owner].length();
  }

  /// @dev Calculates how many items fit into a page slice.
  ///      - If `start >= total`, returns `0` to indicate an empty slice.
  ///      - Otherwise, clamps the page end to `total` when `start + limit` exceeds it.
  /// @param total The total number of items available.
  /// @param start The zero-based index at which the page begins.
  /// @param limit The maximum number of items to include in the page.
  /// @return count The number of items from `start` before hitting `total` (zero if `start >= total`).
  function _getPageCount(uint256 total, uint256 start, uint256 limit) internal pure returns (uint256 count) {
    if (start >= total) {
      return 0;
    }
    uint256 end = start + limit > total ? total : start + limit;
    return end - start;
  }

  function _enforceLimits(address owner, bytes32 donHash, string memory donFamily, uint32 pending) internal view {
    DonConfig storage cfg = s_donConfigs[donHash];
    // Global limit must be explicitly enabled (zero or missing = disallowed)
    if (!cfg.limitValue.enabled) revert DonLimitNotSet(donFamily);

    // Determine the effective cap: start with the global cap, but if this user has an override, use that instead
    uint32 cap = cfg.userOverride[owner].enabled ? cfg.userOverride[owner].value : cfg.limitValue.value;

    // <= to include the pending addition(s)
    if (s_userDONCount[owner][donHash] + pending > cap) {
      revert MaxWorkflowsPerUserDONExceeded(owner, donFamily);
    }
  }

  /// @dev Internal helper that converts a storage record (indexed by RID)
  ///      into an in‑memory `WorkflowMetadataView`.
  function _workflowMetadataView(
    bytes32 rid
  ) internal view returns (WorkflowMetadataView memory v) {
    WorkflowMetadata storage rec = s_workflows[rid];
    if (rec.owner == address(0)) return v;

    return WorkflowMetadataView({
      workflowId: rec.workflowId,
      owner: rec.owner,
      createdAt: rec.createdAt,
      status: rec.status,
      workflowName: rec.workflowName,
      binaryUrl: rec.binaryUrl,
      configUrl: rec.configUrl,
      tag: rec.tag,
      attributes: rec.attributes,
      // For ACTIVE workflows this will resolve to the correct DON label.
      // For PAUSED/never‑assigned workflows the label is the empty string.;
      donFamily: s_donConfigs[s_donByWorkflowRid[rid]].family
    });
  }

  /// @dev Internal helper to get a workflow record from storage based on the RID. It also checks if the caller is the
  /// owner of the workflow and that the owner is not 0.
  /// @param rid The RID of the workflow to get.
  /// @return rec The workflow record.
  /// @custom:revert WorkflowDoesNotExist If the workflow does not exist.
  /// @custom:revert CallerIsNotWorkflowOwner If the caller is not the owner of the workflow.
  function _getRecord(address owner, bytes32 rid) internal view returns (WorkflowMetadata storage rec) {
    rec = s_workflows[rid];
    if (rec.owner == address(0)) revert WorkflowDoesNotExist();
    if (rec.owner != owner) revert CallerIsNotWorkflowOwner(owner);
    return rec;
  }

  /// @notice Computes a unique workflow key for indexing by owner and name.
  /// @param owner Address of the workflow owner.
  /// @param name  Human-readable name of the workflow.
  /// @return key  Keccak256 hash combining `owner` and `name`.
  function _workflowKey(address owner, string memory name) internal pure returns (bytes32 key) {
    return keccak256(abi.encode(owner, name));
  }

  /// @notice Hashes an arbitrary-length DON label string into a fixed-size bytes32.
  /// @param str   DON label as a UTF-8 string.
  /// @return hash Keccak256 hash of `bytes(str)`.
  function _hash(
    string memory str
  ) internal pure returns (bytes32 hash) {
    return keccak256(bytes(str));
  }
}