<?php

declare(strict_types=1);

namespace Drupal\flowdrop_pipeline\Entity;

use Drupal\Component\Serialization\Json;
use Drupal\Core\Entity\Attribute\ContentEntityType;
use Drupal\Core\Entity\ContentEntityBase;
use Drupal\Core\Entity\ContentEntityDeleteForm;
use Drupal\Core\Entity\EntityChangedTrait;
use Drupal\Core\Entity\EntityStorageInterface;
use Drupal\Core\Entity\EntityTypeInterface;
use Drupal\Core\Entity\Form\DeleteMultipleForm;
use Drupal\Core\Entity\Routing\AdminHtmlRouteProvider;
use Drupal\Core\Field\BaseFieldDefinition;
use Drupal\Core\Field\FieldStorageDefinitionInterface;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\FlowDropPipelineAccessControlHandler;
use Drupal\flowdrop_pipeline\FlowDropPipelineListBuilder;
use Drupal\flowdrop_pipeline\Form\FlowDropPipelineForm;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;
use Drupal\Core\Logger\LoggerChannelTrait;
use Drupal\user\EntityOwnerTrait;
use Drupal\views\EntityViewsData;

/**
 * FlowDrop Pipeline Entity.
 *
 * @internal
 *   This class is not part of the public API and may change without notice.
 */
#[ContentEntityType(
  id: 'flowdrop_pipeline',
  label: new TranslatableMarkup('FlowDrop Pipeline'),
  label_collection: new TranslatableMarkup('FlowDrop Pipelines'),
  label_singular: new TranslatableMarkup('flowdrop pipeline'),
  label_plural: new TranslatableMarkup('flowdrop pipelines'),
  entity_keys: [
    'id' => 'id',
    'bundle' => 'bundle',
    'label' => 'label',
    'owner' => 'uid',
    'published' => 'status',
    'uuid' => 'uuid',
  ],
  handlers: [
    'list_builder' => FlowDropPipelineListBuilder::class,
    'views_data' => EntityViewsData::class,
    'access' => FlowDropPipelineAccessControlHandler::class,
    'form' => [
      'add' => FlowDropPipelineForm::class,
      'edit' => FlowDropPipelineForm::class,
      'delete' => ContentEntityDeleteForm::class,
      'delete-multiple-confirm' => DeleteMultipleForm::class,
    ],
    'route_provider' => [
      'html' => AdminHtmlRouteProvider::class,
    ],
  ],
  links: [
    'collection' => '/admin/flowdrop/pipelines',
    'add-form' => '/admin/flowdrop/pipelines/add/{flowdrop_pipeline_type}',
    'add-page' => '/admin/flowdrop/pipelines/add',
    'canonical' => '/admin/flowdrop/pipelines/{flowdrop_pipeline}',
    'edit-form' => '/admin/flowdrop/pipelines/{flowdrop_pipeline}/edit',
    'delete-form' => '/admin/flowdrop/pipelines/{flowdrop_pipeline}/delete',
    'delete-multiple-form' => '/admin/flowdrop/pipelines/delete-multiple',
  ],
  admin_permission: 'administer flowdrop_pipeline types',
  bundle_entity_type: 'flowdrop_pipeline_type',
  bundle_label: new TranslatableMarkup('FlowDrop Pipeline type'),
  base_table: 'flowdrop_pipeline',
  label_count: [
    'singular' => '@count flowdrop pipelines',
    'plural' => '@count flowdrop pipelines',
  ],
  field_ui_base_route: 'entity.flowdrop_pipeline_type.edit_form',
)]
class FlowDropPipeline extends ContentEntityBase implements FlowDropPipelineInterface {

  use EntityChangedTrait;
  use EntityOwnerTrait;
  use LoggerChannelTrait;

