ClearUserWatchlistJob.php

Go to the documentation of this file.

<?php
 
use MediaWiki\MediaWikiServices;
use MediaWiki\User\UserIdentity;
 
class ClearUserWatchlistJob extends Job implements GenericParameterJob {
    public function __construct( array $params ) {
        parent::__construct( 'clearUserWatchlist', $params );
 
        $this->removeDuplicates = true;
    }
 
    public static function newForUser( UserIdentity $user, $maxWatchlistId ) {
        return new self( [ 'userId' => $user->getId(), 'maxWatchlistId' => $maxWatchlistId ] );
    }
 
    public function run() {
        $updateRowsPerQuery = MediaWikiServices::getInstance()->getMainConfig()->get( 'UpdateRowsPerQuery' );
        $userId = $this->params['userId'];
        $maxWatchlistId = $this->params['maxWatchlistId'];
        $batchSize = $updateRowsPerQuery;
 
        $loadBalancer = MediaWikiServices::getInstance()->getDBLoadBalancer();
        $dbw = $loadBalancer->getConnectionRef( DB_PRIMARY );
        $dbr = $loadBalancer->getConnectionRef( DB_REPLICA, [ 'watchlist' ] );
 
        // Wait before lock to try to reduce time waiting in the lock.
        if ( !$loadBalancer->waitForPrimaryPos( $dbr ) ) {
            $this->setLastError( 'Timed out waiting for replica to catch up before lock' );
            return false;
        }
 
        // Use a named lock so that jobs for this user see each others' changes
        $lockKey = "{{$dbw->getDomainID()}}:ClearUserWatchlist:$userId"; // per-wiki
        $scopedLock = $dbw->getScopedLockAndFlush( $lockKey, __METHOD__, 10 );
        if ( !$scopedLock ) {
            $this->setLastError( "Could not acquire lock '$lockKey'" );
            return false;
        }
 
        if ( !$loadBalancer->waitForPrimaryPos( $dbr ) ) {
            $this->setLastError( 'Timed out waiting for replica to catch up within lock' );
            return false;
        }
 
        // Clear any stale REPEATABLE-READ snapshot
        $dbr->flushSnapshot( __METHOD__ );
 
        $watchlistIds = $dbr->selectFieldValues(
            'watchlist',
            'wl_id',
            [
                'wl_user' => $userId,
                'wl_id <= ' . $maxWatchlistId
            ],
            __METHOD__,
            [
                'LIMIT' => $batchSize,
            ]
        );
 
        if ( count( $watchlistIds ) == 0 ) {
            return true;
        }
 
        $dbw->delete( 'watchlist', [ 'wl_id' => $watchlistIds ], __METHOD__ );
        if ( MediaWikiServices::getInstance()->getMainConfig()->get( 'WatchlistExpiry' ) ) {
            $dbw->delete( 'watchlist_expiry', [ 'we_item' => $watchlistIds ], __METHOD__ );
        }
 
        // Commit changes and remove lock before inserting next job.
        $lbf = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
        $lbf->commitPrimaryChanges( __METHOD__ );
        unset( $scopedLock );
 
        if ( count( $watchlistIds ) === (int)$batchSize ) {
            // Until we get less results than the limit, recursively push
            // the same job again.
            JobQueueGroup::singleton()->push( new self( $this->getParams() ) );
        }
 
        return true;
    }
 
    public function getDeduplicationInfo() {
        $info = parent::getDeduplicationInfo();
        // This job never has a namespace or title so we can't use it for deduplication
        unset( $info['namespace'] );
        unset( $info['title'] );
        return $info;
    }
 
}