257 lines
13 KiB
PHP
257 lines
13 KiB
PHP
<?php
|
|
|
|
namespace App\Services;
|
|
|
|
use App\Models\StockItem;
|
|
use App\Models\Tenant;
|
|
use App\Models\User;
|
|
use App\Models\Warehouse;
|
|
use Carbon\CarbonImmutable;
|
|
use Carbon\Exceptions\InvalidFormatException;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use Throwable;
|
|
|
|
/** One durable, coalescing work item per tenant/article/warehouse. No leased human identity. */
|
|
class ReplenishmentQueue
|
|
{
|
|
/** Read-only shared status for the workdesk and pilot diagnostics.
|
|
* @return array<string,mixed>
|
|
*/
|
|
public function status(User $actor): array
|
|
{
|
|
abort_unless(auth()->id() === $actor->id, 403);
|
|
$observation = $this->observation();
|
|
|
|
return ['recoveryReady' => $this->ready(), 'scanMinutes' => config('replenishment.scan_minutes'),
|
|
'schedulerStatus' => $observation['schedulerStatus'], 'schedulerRecent' => $observation['schedulerRecent'],
|
|
'work' => $this->ready() ? $this->workStatus($actor->tenant_id) : null];
|
|
}
|
|
|
|
/** Workdesk sources, with a bound observation required for technical monitoring.
|
|
* @return array<string,mixed>
|
|
*/
|
|
public function health(): array
|
|
{
|
|
abort_unless(app()->runningInConsole() && ! auth()->check() && ! request()->hasSession(), 403);
|
|
|
|
return DB::transaction(function () {
|
|
if (DB::getDriverName() === 'pgsql' && DB::transactionLevel() === 1) {
|
|
DB::statement('SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY');
|
|
}
|
|
$ready = $this->ready();
|
|
$observation = $this->observation();
|
|
$work = $ready ? $this->workStatus(null, true) : null;
|
|
$reasons = [];
|
|
if (! $ready) {
|
|
$reasons[] = 'not_installed';
|
|
}
|
|
if (! $observation['schedulerRecent']) {
|
|
$reasons[] = 'not_recent';
|
|
}
|
|
if (($observation['schedulerStatus']['result'] ?? null) !== 'passed') {
|
|
$reasons[] = 'last_run_not_successful';
|
|
}
|
|
if (! $observation['schedulerBound']) {
|
|
$reasons[] = 'unbound_observation';
|
|
}
|
|
if ((int) ($work->pending ?? 0) > 0) {
|
|
$reasons[] = 'pending_signals';
|
|
}
|
|
if ((int) ($work->failed ?? 0) > 0) {
|
|
$reasons[] = 'failed_work';
|
|
}
|
|
|
|
return ['checked_at' => now()->toIso8601String(), 'ok' => $reasons === [], 'reasons' => $reasons,
|
|
'recovery_ready' => $ready, 'scheduler_recent' => $observation['schedulerRecent'],
|
|
'last_result' => $observation['schedulerStatus']['result'] ?? null,
|
|
'last_finished_at' => $observation['schedulerStatus']['finished_at'] ?? null,
|
|
'work' => $work === null ? null : ['total' => (int) $work->total, 'pending' => (int) ($work->pending ?? 0), 'failed' => (int) ($work->failed ?? 0)]];
|
|
});
|
|
}
|
|
|
|
private function workStatus(?int $tenant, bool $global = false): \stdClass
|
|
{
|
|
return DB::table('replenishment_work')->when(! $global, fn ($query) => $query->where('tenant_id', $tenant))
|
|
->selectRaw('COUNT(*) as total, SUM(CASE WHEN generation > processed_generation THEN 1 ELSE 0 END) as pending, SUM(CASE WHEN failures > 0 THEN 1 ELSE 0 END) as failed, MAX(last_run_at) as last_run_at')->first() ?? throw new \LogicException('Estado técnico indisponível.');
|
|
}
|
|
|
|
/** @return array{schedulerStatus:array<string,mixed>|null,schedulerRecent:bool,schedulerBound:bool} */
|
|
private function observation(): array
|
|
{
|
|
$scheduler = null;
|
|
$recent = false;
|
|
$bound = false;
|
|
$path = storage_path('framework/replenishment-command-status.json');
|
|
if (! is_file($path) && app()->environment('payable-testing')) {
|
|
$path = storage_path('framework/replenishment-status.json');
|
|
}
|
|
if (is_file($path)) {
|
|
$text = file_get_contents($path);
|
|
$data = $text === false ? null : json_decode($text, true);
|
|
if (is_array($data) && is_string($data['finished_at'] ?? null) && in_array($data['result'] ?? null, ['passed', 'retry_pending'], true)) {
|
|
if (array_key_exists('installation', $data)) {
|
|
if (! is_string($data['installation']) || ! hash_equals(app(ReplenishmentHeartbeat::class)->fingerprint(), $data['installation'])) {
|
|
return ['schedulerStatus' => null, 'schedulerRecent' => false, 'schedulerBound' => false];
|
|
}
|
|
$bound = true;
|
|
}
|
|
try {
|
|
$finished = CarbonImmutable::parse($data['finished_at']);
|
|
$recent = $finished->greaterThanOrEqualTo(now()->subMinutes(3)) && $finished->lessThanOrEqualTo(now()->addMinute());
|
|
$scheduler = array_intersect_key($data, array_flip(['started_at', 'finished_at', 'pid', 'scope', 'result', 'exit_code']));
|
|
} catch (InvalidFormatException) {
|
|
} // Optional telemetry must not prevent operational consultation.
|
|
}
|
|
}
|
|
|
|
return ['schedulerStatus' => $scheduler, 'schedulerRecent' => $recent, 'schedulerBound' => $bound];
|
|
}
|
|
|
|
public function ready(): bool
|
|
{
|
|
return config('work_context.composition') === 'sgc' && Schema::hasTable('replenishment_work');
|
|
}
|
|
|
|
public function enqueue(User $actor, int $item, int $warehouse): int
|
|
{
|
|
return DB::transaction(function () use ($actor, $item, $warehouse) {
|
|
app(ConflictGuard::class)->lockCompany($actor);
|
|
$actor = app(CurrentActor::class)->lock($actor);
|
|
StockItem::query()->whereKey($item)->firstOrFail();
|
|
Warehouse::query()->whereKey($warehouse)->firstOrFail();
|
|
|
|
return $this->touch($actor->tenant_id, $item, $warehouse, $actor->id, true);
|
|
});
|
|
}
|
|
|
|
private function touch(int $tenant, int $item, int $warehouse, int $responsible, bool $signal): int
|
|
{
|
|
$row = DB::table('replenishment_work')->where('tenant_id', $tenant)->where('stock_item_id', $item)->where('warehouse_id', $warehouse)->first();
|
|
if ($row === null) {
|
|
return DB::table('replenishment_work')->insertGetId(['tenant_id' => $tenant, 'stock_item_id' => $item, 'warehouse_id' => $warehouse, 'responsible_id' => $responsible, 'generation' => 1, 'processed_generation' => 0, 'due_at' => now(), 'created_at' => now(), 'updated_at' => now()]);
|
|
}
|
|
if ($signal) {
|
|
DB::table('replenishment_work')->where('id', $row->id)->update(['generation' => $row->generation + 1, 'responsible_id' => $responsible, 'due_at' => now(), 'updated_at' => now()]);
|
|
}
|
|
|
|
return $row->id;
|
|
}
|
|
|
|
/** CLI discovery recovers pre-installation interruptions and adds approved policies without stock events. */
|
|
public function discover(): void
|
|
{
|
|
abort_unless(app()->runningInConsole() && ! auth()->check() && ! request()->hasSession(), 403);
|
|
if (! $this->ready()) {
|
|
return;
|
|
}
|
|
$latest = DB::table('supply_records')->selectRaw('MAX(id)')->whereIn('kind', ['policy', 'auto_rule', 'auto_need'])->groupBy('tenant_id', 'kind', 'record_key');
|
|
DB::table('supply_records')->whereIn('id', $latest)->whereNotNull('stock_item_id')->whereNotNull('warehouse_id')->orderBy('id')->chunkById(100, function ($rows) {
|
|
foreach ($rows as $row) {
|
|
$data = json_decode($row->data, true, 64, JSON_THROW_ON_ERROR);
|
|
if ($row->kind === 'policy' && ($row->state !== 'approved' || ($data['purpose'] ?? null) !== 'internal')) {
|
|
continue;
|
|
}
|
|
if ($row->kind === 'auto_rule' && $row->state !== 'approved') {
|
|
continue;
|
|
}
|
|
$this->discoverPair($row->tenant_id, $row->stock_item_id, $row->warehouse_id, $row->created_by);
|
|
}
|
|
});
|
|
DB::table('warehouse_requests')->where('work_context_key', 'procurement.internal')->orderBy('id')->chunkById(100, function ($rows) {
|
|
foreach ($rows as $row) {
|
|
foreach (json_decode($row->snapshot, true, 64, JSON_THROW_ON_ERROR)['lines'] as $line) {
|
|
$this->discoverPair($row->tenant_id, $line['item']['id'], $row->warehouse_id, $row->created_by);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
private function enabled(int $tenant): bool
|
|
{
|
|
$company = Tenant::query()->findOrFail($tenant);
|
|
|
|
return $company->work_context_policy === null || (in_array('procurement.internal', $company->work_context_policy['enabled'] ?? [], true) && in_array('procurement.internal', config('work_context.entitlements.'.$tenant, []), true));
|
|
}
|
|
|
|
private function discoverPair(int $tenant, int $item, int $warehouse, int $responsible): void
|
|
{
|
|
DB::transaction(function () use ($tenant, $item, $warehouse, $responsible) {
|
|
Tenant::query()->whereKey($tenant)->lockForUpdate()->firstOrFail();
|
|
if (! $this->enabled($tenant)) {
|
|
return;
|
|
}
|
|
if (! DB::table('users')->where('tenant_id', $tenant)->where('id', $responsible)->exists() || ! DB::table('stock_items')->where('tenant_id', $tenant)->where('id', $item)->exists() || ! DB::table('warehouses')->where('tenant_id', $tenant)->where('id', $warehouse)->exists()) {
|
|
return;
|
|
}
|
|
$this->touch($tenant, $item, $warehouse, $responsible, false);
|
|
});
|
|
}
|
|
|
|
public function interactive(User $actor, int $id): bool
|
|
{
|
|
return $this->process($id, $actor);
|
|
}
|
|
|
|
public function scheduled(int $id): bool
|
|
{
|
|
return app(ReplenishmentRun::class)->execute($id, fn () => $this->process($id, null));
|
|
}
|
|
|
|
private function process(int $id, ?User $actor): bool
|
|
{
|
|
$work = DB::table('replenishment_work')->where('id', $id)->firstOrFail();
|
|
if ($actor !== null) {
|
|
abort_unless($actor->tenant_id === $work->tenant_id && auth()->id() === $actor->id, 403);
|
|
}
|
|
$generation = null;
|
|
try {
|
|
DB::transaction(function () use ($id, $actor, &$generation) {
|
|
$first = DB::table('replenishment_work')->where('id', $id)->firstOrFail();
|
|
Tenant::query()->whereKey($first->tenant_id)->lockForUpdate()->firstOrFail();
|
|
$row = DB::table('replenishment_work')->where('id', $id)->lockForUpdate()->firstOrFail();
|
|
if (CarbonImmutable::parse($row->due_at)->isFuture() && $row->generation === $row->processed_generation) {
|
|
return;
|
|
}
|
|
$generation = $row->generation;
|
|
if ($actor === null) {
|
|
$actor = User::query()->where('tenant_id', $row->tenant_id)->whereKey($row->responsible_id)->firstOrFail();
|
|
}
|
|
if ($this->enabled($row->tenant_id)) {
|
|
app(AutomaticPurchases::class)->evaluate($actor, $row->stock_item_id, $row->warehouse_id);
|
|
}
|
|
$latest = app(SupplyPlanning::class)->latest('auto_need')->where('stock_item_id', $row->stock_item_id)->where('warehouse_id', $row->warehouse_id)->first();
|
|
DB::table('replenishment_work')->where('id', $id)->update(['processed_generation' => $generation, 'attempts' => $row->attempts + 1, 'failures' => 0, 'last_outcome' => $this->enabled($row->tenant_id) ? ($latest->state ?? 'no_need') : 'context_disabled', 'last_error' => null, 'last_run_at' => now(), 'due_at' => now()->addMinutes(config('replenishment.scan_minutes')), 'updated_at' => now()]);
|
|
});
|
|
|
|
return true;
|
|
} catch (Throwable $error) {
|
|
report($error);
|
|
// The failed transaction rolled back both draft and checkpoint. Keep the durable signal.
|
|
DB::transaction(function () use ($id, $work, $generation) {
|
|
Tenant::query()->whereKey($work->tenant_id)->lockForUpdate()->firstOrFail();
|
|
$row = DB::table('replenishment_work')->where('id', $id)->lockForUpdate()->firstOrFail();
|
|
$fields = ['attempts' => $row->attempts + 1, 'failures' => $row->failures + 1, 'last_outcome' => 'error', 'last_error' => 'Avaliação interrompida. Recuperação automática pendente; consulte o registo técnico.', 'updated_at' => now()];
|
|
if ($generation !== null && $generation === $row->generation) {
|
|
$fields['due_at'] = now()->addSeconds(min(3600, 60 * (2 ** min($row->failures, 6))));
|
|
}
|
|
DB::table('replenishment_work')->where('id', $id)->update($fields);
|
|
});
|
|
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/** @return array{processed:int,failed:int} */
|
|
public function drain(int $limit): array
|
|
{
|
|
abort_unless(app()->runningInConsole() && ! auth()->check() && ! request()->hasSession(), 403);
|
|
$result = ['processed' => 0, 'failed' => 0];
|
|
foreach (DB::table('replenishment_work')->where('due_at', '<=', now())->orderBy('due_at')->orderBy('id')->limit($limit)->pluck('id') as $id) {
|
|
$result[$this->scheduled($id) ? 'processed' : 'failed']++;
|
|
}
|
|
|
|
return $result;
|
|
}
|
|
}
|