<?php

namespace Blesta\App\Models;

use Blesta\App\AppModel;
use DateTime;
use DateTimeZone;
use Language;
use Loader;
use PDO;
use Throwable;

/**
 * Revenue Intelligence snapshot pipeline.
 *
 * Builds the analytics_* fact/rollup tables from a set-based read of the
 * services, pricings, transactions and invoices tables. Everything here is
 * computed with a handful of SQL statements per company (no per-service
 * pricing-presenter calls) and published atomically per snapshot date so
 * readers never observe a half-built snapshot.
 *
 * All "by date" bucketing is done in the company's local timezone. Datetimes
 * stored in transactions/invoices/services are UTC and are converted to the
 * company-local calendar day before bucketing.
 *
 * @package blesta
 * @subpackage app.models
 * @copyright Copyright (c) 2026, Phillips Data, Inc.
 * @license http://www.blesta.com/license/ The Blesta License Agreement
 * @link http://www.blesta.com/ Blesta
 */
class AnalyticsSnapshots extends AppModel
{
    /**
     * @var int The number of days of revenue facts to (re)build behind the snapshot date
     */
    private const REVENUE_LOOKBACK_CAP = 90;

    /**
     * @var int The number of services to insert into analytics_service_mrr per batch
     */
    private const SERVICE_BATCH_SIZE = 1000;

    /**
     * @var int The number of forward months to precompute the renewal forecast for
     */
    private const FORECAST_MONTHS = 12;

    /**
     * @var int Number of log_transactions rows consumed per batch when scanning for
     *  edited transactions (cursor-paginated on the indexed primary key)
     */
    private const TRANSACTION_LOG_BATCH_SIZE = 1000;

    /**
     * @var int Maximum log batches consumed per run; unconsumed rows resume from the
     *  persisted cursor on the next run
     */
    private const TRANSACTION_LOG_MAX_BATCHES = 10;

    /**
     * @var int Maximum distinct company-local dates rebuilt per run due to transaction
     *  edits (each date rebuild scans the transactions table); excess dates resume on
     *  the next run via the cursor
     */
    private const EDITED_DATE_REBUILD_CAP = 30;

    /**
     * @var string The company setting holding the last-consumed log_transactions ID
     */
    private const TRANSACTION_LOG_CURSOR_SETTING = 'analytics_transaction_log_cursor';

    /**
     * Initialize AnalyticsSnapshots
     */
    public function __construct()
    {
        parent::__construct();

        Loader::loadModels($this, ['Companies']);
        Language::loadLang(['analytics_snapshots']);
    }

    /**
     * Runs a full snapshot for the given company for the given company-local date.
     *
     * The workflow is:
     *  1. Acquire a per-company advisory lock (returns false immediately if held).
     *  2. Mark an analytics_snapshot_runs row 'running' for the date.
     *  3. Build analytics_service_mrr rows set-based, chunked by service id.
     *  4. In a single transaction, replace the date's rollup/movement/revenue/fx/
     *     forecast rows and mark the run 'completed'.
     *  5. Prune per-retention old per-service rows (monthly last row kept forever).
     *
     * On any failure the run is marked 'failed', previously completed snapshots are
     * left untouched, and false is returned.
     *
     * @param int $company_id The ID of the company to snapshot
     * @param string $date The company-local snapshot date (Y-m-d); null for "today" in the company timezone
     * @return bool True on success, false if busy or on failure
     */
    public function snapshot($company_id, $date = null)
    {
        $company_id = (int) $company_id;
        $timezone = $this->getCompanyTimezone($company_id);

        // Resolve the company-local snapshot date
        $snapshot_date = ($date !== null)
            ? substr($date, 0, 10)
            : $this->localNow($timezone, 'Y-m-d');

        $lock_name = 'analytics_snapshot_' . $company_id;

        // Acquire the advisory lock; bail quietly if another run holds it
        if (!$this->acquireLock($lock_name)) {
            return false;
        }

        try {
            return $this->runSnapshot($company_id, $snapshot_date, $timezone);
        } finally {
            $this->releaseLock($lock_name);
        }
    }

    /**
     * Executes the snapshot pipeline for a company/date. Assumes the advisory lock is
     * already held by the caller, which guarantees its release.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string $timezone The company timezone
     * @return bool True on success, false on failure (errors set via Input)
     */
    private function runSnapshot($company_id, $snapshot_date, $timezone)
    {
        // Capture the run's prior state so a failed re-run of an already-completed date
        // can be restored (the rollback below preserves that date's published data)
        $prior_run = $this->getRun($company_id, $snapshot_date);

        try {
            // Determine the previous completed snapshot to diff movements against and
            // the revenue window to (re)build
            $previous_date = $this->getPreviousCompletedDate($company_id, $snapshot_date);
            $revenue_dates = $this->getRevenueDates($previous_date, $snapshot_date, $timezone);

            // Consume transaction edits logged since the last run (refunds/returns/voids
            // mutate the original row) from log_transactions via a persisted cursor so no
            // edit is ever skipped. This yields the (possibly much older) original dates
            // to rebuild plus the refund events to post on their refund dates
            $edited = $this->consumeTransactionEdits($company_id, $snapshot_date, $timezone);
            foreach ($edited['dates'] as $edited_date) {
                if ($edited_date <= $snapshot_date && !in_array($edited_date, $revenue_dates, true)) {
                    $revenue_dates[] = $edited_date;
                }
            }
            sort($revenue_dates);

            // Build and publish atomically: everything from the run-state transition to
            // the 'completed' mark commits or rolls back together, so neither a failure
            // nor a hard interruption (killed process, dropped connection) can damage a
            // previously completed snapshot or leave a stray 'running' row behind
            $this->Record->begin();

            $this->upsertRun($company_id, $snapshot_date);
            $this->buildServiceMrr($company_id, $snapshot_date, $timezone);
            $this->buildMrrDaily($company_id, $snapshot_date);
            $this->buildMovements($company_id, $snapshot_date, $previous_date, $timezone);
            foreach ($revenue_dates as $revenue_date) {
                $this->buildRevenueForDate($company_id, $revenue_date, $timezone);
            }
            // Post refund events on their refund dates after the rebuilds (rebuilds carry
            // previously posted refunds forward; the cursor guarantees exactly-once posting)
            $this->applyRefundEvents($company_id, $edited['events']);
            $this->buildFxRates($company_id, $snapshot_date);
            $this->buildRenewalForecast($company_id, $snapshot_date, $timezone);

            $this->markRunCompleted($company_id, $snapshot_date);

            $this->Record->commit();

            // Advance the edit-log cursor only after a successful publish so a failed
            // run re-examines the same log rows next time
            if ($edited['cursor'] !== null) {
                $this->Companies->setSetting(
                    $company_id,
                    self::TRANSACTION_LOG_CURSOR_SETTING,
                    (string) $edited['cursor'],
                    false
                );
            }
        } catch (Throwable $e) {
            if ($this->Record->getConnection()->inTransaction()) {
                $this->Record->rollBack();
            }

            if ($prior_run && $prior_run->status == 'completed') {
                // The rollback preserved the prior completed data — restore the run
                // state so readers and future movement diffs stay consistent with it
                $this->markRunRestored($company_id, $snapshot_date, $prior_run->completed_at, $e->getMessage());
            } else {
                $this->markRunFailed($company_id, $snapshot_date, $e->getMessage());
            }

            $this->Input->setErrors([
                'snapshot' => ['failed' => $this->_('AnalyticsSnapshots.!error.snapshot.failed', $e->getMessage())]
            ]);

            return false;
        }

        // Prune old per-service rows per retention policy (outside the transaction)
        try {
            $this->pruneServiceMrr($company_id, $snapshot_date, $timezone);
        } catch (Throwable $e) {
            // Pruning is best-effort; a failure here must not fail the snapshot
        }

        return true;
    }

    /**
     * Acquires a MySQL named advisory lock without waiting.
     *
     * @param string $lock_name The lock name
     * @return bool True if acquired, false if held by another connection
     */
    private function acquireLock($lock_name)
    {
        $result = $this->Record->query('SELECT GET_LOCK(?, 0) AS acquired', [$lock_name])
            ->fetch(PDO::FETCH_OBJ);

        return $result && (int) $result->acquired === 1;
    }

    /**
     * Releases a MySQL named advisory lock.
     *
     * @param string $lock_name The lock name
     */
    private function releaseLock($lock_name)
    {
        try {
            $this->Record->query('SELECT RELEASE_LOCK(?)', [$lock_name]);
        } catch (Throwable $e) {
            // Ignore release errors
        }
    }

