feat(dispatcher): WP-Cron tick with backoff and retention cleanup

2026-04-17 22:45:38 +05:00
parent 786a306f29
commit 4ce2fac42d
1 changed files with 72 additions and 0 deletions
@@ -0,0 +1,72 @@
+<?php
+declare(strict_types=1);
+
+namespace Cf7stg;
+
+final class Dispatcher
+{
+    private const BACKOFF_SECONDS = [60, 300, 1800, 3600, 3600];
+    private const MAX_ATTEMPTS = 5;
+
+    public static function register_hooks(): void
+    {
+        add_filter('cron_schedules', [__CLASS__, 'register_schedule']);
+        add_action('cf7stg_dispatch', [__CLASS__, 'tick']);
+        add_action('cf7stg_cleanup', [__CLASS__, 'cleanup']);
+    }
+
+    public static function register_schedule(array $schedules): array
+    {
+        $schedules['cf7stg_every_minute'] = [
+            'interval' => 60,
+            'display'  => 'Раз в минуту (CF7 Spam → TG)',
+        ];
+        return $schedules;
+    }
+
+    public static function tick(): void
+    {
+        if (!Settings::is_configured()) {
+            return;
+        }
+        $items = Queue::fetch_pending(5);
+        $chat_id = Settings::get_chat_id();
+        foreach ($items as $item) {
+            $payload = json_decode((string)$item->payload_json, true);
+            if (!is_array($payload) || empty($payload['text'])) {
+                Queue::mark_failed((int)$item->id, 'Invalid payload JSON', (int)$item->attempts + 1);
+                continue;
+            }
+            try {
+                TelegramClient::send_message(
+                    $chat_id,
+                    (string)$payload['text'],
+                    isset($payload['reply_markup']) && is_array($payload['reply_markup']) ? $payload['reply_markup'] : null
+                );
+                Queue::mark_sent((int)$item->id);
+            } catch (TelegramException $e) {
+                self::handle_failure($item, $e);
+            }
+            usleep(100000); // 100ms between sends
+        }
+    }
+
+    private static function handle_failure(object $item, TelegramException $e): void
+    {
+        $attempts = (int)$item->attempts + 1;
+        if ($e->is_permanent() || $attempts >= self::MAX_ATTEMPTS) {
+            Queue::mark_failed((int)$item->id, $e->getMessage(), $attempts);
+            return;
+        }
+        $backoff = self::BACKOFF_SECONDS[$attempts - 1] ?? end(self::BACKOFF_SECONDS);
+        if ($e->retry_after() > $backoff) {
+            $backoff = $e->retry_after();
+        }
+        Queue::reschedule((int)$item->id, $attempts, $e->getMessage(), $backoff);
+    }
+
+    public static function cleanup(): void
+    {
+        Queue::cleanup_old_sent(30);
+    }
+}