contactcats/Civi/Api4/Action/ContactCategory/GetFlows.php

<?php
namespace Civi\Api4\Action\ContactCategory;

use CRM_Contactcats_ExtensionUtil as E;
use Civi;
use Civi\Api4\Generic\Result;
use CRM_Core_DAO;
use CRM_Core_Exception;
use DateTimeImmutable;

/**
 * Get counts of contacts moving between categories between two given dates.
 *
 * @see \Civi\Api4\Generic\AbstractAction
 *
 */
class GetFlows extends \Civi\Api4\Generic\AbstractAction {

  /**
   * Start Date
   *
   * @var string
   * @required
   * @default NULL
   */
  protected $startDate = '';

  /**
   * End Date (optional)
   *
   * @var string|null
   * @default NULL
   */
  protected ?string $endDate = NULL;

  private string $catChangesTableName;
  private int $activityTypeId;

  public function _run(Result $result) {
    Civi::log()->debug('Begin', ['=start' => 'GetFlows', '=timed' => 1, 'from' => $this->startDate, 'to' => $this->endDate]);

    $this->catChangesTableName = \Civi\Api4\CustomGroup::get(FALSE)
      ->addWhere('name', '=', 'Category_changes')
      ->execute()->single()['table_name'];
    $this->activityTypeId = (int) \Civi\Api4\OptionValue::get(FALSE)
      ->addWhere('name', '=', 'changed_contact_category')
      ->addWhere('option_group_id:name', '=', 'activity_type')
      ->execute()->single()['value'];
    // Check dates are valid and get a Ymd format of the day *following* the specified one.
    // This is so we can find activities *before* this date and thereby capture activities that
    // occurred *on* the given date.
    $getValidDate = fn(?string $input) => $input ? (new DateTimeImmutable($input))->modify('+1 day')->format('Ymd') : FALSE;

    $endDate = NULL;
    $startDate = $getValidDate($this->startDate);
    if (!$startDate) {
      throw new CRM_Core_Exception(E::ts("Cannot parse given start date."));
    }
    if (!empty($this->endDate)) {
      $endDate = $getValidDate($this->endDate);
      if (!$endDate) {
        throw new CRM_Core_Exception(E::ts("Cannot parse given end date."));
      }
      if (!($endDate > $startDate)) {
        throw new CRM_Core_Exception(E::ts("This version of CiviCRM does not support twisting the space time continuum. End date cannot be before start date."));
      }
    }
    elseif ($startDate > date('Ymd', strtotime('tomorrow'))) {
      throw new CRM_Core_Exception(E::ts("This version of CiviCRM does not support predicting the future. Start time cannot be in the future."));
    }

    // $result['windowFunctionsSupported'] = $this->windowFunctionsSupported();
    // if ($result['windowFunctionsSupported']) {
    if ($this->windowFunctionsSupported()) {
      // $this->debug1($startDate, $endDate);
      $this->solveWithWindowFunctions($result, $startDate, $endDate);
    }
    else {
      // TODO: make a slow version that doesn't need window functions.
      // $this->solveWithoutWindowFunctions($result);
      throw new CRM_Core_Exception("your database does not support Window functions.");
    }

    Civi::log()->debug("Complete.", ['=' => 'set', 'result' => $result->getArrayCopy()]);
  }

  /**
   * Does the database support window functions in the way we need?
   *
   * MariaDB 10.2+ does.
   */
  protected function windowFunctionsSupported() {
    $cache = \CRM_Utils_Cache::create(['type' => ['SqlGroup'], 'name' => 'contactcats']);
    // Will return default if cached value expired.
    $supported = $cache->get('windowFunctionsSupported');
    $supported = NULL;
    if ($supported === NULL) {
      // Need to test for window functions.
      $supported = FALSE;
      // Prevent Civi handling possible execption
      $handler = set_exception_handler(fn() => FALSE);
      try {
        \CRM_Core_DAO::executeQuery("CREATE TEMPORARY TABLE contactcats_window_check (i int unsigned not null, x int unsigned not null)");
        \CRM_Core_DAO::executeQuery("INSERT INTO contactcats_window_check VALUES (1, 1), (2, 1), (3, 2);");
        $rows = \CRM_Core_DAO::executeQuery("SELECT x, ROW_NUMBER() OVER ( PARTITION BY (x) ORDER BY i DESC) rn FROM contactcats_window_check ORDER BY x, rn")->fetchAll();
        $supported = ($rows === [['x' => "1", 'rn' => "1"], ['x' => "1", 'rn' => "2"], ['x' => "2", 'rn' => "1"]]);
      }
      catch (\Exception $e) {
        // Silent fail.
      }
      // Restore exception handler
      set_exception_handler($handler);
      $cache->set('windowFunctionsSupported', $supported);
    }
    return $supported;
  }