    /**
     * Inserts or resets the run-state row for the given date to 'running'.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     */
    private function upsertRun($company_id, $snapshot_date)
    {
        $this->Record->query(
            'INSERT INTO analytics_snapshot_runs (company_id, snapshot_date, status, started_at)'
            . " VALUES (?, ?, 'running', ?)"
            . " ON DUPLICATE KEY UPDATE status = 'running', started_at = VALUES(started_at),"
            . ' completed_at = NULL, error = NULL',
            [$company_id, $snapshot_date, gmdate('Y-m-d H:i:s')]
        );
    }

    /**
     * Marks the run-state row 'completed' with the current UTC time.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     */
    private function markRunCompleted($company_id, $snapshot_date)
    {
        $this->Record->query(
            "UPDATE analytics_snapshot_runs SET status = 'completed', completed_at = ?, error = NULL"
            . ' WHERE company_id = ? AND snapshot_date = ?',
            [gmdate('Y-m-d H:i:s'), $company_id, $snapshot_date]
        );
    }

    /**
     * Fetches the run-state row for a company/date, if any
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @return stdClass|null The run row, or null if none exists
     */
    private function getRun($company_id, $snapshot_date)
    {
        $row = $this->Record->query(
            'SELECT * FROM analytics_snapshot_runs WHERE company_id = ? AND snapshot_date = ?',
            [$company_id, $snapshot_date]
        )->fetch(PDO::FETCH_OBJ);

        return $row ?: null;
    }

    /**
     * Restores a run to 'completed' after a failed re-run whose rollback preserved the
     * previously published data. The failure message is retained for diagnostics.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string $completed_at The original UTC completion datetime to restore
     * @param string $error The failure message from the aborted re-run
     */
    private function markRunRestored($company_id, $snapshot_date, $completed_at, $error)
    {
        try {
            $this->Record->query(
                "UPDATE analytics_snapshot_runs SET status = 'completed', completed_at = ?, error = ?"
                . ' WHERE company_id = ? AND snapshot_date = ?',
                [$completed_at, $error, $company_id, $snapshot_date]
            );
        } catch (Throwable $e) {
            // Nothing further we can do if the restore itself fails
        }
    }

    /**
     * Marks the run-state row 'failed' with the given error message.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string $error The error message
     */
    private function markRunFailed($company_id, $snapshot_date, $error)
    {
        try {
            // Upsert: the 'running' transition is now part of the rolled-back
            // transaction, so a first-time run's row may not exist at this point
            $this->Record->query(
                'INSERT INTO analytics_snapshot_runs (company_id, snapshot_date, status, started_at, error)'
                . " VALUES (?, ?, 'failed', ?, ?)"
                . " ON DUPLICATE KEY UPDATE status = 'failed', completed_at = NULL, error = VALUES(error)",
                [$company_id, $snapshot_date, gmdate('Y-m-d H:i:s'), $error]
            );
        } catch (Throwable $e) {
            // Nothing further we can do if even the failure state cannot be recorded
        }
    }

    /**
     * Returns the most recent completed snapshot date strictly before the given date.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @return string|null The previous completed date (Y-m-d), or null if none
     */
    private function getPreviousCompletedDate($company_id, $snapshot_date)
    {
        $row = $this->Record->query(
            'SELECT MAX(snapshot_date) AS snapshot_date FROM analytics_snapshot_runs'
            . " WHERE company_id = ? AND status = 'completed' AND snapshot_date < ?",
            [$company_id, $snapshot_date]
        )->fetch(PDO::FETCH_OBJ);

        return ($row && $row->snapshot_date) ? $row->snapshot_date : null;
    }

    /**
     * Builds the inclusive list of company-local dates for which cash/invoice facts
     * should be (re)built: from the last completed snapshot (capped at REVENUE_LOOKBACK_CAP
     * days back) through the snapshot date. Dates dirtied by transaction edits are added
     * separately by the caller via consumeTransactionEdits().
     *
     * @param string|null $previous_date The previous completed snapshot date (Y-m-d) or null
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string $timezone The company timezone
     * @return array A list of Y-m-d date strings (ascending)
     */
    private function getRevenueDates($previous_date, $snapshot_date, $timezone)
    {
        $cap_date = $this->localModify($snapshot_date, '-' . self::REVENUE_LOOKBACK_CAP . ' days', $timezone);

        $start = $previous_date ?? $snapshot_date;
        if ($start < $cap_date) {
            $start = $cap_date;
        }

        $dates = [];
        $current = $start;
        while ($current <= $snapshot_date) {
            $dates[] = $current;
            $current = $this->localModify($current, '+1 day', $timezone);

            // Hard cap the number of dates as a safety net
            if (count($dates) > self::REVENUE_LOOKBACK_CAP + 1) {
                break;
            }
        }

        return $dates;
    }

    /**
     * Consumes transaction edits (refunds, returns, voids and amount changes are logged
     * to log_transactions) exhaustively via a per-company cursor over the log's indexed
     * primary key. Yields both the distinct company-local ORIGINAL transaction dates to
     * rebuild and the refund EVENTS (Phase 2, §2.2) parsed from the consumed rows'
     * prev/cur field diffs, dated on the edit itself so refunds bucket on the refund
     * date rather than the transaction date.
     *
     * The cursor value is returned rather than persisted — the caller persists it only
     * after the rebuilt dates and posted events have been committed, so a failed run
     * re-examines the same rows and events are posted exactly once. Work per run is
     * bounded by batch and date caps; any remainder resumes from the cursor next run.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string $timezone The company timezone
     * @return array An array containing:
     *  - dates A list of company-local Y-m-d date strings to rebuild
     *  - events A list of refund events: ['date', 'currency', 'gateway_id', 'amount']
     *      (amount may be negative for refund reversals)
     *  - cursor The new cursor value to persist after a successful publish (null if unchanged)
     */
    private function consumeTransactionEdits($company_id, $snapshot_date, $timezone)
    {
        $start_cursor = $this->getCompanySettingValue($company_id, self::TRANSACTION_LOG_CURSOR_SETTING, null);

        // First run: baseline the cursor at the current end of the log; historical
        // edits predate observation and are not replayed (no-backfill policy)
        if ($start_cursor === null || $start_cursor === '') {
            $max = $this->Record->query('SELECT MAX(id) AS id FROM log_transactions')->fetch(PDO::FETCH_OBJ);

            return ['dates' => [], 'events' => [], 'cursor' => ($max && $max->id ? (int) $max->id : 0)];
        }

        $start_cursor = (int) $start_cursor;
        $cursor = $start_cursor;
        $dates = [];
        $transaction_ids = [];
        $advanced = false;
        $capped = false;

        // The end of the log as of now; the end-of-scan jump below must not pass this,
        // or rows appended mid-scan could be skipped
        $max = $this->Record->query('SELECT MAX(id) AS id FROM log_transactions')->fetch(PDO::FETCH_OBJ);
        $log_end = ($max && $max->id ? (int) $max->id : $cursor);

        for ($batch = 0; $batch < self::TRANSACTION_LOG_MAX_BATCHES && !$capped; $batch++) {
            $rows = $this->Record->query(
                'SELECT lt.id, lt.transaction_id, t.date_added FROM log_transactions lt'
                . ' INNER JOIN transactions t ON t.id = lt.transaction_id'
                . ' INNER JOIN clients c ON c.id = t.client_id'
                . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
                . ' WHERE cg.company_id = ? AND lt.id > ?'
                . ' ORDER BY lt.id ASC'
                . ' LIMIT ' . self::TRANSACTION_LOG_BATCH_SIZE,
                [$company_id, $cursor]
            )->fetchAll(PDO::FETCH_OBJ);

            foreach ($rows as $row) {
                $local_date = $this->utcToLocalDate($row->date_added, $timezone);

                // Stop before exceeding the per-run rebuild cap; the cursor stays on
                // the last consumed row so the remainder resumes next run
                if (!isset($dates[$local_date]) && count($dates) >= self::EDITED_DATE_REBUILD_CAP) {
                    $capped = true;
                    break;
                }

                $dates[$local_date] = true;
                $transaction_ids[(int) $row->transaction_id] = true;
                $cursor = (int) $row->id;
                $advanced = true;
            }

            if (!$capped && count($rows) < self::TRANSACTION_LOG_BATCH_SIZE) {
                // The scan reached the end of the log: every remaining row up to the
                // pre-scan end was either consumed or belongs to another company
                if ($log_end > $cursor) {
                    $cursor = $log_end;
                    $advanced = true;
                }
                break;
            }
        }

        // Parse refund events from the consumed window (start_cursor, cursor]
        $events = $this->buildRefundEvents(array_keys($transaction_ids), $start_cursor, $cursor, $timezone);

        return ['dates' => array_keys($dates), 'events' => $events, 'cursor' => ($advanced ? $cursor : null)];
    }

