232 lines
9.1 KiB
PHP
232 lines
9.1 KiB
PHP
<?php
|
|
/**
|
|
* EtlStockManager
|
|
*
|
|
* Manages etl_stock_summary — the materialized aggregate of td_stock_<warehouse_id>
|
|
* by (company_id, warehouse_id, product_sku, month). Mirrors EtlManager for GL.
|
|
*
|
|
* etl_stock_summary is maintained two ways:
|
|
* 1. Write-through — adjustBalance() in WarehouseManager updates it on every approve/delete.
|
|
* 2. Batch repair — checkAndRepair() scans td_stock_* for drift and rebuilds stale months.
|
|
*
|
|
* Gap detection is timestamp-based (identical to GL):
|
|
* A month is stale when any td_stock_N row (updated_at) is newer than the
|
|
* oldest etl_stock_summary entry for that warehouse+month (source_updated_at).
|
|
*
|
|
* checkAndRepair() — scan all warehouses, rebuild stale months, save status
|
|
* getGaps() — return [['warehouse_id'=>N,'period'=>'YYYY-MM'], ...]
|
|
* rebuildPeriod(wh_id, period) — DELETE + reinsert one (warehouse, month) slice
|
|
* rebuildFull(wh_id) — DELETE + rebuild all months for one warehouse
|
|
* getDistinctPeriods(wh_id) — distinct approved months in td_stock_N
|
|
* getStatus() — read etl_stock_* keys from company_setting
|
|
*/
|
|
class EtlStockManager
|
|
{
|
|
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 $gap) {
|
|
$this->rebuildPeriod($gap['warehouse_id'], $gap['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
|
|
{
|
|
$stock_tables = $this->discoverStockTables();
|
|
$gaps = [];
|
|
|
|
foreach ($stock_tables as $table) {
|
|
$warehouse_id = (int)substr($table, strlen('td_stock_'));
|
|
|
|
$sth = $this->pdo->prepare("
|
|
SELECT DISTINCT DATE_FORMAT(s.date, '%Y-%m') AS period
|
|
FROM `{$table}` s
|
|
LEFT JOIN (
|
|
SELECT month, MAX(source_updated_at) AS etl_ts
|
|
FROM etl_stock_summary
|
|
WHERE company_id = :cid2 AND warehouse_id = :wh2
|
|
GROUP BY month
|
|
) etl ON etl.month = DATE_FORMAT(s.date, '%Y-%m')
|
|
WHERE s.company_id = :cid
|
|
AND s.status = 1
|
|
AND (etl.month IS NULL OR s.updated_at > etl.etl_ts)
|
|
ORDER BY period
|
|
");
|
|
$sth->execute([
|
|
':cid' => $this->companyId,
|
|
':cid2' => $this->companyId,
|
|
':wh2' => $warehouse_id,
|
|
]);
|
|
|
|
foreach ($sth->fetchAll(PDO::FETCH_COLUMN) as $period) {
|
|
$gaps[] = ['warehouse_id' => $warehouse_id, 'period' => $period];
|
|
}
|
|
}
|
|
|
|
return $gaps;
|
|
}
|
|
|
|
public function rebuildPeriod(int $warehouse_id, string $period): void
|
|
{
|
|
$table = 'td_stock_' . $warehouse_id;
|
|
|
|
$this->pdo->prepare(
|
|
"DELETE FROM etl_stock_summary
|
|
WHERE company_id = :cid AND warehouse_id = :wh AND month = :month"
|
|
)->execute([':cid' => $this->companyId, ':wh' => $warehouse_id, ':month' => $period]);
|
|
|
|
$this->pdo->prepare("
|
|
INSERT INTO etl_stock_summary
|
|
(company_id, warehouse_id, product_sku, month, total_in, total_out, source_updated_at)
|
|
SELECT
|
|
company_id,
|
|
:wh,
|
|
product_sku,
|
|
DATE_FORMAT(`date`, '%Y-%m') AS month,
|
|
ROUND(SUM(`in`), 4) AS total_in,
|
|
ROUND(SUM(`out`), 4) AS total_out,
|
|
MAX(updated_at) AS source_updated_at
|
|
FROM `{$table}`
|
|
WHERE company_id = :cid
|
|
AND status = 1
|
|
AND DATE_FORMAT(`date`, '%Y-%m') = :month
|
|
GROUP BY company_id, product_sku, DATE_FORMAT(`date`, '%Y-%m')
|
|
")->execute([':cid' => $this->companyId, ':wh' => $warehouse_id, ':month' => $period]);
|
|
}
|
|
|
|
public function rebuildFull(int $warehouse_id): array
|
|
{
|
|
$table = 'td_stock_' . $warehouse_id;
|
|
|
|
$this->pdo->prepare(
|
|
"DELETE FROM etl_stock_summary WHERE company_id = :cid AND warehouse_id = :wh"
|
|
)->execute([':cid' => $this->companyId, ':wh' => $warehouse_id]);
|
|
|
|
$this->pdo->prepare("
|
|
INSERT INTO etl_stock_summary
|
|
(company_id, warehouse_id, product_sku, month, total_in, total_out, source_updated_at)
|
|
SELECT
|
|
company_id,
|
|
:wh,
|
|
product_sku,
|
|
DATE_FORMAT(`date`, '%Y-%m') AS month,
|
|
ROUND(SUM(`in`), 4) AS total_in,
|
|
ROUND(SUM(`out`), 4) AS total_out,
|
|
MAX(updated_at) AS source_updated_at
|
|
FROM `{$table}`
|
|
WHERE company_id = :cid AND status = 1
|
|
GROUP BY company_id, product_sku, DATE_FORMAT(`date`, '%Y-%m')
|
|
")->execute([':cid' => $this->companyId, ':wh' => $warehouse_id]);
|
|
|
|
$sth = $this->pdo->prepare(
|
|
"SELECT COUNT(DISTINCT month) FROM etl_stock_summary
|
|
WHERE company_id = :cid AND warehouse_id = :wh"
|
|
);
|
|
$sth->execute([':cid' => $this->companyId, ':wh' => $warehouse_id]);
|
|
$periods_rebuilt = (int)$sth->fetchColumn();
|
|
|
|
$this->saveStatus('ok', 0, $periods_rebuilt);
|
|
|
|
return ['periods_rebuilt' => $periods_rebuilt];
|
|
}
|
|
|
|
public function getDistinctPeriods(int $warehouse_id): array
|
|
{
|
|
$table = 'td_stock_' . $warehouse_id;
|
|
$sth = $this->pdo->prepare(
|
|
"SELECT DISTINCT DATE_FORMAT(`date`, '%Y-%m') AS period
|
|
FROM `{$table}`
|
|
WHERE company_id = :cid AND status = 1
|
|
ORDER BY period"
|
|
);
|
|
$sth->execute([':cid' => $this->companyId]);
|
|
return $sth->fetchAll(PDO::FETCH_COLUMN);
|
|
}
|
|
|
|
public function getStatus(): array
|
|
{
|
|
$keys = [
|
|
'etl_stock_last_ran',
|
|
'etl_stock_last_result',
|
|
'etl_stock_gaps_found',
|
|
'etl_stock_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_stock_last_ran'] ?? null;
|
|
$next_run = $last_ran
|
|
? date('Y-m-d', strtotime($last_ran . ' +1 day')) . sprintf(' %02d:30', $slot_hour)
|
|
: date('Y-m-d') . sprintf(' %02d:30', $slot_hour);
|
|
|
|
return [
|
|
'slot_hour' => $slot_hour,
|
|
'last_ran' => $last_ran,
|
|
'last_result' => $rows['etl_stock_last_result'] ?? null,
|
|
'gaps_found' => (int)($rows['etl_stock_gaps_found'] ?? 0),
|
|
'periods_rebuilt' => (int)($rows['etl_stock_periods_rebuilt'] ?? 0),
|
|
'next_run' => $next_run,
|
|
];
|
|
}
|
|
|
|
// ── Private ──────────────────────────────────────────────────────────────
|
|
|
|
private function discoverStockTables(): array
|
|
{
|
|
$sth = $this->pdo->prepare(
|
|
"SELECT t.table_name
|
|
FROM md_warehouse w
|
|
JOIN information_schema.tables t
|
|
ON t.table_schema = DATABASE()
|
|
AND t.table_name = CONCAT('td_stock_', w.id)
|
|
WHERE w.company_id = :cid"
|
|
);
|
|
$sth->execute([':cid' => $this->companyId]);
|
|
return $sth->fetchAll(PDO::FETCH_COLUMN);
|
|
}
|
|
|
|
private function saveStatus(string $result, int $gaps_found, int $periods_rebuilt): void
|
|
{
|
|
$now = date('Y-m-d H:i:s');
|
|
$data = [
|
|
'etl_stock_last_ran' => $now,
|
|
'etl_stock_last_result' => $result,
|
|
'etl_stock_gaps_found' => (string)$gaps_found,
|
|
'etl_stock_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]);
|
|
}
|
|
}
|
|
}
|