fluent-support / app / Services / Integrations / FluentBot / FluentBotAPI.php

FluentBotAPI.php in Fluent Support – Helpdesk & Customer Support Ticket System trunk, at app/Services/Integrations/FluentBot/FluentBotAPI.php

299 lines 11.4 KB

Raw Download Zip

1	<?php
2
3	namespace FluentSupport\App\Services\Integrations\FluentBot;
4
5	use WP_Error;
6
7	class FluentBotAPI
8	{
9	protected $apiUrl;
10
11	protected $apiKey;
12
13	public function __construct(string $apiUrl, string $apiKey = '')
14	{
15	$this->apiUrl = $apiUrl;
16	$this->apiKey = $apiKey;
17	}
18
19	public function makeRequest(int $ticketId, $prompt, array $args = [])
20	{
21	$response = $this->sendRequest($args);
22
23	if (is_wp_error($response)) {
24	$message = $response->get_error_message();
25	$code = $response->get_error_code();
26	return new \WP_Error($code, $message);
27	}
28
29	$rawBody = wp_remote_retrieve_body($response);
30	$responseBody = json_decode($rawBody, true) ?? [];
31
32	if (!$responseBody \|\| !is_array($responseBody)) {
33	return new \WP_Error('fluent_bot_error', __('Invalid or empty response from API', 'fluent-support'));
34	}
35
36	if (!empty($responseBody['error'])) {
37	$message = $responseBody['error']['message'] ?? __('Unknown error occurred', 'fluent-support');
38	return new \WP_Error('fluent_bot_error', $message);
39	}
40
41	$statusCode = wp_remote_retrieve_response_code($response);
42
43	if ($statusCode !== 200) {
44	$error = $responseBody['message'] ?? __('Something went wrong.', 'fluent-support');
45	return new \WP_Error($statusCode, $error);
46	}
47
48	$content = $responseBody['response'] ?? '';
49
50	if (empty($content)) {
51	return new \WP_Error('fluent_bot_error', __('No AI response found in the API response.', 'fluent-support'));
52	}
53
54	$tokenUsage = $responseBody['token_usage'] ?? [];
55	$totalTokens = ($tokenUsage['input_tokens'] ?? 0) + ($tokenUsage['output_tokens'] ?? 0);
56	do_action('fluent_support/ai_response_success', $ticketId, $prompt, $totalTokens, "FluentBot");
57
58	// Return both content and chat_id if available
59	return [
60	'content' => $content,
61	'chat_id' => $responseBody['chat_id'] ?? null
62	];
63	}
64
65	public function makeStreamRequest(int $ticketId, $prompt, array $args = [])
66	{
67	$timeout = apply_filters('fs_ai_request_timeout', 120);
68
69	// Use cURL for streaming
70	// Note: Using cURL here because WordPress HTTP API doesn't support streaming SSE responses
71	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init
72	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt
73	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec
74	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close
75	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_getinfo
76	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_error
77	// PluginCheck:ignoreFile
78	$ch = curl_init();
79	curl_setopt($ch, CURLOPT_URL, $this->apiUrl);
80	curl_setopt($ch, CURLOPT_POST, true);
81	curl_setopt($ch, CURLOPT_POSTFIELDS, wp_json_encode($args));
82	$streamHeaders = ['Content-Type: application/json'];
83	if ($this->apiKey !== '') {
84	$streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey;
85	}
86	curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders);
87
88	$buffer = '';
89	$conversationId = null;
90	$streamTokens = 0;
91
92	curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer, &$conversationId, &$streamTokens) {
93	foreach ($this->drainSseChunk($data, $buffer) as $parsed) {
94	// Store chat_id for later use
95	if ($parsed['event'] === 'chat_id' && !empty($parsed['data'])) {
96	$conversationId = $parsed['data'][0];
97	} elseif ($parsed['event'] === 'token_usage' && !empty($parsed['data'])) {
98	$usage = json_decode($parsed['data'][0], true);
99	if (is_array($usage)) {
100	$streamTokens = ($usage['input_tokens'] ?? 0) + ($usage['output_tokens'] ?? 0);
101	}
102	}
103	}
104
105	return strlen($data);
106	});
107
108	curl_setopt($ch, CURLOPT_TIMEOUT, $timeout);
109	curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
110	curl_setopt($ch, CURLOPT_BUFFERSIZE, 128); // Smaller buffer for faster streaming
111
112	$result = curl_exec($ch);
113	$httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
114
115	if (curl_error($ch)) {
116	echo "event: error\n";
117	echo "data: " . json_encode(['error' => curl_error($ch)]) . "\n\n";
118	flush();
119	}
120
121	curl_close($ch);
122
123	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_init
124	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_setopt
125	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_exec
126	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_close
127	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_getinfo
128	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_error
129
130	if ($httpCode === 200) {
131	do_action('fluent_support/ai_response_success', $ticketId, $prompt, $streamTokens, "FluentBot");
132	}
133	}
134
135	/**
136	* Forward every complete SSE frame in $data to the browser, return the parsed frames.
137	* $buffer holds the partial trailing frame across cURL writes, so pass it by reference.
138	*
139	* @param string $data raw bytes from cURL
140	* @param string $buffer incomplete frame carried over from the previous write
141	* @return array<int, array{event: string, data: array<int, string>}>
142	*/
143	private function drainSseChunk(string $data, string &$buffer): array
144	{
145	$buffer .= $data;
146
147	// Process complete SSE events from the AI API
148	$events = explode("\n\n", $buffer);
149	$buffer = array_pop($events); // Keep incomplete event in buffer
150
151	$parsed = [];
152
153	foreach ($events as $event) {
154	if (!trim($event)) {
155	continue;
156	}
157
158	// SSE comment/keepalive (starts ":"): forward raw so proxies keep seeing bytes, no idle-timeout.
159	if (strpos(ltrim($event), ':') === 0) {
160	echo $event . "\n\n";
161	flush();
162	continue;
163	}
164
165	$lines = explode("\n", $event);
166	$eventType = '';
167	$eventId = '';
168	$eventDataLines = [];
169
170	foreach ($lines as $line) {
171	if (strpos($line, 'event: ') === 0) {
172	$eventType = trim((string)substr($line, 7));
173	} elseif (strpos($line, 'id: ') === 0) {
174	$eventId = trim((string)substr($line, 4));
175	} elseif (strpos($line, 'data: ') === 0) {
176	$eventDataLines[] = (string)substr($line, 6);
177	}
178	}
179
180	// Forward the event to the browser with proper formatting
181	if (!$eventType) {
182	continue;
183	}
184
185	echo "event: ".esc_html($eventType)."\n";
186
187	// Include ID if present
188	if ($eventId !== '') {
189	echo "id: ".esc_html($eventId)."\n";
190	}
191
192	// Handle multiple data lines properly
193	if (!empty($eventDataLines)) {
194	foreach ($eventDataLines as $dataLine) {
195	// phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- raw SSE payload from trusted bot endpoint; rendered output sanitized client-side
196	echo "data: ".$dataLine."\n";
197	}
198	} else {
199	echo "data: \n";
200	}
201
202	echo "\n";
203	flush();
204
205	$parsed[] = ['event' => $eventType, 'data' => $eventDataLines];
206	}
207
208	return $parsed;
209	}
210
211	/**
212	* Reconnect to an in-flight turn: replay its buffered SSE and tail it live. Same wire
213	* format as makeStreamRequest. Upstream ends on `__done__` or an `idle` frame; a total
214	* cap still applies — see the timeout block below.
215	*/
216	public function makeResumeStreamRequest()
217	{
218	// Use cURL for streaming
219	// Note: Using cURL here because WordPress HTTP API doesn't support streaming SSE responses
220	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init
221	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt
222	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec
223	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close
224	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_error
225	// PluginCheck:ignoreFile
226	$ch = curl_init();
227	curl_setopt($ch, CURLOPT_URL, $this->apiUrl);
228	curl_setopt($ch, CURLOPT_HTTPGET, true);
229
230	$streamHeaders = ['Accept: text/event-stream'];
231	if ($this->apiKey !== '') {
232	$streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey;
233	}
234	curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders);
235
236	$buffer = '';
237
238	curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer) {
239	// Stop as soon as the agent navigates away — nothing here needs to
240	// outlive the client, the turn itself is persisted upstream.
241	if (connection_aborted()) {
242	return 0;
243	}
244
245	$this->drainSseChunk($data, $buffer);
246
247	return strlen($data);
248	});
249
250	// Bounded on purpose: connection_aborted() can lag (abort travels browser ->
251	// proxy -> fluent-bot, Apache buffers writes), so a refresh mid-turn can leave
252	// the old tail pinning a worker. The cap makes that self-limiting — the client
253	// treats a cut tail as a body ended without __done__ and refetches.
254	curl_setopt($ch, CURLOPT_TIMEOUT, apply_filters('fs_ai_resume_timeout', 120));
255	curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 15);
256	// Abort sooner when genuinely stalled: the upstream sends ': keepalive'
257	// during quiet gaps, so a healthy tail always beats this window.
258	curl_setopt($ch, CURLOPT_LOW_SPEED_LIMIT, 1);
259	curl_setopt($ch, CURLOPT_LOW_SPEED_TIME, apply_filters('fs_ai_resume_stall_timeout', 60));
260	curl_setopt($ch, CURLOPT_TCP_KEEPALIVE, 1);
261	curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
262	curl_setopt($ch, CURLOPT_BUFFERSIZE, 128); // Smaller buffer for faster streaming
263
264	curl_exec($ch);
265
266	if (curl_error($ch)) {
267	echo "event: error\n";
268	echo "data: " . wp_json_encode(['error' => curl_error($ch)]) . "\n\n";
269	flush();
270	}
271
272	curl_close($ch);
273	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_init
274	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_setopt
275	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_exec
276	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_close
277	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_error
278	}
279
280	protected function sendRequest(array $payload)
281	{
282	$headers = [
283	'Content-Type' => 'application/json',
284	];
285
286	if ($this->apiKey !== '') {
287	$headers['Authorization'] = 'Bearer ' . $this->apiKey;
288	}
289
290	$timeout = apply_filters('fs_ai_request_timeout', 60);
291
292	return wp_remote_post($this->apiUrl, [
293	'headers' => $headers,
294	'body' => wp_json_encode($payload),
295	'timeout' => $timeout,
296	]);
297	}
298	}
299