    /**
     * Derives refund events for a set of edited transactions by replaying each
     * transaction's full log history in order (the transaction_id index makes this
     * cheap) and emitting deltas only for log rows within the consumed cursor window,
     * so each event is produced exactly once across runs.
     *
     * Refund semantics per transaction, walking rows in id order while tracking the
     * amount and status in effect:
     *  - status transitions INTO refunded/returned emit +amount-after-the-edit (a
     *    partial refund replaces the amount with the refunded portion in the same
     *    edit; a full refund leaves the amount unchanged)
     *  - amount changes WHILE refunded/returned emit the delta (a further partial refund)
     *  - status transitions OUT of refunded/returned emit the negated amount (reversal)
     *
     * @param array $transaction_ids The IDs of transactions with consumed log rows
     * @param int $start_cursor Only log rows with id greater than this emit events
     * @param int $end_cursor Only log rows with id at or below this emit events
     * @param string $timezone The company timezone
     * @return array A list of ['date' => Y-m-d, 'currency', 'gateway_id', 'amount']
     */
    private function buildRefundEvents(array $transaction_ids, $start_cursor, $end_cursor, $timezone)
    {
        if (empty($transaction_ids)) {
            return [];
        }

        $refund_statuses = ['refunded', 'returned'];
        $events = [];

        foreach (array_chunk($transaction_ids, 500) as $chunk) {
            $placeholders = implode(', ', array_fill(0, count($chunk), '?'));

            $transactions = $this->Record->query(
                'SELECT id, amount, currency, gateway_id, status FROM transactions'
                . ' WHERE id IN (' . $placeholders . ')',
                $chunk
            )->fetchAll(PDO::FETCH_OBJ);
            $meta = [];
            foreach ($transactions as $transaction) {
                $meta[(int) $transaction->id] = $transaction;
            }

            $log_rows = $this->Record->query(
                'SELECT id, transaction_id, `change`, date_changed FROM log_transactions'
                . ' WHERE transaction_id IN (' . $placeholders . ') AND id <= ?'
                . ' ORDER BY transaction_id ASC, id ASC',
                array_merge($chunk, [$end_cursor])
            )->fetchAll(PDO::FETCH_OBJ);

            $histories = [];
            foreach ($log_rows as $log_row) {
                $histories[(int) $log_row->transaction_id][] = $log_row;
            }

            foreach ($histories as $transaction_id => $history) {
                if (!isset($meta[$transaction_id])) {
                    continue;
                }
                $transaction = $meta[$transaction_id];

                // Reconstruct the initial state from the earliest logged prev values,
                // falling back to the current row when a field was never edited
                $amount = (float) $transaction->amount;
                $status = $transaction->status;
                $parsed = [];
                foreach ($history as $log_row) {
                    $parsed[$log_row->id] = $this->parseTransactionLogChange($log_row->change);
                }
                foreach ($history as $log_row) {
                    $fields = $parsed[$log_row->id];
                    if (isset($fields['amount']['prev'])) {
                        $amount = (float) $fields['amount']['prev'];
                        break;
                    }
                }
                foreach ($history as $log_row) {
                    $fields = $parsed[$log_row->id];
                    if (isset($fields['status']['prev'])) {
                        $status = $fields['status']['prev'];
                        break;
                    }
                }

                foreach ($history as $log_row) {
                    $fields = $parsed[$log_row->id];
                    if ($fields === null) {
                        continue;
                    }

                    $new_amount = isset($fields['amount']['cur']) ? (float) $fields['amount']['cur'] : $amount;
                    $new_status = isset($fields['status']['cur']) ? $fields['status']['cur'] : $status;

                    $was_refunded = in_array($status, $refund_statuses, true);
                    $is_refunded = in_array($new_status, $refund_statuses, true);

                    $delta = 0.0;
                    if (!$was_refunded && $is_refunded) {
                        $delta = $new_amount;
                    } elseif ($was_refunded && $is_refunded && $new_amount != $amount) {
                        $delta = $new_amount - $amount;
                    } elseif ($was_refunded && !$is_refunded) {
                        $delta = -$amount;
                    }

                    if ($delta != 0.0 && $log_row->id > $start_cursor && $log_row->id <= $end_cursor) {
                        $events[] = [
                            'date' => $this->utcToLocalDate($log_row->date_changed, $timezone),
                            'currency' => $transaction->currency,
                            'gateway_id' => (int) ($transaction->gateway_id ?? 0),
                            'amount' => $delta
                        ];
                    }

                    $amount = $new_amount;
                    $status = $new_status;
                }
            }
        }

        return $events;
    }

    /**
     * Posts refund events additively into analytics_revenue_daily on their refund dates
     * (dimension 'total' and 'gateway'). Uses upserts so a date that has never been
     * built still records its refunds; date rebuilds carry these values forward.
     *
     * @param int $company_id The company ID
     * @param array $events Refund events from consumeTransactionEdits()
     */
    private function applyRefundEvents($company_id, array $events)
    {
        if (empty($events)) {
            return;
        }

        // Aggregate per (date, currency) for totals and per (date, currency, gateway)
        $totals = [];
        $gateways = [];
        foreach ($events as $event) {
            $date = $event['date'];
            $currency = $event['currency'];
            $gateway_id = (int) $event['gateway_id'];

            $totals[$date][$currency] = ($totals[$date][$currency] ?? 0.0) + $event['amount'];
            $gateways[$date][$currency][$gateway_id]
                = ($gateways[$date][$currency][$gateway_id] ?? 0.0) + $event['amount'];
        }

        $upsert = 'INSERT INTO analytics_revenue_daily'
            . ' (company_id, date, currency, dimension, dimension_id, collected, refunded,'
            . ' invoiced, invoiced_tax, credited)'
            . ' VALUES (?, ?, ?, ?, ?, 0, ?, 0, 0, 0)'
            . ' ON DUPLICATE KEY UPDATE refunded = ROUND(refunded + VALUES(refunded), 4)';

        foreach ($totals as $date => $currencies) {
            foreach ($currencies as $currency => $amount) {
                if (round($amount, 4) == 0.0) {
                    continue;
                }
                $this->Record->query(
                    $upsert,
                    [$company_id, $date, $currency, 'total', 0, round($amount, 4)]
                );
            }
        }
        foreach ($gateways as $date => $currencies) {
            foreach ($currencies as $currency => $gateway_amounts) {
                foreach ($gateway_amounts as $gateway_id => $amount) {
                    if (round($amount, 4) == 0.0) {
                        continue;
                    }
                    $this->Record->query(
                        $upsert,
                        [$company_id, $date, $currency, 'gateway', $gateway_id, round($amount, 4)]
                    );
                }
            }
        }
    }

    /**
     * Decodes a log_transactions.change payload (base64-encoded serialized field diff)
     *
     * @param string $change The raw change payload
     * @return array|null A map of field => ['prev' => ..., 'cur' => ...], or null when undecodable
     */
    private function parseTransactionLogChange($change)
    {
        $decoded = base64_decode((string) $change, true);
        if ($decoded === false) {
            return null;
        }

        try {
            $fields = unserialize($decoded, ['allowed_classes' => false]);
        } catch (Throwable $e) {
            return null;
        }

        return is_array($fields) ? $fields : null;
    }

