cambios importantes

2026-06-12 12:18:21 -05:00
parent efebec3c1b
commit bc422e0a1c
20 changed files with 5348 additions and 99 deletions
@@ -0,0 +1,96 @@
+<?php
+declare(strict_types=1);
+
+class OutboundWorker
+{
+    private const BATCH_SIZE = 10;
+
+    public static function processQueue(): array
+    {
+        $processed = 0;
+        $errors    = [];
+
+        $items = self::dequeue();
+
+        foreach ($items as $item) {
+            try {
+                $result = self::process($item);
+
+                if ($result['success']) {
+                    $stmt = db()->prepare("UPDATE outbound_queue SET status = 'sent', wam_id = ?, updated_at = NOW() WHERE id = ?");
+                    $stmt->execute([$result['wam_id'], $item['id']]);
+                    $processed++;
+                } else {
+                    $attempts = (int)$item['attempts'] + 1;
+                    $max      = (int)$item['max_attempts'];
+
+                    if ($attempts >= $max) {
+                        $stmt = db()->prepare("UPDATE outbound_queue SET status = 'failed', attempts = ?, last_error = ?, updated_at = NOW() WHERE id = ?");
+                        $stmt->execute([$attempts, $result['error'], $item['id']]);
+                    } else {
+                        $stmt = db()->prepare("UPDATE outbound_queue SET status = 'queued', attempts = ?, last_error = ?, updated_at = NOW() WHERE id = ?");
+                        $stmt->execute([$attempts, $result['error'], $item['id']]);
+                    }
+
+                    $errors[] = "msg#{$item['id']}: {$result['error']}";
+                }
+            } catch (\Exception $e) {
+                $errors[] = "msg#{$item['id']}: " . $e->getMessage();
+            }
+        }
+
+        return ['processed' => $processed, 'errors' => $errors];
+    }
+
+    private static function dequeue(): array
+    {
+        $db = db();
+        $db->beginTransaction();
+
+        try {
+            $stmt = $db->prepare("
+                SELECT q.*, c.phone_number_id
+                FROM outbound_queue q
+                JOIN companies c ON c.id = q.company_id
+                WHERE q.status = 'queued' AND c.is_active = 1
+                ORDER BY q.id ASC
+                LIMIT ?
+                FOR UPDATE SKIP LOCKED
+            ");
+            $stmt->execute([self::BATCH_SIZE]);
+            $items = $stmt->fetchAll();
+
+            if (!empty($items)) {
+                $ids = array_column($items, 'id');
+                $placeholders = implode(',', array_fill(0, count($ids), '?'));
+                $db->prepare("UPDATE outbound_queue SET status = 'sending', updated_at = NOW() WHERE id IN ({$placeholders})")->execute($ids);
+            }
+
+            $db->commit();
+            return $items;
+        } catch (\Exception $e) {
+            $db->rollBack();
+            throw $e;
+        }
+    }
+
+    private static function process(array $item): array
+    {
+        $payload = json_decode($item['payload'], true);
+        if (!is_array($payload)) {
+            return ['success' => false, 'error' => 'Payload inválido'];
+        }
+
+        $phoneNumberId = $item['phone_number_id'];
+        $type          = $item['message_type'];
+        $to            = $item['to_number'];
+
+        return match ($type) {
+            'text'    => WhatsAppSender::sendText($to, $payload['text'] ?? '', $phoneNumberId),
+            'image'   => WhatsAppSender::sendImage($to, $payload['media_id'] ?? $payload['url'] ?? '', $phoneNumberId, $payload['caption'] ?? null),
+            'template'=> WhatsAppSender::sendTemplate($to, $payload['template_name'] ?? '', $phoneNumberId, $payload['components'] ?? []),
+            'interactive' => WhatsAppSender::sendInteractive($to, $payload['interactive'] ?? [], $phoneNumberId),
+            default   => ['success' => false, 'error' => "Tipo no soportado: {$type}"],
+        };
+    }
+}