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

namespace Civi\DataProcessor\DataFlow;

use Civi\DataProcessor\DataFlow\Sort\SortSpecification;
use Civi\DataProcessor\DataFlow\SqlDataFlow\WhereClauseInterface;
use Civi\DataProcessor\DataSpecification\DataSpecification;
use Civi\DataProcessor\DataSpecification\FieldExistsException;
use Civi\DataProcessor\Exception\DataFlowException;
use Civi\DataProcessor\FieldOutputHandler\OutputHandlerAggregate;
use Civi\DataProcessor\Utils\Sql;
use CRM_Core_DAO;
use Exception;

abstract class SqlDataFlow extends AbstractDataFlow {

  /**
   * @var null|CRM_Core_DAO
   */
  protected $dao = null;

  /**
   * @var null|int
   */
  protected $count = null;

  protected $whereClauses = array();

  protected $sqlStatements = array();

  protected $sqlCountStatements = array();

  /**
   * @var float
   */
  protected $countQueryTime = 0.00;

  /**
   * @var float
   */
  protected $queryTime = 0.00;

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

  /**
   * Returns an array with the fields for in the select statement in the sql
   * query.
   *
   * @param bool $isCountQuery
   * @return string[]
   */
  abstract public function getFieldsForSelectStatement(bool $isCountQuery=false): array;

  /**
   * Returns an array with the fields for in the group by statement in the sql
   * query.
   *
   * @return string[]
   */
  public function getFieldsForGroupByStatement(): array {
    $fields = array();
    foreach($this->getGroupByDataSpecification()->getFields() as $fieldSpec) {
      $fields[] = $fieldSpec->getSqlGroupByStatement($this->getName());
    }
    return $fields;
  }

  /**
   * @param \Civi\DataProcessor\FieldOutputHandler\OutputHandlerAggregate $aggregateOutputHandler
   */
  public function addAggregateOutputHandler(OutputHandlerAggregate $aggregateOutputHandler) {
    parent::addAggregateOutputHandler($aggregateOutputHandler);
    $fieldSpec = $aggregateOutputHandler->getAggregateFieldSpec();
    try {
      $this->getGroupByDataSpecification()->addFieldSpecification($fieldSpec->name, $fieldSpec);
    } catch (FieldExistsException $e) {
      // Do nothing.
    }
  }

  /**
   * @param \Civi\DataProcessor\FieldOutputHandler\OutputHandlerAggregate $aggregateOutputHandler
   */
  public function removeAggregateOutputHandler(OutputHandlerAggregate $aggregateOutputHandler) {
    parent::removeAggregateOutputHandler($aggregateOutputHandler);
    $fieldSpec = $aggregateOutputHandler->getAggregateFieldSpec();
    $this->getGroupByDataSpecification()->removeFieldSpecification($fieldSpec);
  }

  /**
   * @return DataSpecification
   */
  public function getGroupByDataSpecification(): DataSpecification {
    if (!$this->groupByDataSpecification) {
      $this->groupByDataSpecification = new DataSpecification();
    }
    return $this->groupByDataSpecification;
  }

  /**
   * Returns the Table part in the from statement.
   *
   * @return string
   */
  abstract public function getTableStatement(): string;

  /**
   * Returns the From Statement.
   *
   * @return string
   */
  public function getFromStatement(): string {
    return "FROM {$this->getTableStatement()}";
  }

