From ffc7bad0e65d3d509f2d3d1227de4652b39842c1 Mon Sep 17 00:00:00 2001
From: paladox <paladox@users.noreply.github.com>
Date: Tue, 24 Sep 2024 16:17:15 +0100
Subject: [PATCH 1/7] Support new event/jobrunner for beta

We want to be able to use it for beta in addition to production.
---
 JobQueueEventBus.php | 171 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 171 insertions(+)
 create mode 100644 JobQueueEventBus.php

diff --git a/JobQueueEventBus.php b/JobQueueEventBus.php
new file mode 100644
index 0000000000..f0d2554ab7
--- /dev/null
+++ b/JobQueueEventBus.php
@@ -0,0 +1,171 @@
+<?php
+
+use ArrayIterator;
+use IJobSpecification;
+use Iterator;
+use Job;
+use JobQueue;
+use JobQueueError;
+use MediaWiki\Extension\EventBus\EventBus;
+use RunnableJob;
+
+/**
+ * Class is a copy of https://github.com/wikimedia/mediawiki-extensions-EventBus/blob/ea8ab686e80f56ac1e1c44296b8c823d4d769ee7/includes/Adapters/JobQueue/JobQueueEventBus.php
+ * with changes for beta.
+ */
+class JobQueueEventBusBeta extends JobQueue {
+	/**
+	 * Get the allowed queue orders for configuration validation
+	 *
+	 * @return array Subset of (random, timestamp, fifo, undefined)
+	 */
+	protected function supportedOrders() {
+		return [ 'fifo' ];
+	}
+
+	/**
+	 * Find out if delayed jobs are supported for configuration validation
+	 *
+	 * @return bool Whether delayed jobs are supported
+	 */
+	protected function supportsDelayedJobs() {
+		return true;
+	}
+
+	/**
+	 * Get the default queue order to use if configuration does not specify one
+	 *
+	 * @return string One of (random, timestamp, fifo, undefined)
+	 */
+	protected function optimalOrder() {
+		return 'fifo';
+	}
+
+	/**
+	 * @see JobQueue::isEmpty()
+	 * @return bool
+	 */
+	protected function doIsEmpty() {
+		// not implemented
+		return false;
+	}
+
+	/**
+	 * @see JobQueue::getSize()
+	 * @return int
+	 */
+	protected function doGetSize() {
+		// not implemented
+		return 0;
+	}
+
+	/**
+	 * @see JobQueue::getAcquiredCount()
+	 * @return int
+	 */
+	protected function doGetAcquiredCount() {
+		// not implemented
+		return 0;
+	}
+
+	/**
+	 * @see JobQueue::supportsTypeAgnostic()
+	 * @return bool
+	 */
+	protected function supportsTypeAgnostic(): bool {
+		return true;
+	}
+
+	/**
+	 * @param IJobSpecification[] $jobs
+	 * @param int $flags
+	 * @throws JobQueueError
+	 * @see JobQueue::batchPush()
+	 */
+	protected function doBatchPush( array $jobs, $flags ) {
+		$streamEvents = [];
+		$streamBuses = [];
+		$count = 0;
+
+		foreach ( $jobs as $job ) {
+			$stream = 'mediawiki.beta.job.' . $job->getType();
+			if ( !isset( $streamBuses[$stream] ) ) {
+				$streamBuses[$stream] = EventBus::getInstanceForStream( $stream );
+			}
+			$item = $streamBuses[$stream]->getFactory()->createJobEvent(
+				$stream,
+				$this->getDomain(),
+				$job
+			);
+
+			if ( $item === null ) {
+				continue;
+			}
+
+			$count++;
+			// hash identifier => de-duplicate
+			if ( isset( $item['sha1'] ) ) {
+				$streamEvents[$stream][$item['sha1']] = $item;
+			} else {
+				$streamEvents[$stream][$item['meta']['id']] = $item;
+			}
+		}
+
+		if ( !$count ) {
+			// nothing to do
+			return;
+		}
+
+		foreach ( array_keys( $streamEvents ) as $stream ) {
+			$result = $streamBuses[$stream]->send(
+				array_values( $streamEvents[$stream] ),
+				EventBus::TYPE_JOB
+			);
+
+			// This means sending jobs to the $stream has failed.
+			if ( is_array( $result ) || is_string( $result ) ) {
+				// Details of backend failure are logged by EventBus::send().
+				// Details of which job failed is logged here.
+				EventBus::logger()->error( 'Could not enqueue jobs for stream {stream}', [
+					'stream' => $stream,
+					'exception' => new JobQueueError( "Could not enqueue jobs" ),
+					'invalidresponse' => (object)$result
+				] );
+				// Avoid fragmenting exception by job or stream name, since backend
+				// issues are generally unrelated to the job (T249745).
+				throw new JobQueueError( "Could not enqueue jobs" );
+			}
+		}
+	}
+
+	/**
+	 * @see JobQueue::pop()
+	 * @return Job|bool
+	 */
+	protected function doPop() {
+		// not implemented
+		return false;
+	}
+
+	/**
+	 * @see JobQueue::ack()
+	 *
+	 * @param RunnableJob $job
+	 */
+	protected function doAck( RunnableJob $job ) {
+		// not implemented
+	}
+
+	/**
+	 * Get an iterator to traverse over all available jobs in this queue.
+	 * This does not include jobs that are currently acquired or delayed.
+	 * Note: results may be stale if the queue is concurrently modified.
+	 *
+	 * @return Iterator
+	 * @throws JobQueueError
+	 */
+	public function getAllQueuedJobs() {
+		// not implemented
+		return new ArrayIterator( [] );
+	}
+}

