← All changes | app/Services/WPAsync/FluentFormAsyncRequest.php +199 -12 3.6.64 → 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	/**
		@@ -28,15 +29,19 @@
28	29	protected $actions = array();
29	30
30	31	/**
31	32	* $app Instance of Application/Framework
32		- * @var FluentForm\Framework\Foundation\Application
	33	+ * @var \FluentForm\Framework\Foundation\Application
33	34	*/
34	35	protected $app = null;
35	36
	37	+ static $formCache = [];
	38	+ static $entryCache = [];
	39	+ static $submissionCache = [];
	40	+
36	41	/**
37	42	* Construct the Object
38		- * @param FluentForm\Framework\Foundation\Application $app
	43	+ * @param \FluentForm\Framework\Foundation\Application $app
39	44	*/
40	45	public function __construct(Application $app)
41	46	{
42	47	$this->app = $app;
		@@ -41,8 +46,13 @@
41	46	{
42	47	$this->app = $app;
43	48	}
44	49
	50	+ public function queue($feed)
	51	+ {
	52	+ return wpFluent()->table($this->table)->insertGetId($feed);
	53	+ }
	54	+
45	55	public function queueFeeds($feeds)
46	56	{
47	57	return wpFluent()->table($this->table)
48	58	->insert($feeds);
		@@ -49,14 +59,17 @@
49	59	}
50	60
51	61	public function dispatchAjax($data = [])
52	62	{
	63	+ /* This hook is deprecated and will be removed soon */
	64	+ $sslVerify = apply_filters('fluentform_https_local_ssl_verify', false);
	65	+
53	66	$args = array(
54	67	'timeout' => 0.1,
55	68	'blocking' => false,
56	69	'body' => $data,
57	70	'cookies' => $_COOKIE,
58		- 'sslverify' => apply_filters('fluentform_https_local_ssl_verify', false),
	71	+ 'sslverify' => apply_filters('fluentform/https_local_ssl_verify', $sslVerify),
59	72	);
60	73
61	74	$queryArgs = array(
62	75	'action' => $this->action,
		@@ -62,19 +75,21 @@
62	75	'action' => $this->action,
63	76	'nonce' => wp_create_nonce($this->action),
64	77	);
65	78
66		- $url = add_query_arg($queryArgs, admin_url( 'admin-ajax.php' ));
	79	+ $url = add_query_arg($queryArgs, Helper::getAjaxUrl());
67	80	wp_remote_post(esc_url_raw($url), $args);
68	81	}
69	82
70	83	public function handleBackgroundCall()
71	84	{
72		- $originId = false;
73		- if(isset($_REQUEST['origin_id'])) {
74		- $originId = intval($_REQUEST['origin_id']);
	85	+ $nonce = sanitize_text_field(wpFluentForm('request')->get('nonce', ''));
	86	+ if (!wp_verify_nonce($nonce, $this->action)) {
	87	+ die('invalid');
75	88	}
76	89
	90	+ $originId = wpFluentForm('request')->get('origin_id', false);
	91	+
77	92	$this->processActions($originId);
78	93	echo 'success';
79	94	die();
80	95	}
		@@ -88,9 +103,9 @@
88	103	}
89	104
90	105	$actionFeeds = $actionFeedQuery->get();
91	106
92		- if(!$actionFeeds) {
	107	+ if(count($actionFeeds) === 0) {
93	108	return;
94	109	}
95	110
96	111	$formCache = [];
		@@ -95,17 +110,23 @@
95	110
96	111	$formCache = [];
97	112	$submissionCache = [];
98	113	$entryCache = [];
	114	+ $formDataCache = [];
	115	+ $feedCache = $this->loadFeedRows($actionFeeds);
99	116
100	117	foreach ($actionFeeds as $actionFeed) {
101	118	$action = $actionFeed->action;
102		- $feed = maybe_unserialize($actionFeed->data);
	119	+ $feed = Helper::safeUnserialize($actionFeed->data);
103	120	$feed['scheduled_action_id'] = $actionFeed->id;
104	121	if(isset($submissionCache[$actionFeed->origin_id])) {
105	122	$submission = $submissionCache[$actionFeed->origin_id];
106	123	} else {
107	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	+ }
108	129	$submissionCache[$submission->id] = $submission;
109	130	}
110	131	if(isset($formCache[$submission->form_id])) {
111	132	$form = $formCache[$submission->form_id];
		@@ -110,11 +131,25 @@
110	131	if(isset($formCache[$submission->form_id])) {
111	132	$form = $formCache[$submission->form_id];
112	133	} else {
113	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	+ }
114	139	$formCache[$form->id] = $form;
115	140	}
116	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	+
117	152	if(isset($entryCache[$submission->id])) {
118	153	$entry = $entryCache[$submission->id];
119	154	} else {
120	155	$entry = $this->getEntry($submission, $form);
		@@ -119,12 +154,13 @@
119	154	} else {
120	155	$entry = $this->getEntry($submission, $form);
121	156	$entryCache[$submission->id] = $entry;
122	157	}
123		- $formData = json_decode($submission->response, true);
124	158
125		- wpFluent()->table($this->table)
	159	+ // Same atomic claim as process(); this path is a public nopriv ajax endpoint.
	160	+ $claimed = wpFluent()->table($this->table)
126	161	->where('id', $actionFeed->id)
	162	+ ->whereIn('status', ['pending', 'failed'])
127	163	->update([
128	164	'status' => 'processing',
129	165	'retry_count' => $actionFeed->retry_count + 1,
130	166	'updated_at' => current_time('mysql')
		@@ -129,13 +165,21 @@
129	165	'retry_count' => $actionFeed->retry_count + 1,
130	166	'updated_at' => current_time('mysql')
131	167	]);
132	168
	169	+ if (!$claimed) {
	170	+ continue;
	171	+ }
	172	+
	173	+ // phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.DynamicHooknameFound -- Dynamic hook name for async request
133	174	do_action($action, $feed, $formData, $entry, $form);
134	175	}
135	176
136	177	if($originId && !empty($form) && !empty($submission)) {
	178	+ /* This hook is deprecated and will be removed soon */
137	179	do_action('fluentform_global_notify_completed', $submission->id, $form);
	180	+
	181	+ do_action('fluentform/global_notify_completed', $submission->id, $form);
138	182	}
139	183	}
140	184
141	185	private function getEntry($submission, $form)
		@@ -142,5 +186,148 @@
