feat: tick planner, delivery job, scheduler and purge

Also: test uses disabled channels instead of deleting them (FK cascade); ChannelRegistry no longer final so tests can extend it (controller ruling).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BCrwHHnGCB5XH968Nxokqw
This commit is contained in:
nikita.hohlov
2026-09-04 07:38:16 -03:00
parent 1b0a559e3c
commit affac4ce1a
13 changed files with 611 additions and 6 deletions

View File

@@ -5,7 +5,7 @@ namespace App\Channels;
use App\Enums\ChannelType;
use Illuminate\Contracts\Container\Container;
final class ChannelRegistry
class ChannelRegistry
{
public function __construct(private readonly Container $container) {}

View File

@@ -0,0 +1,26 @@
<?php
namespace App\Console\Commands;
use App\Enums\EventState;
use App\Models\Event;
use Carbon\CarbonImmutable;
use Illuminate\Console\Command;
class PurgeCommand extends Command
{
protected $signature = 'hado:purge';
protected $description = 'Удаляет терминальные события старше HADO_RETENTION_DAYS';
public function handle(): int
{
$terminal = array_map(fn (EventState $s) => $s->value, array_filter(EventState::cases(), fn (EventState $s) => $s->isTerminal()));
$cutoff = CarbonImmutable::now()->subDays((int) config('hado.retention_days'));
$deleted = Event::whereIn('state', $terminal)->where('updated_at', '<', $cutoff)->delete();
$this->info("Удалено событий: {$deleted}");
return self::SUCCESS;
}
}

View File

@@ -0,0 +1,20 @@
<?php
namespace App\Console\Commands;
use App\Delivery\Tick;
use Illuminate\Console\Command;
class TickCommand extends Command
{
protected $signature = 'hado:tick';
protected $description = 'Один тик: переходы состояний и доставка напоминаний в открытые окна';
public function handle(Tick $tick): int
{
$tick->run();
return self::SUCCESS;
}
}

View File

@@ -0,0 +1,58 @@
<?php
namespace App\Delivery;
use App\Channels\ChannelRegistry;
use App\Enums\DeliveryResult;
use App\Models\Delivery;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Throwable;
final class DeliverJob implements ShouldQueue
{
use Queueable;
public int $tries = 3;
public function __construct(public readonly int $deliveryId) {}
/** @return list<int> */
public function backoff(): array
{
return [30, 120, 600];
}
public function handle(ChannelRegistry $registry): void
{
$delivery = Delivery::with(['event.user', 'event.source', 'channel'])->findOrFail($this->deliveryId);
if ($delivery->result !== DeliveryResult::Pending) {
return;
}
if ($delivery->event->isTerminal()) {
$delivery->update(['result' => DeliveryResult::Failed, 'error' => 'skipped: событие уже закрыто']);
return;
}
$outcome = $registry->for($delivery->channel->type)
->deliver($delivery->event, $delivery->event->user, $delivery->channel->config, $delivery);
if ($outcome->ok) {
$delivery->update(['result' => DeliveryResult::Ok, 'sent_at' => now(), 'error' => null, 'meta' => $outcome->meta]);
return;
}
$delivery->update(['error' => $outcome->error]);
throw new DeliveryFailed($outcome->error ?? 'delivery failed');
}
public function failed(Throwable $e): void
{
Delivery::whereKey($this->deliveryId)
->where('result', DeliveryResult::Pending->value)
->update(['result' => DeliveryResult::Failed, 'error' => $e->getMessage()]);
}
}

View File

@@ -0,0 +1,7 @@
<?php
namespace App\Delivery;
use RuntimeException;
final class DeliveryFailed extends RuntimeException {}

108
core/app/Delivery/Tick.php Normal file
View File

@@ -0,0 +1,108 @@
<?php
namespace App\Delivery;
use App\Channels\ChannelRegistry;
use App\Enums\DeliveryResult;
use App\Enums\EventState;
use App\Enums\Presence;
use App\Models\Channel;
use App\Models\Delivery;
use App\Models\Event;
use App\Models\User;
use App\Scheduling\DueResolver;
use App\Scheduling\StateResolver;
use App\Scheduling\Window;
use App\Scheduling\WindowResolver;
use Illuminate\Contracts\Cache\Repository as Cache;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Str;
/**
* Один тик планировщика (раз в минуту): переходы состояний по датам и решение,
* доставлять ли напоминание в открытое окно и куда.
*/
final class Tick
{
public function __construct(
private readonly DueResolver $due,
private readonly StateResolver $states,
private readonly WindowResolver $windows,
private readonly ChannelRegistry $registry,
private readonly Cache $cache,
private readonly int $presenceTtl,
) {}
public function run(): void
{
Event::query()
->whereIn('state', EventState::nonTerminalValues())
->with('user')
->chunkById(200, fn (Collection $events) => $events->each(fn (Event $e) => $this->process($e)));
}
private function process(Event $event): void
{
$user = $event->user;
$now = $user->now();
$state = $this->states->resolve($event, $user, $now);
if ($state !== $event->state) {
$event->state = $state;
$event->save();
}
if (! $state->isActive() || $now->lt($event->quiet_until)) {
return;
}
$window = $this->windows->current($state, $user, $this->due->resolve($event, $user), $now);
if ($window === null) {
return;
}
if (Delivery::where('event_id', $event->id)->where('window_start', $window->start->utc())->exists()) {
return;
}
$channels = $user->channels()->where('enabled', true)->get();
if ($channels->isEmpty()) {
return;
}
$present = $channels->filter(fn (Channel $c) => $this->presence($c, $user) === Presence::Present);
if ($present->isNotEmpty()) {
$this->dispatch($event, $present, $window);
return;
}
if ($window->fallbackAllowed && $now->gte($window->fallbackAt)) {
$this->dispatch($event, $channels, $window);
}
}
private function presence(Channel $channel, User $user): Presence
{
return $this->cache->remember(
"presence:{$channel->id}",
$this->presenceTtl,
fn () => $this->registry->for($channel->type)->presence($user, $channel->config),
);
}
/** Строка pending — в транзакции ДО постановки job-а: следующий тик её уже видит. */
private function dispatch(Event $event, Collection $channels, Window $window): void
{
DB::transaction(function () use ($event, $channels, $window) {
foreach ($channels as $channel) {
$delivery = Delivery::create([
'event_id' => $event->id,
'channel_id' => $channel->id,
'window_start' => $window->start->utc(),
'action_token' => Str::random(48),
'result' => DeliveryResult::Pending,
]);
DeliverJob::dispatch($delivery->id)->afterCommit();
}
});
}
}

View File

@@ -22,6 +22,15 @@ class AppServiceProvider extends ServiceProvider
config('hado.telegram.token'),
config('hado.telegram.seen_ttl'),
));
$this->app->bind(\App\Delivery\Tick::class, fn ($app) => new \App\Delivery\Tick(
$app->make(\App\Scheduling\DueResolver::class),
$app->make(\App\Scheduling\StateResolver::class),
$app->make(\App\Scheduling\WindowResolver::class),
$app->make(\App\Channels\ChannelRegistry::class),
$app->make('cache.store'),
config('hado.presence_ttl'),
));
}
/**

View File

@@ -1,8 +1,6 @@
<?php
use Illuminate\Foundation\Inspiring;
use Illuminate\Support\Facades\Artisan;
use Illuminate\Support\Facades\Schedule;
Artisan::command('inspire', function () {
$this->comment(Inspiring::quote());
})->purpose('Display an inspiring quote');
Schedule::command('hado:tick')->everyMinute()->withoutOverlapping()->runInBackground();
Schedule::command('hado:purge')->dailyAt('04:00');

View File

@@ -0,0 +1,93 @@
<?php
namespace Tests\Feature;
use App\Channels\ChannelRegistry;
use App\Delivery\DeliverJob;
use App\Delivery\DeliveryFailed;
use App\Enums\AfterDue;
use App\Enums\ChannelType;
use App\Enums\DeliveryResult;
use App\Enums\DueMode;
use App\Enums\EventState;
use App\Models\Channel;
use App\Models\Delivery;
use App\Models\Event;
use App\Models\Source;
use App\Models\User;
use Carbon\CarbonImmutable;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Tests\Support\FakeChannel;
use Tests\Support\FakeRegistry;
use Tests\TestCase;
class DeliverJobTest extends TestCase
{
use RefreshDatabase;
private Delivery $delivery;
private FakeChannel $driver;
protected function setUp(): void
{
parent::setUp();
$source = Source::create(['name' => 'docs', 'token_hash' => hash('sha256', 'x')]);
$user = User::create(['login' => 'n', 'tz' => 'UTC', 'quiet_start' => '22:00:00', 'quiet_end' => '09:00:00']);
$channel = Channel::create(['user_id' => $user->id, 'type' => ChannelType::Webhook, 'config' => ['deliver_url' => 'u', 'presence_url' => 'p'], 'enabled' => true]);
$event = Event::create([
'source_id' => $source->id, 'user_id' => $user->id, 'source_ref' => 'r', 'topic' => '',
'due_mode' => DueMode::Local, 'due_date' => '2026-09-30', 'due_time' => '24:00:00', 'fire_on' => '2026-08-31',
'after_due' => AfterDue::Keep, 'payload' => ['title' => 'T'], 'state' => EventState::Preparing,
'quiet_until' => CarbonImmutable::parse('2000-01-01Z'),
]);
$this->delivery = Delivery::create([
'event_id' => $event->id, 'channel_id' => $channel->id, 'window_start' => CarbonImmutable::parse('2026-09-05T09:00:00Z'),
'action_token' => 'tok', 'result' => DeliveryResult::Pending,
]);
$this->driver = new FakeChannel;
$this->app->instance(ChannelRegistry::class, new FakeRegistry(['webhook' => $this->driver]));
}
public function test_success_marks_ok_with_meta(): void
{
(new DeliverJob($this->delivery->id))->handle(app(ChannelRegistry::class));
$d = $this->delivery->fresh();
$this->assertSame(DeliveryResult::Ok, $d->result);
$this->assertNotNull($d->sent_at);
$this->assertSame(['n' => 1], $d->meta);
}
public function test_failure_records_error_and_throws_for_retry(): void
{
$this->driver->succeed = false;
$this->expectException(DeliveryFailed::class);
try {
(new DeliverJob($this->delivery->id))->handle(app(ChannelRegistry::class));
} finally {
$d = $this->delivery->fresh();
$this->assertSame(DeliveryResult::Pending, $d->result, 'ещё pending — будут ретраи');
$this->assertSame('boom', $d->error);
}
}
public function test_failed_hook_marks_failed(): void
{
(new DeliverJob($this->delivery->id))->failed(new DeliveryFailed('boom'));
$this->assertSame(DeliveryResult::Failed, $this->delivery->fresh()->result);
}
public function test_skips_when_already_done_or_event_terminal(): void
{
$this->delivery->update(['result' => DeliveryResult::Ok]);
(new DeliverJob($this->delivery->id))->handle(app(ChannelRegistry::class));
$this->assertSame([], $this->driver->delivered);
$this->delivery->update(['result' => DeliveryResult::Pending]);
$this->delivery->event->update(['state' => EventState::Done]);
(new DeliverJob($this->delivery->id))->handle(app(ChannelRegistry::class));
$this->assertSame([], $this->driver->delivered);
$this->assertSame(DeliveryResult::Failed, $this->delivery->fresh()->result);
}
}

View File

@@ -0,0 +1,42 @@
<?php
namespace Tests\Feature;
use App\Enums\AfterDue;
use App\Enums\DueMode;
use App\Enums\EventState;
use App\Models\Event;
use App\Models\Source;
use App\Models\User;
use Carbon\CarbonImmutable;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Tests\TestCase;
class PurgeCommandTest extends TestCase
{
use RefreshDatabase;
public function test_deletes_old_terminal_events_only(): void
{
config(['hado.retention_days' => 90]);
$source = Source::create(['name' => 'docs', 'token_hash' => hash('sha256', 'x')]);
$user = User::create(['login' => 'n', 'tz' => 'UTC']);
$make = fn (string $topic, EventState $state, string $updatedAt) => tap(Event::create([
'source_id' => $source->id, 'user_id' => $user->id, 'source_ref' => 'r', 'topic' => $topic,
'due_mode' => DueMode::Local, 'due_date' => '2026-01-01', 'due_time' => '24:00:00', 'fire_on' => '2026-01-01',
'after_due' => AfterDue::Keep, 'payload' => ['title' => 'T'], 'state' => $state,
'quiet_until' => CarbonImmutable::parse('2000-01-01Z'),
]), fn (Event $e) => Event::whereKey($e->id)->update(['updated_at' => $updatedAt]));
CarbonImmutable::setTestNow('2026-09-04T12:00:00Z');
$make('old-done', EventState::Done, '2026-05-01 00:00:00');
$make('old-expired', EventState::Expired, '2026-05-01 00:00:00');
$make('fresh-done', EventState::Done, '2026-08-01 00:00:00');
$make('old-live', EventState::Overdue, '2026-05-01 00:00:00');
$this->artisan('hado:purge')->assertSuccessful();
$this->assertSame(['fresh-done', 'old-live'], Event::orderBy('topic')->pluck('topic')->all());
CarbonImmutable::setTestNow();
}
}

View File

@@ -0,0 +1,190 @@
<?php
namespace Tests\Feature;
use App\Channels\ChannelRegistry;
use App\Delivery\Tick;
use App\Enums\AfterDue;
use App\Enums\ChannelType;
use App\Enums\DeliveryResult;
use App\Enums\DueMode;
use App\Enums\EventState;
use App\Enums\Presence;
use App\Models\Channel;
use App\Models\Delivery;
use App\Models\Event;
use App\Models\Source;
use App\Models\User;
use Carbon\CarbonImmutable;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Illuminate\Support\Facades\Cache;
use Tests\Support\FakeChannel;
use Tests\Support\FakeRegistry;
use Tests\TestCase;
class TickTest extends TestCase
{
use RefreshDatabase;
private User $user;
private Source $source;
private FakeChannel $web;
private FakeChannel $telegram;
private FakeChannel $webhook;
protected function setUp(): void
{
parent::setUp();
$this->source = Source::create(['name' => 'docs', 'token_hash' => hash('sha256', 'x')]);
$this->user = User::create(['login' => 'nikita', 'tz' => 'UTC', 'quiet_start' => '22:00:00', 'quiet_end' => '09:00:00']);
Channel::create(['user_id' => $this->user->id, 'type' => ChannelType::Web, 'config' => [], 'enabled' => true]);
Channel::create(['user_id' => $this->user->id, 'type' => ChannelType::Telegram, 'config' => ['chat_id' => 1], 'enabled' => true]);
Channel::create(['user_id' => $this->user->id, 'type' => ChannelType::Webhook, 'config' => ['deliver_url' => 'u', 'presence_url' => 'p'], 'enabled' => true]);
$this->web = new FakeChannel(Presence::Absent);
$this->telegram = new FakeChannel(Presence::Unknown);
$this->webhook = new FakeChannel(Presence::Absent);
$this->app->instance(ChannelRegistry::class, new FakeRegistry([
'web' => $this->web, 'telegram' => $this->telegram, 'webhook' => $this->webhook,
]));
}
protected function tearDown(): void
{
CarbonImmutable::setTestNow();
parent::tearDown();
}
private function event(string $due, string $fireOn, AfterDue $afterDue = AfterDue::Keep, string $time = '24:00:00'): Event
{
return Event::create([
'source_id' => $this->source->id, 'user_id' => $this->user->id, 'source_ref' => 'r', 'topic' => uniqid(),
'due_mode' => DueMode::Local, 'due_date' => $due, 'due_time' => $time, 'fire_on' => $fireOn,
'after_due' => $afterDue, 'payload' => ['title' => 'T'], 'state' => EventState::Scheduled,
'quiet_until' => CarbonImmutable::parse('2000-01-01Z'),
]);
}
private function tickAt(string $utc): void
{
CarbonImmutable::setTestNow($utc);
Cache::flush();
app(Tick::class)->run();
}
public function test_moves_states_without_delivering_before_windows(): void
{
$e = $this->event('2026-09-30', '2026-08-31');
$this->tickAt('2026-08-31T03:00:00Z');
$this->assertSame(EventState::Preparing, $e->fresh()->state);
$this->assertSame(0, Delivery::count(), 'тихие часы — доставок нет');
$this->tickAt('2026-10-01T00:00:00Z');
$this->assertSame(EventState::Overdue, $e->fresh()->state);
}
public function test_delivers_only_to_present_channels_and_once_per_window(): void
{
$e = $this->event('2026-09-30', '2026-08-31');
$this->webhook->presence = Presence::Present;
$this->tickAt('2026-09-05T10:00:00Z');
$deliveries = Delivery::all();
$this->assertCount(1, $deliveries);
$this->assertSame(ChannelType::Webhook, $deliveries->sole()->channel->type);
$this->assertSame(DeliveryResult::Ok, $deliveries->sole()->result, 'sync-очередь: job уже отработал');
$this->assertSame('2026-09-05T09:00:00+00:00', $deliveries->sole()->window_start->toIso8601String());
$this->tickAt('2026-09-05T15:00:00Z');
$this->assertSame(1, Delivery::count(), 'второй раз в то же окно не шлём');
$this->tickAt('2026-09-06T10:00:00Z');
$this->assertSame(2, Delivery::count(), 'новое окно — новая доставка');
}
public function test_falls_back_to_all_channels_near_window_end(): void
{
$this->event('2026-09-30', '2026-08-31');
$this->tickAt('2026-09-05T21:00:00Z');
$this->assertSame(0, Delivery::count(), 'никто не present, до fallback ещё далеко');
$this->tickAt('2026-09-05T21:31:00Z');
$this->assertSame(3, Delivery::count(), 'fallback во все три канала');
$this->assertSame([1, 1, 1], [count($this->web->delivered), count($this->telegram->delivered), count($this->webhook->delivered)]);
}
public function test_today_fallback_limited_to_listed_hours(): void
{
$this->event('2026-09-30', '2026-09-30');
$this->tickAt('2026-09-30T11:55:00Z');
$this->assertSame(0, Delivery::count(), '11:00 не fallback-час');
$this->tickAt('2026-09-30T14:55:00Z');
$this->assertSame(3, Delivery::count(), '14:00 — fallback-час');
}
public function test_today_hourly_when_present(): void
{
$this->event('2026-09-30', '2026-09-30');
$this->web->presence = Presence::Present;
$this->tickAt('2026-09-30T11:05:00Z');
$this->tickAt('2026-09-30T11:45:00Z');
$this->tickAt('2026-09-30T12:05:00Z');
$this->assertSame(2, Delivery::count());
$this->assertSame(0, count($this->telegram->delivered), 'в telegram не шли: web present');
}
public function test_quiet_until_blocks_delivery(): void
{
$e = $this->event('2026-09-30', '2026-08-31');
$e->update(['quiet_until' => CarbonImmutable::parse('2026-09-15T00:00:00Z')]);
$this->webhook->presence = Presence::Present;
$this->tickAt('2026-09-10T10:00:00Z');
$this->assertSame(0, Delivery::count());
$this->tickAt('2026-09-15T10:00:00Z');
$this->assertSame(1, Delivery::count());
}
public function test_disabled_channel_is_skipped_and_user_without_channels_is_fine(): void
{
Channel::where('type', ChannelType::Webhook->value)->update(['enabled' => false]);
$this->webhook->presence = Presence::Present;
$this->event('2026-09-30', '2026-08-31');
$this->tickAt('2026-09-05T21:31:00Z');
$this->assertSame(2, Delivery::count(), 'fallback только в включённые');
Channel::query()->update(['enabled' => false]);
$this->tickAt('2026-09-06T21:31:00Z');
$this->assertSame(2, Delivery::count());
}
public function test_user_timezone_shifts_windows(): void
{
$this->user->update(['tz' => 'Asia/Shanghai']); // UTC+8
$this->event('2026-09-30', '2026-08-31');
$this->webhook->presence = Presence::Present;
$this->tickAt('2026-09-05T00:30:00Z'); // 08:30 в Шанхае — тихие часы
$this->assertSame(0, Delivery::count());
$this->tickAt('2026-09-05T01:30:00Z'); // 09:30 в Шанхае
$this->assertSame(1, Delivery::count());
}
public function test_presence_is_cached_for_a_minute(): void
{
$this->event('2026-09-30', '2026-08-31');
$this->webhook->presence = Presence::Present;
CarbonImmutable::setTestNow('2026-09-05T10:00:00Z');
app(Tick::class)->run();
$this->webhook->presence = Presence::Absent;
$this->assertSame(Presence::Present, Cache::get('presence:'.Channel::where('type', 'webhook')->sole()->id));
}
}

View File

@@ -0,0 +1,32 @@
<?php
namespace Tests\Support;
use App\Channels\ChannelDriver;
use App\Channels\DeliveryOutcome;
use App\Enums\Presence;
use App\Models\Delivery;
use App\Models\Event;
use App\Models\User;
final class FakeChannel implements ChannelDriver
{
/** @var list<int> event ids delivered */
public array $delivered = [];
public function __construct(public Presence $presence = Presence::Unknown, public bool $succeed = true) {}
public function presence(User $user, array $config): Presence
{
return $this->presence;
}
public function deliver(Event $event, User $user, array $config, Delivery $delivery): DeliveryOutcome
{
$this->delivered[] = $event->id;
return $this->succeed ? DeliveryOutcome::ok(['n' => count($this->delivered)]) : DeliveryOutcome::failed('boom');
}
public function onDone(Event $event, Delivery $delivery): void {}
}

View File

@@ -0,0 +1,22 @@
<?php
namespace Tests\Support;
use App\Channels\ChannelDriver;
use App\Channels\ChannelRegistry;
use App\Enums\ChannelType;
use Illuminate\Container\Container;
final class FakeRegistry extends ChannelRegistry
{
/** @param array<string, ChannelDriver> $drivers keyed by ChannelType value */
public function __construct(private readonly array $drivers)
{
parent::__construct(Container::getInstance());
}
public function for(ChannelType $type): ChannelDriver
{
return $this->drivers[$type->value];
}
}