stock aggregate table
This commit is contained in:
Thanakorn S
@@ -0,0 +1,227 @@
|
||||
<?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 table_name FROM information_schema.tables
|
||||
WHERE table_schema = DATABASE() AND table_name LIKE 'td_stock_%'"
|
||||
);
|
||||
$sth->execute();
|
||||
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]);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user