142	186	{
143	187	$formInputs = FormFieldsParser::getEntryInputs($form, ['admin_label', 'raw']);
144	188	return FormDataParser::parseFormEntry($submission, $form, $formInputs);
145	189	}
146		-}
	190	+
	191	+ public function process($queue)
	192	+ {
	193	+ if (is_numeric($queue)) {
	194	+ $queue = wpFluent()->table($this->table)->where('status', 'pending')->find($queue);
	195	+ }
	196	+
	197	+ if (!$queue || empty($queue->action)) {
	198	+ return;
	199	+ }
	200	+
	201	+ $action = $queue->action;
	202	+ $feed = Helper::safeUnserialize($queue->data);
	203	+ $feed['scheduled_action_id'] = $queue->id;
	204	+
	205	+ if (isset(static::$submissionCache[$queue->origin_id])) {
	206	+ $submission = static::$submissionCache[$queue->origin_id];
	207	+ } else {
	208	+ $submission = wpFluent()->table('fluentform_submissions')->find($queue->origin_id);
	209	+ if (!$submission) {
	210	+ $this->abandonStaleRow($queue, 'Skipped: the submission or form no longer exists');
	211	+ return;
	212	+ }
	213	+ static::$submissionCache[$submission->id] = $submission;
	214	+ }
	215	+
	216	+ if (isset(static::$formCache[$submission->form_id])) {
	217	+ $form = static::$formCache[$submission->form_id];
	218	+ } else {
	219	+ $form = wpFluent()->table('fluentform_forms')->find($submission->form_id);
	220	+ if (!$form) {
	221	+ $this->abandonStaleRow($queue, 'Skipped: the submission or form no longer exists');
	222	+ return;
	223	+ }
	224	+ static::$formCache[$form->id] = $form;
	225	+ }
	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	+
	236	+ if (isset(static::$entryCache[$submission->id])) {
	237	+ $entry = static::$entryCache[$submission->id];
	238	+ } else {
	239	+ $entry = $this->getEntry($submission, $form);
	240	+
	241	+ static::$entryCache[$submission->id] = $entry;
	242	+ }
	243	+
	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)
	247	+ ->where('id', $queue->id)
	248	+ ->whereIn('status', ['pending', 'failed'])
	249	+ ->update([
	250	+ 'status' => 'processing',
	251	+ 'retry_count' => $queue->retry_count + 1,
	252	+ 'updated_at' => current_time('mysql')
	253	+ ]);
	254	+
	255	+ if (!$claimed) {
	256	+ return;
	257	+ }
	258	+
	259	+ // phpcs:ignore WordPress.NamingConventions.PrefixAllGlobals.DynamicHooknameFound -- Dynamic hook name for async request
	260	+ do_action($action, $feed, $formData, $entry, $form);
	261	+
	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);
	320	+ }
	321	+
	322	+ public function maybeFinished($originId, $form)
	323	+ {
	324	+ $pendingFeeds = wpFluent()->table($this->table)->where([
	325	+ 'status' => 'pending',
	326	+ 'origin_id' => $originId
	327	+ ])->get();
	328	+
	329	+ if (count($pendingFeeds) === 0) {
	330	+ do_action('fluentform/global_notify_completed', $originId, $form);
	331	+ }
	332	+ }
	333	+}