  protected function debug1(string $startDateYmd, ?string $endDateYmd) {

    if (!$endDateYmd) {
      $endDateYmd = date('Ymd', strtotime('tomorrow'));
    }
    $params = [1 => [$startDateYmd, 'Int'], 2 => [$endDateYmd, 'Int']];
    $params1 = [1 => [$startDateYmd, 'Int']];

    $this->dump("all activities", <<<SQL
      SELECT ac.contact_id, a.activity_date_time, a.id activity_id
      FROM civicrm_activity a
      INNER JOIN civicrm_activity_contact ac 
        ON a.id = ac.activity_id AND ac.record_type_id = 3
      WHERE a.activity_type_id = $this->activityTypeId
    SQL);

    $this->dump("all start activities", <<<SQL
    WITH activities AS (
      SELECT ac.contact_id, a.activity_date_time, a.id activity_id
      FROM civicrm_activity a
      INNER JOIN civicrm_activity_contact ac 
        ON a.id = ac.activity_id AND ac.record_type_id = 3
      WHERE a.activity_type_id = $this->activityTypeId
    ),

    /* startActivity is the latest activity before the window's start date */
    startActivity AS (
        SELECT contact_id, activity_id,
          ROW_NUMBER() OVER (
            PARTITION BY (contact_id)
            ORDER BY activity_date_time DESC
          ) rn
        FROM activities a1
        WHERE a1.activity_date_time < %1
        )

      select * from startActivity;
    SQL, $params1);

    $this->dump("all end activities", <<<SQL
    WITH activities AS (
      SELECT ac.contact_id, a.activity_date_time, a.id activity_id
      FROM civicrm_activity a
      INNER JOIN civicrm_activity_contact ac 
        ON a.id = ac.activity_id AND ac.record_type_id = 3
      WHERE a.activity_type_id = $this->activityTypeId
    ),

    /* endActivity is the latest activity before the window's end date */
    endActivity AS (
        SELECT contact_id, activity_id,
          ROW_NUMBER() OVER (
            PARTITION BY (contact_id)
            ORDER BY activity_date_time DESC
          ) rn
        FROM activities a2
        WHERE a2.activity_date_time < %2
        )

      select * from endActivity;
    SQL, $params);

    $this->dump("all 2", <<<SQL
    /* Identify the relevant activities for the contacts */
    WITH activities AS (
      SELECT ac.contact_id, a.activity_date_time, a.id activity_id
      FROM civicrm_activity a
      INNER JOIN civicrm_activity_contact ac 
        ON a.id = ac.activity_id AND ac.record_type_id = 3
      WHERE a.activity_type_id = $this->activityTypeId
    ),

    /* startActivity is the latest activity before the window's start date */
    startActivity AS (
        SELECT contact_id, activity_id,
          ROW_NUMBER() OVER (
            PARTITION BY (contact_id)
            ORDER BY activity_date_time DESC
          ) rn
        FROM activities a1
        WHERE a1.activity_date_time < %1
        ),

    /* endActivity is the latest activity before the window's end date */
    endActivity AS (
        SELECT contact_id, activity_id,
          ROW_NUMBER() OVER (
            PARTITION BY (contact_id)
            ORDER BY activity_date_time DESC
          ) rn
        FROM activities a2
        WHERE a2.activity_date_time < %2
        )

    SELECT /*startCat.new_category_id from_category_id, endCat.new_category_id to_category_id,*/
      endActivity.contact_id endCtID, endActivity.activity_id endAcID,
      startActivity.contact_id startCtID, startActivity.activity_id startAcID
    FROM endActivity 
    /*    INNER JOIN $this->catChangesTableName endCat
      ON endActivity.activity_id = endCat.entity_id */
    LEFT JOIN (
      startActivity
      /* INNER JOIN $this->catChangesTableName startCat
        ON startActivity.activity_id = startCat.entity_id */
        )
      ON startActivity.contact_id = endActivity.contact_id AND startActivity.rn = 1
    WHERE endActivity.rn = 1
    /*ORDER BY from_category_id, to_category_id*/
    ;
    SQL, $params);

  }