  /**
   * {@inheritdoc}
   */
  public static function baseFieldDefinitions(EntityTypeInterface $entity_type) {
    $fields = parent::baseFieldDefinitions($entity_type);

    $fields['label'] = BaseFieldDefinition::create('string')
      ->setLabel(t('Label'))
      ->setRequired(TRUE)
      ->setSetting('max_length', 255)
      ->setDisplayOptions('form', [
        'type' => 'string_textfield',
        'weight' => -5,
      ])
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayOptions('view', [
        'label' => 'hidden',
        'type' => 'string',
        'weight' => -5,
      ])
      ->setDisplayConfigurable('view', TRUE);

    $fields['status'] = BaseFieldDefinition::create('boolean')
      ->setLabel(t('Status'))
      ->setDefaultValue(TRUE)
      ->setSetting('on_label', 'Enabled')
      ->setDisplayOptions('form', [
        'type' => 'boolean_checkbox',
        'settings' => [
          'display_label' => FALSE,
        ],
        'weight' => 0,
      ])
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayOptions('view', [
        'type' => 'boolean',
        'label' => 'above',
        'weight' => 0,
        'settings' => [
          'format' => 'enabled-disabled',
        ],
      ])
      ->setDisplayConfigurable('view', TRUE);

    $fields['description'] = BaseFieldDefinition::create('string_long')
      ->setLabel(t('Description'));

    $fields['uid'] = BaseFieldDefinition::create('entity_reference')
      ->setLabel(t('Author'))
      ->setSetting('target_type', 'user')
      ->setDefaultValueCallback(self::class . '::getDefaultEntityOwner')
      ->setDisplayOptions('form', [
        'type' => 'entity_reference_autocomplete',
        'settings' => [
          'match_operator' => 'CONTAINS',
          'size' => 60,
          'placeholder' => '',
        ],
        'weight' => 15,
      ])
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayOptions('view', [
        'label' => 'above',
        'type' => 'author',
        'weight' => 15,
      ])
      ->setDisplayConfigurable('view', TRUE);

    $fields['created'] = BaseFieldDefinition::create('created')
      ->setLabel(t('Authored on'))
      ->setDescription(t('The time that the flowdrop pipeline was created.'))
      ->setDisplayOptions('view', [
        'label' => 'above',
        'type' => 'timestamp',
        'weight' => 20,
      ])
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayOptions('form', [
        'type' => 'datetime_timestamp',
        'weight' => 20,
      ])
      ->setDisplayConfigurable('view', TRUE);

    $fields['changed'] = BaseFieldDefinition::create('changed')
      ->setLabel(t('Changed'))
      ->setDescription(t('The time that the flowdrop pipeline was last edited.'))
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Workflow ID - References the workflow this pipeline is based on.
    $fields['workflow_id'] = BaseFieldDefinition::create('entity_reference')
      ->setLabel(t('Workflow ID'))
      ->setDescription(t('The ID of the workflow this pipeline is based on.'))
      ->setRequired(TRUE)
      ->setSetting('target_type', 'flowdrop_workflow')
      ->setSetting('handler', 'default')
      ->setDisplayOptions('view', [
        'label' => 'hidden',
        'type' => 'author',
        'weight' => 0,
      ])
      ->setDisplayOptions('form', [
        'type' => 'entity_reference_autocomplete',
        'weight' => 5,
        'settings' => [
          'match_operator' => 'CONTAINS',
          'size' => '60',
          'autocomplete_type' => 'tags',
          'placeholder' => '',
        ],
      ])
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Pipeline ID - References the pipeline this job belongs to.
    $fields['job_id'] = BaseFieldDefinition::create('entity_reference')
      ->setLabel(t('Job ID'))
      ->setDescription(t('The ID of the job.'))
      ->setRequired(TRUE)
      ->setSetting('target_type', 'flowdrop_job')
      ->setCardinality(FieldStorageDefinitionInterface::CARDINALITY_UNLIMITED)
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Status - Current status of the pipeline.
    $fields['status'] = BaseFieldDefinition::create('list_string')
      ->setLabel(t('Status'))
      ->setDescription(t('The current status of the pipeline.'))
      ->setRequired(TRUE)
      ->setDefaultValue(PipelineStatus::PENDING)
      ->setSettings([
        'allowed_values' => PipelineStatus::allowedValues(),
      ])
      ->setDisplayOptions('view', [
        'label' => 'above',
        'type' => 'list_default',
        'weight' => -4,
      ])
      ->setDisplayOptions('form', [
        'type' => 'options_select',
        'weight' => -4,
      ])
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Input Data - JSON field for pipeline input data.
    $fields['input_data'] = BaseFieldDefinition::create('string_long')
      ->setLabel(t('Input Data'))
      ->setDescription(t('The input data for this pipeline (JSON format).'))
      ->setDefaultValue('{}')
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Output Data - JSON field for pipeline output data.
    $fields['output_data'] = BaseFieldDefinition::create('string_long')
      ->setLabel(t('Output Data'))
      ->setDescription(t('The output data from this pipeline (JSON format).'))
      ->setDefaultValue('{}')
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Error Message - Error details if pipeline failed.
    $fields['error_message'] = BaseFieldDefinition::create('string_long')
      ->setLabel(t('Error Message'))
      ->setDescription(t('Error message if the pipeline failed.'))
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Started Time - When the pipeline started execution.
    $fields['started'] = BaseFieldDefinition::create('created')
      ->setLabel(t('Started'))
      ->setDescription(t('When the pipeline started execution.'))
      ->setDefaultValue(NULL)
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Completed Time - When the pipeline completed execution.
    $fields['completed'] = BaseFieldDefinition::create('created')
      ->setLabel(t('Completed'))
      ->setDescription(t('When the pipeline completed execution.'))
      ->setDefaultValue(NULL)
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Max Concurrent Jobs - Maximum number of jobs to run simultaneously.
    $fields['max_concurrent_jobs'] = BaseFieldDefinition::create('integer')
      ->setLabel(t('Max Concurrent Jobs'))
      ->setDescription(t('Maximum number of jobs to run simultaneously.'))
      ->setDefaultValue(5)
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Job Priority Strategy - How to prioritize jobs.
    $fields['job_priority_strategy'] = BaseFieldDefinition::create('list_string')
      ->setLabel(t('Job Priority Strategy'))
      ->setDescription(t('How to prioritize jobs in this pipeline.'))
      ->setDefaultValue('dependency_order')
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Retry Strategy - How to handle job failures.
    $fields['retry_strategy'] = BaseFieldDefinition::create('list_string')
      ->setLabel(t('Retry Strategy'))
      ->setDescription(t('How to handle job failures in this pipeline.'))
      ->setDefaultValue('individual')
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Execution Context - JSON field for pipeline execution context.
    $fields['execution_context'] = BaseFieldDefinition::create('string_long')
      ->setLabel(t('Execution Context'))
      ->setDescription(t('Execution context data for the pipeline (JSON format).'))
      ->setDefaultValue('{}')
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // The session_id field is provided by flowdrop_session via
    // hook_entity_base_field_info() to avoid a circular dependency:
    // flowdrop_session depends on flowdrop_pipeline, so the pipeline
    // module cannot reference flowdrop_session here without making
    // installation order impossible.
    //
    // Pipeline lineage. parent_pipeline_id is NULL for a top-level run and
    // points at the spawning pipeline for a nested (shared-session
    // sub-workflow) run. root_pipeline_id is the top of the chain and is
    // always populated: a root pipeline stores its own id (stamped in
    // postSave() on insert), a nested one inherits its parent's root. This
    // lets every message in a run be correlated to the run's root pipeline
    // without walking the chain.
    $fields['parent_pipeline_id'] = BaseFieldDefinition::create('integer')
      ->setLabel(t('Parent Pipeline ID'))
      ->setDescription(t('The pipeline that spawned this one, or NULL for a top-level run.'))
      ->setDisplayConfigurable('view', TRUE);

    $fields['root_pipeline_id'] = BaseFieldDefinition::create('integer')
      ->setLabel(t('Root Pipeline ID'))
      ->setDescription(t('The top-level pipeline of this run. A root pipeline stores its own id.'))
      ->setDisplayConfigurable('view', TRUE);

    // Pending Interrupt ID - UUID of the interrupt this pipeline is paused
    // on. NULL when the pipeline is not paused. Lifted off session metadata
    // so concurrent sub-pipelines in the same session can each carry their
    // own pending state.
    $fields['pending_interrupt_id'] = BaseFieldDefinition::create('string')
      ->setLabel(t('Pending Interrupt ID'))
      ->setDescription(t('UUID of the interrupt this pipeline is paused on.'))
      ->setSetting('max_length', 128)
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    // Pending Interrupt Type - String form of the InterruptType enum.
    // Stored alongside the id so resume callers can dispatch without
    // loading the interrupt entity first.
    $fields['pending_interrupt_type'] = BaseFieldDefinition::create('string')
      ->setLabel(t('Pending Interrupt Type'))
      ->setDescription(t('Type of the interrupt this pipeline is paused on.'))
      ->setSetting('max_length', 64)
      ->setDisplayConfigurable('form', TRUE)
      ->setDisplayConfigurable('view', TRUE);

    return $fields;
  }

