7d1df7828a
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BCrwHHnGCB5XH968Nxokqw
119 lines
3.8 KiB
PHP
119 lines
3.8 KiB
PHP
<?php
|
||
|
||
namespace App\Ingest;
|
||
|
||
use App\Enums\AfterDue;
|
||
use App\Enums\ChannelType;
|
||
use App\Enums\EventState;
|
||
use App\Models\Event;
|
||
use App\Models\Source;
|
||
use App\Models\User;
|
||
use App\Scheduling\StateResolver;
|
||
use Carbon\CarbonImmutable;
|
||
use Illuminate\Support\Facades\DB;
|
||
|
||
final class EventUpserter
|
||
{
|
||
public function __construct(private readonly StateResolver $states) {}
|
||
|
||
/**
|
||
* @param list<array{user:string,source_ref:string,topic?:string,due_at:string,fire_at:string,after_due:string,payload:array}> $items
|
||
* @return list<array{source_ref:string,topic:string,id:int,state:string}>
|
||
*/
|
||
public function upsert(Source $source, array $items): array
|
||
{
|
||
return DB::transaction(function () use ($source, $items) {
|
||
$out = [];
|
||
foreach ($items as $item) {
|
||
$event = $this->upsertOne($source, $item);
|
||
$out[] = [
|
||
'source_ref' => $event->source_ref,
|
||
'topic' => $event->topic,
|
||
'id' => $event->id,
|
||
'state' => $event->state->value,
|
||
];
|
||
}
|
||
|
||
return $out;
|
||
});
|
||
}
|
||
|
||
public function withdraw(Source $source, string $sourceRef, string $topic): bool
|
||
{
|
||
$event = $this->find($source, $sourceRef, $topic);
|
||
if ($event === null) {
|
||
return false;
|
||
}
|
||
$event->update(['state' => EventState::Withdrawn]);
|
||
|
||
return true;
|
||
}
|
||
|
||
public function find(Source $source, string $sourceRef, string $topic): ?Event
|
||
{
|
||
return Event::where('source_id', $source->id)
|
||
->where('source_ref', $sourceRef)
|
||
->where('topic', $topic)
|
||
->first();
|
||
}
|
||
|
||
/** Пользователь по логину; создаётся с поясом по умолчанию и web-каналом. */
|
||
public function ensureUser(string $login): User
|
||
{
|
||
$user = User::firstOrCreate(
|
||
['login' => $login],
|
||
['tz' => config('hado.default_tz'), 'quiet_start' => '22:00:00', 'quiet_end' => '09:00:00'],
|
||
);
|
||
$user->channels()->firstOrCreate(['type' => ChannelType::Web->value], ['config' => [], 'enabled' => true]);
|
||
|
||
return $user;
|
||
}
|
||
|
||
private function upsertOne(Source $source, array $item): Event
|
||
{
|
||
$user = $this->ensureUser($item['user']);
|
||
$due = DueAtParser::parse($item['due_at']);
|
||
$topic = $item['topic'] ?? '';
|
||
$now = $user->now();
|
||
|
||
$attributes = [
|
||
'user_id' => $user->id,
|
||
'fire_on' => $item['fire_at'],
|
||
'after_due' => AfterDue::from($item['after_due']),
|
||
'payload' => $item['payload'],
|
||
] + $due->columns();
|
||
|
||
$event = $this->find($source, $item['source_ref'], $topic);
|
||
|
||
if ($event === null) {
|
||
$event = new Event([
|
||
'source_id' => $source->id,
|
||
'source_ref' => $item['source_ref'],
|
||
'topic' => $topic,
|
||
'state' => EventState::Scheduled,
|
||
'quiet_until' => $now->subMinute(),
|
||
] + $attributes);
|
||
$event->state = $this->states->resolve($event, $user, $now);
|
||
$event->save();
|
||
|
||
return $event;
|
||
}
|
||
|
||
$dueChanged = ! $due->equals($event);
|
||
$event->fill($attributes);
|
||
|
||
if ($dueChanged) {
|
||
// Перенос: новая дата — новый заход, даже из терминального состояния.
|
||
$event->done_at = null;
|
||
$event->quiet_until = $now->subMinute();
|
||
$event->state = $this->states->resolve($event, $user, $now);
|
||
} elseif (! $event->isTerminal()) {
|
||
$event->state = $this->states->resolve($event, $user, $now);
|
||
}
|
||
|
||
$event->save();
|
||
|
||
return $event;
|
||
}
|
||
}
|