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

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

286 lines 10.5 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 [
59	'content' => $content,
60	'chat_id' => $responseBody['chat_id'] ?? null
61	];
62	}
63
64	public function makeStreamRequest(int $ticketId, $prompt, array $args = [])
65	{
66	$timeout = apply_filters('fs_ai_request_timeout', 120);
67
68	// Raw cURL because the WP HTTP API can't stream SSE.
69	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init
70	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt
71	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec
72	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close
73	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_getinfo
74	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_error
75	// PluginCheck:ignoreFile
76	$ch = curl_init();
77	curl_setopt($ch, CURLOPT_URL, $this->apiUrl);
78	curl_setopt($ch, CURLOPT_POST, true);
79	curl_setopt($ch, CURLOPT_POSTFIELDS, wp_json_encode($args));
80	$streamHeaders = ['Content-Type: application/json'];
81	if ($this->apiKey !== '') {
82	$streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey;
83	}
84	curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders);
85
86	$buffer = '';
87	$conversationId = null;
88	$streamTokens = 0;
89
90	curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer, &$conversationId, &$streamTokens) {
91	foreach ($this->drainSseChunk($data, $buffer) as $parsed) {
92	if ($parsed['event'] === 'chat_id' && !empty($parsed['data'])) {
93	$conversationId = $parsed['data'][0];
94	} elseif ($parsed['event'] === 'token_usage' && !empty($parsed['data'])) {
95	$usage = json_decode($parsed['data'][0], true);
96	if (is_array($usage)) {
97	$streamTokens = ($usage['input_tokens'] ?? 0) + ($usage['output_tokens'] ?? 0);
98	}
99	}
100	}
101
102	return strlen($data);
103	});
104
105	curl_setopt($ch, CURLOPT_TIMEOUT, $timeout);
106	curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
107	curl_setopt($ch, CURLOPT_BUFFERSIZE, 128); // Smaller buffer for faster streaming
108
109	$result = curl_exec($ch);
110	$httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
111
112	if (curl_error($ch)) {
113	echo "event: error\n";
114	echo "data: " . json_encode(['error' => curl_error($ch)]) . "\n\n";
115	flush();
116	}
117
118	curl_close($ch);
119
120	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_init
121	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_setopt
122	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_exec
123	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_close
124	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_getinfo
125	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_error
126
127	if ($httpCode === 200) {
128	do_action('fluent_support/ai_response_success', $ticketId, $prompt, $streamTokens, "FluentBot");
129	}
130	}
131
132	/**
133	* Forward every complete SSE frame in $data to the browser, return the parsed frames.
134	* $buffer holds the partial trailing frame across cURL writes, so pass it by reference.
135	*
136	* @param string $data raw bytes from cURL
137	* @param string $buffer incomplete frame carried over from the previous write
138	* @return array<int, array{event: string, data: array<int, string>}>
139	*/
140	private function drainSseChunk(string $data, string &$buffer): array
141	{
142	$buffer .= $data;
143
144	$events = explode("\n\n", $buffer);
145	$buffer = array_pop($events); // Keep incomplete event in buffer
146
147	$parsed = [];
148
149	foreach ($events as $event) {
150	if (!trim($event)) {
151	continue;
152	}
153
154	// Forward keepalive comments as-is so proxies don't hit an idle timeout.
155	if (strpos(ltrim($event), ':') === 0) {
156	echo $event . "\n\n";
157	flush();
158	continue;
159	}
160
161	$lines = explode("\n", $event);
162	$eventType = '';
163	$eventId = '';
164	$eventDataLines = [];
165
166	foreach ($lines as $line) {
167	if (strpos($line, 'event: ') === 0) {
168	$eventType = trim((string)substr($line, 7));
169	} elseif (strpos($line, 'id: ') === 0) {
170	$eventId = trim((string)substr($line, 4));
171	} elseif (strpos($line, 'data: ') === 0) {
172	$eventDataLines[] = (string)substr($line, 6);
173	}
174	}
175
176	if (!$eventType) {
177	continue;
178	}
179
180	echo "event: ".esc_html($eventType)."\n";
181
182	if ($eventId !== '') {
183	echo "id: ".esc_html($eventId)."\n";
184	}
185
186	if (!empty($eventDataLines)) {
187	foreach ($eventDataLines as $dataLine) {
188	// phpcs:ignore WordPress.Security.EscapeOutput.OutputNotEscaped -- raw SSE payload from trusted bot endpoint; rendered output sanitized client-side
189	echo "data: ".$dataLine."\n";
190	}
191	} else {
192	echo "data: \n";
193	}
194
195	echo "\n";
196	flush();
197
198	$parsed[] = ['event' => $eventType, 'data' => $eventDataLines];
199	}
200
201	return $parsed;
202	}
203
204	/**
205	* Reconnect to an in-flight turn: replay its buffered SSE and tail it live.
206	* Same wire format as makeStreamRequest. Upstream ends on `__done__` or an `idle` frame.
207	*/
208	public function makeResumeStreamRequest()
209	{
210	// Raw cURL because the WP HTTP API can't stream SSE.
211	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_init
212	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_setopt
213	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_exec
214	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_close
215	// phpcs:disable WordPress.WP.AlternativeFunctions.curl_curl_error
216	// PluginCheck:ignoreFile
217	$ch = curl_init();
218	curl_setopt($ch, CURLOPT_URL, $this->apiUrl);
219	curl_setopt($ch, CURLOPT_HTTPGET, true);
220
221	$streamHeaders = ['Accept: text/event-stream'];
222	if ($this->apiKey !== '') {
223	$streamHeaders[] = 'Authorization: Bearer ' . $this->apiKey;
224	}
225	curl_setopt($ch, CURLOPT_HTTPHEADER, $streamHeaders);
226
227	$buffer = '';
228
229	curl_setopt($ch, CURLOPT_WRITEFUNCTION, function($ch, $data) use (&$buffer) {
230	// Stop when the agent leaves. The turn itself is persisted upstream.
231	if (connection_aborted()) {
232	return 0;
233	}
234
235	$this->drainSseChunk($data, $buffer);
236
237	return strlen($data);
238	});
239
240	// connection_aborted() can lag behind proxies and buffering, so cap the tail
241	// or a refresh mid-turn can pin a worker. The client refetches a cut tail.
242	curl_setopt($ch, CURLOPT_TIMEOUT, apply_filters('fs_ai_resume_timeout', 120));
243	curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 15);
244	// Upstream sends keepalives during quiet gaps, so a stall here means it's dead.
245	curl_setopt($ch, CURLOPT_LOW_SPEED_LIMIT, 1);
246	curl_setopt($ch, CURLOPT_LOW_SPEED_TIME, apply_filters('fs_ai_resume_stall_timeout', 60));
247	curl_setopt($ch, CURLOPT_TCP_KEEPALIVE, 1);
248	curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
249	curl_setopt($ch, CURLOPT_BUFFERSIZE, 128); // Smaller buffer for faster streaming
250
251	curl_exec($ch);
252
253	if (curl_error($ch)) {
254	echo "event: error\n";
255	echo "data: " . wp_json_encode(['error' => curl_error($ch)]) . "\n\n";
256	flush();
257	}
258
259	curl_close($ch);
260	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_init
261	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_setopt
262	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_exec
263	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_close
264	// phpcs:enable WordPress.WP.AlternativeFunctions.curl_curl_error
265	}
266
267	protected function sendRequest(array $payload)
268	{
269	$headers = [
270	'Content-Type' => 'application/json',
271	];
272
273	if ($this->apiKey !== '') {
274	$headers['Authorization'] = 'Bearer ' . $this->apiKey;
275	}
276
277	$timeout = apply_filters('fs_ai_request_timeout', 60);
278
279	return wp_remote_post($this->apiUrl, [
280	'headers' => $headers,
281	'body' => wp_json_encode($payload),
282	'timeout' => $timeout,
283	]);
284	}
285	}
286