<?php
/**
 * @author Jaap Jansma <jaap.jansma@civicoop.org>
 * @license AGPL-3.0
 */

namespace Civi\DataProcessor\DataFlow\CombinedDataFlow;

use Civi\DataProcessor\DataFlow\AbstractDataFlow;
use Civi\DataProcessor\DataFlow\MultipleDataFlows\JoinInterface;
use Civi\DataProcessor\DataFlow\MultipleDataFlows\SimpleJoin;
use Civi\DataProcessor\DataFlow\SqlDataFlow;
use Civi\DataProcessor\Exception\DataFlowException;
use Civi\DataProcessor\DataFlow\EndOfFlowException;
use Civi\DataProcessor\DataFlow\MultipleDataFlows\DataFlowDescription;
use Civi\DataProcessor\DataFlow\MultipleDataFlows\MultipleSourceDataFlows;
use Civi\DataProcessor\DataSpecification\DataSpecification;


class CombinedDataFlow extends AbstractDataFlow implements MultipleSourceDataFlows {

  /**
   * @var DataFlowDescription[]
   */
  protected $sourceDataFlowDescriptions = array();

  /**
   * @var int
   */
  protected $currentRecordIndex = 0;

  /**
   * @var array
   */
  protected $recordSet = array();

  /**
   * @var int
   */
  protected $recordCount = 0;

  /**
   * @var bool
   */
  protected $initialized = false;

  /**
   * @var \Civi\DataProcessor\DataSpecification\DataSpecification
   */
  protected $dataSpecification;

  /**
   * @var int
   */
  protected $batchSize = 100;

  /**
   * @var array
   */
  protected $dataFlows = [];

  public function __construct() {
    parent::__construct();
    $this->dataSpecification = new DataSpecification();
  }

  /**
   * Adds a source data flow
   *
   * @param \Civi\DataProcessor\DataFlow\MultipleDataFlows\DataFlowDescription $dataFlowDescription
   * @return void
   */
  public function addSourceDataFlow(DataFlowDescription $dataFlowDescription) {
    $this->sourceDataFlowDescriptions[] = $dataFlowDescription;
  }

  /**
   * Removes a source data flow
   *
   * @param \Civi\DataProcessor\DataFlow\MultipleDataFlows\DataFlowDescription $dataFlowDescription
   * @return void
   */
  public function removeSourceDataFlow(DataFlowDescription $dataFlowDescription) {
    foreach($this->sourceDataFlowDescriptions as $idx => $sourceDataFlowDescription) {
      if ($sourceDataFlowDescription === $dataFlowDescription) {
        unset($this->sourceDataFlowDescriptions[$idx]);
      }
    }
  }

  /**
   * @param \Civi\DataProcessor\FieldOutputHandler\AbstractFieldOutputHandler $handlers[]
   */
  public function setOutputFieldHandlers($handlers) {
    parent::setOutputFieldHandlers($handlers);
    foreach($this->sourceDataFlowDescriptions as $sourceDataFlowDescription) {
      $sourceDataFlowDescription->getDataFlow()->setOutputFieldHandlers($handlers);
    }
  }