    /**
     * Builds analytics_service_mrr rows for the snapshot date, set-based and chunked by
     * service id. MRR is the monthly-normalized recurring contracted price in the pricing's
     * native currency: the effective recurring service price (override -> price_renews ->
     * price) times quantity, plus each recurring configurable option's effective price times
     * its quantity. One-time pricings, taxes, setup/cancel fees and coupons are excluded.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string $timezone The company timezone
     */
    private function buildServiceMrr($company_id, $snapshot_date, $timezone)
    {
        // Idempotent: clear any partial rows for this date first
        $this->Record->from('analytics_service_mrr')
            ->where('company_id', '=', $company_id)
            ->where('snapshot_date', '=', $snapshot_date)
            ->delete();

        $service_term = $this->termInMonthsSql('p');
        $option_term = $this->termInMonthsSql('op');

        // Per-service recurring option MRR (monthly-normalized), summed per option currency.
        // Scoped to the chunk's service id window so the aggregate never scans other
        // companies' options, and grouped by currency so a mismatched option pricing
        // (e.g. a service overridden to another currency) is excluded rather than
        // summed into the wrong currency
        $option_join = '(SELECT so.service_id AS service_id, op.currency AS currency,'
            . ' SUM(COALESCE(op.price_renews, op.price) * so.qty / NULLIF(' . $option_term . ', 0)) AS option_mrr'
            . ' FROM service_options so'
            . ' INNER JOIN package_option_pricing pop ON pop.id = so.option_pricing_id'
            . ' INNER JOIN pricings op ON op.id = pop.pricing_id'
            . " WHERE op.period != 'onetime'"
            . ' AND so.service_id >= ? AND so.service_id <= ?'
            . ' GROUP BY so.service_id, op.currency) opt';

        $base_mrr = 'COALESCE(s.override_price, p.price_renews, p.price) * s.qty / NULLIF(' . $service_term . ', 0)';

        $insert = 'INSERT INTO analytics_service_mrr'
            . ' (service_id, snapshot_date, company_id, currency, mrr, status, client_id, package_id, module_id)'
            . ' SELECT s.id, ?, ?, COALESCE(s.override_currency, p.currency),'
            . ' ROUND(COALESCE(' . $base_mrr . ', 0) + COALESCE(opt.option_mrr, 0), 4),'
            . ' s.status, s.client_id, pp.package_id, COALESCE(pkg.module_id, 0)'
            . ' FROM services s'
            . ' INNER JOIN clients c ON c.id = s.client_id'
            . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
            . ' INNER JOIN package_pricing pp ON pp.id = s.pricing_id'
            . ' INNER JOIN pricings p ON p.id = pp.pricing_id'
            . ' INNER JOIN packages pkg ON pkg.id = pp.package_id'
            . ' LEFT JOIN ' . $option_join
            . ' ON opt.service_id = s.id AND opt.currency = COALESCE(s.override_currency, p.currency)'
            . ' WHERE cg.company_id = ?'
            . " AND s.status IN ('active', 'suspended')"
            . " AND p.period != 'onetime'"
            . ' AND s.id >= ? AND s.id <= ?';

        // Chunk by service id windows to bound each statement's working set
        $last_id = 0;
        do {
            $window = $this->Record->select(['services.id'])
                ->from('services')
                ->innerJoin('clients', 'clients.id', '=', 'services.client_id', false)
                ->innerJoin('client_groups', 'client_groups.id', '=', 'clients.client_group_id', false)
                ->where('client_groups.company_id', '=', $company_id)
                ->where('services.status', 'in', ['active', 'suspended'])
                ->where('services.id', '>', $last_id)
                ->order(['services.id' => 'ASC'])
                ->limit(self::SERVICE_BATCH_SIZE)
                ->fetchAll();

            if (empty($window)) {
                break;
            }

            $low_id = (int) $window[0]->id;
            $high_id = (int) end($window)->id;

            $this->Record->query(
                $insert,
                [$snapshot_date, $company_id, $low_id, $high_id, $company_id, $low_id, $high_id]
            );

            $last_id = $high_id;
        } while (count($window) === self::SERVICE_BATCH_SIZE);
    }

    /**
     * Builds the company MRR rollup for the snapshot date from analytics_service_mrr.
     * "mrr" is contracted MRR (active + suspended); "suspended_mrr" is the suspended
     * portion only. Active-only MRR is mrr - suspended_mrr. Client counts here are
     * per-currency (distinct clients with positive MRR in that currency).
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     */
    private function buildMrrDaily($company_id, $snapshot_date)
    {
        $this->Record->from('analytics_mrr_daily')
            ->where('company_id', '=', $company_id)
            ->where('snapshot_date', '=', $snapshot_date)
            ->delete();

        $this->Record->query(
            'INSERT INTO analytics_mrr_daily'
            . ' (company_id, snapshot_date, currency, mrr, suspended_mrr, active_services,'
            . ' suspended_services, active_clients)'
            . ' SELECT company_id, snapshot_date, currency,'
            . ' ROUND(SUM(mrr), 4),'
            . " ROUND(SUM(CASE WHEN status = 'suspended' THEN mrr ELSE 0 END), 4),"
            . " SUM(CASE WHEN status = 'active' THEN 1 ELSE 0 END),"
            . " SUM(CASE WHEN status = 'suspended' THEN 1 ELSE 0 END),"
            . ' COUNT(DISTINCT CASE WHEN mrr > 0 THEN client_id END)'
            . ' FROM analytics_service_mrr'
            . ' WHERE company_id = ? AND snapshot_date = ?'
            . ' GROUP BY company_id, snapshot_date, currency',
            [$company_id, $snapshot_date]
        );
    }

