thinkrank / includes / ai / class-brand-visibility-runner.php

class-brand-visibility-runner.php in ThinkRank AI SEO – AI SEO Plugin for WordPress: Schema, XML Sitemaps, Meta Tags, Search Console & Local SEO 1.28.0, at includes/ai/class-brand-visibility-runner.php

797 lines 27.2 KB

Raw Download Zip

1	<?php
2	/**
3	* Brand Visibility v2 — queued run engine.
4	*
5	* A full analysis is queries × platforms × samples, which is routinely 100+
6	* AI calls. v1 ran that batch synchronously inside a REST request: on any real
7	* configuration it would exhaust max_execution_time, and an interruption lost
8	* everything. This engine persists every probe as a task row and drains them
9	* off-request in cron ticks with a wall-clock budget, so a run makes forward
10	* progress, survives interruption, and reports honest progress while it works.
11	*
12	* The shape mirrors the standard batch/sweeper/finalizer trio:
13	* start() — snapshot config, fan out task rows, kick the first tick
14	* tick() — process what fits in the budget, then reschedule if unfinished
15	* sweep() — return tasks stuck in `running` (a killed worker) to `pending`
16	* finalize() — aggregate task rows into the run's results, once
17	*
18	* @package ThinkRank\AI
19	* @since 1.28.0
20	*/
21
22	declare(strict_types=1);
23
24	namespace ThinkRank\AI;
25
26	use ThinkRank\Core\Plan_Config;
27	use ThinkRank\Core\Settings;
28
29	if (!defined('ABSPATH')) {
30	exit;
31	}
32
33	/**
34	* Creates and drains Brand Visibility analysis runs.
35	*/
36	class Brand_Visibility_Runner {
37
38	/**
39	* Cron hook that drains pending tasks.
40	*/
41	public const TICK_HOOK = 'thinkrank_bv_tick';
42
43	/**
44	* Recurring safety net that restarts a stalled drain.
45	*/
46	public const WATCHDOG_HOOK = 'thinkrank_bv_watchdog';
47
48	/**
49	* Interval slug for the watchdog.
50	*/
51	public const WATCHDOG_INTERVAL = 'thinkrank_bv_five_minutes';
52
53	/**
54	* Wall-clock budget for one tick, seconds. Comfortably inside the 30s
55	* PHP default while still landing several probes per tick.
56	*/
57	private const TICK_BUDGET = 20;
58
59	/**
60	* Hard ceiling on probes per tick, so a fast provider can't run a tick
61	* long enough to collide with the next scheduled one.
62	*/
63	private const TICK_MAX_TASKS = 12;
64
65	/**
66	* Attempts before a task is abandoned.
67	*/
68	private const MAX_ATTEMPTS = 2;
69
70	/**
71	* Minutes after which a `running` task is presumed dead.
72	*/
73	private const STUCK_AFTER_MINUTES = 10;
74
75	/**
76	* Output-token budget per probe. Reasoning-safe: see 1942a44/6094d9d —
77	* GPT-5-family models spend output tokens on hidden reasoning first and
78	* return empty text if the budget is too small.
79	*/
80	private const ANSWER_TOKENS = 8000;
81
82	/**
83	* Settings accessor.
84	*
85	* @var Settings
86	*/
87	private Settings $settings;
88
89	/**
90	* Provider resolver.
91	*
92	* @var Brand_Visibility_Providers
93	*/
94	private Brand_Visibility_Providers $providers;
95
96	/**
97	* Constructor.
98	*
99	* @param Settings\|null $settings Settings.
100	* @param Brand_Visibility_Providers\|null $providers Provider resolver.
101	*/
102	public function __construct(?Settings $settings = null, ?Brand_Visibility_Providers $providers = null) {
103	$this->settings = $settings ?? Settings::instance();
104	$this->providers = $providers ?? new Brand_Visibility_Providers($this->settings);
105	}
106
107	/**
108	* Register the cron listener.
109	*
110	* @return void
111	*/
112	public function init(): void {
113	add_action(self::TICK_HOOK, [$this, 'tick']);
114	add_action(self::WATCHDOG_HOOK, [$this, 'watchdog']);
115	add_filter('cron_schedules', [self::class, 'add_cron_interval']); // phpcs:ignore WordPress.WP.CronInterval.ChangeDetected
116	}
117
118	/**
119	* Register the watchdog's five-minute interval.
120	*
121	* @param array $schedules Registered cron schedules.
122	* @return array
123	*/
124	public static function add_cron_interval(array $schedules): array {
125	$schedules[self::WATCHDOG_INTERVAL] ??= [
126	'interval' => 5 * MINUTE_IN_SECONDS,
127	'display' => __('Every five minutes (ThinkRank Brand Visibility)', 'thinkrank'),
128	];
129
130	return $schedules;
131	}
132
133	/**
134	* Recover a run whose drain stopped.
135	*
136	* `tick()` reschedules itself, which is enough while ticks complete — but
137	* a tick killed by a fatal, an OOM or a worker timeout never reaches that
138	* line, and the run then sits at whatever percentage it reached forever.
139	* This recurring hook re-sweeps abandoned claims, closes runs whose tasks
140	* all resolved, and re-arms the drain. It unschedules itself once there is
141	* no outstanding work, so it costs nothing on an idle site.
142	*
143	* @return void
144	*/
145	public function watchdog(): void {
146	$this->sweep();
147	$this->finalize_completed_runs();
148
149	if ($this->has_outstanding_tasks()) {
150	$this->schedule_tick();
151	return;
152	}
153
154	self::unschedule_watchdog();
155	}
156
157	/**
158	* Arm the watchdog while a run is in flight.
159	*
160	* @return void
161	*/
162	private function schedule_watchdog(): void {
163	if (!wp_next_scheduled(self::WATCHDOG_HOOK)) {
164	wp_schedule_event(time() + (5 * MINUTE_IN_SECONDS), self::WATCHDOG_INTERVAL, self::WATCHDOG_HOOK);
165	}
166	}
167
168	/**
169	* Disarm the watchdog.
170	*
171	* @return void
172	*/
173	public static function unschedule_watchdog(): void {
174	wp_clear_scheduled_hook(self::WATCHDOG_HOOK);
175	}
176
177	/**
178	* Runs table name.
179	*
180	* @return string
181	*/
182	public static function runs_table(): string {
183	global $wpdb;
184	return $wpdb->prefix . 'thinkrank_bv_runs';
185	}
186
187	/**
188	* Tasks table name.
189	*
190	* @return string
191	*/
192	public static function tasks_table(): string {
193	global $wpdb;
194	return $wpdb->prefix . 'thinkrank_bv_tasks';
195	}
196
197	/**
198	* Expand a configuration into the flat list of probes it implies.
199	*
200	* Pure and public so the wizard's cost preview and the runner agree by
201	* construction — the number shown before spending is the number of calls
202	* actually made.
203	*
204	* @param array $config queries[{text,type}], platforms[], samples.
205	* @return array<int, array{query_text: string, query_type: string, platform: string, sample_index: int}>
206	*/
207	public static function plan_tasks(array $config): array {
208	$queries = $config['queries'] ?? [];
209	$platforms = $config['platforms'] ?? [];
210	$samples = max(1, (int) ($config['samples'] ?? 1));
211
212	$tasks = [];
213
214	foreach ($queries as $query) {
215	$text = trim((string) ($query['text'] ?? ''));
216	if ('' === $text) {
217	continue;
218	}
219	$type = (string) ($query['type'] ?? 'branded');
220	if (!in_array($type, Brand_Visibility_Scorer::QUERY_TYPES, true)) {
221	$type = 'branded';
222	}
223
224	foreach ($platforms as $platform) {
225	for ($i = 0; $i < $samples; $i++) {
226	$tasks[] = [
227	'query_text' => $text,
228	'query_type' => $type,
229	'platform' => (string) $platform,
230	'sample_index' => $i,
231	];
232	}
233	}
234	}
235
236	return $tasks;
237	}
238
239	/**
240	* Start a run: snapshot the config, fan out tasks, schedule the drain.
241	*
242	* @param array $config Full run configuration.
243	* @return int New run id.
244	* @throws \Exception When the config yields no work.
245	*/
246	public function start(array $config): int {
247	global $wpdb;
248
249	$planned = self::plan_tasks($config);
250	if (empty($planned)) {
251	throw new \Exception(esc_html__('Nothing to run: add at least one question and one AI platform.', 'thinkrank'));
252	}
253
254	// Pre-flight: repair a missing schema rather than failing with a bare
255	// "could not start" (#270). Self-healing beats an error message here —
256	// the user can't act on "a table is missing" anyway.
257	$this->ensure_schema();
258
259	$wpdb->insert(
260	self::runs_table(),
261	[
262	'status' => 'running',
263	'started_at' => current_time('mysql'),
264	'tasks_total' => count($planned),
265	'tasks_done' => 0,
266	'config' => wp_json_encode($config),
267	],
268	['%s', '%s', '%d', '%d', '%s']
269	);
270
271	$run_id = (int) $wpdb->insert_id;
272	if ($run_id <= 0) {
273	throw new \Exception(esc_html__('Could not start the analysis run.', 'thinkrank'));
274	}
275
276	foreach ($planned as $task) {
277	$wpdb->insert(
278	self::tasks_table(),
279	[
280	'run_id' => $run_id,
281	'query_text' => $task['query_text'],
282	'query_type' => $task['query_type'],
283	'platform' => $task['platform'],
284	'sample_index' => $task['sample_index'],
285	'status' => 'pending',
286	'updated_at' => current_time('mysql'),
287	],
288	['%d', '%s', '%s', '%s', '%d', '%s', '%s']
289	);
290	}
291
292	$this->schedule_tick();
293	$this->schedule_watchdog();
294
295	return $run_id;
296	}
297
298	/**
299	* Make sure this feature's tables exist before a run depends on them.
300	*
301	* Normally the schema upgrade path has already handled it; this is the
302	* backstop for an install whose stored db_version matched while a table
303	* was absent — the exact state that made Brand Visibility dead on arrival
304	* in #270. dbDelta creates only what's missing, so re-running is cheap and
305	* safe.
306	*
307	* @since 1.28.0
308	*
309	* @return void
310	* @throws \Exception When the tables still cannot be created.
311	*/
312	private function ensure_schema(): void {
313	global $wpdb;
314
315	$table = self::runs_table();
316
317	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- one-off existence probe before a run.
318	if ($wpdb->get_var($wpdb->prepare('SHOW TABLES LIKE %s', $table))) {
319	return;
320	}
321
322	(new \ThinkRank\Database\Database_Schema())->create_tables();
323
324	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching -- verify the repair worked.
325	if (!$wpdb->get_var($wpdb->prepare('SHOW TABLES LIKE %s', $table))) {
326	throw new \Exception(esc_html__(
327	'Brand Visibility storage is missing and could not be created. Please deactivate and reactivate ThinkRank, then try again.',
328	'thinkrank'
329	));
330	}
331	}
332
333	/**
334	* Process a slice of pending work, then reschedule while work remains.
335	*
336	* @return void
337	*/
338	public function tick(): void {
339	$this->sweep();
340
341	$started = microtime(true);
342	$handled = 0;
343
344	while ($handled < self::TICK_MAX_TASKS && (microtime(true) - $started) < self::TICK_BUDGET) {
345	$task = $this->claim_next_task();
346	if (null === $task) {
347	break;
348	}
349
350	$this->process_task($task);
351	$handled++;
352	}
353
354	// Finalize every run whose tasks are all resolved, then keep ticking
355	// only while work is genuinely outstanding.
356	$this->finalize_completed_runs();
357
358	if ($this->has_outstanding_tasks()) {
359	$this->schedule_tick();
360	return;
361	}
362
363	// Nothing left to drain — stop the recurring safety net.
364	self::unschedule_watchdog();
365	}
366
367	/**
368	* Return long-`running` tasks to the pool.
369	*
370	* A worker killed mid-probe would otherwise leave its task claimed
371	* forever, stalling the run at 97%.
372	*
373	* @return void
374	*/
375	public function sweep(): void {
376	global $wpdb;
377
378	// `updated_at` is written with current_time('mysql') — site-local, not
379	// UTC — so the cutoff must be site-local too. Comparing a UTC cutoff
380	// against local timestamps skews the sweep by the gmt_offset: east of
381	// UTC nothing is ever swept (a killed worker stalls the run forever),
382	// west of it live tasks get yanked back to `pending` mid-probe.
383	$cutoff = $this->local_time_ago(self::STUCK_AFTER_MINUTES * MINUTE_IN_SECONDS);
384	$table = self::tasks_table();
385
386	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- maintenance sweep on our own table.
387	$wpdb->query($wpdb->prepare(
388	"UPDATE `{$table}` SET status = 'pending' WHERE status = 'running' AND updated_at < %s",
389	$cutoff
390	));
391	}
392
393	/**
394	* Claim the next pending task (marking it `running` so parallel ticks
395	* can't both take it).
396	*
397	* @return array\|null Task row, or null when the queue is drained.
398	*/
399	private function claim_next_task(): ?array {
400	global $wpdb;
401
402	$table = self::tasks_table();
403
404	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- queue read on our own table.
405	$task = $wpdb->get_row("SELECT * FROM `{$table}` WHERE status = 'pending' ORDER BY id ASC LIMIT 1", ARRAY_A);
406
407	if (!$task) {
408	return null;
409	}
410
411	// Conditional UPDATE is the claim: if another tick got there first the
412	// affected-row count is 0 and we simply skip this one.
413	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- claim on our own table.
414	$claimed = $wpdb->query($wpdb->prepare(
415	"UPDATE `{$table}` SET status = 'running', attempts = attempts + 1, updated_at = %s WHERE id = %d AND status = 'pending'",
416	current_time('mysql'),
417	(int) $task['id']
418	));
419
420	return $claimed ? $task : null;
421	}
422
423	/**
424	* Run one probe and record its verdict.
425	*
426	* @param array $task Task row.
427	* @return void
428	*/
429	private function process_task(array $task): void {
430	global $wpdb;
431
432	$run = $this->get_run((int) $task['run_id']);
433	$config = is_array($run['config'] ?? null) ? $run['config'] : [];
434
435	$brand = (string) ($config['brand'] ?? get_bloginfo('name'));
436	$variants = (array) ($config['variants'] ?? []);
437	$competitors = (array) ($config['competitors'] ?? []);
438	$host = (string) wp_parse_url(home_url(), PHP_URL_HOST);
439
440	try {
441	$client = $this->providers->client_for((string) $task['platform']);
442	$answer = $this->ask($client, (string) $task['query_text'], (string) $task['platform']);
443
444	if ('' === trim($answer)) {
445	throw new \Exception(esc_html__('The AI returned an empty answer.', 'thinkrank'));
446	}
447
448	$mentioned = Brand_Visibility_Scorer::is_mentioned($answer, $brand, $variants);
449	$sentiment = '';
450
451	// Sentiment costs an extra call, so only ask when the brand was
452	// actually named — an answer that ignored you has no opinion.
453	if ($mentioned && Plan_Config::can('brand_sentiment', 'ai_visibility')) {
454	$sentiment = $this->classify_sentiment($client, $answer, $brand);
455	}
456
457	$this->update_task((int) $task['id'], [
458	'status' => 'done',
459	'mentioned' => $mentioned ? 1 : 0,
460	'cited' => Brand_Visibility_Scorer::is_cited($answer, $host) ? 1 : 0,
461	'sentiment' => $sentiment,
462	'competitors' => wp_json_encode(Brand_Visibility_Scorer::competitors_in($answer, $competitors)),
463	'excerpt' => Brand_Visibility_Scorer::excerpt($answer, $brand, $variants),
464	'answer' => substr($answer, 0, 20000),
465	'error' => '',
466	'updated_at' => current_time('mysql'),
467	]);
468
469	$this->bump_run((int) $task['run_id'], 'tasks_done');
470	} catch (\Throwable $e) {
471	$retryable = (int) $task['attempts'] < self::MAX_ATTEMPTS;
472
473	$this->update_task((int) $task['id'], [
474	'status' => $retryable ? 'pending' : 'failed',
475	'error' => substr($e->getMessage(), 0, 500),
476	'updated_at' => current_time('mysql'),
477	]);
478
479	if (!$retryable) {
480	$this->bump_run((int) $task['run_id'], 'tasks_failed');
481	}
482	}
483	}
484
485	/**
486	* Ask one platform one question.
487	*
488	* @param object $client Provider client.
489	* @param string $question User-style question.
490	* @param string $platform Platform slug (for provider-specific options).
491	* @return string Answer text.
492	*/
493	private function ask($client, string $question, string $platform): string {
494	// A user-style question, not an SEO prompt: the point is to observe an
495	// ordinary answer, exactly as a shopper would receive it.
496	$prompt = sprintf(
497	"Answer the following question the way you would answer any user. Be specific and name relevant products, companies or websites where appropriate.\n\nQuestion: %s",
498	$question
499	);
500
501	$options = [
502	'max_tokens' => self::ANSWER_TOKENS,
503	'temperature' => 0.4,
504	];
505
506	// GPT-5 reasoning models will otherwise spend the whole budget on
507	// hidden reasoning and return empty text (see 6094d9d).
508	if ('chatgpt' === $platform) {
509	$options['reasoning_effort'] = 'minimal';
510	}
511
512	$response = $client->generate_completion($prompt, $options);
513
514	return $this->extract_text($response);
515	}
516
517	/**
518	* One-word sentiment of an answer toward the brand.
519	*
520	* @param object $client Provider client.
521	* @param string $answer Answer text.
522	* @param string $brand Brand name.
523	* @return string positive\|neutral\|negative, or '' when unclear.
524	*/
525	private function classify_sentiment($client, string $answer, string $brand): string {
526	try {
527	$prompt = sprintf(
528	"Classify the sentiment toward \"%s\" in the text below. Reply with exactly one word: positive, neutral, or negative.\n\nText:\n%s",
529	$brand,
530	substr($answer, 0, 4000)
531	);
532
533	$verdict = strtolower(trim($this->extract_text(
534	$client->generate_completion($prompt, ['max_tokens' => 2000, 'temperature' => 0])
535	)));
536
537	foreach (['positive', 'negative', 'neutral'] as $mood) {
538	if (false !== strpos($verdict, $mood)) {
539	return $mood;
540	}
541	}
542	} catch (\Throwable $e) {
543	// Sentiment is a nice-to-have; never fail a probe over it.
544	return '';
545	}
546
547	return '';
548	}
549
550	/**
551	* Pull assistant text out of any supported provider response shape.
552	*
553	* @param array $response Decoded provider response.
554	* @return string
555	*/
556	private function extract_text(array $response): string {
557	if (isset($response['choices'][0]['message']['content'])) {
558	$content = $response['choices'][0]['message']['content'];
559	if (is_array($content)) {
560	return implode(' ', array_map(
561	static fn($part) => is_array($part) ? (string) ($part['text'] ?? '') : (string) $part,
562	$content
563	));
564	}
565	return (string) $content;
566	}
567
568	if (isset($response['content'][0]['text'])) {
569	return (string) $response['content'][0]['text'];
570	}
571
572	if (isset($response['candidates'][0]['content']['parts'][0]['text'])) {
573	return (string) $response['candidates'][0]['content']['parts'][0]['text'];
574	}
575
576	if (isset($response['content']) && is_string($response['content'])) {
577	return $response['content'];
578	}
579
580	return '';
581	}
582
583	/**
584	* Aggregate and close any run whose tasks are all resolved.
585	*
586	* @return void
587	*/
588	public function finalize_completed_runs(): void {
589	global $wpdb;
590
591	$runs = self::runs_table();
592	$tasks = self::tasks_table();
593
594	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- read on our own tables.
595	$ids = $wpdb->get_col(
596	"SELECT r.id FROM `{$runs}` r
597	WHERE r.status = 'running'
598	AND NOT EXISTS (
599	SELECT 1 FROM `{$tasks}` t
600	WHERE t.run_id = r.id AND t.status IN ('pending', 'running')
601	)"
602	);
603
604	foreach ((array) $ids as $id) {
605	$this->finalize((int) $id);
606	}
607	}
608
609	/**
610	* Compute and store a run's aggregates.
611	*
612	* @param int $run_id Run id.
613	* @return void
614	*/
615	public function finalize(int $run_id): void {
616	global $wpdb;
617
618	$run = $this->get_run($run_id);
619	$config = is_array($run['config'] ?? null) ? $run['config'] : [];
620	$tasks = $this->get_tasks($run_id);
621	$results = Brand_Visibility_Scorer::aggregate($tasks, $config);
622
623	$done = count(array_filter($tasks, static fn(array $t): bool => 'done' === $t['status']));
624	$failed = count(array_filter($tasks, static fn(array $t): bool => 'failed' === $t['status']));
625
626	// A run where every probe failed is a FAILED run, not a run reporting
627	// zero visibility — the distinction v1 got wrong.
628	$status = (0 === $done && $failed > 0) ? 'failed' : 'complete';
629
630	$wpdb->update(
631	self::runs_table(),
632	[
633	'status' => $status,
634	'finished_at' => current_time('mysql'),
635	'tasks_done' => $done,
636	'tasks_failed' => $failed,
637	'results' => wp_json_encode($results),
638	'error' => 'failed' === $status
639	? (string) ($tasks[0]['error'] ?? __('Every probe failed.', 'thinkrank'))
640	: '',
641	],
642	['id' => $run_id],
643	['%s', '%s', '%d', '%d', '%s', '%s'],
644	['%d']
645	);
646	}
647
648	/**
649	* A run with its config/results decoded.
650	*
651	* @param int $run_id Run id.
652	* @return array
653	*/
654	public function get_run(int $run_id): array {
655	global $wpdb;
656
657	$table = self::runs_table();
658
659	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- read on our own table.
660	$row = $wpdb->get_row($wpdb->prepare("SELECT * FROM `{$table}` WHERE id = %d", $run_id), ARRAY_A);
661
662	if (!$row) {
663	return [];
664	}
665
666	$row['config'] = json_decode((string) $row['config'], true) ?: [];
667	$row['results'] = json_decode((string) $row['results'], true) ?: [];
668
669	return $row;
670	}
671
672	/**
673	* Task rows for a run.
674	*
675	* @param int $run_id Run id.
676	* @return array
677	*/
678	public function get_tasks(int $run_id): array {
679	global $wpdb;
680
681	$table = self::tasks_table();
682
683	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- read on our own table.
684	return (array) $wpdb->get_results(
685	$wpdb->prepare("SELECT * FROM `{$table}` WHERE run_id = %d ORDER BY id ASC", $run_id),
686	ARRAY_A
687	);
688	}
689
690	/**
691	* Recent runs, newest first (feeds the over-time chart).
692	*
693	* @param int $limit Max runs.
694	* @return array
695	*/
696	public function recent_runs(int $limit = 20): array {
697	global $wpdb;
698
699	$table = self::runs_table();
700	$limit = max(1, min(100, $limit));
701
702	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- read on our own table.
703	$rows = (array) $wpdb->get_results(
704	$wpdb->prepare("SELECT id, status, started_at, finished_at, tasks_total, tasks_done, tasks_failed, results FROM `{$table}` ORDER BY id DESC LIMIT %d", $limit),
705	ARRAY_A
706	);
707
708	return array_map(static function (array $row): array {
709	$row['results'] = json_decode((string) $row['results'], true) ?: [];
710	return $row;
711	}, $rows);
712	}
713
714	/**
715	* Whether any run still has work outstanding.
716	*
717	* @return bool
718	*/
719	private function has_outstanding_tasks(): bool {
720	global $wpdb;
721
722	$table = self::tasks_table();
723
724	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- read on our own table.
725	return (int) $wpdb->get_var("SELECT COUNT(*) FROM `{$table}` WHERE status IN ('pending','running')") > 0;
726	}
727
728	/**
729	* Schedule the next drain if one isn't already due.
730	*
731	* @return void
732	*/
733	private function schedule_tick(): void {
734	if (!wp_next_scheduled(self::TICK_HOOK)) {
735	wp_schedule_single_event(time() + 5, self::TICK_HOOK);
736	}
737
738	// Cron on a low-traffic site only fires on a visit; nudge it so a run
739	// started from the dashboard begins immediately.
740	spawn_cron();
741	}
742
743	/**
744	* Format a UTC timestamp as a site-local MySQL datetime.
745	*
746	* The datetime columns on this feature's tables are written with
747	* current_time('mysql'), so every comparison value has to be built in the
748	* same frame. Deriving it from current_time() rather than the gmt_offset
749	* option keeps it correct under DST — that option is only re-synced when
750	* the timezone is saved. WordPress runs PHP in UTC, so parsing the local
751	* string with strtotime() and re-formatting with gmdate() round-trips the
752	* same wall clock, minus the offset asked for.
753	*
754	* @param int $offset_seconds Seconds to subtract from "now".
755	* @return string Site-local `Y-m-d H:i:s`.
756	*/
757	private function local_time_ago(int $offset_seconds): string {
758	return gmdate('Y-m-d H:i:s', strtotime(current_time('mysql')) - $offset_seconds);
759	}
760
761	/**
762	* Update a task row.
763	*
764	* @param int $task_id Task id.
765	* @param array $data Column => value.
766	* @return void
767	*/
768	private function update_task(int $task_id, array $data): void {
769	global $wpdb;
770
771	$wpdb->update(self::tasks_table(), $data, ['id' => $task_id], null, ['%d']);
772	}
773
774	/**
775	* Increment a run counter.
776	*
777	* @param int $run_id Run id.
778	* @param string $column tasks_done\|tasks_failed.
779	* @return void
780	*/
781	private function bump_run(int $run_id, string $column): void {
782	global $wpdb;
783
784	if (!in_array($column, ['tasks_done', 'tasks_failed'], true)) {
785	return;
786	}
787
788	$table = self::runs_table();
789
790	// phpcs:ignore WordPress.DB.DirectDatabaseQuery.DirectQuery, WordPress.DB.DirectDatabaseQuery.NoCaching, WordPress.DB.PreparedSQL.InterpolatedNotPrepared -- counter bump on our own table; column whitelisted above.
791	$wpdb->query($wpdb->prepare(
792	"UPDATE `{$table}` SET {$column} = {$column} + 1 WHERE id = %d",
793	$run_id
794	));
795	}
796	}
797