feat(queue): CRUD, dedup, retention, status counters

2026-04-17 22:41:30 +05:00
parent cc37aa2647
commit 09f2638066
1 changed files with 182 additions and 0 deletions
@@ -0,0 +1,182 @@
+<?php
+declare(strict_types=1);
+
+namespace Cf7stg;
+
+final class Queue
+{
+    public const STATUS_PENDING = 'pending';
+    public const STATUS_SENT    = 'sent';
+    public const STATUS_FAILED  = 'failed';
+
+    public static function table_name(): string
+    {
+        global $wpdb;
+        return $wpdb->prefix . 'cf7stg_queue';
+    }
+
+    public static function enqueue(array $row): int
+    {
+        global $wpdb;
+        $defaults = [
+            'flamingo_post_id' => null,
+            'form_id'          => null,
+            'source_key'       => 'cf7',
+            'payload_json'     => '',
+            'label'            => 'unclear',
+            'is_retro'         => 0,
+            'status'           => self::STATUS_PENDING,
+            'attempts'         => 0,
+            'last_error'       => null,
+            'next_try_at'      => current_time('mysql'),
+            'created_at'       => current_time('mysql'),
+            'sent_at'          => null,
+        ];
+        $data = array_merge($defaults, $row);
+        $wpdb->insert(self::table_name(), $data);
+        return (int)$wpdb->insert_id;
+    }
+
+    public static function exists_for_flamingo(int $post_id, bool $is_retro): bool
+    {
+        global $wpdb;
+        $t = self::table_name();
+        $sql = $wpdb->prepare(
+            "SELECT id FROM {$t} WHERE flamingo_post_id = %d AND is_retro = %d LIMIT 1",
+            $post_id,
+            $is_retro ? 1 : 0
+        );
+        return (bool)$wpdb->get_var($sql);
+    }
+
+    /** @return array<int,object> */
+    public static function fetch_pending(int $limit = 5): array
+    {
+        global $wpdb;
+        $t = self::table_name();
+        $sql = $wpdb->prepare(
+            "SELECT * FROM {$t}
+             WHERE status = %s
+               AND (next_try_at IS NULL OR next_try_at <= %s)
+             ORDER BY id ASC
+             LIMIT %d",
+            self::STATUS_PENDING,
+            current_time('mysql'),
+            $limit
+        );
+        $rows = $wpdb->get_results($sql);
+        return is_array($rows) ? $rows : [];
+    }
+
+    public static function mark_sent(int $id): void
+    {
+        global $wpdb;
+        $wpdb->update(
+            self::table_name(),
+            ['status' => self::STATUS_SENT, 'sent_at' => current_time('mysql'), 'last_error' => null],
+            ['id' => $id]
+        );
+    }
+
+    public static function reschedule(int $id, int $attempts, string $error, int $backoff_seconds): void
+    {
+        global $wpdb;
+        $next = gmdate('Y-m-d H:i:s', time() + $backoff_seconds);
+        $wpdb->update(
+            self::table_name(),
+            [
+                'attempts'    => $attempts,
+                'last_error'  => $error,
+                'next_try_at' => $next,
+            ],
+            ['id' => $id]
+        );
+    }
+
+    public static function mark_failed(int $id, string $error, int $attempts): void
+    {
+        global $wpdb;
+        $wpdb->update(
+            self::table_name(),
+            [
+                'status'     => self::STATUS_FAILED,
+                'last_error' => $error,
+                'attempts'   => $attempts,
+            ],
+            ['id' => $id]
+        );
+    }
+
+    public static function count_by_status(string $status): int
+    {
+        global $wpdb;
+        $t = self::table_name();
+        return (int)$wpdb->get_var($wpdb->prepare(
+            "SELECT COUNT(*) FROM {$t} WHERE status = %s", $status
+        ));
+    }
+
+    public static function count_failed(): int
+    {
+        return self::count_by_status(self::STATUS_FAILED);
+    }
+
+    public static function count_sent_last_24h(): int
+    {
+        global $wpdb;
+        $t = self::table_name();
+        $cutoff = gmdate('Y-m-d H:i:s', time() - 86400);
+        return (int)$wpdb->get_var($wpdb->prepare(
+            "SELECT COUNT(*) FROM {$t} WHERE status = %s AND sent_at >= %s",
+            self::STATUS_SENT, $cutoff
+        ));
+    }
+
+    /** @return array<int,object> */
+    public static function list_failed(int $limit = 10): array
+    {
+        global $wpdb;
+        $t = self::table_name();
+        $rows = $wpdb->get_results($wpdb->prepare(
+            "SELECT id, created_at, last_error FROM {$t}
+             WHERE status = %s
+             ORDER BY id DESC LIMIT %d",
+            self::STATUS_FAILED, $limit
+        ));
+        return is_array($rows) ? $rows : [];
+    }
+
+    public static function retry_all_failed(): int
+    {
+        global $wpdb;
+        $t = self::table_name();
+        $now = current_time('mysql');
+        $count = $wpdb->query($wpdb->prepare(
+            "UPDATE {$t}
+             SET status = %s, attempts = 0, last_error = NULL, next_try_at = %s
+             WHERE status = %s",
+            self::STATUS_PENDING, $now, self::STATUS_FAILED
+        ));
+        return (int)$count;
+    }
+
+    public static function purge_sent(): int
+    {
+        global $wpdb;
+        $t = self::table_name();
+        return (int)$wpdb->query($wpdb->prepare(
+            "DELETE FROM {$t} WHERE status = %s", self::STATUS_SENT
+        ));
+    }
+
+    public static function cleanup_old_sent(int $days = 30): int
+    {
+        global $wpdb;
+        $t = self::table_name();
+        $cutoff = gmdate('Y-m-d H:i:s', time() - $days * 86400);
+        return (int)$wpdb->query($wpdb->prepare(
+            "DELETE FROM {$t} WHERE status = %s AND sent_at < %s",
+            self::STATUS_SENT, $cutoff
+        ));
+    }
+}