    /**
     * Derives MRR movements for the snapshot date by diffing it against the previous
     * completed snapshot. Movements are decomposed into new/expansion/contraction/
     * churn/reactivation; every amount is stored as a positive magnitude. A currency
     * change on a service is recorded as churn in the old currency plus new in the new
     * currency. No movements are produced when there is no prior completed snapshot.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string|null $previous_date The previous completed snapshot date (Y-m-d) or null
     * @param string $timezone The company timezone
     */
    private function buildMovements($company_id, $snapshot_date, $previous_date, $timezone)
    {
        // Regeneration is idempotent per date
        $this->Record->from('analytics_mrr_movements')
            ->where('company_id', '=', $company_id)
            ->where('date', '=', $snapshot_date)
            ->delete();

        if ($previous_date === null) {
            // First completed snapshot: nothing to diff against
            return;
        }

        $prev_start_utc = $this->localToUtc($previous_date, '00:00:00', $timezone);

        // Expansion: same service, same currency, MRR increased
        $this->Record->query(
            'INSERT INTO analytics_mrr_movements'
            . ' (company_id, date, currency, type, amount, service_id, client_id, package_id)'
            . " SELECT t.company_id, t.snapshot_date, t.currency, 'expansion', ROUND(t.mrr - p.mrr, 4),"
            . ' t.service_id, t.client_id, t.package_id'
            . ' FROM analytics_service_mrr t'
            . ' INNER JOIN analytics_service_mrr p'
            . ' ON p.service_id = t.service_id AND p.snapshot_date = ? AND p.currency = t.currency'
            . ' WHERE t.company_id = ? AND t.snapshot_date = ? AND (t.mrr - p.mrr) > 0',
            [$previous_date, $company_id, $snapshot_date]
        );

        // Contraction: same service, same currency, MRR decreased
        $this->Record->query(
            'INSERT INTO analytics_mrr_movements'
            . ' (company_id, date, currency, type, amount, service_id, client_id, package_id)'
            . " SELECT t.company_id, t.snapshot_date, t.currency, 'contraction', ROUND(p.mrr - t.mrr, 4),"
            . ' t.service_id, t.client_id, t.package_id'
            . ' FROM analytics_service_mrr t'
            . ' INNER JOIN analytics_service_mrr p'
            . ' ON p.service_id = t.service_id AND p.snapshot_date = ? AND p.currency = t.currency'
            . ' WHERE t.company_id = ? AND t.snapshot_date = ? AND (t.mrr - p.mrr) < 0',
            [$previous_date, $company_id, $snapshot_date]
        );

        // Churn: present in prior currency but not in the current snapshot in that currency
        // (covers cancellations and the old-currency side of a currency change)
        $this->Record->query(
            'INSERT INTO analytics_mrr_movements'
            . ' (company_id, date, currency, type, amount, service_id, client_id, package_id)'
            . " SELECT p.company_id, ?, p.currency, 'churn', ROUND(p.mrr, 4),"
            . ' p.service_id, p.client_id, p.package_id'
            . ' FROM analytics_service_mrr p'
            . ' LEFT JOIN analytics_service_mrr t'
            . ' ON t.service_id = p.service_id AND t.snapshot_date = ? AND t.currency = p.currency'
            . ' WHERE p.company_id = ? AND p.snapshot_date = ? AND t.service_id IS NULL',
            [$snapshot_date, $snapshot_date, $company_id, $previous_date]
        );

        // New (currency change): appears in a currency absent last snapshot, but the service
        // did exist last snapshot (in another currency)
        $this->Record->query(
            'INSERT INTO analytics_mrr_movements'
            . ' (company_id, date, currency, type, amount, service_id, client_id, package_id)'
            . " SELECT t.company_id, t.snapshot_date, t.currency, 'new', ROUND(t.mrr, 4),"
            . ' t.service_id, t.client_id, t.package_id'
            . ' FROM analytics_service_mrr t'
            . ' LEFT JOIN analytics_service_mrr p'
            . ' ON p.service_id = t.service_id AND p.snapshot_date = ? AND p.currency = t.currency'
            . ' WHERE t.company_id = ? AND t.snapshot_date = ? AND p.service_id IS NULL'
            . ' AND EXISTS (SELECT 1 FROM analytics_service_mrr pp'
            . ' WHERE pp.service_id = t.service_id AND pp.snapshot_date = ?)',
            [$previous_date, $company_id, $snapshot_date, $previous_date]
        );

        // Reactivation: genuinely re-appears and has prior history (any earlier snapshot row,
        // or a cancellation predating the previous snapshot)
        $this->Record->query(
            'INSERT INTO analytics_mrr_movements'
            . ' (company_id, date, currency, type, amount, service_id, client_id, package_id)'
            . " SELECT t.company_id, t.snapshot_date, t.currency, 'reactivation', ROUND(t.mrr, 4),"
            . ' t.service_id, t.client_id, t.package_id'
            . ' FROM analytics_service_mrr t'
            . ' LEFT JOIN analytics_service_mrr p'
            . ' ON p.service_id = t.service_id AND p.snapshot_date = ? AND p.currency = t.currency'
            . ' WHERE t.company_id = ? AND t.snapshot_date = ? AND p.service_id IS NULL'
            . ' AND NOT EXISTS (SELECT 1 FROM analytics_service_mrr pp'
            . ' WHERE pp.service_id = t.service_id AND pp.snapshot_date = ?)'
            . ' AND (EXISTS (SELECT 1 FROM analytics_service_mrr h'
            . ' WHERE h.service_id = t.service_id AND h.snapshot_date < ?)'
            . ' OR EXISTS (SELECT 1 FROM services s2'
            . ' WHERE s2.id = t.service_id AND s2.date_canceled IS NOT NULL AND s2.date_canceled < ?))',
            [$previous_date, $company_id, $snapshot_date, $previous_date, $previous_date, $prev_start_utc]
        );

        // New (genuine): re-appears with no prior history at all
        $this->Record->query(
            'INSERT INTO analytics_mrr_movements'
            . ' (company_id, date, currency, type, amount, service_id, client_id, package_id)'
            . " SELECT t.company_id, t.snapshot_date, t.currency, 'new', ROUND(t.mrr, 4),"
            . ' t.service_id, t.client_id, t.package_id'
            . ' FROM analytics_service_mrr t'
            . ' LEFT JOIN analytics_service_mrr p'
            . ' ON p.service_id = t.service_id AND p.snapshot_date = ? AND p.currency = t.currency'
            . ' WHERE t.company_id = ? AND t.snapshot_date = ? AND p.service_id IS NULL'
            . ' AND NOT EXISTS (SELECT 1 FROM analytics_service_mrr pp'
            . ' WHERE pp.service_id = t.service_id AND pp.snapshot_date = ?)'
            . ' AND NOT (EXISTS (SELECT 1 FROM analytics_service_mrr h'
            . ' WHERE h.service_id = t.service_id AND h.snapshot_date < ?)'
            . ' OR EXISTS (SELECT 1 FROM services s2'
            . ' WHERE s2.id = t.service_id AND s2.date_canceled IS NOT NULL AND s2.date_canceled < ?))',
            [$previous_date, $company_id, $snapshot_date, $previous_date, $previous_date, $prev_start_utc]
        );
    }

    /**
     * Records the snapshot-date exchange rates for every company currency so consolidated
     * (converted-to-default) restatements use a stable historical rate. The stored "rate"
     * is the currencies.exchange_rate value (relative to the company primary/default
     * currency; the default currency's rate is 1). Consolidation converts a native amount to
     * the default currency as amount * (default_rate / native_rate), matching Currencies::convert().
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     */
    private function buildFxRates($company_id, $snapshot_date)
    {
        $this->Record->from('analytics_fx_rates')
            ->where('company_id', '=', $company_id)
            ->where('snapshot_date', '=', $snapshot_date)
            ->delete();

        $this->Record->query(
            'INSERT INTO analytics_fx_rates (company_id, snapshot_date, currency, rate)'
            . ' SELECT ?, ?, code, exchange_rate FROM currencies WHERE company_id = ?',
            [$company_id, $snapshot_date, $company_id]
        );
    }

