Merge "Abstracting MySQL in Echo so storage type can be swapped easily"

Echo.php

@@ -49,6 +49,7 @@ $wgAutoloadClasses['EchoSubscription'] = $dir . 'model/Subscription.php';
 $wgAutoloadClasses['EchoEvent'] = $dir . 'model/Event.php';
 $wgAutoloadClasses['EchoNotification'] = $dir . 'model/Notification.php';
 $wgAutoloadClasses['MWEchoEmailBatch'] = $dir . 'includes/EmailBatch.php';
+$wgAutoloadClasses['MWDbEchoEmailBatch'] = $dir . 'includes/DbEmailBatch.php';
 
 // Formatters
 $wgAutoloadClasses['EchoNotificationFormatter'] = $dir . 'formatters/NotificationFormatter.php';
@@ -75,6 +76,10 @@ $wgAutoloadClasses['SpecialNotifications'] = $dir . 'special/SpecialNotification
 $wgSpecialPages['Notifications'] = 'SpecialNotifications';
 $wgSpecialPageGroups['Notifications'] = 'users';
 
+// Backend support
+$wgAutoloadClasses['MWEchoBackend'] = $dir . 'includes/EchoBackend.php';
+$wgAutoloadClasses['MWDbEchoBackend'] = $dir . 'includes/DbEchoBackend.php';
+
 // Housekeeping hooks
 $wgHooks['LoadExtensionSchemaUpdates'][] = 'EchoHooks::getSchemaUpdates';
 $wgHooks['GetPreferences'][] = 'EchoHooks::getPreferences';
@@ -83,6 +88,9 @@ $wgHooks['BeforePageDisplay'][] = 'EchoHooks::beforePageDisplay';
 $wgHooks['MakeGlobalVariablesScript'][] = 'EchoHooks::makeGlobalVariablesScript';
 $wgHooks['UnitTestsList'][] = 'EchoHooks::getUnitTests';
 
