mlsimport / includes / mlsimport-telemetry-connections.php

mlsimport-telemetry-connections.php in MLSImport: IDX Plugin & MLS Plugin for Real Estate Listings trunk, at includes/mlsimport-telemetry-connections.php

240 lines 11.1 KB

Raw Download Zip

1	<?php
2	/**
3	* MLSImport Per-Connection Telemetry (issue #283, decision #272).
4	*
5	* WHY THIS FILE EXISTS
6	* --------------------
7	* With multi-MLS support (connections keyed by mls_id) the daily heartbeat
8	* must report per-MLS health so the portal can tell WHICH connection is
9	* failing while the others keep syncing. Decision #272: the heartbeat stays
10	* ONE aggregated daily payload; it gains a `connections` array with one full
11	* health entry per registered connection, while install-wide facts stay
12	* global and the legacy singular fields keep being sent.
13	*
14	* This file owns the per-connection half of that:
15	* - the per-connection sync-stamp writer (success time / failure time+code
16	* maps on mlsimport_telemetry_state), shared by the request choke point
17	* and the cron connection gate so both write the maps the same one way,
18	* - the workload gatherer (bound-task / paused-task / listing counts per
19	* connection, resolved through the #277 cron gate and the #278
20	* provenance stamp),
21	* - the PURE payload-assembly seam that turns registry records + telemetry
22	* state + the gathered workload into the `connections` array.
23	*
24	* The counter buckets themselves ('daily_mls' nested per mls_id) are written
25	* by mlsimport_telemetry_flush() in mlsimport-telemetry.php — bump callers
26	* pass the connection id they already have in hand from the task binding.
27	*
28	* @since 7.2.0
29	* @package Mlsimport
30	* @subpackage Mlsimport/includes
31	*/
32
33	if (! defined('ABSPATH') ) {
34	exit; // Exit if accessed directly.
35	}
36
37	/**
38	* Stamp one connection's sync outcome into the per-connection maps.
39	*
40	* Step by step:
41	* 1. Ignore a non-positive id — an unattributed pull has no connection entry
42	* (the GLOBAL sync_health stamps are written by the caller either way).
43	* 2. Success: record the time in the 'connection_sync_success' map
44	* (mls_id => epoch).
45	* 3. Failure: record time + failure class in the 'connection_sync_failures'
46	* map (mls_id => {at, code}) — the same map the #277 cron gate writes, so
47	* the heartbeat reads ONE map no matter which seam saw the failure.
48	* 4. Persist through mlsimport_telemetry_set() (single write, autoload no).
49	*
50	* @param int $mls_id The connection the pull ran for (0 = unattributed).
51	* @param bool $ok Whether the pull succeeded.
52	* @param string $code Failure class when $ok is false ('token', 'auth', ...).
53	* @return void
54	*/
55	function mlsimport_telemetry_record_connection_sync( int $mls_id, bool $ok, string $code = '' ): void
56	{
57	// Step 1: no connection, no per-connection entry.
58	if ($mls_id <= 0 ) {
59	return;
60	}
61
62	// Load the persisted state; coerce a non-array back to an array.
63	$state = get_option('mlsimport_telemetry_state', array());
64	if (! is_array($state) ) {
65	$state = array();
66	}
67
68	if ($ok ) {
69	// Step 2: success map — mls_id => epoch of the last good pull.
70	$map = is_array($state['connection_sync_success'] ?? null) ? $state['connection_sync_success'] : array();
71	$map[ $mls_id ] = time();
72	mlsimport_telemetry_set('connection_sync_success', $map);
73	return;
74	}
75
76	// Step 3: failure map — mls_id => when it failed and why.
77	$map = is_array($state['connection_sync_failures'] ?? null) ? $state['connection_sync_failures'] : array();
78	$map[ $mls_id ] = array(
79	'at' => time(),
80	'code' => $code,
81	);
82	mlsimport_telemetry_set('connection_sync_failures', $map);
83	}
84
85	/**
86	* Gather each connection's workload for the heartbeat: how many Import Tasks
87	* are bound to it, how many of those the cron gate would skip, and how many
88	* listings carry its provenance stamp.
89	*
90	* Step by step:
91	* 1. Zero-base one workload slot per registered connection.
92	* 2. Resolve every task through the #277 cron gate wrapper — the SAME
93	* binding resolution (stamped meta, else the current connection) and the
94	* SAME import/skip decision the hourly cron applies, so the heartbeat
95	* counts exactly what the cron would do. A task bound to a deleted /
96	* unregistered connection has no entry to count against and is skipped
97	* (its connection is absent from the array the portal reads anyway).
98	* 3. Count each connection's listings: published posts carrying its
99	* 'mlsimport_mls_id' provenance stamp (#278). Hidden dedupe losers (#282)
100	* are stored listings and must count, hence the include-hidden flag.
101	*
102	* Only called with a non-empty registry, from the payload collector — once
103	* per heartbeat, so the per-connection count queries stay cheap.
104	*
105	* @param array $records Registry records keyed by mls_id.
106	* @param array $task_ids Every Import Task post id on the install.
107	* @param string $post_type The theme's property post type.
108	* @return array<int, array{tasks: int, paused: int, listings: int}> Workload per mls_id.
109	*/
110	function mlsimport_telemetry_gather_connection_workload( array $records, array $task_ids, string $post_type ): array
111	{
112	// Step 1: one zeroed slot per registered connection.
113	$workload = array();
114	foreach ( $records as $record ) {
115	$workload[ (int) $record['mls_id'] ] = array(
116	'tasks' => 0,
117	'paused' => 0,
118	'listings' => 0,
119	);
120	}
121
122	// Step 2: gate every task exactly as the hourly cron would.
123	foreach ( $task_ids as $task_id ) {
124	$gate = mlsimport_cron_task_gate((int) $task_id);
125	$mls_id = (int) $gate['mls_id'];
126	if (! isset($workload[ $mls_id ]) ) {
127	continue;
128	}
129	$workload[ $mls_id ]['tasks']++;
130	// Paused/dead = the gate would skip it (broken, unentitled, deleted).
131	if ('import' !== $gate['action'] ) {
132	$workload[ $mls_id ]['paused']++;
133	}
134	}
135
136	// Step 3: provenance-stamped listing count per connection.
137	foreach ( array_keys($workload) as $mls_id ) {
138	$provenance_query = new WP_Query(
139	array(
140	'post_type' => $post_type,
141	'post_status' => 'publish',
142	'posts_per_page' => 1,
143	'fields' => 'ids',
144	'no_found_rows' => false,
145	'meta_query' => array(
146	array(
147	'key' => 'mlsimport_mls_id',
148	'value' => (string) $mls_id,
149	),
150	),
151	'mlsimport_include_hidden' => true,
152	)
153	);
154	$workload[ $mls_id ]['listings'] = (int) $provenance_query->found_posts;
155	}
156
157	return $workload;
158	}
159
160	/**
161	* Build the heartbeat `connections` array — the PURE payload-assembly seam.
162	*
163	* Everything WordPress-dependent (registry read, task gating, listing
164	* counts) is gathered by the caller and passed in, so this function can be
165	* unit-tested with multi-connection state fixtures (issue #283 acceptance)
166	* and never touches the DB.
167	*
168	* Step by step, per registered connection (records arrive priority-sorted
169	* from Mlsimport_Connections::all(), so the array keeps that order):
170	* 1. Identity: mls_id, provider type, priority.
171	* 2. Connection status: the record's test result + when it was last tested.
172	* 3. Sync health: last successful / last failed pull (+ failure class) from
173	* the per-connection maps written above.
174	* 4. 7-day counters: sum this connection's 'daily_mls' buckets with the same
175	* summer the global fields use — so global sums equal the sum of the
176	* per-connection buckets by construction. token_failures is fed only by
177	* token failures attributable to ONE connection; the plugin's own SaaS
178	* JWT is account-level and counts globally only, so per-connection token
179	* pain currently surfaces through last_failure_code = 'token' instead.
180	* 5. Workload: bound task count, how many of those tasks the cron gate would
181	* skip (paused/dead), the count of listings stamped with this mls_id, and
182	* the last time an import touched a listing for this connection.
183	*
184	* @param array $records Registry records keyed by mls_id, priority order.
185	* @param array $state The full mlsimport_telemetry_state array.
186	* @param string $today Reference UTC date 'Y-m-d' for the 7-day sums.
187	* @param array $workload mls_id => {tasks, paused, listings} counts.
188	* @return array<int, array> One complete health entry per connection.
189	*/
190	function mlsimport_telemetry_connections_payload( array $records, array $state, string $today, array $workload ): array
191	{
192	// The per-connection daily buckets and stamp maps (all optional in state).
193	$daily_mls = is_array($state['daily_mls'] ?? null) ? $state['daily_mls'] : array();
194	$success_map = is_array($state['connection_sync_success'] ?? null) ? $state['connection_sync_success'] : array();
195	$failure_map = is_array($state['connection_sync_failures'] ?? null) ? $state['connection_sync_failures'] : array();
196	$import_map = is_array($state['connection_last_import'] ?? null) ? $state['connection_last_import'] : array();
197
198	$entries = array();
199	foreach ( $records as $record ) {
200	$mls_id = (int) ( $record['mls_id'] ?? 0 );
201
202	// Step 4: this connection's own 7-day sums (same summer as the globals).
203	$mls_daily = is_array($daily_mls[ $mls_id ] ?? null) ? $daily_mls[ $mls_id ] : array();
204	$sums = mlsimport_telemetry_sum_buckets($mls_daily, $today, 7);
205
206	// Step 3: this connection's failure record ({at, code} or absent).
207	$failure = is_array($failure_map[ $mls_id ] ?? null) ? $failure_map[ $mls_id ] : array();
208
209	// Step 5: this connection's gathered workload counts (or all-zero).
210	$load = is_array($workload[ $mls_id ] ?? null) ? $workload[ $mls_id ] : array();
211
212	$entries[] = array(
213	// Step 1: identity.
214	'mls_id' => $mls_id,
215	'provider' => (string) ( $record['provider_type'] ?? '' ),
216	'priority' => (int) ( $record['priority'] ?? 0 ),
217	// Step 2: connection-test status.
218	'status' => (string) ( $record['status'] ?? '' ),
219	'last_test' => mlsimport_telemetry_iso((int) ( $record['tested_at'] ?? 0 )),
220	// Step 3: per-connection sync health.
221	'last_successful_sync' => mlsimport_telemetry_iso((int) ( $success_map[ $mls_id ] ?? 0 )),
222	'last_failed_sync' => mlsimport_telemetry_iso((int) ( $failure['at'] ?? 0 )),
223	'last_failure_code' => (string) ( $failure['code'] ?? '' ),
224	// Step 4: 7-day counters from this connection's own buckets.
225	'imported_last_7_days' => (int) $sums['imported'],
226	'updated_last_7_days' => (int) $sums['updated'],
227	'deleted_last_7_days' => (int) $sums['deleted'],
228	'syncs_last_7_days' => (int) $sums['syncs'],
229	'token_failures_last_7_days' => (int) $sums['token_failures'],
230	// Step 5: workload.
231	'task_count' => (int) ( $load['tasks'] ?? 0 ),
232	'tasks_paused_dead' => (int) ( $load['paused'] ?? 0 ),
233	'listing_count' => (int) ( $load['listings'] ?? 0 ),
234	'last_import' => mlsimport_telemetry_iso((int) ( $import_map[ $mls_id ] ?? 0 )),
235	);
236	}
237
238	return $entries;
239	}
240