    /**
     * Precomputes the forward renewal forecast as a 12-month PROJECTION (revised July 18,
     * 2026): each active/suspended recurring service contributes every renewal occurrence
     * within the next FORECAST_MONTHS company-local months, stepped forward from
     * services.date_renews by its term (the service row itself is never modified). The
     * full per-term expected amount (NOT normalized to a month) is bucketed by occurrence
     * month and native currency; renewal_count counts distinct services per month.
     *
     * Occurrences stop at a scheduled cancellation (services.date_canceled). Overdue
     * services (date_renews in the past) contribute only their occurrences that land in
     * the window and are reachable within the per-period occurrence bound below; services
     * overdue by more than the projection window do not project.
     *
     * Performance: occurrences are generated by a small cross-joined sequence, split per
     * period class so the multiplier matches the need — month/year services join at most
     * 13 sequence rows, week 53, day 366 (the latter two only against services actually
     * priced in those periods). Everything remains a handful of set-based statements.
     *
     * Amounts are net of recurring coupons ("expected renewal MRR"): a service's coupon
     * discounts an occurrence when the coupon is active and recurring, subject to its
     * date/quantity limits only when limit_recurring is set — mirroring
     * Coupons::getRecurring(), which does not re-check package membership on renewal.
     * Date limits are evaluated against each occurrence date, so an expiring coupon stops
     * discounting later months. Percent discounts apply to the base (and to options when
     * apply_package_options is set); fixed-amount discounts are subtracted once per
     * occurrence (an accepted approximation of per-invoice application), floored at zero.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (used as computed_date)
     * @param string $timezone The company timezone
     */
    private function buildRenewalForecast($company_id, $snapshot_date, $timezone)
    {
        $this->Record->from('analytics_renewal_forecast')
            ->where('company_id', '=', $company_id)
            ->where('computed_date', '=', $snapshot_date)
            ->delete();

        $start_utc = $this->localToUtc($snapshot_date, '00:00:00', $timezone);
        $end_local = $this->localModify($snapshot_date, '+' . self::FORECAST_MONTHS . ' months', $timezone);
        $end_utc = $this->localToUtc($end_local, '00:00:00', $timezone);

        // Bucket occurrence months in company-local time using the zone's current UTC
        // offset (a fixed offset per build; rows within a DST shift of a month boundary
        // may land one month off — an accepted approximation for a forward-looking forecast)
        $offset_minutes = (int) round(
            (new DateTimeZone($timezone))->getOffset(new DateTime($start_utc, new DateTimeZone('UTC'))) / 60
        );
        $month_bucket = "DATE_FORMAT(DATE_ADD(o.occurrence, INTERVAL {$offset_minutes} MINUTE), '%Y-%m-01')";

        // Full per-term renewal amount of each recurring configurable option, scoped to
        // this company and grouped per option currency (mismatched currencies excluded
        // at the join rather than mis-summed)
        $option_join = '(SELECT so.service_id AS service_id, op.currency AS currency,'
            . ' SUM(COALESCE(op.price_renews, op.price) * so.qty) AS option_amount'
            . ' FROM service_options so'
            . ' INNER JOIN services sv ON sv.id = so.service_id'
            . ' INNER JOIN clients cc ON cc.id = sv.client_id'
            . ' INNER JOIN client_groups cgg ON cgg.id = cc.client_group_id'
            . ' INNER JOIN package_option_pricing pop ON pop.id = so.option_pricing_id'
            . ' INNER JOIN pricings op ON op.id = pop.pricing_id'
            . " WHERE op.period != 'onetime'"
            . ' AND cgg.company_id = ?'
            . ' GROUP BY so.service_id, op.currency) optf';

        $base_amount = '(o.base_amount)';
        $option_amount = '(o.option_amount)';
        $gross_amount = '(o.base_amount + o.option_amount)';

        // Expected occurrence amount net of an applicable recurring coupon
        $expected_amount = 'GREATEST(0, CASE'
            . ' WHEN ca.coupon_id IS NULL THEN ' . $gross_amount
            . " WHEN ca.type = 'percent' THEN CASE WHEN cp.apply_package_options = 1"
            . ' THEN ' . $gross_amount . ' * (1 - ca.amount / 100)'
            . ' ELSE ' . $base_amount . ' * (1 - ca.amount / 100) + ' . $option_amount . ' END'
            . ' ELSE ' . $gross_amount . ' - ca.amount END)';

        // Period classes: bounded occurrence sequence per term family. The occurrence
        // date advances by (k * term) months for month/year terms and (k * term) days
        // for day/week terms (weeks as 7 days).
        $classes = [
            [
                'periods' => "('month', 'year')",
                'max_k' => self::FORECAST_MONTHS + 1,
                'occurrence' => 'DATE_ADD(s.date_renews, INTERVAL (seq.k * p.term'
                    . " * (CASE p.period WHEN 'year' THEN 12 ELSE 1 END)) MONTH)"
            ],
            [
                'periods' => "('week')",
                'max_k' => 53,
                'occurrence' => 'DATE_ADD(s.date_renews, INTERVAL (seq.k * p.term * 7) DAY)'
            ],
            [
                'periods' => "('day')",
                'max_k' => 366,
                'occurrence' => 'DATE_ADD(s.date_renews, INTERVAL (seq.k * p.term) DAY)'
            ]
        ];

        foreach ($classes as $class) {
            $occurrence = $class['occurrence'];

            $occurrences = '(SELECT s.id AS service_id, s.coupon_id AS coupon_id,'
                . ' COALESCE(s.override_currency, p.currency) AS currency,'
                . ' (COALESCE(s.override_price, p.price_renews, p.price) * s.qty) AS base_amount,'
                . ' COALESCE(optf.option_amount, 0) AS option_amount,'
                . ' ' . $occurrence . ' AS occurrence'
                . ' FROM services s'
                . ' INNER JOIN clients c ON c.id = s.client_id'
                . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
                . ' INNER JOIN package_pricing pp ON pp.id = s.pricing_id'
                . ' INNER JOIN pricings p ON p.id = pp.pricing_id'
                . ' LEFT JOIN ' . $option_join
                . ' ON optf.service_id = s.id AND optf.currency = COALESCE(s.override_currency, p.currency)'
                . ' CROSS JOIN ' . $this->sequenceSql($class['max_k']) . ' seq'
                . ' WHERE cg.company_id = ?'
                . " AND s.status IN ('active', 'suspended')"
                . ' AND p.period IN ' . $class['periods']
                . ' AND s.date_renews IS NOT NULL AND s.date_renews < ?'
                . ' AND ' . $occurrence . ' >= ? AND ' . $occurrence . ' < ?'
                . ' AND (s.date_canceled IS NULL OR ' . $occurrence . ' < s.date_canceled)'
                . ') o';

            $this->Record->query(
                'INSERT INTO analytics_renewal_forecast'
                . ' (company_id, computed_date, month, currency, renewal_amount, renewal_count)'
                . ' SELECT ?, ?, ' . $month_bucket . ', o.currency,'
                . ' ROUND(SUM(' . $expected_amount . '), 4),'
                . ' COUNT(DISTINCT o.service_id)'
                . ' FROM ' . $occurrences
                . " LEFT JOIN coupons cp ON cp.id = o.coupon_id AND cp.status = 'active' AND cp.recurring = 1"
                . ' AND (cp.limit_recurring = 0 OR ((cp.max_qty = 0 OR cp.used_qty < cp.max_qty)'
                . ' AND (cp.start_date IS NULL OR o.occurrence >= cp.start_date)'
                . ' AND (cp.end_date IS NULL OR o.occurrence <= cp.end_date)))'
                . ' LEFT JOIN coupon_amounts ca ON ca.coupon_id = cp.id AND ca.currency = o.currency'
                . ' GROUP BY ' . $month_bucket . ', o.currency'
                . ' ON DUPLICATE KEY UPDATE'
                . ' renewal_amount = ROUND(renewal_amount + VALUES(renewal_amount), 4),'
                . ' renewal_count = renewal_count + VALUES(renewal_count)',
                [$company_id, $snapshot_date, $company_id, $company_id, $end_utc, $start_utc, $end_utc]
            );
        }
    }

    /**
     * Builds a derived table of sequential integers 0..$max as a digits cross join
     * (bounded at 0..999), aliased column k.
     *
     * @param int $max The maximum sequence value (inclusive)
     * @return string The derived-table SQL (unaliased)
     */
    private function sequenceSql($max)
    {
        $max = max(0, min(999, (int) $max));
        $digits = '(SELECT 0 AS d UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3'
            . ' UNION ALL SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7'
            . ' UNION ALL SELECT 8 UNION ALL SELECT 9)';

        return '(SELECT u.d + t.d * 10 + h.d * 100 AS k'
            . ' FROM ' . $digits . ' u, ' . $digits . ' t, ' . $digits . ' h'
            . ' WHERE u.d + t.d * 10 + h.d * 100 <= ' . $max . ')';
    }