  /**
   * Initialize the data flow
   *
   * @return void
   */
  public function initialize() {
    if ($this->isInitialized()) {
      return;
    }

    $this->recordSet = array();
    $this->recordCount = 0;

    if (count($this->sourceDataFlowDescriptions) < 1) {
      $this->initialized = true;
      return;
    }

    $allRecords = array();
    for($i=0; $i<count($this->sourceDataFlowDescriptions); $i++) {
      $joinSpec = $this->sourceDataFlowDescriptions[$i]->getJoinSpecification();
      if ($joinSpec) {
        $joinSpec->ensureFields();
      }
      $dataFlow = $this->sourceDataFlowDescriptions[$i]->getDataFlow();
      if ($dataFlow instanceof SqlDataFlow) {
        $dataFlow->convertJoinClausesToWhereClause();
      }
    }
    for($i=0; $i<count($this->sourceDataFlowDescriptions); $i++) {
      do {
        $batch = [];
        try {
          $batch = $this->getAllRecordsFromDataFlowAsArray($this->sourceDataFlowDescriptions[$i]->getDataFlow(), $this->batchSize);
          for($j=$i+1; $j<count($this->sourceDataFlowDescriptions); $j++) {
            $this->sourceDataFlowDescriptions[$j]->getJoinSpecification()->prepareRightDataFlow($batch, $this->sourceDataFlowDescriptions[$j]->getDataFlow());
            $rightRecords = $this->getAllRecordsFromDataFlowAsArray($this->sourceDataFlowDescriptions[$j]->getDataFlow());
            $batch = $this->sourceDataFlowDescriptions[$j]->getJoinSpecification()->join($batch, $rightRecords);
          }
          $allRecords = array_merge($allRecords, $batch);
        } catch (DataFlowException $e) {
        }
      } while(count($batch) >= $this->batchSize || $this->batchSize == 0);
    }
    $this->recordCount = count($allRecords);

    if ($this->offset || $this->limit) {
      $offset = $this->offset !== FALSE ? $this->offset : 0;
      for ($i = $offset; $i < count($allRecords); $i++) {
        $this->recordSet[] = $allRecords[$i];
        if ($this->limit && count($this->recordSet) >= $this->limit) {
          break;
        }
      }
    } else {
      $this->recordSet = $allRecords;
    }

    $this->initialized = true;
  }

  /**
   * Return all records for a given data flow.
   *
   * @param \Civi\DataProcessor\DataFlow\AbstractDataFlow $dataFlow
   * @param int $batchSize 0 for unlimited
   *
   * @return array
   * @throws DataFlowException
   */
  protected function getAllRecordsFromDataFlowAsArray(AbstractDataFlow $dataFlow, int $batchSize=0): array {
    $records = array();
    $i = 0;
    try {
      while (($record = $dataFlow->retrieveNextRecord()) && ($i < $batchSize || $batchSize == 0)) {
        $records[] = $record;
        $i++;
      }
    } catch (EndOfFlowException $e) {
      // Do nothing
    }
    $records = array_unique($records, SORT_REGULAR);
    return $records;
  }

  /**
   * Returns whether this flow has been initialized or not
   *
   * @return bool
   */
  public function isInitialized(): bool {
    return $this->initialized;
  }

  /**
   * Resets the initialized state. This function is called
   * when a setting has changed. E.g. when offset or limit are set.
   *
   * @return void
   */
  public function resetInitializeState() {
    parent::resetInitializeState();
    $this->initialized = false;
  }

  /**
   * Returns the next record in an associative array
   *
   * @param string $fieldNameprefix
   *   The prefix before the name of the field within the record
   *
   * @return array
   * @throws EndOfFlowException
   */
  public function retrieveNextRecord($fieldNameprefix=''): array {
    if (!$this->isInitialized()) {
      $this->initialize();
    }

    if (!isset($this->recordSet[$this->currentRecordIndex])) {
      throw new EndOfFlowException();
    }
    $record = $this->recordSet[$this->currentRecordIndex];
    $out = array();
    if (strlen($fieldNameprefix)) {
      foreach ($record as $field => $value) {
        $out[$fieldNameprefix . $field] = $value;
      }
    } else {
      $out = $record;
    }
    $this->currentRecordIndex++;
    return $out;
  }

  /**
   * @return int
   */
  public function recordCount(): int {
    if (!$this->isInitialized()) {
      $this->initialize();
    }
    return $this->recordCount;
  }


  /**
   * @return DataSpecification
   * @throws \Civi\DataProcessor\DataSpecification\FieldExistsException
   */
  public function getDataSpecification(): DataSpecification {
    $dataSpecification = new DataSpecification();
    foreach($this->dataFlows as $dataFlow) {
      $namePrefix = $dataFlow['dataflow']->getName();
      $dataSpecification->merge($dataFlow['dataflow']->getDataSpecification(), $namePrefix);
    }
    return $dataSpecification;
  }

  public function getName(): string {
    return 'combined_data_flow';
  }

  public function getDebugInformation(): array {
    $debug = array();
    foreach($this->sourceDataFlowDescriptions as $sourceDataFlowDescription) {
      $debug[$sourceDataFlowDescription->getDataFlow()->getName()] = $sourceDataFlowDescription->getDataFlow()->getDebugInformation();
    }
    return $debug;

  }

}