  /**
   * {@inheritDoc}
   */
  public function getWorkflowId(): string {
    return $this->get('workflow_id')->target_id;
  }

  /**
   * {@inheritDoc}
   */
  public function setWorkflowId(string $workflow_id): static {
    $this->set('workflow_id', ['target_id' => $workflow_id]);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getStatus(): string {
    return $this->get('status')->value;
  }

  /**
   * {@inheritDoc}
   */
  public function setStatus(string $status): static {
    $this->set('status', $status);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getInputData(): array {
    $data = $this->get('input_data')->value;
    if ($data === NULL || $data === '') {
      return [];
    }
    $decoded = Json::decode($data);
    return is_array($decoded) ? $decoded : [];
  }

  /**
   * {@inheritDoc}
   */
  public function setInputData(array $input_data): static {
    $this->set('input_data', Json::encode($input_data));
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getOutputData(): array {
    $data = $this->get('output_data')->value;
    if ($data === NULL || $data === '') {
      return [];
    }
    $decoded = Json::decode($data);
    return is_array($decoded) ? $decoded : [];
  }

  /**
   * {@inheritDoc}
   */
  public function setOutputData(array $output_data): static {
    $this->set('output_data', Json::encode($output_data));
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getErrorMessage(): string {
    return $this->get('error_message')->value ?? '';
  }

  /**
   * {@inheritDoc}
   */
  public function setErrorMessage(string $error_message): static {
    $this->set('error_message', $error_message);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getStarted(): int {
    return (int) $this->get('started')->value;
  }

  /**
   * {@inheritDoc}
   */
  public function setStarted(int $started): static {
    $this->set('started', $started);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getCompleted(): int {
    return (int) $this->get('completed')->value;
  }

  /**
   * {@inheritDoc}
   */
  public function setCompleted(int $completed): static {
    $this->set('completed', $completed);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getMaxConcurrentJobs(): int {
    return (int) $this->get('max_concurrent_jobs')->value;
  }

  /**
   * {@inheritDoc}
   */
  public function setMaxConcurrentJobs(int $max_concurrent_jobs): static {
    $this->set('max_concurrent_jobs', $max_concurrent_jobs);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getJobPriorityStrategy(): string {
    return $this->get('job_priority_strategy')->value;
  }

  /**
   * {@inheritDoc}
   */
  public function setJobPriorityStrategy(string $strategy): static {
    $this->set('job_priority_strategy', $strategy);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getRetryStrategy(): string {
    return $this->get('retry_strategy')->value;
  }

  /**
   * {@inheritDoc}
   */
  public function setRetryStrategy(string $strategy): static {
    $this->set('retry_strategy', $strategy);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getExecutionContext(): array {
    $data = $this->get('execution_context')->value;
    if ($data === NULL || $data === '') {
      return [];
    }
    $decoded = Json::decode($data);
    return is_array($decoded) ? $decoded : [];
  }

  /**
   * {@inheritDoc}
   */
  public function setExecutionContext(array $context): static {
    $this->set('execution_context', Json::encode($context));
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function isReadyToStart(): bool {
    return $this->getStatus() === PipelineStatus::PENDING;
  }

  /**
   * {@inheritDoc}
   */
  public function isRunning(): bool {
    return $this->getStatus() === PipelineStatus::RUNNING;
  }

  /**
   * {@inheritDoc}
   */
  public function isCompleted(): bool {
    return $this->getStatus() === PipelineStatus::COMPLETED;
  }

  /**
   * {@inheritDoc}
   */
  public function isFailed(): bool {
    return $this->getStatus() === PipelineStatus::FAILED;
  }

  /**
   * {@inheritDoc}
   */
  public function isCancelled(): bool {
    return $this->getStatus() === PipelineStatus::CANCELLED;
  }

  /**
   * {@inheritDoc}
   */
  public function isPaused(): bool {
    return $this->getStatus() === PipelineStatus::PAUSED;
  }

  /**
   * {@inheritDoc}
   */
  public function markAsStarted(): static {
    $this->setStatus(PipelineStatus::RUNNING)
      ->setStarted(time());
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function markAsCompleted(array $output_data = []): static {
    $this->setStatus(PipelineStatus::COMPLETED)
      ->setCompleted(time())
      ->setOutputData($output_data);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function markAsFailed(string $error_message = ''): static {
    $this->setStatus(PipelineStatus::FAILED)
      ->setCompleted(time())
      ->setErrorMessage($error_message);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function markAsCancelled(): static {
    $this->setStatus(PipelineStatus::CANCELLED)
      ->setCompleted(time());
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function pause(): static {
    $this->setStatus(PipelineStatus::PAUSED);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function resume(): static {
    if ($this->getStatus() === PipelineStatus::PAUSED) {
      $this->setStatus(PipelineStatus::RUNNING);
    }
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getWorkflow(): ?WorkflowDefinitionInterface {
    try {
      $workflow = $this->get('workflow_id')->entity;
      return $workflow instanceof WorkflowDefinitionInterface ? $workflow : NULL;
    }
    catch (\Exception $e) {
      $this->getLogger('flowdrop_pipeline')->error('Failed to load workflow for pipeline @id: @message', [
        '@id' => $this->id(),
        '@message' => $e->getMessage(),
      ]);
      return NULL;
    }
  }

  /**
   * {@inheritdoc}
   */
  public function getJobs(): array {
    $output = [];
    try {
      $job_field = $this->get('job_id');
      if ($job_field->isEmpty()) {
        return [];
      }

      $job_storage = $this->entityTypeManager()->getStorage('flowdrop_job');
      $job_ids = [];

      foreach ($job_field as $item) {
        if (!empty($item->target_id)) {
          $job_ids[] = $item->target_id;
        }
      }

      if (empty($job_ids)) {
        return [];
      }

      $jobs = $job_storage->loadMultiple($job_ids);

      // Sort by priority and created date.
      usort($jobs, function ($a, $b) {
        if (!$a instanceof FlowDropJobInterface || !$b instanceof FlowDropJobInterface) {
          return 0;
        }
        $priority_comparison = $a->getPriority() <=> $b->getPriority();
        if ($priority_comparison !== 0) {
          return $priority_comparison;
        }
        return (int) $a->get('created')->value <=> (int) $b->get('created')->value;
      });

      foreach ($jobs as $job) {
        if ($job instanceof FlowDropJobInterface) {
          $output[] = $job;
        }
      }
      return $output;
    }
    catch (\Exception $e) {
      $this->getLogger('flowdrop_pipeline')->error('Failed to load jobs for pipeline @id: @message', [
        '@id' => $this->id(),
        '@message' => $e->getMessage(),
      ]);
      return [];
    }
  }

  /**
   * {@inheritdoc}
   */
  public function getJobsByStatus(string $status): array {
    $all_jobs = $this->getJobs();
    $filtered_jobs = [];

    foreach ($all_jobs as $job) {
      if ($job->getStatus() === $status) {
        $filtered_jobs[] = $job;
      }
    }

    return $filtered_jobs;
  }

  /**
   * {@inheritDoc}
   */
  public function hasRunningJobs(): bool {
    $running_jobs = $this->getJobsByStatus(JobStatus::RUNNING);
    return !empty($running_jobs);
  }

  /**
   * {@inheritDoc}
   */
  public function hasPendingJobs(): bool {
    $pending_jobs = $this->getJobsByStatus(JobStatus::PENDING);
    return !empty($pending_jobs);
  }

  /**
   * Check if the pipeline has any idle jobs.
   *
   * Idle jobs are jobs that have been created but are not yet ready to run
   * (dependencies not met).
   *
   * @return bool
   *   TRUE if the pipeline has idle jobs.
   */
  public function hasIdleJobs(): bool {
    $idle_jobs = $this->getJobsByStatus(JobStatus::IDLE);
    return !empty($idle_jobs);
  }

  /**
   * Check if the pipeline has any jobs that haven't finished yet.
   *
   * This includes idle, pending, running, and interrupted jobs.
   *
   * @return bool
   *   TRUE if the pipeline has unfinished jobs.
   */
  public function hasUnfinishedJobs(): bool {
    $jobs = $this->getJobs();
    foreach ($jobs as $job) {
      $status = $job->getStatus();
      if (in_array($status, JobStatus::activeStatuses(), TRUE)) {
        return TRUE;
      }
    }
    return FALSE;
  }

  /**
   * Check if the pipeline has any interrupted jobs.
   *
   * Interrupted jobs are waiting for human-in-the-loop input.
   *
   * @return bool
   *   TRUE if the pipeline has interrupted jobs.
   */
  public function hasInterruptedJobs(): bool {
    $interrupted_jobs = $this->getJobsByStatus(JobStatus::INTERRUPTED);
    return !empty($interrupted_jobs);
  }

  /**
   * Add a job to this pipeline.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to add.
   *
   * @return static
   *   The pipeline entity.
   */
  public function addJob(FlowDropJobInterface $job): static {
    $current_jobs = $this->get('job_id')->getValue();
    $job_ids = array_column($current_jobs, 'target_id');

    // Only add if not already present.
    if (!in_array($job->id(), $job_ids)) {
      $this->get('job_id')->appendItem(['target_id' => $job->id()]);
    }

    return $this;
  }

  /**
   * Remove a job from this pipeline.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to remove.
   *
   * @return static
   *   The pipeline entity.
   */
  public function removeJob(FlowDropJobInterface $job): static {
    $job_field = $this->get('job_id');
    $items_to_remove = [];

    foreach ($job_field as $delta => $item) {
      $item_value = $item->getValue();
      if (isset($item_value['target_id']) && (string) $item_value['target_id'] === (string) $job->id()) {
        $items_to_remove[] = $delta;
      }
    }

    // Remove in reverse order to maintain indexes.
    foreach (array_reverse($items_to_remove) as $delta) {
      $job_field->removeItem($delta);
    }

    return $this;
  }

  /**
   * Clear all jobs from this pipeline.
   *
   * @return static
   *   The pipeline entity.
   */
  public function clearJobs(): static {
    $this->get('job_id')->setValue([]);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function allJobsCompleted(): bool {
    $jobs = $this->getJobs();
    if (empty($jobs)) {
      return FALSE;
    }

    // A job is considered "not complete" if it's in a non-terminal state:
    // - idle: Job hasn't started yet (waiting for dependencies).
    // - pending: Job is ready to run but hasn't started.
    // - running: Job is currently executing.
    // - interrupted: Job is waiting for human-in-the-loop input.
    // Terminal states (completed, failed, cancelled, skipped) are considered
    // done and don't block this check.
    foreach ($jobs as $job) {
      $status = $job->getStatus();
      if (in_array($status, JobStatus::activeStatuses(), TRUE)) {
        return FALSE;
      }
    }

    return TRUE;
  }

  /**
   * {@inheritDoc}
   */
  public function hasFailedJobs(): bool {
    $failed_jobs = $this->getJobsByStatus(JobStatus::FAILED);
    return !empty($failed_jobs);
  }

  /**
   * Calculate job counts dynamically.
   *
   * @return array<string, int>
   *   Array with job counts by status.
   */
  public function calculateJobCounts(): array {
    $jobs = $this->getJobs();
    $counts = [
      'total' => 0,
      JobStatus::IDLE => 0,
      JobStatus::PENDING => 0,
      JobStatus::RUNNING => 0,
      JobStatus::COMPLETED => 0,
      JobStatus::FAILED => 0,
      JobStatus::CANCELLED => 0,
      JobStatus::SKIPPED => 0,
      JobStatus::INTERRUPTED => 0,
    ];

    foreach ($jobs as $job) {
      $status = $job->getStatus();
      $counts['total']++;
      if (isset($counts[$status])) {
        $counts[$status]++;
      }
    }

    return $counts;
  }

  /**
   * {@inheritDoc}
   */
  public function getSnapshotId(): ?string {
    $value = $this->get('snapshot_id')->target_id;
    return $value !== NULL ? (string) $value : NULL;
  }

  /**
   * {@inheritDoc}
   */
  public function setSnapshotId(?string $snapshot_id): static {
    if ($snapshot_id === NULL) {
      $this->set('snapshot_id', NULL);
    }
    else {
      $this->set('snapshot_id', ['target_id' => $snapshot_id]);
    }
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getSnapshot(): ?object {
    try {
      $snapshot = $this->get('snapshot_id')->entity;
      return $snapshot ?? NULL;
    }
    catch (\Exception $e) {
      $this->getLogger('flowdrop_pipeline')->error('Failed to load snapshot for pipeline @id: @message', [
        '@id' => $this->id(),
        '@message' => $e->getMessage(),
      ]);
      return NULL;
    }
  }

  /**
   * {@inheritDoc}
   */
  public function getSessionId(): ?string {
    // The session_id field is provided by flowdrop_session. Tolerate it
    // being absent so callers don't blow up on pipeline-only installs or
    // after flowdrop_session is uninstalled.
    if (!$this->hasField('session_id')) {
      return NULL;
    }
    $value = $this->get('session_id')->target_id;
    return $value !== NULL ? (string) $value : NULL;
  }

  /**
   * {@inheritDoc}
   */
  public function setSessionId(?string $session_id): static {
    // No-op when flowdrop_session is not enabled — the field doesn't exist.
    if (!$this->hasField('session_id')) {
      return $this;
    }
    if ($session_id === NULL || $session_id === '') {
      $this->set('session_id', NULL);
    }
    else {
      $this->set('session_id', ['target_id' => $session_id]);
    }
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getParentPipelineId(): ?string {
    $value = $this->get('parent_pipeline_id')->value;
    return $value !== NULL ? (string) $value : NULL;
  }

  /**
   * {@inheritDoc}
   */
  public function setParentPipelineId(?string $parent_pipeline_id): static {
    $this->set('parent_pipeline_id', ($parent_pipeline_id === NULL || $parent_pipeline_id === '') ? NULL : (int) $parent_pipeline_id);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getRootPipelineId(): ?string {
    $value = $this->get('root_pipeline_id')->value;
    return $value !== NULL ? (string) $value : NULL;
  }

  /**
   * {@inheritDoc}
   */
  public function setRootPipelineId(?string $root_pipeline_id): static {
    $this->set('root_pipeline_id', ($root_pipeline_id === NULL || $root_pipeline_id === '') ? NULL : (int) $root_pipeline_id);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getPendingInterruptId(): ?string {
    $value = $this->get('pending_interrupt_id')->value;
    return $value !== NULL && $value !== '' ? (string) $value : NULL;
  }

  /**
   * {@inheritDoc}
   */
  public function setPendingInterruptId(?string $interrupt_id): static {
    $this->set('pending_interrupt_id', $interrupt_id ?: NULL);
    return $this;
  }

  /**
   * {@inheritDoc}
   */
  public function getPendingInterruptType(): ?string {
    $value = $this->get('pending_interrupt_type')->value;
    return $value !== NULL && $value !== '' ? (string) $value : NULL;
  }

  /**
   * {@inheritDoc}
   */
  public function setPendingInterruptType(?string $type): static {
    $this->set('pending_interrupt_type', $type ?: NULL);
    return $this;
  }

  /**
   * {@inheritdoc}
   */
  // phpcs:ignore Generic.CodeAnalysis.UselessOverridingMethod.Found -- adds return type for static analysis
  public function id(): int|string|null {
    return parent::id();
  }

  /**
   * {@inheritdoc}
   */
  // phpcs:ignore Generic.CodeAnalysis.UselessOverridingMethod.Found -- adds return type for static analysis
  public function uuid(): ?string {
    return parent::uuid();
  }

  /**
   * {@inheritdoc}
   */
  // phpcs:ignore Generic.CodeAnalysis.UselessOverridingMethod.Found -- adds return type for static analysis
  public function save(): int {
    return parent::save();
  }

  /**
   * {@inheritDoc}
   */
  public function preSave(EntityStorageInterface $storage): void {
    parent::preSave($storage);

    // Set default values if not set.
    if ($this->isNew()) {
      if (!$this->get('status')->value) {
        $this->setStatus(PipelineStatus::PENDING);
      }
      if (!$this->get('max_concurrent_jobs')->value) {
        $this->setMaxConcurrentJobs(5);
      }
      if (!$this->get('job_priority_strategy')->value) {
        $this->setJobPriorityStrategy('dependency_order');
      }
      if (!$this->get('retry_strategy')->value) {
        $this->setRetryStrategy('individual');
      }
    }
  }

  /**
   * {@inheritDoc}
   */
  public function postSave(EntityStorageInterface $storage, $update = TRUE): void {
    parent::postSave($storage, $update);

    // Self-root: a top-level pipeline's root is itself. The id only exists
    // once the insert assigns it, so a run that did not inherit a root from a
    // parent gets stamped here. A direct single-column write is used rather
    // than save() — calling save() from postSave() would recurse. The window
    // between the insert and this UPDATE is sub-millisecond and closes before
    // the caller creates any message referencing this pipeline, so a reader
    // never observes a NULL root. The in-memory entity is updated too so the
    // value is visible without a reload.
    if ($update === FALSE && $this->get('root_pipeline_id')->value === NULL) {
      $id = (int) $this->id();
      $this->set('root_pipeline_id', $id);
      \Drupal::database()
        ->update($this->getEntityType()->getBaseTable())
        ->fields(['root_pipeline_id' => $id])
        ->condition('id', $id)
        ->execute();
    }
  }

}
