Feature: Implement multi-stage Conversation Flow Engine with TestFlow

2026-05-22 05:11:35 +03:00
parent b82a02f6fa
commit 7ec4d9becb
11 changed files with 737 additions and 0 deletions
--- a/backend/app/Controllers/WhatsAppController.php
+++ b/backend/app/Controllers/WhatsAppController.php
@@ -299,6 +299,11 @@ class WhatsAppController extends BaseController
    private function triggerAutoReply(array $session, array $msgData)
    {
        try {
+            // Hook the Conversation Flow Engine to intercept messages for active or new flows
+            if (\App\Core\Flows\ConversationFlowEngine::processMessage($session, $msgData)) {
+                return;
+            }
+
            $rule = \App\Models\ChatbotRule::findActiveForRule($session['company_id'], $session['id']);
            if (!$rule || !$rule['is_active']) {
                return;
--- a/backend/app/Core/Flows/BaseFlow.php
+++ b/backend/app/Core/Flows/BaseFlow.php
@@ -0,0 +1,20 @@
+<?php
+
+namespace App\Core\Flows;
+
+/**
+ * BaseFlow
+ * Abstract base class for all conversation flow states.
+ */
+abstract class BaseFlow
+{
+    /**
+     * Process a step in the conversation flow.
+     *
+     * @param string $step The current step identifier ('start' for new flows)
+     * @param array $messageData The incoming WhatsApp message payload (body, phone, etc.)
+     * @param array $context Reference to the persistent JSON context array
+     * @return FlowResult
+     */
+    abstract public function handleStep(string $step, array $messageData, array &$context): FlowResult;
+}
--- a/backend/app/Core/Flows/ConversationFlowEngine.php
+++ b/backend/app/Core/Flows/ConversationFlowEngine.php
@@ -0,0 +1,177 @@
+<?php
+
+namespace App\Core\Flows;
+
+use App\Models\ConversationState;
+use App\Models\MessageLog;
+
+/**
+ * ConversationFlowEngine
+ * Orchestrates multi-stage interactive conversation flows.
+ */
+class ConversationFlowEngine
+{
+    /**
+     * Map of registered flow names to their classes
+     */
+    private static array $flows = [
+        'test_flow' => TestFlow::class,
+    ];
+
+    /**
+     * Map of keyword triggers to start a specific flow
+     */
+    private static array $startTriggers = [
+        'test' => 'test_flow',
+        'اختبار' => 'test_flow',
+    ];
+
+    /**
+     * Process incoming message.
+     * Returns true if handled by the flow engine, false otherwise.
+     */
+    public static function processMessage(array $session, array $msgData): bool
+    {
+        // 1. Housekeeping: remove expired sessions
+        ConversationState::cleanExpired();
+
+        $phone = $msgData['phone'];
+        $companyId = $session['company_id'];
+        $text = isset($msgData['body']) ? trim($msgData['body']) : '';
+
+        // 2. Lookup existing active flow
+        $state = ConversationState::findActive($companyId, $phone);
+
+        $flowName = '';
+        $currentStep = '';
+        $context = [];
+
+        if ($state) {
+            $flowName = $state['flow_name'];
+            $currentStep = $state['current_step'];
+            $context = json_decode($state['context_data'] ?: '{}', true) ?: [];
+        } else {
+            // Check if message is a starting trigger for a new flow
+            $normalizedText = strtolower(trim($text));
+            if (isset(self::$startTriggers[$normalizedText])) {
+                $flowName = self::$startTriggers[$normalizedText];
+                $currentStep = 'start';
+                $context = [];
+            }
+        }
+
+        // If no active flow and no trigger matches, pass control back to normal bot rules
+        if (empty($flowName) || !isset(self::$flows[$flowName])) {
+            return false;
+        }
+
+        // 3. User cancel flow option
+        $normalizedCancel = strtolower(trim($text));
+        if (in_array($normalizedCancel, ['إلغاء', 'خروج', 'cancel', 'exit'])) {
+            if ($state) {
+                ConversationState::deleteState($state['id']);
+                $cancelMsg = in_array($normalizedCancel, ['cancel', 'exit']) 
+                    ? 'Interactive flow cancelled.' 
+                    : 'تم إلغاء المحادثة التفاعلية.';
+                self::sendReply($session, $phone, $cancelMsg);
+                return true;
+            }
+        }
+
+        // 4. Instantiate and execute the flow
+        $flowClass = self::$flows[$flowName];
+        /** @var BaseFlow $flowInstance */
+        $flowInstance = new $flowClass();
+
+        try {
+            $result = $flowInstance->handleStep($currentStep, $msgData, $context);
+
+            if ($result->isFinished()) {
+                // Flow has reached terminal state: clean up DB record
+                if ($state) {
+                    ConversationState::deleteState($state['id']);
+                }
+            } else {
+                // Flow has next step: save or update state record (TTL: 1 hour)
+                ConversationState::saveState([
+                    'company_id' => $companyId,
+                    'contact_phone' => $phone,
+                    'flow_name' => $flowName,
+                    'current_step' => $result->getNextStep(),
+                    'context_data' => json_encode($context, JSON_UNESCAPED_UNICODE),
+                    'expires_at' => date('Y-m-d H:i:s', strtotime('+1 hour'))
+                ]);
+            }
+
+            // 5. Send reply if one is provided
+            if ($result->getReplyText() !== '') {
+                self::sendReply($session, $phone, $result->getReplyText());
+            }
+
+            return true;
+        } catch (\Exception $e) {
+            error_log("[ConversationFlowEngine Exception] Flow '{$flowName}' failed: " . $e->getMessage());
+            return false;
+        }
+    }
+
+    /**
+     * Send outbound message reply via the Baileys Gateway
+     */
+    private static function sendReply(array $session, string $phone, string $message): void
+    {
+        $gatewayUrl = rtrim(getenv('WHATSAPP_GATEWAY_URL') ?: 'http://localhost:3722', '/');
+        if (substr($gatewayUrl, -4) === '/api') {
+            $sendUrl = $gatewayUrl . '/messages/send';
+        } else {
+            $sendUrl = $gatewayUrl . '/api/messages/send';
+        }
+
+        $payload = json_encode([
+            'session_key' => $session['session_key'],
+            'phone' => $phone,
+            'message' => $message
+        ]);
+
+        $ch = curl_init($sendUrl);
+        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+        curl_setopt($ch, CURLOPT_POST, true);
+        curl_setopt($ch, CURLOPT_POSTFIELDS, $payload);
+        curl_setopt($ch, CURLOPT_HTTPHEADER, [
+            'Content-Type: application/json',
+            'X-Webhook-Secret: ' . getenv('WEBHOOK_SECRET')
+        ]);
+        curl_setopt($ch, CURLOPT_TIMEOUT, 10);
+        
+        $response = curl_exec($ch);
+        $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
+        curl_close($ch);
+
+        $status = 'failed';
+        $errorMsg = null;
+        $waMsgId = null;
+
+        if ($httpCode === 200) {
+            $status = 'sent';
+            $resData = json_decode($response, true);
+            $waMsgId = $resData['data']['key']['id'] ?? null;
+        } else {
+            $resData = json_decode($response, true);
+            $errorMsg = $resData['error'] ?? 'HTTP Code ' . $httpCode;
+            error_log("[Flow Engine Gateway Error] Failed to send: " . $errorMsg);
+        }
+
+        // Log the outbound auto-reply message
+        MessageLog::logMessage([
+            'company_id' => $session['company_id'],
+            'session_id' => $session['id'],
+            'contact_phone' => $phone,
+            'direction' => 'outbound',
+            'message_type' => 'text',
+            'message_body' => $message,
+            'whatsapp_message_id' => $waMsgId,
+            'status' => $status,
+            'error_message' => $errorMsg
+        ]);
+    }
+}
--- a/backend/app/Core/Flows/FlowResult.php
+++ b/backend/app/Core/Flows/FlowResult.php
@@ -0,0 +1,55 @@
+<?php
+
+namespace App\Core\Flows;
+
+/**
+ * FlowResult
+ * Holds response details after handling a single step in a conversation flow.
+ */
+class FlowResult
+{
+    private string $replyText;
+    private string $nextStep;
+    private bool $finished;
+    private ?string $mediaUrl;
+
+    public function __construct(string $replyText, string $nextStep, bool $finished = false, ?string $mediaUrl = null)
+    {
+        $this->replyText = $replyText;
+        $this->nextStep = $nextStep;
+        $this->finished = $finished;
+        $this->mediaUrl = $mediaUrl;
+    }
+
+    /**
+     * Get the reply message text to send to the contact
+     */
+    public function getReplyText(): string
+    {
+        return $this->replyText;
+    }
+
+    /**
+     * Get the next step key
+     */
+    public function getNextStep(): string
+    {
+        return $this->nextStep;
+    }
+
+    /**
+     * Check if the flow is finished (to be destroyed)
+     */
+    public function isFinished(): bool
+    {
+        return $this->finished;
+    }
+
+    /**
+     * Get target media URL (if any)
+     */
+    public function getMediaUrl(): ?string
+    {
+        return $this->mediaUrl;
+    }
+}
--- a/backend/app/Core/Flows/TestFlow.php
+++ b/backend/app/Core/Flows/TestFlow.php
@@ -0,0 +1,38 @@
+<?php
+
+namespace App\Core\Flows;
+
+/**
+ * TestFlow
+ * A basic interactive flow for testing the multi-stage system.
+ */
+class TestFlow extends BaseFlow
+{
+    public function handleStep(string $step, array $messageData, array &$context): FlowResult
+    {
+        $text = isset($messageData['body']) ? trim($messageData['body']) : '';
+
+        switch ($step) {
+            case 'start':
+                // Initiate step
+                return new FlowResult("أهلاً بك في اختبار المسار التفاعلي! ما هو اسمك الكريم؟", "ask_name");
+
+            case 'ask_name':
+                if (empty($text)) {
+                    return new FlowResult("يرجى إدخال اسمك للاستمرار:", "ask_name");
+                }
+                $context['name'] = $text;
+                return new FlowResult("تشرفنا بك يا {$text}! من فضلك قم بتقييم خدمتنا من 1 إلى 5:", "ask_feedback");
+
+            case 'ask_feedback':
+                if (!preg_match('/^[1-5]$/', $text)) {
+                    return new FlowResult("الرجاء إدخال رقم من 1 إلى 5 فقط للتقييم:", "ask_feedback");
+                }
+                $context['rating'] = (int)$text;
+                return new FlowResult("شكراً لك يا {$context['name']}! لقد تم تسجيل تقييمك ({$text}/5) بنجاح. يومك سعيد!", "finished", true);
+
+            default:
+                return new FlowResult("خطأ في تحديد خطوة المسار.", "finished", true);
+        }
+    }
+}
--- a/backend/app/Models/ConversationState.php
+++ b/backend/app/Models/ConversationState.php
@@ -0,0 +1,119 @@
+<?php
+
+namespace App\Models;
+
+use App\Core\Database;
+use App\Core\Security;
+
+/**
+ * ConversationState Model
+ * Manages conversation states for multi-step bot flows.
+ */
+class ConversationState extends BaseModel
+{
+    protected static string $table = 'conversation_states';
+
+    /**
+     * Find active conversation state for a company and contact phone
+     */
+    public static function findActive(int $companyId, string $phone): ?array
+    {
+        self::ensureTableExists();
+        $hash = Security::blindIndex($phone);
+        $state = Database::selectOne(
+            "SELECT * FROM " . static::$table . " WHERE company_id = ? AND contact_phone_hash = ? AND expires_at > NOW() LIMIT 1",
+            [$companyId, $hash]
+        );
+        return self::decryptState($state);
+    }
+
+    /**
+     * Save or update conversation state securely
+     */
+    public static function saveState(array $data): string
+    {
+        self::ensureTableExists();
+
+        if (!empty($data['contact_phone'])) {
+            $data['contact_phone_hash'] = Security::blindIndex($data['contact_phone']);
+            $data['contact_phone'] = Security::encrypt($data['contact_phone']);
+        }
+
+        $existing = Database::selectOne(
+            "SELECT id FROM " . static::$table . " WHERE company_id = ? AND contact_phone_hash = ? LIMIT 1",
+            [$data['company_id'], $data['contact_phone_hash']]
+        );
+
+        if ($existing) {
+            self::update($existing['id'], $data);
+            return $existing['id'];
+        } else {
+            return self::create($data);
+        }
+    }
+
+    /**
+     * Delete conversation state by ID
+     */
+    public static function deleteState(int $id): int
+    {
+        self::ensureTableExists();
+        return self::delete($id);
+    }
+
+    /**
+     * Clean up all expired states
+     */
+    public static function cleanExpired(): void
+    {
+        self::ensureTableExists();
+        try {
+            Database::execute("DELETE FROM " . static::$table . " WHERE expires_at < NOW()");
+        } catch (\Exception $e) {
+            error_log("Failed to clean expired conversation states: " . $e->getMessage());
+        }
+    }
+
+    /**
+     * Helper to decrypt sensitive fields
+     */
+    private static function decryptState(?array $state): ?array
+    {
+        if ($state) {
+            $state['contact_phone'] = !empty($state['contact_phone']) ? Security::decrypt($state['contact_phone']) : null;
+        }
+        return $state;
+    }
+
+    /**
+     * Ensure the conversation_states table exists in the database
+     */
+    public static function ensureTableExists(): void
+    {
+        static $checked = false;
+        if ($checked) return;
+
+        try {
+            Database::execute("
+                CREATE TABLE IF NOT EXISTS `conversation_states` (
+                    `id` INT AUTO_INCREMENT PRIMARY KEY,
+                    `company_id` INT NOT NULL,
+                    `contact_phone` VARCHAR(512) NOT NULL,
+                    `contact_phone_hash` VARCHAR(64) NOT NULL,
+                    `flow_name` VARCHAR(100) NOT NULL,
+                    `current_step` VARCHAR(100) NOT NULL,
+                    `context_data` JSON DEFAULT NULL,
+                    `expires_at` TIMESTAMP NOT NULL,
+                    `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+                    UNIQUE KEY `unique_company_contact` (`company_id`, `contact_phone_hash`),
+                    FOREIGN KEY (`company_id`) REFERENCES `companies` (`id`) ON DELETE CASCADE,
+                    INDEX `idx_conv_expires` (`expires_at`)
+                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+            ");
+            $checked = true;
+        } catch (\Exception $e) {
+            error_log("Failed to ensure conversation_states table: " . $e->getMessage());
+        }
+    }
+}