  /**
   * Initialize the data flow
   *
   * @throws \Civi\DataProcessor\Exception\DataFlowException
   */
  public function initialize() {
    if ($this->isInitialized()) {
      return;
    }

    $countSql = "";
    $sql = "";
    $this->sqlCountStatements = [];
    $this->sqlStatements = [];

    try {
      $selectAndFrom = $this->getSelectQueryStatement();
      $selectAndFromForCountQuery = $this->getSelectQueryStatement(true);
      $where = $this->getWhereStatement();
      $groupBy = $this->getGroupByStatement();
      $orderBy = $this->getOrderByStatement();
      $countName = 'count_'.$this->getName();
      $sql = "$selectAndFrom $where $groupBy $orderBy";
      $countSql = "SELECT COUNT(*) AS `count` FROM ($selectAndFromForCountQuery $where $groupBy) `$countName`";
      $this->sqlCountStatements[] = $countSql;
      $startTime = microtime(true);
      $countDao = CRM_Core_DAO::executeQuery($countSql, [], true, NULL, false, true);
      $this->count = 0;
      if (!is_a($countDao, 'DB_Error') && $countDao) {
        while ($countDao->fetch()) {
          $this->count = $this->count + $countDao->count;
        }
      }
      $this->countQueryTime = microtime(true) - $startTime;

      // Build Limit and Offset.
      $limitStatement = "";
      if ($this->offset !== FALSE && $this->limit !== FALSE) {
        $limitStatement = "LIMIT $this->offset, $this->limit";
      }
      elseif ($this->offset === FALSE && $this->limit !== FALSE) {
        $limitStatement = "LIMIT 0, $this->limit";
      }
      elseif ($this->offset !== FALSE && $this->limit === FALSE) {
        $calculatedLimit = $this->count - $this->offset;
        $limitStatement = "LIMIT $this->offset, $calculatedLimit";
      }
      $sql .= " $limitStatement";
      $this->sqlStatements[] = $sql;
      $startTime = microtime(true);
      $this->dao = CRM_Core_DAO::executeQuery($sql, [], true, NULL, false, true);
      $this->queryTime = microtime(true) - $startTime;
      if (is_a($this->dao, 'DB_Error') || !$this->dao) {
        throw new DataFlowException('Error in dataflow: '.$sql);
      }
    } catch (Exception $e) {
      throw new DataFlowException(
        "Error in DataFlow query.
        \r\nData flow: {$this->getName()}
        \r\nQuery: $sql
        \r\nCount query: $countSql");
    }
  }

  /**
   * Returns whether this flow has been initialized or not
   *
   * @return bool
   */
  public function isInitialized(): bool {
    if ($this->dao !== null) {
      return true;
    }
    return false;
  }

  /**
   * 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->dao = null;
    $this->queryTime = 0.00;
    $this->countQueryTime = 0.00;
    $this->sqlStatements = [];
    $this->sqlCountStatements = [];
  }

  /**
   * 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|DataFlowException
   */
  public function retrieveNextRecord($fieldNameprefix=''): array {
    if (!$this->isInitialized()) {
      $this->initialize();
    }

    if (!$this->dao->fetch()) {
      throw new EndOfFlowException();
    }
    $record = array();
    foreach($this->dataSpecification->getFields() as $field) {
      $alias = $field->alias;
      $record[$fieldNameprefix.$alias] = $this->dao->$alias;
    }
    return $record;
  }

  /**
   * @return int
   */
  public function recordCount(): int {
    if (!$this->isInitialized()) {
      try {
        $this->initialize();
      } catch (DataFlowException $e) {
        return 0;
      }
    }
    return $this->count;
  }

  /**
   * @param bool $isCountQuery
   *
   * @return string
   */
  public function getSelectQueryStatement(bool $isCountQuery = false): string {
    $select = implode(", ", $this->getFieldsForSelectStatement($isCountQuery));
    $from = $this->getFromStatement();
    return "SELECT DISTINCT $select $from";
  }

  public function getGroupByStatement(): string {
    $groupByFields = $this->getFieldsForGroupByStatement();
    if (!count($groupByFields)) {
      return "";
    }
    return "GROUP BY ".implode(", ", $groupByFields);
  }

  /**
   * Returns the where statement for this query.
   *
   * @return string
   */
  public function getWhereStatement(): string {
    $whereStatement = Sql::generateConditionStatement($this->getWhereClauses(FALSE, TRUE));
    if (strlen($whereStatement)) {
      return "WHERE ".$whereStatement;
    }
    return "";
  }

  public function convertJoinClausesToWhereClause(): void {
    /** @var WhereClauseInterface $c */
    foreach($this->whereClauses as $c) {
      if ($c->isJoinClause()) {
        $c->setJoinClause(FALSE);
      }
    }
  }

  /**
   * @param \Civi\DataProcessor\DataFlow\SqlDataFlow\WhereClauseInterface $clause
   *
   * @return \Civi\DataProcessor\DataFlow\SqlDataFlow
   */
  public function addWhereClause(WhereClauseInterface $clause): SqlDataFlow {
    foreach($this->whereClauses as $c) {
      if (spl_object_id($clause) == spl_object_id($c) || $clause->getWhereClause() == $c->getWhereClause()) {
        return $this; // Where clause is already added do not add it again.
      }
    }
    $this->whereClauses[] = $clause;
    $this->resetInitializeState();
    return $this;
  }

  /**
   * @param \Civi\DataProcessor\DataFlow\SqlDataFlow\WhereClauseInterface $clause
   *
   * @return \Civi\DataProcessor\DataFlow\SqlDataFlow
   */
  public function removeWhereClause(WhereClauseInterface $clause): SqlDataFlow {
    foreach($this->whereClauses as  $i => $c) {
      if (spl_object_id($clause) == spl_object_id($c) || $clause->getWhereClause() == $c->getWhereClause()) {
        unset($this->whereClauses[$i]);
      }
    }
    $this->resetInitializeState();
    return $this;
  }

  /**
   * Return all the where clauses
   *
   * @param bool $includeJoinClause
   * @param bool $includeNonJoinClause
   *
   * @return array
   */
  public function getWhereClauses(bool $includeJoinClause=TRUE, bool $includeNonJoinClause=TRUE): array {
    $clauses = [];
    $joinClauses = [];
    foreach($this->whereClauses as $clause) {
      if ($clause->isJoinClause() && $includeJoinClause) {
        $joinClauses[] = $clause;
      }
      if (!$clause->isJoinClause() && $includeNonJoinClause) {
        $clauses[] = $clause;
      }
    }
    return array_merge($joinClauses, $clauses);
  }

  /**
   * Get the order by statement
   *
   * @return string
   */
  public function getOrderByStatement(): string {
    $orderBys = array();
    foreach($this->sortSpecifications as $sortSpecification) {
      $dir = 'ASC';
      if ($sortSpecification->getDirection() == SortSpecification::DESC) {
        $dir = 'DESC';
      }
      if ($sortSpecification->getField()) {
        $orderBy = $sortSpecification->getField()->getSqlOrderBy();
        $orderBys[] = "$orderBy $dir";
      }
    }
    if (count($orderBys)) {
      return "ORDER BY ".implode(", ", $orderBys);
    }
    return "";
  }

  /**
   * Returns debug information
   *
   * @return array
   */
  public function getDebugInformation(): array {
    return array(
      'query_time' => number_format($this->queryTime, 3, '.', ',') . ' seconds',
      'count_query_time' => number_format($this->countQueryTime, 3, '.', ',') . ' seconds',
      'query' => $this->sqlStatements,
      'count_query' => $this->sqlCountStatements,
    );
  }

  /**
   * @param $records
   * @param string $fieldNameprefix
   *
   * @return array();
   */
  protected function aggregate($records, $fieldNameprefix=""): array {
    // Aggregation is done in the database.
    return $records;
  }

}