  /**
   * This SQL uses the change activities to determine the flows between two dates.
   *
   * It does so by finding the latest activity before the start date and using it's new_category_id
   * as the category for that contact on that date, and likewise for the end date.
   *
   * It is assumed that a change activity is always present.
   */
  protected function solveWithWindowFunctions(Result $result, string $startDateYmd, ?string $endDateYmd) {

    $sql = <<<SQL
    /* Identify the relevant activities for the contacts */
    WITH activities AS (
      SELECT ac.contact_id, a.activity_date_time, a.id activity_id
      FROM civicrm_activity a
      INNER JOIN civicrm_activity_contact ac 
        ON a.id = ac.activity_id AND ac.record_type_id = 3
      WHERE a.activity_type_id = $this->activityTypeId
    ),

    /* startActivity is the latest activity before the window's start date */
    startActivity AS (
        SELECT contact_id, activity_id,
          ROW_NUMBER() OVER (
            PARTITION BY (contact_id)
            ORDER BY activity_date_time DESC
          ) rn
        FROM activities a1
        WHERE a1.activity_date_time < %1
        ),

    /* endActivity is the latest activity before the window's end date */
    endActivity AS (
        SELECT contact_id, activity_id,
          ROW_NUMBER() OVER (
            PARTITION BY (contact_id)
            ORDER BY activity_date_time DESC
          ) rn
        FROM activities a2
        WHERE a2.activity_date_time < %2
        )

    /* join startActivity and endActivity to count changes between each shift */
    SELECT startCat.new_category_id from_category_id, endCat.new_category_id to_category_id, count(*) contact_count
    FROM endActivity 
    INNER JOIN $this->catChangesTableName endCat
      ON endActivity.activity_id = endCat.entity_id
    LEFT JOIN (
      startActivity
      INNER JOIN $this->catChangesTableName startCat
        ON startActivity.activity_id = startCat.entity_id
        )
      ON startActivity.contact_id = endActivity.contact_id AND startActivity.rn = 1
    WHERE endActivity.rn = 1
    GROUP BY from_category_id, to_category_id
    ;
    SQL;

    // print "\n$sql\n %1 $startDateYmd, %2 $endDateYmd\n";

    if (!$endDateYmd) {
      $endDateYmd = date('Ymd', strtotime('tomorrow'));
    }
    $data = CRM_Core_DAO::executeQuery($sql, [
      1 => [$startDateYmd, 'Int'],
      2 => [$endDateYmd, 'Int'],
    ])->fetchAll();

    // Don't use exchange array so as not to gazump non-array data(?)
    foreach ($data as $row) {
      $result[] = [
        'from_category_id' => (int) $row['from_category_id'],
        'to_category_id' => (int) $row['to_category_id'],
        'contact_count' => (int) $row['contact_count'],
      ];
    }
  }

  /**
   * useful in debugging phpunit tests only; unused in normal operation.
   */
  protected function dump(string $msg, string $sql, $params = []) {
    try {
      $data = CRM_Core_DAO::executeQuery($sql, $params)->fetchAll();
    }
    catch (\Exception $e) {
      print $e->getCause()->userinfo;
    }
    print "\n$msg (" . count($data) . ") records ===============================\n";
    foreach ($data as $row) {
      print json_encode($row) . "\n";
    }
    print "===============================\n";
  }

}