    /**
     * Builds the dimensional cash/invoice facts for a single company-local date, replacing
     * any existing rows for that date. Populates the 'total', 'gateway' and 'package'
     * dimensions ('package_group' is reserved and not populated in Phase 1).
     *
     * The 'refunded' measure is NOT regenerated here (Phase 2, §2.2): refunds are posted
     * on their refund dates by applyRefundEvents() as edits are consumed from the
     * transaction log, and a rebuild carries the date's previously posted values forward.
     * Collected amounts are restated to the original gross for transactions whose amount
     * was mutated by refund processing (see the gross recovery below).
     *
     * @param int $company_id The company ID
     * @param string $date The company-local fact date (Y-m-d)
     * @param string $timezone The company timezone
     */
    private function buildRevenueForDate($company_id, $date, $timezone)
    {
        $start_utc = $this->localToUtc($date, '00:00:00', $timezone);
        $end_utc = $this->localToUtc($this->localModify($date, '+1 day', $timezone), '00:00:00', $timezone);

        // Preserve the date's accumulated refund postings across the rebuild
        $carried = $this->Record->query(
            'SELECT currency, dimension, dimension_id, refunded FROM analytics_revenue_daily'
            . ' WHERE company_id = ? AND date = ? AND refunded != 0'
            . " AND dimension IN ('total', 'gateway')",
            [$company_id, $date]
        )->fetchAll(PDO::FETCH_OBJ);

        // Clear the date's rows first (idempotent regeneration)
        $this->Record->from('analytics_revenue_daily')
            ->where('company_id', '=', $company_id)
            ->where('date', '=', $date)
            ->delete();

        // Settled revenue-type transaction predicate. Refunds mutate the original
        // transaction in place (status -> refunded/returned, and a partial refund
        // replaces the amount with the refunded portion — see GatewayPayments::refundPayment),
        // so refunded/returned rows must still count as collected or a refund would
        // erase the original collection and net negative. Partial-refund rows carry only
        // the refunded portion; the gross recovery below restates them to the original
        // amount recovered from the transaction edit log.
        // The type filter mirrors BillingOverviewStatistics::getRevenue
        $revenue_predicate = " AND t.status IN ('approved', 'refunded', 'returned')"
            . " AND (t.type != 'other' OR (t.type = 'other'"
            . ' AND (tt.type = ? OR t.transaction_type_id IS NULL)))';

        // -- Gross recovery (§2.2): per-currency/gateway corrections restoring the
        // original gross of edited transactions dated this date --
        $gross_deltas = $this->getGrossRecoveryDeltas($company_id, $start_utc, $end_utc, $revenue_predicate);

        // -- Totals (dimension 'total', dimension_id 0) --
        $totals = [];

        $collected = $this->Record->query(
            'SELECT t.currency AS currency, ROUND(SUM(t.amount), 4) AS amount'
            . ' FROM transactions t'
            . ' INNER JOIN clients c ON c.id = t.client_id'
            . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
            . ' LEFT JOIN transaction_types tt ON tt.id = t.transaction_type_id'
            . ' WHERE cg.company_id = ? AND t.date_added >= ? AND t.date_added < ?'
            . $revenue_predicate
            . ' GROUP BY t.currency',
            [$company_id, $start_utc, $end_utc, 'debit']
        )->fetchAll(PDO::FETCH_OBJ);
        foreach ($collected as $row) {
            $totals[$row->currency]['collected'] = $row->amount;
        }

        // Apply gross recovery to the totals
        foreach ($gross_deltas['totals'] as $currency => $delta) {
            $totals[$currency]['collected'] = round(($totals[$currency]['collected'] ?? 0) + $delta, 4);
        }

        $credited = $this->Record->query(
            'SELECT t.currency AS currency, ROUND(SUM(t.amount), 4) AS amount'
            . ' FROM transactions t'
            . ' INNER JOIN clients c ON c.id = t.client_id'
            . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
            . ' INNER JOIN transaction_types tt ON tt.id = t.transaction_type_id'
            . " WHERE cg.company_id = ? AND t.status = 'approved' AND t.type = 'other' AND tt.type = 'credit'"
            . ' AND t.date_added >= ? AND t.date_added < ?'
            . ' GROUP BY t.currency',
            [$company_id, $start_utc, $end_utc]
        )->fetchAll(PDO::FETCH_OBJ);
        foreach ($credited as $row) {
            $totals[$row->currency]['credited'] = $row->amount;
        }

        $invoiced = $this->Record->query(
            'SELECT i.currency AS currency, ROUND(SUM(i.total), 4) AS total,'
            . ' ROUND(SUM(i.total - i.subtotal), 4) AS tax'
            . ' FROM invoices i'
            . ' INNER JOIN clients c ON c.id = i.client_id'
            . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
            . " WHERE cg.company_id = ? AND i.status IN ('active', 'proforma')"
            . ' AND i.date_billed >= ? AND i.date_billed < ?'
            . ' GROUP BY i.currency',
            [$company_id, $start_utc, $end_utc]
        )->fetchAll(PDO::FETCH_OBJ);
        foreach ($invoiced as $row) {
            $totals[$row->currency]['invoiced'] = $row->total;
            $totals[$row->currency]['invoiced_tax'] = $row->tax;
        }

        // Carry the previously posted refund totals forward
        foreach ($carried as $row) {
            if ($row->dimension === 'total') {
                $totals[$row->currency]['refunded'] = $row->refunded;
            }
        }

        foreach ($totals as $currency => $measures) {
            $this->insertRevenueRow($company_id, $date, $currency, 'total', 0, $measures);
        }

        // -- By gateway (dimension 'gateway') --
        $gateways = [];

        $gw_collected = $this->Record->query(
            'SELECT COALESCE(t.gateway_id, 0) AS dimension_id, t.currency AS currency,'
            . ' ROUND(SUM(t.amount), 4) AS amount'
            . ' FROM transactions t'
            . ' INNER JOIN clients c ON c.id = t.client_id'
            . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
            . ' LEFT JOIN transaction_types tt ON tt.id = t.transaction_type_id'
            . ' WHERE cg.company_id = ? AND t.date_added >= ? AND t.date_added < ?'
            . $revenue_predicate
            . ' GROUP BY COALESCE(t.gateway_id, 0), t.currency',
            [$company_id, $start_utc, $end_utc, 'debit']
        )->fetchAll(PDO::FETCH_OBJ);
        foreach ($gw_collected as $row) {
            $gateways[$row->dimension_id][$row->currency]['collected'] = $row->amount;
        }

        // Apply gross recovery per gateway
        foreach ($gross_deltas['gateways'] as $gateway_id => $currencies) {
            foreach ($currencies as $currency => $delta) {
                $gateways[$gateway_id][$currency]['collected']
                    = round(($gateways[$gateway_id][$currency]['collected'] ?? 0) + $delta, 4);
            }
        }

        // Carry the previously posted per-gateway refunds forward
        foreach ($carried as $row) {
            if ($row->dimension === 'gateway') {
                $gateways[(int) $row->dimension_id][$row->currency]['refunded'] = $row->refunded;
            }
        }

        foreach ($gateways as $dimension_id => $currencies) {
            foreach ($currencies as $currency => $measures) {
                $this->insertRevenueRow($company_id, $date, $currency, 'gateway', (int) $dimension_id, $measures);
            }
        }

        // -- By package (dimension 'package'; collected only, allocated pro-rata by ex-tax line total) --
        $pkg_collected = $this->Record->query(
            'SELECT COALESCE(pp.package_id, 0) AS dimension_id, x.currency AS currency,'
            . ' ROUND(SUM(x.alloc), 4) AS amount'
            . ' FROM ('
            . ' SELECT t.currency AS currency, il.service_id AS service_id,'
            . ' ta.amount * (il.qty * il.amount) / lt.line_total AS alloc'
            . ' FROM transactions t'
            . ' INNER JOIN clients c ON c.id = t.client_id'
            . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
            . ' LEFT JOIN transaction_types tt ON tt.id = t.transaction_type_id'
            . ' INNER JOIN transaction_applied ta ON ta.transaction_id = t.id'
            . ' INNER JOIN invoice_lines il ON il.invoice_id = ta.invoice_id'
            . ' INNER JOIN (SELECT invoice_id, SUM(qty * amount) AS line_total'
            . ' FROM invoice_lines GROUP BY invoice_id) lt ON lt.invoice_id = ta.invoice_id'
            . ' WHERE cg.company_id = ? AND t.date_added >= ? AND t.date_added < ?'
            . $revenue_predicate
            . ' AND lt.line_total != 0'
            . ' ) x'
            . ' LEFT JOIN services s ON s.id = x.service_id'
            . ' LEFT JOIN package_pricing pp ON pp.id = s.pricing_id'
            . ' GROUP BY COALESCE(pp.package_id, 0), x.currency',
            [$company_id, $start_utc, $end_utc, 'debit']
        )->fetchAll(PDO::FETCH_OBJ);
        foreach ($pkg_collected as $row) {
            $this->insertRevenueRow(
                $company_id,
                $date,
                $row->currency,
                'package',
                (int) $row->dimension_id,
                ['collected' => $row->amount]
            );
        }
    }

    /**
     * Computes per-currency and per-gateway 'collected' corrections restoring the
     * original gross of transactions (dated within the given UTC window) whose amount
     * was mutated by refund processing (§2.2). A partial refund replaces the row's
     * amount with the refunded portion, so the base collected query understates gross;
     * the correction is (original gross - current amount) per transaction, where gross
     * follows genuine (non-refund) amount edits but ignores refund bookkeeping.
     *
     * @param int $company_id The company ID
     * @param string $start_utc The window start (UTC datetime, inclusive)
     * @param string $end_utc The window end (UTC datetime, exclusive)
     * @param string $revenue_predicate The shared settled-revenue SQL predicate
     * @return array An array containing:
     *  - totals A map of currency => collected delta
     *  - gateways A map of gateway_id => currency => collected delta
     */
    private function getGrossRecoveryDeltas($company_id, $start_utc, $end_utc, $revenue_predicate)
    {
        $deltas = ['totals' => [], 'gateways' => []];

        $edited = $this->Record->query(
            'SELECT t.id, t.currency, COALESCE(t.gateway_id, 0) AS gateway_id, t.amount, t.status'
            . ' FROM transactions t'
            . ' INNER JOIN clients c ON c.id = t.client_id'
            . ' INNER JOIN client_groups cg ON cg.id = c.client_group_id'
            . ' LEFT JOIN transaction_types tt ON tt.id = t.transaction_type_id'
            . ' WHERE cg.company_id = ? AND t.date_added >= ? AND t.date_added < ?'
            . $revenue_predicate
            . ' AND EXISTS (SELECT 1 FROM log_transactions lt WHERE lt.transaction_id = t.id)',
            [$company_id, $start_utc, $end_utc, 'debit']
        )->fetchAll(PDO::FETCH_OBJ);

        if (empty($edited)) {
            return $deltas;
        }

        foreach (array_chunk($edited, 500) as $chunk) {
            $ids = [];
            foreach ($chunk as $transaction) {
                $ids[] = (int) $transaction->id;
            }
            $placeholders = implode(', ', array_fill(0, count($ids), '?'));

            $log_rows = $this->Record->query(
                'SELECT id, transaction_id, `change` FROM log_transactions'
                . ' WHERE transaction_id IN (' . $placeholders . ')'
                . ' ORDER BY transaction_id ASC, id ASC',
                $ids
            )->fetchAll(PDO::FETCH_OBJ);

            $histories = [];
            foreach ($log_rows as $log_row) {
                $histories[(int) $log_row->transaction_id][] = $log_row;
            }

            foreach ($chunk as $transaction) {
                $history = $histories[(int) $transaction->id] ?? [];
                if (empty($history)) {
                    continue;
                }

                $gross = $this->computeGrossAmount($transaction, $history);
                $delta = round($gross - (float) $transaction->amount, 4);
                if ($delta == 0.0) {
                    continue;
                }

                $currency = $transaction->currency;
                $gateway_id = (int) $transaction->gateway_id;
                $deltas['totals'][$currency] = ($deltas['totals'][$currency] ?? 0.0) + $delta;
                $deltas['gateways'][$gateway_id][$currency]
                    = ($deltas['gateways'][$gateway_id][$currency] ?? 0.0) + $delta;
            }
        }

        return $deltas;
    }

