rateLimitService = $rateLimitService; $this->botService = $botService; $this->botMapper = $botMapper; $this->talkHandler = $talkHandler; $this->logger = $logger; // Run every minute (60 seconds) $this->setInterval(60); // Allow some flexibility in timing $this->setTimeSensitivity(self::TIME_SENSITIVE); } /** * @param array $arguments */ protected function run($arguments): void { // Check if rate limiting is enabled if (!$this->rateLimitService->isEnabled()) { $this->logger->debug('EducAI: Rate limiting disabled, skipping queue processing'); return; } // Get queue stats $stats = $this->rateLimitService->getQueueStats(); if ($stats['pending'] === 0 && $stats['processing'] === 0) { $this->logger->debug('EducAI: No queued requests to process'); return; } $this->logger->info('EducAI: Starting queue processing job', [ 'pending' => $stats['pending'], 'processing' => $stats['processing'], ]); // First, handle stale processing requests (reset them) $this->rateLimitService->cleanup(self::MAX_REQUEST_AGE_SECONDS); // Process pending requests $processedCount = 0; $maxToProcess = min(self::MAX_REQUESTS_PER_RUN, $stats['pending']); for ($i = 0; $i < $maxToProcess; $i++) { // Check if we have rate limit capacity if (!$this->rateLimitService->canProcess()) { $waitSeconds = $this->rateLimitService->getSecondsUntilAvailable(); $this->logger->info('EducAI: Rate limit reached, stopping queue processing', [ 'processed' => $processedCount, 'wait_seconds' => $waitSeconds, ]); break; } // Get next pending request $request = $this->rateLimitService->getNextPending(); if ($request === null) { break; } // Skip stale requests if ($request->isStale(self::MAX_REQUEST_AGE_SECONDS)) { $this->rateLimitService->markFailed($request, 'Request expired (too old)'); $this->sendFailureNotification($request, 'Your request has expired. Please try again.'); continue; } // Process the request $this->processQueuedRequest($request); $processedCount++; // Small delay between requests to respect rate limits usleep(100000); // 100ms } $this->logger->info('EducAI: Queue processing completed', [ 'processed_count' => $processedCount, 'remaining_pending' => $stats['pending'] - $processedCount, ]); } /** * Process a single queued request */ private function processQueuedRequest(QueuedRequest $request): void { $requestId = $request->getId(); $this->logger->info('EducAI: Processing queued request', [ 'request_id' => $requestId, 'bot_id' => $request->getBotId(), 'attempts' => $request->getAttempts(), ]); // Mark as processing $this->rateLimitService->markProcessing($request); // Record rate limit usage $this->rateLimitService->recordUsage(); try { // Get the bot $bot = $this->botMapper->findById($request->getBotId()); if (!$bot->getIsActive()) { throw new \Exception('Bot is no longer active'); } // Process the message (with isFromQueue=true to skip rate limit check) $response = $this->botService->processMessage( $bot, $request->getMessage(), $request->getRoomToken(), $request->getUserId(), $request->getOriginalMessage(), null, // No streaming for queued requests true, // isFromQueue = true null, null, $request->getThreadRootMessageId(), $request->getReplyToMessageId() ); // Mark as completed $this->rateLimitService->markCompleted($request, $response); // Send the response to Talk $this->talkHandler->sendReplyToTalk( $request->getRoomToken(), $response, $request->getReplyToMessageId() ?? 0 ); $this->logger->info('EducAI: Successfully processed queued request', [ 'request_id' => $requestId, 'response_length' => strlen($response), ]); } catch (DoesNotExistException $e) { $error = 'Bot no longer exists'; $this->rateLimitService->markFailed($request, $error); $this->sendFailureNotification($request, 'The bot is no longer available.'); $this->logger->warning('EducAI: Queued request failed - bot not found', [ 'request_id' => $requestId, 'bot_id' => $request->getBotId(), ]); } catch (\Exception $e) { $error = $e->getMessage(); // Check if we should retry (note: attempts already incremented in markProcessing) if ($request->getAttempts() < self::MAX_RETRY_ATTEMPTS) { // Reset to pending so it will be picked up again $this->rateLimitService->markForRetry($request, $error); $this->logger->warning('EducAI: Queued request failed, will retry', [ 'request_id' => $requestId, 'attempts' => $request->getAttempts(), 'max_attempts' => self::MAX_RETRY_ATTEMPTS, 'error' => $error, ]); } else { // Max retries exceeded $this->rateLimitService->markFailed($request, 'Max retries exceeded: ' . $error); $this->sendFailureNotification( $request, self::GENERIC_FAILURE_NOTIFICATION ); $this->logger->error('EducAI: Queued request permanently failed', [ 'request_id' => $requestId, 'attempts' => $request->getAttempts(), 'error' => $error, ]); } } } /** * Send a failure notification to the Talk room */ private function sendFailureNotification(QueuedRequest $request, string $message): void { try { $this->talkHandler->sendReplyToTalk( $request->getRoomToken(), '⚠️ ' . $message, $request->getReplyToMessageId() ?? 0 ); } catch (\Exception $e) { $this->logger->error('EducAI: Failed to send failure notification', [ 'request_id' => $request->getId(), 'error' => $e->getMessage(), ]); } } }