fluent-community / database / Migrations / NotificationPrefMigrator.php

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

384 lines 14.4 KB

Raw Download Zip

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