    /**
     * Replays a transaction's log history to determine its original gross amount:
     * the initial amount, updated by genuine amount corrections but NOT by refund
     * bookkeeping (amount changes made while entering or already in a
     * refunded/returned status replace the amount with the refunded portion).
     *
     * @param stdClass $transaction The current transaction row (amount, status)
     * @param array $log_rows The transaction's log rows, ordered by id ascending
     * @return float The reconstructed gross amount
     */
    private function computeGrossAmount($transaction, array $log_rows)
    {
        $refund_statuses = ['refunded', 'returned'];

        $parsed = [];
        foreach ($log_rows as $log_row) {
            $parsed[$log_row->id] = $this->parseTransactionLogChange($log_row->change);
        }

        // Initial state from the earliest logged prev values (fall back to current)
        $amount = (float) $transaction->amount;
        $status = $transaction->status;
        foreach ($log_rows as $log_row) {
            if (isset($parsed[$log_row->id]['amount']['prev'])) {
                $amount = (float) $parsed[$log_row->id]['amount']['prev'];
                break;
            }
        }
        foreach ($log_rows as $log_row) {
            if (isset($parsed[$log_row->id]['status']['prev'])) {
                $status = $parsed[$log_row->id]['status']['prev'];
                break;
            }
        }

        $gross = $amount;
        foreach ($log_rows as $log_row) {
            $fields = $parsed[$log_row->id];
            if ($fields === null) {
                continue;
            }

            $new_amount = isset($fields['amount']['cur']) ? (float) $fields['amount']['cur'] : $amount;
            $new_status = isset($fields['status']['cur']) ? $fields['status']['cur'] : $status;

            if ($new_amount != $amount
                && !in_array($status, $refund_statuses, true)
                && !in_array($new_status, $refund_statuses, true)
            ) {
                // A genuine correction outside refund processing moves the gross
                $gross = $new_amount;
            }

            $amount = $new_amount;
            $status = $new_status;
        }

        return $gross;
    }

    /**
     * Inserts a single analytics_revenue_daily row from a measure map.
     *
     * @param int $company_id The company ID
     * @param string $date The company-local fact date (Y-m-d)
     * @param string $currency The currency code
     * @param string $dimension The dimension (total|gateway|package)
     * @param int $dimension_id The dimension id (0 = total/unassigned)
     * @param array $measures Any of collected/refunded/invoiced/invoiced_tax/credited
     */
    private function insertRevenueRow($company_id, $date, $currency, $dimension, $dimension_id, array $measures)
    {
        $this->Record->insert('analytics_revenue_daily', [
            'company_id' => $company_id,
            'date' => $date,
            'currency' => $currency,
            'dimension' => $dimension,
            'dimension_id' => $dimension_id,
            'collected' => $measures['collected'] ?? 0,
            'refunded' => $measures['refunded'] ?? 0,
            'invoiced' => $measures['invoiced'] ?? 0,
            'invoiced_tax' => $measures['invoiced_tax'] ?? 0,
            'credited' => $measures['credited'] ?? 0
        ]);
    }

    /**
     * Prunes per-service daily rows older than the retention window, preserving the last
     * completed snapshot row of each company-local month (kept forever) so cohort/retention
     * reads always have a monthly anchor.
     *
     * @param int $company_id The company ID
     * @param string $snapshot_date The company-local snapshot date (Y-m-d)
     * @param string $timezone The company timezone
     */
    private function pruneServiceMrr($company_id, $snapshot_date, $timezone)
    {
        $retention_days = (int) $this->getCompanySettingValue($company_id, 'analytics_retention_days', 90);
        if ($retention_days < 1) {
            $retention_days = 90;
        }

        $cutoff = $this->localModify($snapshot_date, '-' . $retention_days . ' days', $timezone);

        $this->Record->query(
            'DELETE FROM analytics_service_mrr'
            . ' WHERE company_id = ? AND snapshot_date < ?'
            . ' AND snapshot_date NOT IN ('
            . ' SELECT keep_date FROM ('
            . ' SELECT MAX(snapshot_date) AS keep_date FROM analytics_snapshot_runs'
            . " WHERE company_id = ? AND status = 'completed'"
            . ' GROUP BY YEAR(snapshot_date), MONTH(snapshot_date)) keep)',
            [$company_id, $cutoff, $company_id]
        );
    }

    /**
     * Builds the CASE expression converting a pricing's term/period into a number of months.
     *
     * @param string $alias The pricings table alias
     * @return string The SQL CASE expression
     */
    private function termInMonthsSql($alias)
    {
        return 'CASE ' . $alias . '.period'
            . ' WHEN ' . "'month' THEN " . $alias . '.term'
            . ' WHEN ' . "'year' THEN " . $alias . '.term * 12'
            . ' WHEN ' . "'week' THEN " . $alias . '.term * 7 / 30.4375'
            . ' WHEN ' . "'day' THEN " . $alias . '.term / 30.4375'
            . ' ELSE NULL END';
    }

    /**
     * Returns the company timezone setting (defaults to UTC).
     *
     * @param int $company_id The company ID
     * @return string The timezone identifier
     */
    private function getCompanyTimezone($company_id)
    {
        $value = $this->getCompanySettingValue($company_id, 'timezone', 'UTC');

        return ($value !== null && $value !== '') ? $value : 'UTC';
    }

    /**
     * Returns a single company setting value, or a default if unset.
     *
     * @param int $company_id The company ID
     * @param string $key The setting key
     * @param mixed $default The value to return when the setting is missing
     * @return mixed The setting value or the default
     */
    private function getCompanySettingValue($company_id, $key, $default = null)
    {
        $setting = $this->Companies->getSetting($company_id, $key);

        return ($setting && isset($setting->value)) ? $setting->value : $default;
    }

    /**
     * Returns the current instant rendered in the given timezone.
     *
     * @param string $timezone The timezone identifier
     * @param string $format The date format to return
     * @return string The formatted local date/time
     */
    private function localNow($timezone, $format = 'Y-m-d')
    {
        $date = clone $this->Date;
        $date->setTimezone('UTC', $timezone);

        return $date->format($format, gmdate('Y-m-d H:i:s'));
    }

    /**
     * Applies a relative modification to a company-local date, staying in local time.
     *
     * @param string $local_date The company-local date (Y-m-d)
     * @param string $modifier A strtotime-style modifier (e.g. '+1 day')
     * @param string $timezone The timezone identifier
     * @return string The resulting company-local date (Y-m-d)
     */
    private function localModify($local_date, $modifier, $timezone)
    {
        $date = clone $this->Date;
        $date->setTimezone($timezone, $timezone);

        return $date->modify($local_date . ' 00:00:00', $modifier, 'Y-m-d');
    }

    /**
     * Converts a company-local date + time to a UTC datetime string for querying UTC columns.
     *
     * @param string $local_date The company-local date (Y-m-d)
     * @param string $time The company-local time (H:i:s)
     * @param string $timezone The timezone identifier
     * @return string The UTC datetime (Y-m-d H:i:s)
     */
    private function localToUtc($local_date, $time, $timezone)
    {
        $date = clone $this->Date;
        $date->setTimezone($timezone, 'UTC');

        return $date->format('Y-m-d H:i:s', $local_date . ' ' . $time);
    }

    /**
     * Converts a UTC datetime to the company-local date
     *
     * @param string $utc_datetime The UTC datetime (Y-m-d H:i:s)
     * @param string $timezone The company timezone
     * @return string The company-local date (Y-m-d)
     */
    private function utcToLocalDate($utc_datetime, $timezone)
    {
        $date = clone $this->Date;
        $date->setTimezone('UTC', $timezone);

        return $date->format('Y-m-d', $utc_datetime);
    }
}
