77 lines
2.1 KiB
PHP
77 lines
2.1 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace App\Service;
|
||
|
||
use App\Model\AggregateRefreshQueue;
|
||
use App\Utils\Log;
|
||
use Carbon\Carbon;
|
||
use Throwable;
|
||
|
||
/**
|
||
* 订单聚合刷新任务
|
||
*
|
||
* 消费 aggregate_refresh_queue 中 created_at < now() - 1 hour 的项，
|
||
* 逐条调用 AggregateRefresherInterface 刷新对应日期的连续聚合，
|
||
* 成功后从队列中删除；失败保留队列项由下次任务重试。
|
||
*
|
||
* 由 OrderAggregatesRefreshCommand（CLI 入口）和 Crontab（定时入口）共同调用。
|
||
*/
|
||
class OrderAggregatesRefreshJob
|
||
{
|
||
public function __construct(private AggregateRefresherInterface $refresher)
|
||
{
|
||
}
|
||
|
||
/**
|
||
* @return array{processed: int, failed: int} 处理与失败计数
|
||
*/
|
||
public function run(): array
|
||
{
|
||
$items = AggregateRefreshQueue::query()
|
||
->where('created_at', '<', Carbon::now()->subHour())
|
||
->orderBy('refresh_date')
|
||
->orderBy('aggregate_view')
|
||
->get();
|
||
|
||
if ($items->isEmpty()) {
|
||
Log::get()->info('orders:refresh-aggregates queue empty');
|
||
return ['processed' => 0, 'failed' => 0];
|
||
}
|
||
|
||
$processed = 0;
|
||
$failed = 0;
|
||
|
||
foreach ($items as $item) {
|
||
$view = $item->aggregate_view;
|
||
$date = $item->refresh_date;
|
||
|
||
try {
|
||
$this->refresher->refresh($view, $date);
|
||
|
||
AggregateRefreshQueue::query()
|
||
->where('refresh_date', $date)
|
||
->where('aggregate_view', $view)
|
||
->delete();
|
||
|
||
$processed++;
|
||
} catch (Throwable $e) {
|
||
Log::get()->error('orders:refresh-aggregates failed', [
|
||
'view' => $view,
|
||
'date' => $date,
|
||
'error' => $e->getMessage(),
|
||
]);
|
||
$failed++;
|
||
}
|
||
}
|
||
|
||
Log::get()->info('orders:refresh-aggregates done', [
|
||
'processed' => $processed,
|
||
'failed' => $failed,
|
||
]);
|
||
|
||
return ['processed' => $processed, 'failed' => $failed];
|
||
}
|
||
}
|