hado/core/app/Ingest/EventUpserter.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;
        });
    }

    /** Снять событие: withdrawn — терминальное, ядро больше не напоминает. */
    public function withdraw(Event $event): void
    {
        $event->update(['state' => EventState::Withdrawn]);
    }

    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;
    }
}