From 09f26380665fc878147e888e032ce2c33b2ea6ea Mon Sep 17 00:00:00 2001 From: Vladimir Bryzgalov Date: Fri, 17 Apr 2026 22:41:30 +0500 Subject: [PATCH] feat(queue): CRUD, dedup, retention, status counters --- includes/class-queue.php | 182 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 includes/class-queue.php diff --git a/includes/class-queue.php b/includes/class-queue.php new file mode 100644 index 0000000..e5c0d99 --- /dev/null +++ b/includes/class-queue.php @@ -0,0 +1,182 @@ +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 */ + 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 */ + 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 + )); + } +}