From a27b5d01b11a2e5df72e3e664d3b604e32a5ddc6 Mon Sep 17 00:00:00 2001
From: paladox <paladox@users.noreply.github.com>
Date: Tue, 24 Sep 2024 16:18:35 +0100
Subject: [PATCH 2/7] Update EventBus.php

---
 EventBus.php | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/EventBus.php b/EventBus.php
index 3da2f863dc..b47b81b039 100644
--- a/EventBus.php
+++ b/EventBus.php
@@ -5,6 +5,8 @@
 use MediaWiki\Extension\EventBus\Adapters\RCFeed\EventBusRCFeedEngine;
 use MediaWiki\Extension\EventBus\Adapters\RCFeed\EventBusRCFeedFormatter;
 
+require_once '/srv/mediawiki/config/JobQueueEventBusBeta.php';
+
 $wgEnableEventBus = 'TYPE_ALL';
 
 if ( $cwPrivate ) {

From afeca64e7b18bb1aeb236c654e7513d3340b305f Mon Sep 17 00:00:00 2001
From: paladox <paladox@users.noreply.github.com>
Date: Tue, 24 Sep 2024 16:18:48 +0100
Subject: [PATCH 3/7] Rename JobQueueEventBus.php to JobQueueEventBusBeta.php

---
 JobQueueEventBus.php => JobQueueEventBusBeta.php | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename JobQueueEventBus.php => JobQueueEventBusBeta.php (100%)

diff --git a/JobQueueEventBus.php b/JobQueueEventBusBeta.php
similarity index 100%
rename from JobQueueEventBus.php
rename to JobQueueEventBusBeta.php

From d021461fc16c456e4de88a9f2ca1a03e2552a5aa Mon Sep 17 00:00:00 2001
From: github-actions <github-actions@users.noreply.github.com>
Date: Tue, 24 Sep 2024 15:19:53 +0000
Subject: [PATCH 4/7] CI: lint code to MediaWiki standards

Check commit and GitHub actions for more details
---
 JobQueueEventBusBeta.php | 7 -------
 1 file changed, 7 deletions(-)

diff --git a/JobQueueEventBusBeta.php b/JobQueueEventBusBeta.php
index f0d2554ab7..5cfe6b8c0a 100644
--- a/JobQueueEventBusBeta.php
+++ b/JobQueueEventBusBeta.php
@@ -1,13 +1,6 @@
 <?php
 
-use ArrayIterator;
-use IJobSpecification;
-use Iterator;
-use Job;
-use JobQueue;
-use JobQueueError;
 use MediaWiki\Extension\EventBus\EventBus;
-use RunnableJob;
 
 /**
  * Class is a copy of https://github.com/wikimedia/mediawiki-extensions-EventBus/blob/ea8ab686e80f56ac1e1c44296b8c823d4d769ee7/includes/Adapters/JobQueue/JobQueueEventBus.php

From 149027323a23b1271d0b0050fa1d47690e1c33f7 Mon Sep 17 00:00:00 2001
From: paladox <paladox@users.noreply.github.com>
Date: Tue, 24 Sep 2024 16:50:05 +0100
Subject: [PATCH 5/7] Update EventBus.php

---
 EventBus.php | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/EventBus.php b/EventBus.php
index b47b81b039..2b901d770e 100644
--- a/EventBus.php
+++ b/EventBus.php
@@ -41,14 +41,14 @@
 	'class' => EventBusRCFeedEngine::class,
 ];
 
+$beta = preg_match( '/^(.*)\.(mirabeta|nexttide)\.org$/', $wi->server );
 $wgJobTypeConf['default'] = [
-	'class' => JobQueueEventBus::class,
+	'class' => preg_match( '/^(.*)\.(mirabeta|nexttide)\.org$/', $wi->server ) ?
+		JobQueueEventBusBeta::class :
+		JobQueueEventBus::class,
 	'readOnlyReason' => false
 ];
 
 $wgEventBusEnableRunJobAPI =
-	wfHostname() === 'mwtask151' ||
-	wfHostname() === 'mwtask161' ||
-	wfHostname() === 'mwtask171' ||
-	wfHostname() === 'mwtask181' ||
-	wfHostname() === 'test151';
+	strpos( wfHostname(), 'mwtask' ) === 0 ||
+	strpos( wfHostname(), 'test' ) === 0;

From 8c5b9c84a234d494dd78e4589e5ce4161c1c804f Mon Sep 17 00:00:00 2001
From: paladox <paladox@users.noreply.github.com>
Date: Tue, 24 Sep 2024 17:06:21 +0100
Subject: [PATCH 6/7] Update EventStreamConfig.php

---
 EventStreamConfig.php | 53 ++++++++++++++++++++++++-------------------
 1 file changed, 30 insertions(+), 23 deletions(-)

diff --git a/EventStreamConfig.php b/EventStreamConfig.php
index f09ccb86f5..d1d68b3ef5 100644
--- a/EventStreamConfig.php
+++ b/EventStreamConfig.php
@@ -9,26 +9,33 @@
 	'canary_events_enabled' => true,
 ];
 
+$betaStream = preg_match( '/^(.*)\.(mirabeta|nexttide)\.org$/', $wi->server ) ? 'beta/' : '';
+
 $wgEventStreams = [
 	'/^mediawiki\\.job\\..+/' => [
-		'schema_title' => 'mediawiki/job',
+		'schema_title' => "mediawiki/{$betaStream}job",
+		'destination_event_service' => 'eventgate',
+		'canary_events_enabled' => false,
+	],
+	'/^mediawiki\\.beta\\.job\\..+/' => [
+		'schema_title' => "mediawiki/{$betaStream}job",
 		'destination_event_service' => 'eventgate',
 		'canary_events_enabled' => false,
 	],
 	'mediawiki.centralnotice.campaign-change' => [
-		'schema_title' => 'mediawiki/centralnotice/campaign/change',
+		'schema_title' => "mediawiki/{$betaStream}centralnotice/campaign/change",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.centralnotice.campaign-create' => [
-		'schema_title' => 'mediawiki/centralnotice/campaign/create',
+		'schema_title' => "mediawiki/{$betaStream}centralnotice/campaign/create",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.centralnotice.campaign-delete' => [
-		'schema_title' => 'mediawiki/centralnotice/campaign/delete',
+		'schema_title' => "mediawiki/{$betaStream}centralnotice/campaign/delete",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.cirrussearch.page_rerender.v1' => [
-		'schema_title' => 'mediawiki/cirrussearch/page_rerender',
+		'schema_title' => "mediawiki/{$betaStream}cirrussearch/page_rerender",
 		'destination_event_service' => 'eventgate',
 		'message_key_fields' => [
 			'wiki_id' => 'wiki_id',
@@ -36,79 +43,79 @@
 		],
 	],
 	'mediawiki.page-create' => [
-		'schema_title' => 'mediawiki/revision/create',
+		'schema_title' => "mediawiki/{$betaStream}revision/create",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.page-delete' => [
-		'schema_title' => 'mediawiki/page/delete',
+		'schema_title' => "mediawiki/{$betaStream}page/delete",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.page-links-change' => [
-		'schema_title' => 'mediawiki/page/links-change',
+		'schema_title' => "mediawiki/{$betaStream}page/links-change",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.page-move' => [
-		'schema_title' => 'mediawiki/page/move',
+		'schema_title' => "mediawiki/{$betaStream}page/move",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.page-properties-change' => [
-		'schema_title' => 'mediawiki/page/properties-change',
+		'schema_title' => "mediawiki/{$betaStream}page/properties-change",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.page-restrictions-change' => [
-		'schema_title' => 'mediawiki/page/restrictions-change',
+		'schema_title' => "mediawiki/{$betaStream}page/restrictions-change",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.page-suppress' => [
-		'schema_title' => 'mediawiki/page/delete',
+		'schema_title' => "mediawiki/{$betaStream}page/delete",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.page-undelete' => [
-		'schema_title' => 'mediawiki/page/undelete',
+		'schema_title' => "mediawiki/{$betaStream}page/undelete",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.recentchange' => [
-		'schema_title' => 'mediawiki/recentchange',
+		'schema_title' => "mediawiki/{$betaStream}recentchange",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.revision-create' => [
-		'schema_title' => 'mediawiki/revision/create',
+		'schema_title' => "mediawiki/{$betaStream}revision/create",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.revision-tags-change' => [
-		'schema_title' => 'mediawiki/revision/tags-change',
+		'schema_title' => "mediawiki/{$betaStream}revision/tags-change",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.revision-visibility-change' => [
-		'schema_title' => 'mediawiki/revision/visibility-change',
+		'schema_title' => "mediawiki/{$betaStream}revision/visibility-change",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.user-blocks-change' => [
-		'schema_title' => 'mediawiki/user/blocks-change',
+		'schema_title' => "mediawiki/{$betaStream}user/blocks-change",
 		'destination_event_service' => 'eventgate',
 	],
 	'resource_change' => [
-		'schema_title' => 'resource_change',
+		'schema_title' => "{$betaStream}resource_change",
 		'destination_event_service' => 'eventgate',
 		'canary_events_enabled' => false,
 	],
 	'resource-purge' => [
-		'schema_title' => 'resource_change',
+		'schema_title' => "{$betaStream}resource_change",
 		'destination_event_service' => 'eventgate',
 		'canary_events_enabled' => false,
 	],
 	'change-prop.transcludes.resource-change' => [
-		'schema_title' => 'resource_change',
+		'schema_title' => "{$betaStream}resource_change",
 		'destination_event_service' => 'eventgate',
 		'canary_events_enabled' => false,
 	],
 	// These are logging channels
 	'mediawiki.api-request' => [
-		'schema_title' => 'mediawiki/api/request',
+		'schema_title' => "mediawiki/{$betaStream}api/request",
 		'destination_event_service' => 'eventgate',
 	],
 	'mediawiki.cirrussearch-request' => [
-		'schema_title' => 'mediawiki/cirrussearch/request',
+		'schema_title' => "mediawiki/{$betaStream}cirrussearch/request",
 		'destination_event_service' => 'eventgate',
 	],
 ];

From 511af7c0c244218fcffb58648b3366e5718c9436 Mon Sep 17 00:00:00 2001
From: paladox <paladox@users.noreply.github.com>
Date: Tue, 24 Sep 2024 17:43:05 +0100
Subject: [PATCH 7/7] Update EventStreamConfig.php

---
 EventStreamConfig.php | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/EventStreamConfig.php b/EventStreamConfig.php
index d1d68b3ef5..8546b478f4 100644
--- a/EventStreamConfig.php
+++ b/EventStreamConfig.php
@@ -13,12 +13,12 @@
 
 $wgEventStreams = [
 	'/^mediawiki\\.job\\..+/' => [
-		'schema_title' => "mediawiki/{$betaStream}job",
+		'schema_title' => "mediawiki/job",
 		'destination_event_service' => 'eventgate',
 		'canary_events_enabled' => false,
 	],
 	'/^mediawiki\\.beta\\.job\\..+/' => [
-		'schema_title' => "mediawiki/{$betaStream}job",
+		'schema_title' => "mediawiki/beta/job",
 		'destination_event_service' => 'eventgate',
 		'canary_events_enabled' => false,
 	],