diff --git a/core/app/Channels/ChannelRegistry.php b/core/app/Channels/ChannelRegistry.php index 1b9b5cb..e9e39b8 100644 --- a/core/app/Channels/ChannelRegistry.php +++ b/core/app/Channels/ChannelRegistry.php @@ -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) {} diff --git a/core/app/Console/Commands/PurgeCommand.php b/core/app/Console/Commands/PurgeCommand.php new file mode 100644 index 0000000..bba8329 --- /dev/null +++ b/core/app/Console/Commands/PurgeCommand.php @@ -0,0 +1,26 @@ + $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; + } +} diff --git a/core/app/Console/Commands/TickCommand.php b/core/app/Console/Commands/TickCommand.php new file mode 100644 index 0000000..b839d0f --- /dev/null +++ b/core/app/Console/Commands/TickCommand.php @@ -0,0 +1,20 @@ +run(); + + return self::SUCCESS; + } +} diff --git a/core/app/Delivery/DeliverJob.php b/core/app/Delivery/DeliverJob.php new file mode 100644 index 0000000..7d307bb --- /dev/null +++ b/core/app/Delivery/DeliverJob.php @@ -0,0 +1,58 @@ + */ + 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()]); + } +} diff --git a/core/app/Delivery/DeliveryFailed.php b/core/app/Delivery/DeliveryFailed.php new file mode 100644 index 0000000..5c06c73 --- /dev/null +++ b/core/app/Delivery/DeliveryFailed.php @@ -0,0 +1,7 @@ +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(); + } + }); + } +} diff --git a/core/app/Providers/AppServiceProvider.php b/core/app/Providers/AppServiceProvider.php index ac4a4ad..32d9cce 100644 --- a/core/app/Providers/AppServiceProvider.php +++ b/core/app/Providers/AppServiceProvider.php @@ -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'), + )); } /** diff --git a/core/routes/console.php b/core/routes/console.php index 3c9adf1..b651cf7 100644 --- a/core/routes/console.php +++ b/core/routes/console.php @@ -1,8 +1,6 @@ comment(Inspiring::quote()); -})->purpose('Display an inspiring quote'); +Schedule::command('hado:tick')->everyMinute()->withoutOverlapping()->runInBackground(); +Schedule::command('hado:purge')->dailyAt('04:00'); diff --git a/core/tests/Feature/DeliverJobTest.php b/core/tests/Feature/DeliverJobTest.php new file mode 100644 index 0000000..19fcc38 --- /dev/null +++ b/core/tests/Feature/DeliverJobTest.php @@ -0,0 +1,93 @@ + '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); + } +} diff --git a/core/tests/Feature/PurgeCommandTest.php b/core/tests/Feature/PurgeCommandTest.php new file mode 100644 index 0000000..f2133e4 --- /dev/null +++ b/core/tests/Feature/PurgeCommandTest.php @@ -0,0 +1,42 @@ + 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(); + } +} diff --git a/core/tests/Feature/TickTest.php b/core/tests/Feature/TickTest.php new file mode 100644 index 0000000..72c8854 --- /dev/null +++ b/core/tests/Feature/TickTest.php @@ -0,0 +1,190 @@ +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)); + } +} diff --git a/core/tests/Support/FakeChannel.php b/core/tests/Support/FakeChannel.php new file mode 100644 index 0000000..5e83462 --- /dev/null +++ b/core/tests/Support/FakeChannel.php @@ -0,0 +1,32 @@ + 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 {} +} diff --git a/core/tests/Support/FakeRegistry.php b/core/tests/Support/FakeRegistry.php new file mode 100644 index 0000000..a8207dc --- /dev/null +++ b/core/tests/Support/FakeRegistry.php @@ -0,0 +1,22 @@ + $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]; + } +}