fluent-community / database / Migrations / NotificationPrefMigrator.php

NotificationPrefMigrator.php in FluentCommunity – Ultra-Fast High-Performance Social Network, Community, LMS & Online Courses 2.10.01, at database/Migrations/NotificationPrefMigrator.php

467 lines 17.9 KB

Raw Download Zip

1	<?php
2	// phpcs:disable
3
4	namespace FluentCommunity\Database\Migrations;
5
6	use FluentCommunity\App\Services\NotificationPref;
7
8	class NotificationPrefMigrator
9	{
10	/**
11	* Legacy pref rows in fcom_notification_users mapped onto the
12	* (channel, event_key) grid. The legacy keys baked the channel into the key
13	* name - com_my_post_mail / com_my_post_push - which is what this table undoes.
14	*
15	* Deliberately a frozen literal rather than derived from
16	* NotificationPref::prefKeyMap(): this describes what the rows meant at the
17	* time of the move, and must not shift if the live vocabulary later does.
18	*
19	* legacy notification_type => [channel, event_key, is_object_scoped]
20	*/
21	private static $legacyMap = [
22	'com_my_post_mail' => ['mail', 'comment', false],
23	'com_my_post_push' => ['push', 'comment', false],
24	'reply_my_com_mail' => ['mail', 'reply', false],
25	'reply_my_com_push' => ['push', 'reply', false],
26	'mention_mail' => ['mail', 'mention', false],
27	'mention_push' => ['push', 'mention', false],
28	'digest_mail' => ['mail', 'digest', false],
29	'message_email_frequency' => ['mail', 'message_frequency', false],
30	'np_by_member_mail' => ['mail', 'np_by_member', true],
31	'np_by_admin_mail' => ['mail', 'np_by_admin', true],
32	];
33
34	/** Legacy rows scanned per statement. Bounded work, not bounded matches. */
35	const BATCH_SIZE = 5000;
36
37	/** Seconds of this request the backfill may use before deferring the rest. */
38	const TIME_BUDGET = 15;
39
40	const DONE_OPTION = 'fluent_community_notification_pref_backfilled';
41
42	const CURSOR_OPTION = 'fluent_community_notification_pref_backfill_cursor';
43
44	const ERROR_OPTION = 'fluent_community_notification_pref_backfill_error';
45
46	const RESUME_HOOK = 'fluent_community/migrate_notification_prefs';
47
48	/** Legacy rows deleted per statement, once the copy is done and verified. */
49	const DELETE_BATCH_SIZE = 2000;
50
51	/**
52	* Migrate the table.
53	*
54	* @return void
55	*/
56	public static function migrate()
57	{
58	global $wpdb;
59
60	$charsetCollate = $wpdb->get_charset_collate();
61
62	$table = $wpdb->prefix . 'fcom_notification_prefs';
63	$indexPrefix = $wpdb->prefix . 'fcom_np_';
64
65	if ($wpdb->get_var("SHOW TABLES LIKE '$table'") != $table) {
66	/*
67	* object_id is NOT NULL DEFAULT 0 rather than nullable on purpose:
68	* MySQL treats NULLs as distinct inside a UNIQUE key, so a nullable
69	* object_id would let duplicate global prefs through the unique index.
70	* 0 means "global / not scoped to an object".
71	*
72	* The `fanout` index is ordered for the recipient-selection queries in
73	* EmailNotificationHandler: channel + event_key + object_id + value are
74	* all equality predicates, and the trailing user_id serves both the
75	* EXISTS probe and the `ID > $lastSentUserId` batch cursor without a
76	* lookup back to the row.
77	*
78	* The `uniq` key doubles as the per-user read index (it leads with
79	* user_id), so no separate index is needed for getUserPrefs().
80	*/
81	$sql = "CREATE TABLE $table (
82	`id` BIGINT UNSIGNED NOT NULL PRIMARY KEY AUTO_INCREMENT,
83	`user_id` BIGINT UNSIGNED NOT NULL,
84	`channel` VARCHAR(20) NOT NULL DEFAULT 'mail',
85	`event_key` VARCHAR(50) NOT NULL,
86	`object_id` BIGINT UNSIGNED NOT NULL DEFAULT 0,
87	`value` TINYINT UNSIGNED NOT NULL DEFAULT 0,
88	`created_at` TIMESTAMP NULL,
89	`updated_at` TIMESTAMP NULL,
90	UNIQUE KEY `{$indexPrefix}_uniq` (`user_id`, `channel`, `event_key`, `object_id`),
91	INDEX `{$indexPrefix}_fanout` (`channel`, `event_key`, `object_id`, `value`, `user_id`)
92	) $charsetCollate;";
93	dbDelta($sql);
94	}
95
96	self::maybeBackfillFromLegacy();
97	}
98
99	/**
100	* Copy object_type = 'notification_pref' rows out of fcom_notification_users.
101	*
102	* Resumable by design. The legacy table has no index leading with
103	* object_type or notification_type, so a single filtered pass would be a full
104	* scan of a table that grows with notification receipts rather than with
105	* preferences. Instead this walks the primary key in fixed-size steps, so the
106	* work per statement is bounded by rows scanned, not by rows matched, and a
107	* partial run resumes from its cursor rather than starting over.
108	*
109	* A failed statement never advances the cursor and never sets the done flag:
110	* a half-copied migration that reported success would silently drop members
111	* back to the site defaults - someone who switched digest mail off would
112	* start receiving it again - which is worse than a migration that retries.
113	*
114	* @return bool true when the backfill is complete
115	*/
116	public static function maybeBackfillFromLegacy()
117	{
118	global $wpdb;
119
120	if (get_option(self::DONE_OPTION)) {
121	return true;
122	}
123
124	$legacyTable = $wpdb->prefix . 'fcom_notification_users';
125
126	if ($wpdb->get_var("SHOW TABLES LIKE '$legacyTable'") != $legacyTable) {
127	// Fresh install: nothing to carry over.
128	self::markComplete();
129	return true;
130	}
131
132	$maxId = (int)$wpdb->get_var("SELECT MAX(id) FROM {$legacyTable}");
133	$cursor = (int)get_option(self::CURSOR_OPTION, 0);
134
135	// One budget for the whole call, timed from here. Anything that resumes
136	// does so in an Action Scheduler request of its own, with a fresh one.
137	$startedAt = microtime(true);
138
139	while ($cursor < $maxId) {
140	// The id this batch ends on. Walks the primary key evenly whether or
141	// not the id space is sparse, and never scans more than BATCH_SIZE.
142	$batchEnd = (int)$wpdb->get_var($wpdb->prepare(
143	"SELECT MAX(id) FROM (
144	SELECT id FROM {$legacyTable} WHERE id > %d ORDER BY id ASC LIMIT %d
145	) AS batch",
146	$cursor,
147	self::BATCH_SIZE
148	));
149
150	if (!$batchEnd) {
151	break;
152	}
153
154	if (!self::copyLegacyRange($cursor, $batchEnd)) {
155	// Leave the cursor where it was so the range is retried intact.
156	update_option(self::ERROR_OPTION, $wpdb->last_error, false);
157	self::scheduleResume();
158	return false;
159	}
160
161	$cursor = $batchEnd;
162	update_option(self::CURSOR_OPTION, $cursor, false);
163
164	if (microtime(true) - $startedAt > self::TIME_BUDGET) {
165	// Out of budget for this call, but the cursor is durable.
166	self::scheduleResume();
167	return false;
168	}
169	}
170
171	/*
172	* The copy is done. Check that every legacy row has a counterpart before
173	* removing anything - it is one query, and it is the difference between
174	* deleting rows we copied and deleting rows we only think we copied.
175	*
176	* Note what it does not compare: the value. A member who changes a
177	* preference after the migration legitimately makes the two differ, so
178	* matching on value would report false gaps forever.
179	*/
180	$unmigrated = self::countUnmigratedRows();
181
182	/*
183	* And separately: rows whose notification_type is not in $legacyMap at all.
184	*
185	* countUnmigratedRows() cannot see these. It joins the same map the copy
186	* joins, so a key the map does not know about is absent from both sides of
187	* that comparison and reads as zero - the copy skips it, the check passes
188	* it, and an unscoped delete then removes a preference nobody carried over.
189	* The check has to be asked about the rows the map does not cover, not only
190	* about the rows it does.
191	*/
192	$unmapped = self::countUnmappedRows();
193
194	if ($unmigrated > 0 \|\| $unmapped > 0) {
195	// Copied data stands and the new table is authoritative, so this is
196	// complete either way - but leave the source alone for inspection.
197	$problems = [];
198
199	if ($unmigrated > 0) {
200	$problems[] = sprintf('%d legacy rows had no counterpart', $unmigrated);
201	}
202
203	if ($unmapped > 0) {
204	$problems[] = sprintf('%d legacy rows used a key this version does not map', $unmapped);
205	}
206
207	update_option(self::ERROR_OPTION, implode('; ', $problems) . '; source left in place', false);
208	self::markComplete();
209
210	return true;
211	}
212
213	if (!self::deleteLegacyRows($startedAt)) {
214	self::scheduleResume();
215
216	return false;
217	}
218
219	// Clear any error recorded by an attempt that has since succeeded.
220	delete_option(self::ERROR_OPTION);
221	self::markComplete();
222
223	return true;
224	}
225
226	/**
227	* Remove the rows the copy read from.
228	*
229	* Batched because a single DELETE over a large table holds locks for as long
230	* as it runs, and this table is read on every portal request by the unread
231	* count. Not otherwise ceremonious: if the request dies partway, the done
232	* flag is never set, the next pass re-runs a copy that is now a no-op and
233	* carries on deleting.
234	*
235	* Public as a test seam, for the same reason copyLegacyRange() is: it is pure
236	* DML, and what it is scoped to is the part worth pinning down.
237	*
238	* @param float $startedAt microtime this call began, for the shared budget
239	* @return bool true when nothing is left to delete
240	*/
241	public static function deleteLegacyRows($startedAt)
242	{
243	global $wpdb;
244
245	$legacyTable = $wpdb->prefix . 'fcom_notification_users';
246
247	$keys = array_keys(self::$legacyMap);
248	$placeholders = implode(', ', array_fill(0, count($keys), '%s'));
249
250	while (true) {
251	/*
252	* Pinned twice over: to the preference rows, so this cannot reach a
253	* notification receipt (live data the ticker and the toast read), and
254	* to the keys the copy above actually knows how to place, so a key
255	* this version does not map survives rather than being deleted
256	* uncopied. The guard in the caller should already have stopped us
257	* before that could happen; this makes it true by construction
258	* instead of by check.
259	*/
260	$args = $keys;
261	$args[] = self::DELETE_BATCH_SIZE;
262
263	$deleted = $wpdb->query($wpdb->prepare(
264	"DELETE FROM {$legacyTable}
265	WHERE `object_type` = 'notification_pref'
266	AND `notification_type` IN ({$placeholders})
267	LIMIT %d",
268	$args
269	));
270
271	if ($deleted === false) {
272	update_option(self::ERROR_OPTION, $wpdb->last_error, false);
273
274	return false;
275	}
276
277	if (!$deleted) {
278	return true;
279	}
280
281	if (microtime(true) - $startedAt > self::TIME_BUDGET) {
282	return false;
283	}
284	}
285	}
286
287	/**
288	* Copy one primary-key range in a single statement.
289	*
290	* ON DUPLICATE KEY UPDATE rather than INSERT IGNORE: both make a re-run of an
291	* already-copied range a no-op, but IGNORE also downgrades
292	* genuine errors (truncation, constraint violations) to warnings, which is
293	* the exact silent failure this method exists to report. The update is a
294	* deliberate self-assignment rather than a write, so a range replayed after
295	* the member has since changed that preference cannot overwrite their newer
296	* choice. The column is table-qualified because the joined legacy table
297	* shares column names with the target.
298	*
299	* Public as a test seam: this is the mapping the whole migration rests on,
300	* and it is pure DML, so the integration tier can exercise it inside its
301	* transaction without the implicit COMMIT that CREATE TABLE would cause.
302	*
303	* @param int $fromId exclusive
304	* @param int $toId inclusive
305	* @return bool
306	*/
307	public static function copyLegacyRange($fromId, $toId)
308	{
309	global $wpdb;
310
311	$table = $wpdb->prefix . 'fcom_notification_prefs';
312	$legacyTable = $wpdb->prefix . 'fcom_notification_users';
313
314	$mapRows = [];
315	$args = [];
316
317	foreach (self::$legacyMap as $legacyKey => $config) {
318	$mapRows[] = 'SELECT %s AS lk, %s AS ch, %s AS ek, %d AS scoped';
319	$args[] = $legacyKey;
320	$args[] = $config[0];
321	$args[] = $config[1];
322	$args[] = $config[2] ? 1 : 0;
323	}
324
325	$args[] = $fromId;
326	$args[] = $toId;
327
328	$sql = "INSERT INTO {$table}
329	(`user_id`, `channel`, `event_key`, `object_id`, `value`, `created_at`, `updated_at`)
330	SELECT l.`user_id`, m.ch, m.ek,
331	CASE WHEN m.scoped = 1 THEN COALESCE(l.`object_id`, 0) ELSE 0 END,
332	COALESCE(l.`is_read`, 0), l.`created_at`, l.`updated_at`
333	FROM {$legacyTable} l
334	INNER JOIN (" . implode(' UNION ALL ', $mapRows) . ") m ON m.lk = l.`notification_type`
335	WHERE l.`object_type` = 'notification_pref'
336	AND l.`user_id` IS NOT NULL
337	AND l.`id` > %d
338	AND l.`id` <= %d
339	ON DUPLICATE KEY UPDATE {$table}.`value` = {$table}.`value`";
340
341	return $wpdb->query($wpdb->prepare($sql, $args)) !== false;
342	}
343
344	/**
345	* Legacy preference rows with no counterpart in the new table. Zero is the
346	* only acceptable answer once the cursor has run out.
347	*
348	* @return int
349	*/
350	public static function countUnmigratedRows()
351	{
352	global $wpdb;
353
354	$table = $wpdb->prefix . 'fcom_notification_prefs';
355	$legacyTable = $wpdb->prefix . 'fcom_notification_users';
356
357	$mapRows = [];
358	$args = [];
359
360	foreach (self::$legacyMap as $legacyKey => $config) {
361	$mapRows[] = 'SELECT %s AS lk, %s AS ch, %s AS ek, %d AS scoped';
362	$args[] = $legacyKey;
363	$args[] = $config[0];
364	$args[] = $config[1];
365	$args[] = $config[2] ? 1 : 0;
366	}
367
368	$sql = "SELECT COUNT(*)
369	FROM {$legacyTable} l
370	INNER JOIN (" . implode(' UNION ALL ', $mapRows) . ") m ON m.lk = l.`notification_type`
371	LEFT JOIN {$table} p
372	ON p.`user_id` = l.`user_id`
373	AND p.`channel` = m.ch
374	AND p.`event_key` = m.ek
375	AND p.`object_id` = CASE WHEN m.scoped = 1 THEN COALESCE(l.`object_id`, 0) ELSE 0 END
376	WHERE l.`object_type` = 'notification_pref'
377	AND l.`user_id` IS NOT NULL
378	AND p.`id` IS NULL";
379
380	return (int)$wpdb->get_var($wpdb->prepare($sql, $args));
381	}
382
383	/**
384	* Legacy preference rows whose key is not in $legacyMap.
385	*
386	* The blind spot in countUnmigratedRows(): that query joins the map, so it can
387	* only ever report on keys the map contains. This one asks the complement, and
388	* a non-zero answer means the vocabulary has drifted from the frozen map and
389	* the source must not be deleted.
390	*
391	* A NULL notification_type counts as unmapped. It is as unplaceable as an
392	* unknown one, and SQL's NOT IN would otherwise return NULL and drop it.
393	*
394	* @return int
395	*/
396	public static function countUnmappedRows()
397	{
398	global $wpdb;
399
400	$legacyTable = $wpdb->prefix . 'fcom_notification_users';
401
402	$keys = array_keys(self::$legacyMap);
403	$placeholders = implode(', ', array_fill(0, count($keys), '%s'));
404
405	$sql = "SELECT COUNT(*)
406	FROM {$legacyTable}
407	WHERE `object_type` = 'notification_pref'
408	AND (`notification_type` IS NULL OR `notification_type` NOT IN ({$placeholders}))";
409
410	return (int)$wpdb->get_var($wpdb->prepare($sql, $keys));
411	}
412
413	/**
414	* Continuation entry point. Deliberately not gated on the plugin's db-version
415	* option: boot/app.php writes that as soon as DBMigrator::run() returns, so a
416	* backfill that deferred work would never be reached through the migrator again.
417	*
418	* @return void
419	*/
420	public static function continueBackfill()
421	{
422	if (get_option(self::DONE_OPTION)) {
423	return;
424	}
425
426	self::maybeBackfillFromLegacy();
427	}
428
429	/**
430	* @return void
431	*/
432	private static function scheduleResume()
433	{
434	if (!function_exists('as_next_scheduled_action') \|\| !function_exists('as_schedule_single_action')) {
435	return;
436	}
437
438	if (\as_next_scheduled_action(self::RESUME_HOOK, [], 'fluent-community')) {
439	return;
440	}
441
442	\as_schedule_single_action(time() + 60, self::RESUME_HOOK, [], 'fluent-community', true);
443	}
444
445	/**
446	* @return void
447	*/
448	private static function markComplete()
449	{
450	update_option(self::DONE_OPTION, 'yes', false);
451	delete_option(self::CURSOR_OPTION);
452
453	/*
454	* Drop any aggregate computed while this table was still filling up.
455	*
456	* NotificationPref::hasAnyEnabled() reads a denormalized option and only
457	* the preference write path refreshes it, so a "nobody has the digest on"
458	* answer derived from a partial table would outlive the migration that
459	* made it wrong - and the hourly scheduler unschedules the digest on it.
460	* Deleting the option rather than recomputing it here keeps the migration
461	* off the read path: the next call recomputes from a table that is now
462	* whole.
463	*/
464	delete_option(NotificationPref::AGGREGATE_OPTION);
465	}
466	}
467