wms-app/app/assets/utils/classes_ac/EtlManager.php

<?php
/**
 * EtlManager
 *
 * Manages etl_gl_summary — the materialized aggregation of td_gl_item by period.
 * Used for fast dashboard queries instead of scanning raw GL rows.
 *
 * rebuildFull()   — wipe and rebuild all periods for this company (manual trigger)
 * rebuildPeriod() — rebuild a single period (nightly scheduler, gap repair)
 * getStatus()     — return last run info from company_setting
 */
class EtlManager
{
    private PDO $pdo;
    private PDO $pdo1;
    private int $companyId;

    public function __construct(PDO $pdo, PDO $pdo1, int $company_id)
    {
        $this->pdo       = $pdo;
        $this->pdo1      = $pdo1;
        $this->companyId = $company_id;
    }

    // ── Public ───────────────────────────────────────────────────────────────

    public function checkAndRepair(): array
    {
        $gaps = $this->getGaps();

        foreach ($gaps as $period) {
            $this->rebuildPeriod($period);
        }

        $gaps_found      = count($gaps);
        $periods_rebuilt = $gaps_found;
        $this->saveStatus('ok', $gaps_found, $periods_rebuilt);

        return ['gaps_found' => $gaps_found, 'periods_rebuilt' => $periods_rebuilt];
    }

    public function getGaps(): array
    {
        $sth = $this->pdo->prepare("
            SELECT DISTINCT g.period
            FROM td_gl g
            LEFT JOIN (
                SELECT period, MIN(source_updated_at) AS etl_ts
                FROM etl_gl_summary
                WHERE company_id = :cid2
                GROUP BY period
            ) etl ON etl.period = g.period
            WHERE g.company_id = :cid
              AND g.source_type NOT IN ('voided', 'reversal')
              AND (
                  etl.period IS NULL
                  OR g.updated_at > etl.etl_ts
              )
            ORDER BY g.period
        ");
        $sth->execute([':cid' => $this->companyId, ':cid2' => $this->companyId]);
        return $sth->fetchAll(PDO::FETCH_COLUMN);
    }

    public function rebuildFull(): array
    {
        $this->pdo->prepare(
            "DELETE FROM etl_gl_summary WHERE company_id = :cid"
        )->execute([':cid' => $this->companyId]);

        $this->pdo->prepare("
            INSERT INTO etl_gl_summary
                (company_id, acc_code, period, debit, credit, count, source_updated_at, updated_at)
            SELECT
                i.company_id,
                i.account_code,
                g.period,
                SUM(i.debit),
                SUM(i.credit),
                COUNT(*),
                MAX(g.updated_at),
                NOW()
            FROM td_gl_item i
            JOIN td_gl g ON g.id = i.gl_id AND g.company_id = i.company_id
            WHERE i.company_id = :cid
              AND g.source_type != 'voided'
            GROUP BY i.company_id, i.account_code, g.period
        ")->execute([':cid' => $this->companyId]);

        $periods_rebuilt = (int)$this->pdo->query(
            "SELECT COUNT(DISTINCT period) FROM etl_gl_summary WHERE company_id = {$this->companyId}"
        )->fetchColumn();

        $this->saveStatus('ok', 0, $periods_rebuilt);

        return ['periods_rebuilt' => $periods_rebuilt];
    }

    public function rebuildPeriod(string $period): void
    {
        $this->pdo->prepare(
            "DELETE FROM etl_gl_summary WHERE company_id = :cid AND period = :period"
        )->execute([':cid' => $this->companyId, ':period' => $period]);

        $this->pdo->prepare("
            INSERT INTO etl_gl_summary
                (company_id, acc_code, period, debit, credit, count, source_updated_at, updated_at)
            SELECT
                i.company_id,
                i.account_code,
                g.period,
                SUM(i.debit),
                SUM(i.credit),
                COUNT(*),
                MAX(g.updated_at),
                NOW()
            FROM td_gl_item i
            JOIN td_gl g ON g.id = i.gl_id AND g.company_id = i.company_id
            WHERE i.company_id = :cid
              AND g.period = :period
              AND g.source_type != 'voided'
            GROUP BY i.company_id, i.account_code, g.period
        ")->execute([':cid' => $this->companyId, ':period' => $period]);
    }

    public function getStatus(): array
    {
        $keys = ['etl_last_ran', 'etl_last_result', 'etl_gaps_found', 'etl_periods_rebuilt'];
        $sth  = $this->pdo1->prepare(
            "SELECT setting_key, value FROM company_setting
             WHERE company_id = :cid AND setting_key IN ('" . implode("','", $keys) . "')"
        );
        $sth->execute([':cid' => $this->companyId]);
        $rows = $sth->fetchAll(PDO::FETCH_KEY_PAIR);

        $slot_hour = $this->companyId % 24;
        $last_ran  = $rows['etl_last_ran'] ?? null;
        $next_run  = $last_ran
            ? date('Y-m-d', strtotime($last_ran . ' +1 day')) . sprintf(' %02d:00', $slot_hour)
            : date('Y-m-d') . sprintf(' %02d:00', $slot_hour);

        return [
            'slot_hour'        => $slot_hour,
            'last_ran'         => $last_ran,
            'last_result'      => $rows['etl_last_result']      ?? null,
            'gaps_found'       => (int)($rows['etl_gaps_found']       ?? 0),
            'periods_rebuilt'  => (int)($rows['etl_periods_rebuilt']  ?? 0),
            'next_run'         => $next_run,
        ];
    }

    // ── Private ──────────────────────────────────────────────────────────────

    private function saveStatus(string $result, int $gaps_found, int $periods_rebuilt): void
    {
        $now  = date('Y-m-d H:i:s');
        $data = [
            'etl_last_ran'        => $now,
            'etl_last_result'     => $result,
            'etl_gaps_found'      => (string)$gaps_found,
            'etl_periods_rebuilt' => (string)$periods_rebuilt,
        ];
        $sth = $this->pdo1->prepare(
            "INSERT INTO company_setting (company_id, setting_key, value, updated_at)
             VALUES (:cid, :key, :val, :ts)
             ON DUPLICATE KEY UPDATE value = VALUES(value), updated_at = VALUES(updated_at)"
        );
        foreach ($data as $key => $val) {
            $sth->execute([':cid' => $this->companyId, ':key' => $key, ':val' => $val, ':ts' => $now]);
        }
    }
}