| 1 |
<?php |
| 2 |
/** |
| 3 |
* File: admin/class-vectorstore-manager.php |
| 4 |
* |
| 5 |
* OpenAI Vector Store write path: initial import + ongoing sync from the |
| 6 |
* knowledge base (plan 15b5c6). Until this class existed the plugin could only |
| 7 |
* READ from a Vector Store (Responses API file_search) — enabling Vector Store |
| 8 |
* mode froze the bot's knowledge at activation time because every KB write |
| 9 |
* still went to WordPress/Pinecone only. |
| 10 |
* |
| 11 |
* Design notes that matter: |
| 12 |
* - One KB entry = ONE file in the store. Content is handed to OpenAI whole |
| 13 |
* (never pre-chunked here) — file_search chunks server-side, and one file |
| 14 |
* per entry is what makes updates/deletes tractable. |
| 15 |
* - Vector Store files are NOT patchable in place. An update is |
| 16 |
* upload-new -> attach-new -> detach+delete-old, and the KB-entry -> file_id |
| 17 |
* mapping lives in {prefix}mxchat_vectorstore_files. Without that mapping |
| 18 |
* the store accumulates stale duplicates and retrieval quality rots with no |
| 19 |
* error anywhere. New file goes in BEFORE the old one is removed so a |
| 20 |
* failure mid-swap leaves the entry answerable (a brief duplicate window is |
| 21 |
* the safe failure, permanent absence is not); a failed old-file delete is |
| 22 |
* parked as status=pending_delete and retried by the sweeper. |
| 23 |
* - Role-restricted entries are NEVER mirrored: file_search has no per-role |
| 24 |
* filtering, so a restricted entry in the store would be served to every |
| 25 |
* visitor. Import skips them, sync skips them, and a restriction change on |
| 26 |
* a mirrored entry deletes its file. |
| 27 |
* - The initial import runs as self-rescheduling WP-Cron ticks (admin kickoff |
| 28 |
* + progress UI), with a WP-CLI command (`wp mxchat vectorstore-import`) |
| 29 |
* that drives the same worker synchronously. A 1,500-entry import will not |
| 30 |
* finish in one request; cursor-based state in an option makes any |
| 31 |
* interruption resumable without duplicates (the mapping table + content |
| 32 |
* hash make re-processing an already-imported entry a no-op). |
| 33 |
* - Import uploads files individually, then attaches each tick's uploads with |
| 34 |
* ONE /file_batches call per tick, per OpenAI's bulk-ingestion guidance. |
| 35 |
*/ |
| 36 |
|
| 37 |
if (!defined('ABSPATH')) { |
| 38 |
exit; // Exit if accessed directly |
| 39 |
} |
| 40 |
|
| 41 |
class MxChat_Vectorstore_Manager { |
| 42 |
|
| 43 |
const TABLE = 'mxchat_vectorstore_files'; |
| 44 |
const STATE_OPTION = 'mxchat_vectorstore_import_state'; |
| 45 |
const TICK_HOOK = 'mxchat_vectorstore_import_tick'; |
| 46 |
const SWEEP_HOOK = 'mxchat_vectorstore_sweep'; |
| 47 |
const LOCK_TRANSIENT = 'mxchat_vectorstore_import_lock'; |
| 48 |
|
| 49 |
/** Entries processed per import tick (also the /file_batches attach size). */ |
| 50 |
const IMPORT_BATCH_SIZE = 20; |
| 51 |
/** Wall-clock budget per cron tick, seconds. */ |
| 52 |
const TICK_TIME_BUDGET = 25; |
| 53 |
|
| 54 |
private static $instance = null; |
| 55 |
|
| 56 |
public static function get_instance() { |
| 57 |
if (self::$instance === null) { |
| 58 |
self::$instance = new self(); |
| 59 |
} |
| 60 |
return self::$instance; |
| 61 |
} |
| 62 |
|
| 63 |
public function __construct() { |
| 64 |
add_action('wp_ajax_mxchat_vectorstore_import_start', array($this, 'ajax_import_start')); |
| 65 |
add_action('wp_ajax_mxchat_vectorstore_import_status', array($this, 'ajax_import_status')); |
| 66 |
add_action('wp_ajax_mxchat_vectorstore_import_resume', array($this, 'ajax_import_resume')); |
| 67 |
add_action('wp_ajax_mxchat_vectorstore_import_cancel', array($this, 'ajax_import_cancel')); |
| 68 |
add_action(self::TICK_HOOK, array($this, 'run_import_tick')); |
| 69 |
add_action(self::SWEEP_HOOK, array($this, 'run_sweep')); |
| 70 |
add_action('admin_init', array($this, 'maybe_schedule_sweep')); |
| 71 |
|
| 72 |
if (defined('WP_CLI') && WP_CLI) { |
| 73 |
WP_CLI::add_command('mxchat vectorstore-import', array($this, 'cli_import')); |
| 74 |
} |
| 75 |
} |
| 76 |
|
| 77 |
// ======================================== |
| 78 |
// CONFIG |
| 79 |
// ======================================== |
| 80 |
|
| 81 |
/** |
| 82 |
* Resolve the sync configuration for a bot. |
| 83 |
* |
| 84 |
* Mirrors get_bot_vectorstore_config()'s resolution shape: default bot (or |
| 85 |
* no multi-bot add-on) reads the global option; other bots go through a |
| 86 |
* filter so the multi-bot add-on can supply a per-bot store — one store per |
| 87 |
* bot, same seam the Pinecone namespace work uses (793b82/d4c6bb approval |
| 88 |
* note). No filter implementation -> non-default bots do NOT fall back to |
| 89 |
* the default store: cross-bot bleed into one store is the bug we just |
| 90 |
* spent two Pinecone plans killing. |
| 91 |
* |
| 92 |
* @return array {enabled: bool, store_id: string, api_key: string} |
| 93 |
*/ |
| 94 |
public static function get_sync_config($bot_id = 'default') { |
| 95 |
$mxchat_options = get_option('mxchat_options', array()); |
| 96 |
$api_key = $mxchat_options['api_key'] ?? ''; |
| 97 |
|
| 98 |
if ($bot_id === 'default' || !class_exists('MxChat_Multi_Bot_Manager')) { |
| 99 |
$vs_options = get_option('mxchat_openai_vectorstore_options', array()); |
| 100 |
$store_id = trim($vs_options['mxchat_vectorstore_sync_store_id'] ?? ''); |
| 101 |
if ($store_id === '') { |
| 102 |
// Fall back to the first configured retrieval store — the |
| 103 |
// common single-store setup shouldn't need the ID twice. |
| 104 |
$ids = array_filter(array_map('trim', explode(',', $vs_options['mxchat_vectorstore_ids'] ?? ''))); |
| 105 |
$store_id = $ids ? reset($ids) : ''; |
| 106 |
} |
| 107 |
return array( |
| 108 |
'enabled' => ($vs_options['mxchat_vectorstore_sync_enabled'] ?? '0') === '1' && $store_id !== '' && $api_key !== '', |
| 109 |
'store_id' => $store_id, |
| 110 |
'api_key' => $api_key, |
| 111 |
); |
| 112 |
} |
| 113 |
|
| 114 |
$bot_config = apply_filters('mxchat_get_bot_vectorstore_sync_config', array(), $bot_id); |
| 115 |
$enabled = !empty($bot_config['enabled']) && !empty($bot_config['store_id']); |
| 116 |
return array( |
| 117 |
'enabled' => $enabled && $api_key !== '', |
| 118 |
'store_id' => $bot_config['store_id'] ?? '', |
| 119 |
'api_key' => $api_key, |
| 120 |
); |
| 121 |
} |
| 122 |
|
| 123 |
/** |
| 124 |
* Stable mapping key for a KB identity (https://, upload://, mxchat://). |
| 125 |
* Deliberately the same md5 the Pinecone path uses as its base vector id, |
| 126 |
* so a Pinecone vector_id doubles as the mapping key for URL-keyed entries. |
| 127 |
*/ |
| 128 |
public static function entry_key($source_url) { |
| 129 |
return md5((string) $source_url); |
| 130 |
} |
| 131 |
|
| 132 |
private static function has_stable_identity($source_url) { |
| 133 |
return !empty($source_url) && preg_match('#^(https?|upload|mxchat)://#i', $source_url); |
| 134 |
} |
| 135 |
|
| 136 |
// ======================================== |
| 137 |
// SYNC ENTRY POINTS (called from MxChat_Utils) |
| 138 |
// ======================================== |
| 139 |
|
| 140 |
/** |
| 141 |
* Mirror a KB write into the Vector Store. Failure here must never fail |
| 142 |
* the primary KB write — callers already stored successfully; we log and |
| 143 |
* return, and the entry self-heals on its next save or via re-import. |
| 144 |
*/ |
| 145 |
public static function sync_upsert_entry($source_url, $content, $bot_id = 'default', $content_type = 'content') { |
| 146 |
$config = self::get_sync_config($bot_id); |
| 147 |
if (!$config['enabled']) { |
| 148 |
return; |
| 149 |
} |
| 150 |
if (!self::has_stable_identity($source_url)) { |
| 151 |
// Brand-new manual content has no identity at this seam; it gains |
| 152 |
// one in storage and is picked up by the import / its next edit. |
| 153 |
return; |
| 154 |
} |
| 155 |
|
| 156 |
// Canonicalize through the stored KB rows when they exist so the save |
| 157 |
// path and the import path hash identical bytes — chunk reassembly |
| 158 |
// differs from pre-chunk content in whitespace, and a byte mismatch |
| 159 |
// here would re-upload every chunked entry on each import/save |
| 160 |
// ping-pong. Pinecone-mode entries have no local rows; the passed |
| 161 |
// content is used as-is there. |
| 162 |
$local = self::read_local_entry($source_url); |
| 163 |
if ($local !== null) { |
| 164 |
$content = $local['content']; |
| 165 |
$content_type = $local['content_type']; |
| 166 |
if ($local['restricted']) { |
| 167 |
self::log('skip restricted entry ' . $source_url); |
| 168 |
return; |
| 169 |
} |
| 170 |
} |
| 171 |
if (self::entry_is_restricted($source_url, $bot_id)) { |
| 172 |
// Never mirror role-restricted content — the store has no per-role |
| 173 |
// filtering, so it would be served to every visitor. |
| 174 |
self::log('skip restricted entry ' . $source_url); |
| 175 |
return; |
| 176 |
} |
| 177 |
|
| 178 |
$result = self::upsert_file($config, self::entry_key($source_url), $source_url, $content, $bot_id, $content_type); |
| 179 |
if (is_wp_error($result)) { |
| 180 |
self::log('sync upsert failed for ' . $source_url . ' — ' . $result->get_error_message()); |
| 181 |
} |
| 182 |
} |
| 183 |
|
| 184 |
/** |
| 185 |
* Mirror a KB entry removal. Same failure isolation as sync_upsert_entry. |
| 186 |
*/ |
| 187 |
public static function sync_delete_entry($source_url, $bot_id = 'default') { |
| 188 |
if (!self::has_stable_identity($source_url)) { |
| 189 |
return; |
| 190 |
} |
| 191 |
self::sync_delete_by_key(self::entry_key($source_url), $bot_id); |
| 192 |
} |
| 193 |
|
| 194 |
/** |
| 195 |
* Removal by mapping key — for callers that hold a Pinecone vector id |
| 196 |
* rather than a URL (base id == md5(url) == our key; chunk ids reduce to |
| 197 |
* their base). Config-independent on purpose: even with sync toggled off, |
| 198 |
* deleting a KB entry should remove a previously-mirrored file rather than |
| 199 |
* strand it in the store. |
| 200 |
*/ |
| 201 |
public static function sync_delete_by_key($entry_key, $bot_id = 'default') { |
| 202 |
if (class_exists('MxChat_Chunker')) { |
| 203 |
$base = MxChat_Chunker::get_base_hash_from_vector_id($entry_key); |
| 204 |
if ($base !== null) { |
| 205 |
$entry_key = $base; |
| 206 |
} |
| 207 |
} |
| 208 |
if (!preg_match('/^[a-f0-9]{32}$/', (string) $entry_key)) { |
| 209 |
return; |
| 210 |
} |
| 211 |
|
| 212 |
global $wpdb; |
| 213 |
$table = $wpdb->prefix . self::TABLE; |
| 214 |
$rows = $wpdb->get_results($wpdb->prepare( |
| 215 |
"SELECT id, store_id, file_id FROM {$table} WHERE entry_key = %s AND bot_id = %s", |
| 216 |
$entry_key, $bot_id |
| 217 |
)); |
| 218 |
if (empty($rows)) { |
| 219 |
return; |
| 220 |
} |
| 221 |
|
| 222 |
$mxchat_options = get_option('mxchat_options', array()); |
| 223 |
$api_key = $mxchat_options['api_key'] ?? ''; |
| 224 |
|
| 225 |
foreach ($rows as $row) { |
| 226 |
// Park first, then attempt: if the API call dies mid-flight the |
| 227 |
// sweeper still knows this file is condemned. |
| 228 |
$wpdb->update($table, array('status' => 'pending_delete'), array('id' => $row->id), array('%s'), array('%d')); |
| 229 |
if ($api_key !== '' && self::remove_remote_file($api_key, $row->store_id, $row->file_id)) { |
| 230 |
$wpdb->delete($table, array('id' => $row->id), array('%d')); |
| 231 |
} |
| 232 |
} |
| 233 |
} |
| 234 |
|
| 235 |
/** |
| 236 |
* A role restriction just changed on a KB entry. Non-public -> pull the |
| 237 |
* mirrored file. Back to public -> best-effort re-mirror from the local |
| 238 |
* KB row (Pinecone-mode content isn't local; it re-mirrors on next save). |
| 239 |
*/ |
| 240 |
public static function handle_role_change($source_url, $bot_id, $new_restriction) { |
| 241 |
if (!self::has_stable_identity($source_url)) { |
| 242 |
return; |
| 243 |
} |
| 244 |
if (!empty($new_restriction) && $new_restriction !== 'public') { |
| 245 |
self::sync_delete_by_key(self::entry_key($source_url), $bot_id); |
| 246 |
return; |
| 247 |
} |
| 248 |
|
| 249 |
$config = self::get_sync_config($bot_id); |
| 250 |
if (!$config['enabled']) { |
| 251 |
return; |
| 252 |
} |
| 253 |
$entry = self::read_local_entry($source_url); |
| 254 |
if ($entry !== null) { |
| 255 |
self::sync_upsert_entry($source_url, $entry['content'], $bot_id, $entry['content_type']); |
| 256 |
} |
| 257 |
} |
| 258 |
|
| 259 |
// ======================================== |
| 260 |
// CORE UPSERT / DELETE ENGINE |
| 261 |
// ======================================== |
| 262 |
|
| 263 |
/** |
| 264 |
* The upload body for an entry — also the input to the change-detection |
| 265 |
* hash, so import and ongoing sync must build it identically. |
| 266 |
*/ |
| 267 |
private static function build_file_body($source_url, $content) { |
| 268 |
$body = ''; |
| 269 |
if (preg_match('#^https?://#i', (string) $source_url)) { |
| 270 |
$body .= 'Source: ' . $source_url . "\n\n"; |
| 271 |
} |
| 272 |
return $body . $content; |
| 273 |
} |
| 274 |
|
| 275 |
/** |
| 276 |
* Upload-new -> attach-new -> record -> detach+delete-old. |
| 277 |
* |
| 278 |
* @param bool $defer_attach Import path: skip the per-file attach and let |
| 279 |
* the caller batch-attach via /file_batches. |
| 280 |
* @return true|string|WP_Error true = swapped/attached, 'unchanged' = hash |
| 281 |
* match no-op, string file_id when $defer_attach (caller attaches). |
| 282 |
*/ |
| 283 |
public static function upsert_file($config, $entry_key, $source_url, $content, $bot_id, $content_type, $defer_attach = false) { |
| 284 |
global $wpdb; |
| 285 |
$table = $wpdb->prefix . self::TABLE; |
| 286 |
|
| 287 |
$body = self::build_file_body($source_url, $content); |
| 288 |
$hash = md5($body); |
| 289 |
|
| 290 |
$live = $wpdb->get_row($wpdb->prepare( |
| 291 |
"SELECT id, file_id, content_hash FROM {$table} WHERE store_id = %s AND bot_id = %s AND entry_key = %s AND status = 'live' LIMIT 1", |
| 292 |
$config['store_id'], $bot_id, $entry_key |
| 293 |
)); |
| 294 |
|
| 295 |
if ($live && $live->content_hash === $hash) { |
| 296 |
return 'unchanged'; |
| 297 |
} |
| 298 |
|
| 299 |
$file_id = self::api_upload_file($config['api_key'], 'mxchat-kb-' . $entry_key . '.txt', $body); |
| 300 |
if (is_wp_error($file_id)) { |
| 301 |
if ($live) { |
| 302 |
$wpdb->update($table, array('last_error' => $file_id->get_error_message()), array('id' => $live->id), array('%s'), array('%d')); |
| 303 |
} |
| 304 |
return $file_id; |
| 305 |
} |
| 306 |
|
| 307 |
if (!$defer_attach) { |
| 308 |
$attached = self::api_attach_file($config['api_key'], $config['store_id'], $file_id); |
| 309 |
if (is_wp_error($attached)) { |
| 310 |
// Orphaned upload: not in the store (harmless to retrieval), |
| 311 |
// delete the file object so it doesn't leak storage. |
| 312 |
self::api_delete_file($config['api_key'], $file_id); |
| 313 |
if ($live) { |
| 314 |
$wpdb->update($table, array('last_error' => $attached->get_error_message()), array('id' => $live->id), array('%s'), array('%d')); |
| 315 |
} |
| 316 |
return $attached; |
| 317 |
} |
| 318 |
} |
| 319 |
|
| 320 |
// Condemn any stranded pending_attach rows for this entry first — a |
| 321 |
// hard-killed earlier import may have uploaded a file that never made |
| 322 |
// it into a batch; its upload is reclaimed by the sweeper. |
| 323 |
$wpdb->query($wpdb->prepare( |
| 324 |
"UPDATE {$table} SET status = 'pending_delete' |
| 325 |
WHERE store_id = %s AND bot_id = %s AND entry_key = %s AND status = 'pending_attach'", |
| 326 |
$config['store_id'], $bot_id, $entry_key |
| 327 |
)); |
| 328 |
|
| 329 |
// Record the new file. Deferred (import) uploads are NOT in the store |
| 330 |
// yet — they become 'live' only after the tick's file_batches call |
| 331 |
// succeeds; recording them as live immediately would make a kill |
| 332 |
// between upload and attach look like a completed import on resume |
| 333 |
// (hash match -> skipped -> file never attached). |
| 334 |
$wpdb->insert($table, array( |
| 335 |
'store_id' => $config['store_id'], |
| 336 |
'bot_id' => $bot_id, |
| 337 |
'entry_key' => $entry_key, |
| 338 |
'source_url' => $source_url, |
| 339 |
'file_id' => $file_id, |
| 340 |
'content_hash' => $hash, |
| 341 |
'status' => $defer_attach ? 'pending_attach' : 'live', |
| 342 |
'updated_at' => current_time('mysql'), |
| 343 |
), array('%s', '%s', '%s', '%s', '%s', '%s', '%s', '%s')); |
| 344 |
|
| 345 |
if ($live) { |
| 346 |
$wpdb->update($table, array('status' => 'pending_delete'), array('id' => $live->id), array('%s'), array('%d')); |
| 347 |
if (self::remove_remote_file($config['api_key'], $config['store_id'], $live->file_id)) { |
| 348 |
$wpdb->delete($table, array('id' => $live->id), array('%d')); |
| 349 |
} |
| 350 |
// Failure path: row stays pending_delete; the sweeper retries. |
| 351 |
} |
| 352 |
|
| 353 |
return $defer_attach ? $file_id : true; |
| 354 |
} |
| 355 |
|
| 356 |
/** |
| 357 |
* Detach from the store AND delete the file object. 404 on either leg |
| 358 |
* counts as success — the goal state is "gone". |
| 359 |
*/ |
| 360 |
private static function remove_remote_file($api_key, $store_id, $file_id) { |
| 361 |
$detached = self::api_detach_file($api_key, $store_id, $file_id); |
| 362 |
if (is_wp_error($detached)) { |
| 363 |
return false; |
| 364 |
} |
| 365 |
$deleted = self::api_delete_file($api_key, $file_id); |
| 366 |
return !is_wp_error($deleted); |
| 367 |
} |
| 368 |
|
| 369 |
/** |
| 370 |
* Sweeper: retry condemned files that survived their first delete attempt. |
| 371 |
* Bounded per run; scheduled only while sync is enabled. |
| 372 |
*/ |
| 373 |
public function run_sweep() { |
| 374 |
global $wpdb; |
| 375 |
$table = $wpdb->prefix . self::TABLE; |
| 376 |
|
| 377 |
$mxchat_options = get_option('mxchat_options', array()); |
| 378 |
$api_key = $mxchat_options['api_key'] ?? ''; |
| 379 |
if ($api_key === '') { |
| 380 |
return; |
| 381 |
} |
| 382 |
|
| 383 |
$rows = $wpdb->get_results( |
| 384 |
"SELECT id, store_id, file_id FROM {$table} WHERE status = 'pending_delete' ORDER BY id ASC LIMIT 25" |
| 385 |
); |
| 386 |
foreach ($rows as $row) { |
| 387 |
if (self::remove_remote_file($api_key, $row->store_id, $row->file_id)) { |
| 388 |
$wpdb->delete($table, array('id' => $row->id), array('%d')); |
| 389 |
} |
| 390 |
} |
| 391 |
} |
| 392 |
|
| 393 |
public function maybe_schedule_sweep() { |
| 394 |
$config = self::get_sync_config('default'); |
| 395 |
$scheduled = wp_next_scheduled(self::SWEEP_HOOK); |
| 396 |
if ($config['enabled'] && !$scheduled) { |
| 397 |
wp_schedule_event(time() + HOUR_IN_SECONDS, 'twicedaily', self::SWEEP_HOOK); |
| 398 |
} elseif (!$config['enabled'] && $scheduled) { |
| 399 |
// Leave scheduled while any condemned rows remain — disabling sync |
| 400 |
// shouldn't strand files the store was already told to forget. |
| 401 |
global $wpdb; |
| 402 |
$table = $wpdb->prefix . self::TABLE; |
| 403 |
$pending = (int) $wpdb->get_var("SELECT COUNT(*) FROM {$table} WHERE status = 'pending_delete'"); |
| 404 |
if ($pending === 0) { |
| 405 |
wp_unschedule_event($scheduled, self::SWEEP_HOOK); |
| 406 |
} |
| 407 |
} |
| 408 |
} |
| 409 |
|
| 410 |
// ======================================== |
| 411 |
// ROLE / LOCAL-ROW HELPERS |
| 412 |
// ======================================== |
| 413 |
|
| 414 |
/** |
| 415 |
* Is this entry role-restricted right now? WP mode: the KB row's column. |
| 416 |
* Pinecone mode: the roles table keyed by base vector id. |
| 417 |
*/ |
| 418 |
private static function entry_is_restricted($source_url, $bot_id) { |
| 419 |
global $wpdb; |
| 420 |
|
| 421 |
$kb_table = $wpdb->prefix . 'mxchat_system_prompt_content'; |
| 422 |
$restriction = $wpdb->get_var($wpdb->prepare( |
| 423 |
"SELECT role_restriction FROM {$kb_table} WHERE source_url = %s LIMIT 1", |
| 424 |
$source_url |
| 425 |
)); |
| 426 |
if ($restriction !== null && $restriction !== '' && $restriction !== 'public') { |
| 427 |
return true; |
| 428 |
} |
| 429 |
|
| 430 |
$roles_table = $wpdb->prefix . 'mxchat_pinecone_roles'; |
| 431 |
if ($wpdb->get_var($wpdb->prepare('SHOW TABLES LIKE %s', $roles_table)) === $roles_table) { |
| 432 |
$restriction = $wpdb->get_var($wpdb->prepare( |
| 433 |
"SELECT role_restriction FROM {$roles_table} WHERE vector_id = %s LIMIT 1", |
| 434 |
self::entry_key($source_url) |
| 435 |
)); |
| 436 |
if ($restriction !== null && $restriction !== '' && $restriction !== 'public') { |
| 437 |
return true; |
| 438 |
} |
| 439 |
} |
| 440 |
return false; |
| 441 |
} |
| 442 |
|
| 443 |
/** |
| 444 |
* Reassemble one entry's full text from the local KB table (chunked rows |
| 445 |
* carry a JSON metadata prefix and are ordered by chunk_index). |
| 446 |
* |
| 447 |
* @return array|null {content, content_type, restricted} or null if absent. |
| 448 |
*/ |
| 449 |
private static function read_local_entry($source_url) { |
| 450 |
global $wpdb; |
| 451 |
$kb_table = $wpdb->prefix . 'mxchat_system_prompt_content'; |
| 452 |
|
| 453 |
$rows = $wpdb->get_results($wpdb->prepare( |
| 454 |
"SELECT article_content, content_type, role_restriction FROM {$kb_table} WHERE source_url = %s ORDER BY id ASC", |
| 455 |
$source_url |
| 456 |
)); |
| 457 |
if (empty($rows)) { |
| 458 |
return null; |
| 459 |
} |
| 460 |
|
| 461 |
$restricted = false; |
| 462 |
$content_type = 'content'; |
| 463 |
$parts = array(); |
| 464 |
foreach ($rows as $row) { |
| 465 |
if (!empty($row->role_restriction) && $row->role_restriction !== 'public') { |
| 466 |
$restricted = true; |
| 467 |
} |
| 468 |
$content_type = $row->content_type ?: $content_type; |
| 469 |
$parsed = MxChat_Chunker::parse_stored_chunk($row->article_content); |
| 470 |
$index = isset($parsed['metadata']['chunk_index']) ? (int) $parsed['metadata']['chunk_index'] : count($parts); |
| 471 |
$parts[$index] = $parsed['text']; |
| 472 |
} |
| 473 |
ksort($parts); |
| 474 |
|
| 475 |
return array( |
| 476 |
'content' => implode("\n\n", $parts), |
| 477 |
'content_type' => $content_type, |
| 478 |
'restricted' => $restricted, |
| 479 |
); |
| 480 |
} |
| 481 |
|
| 482 |
// ======================================== |
| 483 |
// INITIAL IMPORT (cron ticks + CLI) |
| 484 |
// ======================================== |
| 485 |
|
| 486 |
private static function default_state() { |
| 487 |
return array( |
| 488 |
'status' => 'idle', // idle|running|done|cancelled|error |
| 489 |
'store_id' => '', |
| 490 |
'bot_id' => 'default', |
| 491 |
'cursor' => 0, // MIN(id) of the last entry group processed |
| 492 |
'total' => 0, // distinct entries at kickoff |
| 493 |
'processed' => 0, |
| 494 |
'imported' => 0, |
| 495 |
'unchanged' => 0, |
| 496 |
'skipped_restricted' => 0, |
| 497 |
'skipped_no_identity' => 0, |
| 498 |
'failed' => 0, |
| 499 |
'last_error' => '', |
| 500 |
'started_at' => 0, |
| 501 |
'updated_at' => 0, |
| 502 |
); |
| 503 |
} |
| 504 |
|
| 505 |
public static function get_import_state() { |
| 506 |
$state = get_option(self::STATE_OPTION, array()); |
| 507 |
return array_merge(self::default_state(), is_array($state) ? $state : array()); |
| 508 |
} |
| 509 |
|
| 510 |
private static function save_import_state($state) { |
| 511 |
$state['updated_at'] = time(); |
| 512 |
update_option(self::STATE_OPTION, $state, false); |
| 513 |
} |
| 514 |
|
| 515 |
/** |
| 516 |
* Process up to $limit entry groups from the local KB table. Shared by the |
| 517 |
* cron tick and the CLI loop. Returns the updated state. |
| 518 |
*/ |
| 519 |
public function import_work($state, $limit, $deadline = null) { |
| 520 |
global $wpdb; |
| 521 |
$kb_table = $wpdb->prefix . 'mxchat_system_prompt_content'; |
| 522 |
|
| 523 |
$config = self::get_sync_config($state['bot_id']); |
| 524 |
if (!$config['enabled'] || $config['store_id'] !== $state['store_id']) { |
| 525 |
$state['status'] = 'error'; |
| 526 |
$state['last_error'] = __('Sync was disabled or the target store changed mid-import.', 'mxchat'); |
| 527 |
return $state; |
| 528 |
} |
| 529 |
|
| 530 |
// Entry = distinct source_url; cursor over MIN(id) keeps the scan |
| 531 |
// stable while rows are inserted/deleted around it. |
| 532 |
$groups = $wpdb->get_results($wpdb->prepare( |
| 533 |
"SELECT source_url, MIN(id) AS mid FROM {$kb_table} |
| 534 |
GROUP BY source_url HAVING mid > %d ORDER BY mid ASC LIMIT %d", |
| 535 |
(int) $state['cursor'], $limit |
| 536 |
)); |
| 537 |
|
| 538 |
if (empty($groups)) { |
| 539 |
$state['status'] = 'done'; |
| 540 |
return $state; |
| 541 |
} |
| 542 |
|
| 543 |
$batch_file_ids = array(); |
| 544 |
|
| 545 |
foreach ($groups as $group) { |
| 546 |
if ($deadline !== null && microtime(true) > $deadline) { |
| 547 |
break; // budget spent — cursor already reflects finished work |
| 548 |
} |
| 549 |
|
| 550 |
$state['cursor'] = (int) $group->mid; |
| 551 |
$state['processed']++; |
| 552 |
|
| 553 |
$source_url = (string) $group->source_url; |
| 554 |
if (!self::has_stable_identity($source_url)) { |
| 555 |
$state['skipped_no_identity']++; |
| 556 |
continue; |
| 557 |
} |
| 558 |
|
| 559 |
$entry = self::read_local_entry($source_url); |
| 560 |
if ($entry === null) { |
| 561 |
continue; // deleted between the group scan and now |
| 562 |
} |
| 563 |
if ($entry['restricted']) { |
| 564 |
$state['skipped_restricted']++; |
| 565 |
continue; |
| 566 |
} |
| 567 |
|
| 568 |
$result = self::upsert_file( |
| 569 |
$config, |
| 570 |
self::entry_key($source_url), |
| 571 |
$source_url, |
| 572 |
$entry['content'], |
| 573 |
$state['bot_id'], |
| 574 |
$entry['content_type'], |
| 575 |
true // defer attach — batched below |
| 576 |
); |
| 577 |
|
| 578 |
if (is_wp_error($result)) { |
| 579 |
$state['failed']++; |
| 580 |
$state['last_error'] = $result->get_error_message(); |
| 581 |
} elseif ($result === 'unchanged') { |
| 582 |
$state['unchanged']++; |
| 583 |
} else { |
| 584 |
$batch_file_ids[] = $result; |
| 585 |
$state['imported']++; |
| 586 |
} |
| 587 |
|
| 588 |
usleep(50000); // 0.05s between uploads, same pacing as Pinecone ops |
| 589 |
} |
| 590 |
|
| 591 |
// Attach everything this pass uploaded with one file_batches call, |
| 592 |
// then promote the mappings to live. Until that promotion, a killed |
| 593 |
// run's uploads read as pending_attach and are re-imported on resume. |
| 594 |
if (!empty($batch_file_ids)) { |
| 595 |
$table = $wpdb->prefix . self::TABLE; |
| 596 |
$batch = self::api_create_file_batch($config['api_key'], $config['store_id'], $batch_file_ids); |
| 597 |
if (is_wp_error($batch)) { |
| 598 |
// Files exist but aren't in the store: condemn the mappings so |
| 599 |
// the sweeper reclaims the uploads, and count the entries as |
| 600 |
// failed — a re-run re-imports them (no live hash rows remain). |
| 601 |
foreach ($batch_file_ids as $fid) { |
| 602 |
$wpdb->update($table, array('status' => 'pending_delete'), array('file_id' => $fid, 'status' => 'pending_attach'), array('%s'), array('%s', '%s')); |
| 603 |
} |
| 604 |
$state['failed'] += count($batch_file_ids); |
| 605 |
$state['imported'] -= count($batch_file_ids); |
| 606 |
$state['last_error'] = 'file_batches: ' . $batch->get_error_message(); |
| 607 |
} else { |
| 608 |
foreach ($batch_file_ids as $fid) { |
| 609 |
$wpdb->update($table, array('status' => 'live'), array('file_id' => $fid, 'status' => 'pending_attach'), array('%s'), array('%s', '%s')); |
| 610 |
} |
| 611 |
} |
| 612 |
} |
| 613 |
|
| 614 |
return $state; |
| 615 |
} |
| 616 |
|
| 617 |
public function run_import_tick() { |
| 618 |
if (get_transient(self::LOCK_TRANSIENT)) { |
| 619 |
return; // another tick is mid-flight |
| 620 |
} |
| 621 |
set_transient(self::LOCK_TRANSIENT, 1, 5 * MINUTE_IN_SECONDS); |
| 622 |
|
| 623 |
$state = self::get_import_state(); |
| 624 |
if ($state['status'] !== 'running') { |
| 625 |
delete_transient(self::LOCK_TRANSIENT); |
| 626 |
return; |
| 627 |
} |
| 628 |
|
| 629 |
$state = $this->import_work($state, self::IMPORT_BATCH_SIZE, microtime(true) + self::TICK_TIME_BUDGET); |
| 630 |
self::save_import_state($state); |
| 631 |
delete_transient(self::LOCK_TRANSIENT); |
| 632 |
|
| 633 |
if ($state['status'] === 'running') { |
| 634 |
wp_schedule_single_event(time() + 2, self::TICK_HOOK); |
| 635 |
} |
| 636 |
} |
| 637 |
|
| 638 |
// ======================================== |
| 639 |
// AJAX (admin UI) |
| 640 |
// ======================================== |
| 641 |
|
| 642 |
private function ajax_guard() { |
| 643 |
check_ajax_referer('mxchat_admin_nonce', 'nonce'); |
| 644 |
if (!current_user_can('manage_options')) { |
| 645 |
wp_send_json_error(array('message' => __('Permission denied.', 'mxchat'))); |
| 646 |
exit; |
| 647 |
} |
| 648 |
} |
| 649 |
|
| 650 |
public function ajax_import_start() { |
| 651 |
$this->ajax_guard(); |
| 652 |
|
| 653 |
$config = self::get_sync_config('default'); |
| 654 |
if (!$config['enabled']) { |
| 655 |
wp_send_json_error(array('message' => __('Enable sync and set a target Vector Store ID (and OpenAI API key) first, then save settings.', 'mxchat'))); |
| 656 |
} |
| 657 |
|
| 658 |
$state = self::get_import_state(); |
| 659 |
if ($state['status'] === 'running') { |
| 660 |
wp_send_json_error(array('message' => __('An import is already running.', 'mxchat'))); |
| 661 |
} |
| 662 |
|
| 663 |
// Validate the store really exists before burning uploads on a typo. |
| 664 |
$store = self::api_get_store($config['api_key'], $config['store_id']); |
| 665 |
if (is_wp_error($store)) { |
| 666 |
wp_send_json_error(array('message' => sprintf(__('Vector Store check failed: %s', 'mxchat'), $store->get_error_message()))); |
| 667 |
} |
| 668 |
|
| 669 |
global $wpdb; |
| 670 |
$kb_table = $wpdb->prefix . 'mxchat_system_prompt_content'; |
| 671 |
$total = (int) $wpdb->get_var("SELECT COUNT(DISTINCT source_url) FROM {$kb_table}"); |
| 672 |
|
| 673 |
$state = self::default_state(); |
| 674 |
$state['status'] = 'running'; |
| 675 |
$state['store_id'] = $config['store_id']; |
| 676 |
$state['total'] = $total; |
| 677 |
$state['started_at'] = time(); |
| 678 |
self::save_import_state($state); |
| 679 |
|
| 680 |
wp_schedule_single_event(time() + 1, self::TICK_HOOK); |
| 681 |
spawn_cron(); |
| 682 |
|
| 683 |
wp_send_json_success(array('state' => self::get_import_state())); |
| 684 |
} |
| 685 |
|
| 686 |
public function ajax_import_status() { |
| 687 |
$this->ajax_guard(); |
| 688 |
$state = self::get_import_state(); |
| 689 |
// A running import whose last heartbeat is stale has lost its cron |
| 690 |
// chain (server restart, cron blocked) — surface a Resume affordance |
| 691 |
// instead of a forever-spinner. |
| 692 |
$state['stalled'] = ($state['status'] === 'running' && (time() - (int) $state['updated_at']) > 120); |
| 693 |
wp_send_json_success(array('state' => $state, 'mapped' => self::mapped_file_count())); |
| 694 |
} |
| 695 |
|
| 696 |
public function ajax_import_resume() { |
| 697 |
$this->ajax_guard(); |
| 698 |
$state = self::get_import_state(); |
| 699 |
if ($state['status'] !== 'running') { |
| 700 |
wp_send_json_error(array('message' => __('No interrupted import to resume.', 'mxchat'))); |
| 701 |
} |
| 702 |
if (!wp_next_scheduled(self::TICK_HOOK)) { |
| 703 |
wp_schedule_single_event(time() + 1, self::TICK_HOOK); |
| 704 |
} |
| 705 |
spawn_cron(); |
| 706 |
wp_send_json_success(array('state' => self::get_import_state())); |
| 707 |
} |
| 708 |
|
| 709 |
public function ajax_import_cancel() { |
| 710 |
$this->ajax_guard(); |
| 711 |
$state = self::get_import_state(); |
| 712 |
if ($state['status'] === 'running') { |
| 713 |
$state['status'] = 'cancelled'; |
| 714 |
self::save_import_state($state); |
| 715 |
} |
| 716 |
wp_send_json_success(array('state' => self::get_import_state())); |
| 717 |
} |
| 718 |
|
| 719 |
public static function mapped_file_count() { |
| 720 |
global $wpdb; |
| 721 |
$table = $wpdb->prefix . self::TABLE; |
| 722 |
if ($wpdb->get_var($wpdb->prepare('SHOW TABLES LIKE %s', $table)) !== $table) { |
| 723 |
return 0; |
| 724 |
} |
| 725 |
return (int) $wpdb->get_var("SELECT COUNT(*) FROM {$table} WHERE status = 'live'"); |
| 726 |
} |
| 727 |
|
| 728 |
// ======================================== |
| 729 |
// WP-CLI |
| 730 |
// ======================================== |
| 731 |
|
| 732 |
/** |
| 733 |
* Import the local knowledge base into the configured Vector Store. |
| 734 |
* |
| 735 |
* ## OPTIONS |
| 736 |
* |
| 737 |
* [--restart] |
| 738 |
* : Discard prior import progress and start from the beginning (already- |
| 739 |
* mirrored, unchanged entries are still skipped via the mapping table). |
| 740 |
* |
| 741 |
* ## EXAMPLES |
| 742 |
* |
| 743 |
* wp mxchat vectorstore-import |
| 744 |
*/ |
| 745 |
public function cli_import($args, $assoc_args) { |
| 746 |
$config = self::get_sync_config('default'); |
| 747 |
if (!$config['enabled']) { |
| 748 |
WP_CLI::error('Vector Store sync is not enabled (needs the sync toggle, a store ID, and the OpenAI API key).'); |
| 749 |
} |
| 750 |
|
| 751 |
$store = self::api_get_store($config['api_key'], $config['store_id']); |
| 752 |
if (is_wp_error($store)) { |
| 753 |
WP_CLI::error('Vector Store check failed: ' . $store->get_error_message()); |
| 754 |
} |
| 755 |
|
| 756 |
$state = self::get_import_state(); |
| 757 |
if (!empty($assoc_args['restart']) || $state['status'] !== 'running') { |
| 758 |
global $wpdb; |
| 759 |
$kb_table = $wpdb->prefix . 'mxchat_system_prompt_content'; |
| 760 |
$state = self::default_state(); |
| 761 |
$state['status'] = 'running'; |
| 762 |
$state['store_id'] = $config['store_id']; |
| 763 |
$state['total'] = (int) $wpdb->get_var("SELECT COUNT(DISTINCT source_url) FROM {$kb_table}"); |
| 764 |
$state['started_at'] = time(); |
| 765 |
self::save_import_state($state); |
| 766 |
} else { |
| 767 |
WP_CLI::log(sprintf('Resuming interrupted import at %d/%d.', $state['processed'], $state['total'])); |
| 768 |
} |
| 769 |
|
| 770 |
while ($state['status'] === 'running') { |
| 771 |
$state = $this->import_work($state, self::IMPORT_BATCH_SIZE); |
| 772 |
self::save_import_state($state); |
| 773 |
WP_CLI::log(sprintf( |
| 774 |
'%d/%d processed — %d uploaded, %d unchanged, %d restricted-skipped, %d no-identity, %d failed', |
| 775 |
$state['processed'], $state['total'], $state['imported'], $state['unchanged'], |
| 776 |
$state['skipped_restricted'], $state['skipped_no_identity'], $state['failed'] |
| 777 |
)); |
| 778 |
} |
| 779 |
|
| 780 |
if ($state['status'] === 'done') { |
| 781 |
WP_CLI::success(sprintf( |
| 782 |
'Import complete: %d uploaded, %d unchanged, %d skipped (restricted), %d skipped (no identity), %d failed.', |
| 783 |
$state['imported'], $state['unchanged'], $state['skipped_restricted'], $state['skipped_no_identity'], $state['failed'] |
| 784 |
)); |
| 785 |
if ($state['failed'] > 0) { |
| 786 |
WP_CLI::warning('Last error: ' . $state['last_error'] . ' — re-run the command to retry failed entries.'); |
| 787 |
} |
| 788 |
} else { |
| 789 |
WP_CLI::error('Import ended with status "' . $state['status'] . '": ' . $state['last_error']); |
| 790 |
} |
| 791 |
} |
| 792 |
|
| 793 |
// ======================================== |
| 794 |
// OPENAI API LAYER |
| 795 |
// ======================================== |
| 796 |
|
| 797 |
private static function api_headers($api_key, $json = true) { |
| 798 |
$headers = array( |
| 799 |
'Authorization' => 'Bearer ' . $api_key, |
| 800 |
'OpenAI-Beta' => 'assistants=v2', |
| 801 |
); |
| 802 |
if ($json) { |
| 803 |
$headers['Content-Type'] = 'application/json'; |
| 804 |
} |
| 805 |
return $headers; |
| 806 |
} |
| 807 |
|
| 808 |
private static function api_error($context, $response) { |
| 809 |
if (is_wp_error($response)) { |
| 810 |
return new WP_Error('vectorstore_request', $context . ': ' . $response->get_error_message()); |
| 811 |
} |
| 812 |
$code = wp_remote_retrieve_response_code($response); |
| 813 |
$body = json_decode(wp_remote_retrieve_body($response), true); |
| 814 |
$detail = $body['error']['message'] ?? ('HTTP ' . $code); |
| 815 |
return new WP_Error('vectorstore_api', $context . ': ' . $detail, array('status' => $code)); |
| 816 |
} |
| 817 |
|
| 818 |
public static function api_get_store($api_key, $store_id) { |
| 819 |
$response = wp_remote_get('https://api.openai.com/v1/vector_stores/' . rawurlencode($store_id), array( |
| 820 |
'headers' => self::api_headers($api_key), |
| 821 |
'timeout' => 30, |
| 822 |
)); |
| 823 |
if (is_wp_error($response) || wp_remote_retrieve_response_code($response) !== 200) { |
| 824 |
return self::api_error('vector store lookup', $response); |
| 825 |
} |
| 826 |
return json_decode(wp_remote_retrieve_body($response), true); |
| 827 |
} |
| 828 |
|
| 829 |
/** |
| 830 |
* POST /v1/files (purpose=assistants), multipart built by hand — WP has no |
| 831 |
* native multipart support in wp_remote_post. |
| 832 |
* |
| 833 |
* @return string|WP_Error file id |
| 834 |
*/ |
| 835 |
public static function api_upload_file($api_key, $filename, $content) { |
| 836 |
$boundary = 'mxchatvs' . wp_generate_password(16, false); |
| 837 |
|
| 838 |
$body = '--' . $boundary . "\r\n"; |
| 839 |
$body .= "Content-Disposition: form-data; name=\"purpose\"\r\n\r\n"; |
| 840 |
$body .= "assistants\r\n"; |
| 841 |
$body .= '--' . $boundary . "\r\n"; |
| 842 |
$body .= 'Content-Disposition: form-data; name="file"; filename="' . $filename . "\"\r\n"; |
| 843 |
$body .= "Content-Type: text/plain\r\n\r\n"; |
| 844 |
$body .= $content . "\r\n"; |
| 845 |
$body .= '--' . $boundary . "--\r\n"; |
| 846 |
|
| 847 |
$response = wp_remote_post('https://api.openai.com/v1/files', array( |
| 848 |
'headers' => array( |
| 849 |
'Authorization' => 'Bearer ' . $api_key, |
| 850 |
'Content-Type' => 'multipart/form-data; boundary=' . $boundary, |
| 851 |
), |
| 852 |
'body' => $body, |
| 853 |
'timeout' => 60, |
| 854 |
)); |
| 855 |
|
| 856 |
if (is_wp_error($response) || wp_remote_retrieve_response_code($response) !== 200) { |
| 857 |
return self::api_error('file upload', $response); |
| 858 |
} |
| 859 |
$data = json_decode(wp_remote_retrieve_body($response), true); |
| 860 |
if (empty($data['id'])) { |
| 861 |
return new WP_Error('vectorstore_api', 'file upload: response carried no file id'); |
| 862 |
} |
| 863 |
return $data['id']; |
| 864 |
} |
| 865 |
|
| 866 |
public static function api_attach_file($api_key, $store_id, $file_id) { |
| 867 |
$response = wp_remote_post('https://api.openai.com/v1/vector_stores/' . rawurlencode($store_id) . '/files', array( |
| 868 |
'headers' => self::api_headers($api_key), |
| 869 |
'body' => wp_json_encode(array('file_id' => $file_id)), |
| 870 |
'timeout' => 30, |
| 871 |
)); |
| 872 |
if (is_wp_error($response) || wp_remote_retrieve_response_code($response) !== 200) { |
| 873 |
return self::api_error('file attach', $response); |
| 874 |
} |
| 875 |
return true; |
| 876 |
} |
| 877 |
|
| 878 |
public static function api_create_file_batch($api_key, $store_id, $file_ids) { |
| 879 |
$response = wp_remote_post('https://api.openai.com/v1/vector_stores/' . rawurlencode($store_id) . '/file_batches', array( |
| 880 |
'headers' => self::api_headers($api_key), |
| 881 |
'body' => wp_json_encode(array('file_ids' => array_values($file_ids))), |
| 882 |
'timeout' => 60, |
| 883 |
)); |
| 884 |
if (is_wp_error($response) || wp_remote_retrieve_response_code($response) !== 200) { |
| 885 |
return self::api_error('file batch', $response); |
| 886 |
} |
| 887 |
return json_decode(wp_remote_retrieve_body($response), true); |
| 888 |
} |
| 889 |
|
| 890 |
public static function api_detach_file($api_key, $store_id, $file_id) { |
| 891 |
$response = wp_remote_request('https://api.openai.com/v1/vector_stores/' . rawurlencode($store_id) . '/files/' . rawurlencode($file_id), array( |
| 892 |
'method' => 'DELETE', |
| 893 |
'headers' => self::api_headers($api_key), |
| 894 |
'timeout' => 30, |
| 895 |
)); |
| 896 |
$code = is_wp_error($response) ? 0 : wp_remote_retrieve_response_code($response); |
| 897 |
if (is_wp_error($response) || ($code !== 200 && $code !== 404)) { |
| 898 |
return self::api_error('file detach', $response); |
| 899 |
} |
| 900 |
return true; |
| 901 |
} |
| 902 |
|
| 903 |
public static function api_delete_file($api_key, $file_id) { |
| 904 |
$response = wp_remote_request('https://api.openai.com/v1/files/' . rawurlencode($file_id), array( |
| 905 |
'method' => 'DELETE', |
| 906 |
'headers' => self::api_headers($api_key), |
| 907 |
'timeout' => 30, |
| 908 |
)); |
| 909 |
$code = is_wp_error($response) ? 0 : wp_remote_retrieve_response_code($response); |
| 910 |
if (is_wp_error($response) || ($code !== 200 && $code !== 404)) { |
| 911 |
return self::api_error('file delete', $response); |
| 912 |
} |
| 913 |
return true; |
| 914 |
} |
| 915 |
|
| 916 |
private static function log($message) { |
| 917 |
if (class_exists('MxChat_Admin') && method_exists('MxChat_Admin', 'mxchat_log_debug')) { |
| 918 |
MxChat_Admin::mxchat_log_debug('vectorstore_sync', $message); |
| 919 |
} |
| 920 |
} |
| 921 |
} |
| 922 |
|