+// Extension initialization
+$wgExtensionFunctions[] = 'EchoHooks::initEchoExtension';
+
 $echoResourceTemplate = array(
 	'localBasePath' => $dir . 'modules',
 	'remoteExtPath' => 'Echo/modules',
@@ -160,6 +168,12 @@ $wgHooks['LinksUpdateAfterInsert'][] = 'EchoHooks::onLinksUpdateAfterInsert';
 
 // Configuration
 
+// The name of the backend to use for Echo, eg, Db, Redis, Zeromq
+$wgEchoBackendName = 'Db';
+
+// The backend object
+$wgEchoBackend = null;
+
 // Whether to turn on email batch function
 $wgEchoEnableEmailBatch = true;
 

Hooks.php

@@ -6,6 +6,15 @@ class EchoHooks {
 	const EMAIL_DAILY_DIGEST = 1; // Send daily email digests
 	const EMAIL_WEEKLY_DIGEST = 7; // Send weekly email digests
 
+	/**
+	 * Initialize Echo extension with necessary data, this function is invoked
+	 * from $wgExtensionFunctions
+	 */
+	public static function initEchoExtension() {
+		global $wgEchoBackend, $wgEchoBackendName;
+		$wgEchoBackend = MWEchoBackend::factory( $wgEchoBackendName );
+	}
+
 	/**
 	 * @param $updater DatabaseUpdater object
 	 * @return bool true in all cases

api/ApiEchoNotifications.php

@@ -55,42 +55,13 @@ class ApiEchoNotifications extends ApiQueryBase {
 	 * @return array
 	 */
 	public static function getNotifications( $user, $unread = false, $format = false, $limit = 20, $timestamp = 0, $offset = 0 ) {
-		global $wgEchoEventDetails;
+		global $wgEchoBackend;
+
 		$lang = RequestContext::getMain()->getLanguage();
-		$dbr = wfGetDB( DB_SLAVE );
 
 		$output = array();
 
-		$conds = array(
-			'notification_user' => $user->getID(),
-			'event_type' => EchoEvent::gatherValidEchoEvents(),
-		);
-
-		if ( $unread ) {
-			$conds['notification_read_timestamp'] = null;
-		}
-
-		// start points are specified
-		if ( $timestamp && $offset ) {
-			$conds[] = 'notification_timestamp <= ' . $dbr->addQuotes( $dbr->timestamp( $timestamp ) );
-			$conds[] = 'notification_event < ' . intval( $offset );
-		}
-
-		$res = $dbr->select(
-			array( 'echo_notification', 'echo_event' ),
-			'*',
-			$conds,
-			__METHOD__,
-			array(
-				// Todo: check if key ( user, timestamp ) is sufficient, if not,
-				// we need to replace it with ( user, timestamp, event )
-				'ORDER BY' => 'notification_timestamp DESC, notification_event DESC',
-				'LIMIT' => $limit,
-			),
-			array(
-				'echo_event' => array( 'LEFT JOIN', 'notification_event=event_id' ),
-			)
-		);
+		$res = $wgEchoBackend->loadNotifications( $user, $unread, $limit = 20, $timestamp = 0, $offset = 0 );
 
 		foreach ( $res as $row ) {
 			// Make sure the user is eligible to recieve this type of notification

controller/NotificationController.php

@@ -10,7 +10,7 @@ class EchoNotificationController {
 	 * @return Integer: Number of unread notifications.
 	 */
 	public static function getNotificationCount( $user, $cached = true, $dbSource = DB_SLAVE ) {
-		global $wgMemc;
+		global $wgMemc, $wgEchoBackend;
 
 		if ( $user->isAnon() ) {
 			return 0;
@@ -22,26 +22,7 @@ class EchoNotificationController {
 			return $wgMemc->get( $memcKey );
 		}
 
-		// double check
-		if ( !in_array( $dbSource, array( DB_SLAVE, DB_MASTER ) ) ) {
-			$dbSource = DB_SLAVE;
-		}
-
-		$db = wfGetDB( $dbSource );
-		$res = $db->selectRow(
-			array( 'echo_notification', 'echo_event' ),
-			array( 'num' => 'COUNT(notification_event)' ),
-			array(
-				'notification_user' => $user->getId(),
-				'notification_read_timestamp' => null,
-				'event_type' => EchoEvent::gatherValidEchoEvents(),
-			),
-			__METHOD__,
-			array(),
-			array(
-				'echo_event' => array( 'LEFT JOIN', 'notification_event=event_id' ),
-			)
-		);
+		$res = $wgEchoBackend->getNotificationCount( $user, $dbSource );
 
 		if ( $res ) {
 			$count = $res->num;
@@ -116,19 +97,13 @@ class EchoNotificationController {
 	 * @param $eventIDs Array of event IDs to mark read
 	 */
 	public static function markRead( $user, $eventIDs ) {
-		$dbw = wfGetDB( DB_MASTER );
+		global $wgEchoBackend;
 
 		$eventIDs = array_filter( (array)$eventIDs, 'is_numeric' );
-
-		$dbw->update( 'echo_notification',
-			array( 'notification_read_timestamp' => $dbw->timestamp( wfTimestampNow() ) ),
-			array(
-				'notification_user' => $user->getId(),
-				'notification_event' => $eventIDs,
-			),
-			__METHOD__
-		);
-
+		if ( !$eventIDs ) {
+			return;
+		}
+		$wgEchoBackend->markRead( $user, $eventIDs );
 		self::resetNotificationCount( $user, DB_MASTER );
 	}
 
@@ -208,7 +183,7 @@ class EchoNotificationController {
 	 * @return Array of EchoSubscription objects.
 	 */
 	protected static function getSubscriptionsForEvent( $event ) {
-		$dbr = wfGetDB( DB_SLAVE );
+		global $wgEchoBackend;
 
 		$conds = array( 'sub_event_type' => $event->getType() );
 
@@ -217,8 +192,7 @@ class EchoNotificationController {
 			$conds['sub_page_title'] = $event->getTitle()->getDBkey();
 		}
 
-		$res = $dbr->select( 'echo_subscription', '*', $conds, __METHOD__,
-			array( 'order by' => 'sub_user asc' ) );
+		$res = $wgEchoBackend->loadSubscription( $conds );
 
 		$subscriptions = array();
 		$rowCollection = array();

includes/DbEchoBackend.php

@@ -0,0 +1,193 @@
+<?php
+
+/**
+ * Database backend for echo notification
+ */
+class MWDbEchoBackend extends MWEchoBackend {
+
+	/**
+	 * Database object
+	 */
+	private $dbr;
+	private $dbw;
+
+	protected function __construct() {
+		$this->dbr = wfGetDB( DB_SLAVE );
+		$this->dbw = wfGetDB( DB_MASTER );
+	}
+
+	/**
+	 * @param $row array
+	 */
+	public function createNotification( $row ) {
+		$row['notification_timestamp'] = $this->dbw->timestamp( $row['notification_timestamp'] );
+		$this->dbw->insert( 'echo_notification', $row, __METHOD__ );
+	}
+
+	/**
+	 * @param $user User the user to get notifications for
+	 * @param $unread bool true to get only unread notifications
+	 * @param $limit int The maximum number of notifications to return
+	 * @param $timestamp int The timestamp to start from
+	 * @param $offset int The notification event id to start from
+	 * @return array
+	 */
+	public function loadNotifications( $user, $unread, $limit, $timestamp, $offset ) {
+		$conds = array(
+			'notification_user' => $user->getID(),
+			'event_type' => EchoEvent::gatherValidEchoEvents(),
+		);
+
+		if ( $unread ) {
+			$conds['notification_read_timestamp'] = null;
+		}
+
+		// start points are specified
+		if ( $timestamp && $offset ) {
+			$conds[] = 'notification_timestamp <= ' . $this->dbr->addQuotes( $this->dbr->timestamp( $timestamp ) );
+			$conds[] = 'notification_event < ' . intval( $offset );
+		}
+
+		$res = $this->dbr->select(
+			array( 'echo_notification', 'echo_event' ),
+			'*',
+			$conds,
+			__METHOD__,
+			array(
+				// Todo: check if key ( user, timestamp ) is sufficient, if not,
+				// we need to replace it with ( user, timestamp, event )
+				'ORDER BY' => 'notification_timestamp DESC, notification_event DESC',
+				'LIMIT' => $limit,
+			),
+			array(
+				'echo_event' => array( 'LEFT JOIN', 'notification_event=event_id' ),
+			)
+		);
+
+		return iterator_to_array( $res, false );
+	}
+
+	/**
+	 * @param $row array
+	 * @return int
+	 */
+	public function createEvent( $row ) {
+		$id = $this->dbw->nextSequenceValue( 'echo_event_id' );
+
+		$row['event_timestamp'] = $this->dbw->timestamp( $row['event_timestamp'] );
+
+		$this->dbw->insert( 'echo_event', $row, __METHOD__ );
+
+		if ( !$id ) {
+			$id = $this->dbw->insertId();
+		}
+
+		return $id;
+	}
+
+	/**
+	 * @param $id int
+	 * @param $fromMaster bool
+	 * @return ResultWrapper
+	 * @throws MWException
+	 */
+	public function loadEvent( $id, $fromMaster = false ) {
+		$db = $fromMaster ? $this->dbw : $this->dbr;
+
+		$row = $db->selectRow( 'echo_event', '*', array( 'event_id' => $id ), __METHOD__ );
+
+		if ( !$row && !$fromMaster ) {
+			return $this->loadEvent( $id, true );
+		} elseif ( !$row ) {
+			throw new MWException( "No EchoEvent found with ID: $id" );
+		}
+
+		return $row;
+	}
+
+	/**
+	 * @param $event EchoEvent
+	 */
+	public function updateEventExtra( $event ) {
+		$this->dbw->update(
+			'echo_event',
+			array( 'event_extra' => $event->serializeExtra() ),
+			array( 'event_id' => $event->getId() ),
+			__METHOD__
+		);
+	}
+
+	/**
+	 * @param $conds array
+	 * @param $rows array
+	 */
+	public function createSubscription( $conds, $rows ) {
+		$this->dbw->begin();
+		$this->dbw->delete( 'echo_subscription', $conds, __METHOD__ );
+
+		if ( count( $rows ) ) {
+			$this->dbw->insert( 'echo_subscription', $rows, __METHOD__ );
+		}
+
+		$this->dbw->commit();
+	}
+
+	/**
+	 * @param $conds array
+	 * @return ResultWrapper
+	 */
+	public function loadSubscription( $conds ) {
+		$res = $this->dbr->select( 'echo_subscription', '*', $conds, __METHOD__, array( 'order by' => 'sub_user asc' ) );
+		return $res;
+	}
+
+	/**
+	 * @param $user User
+	 * @param $eventIDs array
+	 */
+	public function markRead( $user, $eventIDs ) {
+		if ( !$eventIDs ) {
+			return;
+		}
+		$this->dbw->update(
+			'echo_notification',
+			array( 'notification_read_timestamp' => $this->dbw->timestamp( wfTimestampNow() ) ),
+			array(
+				'notification_user' => $user->getId(),
+				'notification_event' => $eventIDs,
+			),
+			__METHOD__
+		);
+	}
+
+	/**
+	 * @param $user User object to check notifications for
+	 * @param $dbSource string use master or slave storage to pull count
+	 * @return ResultWrapper|bool
+	 */
+	public function getNotificationCount( $user, $dbSource ) {
+		// double check
+		if ( !in_array( $dbSource, array( DB_SLAVE, DB_MASTER ) ) ) {
+			$dbSource = DB_SLAVE;
+		}
+
+		$db = wfGetDB( $dbSource );
+		$res = $db->selectRow(
+			array( 'echo_notification', 'echo_event' ),
+			array( 'num' => 'COUNT(notification_event)' ),
+			array(
+				'notification_user' => $user->getId(),
+				'notification_read_timestamp' => null,
+				'event_type' => EchoEvent::gatherValidEchoEvents(),
+			),
+			__METHOD__,
+			array(),
+			array(
+				'echo_event' => array( 'LEFT JOIN', 'notification_event=event_id' ),
+			)
+		);
+
+		return $res;
+	}
+
+}

includes/DbEmailBatch.php

@@ -0,0 +1,180 @@
+<?php
+
+/**
+ * Handles email batch for a user for database storage
+ */
+class MWDbEchoEmailBatch extends MWEchoEmailBatch {
+
+	/**
+	 * Set the last event of this batch, this is a cutoff point for clearing
+	 * processed/invalid events
+	 *
+	 * @return bool true if event exists false otherwise
+	 */
+	protected function setLastEvent() {
+		$dbr = wfGetDB( DB_SLAVE );
+		$res = $dbr->selectField(
+			array( 'echo_email_batch' ),
+			array( 'MAX( eeb_event_id )' ),
+			array( 'eeb_user_id' => $this->mUser->getId() ),
+			__METHOD__
+		);
+
+		if ( $res ) {
+			$this->lastEvent = $res;
+			return true;
+		} else {
+			return false;
+		}
+	}
+
+	/**
+	 * Get the events queued for the current user
+	 * @return array
+	 */
+	protected function getEvents() {
+		$events = array();
+
+		$validEvents = EchoEvent::gatherValidEchoEvents();
+
+		if ( $validEvents ) {
+			$dbr = wfGetDB( DB_SLAVE );
+
+			$conds = array(
+				'eeb_user_id' => $this->mUser->getId(),
+				'event_id = eeb_event_id',
+				'event_type' => $validEvents
+			);
+
+			if ( $this->lastEvent ) {
+				$conds[] = 'eeb_event_id <= ' . intval( $this->lastEvent );
+			}
+
+			$res = $dbr->select(
+				array( 'echo_email_batch', 'echo_event' ),
+				array( '*' ),
+				$conds,
+				__METHOD__,
+				array( 'ORDER BY' => 'eeb_event_priority, eeb_event_id', 'LIMIT' => self::$displaySize + 1 )
+			);
+
+			foreach( $res as $row ) {
+				$events[$row->eeb_id] = $row;
+			}
+		}
+
+		return $events;
+	}
+
+	/**
+	 * Clear "processed" events in the queue, processed could be: email sent, invalid, users do not want to receive emails
+	 */
+	public function clearProcessedEvent() {
+		$conds = array( 'eeb_user_id' => $this->mUser->getId() );
+
+		// there is a processed cutoff point
+		if ( $this->lastEvent ) {
+			$conds[] = 'eeb_event_id <= ' . intval( $this->lastEvent );
+		}
+
+		$dbw = wfGetDB( DB_MASTER );
+		$dbw->delete(
+			'echo_email_batch',
+			$conds,
+			__METHOD__,
+			array()
+		);
+	}
+
+	/**
+	 * Send the batch email
+	 */
+	public function sendEmail() {
+		global $wgPasswordSender, $wgPasswordSenderName, $wgEchoEmailFooterAddress;
+
+		// global email footer
+		$footer = wfMessage( 'echo-email-footer-default' )
+				->inLanguage( $this->mUser->getOption( 'language' ) )
+				->params( $wgEchoEmailFooterAddress, '' )
+				->text();
+
+		// @Todo - replace them with the CONSTANT in 33810 once it is merged
+		if ( $this->mUser->getOption( 'echo-email-frequency' ) == 7 ) {
+			$frequency = 'weekly';
+		} else {
+			$frequency = 'daily';
+		}
+
+		// email subject
+		if ( $this->count > self::$displaySize ) {
+			$count = wfMessage( 'echo-notification-count' )->params( self::$displaySize )->text();
+		} else {
+			$count = $this->count;
+		}
+		$subject = wfMessage( 'echo-email-batch-subject-' . $frequency )->params( $count, $this->count )->text();
+		$body = wfMessage( 'echo-email-batch-body-' . $frequency )->params(
+				$this->mUser->getName(),
+				$count,
+				$this->count,
+				$this->listToText(),
+				$footer
+			)->text();
+
+		$adminAddress = new MailAddress( $wgPasswordSender, $wgPasswordSenderName );
+		$address = new MailAddress( $this->mUser );
+
+		$params = array(
+			'to' => $address,
+			'from' => $adminAddress,
+			'subj' => $subject,
+			'body' => $body,
+			// no replyto
+			'replyto' => ''
+		);
+		$job = new EmaillingJob( null, $params );
+		JobQueueGroup::singleton()->push( $job );
+	}
+
+	/**
+	 * Insert notification event into email queue
+	 * @param $userId int
+	 * @param $eventId int
+	 * @param $priority int
+	 */
+	public static function actuallyAddToQueue( $userId, $eventId, $priority ) {
+		if ( !$userId || !$eventId ) {
+			return;
+		}
+
+		$dbw = wfGetDB( DB_MASTER );
+		$dbw->insert(
+			'echo_email_batch',
+			array(
+				'eeb_user_id' => $userId,
+				'eeb_event_id' => $eventId,
+				'eeb_event_priority' => $priority
+			),
+			__METHOD__,
+			array( 'IGNORE' )
+		);
+	}
+
+	/**
+	 * Get a list of users to be notified for the batch
+	 * @param $startUserId int
+	 * @param $batchSize int
+	 */
+	public static function actuallyGetUsersToNotify( $startUserId, $batchSize ) {
+		$dbr = wfGetDB( DB_MASTER );
+		$res = $dbr->select(
+			array( 'echo_email_batch' ),
+			array( 'eeb_user_id' ),
+			array( 'eeb_user_id > ' . $startUserId  ),
+			__METHOD__,
+			array( 'ORDER BY' => 'eeb_user_id', 'LIMIT' => $batchSize )
+		);
+
+		return $res;
+	}
+
+}

includes/EchoBackend.php

@@ -0,0 +1,100 @@
+<?php
+
+/**
+ * Base backend class for accessing and saving echo notification data,
+ * this class should only provide all the necessary interfaces and
+ * implementation should be provided in each child class
+ */
+abstract class MWEchoBackend {
+
+	private static $cache = array();
+
+	/**
+	 * Factory to initialize a backend class
+	 * @param $backend string
+	 * @return MWEchoBackend
+	 * @throws MWException
+	 */
+	public static function factory( $backend ) {
+		$backend = strval( $backend );
+
+		$className = 'MW' . $backend . 'EchoBackend';
+
+		if ( !class_exists( $className ) ) {
+			throw new MWException( "$backend backend is not supported" );
+		}
+
+		if ( !isset( self::$cache[$backend] ) ) {
+			self::$cache[$backend] = new $className();
+		}
+
+		return self::$cache[$backend];
+	}
+
+	/**
+	 * Create a new notification
+	 * @param $row array
+	 */
+	abstract public function createNotification( $row );
+
+	/**
+	 * Load notifications based on the parameters
+	 * @param $user User the user to get notifications for
+	 * @param $unread bool true to get only unread notifications
+	 * @param $limit int The maximum number of notifications to return
+	 * @param $timestamp int The timestamp to start from
+	 * @param $offset int The notification event id to start from
+	 * @return array
+	 */
+	abstract public function loadNotifications( $user, $unread, $limit, $timestamp, $offset );
+
+	/**
+	 * Create an Echo event
+	 * @param $row array
+	 * @return int
+	 */
+	abstract public function createEvent( $row );
+
+	/**
+	 * Load an Echo event
+	 * @param $id int
+	 * @param $fromMaster bool
+	 */
+	abstract public function loadEvent( $id, $fromMaster );
+
+	/**
+	 * Update the extra data for an Echo event
+	 * @param $event EchoEvent
+	 */
+	abstract public function updateEventExtra( $event );
+
+	/**
+	 * Create an Echo subscription
+	 * @param $conds array
+	 * @param $rows array
+	 */
+	abstract public function createSubscription( $conds, $rows );
+
+	/**
+	 * Load an Echo subscription
+	 * @param $conds array
+	 * @return ResultWrapper
+	 */
+	abstract public function loadSubscription( $conds );
+
+	/**
+	 * Mark notifications as read for a user
+	 * @param $user User
+	 * @param $eventIDs array
+	 */
+	abstract public function markRead( $user, $eventIDs );
+
+	/**
+	 * Retrieves number of unread notifications that a user has.
+	 * @param $user User object to check notifications for
+	 * @param $dbSource string use master or slave storage to pull count
+	 * @return ResultWrapper|bool
+	 */
+	abstract public function getNotificationCount( $user, $dbSource );
+
+}

includes/EmailBatch.php

@@ -1,9 +1,9 @@
 <?php
 
 /**
- * Handles email batch for a user
+ * Handles email batch for a user for database storage
  */
-class MWEchoEmailBatch {
+abstract class MWEchoEmailBatch {
 
 	// the user to be notified
 	protected $mUser;
@@ -38,13 +38,15 @@ class MWEchoEmailBatch {
 	 * @return MWEchoEmailBatch/false
 	 */
 	public static function newFromUserId( $userId ) {
+		$batchClassName = self::getEmailBatchCalss();
+
 		$user = User::newFromId( intval( $userId ) );
 
 		$userEmailSetting = intval( $user->getOption( 'echo-email-frequency' ) );
 
 		// clear all existing events if user decides not to receive emails
 		if ( $userEmailSetting == -1 ) {
-			$emailBatch = new MWEchoEmailBatch( $user );
+			$emailBatch = new $batchClassName( $user );
 			$emailBatch->clearProcessedEvent();
 			return false;
 		}
@@ -63,7 +65,24 @@ class MWEchoEmailBatch {
 			}
 		}
 
-		return new MWEchoEmailBatch( $user );
+		return new $batchClassName( $user );
+	}
+
+	/**
+	 * Get the name of the email batch class
+	 * @return string
+	 * @throws MWException
+	 */
+	private static function getEmailBatchCalss() {
+		global $wgEchoBackendName;
+
+		$className = 'MW' . $wgEchoBackendName . 'EchoEmailBatch';
+
+		if ( !class_exists( $className ) ) {
+			throw new MWException( "$wgEchoBackendName email batch is not supported!" );
+		}
+
+		return $className;
 	}
 
 	/**
@@ -105,22 +124,7 @@ class MWEchoEmailBatch {
 	 *
 	 * @return bool true if event exists false otherwise
 	 */
-	protected function setLastEvent() {
-		$dbr = wfGetDB( DB_SLAVE );
-		$res = $dbr->selectField(
-			array( 'echo_email_batch' ),
-			array( 'MAX( eeb_event_id )' ),
-			array( 'eeb_user_id' => $this->mUser->getId() ),
-			__METHOD__
-		);
-
-		if ( $res ) {
-			$this->lastEvent = $res;
-			return true;
-		} else {
-			return false;
-		}
-	}
+	abstract protected function setLastEvent();
 
 	/**
 	 * Update the user's last batch timestamp after a successful batch
@@ -135,39 +139,7 @@ class MWEchoEmailBatch {
 	 * Get the events queued for the current user
 	 * @return array
 	 */
-	protected function getEvents() {
-		$events = array();
-
-		$validEvents = EchoEvent::gatherValidEchoEvents();
-
-		if ( $validEvents ) {
-			$dbr = wfGetDB( DB_SLAVE );
-
-			$conds = array(
-				'eeb_user_id' => $this->mUser->getId(),
-				'event_id = eeb_event_id',
-				'event_type' => $validEvents
-			);
-
-			if ( $this->lastEvent ) {
-				$conds[] = 'eeb_event_id <= ' . intval( $this->lastEvent );
-			}
-
-			$res = $dbr->select(
-				array( 'echo_email_batch', 'echo_event' ),
-				array( '*' ),
-				$conds,
-				__METHOD__,
-				array( 'ORDER BY' => 'eeb_event_priority, eeb_event_id', 'LIMIT' => self::$displaySize + 1 )
-			);
-
-			foreach( $res as $row ) {
-				$events[$row->eeb_id] = $row;
-			}
-		}
-
-		return $events;
-	}
+	abstract protected function getEvents();
 
 	/**
 	 * Add individual event template to the big email content
@@ -222,22 +194,7 @@ class MWEchoEmailBatch {
 	/**
 	 * Clear "processed" events in the queue, processed could be: email sent, invalid, users do not want to receive emails
 	 */
-	public function clearProcessedEvent() {
-		$conds = array( 'eeb_user_id' => $this->mUser->getId() );
-
-		// there is a processed cutoff point
-		if ( $this->lastEvent ) {
-			$conds[] = 'eeb_event_id <= ' . intval( $this->lastEvent );
-		}
-
-		$dbw = wfGetDB( DB_MASTER );
-		$dbw->delete(
-			'echo_email_batch',
-			$conds,
-			__METHOD__,
-			array()
-		);
-	}
+	abstract public function clearProcessedEvent();
 
 	/**
 	 * Send the batch email
@@ -295,20 +252,27 @@ class MWEchoEmailBatch {
 	 * @param $priority int
 	 */
 	public static function addToQueue( $userId, $eventId, $priority ) {
-		if ( !$userId || !$eventId ) {
-			return;
+		$batchClassName = self::getEmailBatchCalss();
+
+		if ( !method_exists( $batchClassName, 'actuallyAddToQueue' ) ) {
+			throw new MWException( "$batchClassName must implement method actuallyAddToQueue()" );
 		}
 
-		$dbw = wfGetDB( DB_MASTER );
-		$dbw->insert(
-			'echo_email_batch',
-			array(
-				'eeb_user_id' => $userId,
-				'eeb_event_id' => $eventId,
-				'eeb_event_priority' => $priority
-			),
-			__METHOD__,
-			array( 'IGNORE' )
-		);
+		$batchClassName::actuallyAddToQueue( $userId, $eventId, $priority );
+	}
+
+	/**
+	 * Get a list of users to be notified for the batch
+	 * @param $startUserId int
+	 * @param $batchSize int
+	 */
+	public static function getUsersToNotify( $startUserId, $batchSize ) {
+		$batchClassName = self::getEmailBatchCalss();
+
+		if ( !method_exists( $batchClassName, 'actuallyGetUsersToNotify' ) ) {
+			throw new MWException( "$batchClassName must implement method actuallyGetUsersToNotify()" );
+		}
+
+		return $batchClassName::actuallyGetUsersToNotify( $startUserId, $batchSize );
 	}
 }

model/Event.php

@@ -140,17 +140,15 @@ class EchoEvent {
 	 * Inserts the object into the database.
 	 */
 	protected function insert() {
-		$dbw = wfGetDB( DB_MASTER );
+		global $wgEchoBackend;
 
 		if ( $this->id ) {
 			throw new MWException( "Attempt to insert() an existing event" );
 		}
 
-		$this->id = $dbw->nextSequenceValue( 'echo_event_id' );
-
 		$row = array(
 			'event_id' => $this->id,
-			'event_timestamp' => $dbw->timestamp( $this->timestamp ),
+			'event_timestamp' => $this->timestamp,
 			'event_type' => $this->type,
 			'event_variant' => $this->variant,
 		);
@@ -170,11 +168,7 @@ class EchoEvent {
 			$row['event_page_title'] = $this->title->getDBkey();
 		}
 
-		$dbw->insert( 'echo_event', $row, __METHOD__ );
-
-		if ( !$this->id ) {
-			$this->id = $dbw->insertId();
-		}
+		$this->id = $wgEchoBackend->createEvent( $row );
 	}
 
 	/**
@@ -207,20 +201,11 @@ class EchoEvent {
 	 * Loads data from the database into this object, given the event ID.
 	 * @param $id int Event ID
 	 * @param $fromMaster bool
-	 * @throws MWException
 	 */
 	public function loadFromID( $id, $fromMaster = false ) {
-		$db = wfGetDB( $fromMaster ? DB_MASTER : DB_SLAVE );
+		global $wgEchoBackend;
 
-		$row = $db->selectRow( 'echo_event', '*', array( 'event_id' => $id ), __METHOD__ );
-
-		if ( !$row && !$fromMaster ) {
-			$this->loadFromID( $id, true );
-		} elseif ( !$row ) {
-			throw new MWException( "No EchoEvent found with ID: $id" );
-		}
-
-		$this->loadFromRow( $row );
+		$this->loadFromRow( $wgEchoBackend->loadEvent( $id, $fromMaster ) );
 	}
 
 	/**
@@ -251,16 +236,11 @@ class EchoEvent {
 	 * Update extra data
 	 */
 	public function updateExtra( $extra ) {
-		$dbw = wfGetDB( DB_MASTER );
+		global $wgEchoBackend;
 
 		$this->extra = $extra;
 		if ( $this->id && $this->extra ) {
-			$dbw->update(
-				'echo_event',
-				array( 'event_extra' => $this->serializeExtra() ),
-				array( 'event_id' => $this->id ),
-				__METHOD__
-			);
+			$wgEchoBackend->updateEventExtra( $this );
 		}
 	}
 
@@ -268,7 +248,7 @@ class EchoEvent {
 	 * Serialize the extra data for event
 	 * @return string
 	 */
-	protected function serializeExtra() {
+	public function serializeExtra() {
 		if ( is_array( $this->extra ) || is_object( $this->extra ) ) {
 			$extra = serialize( $this->extra );
 		} elseif ( is_null( $this->extra ) ) {

model/Notification.php

@@ -52,18 +52,18 @@ class EchoNotification {
 	}
 
 	/**
-	 * Adds this new notification object to the database.
+	 * Adds this new notification object to the backend storage.
 	 */
 	protected function insert() {
-		$dbw = wfGetDB( DB_MASTER );
+		global $wgEchoBackend;
 
 		$row = array(
 			'notification_event' => $this->event->getId(),
 			'notification_user' => $this->user->getId(),
-			'notification_timestamp' => $dbw->timestamp( $this->timestamp ),
+			'notification_timestamp' => $this->timestamp,
 			'notification_read_timestamp' => $this->readTimestamp,
 		);
 
-		$dbw->insert( 'echo_notification', $row, __METHOD__ );
+		$wgEchoBackend->createNotification( $row );
 	}
 }

model/Subscription.php

@@ -139,11 +139,10 @@ class EchoSubscription {
 		if ( $this->loaded ) {
 			return;
 		}
-		$dbr = wfGetDB( DB_SLAVE );
+		global $wgEchoBackend;
 
 		$conds = $this->getConds();
-		$res = $dbr->select( 'echo_subscription', '*', $conds, __METHOD__ );
-
+		$res = $wgEchoBackend->loadSubscription( $conds );
 		$this->loadFromRows( $res );
 	}
 
@@ -201,11 +200,7 @@ class EchoSubscription {
 
 		$conds = $this->getConds();
 
-		$dbw = wfGetDB( DB_MASTER );
-		$dbw->begin();
-		$dbw->delete( 'echo_subscription', $conds, __METHOD__ );
-
-		global $wgEchoDefaultNotificationTypes;
+		global $wgEchoDefaultNotificationTypes, $wgEchoBackend;
 		if ( isset( $wgEchoDefaultNotificationTypes[$this->event] ) ) {
 			$defaultState = array_merge(
 				$wgEchoDefaultNotificationTypes['all'],
@@ -229,10 +224,6 @@ class EchoSubscription {
 			}
 		}
 
-		if ( count( $rows ) ) {
-			$dbw->insert( 'echo_subscription', $rows, __METHOD__ );
-		}
-
-		$dbw->commit();
+		$wgEchoBackend->createSubscription( $conds, $rows );
 	}
 }

processEchoEmailBatch.php

@@ -40,13 +40,8 @@ class processEchoEmailBatch extends Maintenance {
 
 		while ( $count === $this->batchSize ) {
 			$count = 0;
-			$res = $this->dbr->select(
-				array( 'echo_email_batch' ),
-				array( 'eeb_user_id' ),
-				array( 'eeb_user_id > ' . $startUserId  ),
-				__METHOD__,
-				array( 'ORDER BY' => 'eeb_user_id', 'LIMIT' => $this->batchSize )
-			);
+
+			$res = MWEchoEmailBatch::getUsersToNotify( $startUserId, $this->batchSize );
 
 			$updated = false;
 			foreach ( $res as $row ) {