← All changes | app/Services/WPAsync/FluentFormAsyncRequest.php +110 -7 6.2.4 → 6.2.14 View file →

		@@ -5,8 +5,9 @@
5	5	use FluentForm\App\Helpers\Helper;
6	6	use FluentForm\App\Modules\Form\FormDataParser;
7	7	use FluentForm\App\Modules\Form\FormFieldsParser;
8	8	use FluentForm\Framework\Foundation\Application;
	9	+use FluentForm\App\Services\Integrations\GlobalNotificationManager;
9	10
10	11	class FluentFormAsyncRequest
11	12	{
12	13	/**
		@@ -109,8 +110,10 @@
109	110
110	111	$formCache = [];
111	112	$submissionCache = [];
112	113	$entryCache = [];
	114	+ $formDataCache = [];
	115	+ $feedCache = $this->loadFeedRows($actionFeeds);
113	116
114	117	foreach ($actionFeeds as $actionFeed) {
115	118	$action = $actionFeed->action;
116	119	$feed = Helper::safeUnserialize($actionFeed->data);
		@@ -118,8 +121,12 @@
118	121	if(isset($submissionCache[$actionFeed->origin_id])) {
119	122	$submission = $submissionCache[$actionFeed->origin_id];
120	123	} else {
121	124	$submission = wpFluent()->table('fluentform_submissions')->find($actionFeed->origin_id);
	125	+ if (!$submission) {
	126	+ $this->abandonStaleRow($actionFeed, 'Skipped: the submission or form no longer exists');
	127	+ continue;
	128	+ }
122	129	$submissionCache[$submission->id] = $submission;
123	130	}
124	131	if(isset($formCache[$submission->form_id])) {
125	132	$form = $formCache[$submission->form_id];
		@@ -124,11 +131,25 @@
124	131	if(isset($formCache[$submission->form_id])) {
125	132	$form = $formCache[$submission->form_id];
126	133	} else {
127	134	$form = wpFluent()->table('fluentform_forms')->find($submission->form_id);
	135	+ if (!$form) {
	136	+ $this->abandonStaleRow($actionFeed, 'Skipped: the submission or form no longer exists');
	137	+ continue;
	138	+ }
128	139	$formCache[$form->id] = $form;
129	140	}
130	141
	142	+ if (!isset($formDataCache[$submission->id])) {
	143	+ $formDataCache[$submission->id] = json_decode($submission->response, true);
	144	+ }
	145	+ $formData = $formDataCache[$submission->id];
	146	+
	147	+ if (!$this->feedStillEnabled($actionFeed, $feedCache, $formData, $submission->id)) {
	148	+ $this->abandonStaleRow($actionFeed, 'Skipped: the feed was disabled or deleted after queueing');
	149	+ continue;
	150	+ }
	151	+
131	152	if(isset($entryCache[$submission->id])) {
132	153	$entry = $entryCache[$submission->id];
133	154	} else {
134	155	$entry = $this->getEntry($submission, $form);
		@@ -133,12 +154,13 @@
133	154	} else {
134	155	$entry = $this->getEntry($submission, $form);
135	156	$entryCache[$submission->id] = $entry;
136	157	}
137		- $formData = json_decode($submission->response, true);
138	158
139		- wpFluent()->table($this->table)
	159	+ // Same atomic claim as process(); this path is a public nopriv ajax endpoint.
	160	+ $claimed = wpFluent()->table($this->table)
140	161	->where('id', $actionFeed->id)
	162	+ ->whereIn('status', ['pending', 'failed'])
141	163	->update([
142	164	'status' => 'processing',
143	165	'retry_count' => $actionFeed->retry_count + 1,
144	166	'updated_at' => current_time('mysql')
		@@ -143,8 +165,12 @@
143	165	'retry_count' => $actionFeed->retry_count + 1,
144	166	'updated_at' => current_time('mysql')
145	167	]);
146	168
	169	+ if (!$claimed) {
	170	+ continue;
	171	+ }
	172	+
147	173	// phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.DynamicHooknameFound -- Dynamic hook name for async request
148	174	do_action($action, $feed, $formData, $entry, $form);
149	175	}
150	176
		@@ -179,9 +205,12 @@
179	205	if (isset(static::$submissionCache[$queue->origin_id])) {
180	206	$submission = static::$submissionCache[$queue->origin_id];
181	207	} else {
182	208	$submission = wpFluent()->table('fluentform_submissions')->find($queue->origin_id);
183		-
	209	+ if (!$submission) {
	210	+ $this->abandonStaleRow($queue, 'Skipped: the submission or form no longer exists');
	211	+ return;
	212	+ }
184	213	static::$submissionCache[$submission->id] = $submission;
185	214	}
186	215
187	216	if (isset(static::$formCache[$submission->form_id])) {
		@@ -187,12 +216,24 @@
187	216	if (isset(static::$formCache[$submission->form_id])) {
188	217	$form = static::$formCache[$submission->form_id];
189	218	} else {
190	219	$form = wpFluent()->table('fluentform_forms')->find($submission->form_id);
191		-
	220	+ if (!$form) {
	221	+ $this->abandonStaleRow($queue, 'Skipped: the submission or form no longer exists');
	222	+ return;
	223	+ }
192	224	static::$formCache[$form->id] = $form;
193	225	}
194	226
	227	+ $formData = json_decode($submission->response, true);
	228	+
	229	+ if (!$this->feedStillEnabled($queue, $this->loadFeedRows([$queue]), $formData, $submission->id)) {
	230	+ if ($this->abandonStaleRow($queue, 'Skipped: the feed was disabled or deleted after queueing')) {
	231	+ $this->maybeFinished($submission->id, $form);
	232	+ }
	233	+ return;
	234	+ }
	235	+
195	236	if (isset(static::$entryCache[$submission->id])) {
196	237	$entry = static::$entryCache[$submission->id];
197	238	} else {
198	239	$entry = $this->getEntry($submission, $form);
		@@ -199,12 +240,13 @@
199	240
200	241	static::$entryCache[$submission->id] = $entry;
201	242	}
202	243
203		- $formData = json_decode($submission->response, true);
204		-
205		- wpFluent()->table($this->table)
	244	+ // Atomic claim: the cron passes a row object, so status is never re-read. Must admit
	245	+ // 'failed' or retries die, and must always change a column - wpdb reports CHANGED rows.
	246	+ $claimed = wpFluent()->table($this->table)
206	247	->where('id', $queue->id)
	248	+ ->whereIn('status', ['pending', 'failed'])
207	249	->update([
208	250	'status' => 'processing',
209	251	'retry_count' => $queue->retry_count + 1,
210	252	'updated_at' => current_time('mysql')
		@@ -209,12 +251,73 @@
209	251	'retry_count' => $queue->retry_count + 1,
210	252	'updated_at' => current_time('mysql')
211	253	]);
212	254
	255	+ if (!$claimed) {
	256	+ return;
	257	+ }
	258	+
213	259	// phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.DynamicHooknameFound -- Dynamic hook name for async request
214	260	do_action($action, $feed, $formData, $entry, $form);
215	261
216	262	$this->maybeFinished($submission->id, $form);
	263	+ }
	264	+
	265	+ // 'skipped' is outside every consumer's selection: cron retries only 'failed', Pro's
	266	+ // failed-integration email reads 'failed'/'error', maybeFinished() counts 'pending'.
	267	+ // Same status predicate as the claim, so a row another worker owns is left alone.
	268	+ private function abandonStaleRow($row, $note)
	269	+ {
	270	+ return (bool) wpFluent()->table($this->table)
	271	+ ->where('id', $row->id)
	272	+ ->whereIn('status', ['pending', 'failed'])
	273	+ ->update([
	274	+ 'status' => 'skipped',
	275	+ 'note' => $note,
	276	+ 'updated_at' => current_time('mysql'),
	277	+ ]);
	278	+ }
	279	+
	280	+ // One query for every distinct feed in the batch, keyed by form_meta id.
	281	+ private function loadFeedRows($actionFeeds)
	282	+ {
	283	+ $feedIds = [];
	284	+ foreach ($actionFeeds as $actionFeed) {
	285	+ if ($actionFeed->feed_id) {
	286	+ $feedIds[(int) $actionFeed->feed_id] = true;
	287	+ }
	288	+ }
	289	+ if (!$feedIds) {
	290	+ return [];
	291	+ }
	292	+
	293	+ $rows = wpFluent()->table('fluentform_form_meta')->whereIn('id', array_keys($feedIds))->get();
	294	+
	295	+ $byId = [];
	296	+ foreach ($rows as $row) {
	297	+ $byId[(int) $row->id] = $row;
	298	+ }
	299	+
	300	+ return $byId;
	301	+ }
	302	+
	303	+ // The row carries a snapshot of the feed taken at submission time; re-run the producer's
	304	+ // own enabled + condition check against the live form_meta row before dispatching it.
	305	+ private function feedStillEnabled($row, $feedRows, $formData, $submissionId)
	306	+ {
	307	+ $feedId = (int) $row->feed_id;
	308	+ if (!$feedId) {
	309	+ return true;
	310	+ }
	311	+
	312	+ $meta = $feedRows[$feedId] ?? null;
	313	+ if (!$meta) {
	314	+ return false;
	315	+ }
	316	+
	317	+ $manager = new GlobalNotificationManager($this->app);
	318	+
	319	+ return (bool) $manager->getEnabledFeeds([$meta], $formData, $submissionId);
217	320	}
218	321
219	322	public function maybeFinished($originId, $form)
220	323	{