Extract subscribers to a separate table in the Bounce worker [MAILPOET-940]

lib/Models/ScheduledTask.php

@@ -11,6 +11,10 @@ class ScheduledTask extends Model {
   const PRIORITY_MEDIUM = 5;
   const PRIORITY_LOW = 10;
 
+  function taskSubscribers() {
+    return $this->has_many(__NAMESPACE__ . '\ScheduledTaskSubscriber', 'task_id', 'id');
+  }
+
   function complete() {
     $this->set('status', self::STATUS_COMPLETED);
     $this->save();
@@ -18,79 +22,28 @@ class ScheduledTask extends Model {
   }
 
   function save() {
-    if(!is_serialized($this->subscribers)) {
-      $this->set('subscribers', serialize($this->subscribers));
-    }
     // set the default priority to medium
     if(!$this->priority) {
       $this->priority = self::PRIORITY_MEDIUM;
     }
     parent::save();
-    $this->subscribers = $this->getSubscribers();
     return $this;
   }
 
-  function getSubscribers() {
-    if(!is_serialized($this->subscribers)) {
-      return $this->subscribers;
-    }
-    $subscribers = unserialize($this->subscribers);
-    if(empty($subscribers['processed'])) {
-      $subscribers['processed'] = array();
-    }
-    return $subscribers;
+  function updateProcessedSubscribers(array $processed_subscribers) {
+    $this->taskSubscribers()
+      ->whereIn('subscriber_id', $processed_subscribers)
+      ->findResultSet()
+      ->set('processed', ScheduledTaskSubscriber::STATUS_PROCESSED)
+      ->save();
+    $this->checkCompleted();
   }
 
-  function isSubscriberProcessed($subscriber_id) {
-    $subscribers = $this->getSubscribers();
-    return in_array($subscriber_id, $subscribers['processed']);
-  }
-
-  function asArray() {
-    $model = parent::asArray();
-    $model['subscribers'] = (is_serialized($this->subscribers))
-      ? unserialize($this->subscribers)
-      : $this->subscribers;
-    return $model;
-  }
-
-  function removeNonexistentSubscribers($subscribers_to_remove) {
-    $subscribers = $this->getSubscribers();
-    $subscribers['to_process'] = array_values(
-      array_diff(
-        $subscribers['to_process'],
-        $subscribers_to_remove
-      )
-    );
-    $this->subscribers = $subscribers;
-    $this->updateCount();
-  }
-
-  function updateProcessedSubscribers($processed_subscribers) {
-    $subscribers = $this->getSubscribers();
-    $subscribers['processed'] = array_merge(
-      $subscribers['processed'],
-      $processed_subscribers
-    );
-    $subscribers['to_process'] = array_values(
-      array_diff(
-        $subscribers['to_process'],
-        $processed_subscribers
-      )
-    );
-    $this->subscribers = $subscribers;
-    $this->updateCount();
-  }
-
-  function updateCount() {
-    $this->subscribers = $this->getSubscribers();
-    $this->count_processed = count($this->subscribers['processed']);
-    $this->count_to_process = count($this->subscribers['to_process']);
-    $this->count_total = $this->count_processed + $this->count_to_process;
-    if(!$this->count_to_process) {
+  private function checkCompleted() {
+    if(!ScheduledTaskSubscriber::getToProcessCount($this->id)) {
       $this->processed_at = current_time('mysql');
       $this->status = self::STATUS_COMPLETED;
+      return $this->save();
     }
-    return $this->save();
   }
-}
+}

lib/Models/ScheduledTaskSubscriber.php

@@ -0,0 +1,61 @@
+<?php
+namespace MailPoet\Models;
+
+if(!defined('ABSPATH')) exit;
+
+class ScheduledTaskSubscriber extends Model {
+  const STATUS_TO_PROCESS = 0;
+  const STATUS_PROCESSED = 1;
+
+  public static $_table = MP_SCHEDULED_TASK_SUBSCRIBERS_TABLE;
+  public static $_id_column = array('task_id', 'subscriber_id');
+
+  static function createOrUpdate($data = array()) {
+    if(!is_array($data) || empty($data['task_id']) || empty($data['subscriber_id'])) return;
+
+    $task_subscriber = self::where('subscriber_id', $data['subscriber_id'])
+      ->where('task_id', $data['task_id'])
+      ->findOne();
+
+    if(empty($task_subscriber)) $task_subscriber = self::create();
+
+    $task_subscriber->task_id = $data['task_id'];
+    $task_subscriber->subscriber_id = $data['subscriber_id'];
+    $task_subscriber->processed = self::STATUS_TO_PROCESS;
+    $task_subscriber->save();
+
+    return $task_subscriber;
+  }
+
+  static function addSubscribers($task_id, array $subscriber_ids) {
+    foreach($subscriber_ids as $subscriber_id) {
+      self::createOrUpdate(array(
+        'task_id' => $task_id,
+        'subscriber_id' => $subscriber_id
+      ));
+    }
+  }
+
+  static function getToProcessCount($task_id) {
+    return self::getCount($task_id, false);
+  }
+
+  static function getProcessedCount($task_id) {
+    return self::getCount($task_id, true);
+  }
+
+  static function getTotalCount($task_id) {
+    return self::getCount($task_id);
+  }
+
+  private static function getCount($task_id, $processed = null) {
+    $orm = self::where('task_id', $task_id);
+    if(!is_null($processed)) {
+      $orm->where(
+        'processed',
+        ($processed) ? self::STATUS_PROCESSED : self::STATUS_TO_PROCESS
+      );
+    }
+    return $orm->count();
+  }
+}