From c37679dabe3c95969460f4538d5969d30c224880 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Mon, 12 Aug 2024 15:10:54 +0200
Subject: [PATCH 01/13] feat(m2t_send): Copy drush script and turn it into a
 cron-job

---
 .../campaignion_m2t_send.info                 |  10 ++
 .../campaignion_m2t_send.module               |  43 ++++++
 .../campaignion_m2t_send.variable.inc         |  22 +++
 campaignion_m2t_send/src/SendMessagesCron.php | 146 ++++++++++++++++++
 4 files changed, 221 insertions(+)
 create mode 100644 campaignion_m2t_send/campaignion_m2t_send.info
 create mode 100644 campaignion_m2t_send/campaignion_m2t_send.module
 create mode 100644 campaignion_m2t_send/campaignion_m2t_send.variable.inc
 create mode 100644 campaignion_m2t_send/src/SendMessagesCron.php

diff --git a/campaignion_m2t_send/campaignion_m2t_send.info b/campaignion_m2t_send/campaignion_m2t_send.info
new file mode 100644
index 00000000..4b986cb4
--- /dev/null
+++ b/campaignion_m2t_send/campaignion_m2t_send.info
@@ -0,0 +1,10 @@
+name = Match to target send
+description = Send M2T messages asynchronously
+core = 7.x
+package = Campaignion
+
+dependencies[] = psr0:psr0
+dependencies[] = campaignion:campaignion_email_to_target
+dependencies[] = little_helpers:little_helpers
+dependencies[] = ultimate_cron:ultimate_cron
+dependencies[] = variable:variable
diff --git a/campaignion_m2t_send/campaignion_m2t_send.module b/campaignion_m2t_send/campaignion_m2t_send.module
new file mode 100644
index 00000000..c8472fa5
--- /dev/null
+++ b/campaignion_m2t_send/campaignion_m2t_send.module
@@ -0,0 +1,43 @@
+<?php
+
+/**
+ * @file
+ * Hook and callback implementations for this module.
+ */
+
+use Drupal\campaignion_m2t_send\SendMessagesCron;
+use Drupal\little_helpers\Services\Container;
+
+/**
+ * Implements hook_little_helpers_services().
+ */
+function campaignion_m2t_send_little_helpers_services() {
+  $info['campaignion_m2t_send.SendMessagesCron'] = [
+    'class' => SendMessagesCron::class,
+    'arguments' => [
+      '!campaignion_m2t_send_content_types',
+    ],
+  ];
+  return $info;
+}
+
+/**
+ * Implements hook_cronapi().
+ */
+function campaignion_m2t_send_cronapi($op, $job = NULL) {
+  $items['campaignion_m2t_send'] = [
+    'description' => 'Send m2t messages',
+    'rule' => '*+@ * * * *',
+    'weight' => 100,
+    'callback' => '_campaignion_m2t_send_run',
+    'arguments' => ['campaignion_m2t_send.SendMessagesCron'],
+  ];
+  return $items;
+}
+
+/**
+ * Helper function to load a cron service and invoke it.
+ */
+function _campaignion_m2t_send_run($service) {
+  Container::get()->loadService($service)->run();
+}
diff --git a/campaignion_m2t_send/campaignion_m2t_send.variable.inc b/campaignion_m2t_send/campaignion_m2t_send.variable.inc
new file mode 100644
index 00000000..3672903c
--- /dev/null
+++ b/campaignion_m2t_send/campaignion_m2t_send.variable.inc
@@ -0,0 +1,22 @@
+<?php
+
+/**
+ * @file
+ * Information about configuration variables defined by this module.
+ */
+
+/**
+ * Implements hook_variable_info().
+ */
+function campaignion_m2t_send_variable_info($options) {
+  $v['campaignion_m2t_send_content_types'] = [
+    'title' => t('Restrict to content-types'),
+    'description' => t('Only process actions of these types.'),
+    'type' => 'options',
+    'default' => ['match_to_target'],
+    'localize' => FALSE,
+    'options callback' => 'node_type_get_names',
+    'multiple' => TRUE,
+  ];
+  return $v;
+}
diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
new file mode 100644
index 00000000..c8d24b03
--- /dev/null
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -0,0 +1,146 @@
+<?php
+
+namespace Drupal\campaignion_m2t_send;
+
+use Drupal\campaignion_email_to_target\Channel\Email;
+use Drupal\campaignion_email_to_target\Component;
+use Drupal\little_helpers\Services\Container;
+use Drupal\little_helpers\Webform\Submission;
+
+/**
+ * Cron job for sending out M2T messages as emails.
+ */
+class SendMessagesCron {
+
+  /**
+   * An array of node type machine names to send data for.
+   *
+   * @var str[]
+   */
+  protected $nodeTypes;
+
+  /**
+   * Create a new cron-job instance based on config.
+   */
+  public function __construct($content_types) {
+    $this->nodeTypes = $content_types;
+  }
+
+  /**
+   * Fetch targets from the e2t service.
+   *
+   * @return array
+   *   Targets with email address grouped by party name.
+   */
+  protected function getCurrentTargets(Submission $submission) {
+    /** @var Drupal\campaignion_email_to_target\Api\Client */
+    $client = Container::get()->loadService('campaignion_email_to_target.api.Client');
+    return $client->getTargets('mp', ['postcode' => $submission->valueByKey('postcode')]);
+  }
+
+  /**
+   * Replace the target in a message with a new target.
+   *
+   * @return array
+   *   The message with the target replaced.
+   */
+  protected function replaceTarget($m, $target) {
+    $m['target'] = $target;
+    $m['sent'] = TRUE;
+    $m['message']['toAddress'] = $target['email'];
+    $m['message']['toName'] = trim("{$target['title']} {$target['first_name']} {$target['last_name']}");
+    $m['message']['header'] = preg_replace('/Dear .*,/', $m['message']['header'], "Dear {$target['salutation']},");
+    return $m;
+  }
+
+  /**
+   * Send the target emails for a submission using new data from the e2t-api.
+   */
+  protected function processSubmission(Submission $submission) {
+    $channel = new Email();
+    $targets = NULL;
+    $email = $submission->valueByKey('email');
+    echo "Checking submission (nid={$submission->nid}, sid={$submission->sid}) by $email …\n";
+    $components = $submission->webform->componentsByType('e2t_selector');
+    foreach ($components as $cid => $component) {
+      echo "\tComponent {$cid} (form_key={$component['form_key']}):\n";
+      $component_o = Component::fromComponent($component);
+      $values = array_filter(array_map('unserialize', $submission->valuesByCid($cid)), function ($m) {
+        return !($m['sent'] ?? FALSE);
+      });
+      if (!$values) {
+        echo "\t\tNo unsent messages found.\n";
+        continue;
+      }
+      $cnt = count($values);
+      echo "\t\t$cnt unsent messages found.\n";
+      if (!$targets) {
+        $targets = $this->getCurrentTargets($submission);
+      }
+      $values = array_filter(array_map(function ($m) use ($targets) {
+        if ($new_target = $targets[0] ?? NULL) {
+          return $this->replaceTarget($m, $new_target);
+        }
+        return NULL;
+      }, $values));
+      $cnt = count($values);
+      echo "\t\t$cnt new targets found.\n";
+      if ($values) {
+        $values_to_send = array_map('serialize', $values);
+        $component_o->sendEmails($values_to_send, $submission, $channel);
+        foreach ($values_to_send as $no => $serialized) {
+          db_update('webform_submitted_data')
+            ->fields(['data' => $serialized])
+            ->condition('nid', $submission->nid)
+            ->condition('sid', $submission->sid)
+            ->condition('cid', $cid)
+            ->condition('no', $no)
+            ->execute();
+        }
+      }
+    }
+  }
+
+  /**
+   * Main function of the cron-job.
+   */
+  public function run() {
+    $q = db_select('field_data_field_email_to_target_options', 'o');
+    $q->join('node', 'n', "o.entity_type='node' AND o.entity_id=n.nid");
+    $nids = $q
+      ->fields('o', ['entity_id'])
+      ->condition('o.field_email_to_target_options_dataset_name', ['mp'])
+      ->condition('n.type', $this->nodeTypes)
+      ->execute()
+      ->fetchCol();
+    $nodes = entity_load('node', $nids);
+
+    $submission_sql = <<<SQL
+    SELECT sid
+    FROM {webform_submissions}
+    WHERE nid=:nid AND sid>:last_sid
+    ORDER BY sid
+    LIMIT 100;
+    SQL;
+
+    $count_processed = 0;
+    foreach ($nodes as $node) {
+      $args = [':last_sid' => 0, ':nid' => $node->nid];
+      while ($sids = db_query($submission_sql, $args)->fetchCol()) {
+        foreach ($sids as $sid) {
+          if ($submission = Submission::load($node->nid, $sid, TRUE)) {
+            $this->processSubmission($submission);
+            $count_processed += 1;
+          }
+          if ($count_processed % 100 == 0) {
+            echo "$count_processed submissions processed.\n";
+          }
+          $args[':last_sid'] = $sid;
+        }
+        gc_collect_cycles();
+      }
+    }
+    echo "$count_processed submissions processed. done.\n";
+  }
+
+}

From 0f2a22db7e21fa4ac225667b131ba4c44bc58134 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Tue, 13 Aug 2024 18:04:34 +0200
Subject: [PATCH 02/13] feat(m2t_send): Use a separate table to quickly find
 unsent messages

---
 .../campaignion_m2t_send.info                 |  1 +
 .../campaignion_m2t_send.install              | 79 +++++++++++++++
 campaignion_m2t_send/src/SendMessagesCron.php | 98 +++++++++----------
 campaignion_m2t_send/src/Submission.php       | 44 +++++++++
 4 files changed, 173 insertions(+), 49 deletions(-)
 create mode 100644 campaignion_m2t_send/campaignion_m2t_send.install
 create mode 100644 campaignion_m2t_send/src/Submission.php

diff --git a/campaignion_m2t_send/campaignion_m2t_send.info b/campaignion_m2t_send/campaignion_m2t_send.info
index 4b986cb4..c07e4c15 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.info
+++ b/campaignion_m2t_send/campaignion_m2t_send.info
@@ -8,3 +8,4 @@ dependencies[] = campaignion:campaignion_email_to_target
 dependencies[] = little_helpers:little_helpers
 dependencies[] = ultimate_cron:ultimate_cron
 dependencies[] = variable:variable
+dependencien[] = webform:webform
diff --git a/campaignion_m2t_send/campaignion_m2t_send.install b/campaignion_m2t_send/campaignion_m2t_send.install
new file mode 100644
index 00000000..7a1734c7
--- /dev/null
+++ b/campaignion_m2t_send/campaignion_m2t_send.install
@@ -0,0 +1,79 @@
+<?php
+
+use Drupal\campaignion_m2t_send\Submission;
+
+/**
+ * Implements hook_install().
+ *
+ * Migrate flags stored in webform_submitted_data.
+ */
+function campaignion_m2t_send_install() {
+  $nids = db_select('webform_component', 'c')
+    ->fields('c', ['nid'])
+    ->groupBy('nid')
+    ->condition('type', 'e2t_selector')
+    ->execute()
+    ->fetchCol();
+  $nodes = entity_load('node', $nids);
+
+  foreach (Submission::iterate($nodes) as $submission) {
+    foreach (array_keys($submission->webform->componentsByType('e2t_selector')) as $cid) {
+      foreach ($submission->valuesByCid($cid) as $no => $data) {
+        $m = unserialize($data);
+        if ($m['sent'] ?? NULL) {
+          db_merge('campaignion_m2t_send')
+            ->key(['nid' => $submission->nid, 'sid' => $submission->sid, 'cid' => $cid, 'no' => $no])
+            ->fields(['sent_at' => 0])
+            ->execute();
+        }
+      }
+    }
+  }
+}
+
+/**
+ * Implements hook_schema().
+ */
+function campaignion_m2t_send_schema() {
+  $tables['campaignion_m2t_send'] = [
+    'description' => 'Stores data about email to target messages that have been sent.',
+    'fields' => [
+      'nid' => [
+        'description' => 'The node identifier of a webform.',
+        'type' => 'int',
+        'unsigned' => TRUE,
+        'not null' => TRUE,
+        'default' => 0,
+      ],
+      'sid' => [
+        'description' => 'The unique identifier for this submission.',
+        'type' => 'int',
+        'unsigned' => TRUE,
+        'not null' => TRUE,
+        'default' => 0,
+      ],
+      'cid' => [
+        'description' => 'The identifier for this component within this node, starts at 0 for each node.',
+        'type' => 'int',
+        'size' => 'small',
+        'unsigned' => TRUE,
+        'not null' => TRUE,
+        'default' => 0,
+      ],
+      'no' => [
+        'description' => 'Usually this value is 0, but if a field has multiple values (such as a time or date), it may require multiple rows in the database.',
+        'type' => 'varchar',
+        'length' => 128,
+        'not null' => TRUE,
+        'default' => '0',
+      ],
+      'sent_at' => [
+        'desrciption' => 'The timestamp for when the email was sent',
+        'type' => 'int',
+        'not null' => TRUE,
+      ],
+    ],
+    'primary key' => ['nid', 'sid', 'cid', 'no'],
+  ];
+  return $tables;
+}
diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index c8d24b03..d210b1ef 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -2,10 +2,10 @@
 
 namespace Drupal\campaignion_m2t_send;
 
+use Drupal\campaignion_m2t_send\Submission;
 use Drupal\campaignion_email_to_target\Channel\Email;
 use Drupal\campaignion_email_to_target\Component;
 use Drupal\little_helpers\Services\Container;
-use Drupal\little_helpers\Webform\Submission;
 
 /**
  * Cron job for sending out M2T messages as emails.
@@ -56,48 +56,33 @@ protected function replaceTarget($m, $target) {
   /**
    * Send the target emails for a submission using new data from the e2t-api.
    */
-  protected function processSubmission(Submission $submission) {
+  protected function processSubmission(Submission $submission, array $data) {
     $channel = new Email();
-    $targets = NULL;
+    $targets = $this->getCurrentTargets($submission, $action->getOptions()['dataset_name']);
     $email = $submission->valueByKey('email');
     echo "Checking submission (nid={$submission->nid}, sid={$submission->sid}) by $email …\n";
-    $components = $submission->webform->componentsByType('e2t_selector');
-    foreach ($components as $cid => $component) {
-      echo "\tComponent {$cid} (form_key={$component['form_key']}):\n";
-      $component_o = Component::fromComponent($component);
-      $values = array_filter(array_map('unserialize', $submission->valuesByCid($cid)), function ($m) {
-        return !($m['sent'] ?? FALSE);
-      });
-      if (!$values) {
-        echo "\t\tNo unsent messages found.\n";
-        continue;
-      }
-      $cnt = count($values);
-      echo "\t\t$cnt unsent messages found.\n";
-      if (!$targets) {
-        $targets = $this->getCurrentTargets($submission);
-      }
-      $values = array_filter(array_map(function ($m) use ($targets) {
-        if ($new_target = $targets[0] ?? NULL) {
-          return $this->replaceTarget($m, $new_target);
-        }
-        return NULL;
-      }, $values));
-      $cnt = count($values);
-      echo "\t\t$cnt new targets found.\n";
-      if ($values) {
-        $values_to_send = array_map('serialize', $values);
-        $component_o->sendEmails($values_to_send, $submission, $channel);
-        foreach ($values_to_send as $no => $serialized) {
-          db_update('webform_submitted_data')
-            ->fields(['data' => $serialized])
-            ->condition('nid', $submission->nid)
-            ->condition('sid', $submission->sid)
-            ->condition('cid', $cid)
-            ->condition('no', $no)
-            ->execute();
-        }
+    $data = array_filter(array_map(function ($d) use ($targets) {
+      $m = unserialize($d->data);
+      if ($new_target = $targets[0] ?? NULL) {
+        $d->new_data = serialize($this->replaceTarget($m, $new_target));
+        return $d;
       }
+      return NULL;
+    }, $data));
+    foreach ($data as $d) {
+      $component_o = Component::fromComponent($submission->webform->component($d->cid));
+      $component_o->sendEmails([$d->new_data], $submission, $channel);
+      db_update('webform_submitted_data')
+        ->fields(['data' => $d->new_data])
+        ->condition('nid', $submission->nid)
+        ->condition('sid', $submission->sid)
+        ->condition('cid', $d->cid)
+        ->condition('no', $d->no)
+        ->execute();
+      db_merge('webform_submitted_data')
+        ->key(['nid' => $submission->nid, 'sid' => $submission->sid, 'cid' => $d->cid, 'no' => $d->no])
+        ->fields(['sent_at' => time()])
+        ->execute();
     }
   }
 
@@ -115,27 +100,42 @@ public function run() {
       ->fetchCol();
     $nodes = entity_load('node', $nids);
 
+    $data_sql = <<<SQL
+    SELECT nid, sid, cid, no, data
+    FROM {webform_submissions} s
+      INNER JOIN {webform_component} c ON c.nid=s.nid AND c.type='e2t_selector'
+      INNER JOIN {webform_submitted_data} d USING(nid, sid, cid)
+      LEFT OUTER JOIN {campaignion_m2t_send} m USING(nid, sid, cid, no)
+    WHERE m.sid IS NULL AND nid=:nid AND sid IN(:sids);
+    SQL;
+
     $submission_sql = <<<SQL
-    SELECT sid
-    FROM {webform_submissions}
-    WHERE nid=:nid AND sid>:last_sid
+    SELECT DISTINCT sid
+    FROM {webform_submissions} s
+      INNER JOIN {webform_component} c ON c.nid=s.nid AND c.type='e2t_selector'
+      INNER JOIN {webform_submitted_data} d USING(nid, sid, cid)
+      LEFT OUTER JOIN {campaignion_m2t_send} m USING(nid, sid, cid, no)
+    WHERE m.sid IS NULL AND s.nid=:nid AND s.sid>:last_sid
     ORDER BY sid
-    LIMIT 100;
+    LIMIT 100
     SQL;
 
     $count_processed = 0;
     foreach ($nodes as $node) {
       $args = [':last_sid' => 0, ':nid' => $node->nid];
       while ($sids = db_query($submission_sql, $args)->fetchCol()) {
-        foreach ($sids as $sid) {
-          if ($submission = Submission::load($node->nid, $sid, TRUE)) {
-            $this->processSubmission($submission);
-            $count_processed += 1;
-          }
+        $data_per_sid = [];
+        foreach (db_query($data_sql, [':nid' => $node->nid, ':sids' => $sids]) as $data) {
+          $data_per_sid[$data->sid][] = $data;
+        }
+        foreach (webform_get_submissions(['ws.sid' => $sids]) as $s) {
+          $submission = new Submission($nodes[$s->nid], $s);
+          $this->processSubmission($submission, $data_per_sid[$submission->sid] ?? []);
+          $count_processed += 1;
           if ($count_processed % 100 == 0) {
             echo "$count_processed submissions processed.\n";
           }
-          $args[':last_sid'] = $sid;
+          $args[':last_sid'] = $s->sid;
         }
         gc_collect_cycles();
       }
diff --git a/campaignion_m2t_send/src/Submission.php b/campaignion_m2t_send/src/Submission.php
new file mode 100644
index 00000000..d54e1e94
--- /dev/null
+++ b/campaignion_m2t_send/src/Submission.php
@@ -0,0 +1,44 @@
+<?php
+
+namespace Drupal\campaignion_m2t_send;
+
+use Drupal\little_helpers\Webform\Submission as _Submission;
+
+/**
+ * Extend submission objects with a scalable iterator.
+ */
+class Submission extends _Submission {
+
+  /**
+   * Efficently iterate over submissions.
+   *
+   * @param object[] $nodes
+   *   The nodes whichs’ submissions should be iterated.
+   * @param string $submission_sql
+   *   A SQL query string that selects sids to iterate over. It must use the
+   *   :nid and :last_sid parameter and sort by sid. If no query is passed,
+   *   then the function iterates over all the submissions of all passed nodes.
+   */
+  public static function iterate(array $nodes, $submission_sql = NULL) {
+    // By default loop over all submissions of each of the passed nodes.
+    $submission_sql = $submission_sql ?? <<<SQL
+    SELECT sid
+    FROM {webform_submissions}
+    WHERE nid=:nid AND sid>:last_sid
+    ORDER BY sid
+    LIMIT 100;
+    SQL;
+
+    foreach ($nodes as $node) {
+      $args = [':last_sid' => 0, ':nid' => $node->nid];
+      while ($sids = db_query($submission_sql, $args)->fetchCol()) {
+        foreach (webform_get_submissions(['ws.sid' => $sids]) as $s) {
+          yield new _Submission($nodes[$s->nid], $s);
+          $args[':last_sid'] = $s->sid;
+        }
+        gc_collect_cycles();
+      }
+    }
+  }
+
+}

From 58b2db6c6532afd3e64f5fb73aec2271ce298a75 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Fri, 16 Aug 2024 10:46:36 +0200
Subject: [PATCH 03/13] feat(m2t_send): Directly configure enabled nodes

---
 .../campaignion_m2t_send.module               |  4 ++--
 .../campaignion_m2t_send.variable.inc         | 13 +++++--------
 campaignion_m2t_send/src/SendMessagesCron.php | 19 ++++++-------------
 3 files changed, 13 insertions(+), 23 deletions(-)

diff --git a/campaignion_m2t_send/campaignion_m2t_send.module b/campaignion_m2t_send/campaignion_m2t_send.module
index c8472fa5..b81a0e18 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.module
+++ b/campaignion_m2t_send/campaignion_m2t_send.module
@@ -15,7 +15,7 @@ function campaignion_m2t_send_little_helpers_services() {
   $info['campaignion_m2t_send.SendMessagesCron'] = [
     'class' => SendMessagesCron::class,
     'arguments' => [
-      '!campaignion_m2t_send_content_types',
+      '!campaignion_m2t_send_enabled_nodes',
     ],
   ];
   return $info;
@@ -26,7 +26,7 @@ function campaignion_m2t_send_little_helpers_services() {
  */
 function campaignion_m2t_send_cronapi($op, $job = NULL) {
   $items['campaignion_m2t_send'] = [
-    'description' => 'Send m2t messages',
+    'description' => 'Send M2T messages of selected nodes',
     'rule' => '*+@ * * * *',
     'weight' => 100,
     'callback' => '_campaignion_m2t_send_run',
diff --git a/campaignion_m2t_send/campaignion_m2t_send.variable.inc b/campaignion_m2t_send/campaignion_m2t_send.variable.inc
index 3672903c..43a211a9 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.variable.inc
+++ b/campaignion_m2t_send/campaignion_m2t_send.variable.inc
@@ -9,14 +9,11 @@
  * Implements hook_variable_info().
  */
 function campaignion_m2t_send_variable_info($options) {
-  $v['campaignion_m2t_send_content_types'] = [
-    'title' => t('Restrict to content-types'),
-    'description' => t('Only process actions of these types.'),
-    'type' => 'options',
-    'default' => ['match_to_target'],
-    'localize' => FALSE,
-    'options callback' => 'node_type_get_names',
-    'multiple' => TRUE,
+  $v['campaignion_m2t_send_enabled_nodes'] = [
+    'title' => t('Enabled nodes'),
+    'description' => t('Only send emails for these nodes.'),
+    'type' => 'unknown',
+    'default' => [],
   ];
   return $v;
 }
diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index d210b1ef..d50a1b31 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -13,17 +13,17 @@
 class SendMessagesCron {
 
   /**
-   * An array of node type machine names to send data for.
+   * An array of nids for which sending is enabled.
    *
-   * @var str[]
+   * @var array
    */
-  protected $nodeTypes;
+  protected $enabledNodes;
 
   /**
    * Create a new cron-job instance based on config.
    */
-  public function __construct($content_types) {
-    $this->nodeTypes = $content_types;
+  public function __construct(array $enabled_nodes) {
+    $this->enabledNodes = $enabled_nodes;
   }
 
   /**
@@ -90,14 +90,7 @@ protected function processSubmission(Submission $submission, array $data) {
    * Main function of the cron-job.
    */
   public function run() {
-    $q = db_select('field_data_field_email_to_target_options', 'o');
-    $q->join('node', 'n', "o.entity_type='node' AND o.entity_id=n.nid");
-    $nids = $q
-      ->fields('o', ['entity_id'])
-      ->condition('o.field_email_to_target_options_dataset_name', ['mp'])
-      ->condition('n.type', $this->nodeTypes)
-      ->execute()
-      ->fetchCol();
+    $nids = array_keys(array_filter($this->enabledNodes));
     $nodes = entity_load('node', $nids);
 
     $data_sql = <<<SQL

From 6368548eceeb0e23fd18e8735e68e7e2b6cc0316 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Fri, 16 Aug 2024 11:39:54 +0200
Subject: [PATCH 04/13] feat(m2t_send): Make implementation independent of the
 dataset

---
 campaignion_m2t_send/src/SendMessagesCron.php | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index d50a1b31..a6d73ab2 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -2,7 +2,9 @@
 
 namespace Drupal\campaignion_m2t_send;
 
+use Drupal\campaignion_action\Loader as ActionLoader;
 use Drupal\campaignion_m2t_send\Submission;
+use Drupal\campaignion_email_to_target\Action;
 use Drupal\campaignion_email_to_target\Channel\Email;
 use Drupal\campaignion_email_to_target\Component;
 use Drupal\little_helpers\Services\Container;
@@ -32,10 +34,10 @@ public function __construct(array $enabled_nodes) {
    * @return array
    *   Targets with email address grouped by party name.
    */
-  protected function getCurrentTargets(Submission $submission) {
+  protected function getCurrentTargets(Submission $submission, string $dataset) {
     /** @var Drupal\campaignion_email_to_target\Api\Client */
     $client = Container::get()->loadService('campaignion_email_to_target.api.Client');
-    return $client->getTargets('mp', ['postcode' => $submission->valueByKey('postcode')]);
+    return $client->getTargets($dataset, ['postcode' => $submission->valueByKey('postcode')]);
   }
 
   /**
@@ -56,7 +58,7 @@ protected function replaceTarget($m, $target) {
   /**
    * Send the target emails for a submission using new data from the e2t-api.
    */
-  protected function processSubmission(Submission $submission, array $data) {
+  protected function processSubmission(Submission $submission, Action $action, array $data) {
     $channel = new Email();
     $targets = $this->getCurrentTargets($submission, $action->getOptions()['dataset_name']);
     $email = $submission->valueByKey('email');
@@ -90,8 +92,12 @@ protected function processSubmission(Submission $submission, array $data) {
    * Main function of the cron-job.
    */
   public function run() {
+    module_load_include('inc', 'webform', 'includes/webform.submissions');
     $nids = array_keys(array_filter($this->enabledNodes));
     $nodes = entity_load('node', $nids);
+    $actions = array_map(function ($node) {
+      return ActionLoader::instance()->actionFromNode($node);
+    }, $nodes);
 
     $data_sql = <<<SQL
     SELECT nid, sid, cid, no, data
@@ -123,7 +129,7 @@ public function run() {
         }
         foreach (webform_get_submissions(['ws.sid' => $sids]) as $s) {
           $submission = new Submission($nodes[$s->nid], $s);
-          $this->processSubmission($submission, $data_per_sid[$submission->sid] ?? []);
+          $this->processSubmission($submission, $actions[$submission->nid], $data_per_sid[$submission->sid] ?? []);
           $count_processed += 1;
           if ($count_processed % 100 == 0) {
             echo "$count_processed submissions processed.\n";

From c0ff0dc99d7690e0965c390dceac88bf03fb4d32 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Fri, 16 Aug 2024 15:52:29 +0200
Subject: [PATCH 05/13] feat(m2t_send): Simple trickle sending ~1 message per
 target and run

---
 campaignion_m2t_send/src/SendMessagesCron.php | 51 ++++++++++++++++---
 1 file changed, 44 insertions(+), 7 deletions(-)

diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index a6d73ab2..750c2e85 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -21,6 +21,21 @@ class SendMessagesCron {
    */
   protected $enabledNodes;
 
+  /**
+   * Number of messages sent for each target.
+   *
+   * @var int[]
+   */
+  protected $sendHistory = [];
+
+  /**
+   * Message statistics.
+   */
+  protected $messageStats = [
+    'sent' => 0,
+    'withheld' => 0,
+  ];
+
   /**
    * Create a new cron-job instance based on config.
    */
@@ -55,17 +70,37 @@ protected function replaceTarget($m, $target) {
     return $m;
   }
 
+  /**
+   * Check whether to send an additional email to the target or not.
+   */
+  protected function rateLimit($target) {
+    $email = $target['email'];
+    $count = $this->sendHistory[$email] ?? 0;
+    if ($count === FALSE) {
+      $this->messageStats['withheld'] += 1;
+      return FALSE;
+    }
+    if (rand(0, 1) === 1) {
+      $this->sendHistory[$email] = $count + 1;
+      $this->messageStats['sent'] += 1;
+      return TRUE;
+    }
+    else {
+      $this->sendHistory[$email] = FALSE;
+      $this->messageStats['withheld'] += 1;
+      return FALSE;
+    }
+  }
+
   /**
    * Send the target emails for a submission using new data from the e2t-api.
    */
   protected function processSubmission(Submission $submission, Action $action, array $data) {
     $channel = new Email();
     $targets = $this->getCurrentTargets($submission, $action->getOptions()['dataset_name']);
-    $email = $submission->valueByKey('email');
-    echo "Checking submission (nid={$submission->nid}, sid={$submission->sid}) by $email …\n";
     $data = array_filter(array_map(function ($d) use ($targets) {
       $m = unserialize($d->data);
-      if ($new_target = $targets[0] ?? NULL) {
+      if (($new_target = $targets[0] ?? NULL) && $this->rateLimit($new_target)) {
         $d->new_data = serialize($this->replaceTarget($m, $new_target));
         return $d;
       }
@@ -131,15 +166,17 @@ public function run() {
           $submission = new Submission($nodes[$s->nid], $s);
           $this->processSubmission($submission, $actions[$submission->nid], $data_per_sid[$submission->sid] ?? []);
           $count_processed += 1;
-          if ($count_processed % 100 == 0) {
-            echo "$count_processed submissions processed.\n";
-          }
           $args[':last_sid'] = $s->sid;
         }
         gc_collect_cycles();
       }
     }
-    echo "$count_processed submissions processed. done.\n";
+    $vars = [
+      '@submissions' => $count_processed,
+      '@sent' => $this->messageStats['sent'],
+      '@withheld' => $this->messageStats['withheld']
+    ];
+    watchdog('campaignion_m2t_send', '@submissions submissions processed, @sent emails sent (@withheld emails not sent).', $vars, WATCHDOG_INFO);
   }
 
 }

From b2581e61d4b47c88c457f196157dcedf3b792da4 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Sun, 18 Aug 2024 13:11:42 +0200
Subject: [PATCH 06/13] feat(m2t_send): Use table as queue for sending (simpler
 queries)

---
 .../campaignion_m2t_send.install              | 28 +++++++++--
 .../campaignion_m2t_send.module               | 50 +++++++++++++++++++
 campaignion_m2t_send/src/SendMessagesCron.php | 19 +++----
 3 files changed, 81 insertions(+), 16 deletions(-)

diff --git a/campaignion_m2t_send/campaignion_m2t_send.install b/campaignion_m2t_send/campaignion_m2t_send.install
index 7a1734c7..50ceee21 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.install
+++ b/campaignion_m2t_send/campaignion_m2t_send.install
@@ -8,6 +8,15 @@ use Drupal\campaignion_m2t_send\Submission;
  * Migrate flags stored in webform_submitted_data.
  */
 function campaignion_m2t_send_install() {
+  $sql = <<<SQL
+  INSERT INTO {campaignion_m2t_send}
+  SELECT nid, sid, cid, no, NULL
+  FROM {webform_submissions} s
+    INNER JOIN {webform_component} c USING(nid)
+    INNER JOIN {webform_submitted_data} d USING(nid, sid, cid)
+  WHERE c.type='e2t_selector'
+  SQL;
+  db_query($sql);
   $nids = db_select('webform_component', 'c')
     ->fields('c', ['nid'])
     ->groupBy('nid')
@@ -21,8 +30,11 @@ function campaignion_m2t_send_install() {
       foreach ($submission->valuesByCid($cid) as $no => $data) {
         $m = unserialize($data);
         if ($m['sent'] ?? NULL) {
-          db_merge('campaignion_m2t_send')
-            ->key(['nid' => $submission->nid, 'sid' => $submission->sid, 'cid' => $cid, 'no' => $no])
+          db_update('campaignion_m2t_send')
+            ->condition('nid', $submission->nid)
+            ->condition('sid', $submission->sid)
+            ->condition('cid', $cid)
+            ->condition('no', $no)
             ->fields(['sent_at' => 0])
             ->execute();
         }
@@ -36,7 +48,7 @@ function campaignion_m2t_send_install() {
  */
 function campaignion_m2t_send_schema() {
   $tables['campaignion_m2t_send'] = [
-    'description' => 'Stores data about email to target messages that have been sent.',
+    'description' => 'Stores the send status of email to target emails.',
     'fields' => [
       'nid' => [
         'description' => 'The node identifier of a webform.',
@@ -68,12 +80,18 @@ function campaignion_m2t_send_schema() {
         'default' => '0',
       ],
       'sent_at' => [
-        'desrciption' => 'The timestamp for when the email was sent',
+        'desrciption' => 'The timestamp for when the email was sent (if)',
         'type' => 'int',
-        'not null' => TRUE,
+        'not null' => FALSE,
       ],
     ],
     'primary key' => ['nid', 'sid', 'cid', 'no'],
+    'indexes' => [
+      'node' => ['nid'],
+      'submission' => ['sid'],
+      'component' => ['nid', 'sid', 'cid'],
+      'unsent' => ['nid', 'sent_at']
+    ],
   ];
   return $tables;
 }
diff --git a/campaignion_m2t_send/campaignion_m2t_send.module b/campaignion_m2t_send/campaignion_m2t_send.module
index b81a0e18..2dfd822d 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.module
+++ b/campaignion_m2t_send/campaignion_m2t_send.module
@@ -7,6 +7,7 @@
 
 use Drupal\campaignion_m2t_send\SendMessagesCron;
 use Drupal\little_helpers\Services\Container;
+use Drupal\little_helpers\Webform\Submission;
 
 /**
  * Implements hook_little_helpers_services().
@@ -41,3 +42,52 @@ function campaignion_m2t_send_cronapi($op, $job = NULL) {
 function _campaignion_m2t_send_run($service) {
   Container::get()->loadService($service)->run();
 }
+
+/**
+ * Implements hook_webform_submission_insert().
+ */
+function campaignion_m2t_send_webform_submission_insert($node, $submission) {
+  $s = new Submission($node, $submission);
+  foreach ($s->webform->componentsByType('e2t_selector') as $component) {
+    $values = $s->valuesByCid($component['cid']);
+    db_delete('campaignion_m2t_send')
+      ->condition('nid', $node->nid)
+      ->condition('sid', $submission->sid)
+      ->condition('cid', $component['cid'])
+      ->condition('no', array_keys($values), 'NOT IN')
+      ->execute();
+    foreach ($values as $no => $value) {
+      db_merge('campaignion_m2t_send')->key([
+        'nid' => $node->nid,
+        'sid' => $submission->sid,
+        'cid' => $component['cid'],
+        'no' => $no,
+      ])->execute();
+    }
+  }
+}
+
+/**
+ * Implements hook_webform_submission_update().
+ */
+function campaignion_m2t_send_webform_submission_update($node, $submission) {
+  campaignion_m2t_send_webform_submission_insert($node, $submission);
+}
+
+/**
+ * Implements hook_webform_submission_delete().
+ */
+function campaignion_m2t_send_webform_submission_delete($node, $submission) {
+  db_delete('campaignion_m2t_send')
+    ->condition('sid', $submission->sid)
+    ->execute();
+}
+
+/**
+ * Implements hook_node_delete().
+ */
+function campaignion_m2t_send_node_delete($node) {
+  db_delete('campaignion_m2t_send')
+    ->condition('nid', $node->nid)
+    ->execute();
+}
diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index 750c2e85..f32a9410 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -136,21 +136,18 @@ public function run() {
 
     $data_sql = <<<SQL
     SELECT nid, sid, cid, no, data
-    FROM {webform_submissions} s
-      INNER JOIN {webform_component} c ON c.nid=s.nid AND c.type='e2t_selector'
-      INNER JOIN {webform_submitted_data} d USING(nid, sid, cid)
-      LEFT OUTER JOIN {campaignion_m2t_send} m USING(nid, sid, cid, no)
-    WHERE m.sid IS NULL AND nid=:nid AND sid IN(:sids);
+    FROM {webform_submitted_data} d
+      INNER JOIN {campaignion_m2t_send} m USING(nid, sid, cid, no)
+    WHERE nid=:nid AND sid IN(:sids) AND sent_at IS NULL
     SQL;
 
     $submission_sql = <<<SQL
-    SELECT DISTINCT sid
+    SELECT s.sid
     FROM {webform_submissions} s
-      INNER JOIN {webform_component} c ON c.nid=s.nid AND c.type='e2t_selector'
-      INNER JOIN {webform_submitted_data} d USING(nid, sid, cid)
-      LEFT OUTER JOIN {campaignion_m2t_send} m USING(nid, sid, cid, no)
-    WHERE m.sid IS NULL AND s.nid=:nid AND s.sid>:last_sid
-    ORDER BY sid
+    WHERE s.sid IN (
+      SELECT sid FROM {campaignion_m2t_send} WHERE nid=:nid AND sent_at IS NULL
+    ) AND s.sid>:last_sid
+    ORDER BY s.sid
     LIMIT 100
     SQL;
 

From 3b1b6d0e2a0c8c278eff001f07901fe39691b596 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Mon, 19 Aug 2024 08:59:45 +0200
Subject: [PATCH 07/13] feat(m2t_send): Keep track of targets emails were sent
 to

---
 campaignion_m2t_send/campaignion_m2t_send.install | 13 ++++++++++---
 campaignion_m2t_send/src/SendMessagesCron.php     |  5 +++--
 2 files changed, 13 insertions(+), 5 deletions(-)

diff --git a/campaignion_m2t_send/campaignion_m2t_send.install b/campaignion_m2t_send/campaignion_m2t_send.install
index 50ceee21..7c81971e 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.install
+++ b/campaignion_m2t_send/campaignion_m2t_send.install
@@ -10,7 +10,7 @@ use Drupal\campaignion_m2t_send\Submission;
 function campaignion_m2t_send_install() {
   $sql = <<<SQL
   INSERT INTO {campaignion_m2t_send}
-  SELECT nid, sid, cid, no, NULL
+  SELECT nid, sid, cid, no, NULL, NULL
   FROM {webform_submissions} s
     INNER JOIN {webform_component} c USING(nid)
     INNER JOIN {webform_submitted_data} d USING(nid, sid, cid)
@@ -35,7 +35,7 @@ function campaignion_m2t_send_install() {
             ->condition('sid', $submission->sid)
             ->condition('cid', $cid)
             ->condition('no', $no)
-            ->fields(['sent_at' => 0])
+            ->fields(['sent_at' => 0, 'target_email' => $m['target']['email']])
             ->execute();
         }
       }
@@ -79,6 +79,12 @@ function campaignion_m2t_send_schema() {
         'not null' => TRUE,
         'default' => '0',
       ],
+      'target_email' => [
+        'description' => 'Email address of the target the email was sent to.',
+        'type' => 'varchar',
+        'length' => 256,
+        'not null' => FALSE,
+      ],
       'sent_at' => [
         'desrciption' => 'The timestamp for when the email was sent (if)',
         'type' => 'int',
@@ -90,7 +96,8 @@ function campaignion_m2t_send_schema() {
       'node' => ['nid'],
       'submission' => ['sid'],
       'component' => ['nid', 'sid', 'cid'],
-      'unsent' => ['nid', 'sent_at']
+      'unsent' => ['nid', 'sent_at'],
+      'target' => ['target_email'],
     ],
   ];
   return $tables;
diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index f32a9410..726f6625 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -102,6 +102,7 @@ protected function processSubmission(Submission $submission, Action $action, arr
       $m = unserialize($d->data);
       if (($new_target = $targets[0] ?? NULL) && $this->rateLimit($new_target)) {
         $d->new_data = serialize($this->replaceTarget($m, $new_target));
+        $d->new_target = $new_target;
         return $d;
       }
       return NULL;
@@ -116,9 +117,9 @@ protected function processSubmission(Submission $submission, Action $action, arr
         ->condition('cid', $d->cid)
         ->condition('no', $d->no)
         ->execute();
-      db_merge('webform_submitted_data')
+      db_merge('campaignion_m2t_send')
         ->key(['nid' => $submission->nid, 'sid' => $submission->sid, 'cid' => $d->cid, 'no' => $d->no])
-        ->fields(['sent_at' => time()])
+        ->fields(['sent_at' => time(), 'target_email' => $d->new_target['email']])
         ->execute();
     }
   }

From caf2a9f5b8df7fff80fa60cea6938056c797729e Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Mon, 19 Aug 2024 10:22:34 +0200
Subject: [PATCH 08/13] feat(m2t_send): Load unsent messages when loading
 submissions

---
 .../campaignion_m2t_send.module               | 15 ++++++++++
 campaignion_m2t_send/src/SendMessagesCron.php | 29 ++++---------------
 campaignion_m2t_send/src/Submission.php       |  2 +-
 3 files changed, 21 insertions(+), 25 deletions(-)

diff --git a/campaignion_m2t_send/campaignion_m2t_send.module b/campaignion_m2t_send/campaignion_m2t_send.module
index 2dfd822d..79303df5 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.module
+++ b/campaignion_m2t_send/campaignion_m2t_send.module
@@ -67,6 +67,21 @@ function campaignion_m2t_send_webform_submission_insert($node, $submission) {
   }
 }
 
+/**
+ * Implements hook_webform_submission_load().
+ */
+function campaignion_m2t_send_webform_submission_load(&$submissions) {
+  $data_sql = <<<SQL
+  SELECT nid, sid, cid, no, data
+  FROM {webform_submitted_data} d
+    INNER JOIN {campaignion_m2t_send} m USING(nid, sid, cid, no)
+  WHERE sid IN(:sids) AND sent_at IS NULL
+  SQL;
+  foreach (db_query($data_sql, [':sids' => array_keys($submissions)]) as $d) {
+    $submissions[$d->sid]->m2t_unsent_messages[] = $d;
+  }
+}
+
 /**
  * Implements hook_webform_submission_update().
  */
diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index 726f6625..d63efbed 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -95,7 +95,7 @@ protected function rateLimit($target) {
   /**
    * Send the target emails for a submission using new data from the e2t-api.
    */
-  protected function processSubmission(Submission $submission, Action $action, array $data) {
+  protected function processSubmission(Submission $submission, Action $action) {
     $channel = new Email();
     $targets = $this->getCurrentTargets($submission, $action->getOptions()['dataset_name']);
     $data = array_filter(array_map(function ($d) use ($targets) {
@@ -106,7 +106,7 @@ protected function processSubmission(Submission $submission, Action $action, arr
         return $d;
       }
       return NULL;
-    }, $data));
+    }, $submission->m2t_unsent_messages));
     foreach ($data as $d) {
       $component_o = Component::fromComponent($submission->webform->component($d->cid));
       $component_o->sendEmails([$d->new_data], $submission, $channel);
@@ -135,13 +135,6 @@ public function run() {
       return ActionLoader::instance()->actionFromNode($node);
     }, $nodes);
 
-    $data_sql = <<<SQL
-    SELECT nid, sid, cid, no, data
-    FROM {webform_submitted_data} d
-      INNER JOIN {campaignion_m2t_send} m USING(nid, sid, cid, no)
-    WHERE nid=:nid AND sid IN(:sids) AND sent_at IS NULL
-    SQL;
-
     $submission_sql = <<<SQL
     SELECT s.sid
     FROM {webform_submissions} s
@@ -153,21 +146,9 @@ public function run() {
     SQL;
 
     $count_processed = 0;
-    foreach ($nodes as $node) {
-      $args = [':last_sid' => 0, ':nid' => $node->nid];
-      while ($sids = db_query($submission_sql, $args)->fetchCol()) {
-        $data_per_sid = [];
-        foreach (db_query($data_sql, [':nid' => $node->nid, ':sids' => $sids]) as $data) {
-          $data_per_sid[$data->sid][] = $data;
-        }
-        foreach (webform_get_submissions(['ws.sid' => $sids]) as $s) {
-          $submission = new Submission($nodes[$s->nid], $s);
-          $this->processSubmission($submission, $actions[$submission->nid], $data_per_sid[$submission->sid] ?? []);
-          $count_processed += 1;
-          $args[':last_sid'] = $s->sid;
-        }
-        gc_collect_cycles();
-      }
+    foreach (Submission::iterate($nodes, $submission_sql) as $submission) {
+      $this->processSubmission($submission, $actions[$submission->nid]);
+      $count_processed += 1;
     }
     $vars = [
       '@submissions' => $count_processed,
diff --git a/campaignion_m2t_send/src/Submission.php b/campaignion_m2t_send/src/Submission.php
index d54e1e94..232fbb63 100644
--- a/campaignion_m2t_send/src/Submission.php
+++ b/campaignion_m2t_send/src/Submission.php
@@ -33,7 +33,7 @@ public static function iterate(array $nodes, $submission_sql = NULL) {
       $args = [':last_sid' => 0, ':nid' => $node->nid];
       while ($sids = db_query($submission_sql, $args)->fetchCol()) {
         foreach (webform_get_submissions(['ws.sid' => $sids]) as $s) {
-          yield new _Submission($nodes[$s->nid], $s);
+          yield new static($nodes[$s->nid], $s);
           $args[':last_sid'] = $s->sid;
         }
         gc_collect_cycles();

From f12dd91550e3ffbcf1103a25d7e157966f75ca78 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Mon, 19 Aug 2024 10:41:47 +0200
Subject: [PATCH 09/13] fix(m2t_send): Filter out targets without email address

---
 campaignion_m2t_send/src/SendMessagesCron.php | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index d63efbed..560bbf46 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -52,7 +52,9 @@ public function __construct(array $enabled_nodes) {
   protected function getCurrentTargets(Submission $submission, string $dataset) {
     /** @var Drupal\campaignion_email_to_target\Api\Client */
     $client = Container::get()->loadService('campaignion_email_to_target.api.Client');
-    return $client->getTargets($dataset, ['postcode' => $submission->valueByKey('postcode')]);
+    return array_filter($client->getTargets($dataset, ['postcode' => $submission->valueByKey('postcode')]), function ($target) {
+      return (bool) ($target['email'] ?? NULL);
+    });
   }
 
   /**

From c1db949f2c2fbd621994e03111fa8f172b9df595 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Mon, 19 Aug 2024 10:57:38 +0200
Subject: [PATCH 10/13] feat(m2t_send): Add a configurable time-limit for the
 cron-runs

---
 campaignion_m2t_send/campaignion_m2t_send.module   |  1 +
 .../campaignion_m2t_send.variable.inc              |  7 +++++++
 campaignion_m2t_send/src/SendMessagesCron.php      | 14 +++++++++++++-
 3 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/campaignion_m2t_send/campaignion_m2t_send.module b/campaignion_m2t_send/campaignion_m2t_send.module
index 79303df5..f55ab937 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.module
+++ b/campaignion_m2t_send/campaignion_m2t_send.module
@@ -17,6 +17,7 @@ function campaignion_m2t_send_little_helpers_services() {
     'class' => SendMessagesCron::class,
     'arguments' => [
       '!campaignion_m2t_send_enabled_nodes',
+      '!campaignion_m2t_send_cron_time_limit',
     ],
   ];
   return $info;
diff --git a/campaignion_m2t_send/campaignion_m2t_send.variable.inc b/campaignion_m2t_send/campaignion_m2t_send.variable.inc
index 43a211a9..a051e22d 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.variable.inc
+++ b/campaignion_m2t_send/campaignion_m2t_send.variable.inc
@@ -9,6 +9,13 @@
  * Implements hook_variable_info().
  */
 function campaignion_m2t_send_variable_info($options) {
+  $v['campaignion_m2t_send_cron_time_limit'] = [
+    'title' => t('Time limit for M2T send cron-jobs'),
+    'description' => t('When a m2t send cron-job has been running for more than this amount of seconds no new batch will be started during this cron-run.'),
+    'type' => 'number',
+    'default' => 20,
+    'localize' => FALSE,
+  ];
   $v['campaignion_m2t_send_enabled_nodes'] = [
     'title' => t('Enabled nodes'),
     'description' => t('Only send emails for these nodes.'),
diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index 560bbf46..f4cf11c6 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -36,11 +36,19 @@ class SendMessagesCron {
     'withheld' => 0,
   ];
 
+  /**
+   * Cron time limit in seconds.
+   *
+   * @var int
+   */
+  protected $timeLimit;
+
   /**
    * Create a new cron-job instance based on config.
    */
-  public function __construct(array $enabled_nodes) {
+  public function __construct(array $enabled_nodes, int $time_limit) {
     $this->enabledNodes = $enabled_nodes;
+    $this->timeLimit = $time_limit;
   }
 
   /**
@@ -131,6 +139,7 @@ protected function processSubmission(Submission $submission, Action $action) {
    */
   public function run() {
     module_load_include('inc', 'webform', 'includes/webform.submissions');
+    $time_limit =  REQUEST_TIME + $this->timeLimit;
     $nids = array_keys(array_filter($this->enabledNodes));
     $nodes = entity_load('node', $nids);
     $actions = array_map(function ($node) {
@@ -151,6 +160,9 @@ public function run() {
     foreach (Submission::iterate($nodes, $submission_sql) as $submission) {
       $this->processSubmission($submission, $actions[$submission->nid]);
       $count_processed += 1;
+      if (time() > $time_limit) {
+        break;
+      }
     }
     $vars = [
       '@submissions' => $count_processed,

From 9291ded80ed3da4630a84b639ba48a0db2defaff Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Mon, 19 Aug 2024 11:18:06 +0200
Subject: [PATCH 11/13] feat(m2t_send): Only send during the day by default

---
 campaignion_m2t_send/campaignion_m2t_send.module | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/campaignion_m2t_send/campaignion_m2t_send.module b/campaignion_m2t_send/campaignion_m2t_send.module
index f55ab937..c4a0cf6a 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.module
+++ b/campaignion_m2t_send/campaignion_m2t_send.module
@@ -29,7 +29,7 @@ function campaignion_m2t_send_little_helpers_services() {
 function campaignion_m2t_send_cronapi($op, $job = NULL) {
   $items['campaignion_m2t_send'] = [
     'description' => 'Send M2T messages of selected nodes',
-    'rule' => '*+@ * * * *',
+    'rule' => '*+@ 8-23 * * *',
     'weight' => 100,
     'callback' => '_campaignion_m2t_send_run',
     'arguments' => ['campaignion_m2t_send.SendMessagesCron'],

From d9eed33251b4d74d06f6a1e1daa515ee852ff23a Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Tue, 20 Aug 2024 15:17:13 +0200
Subject: [PATCH 12/13] fix(m2t_send): Simplify rate limite code

---
 campaignion_m2t_send/src/SendMessagesCron.php | 35 +++++++++----------
 1 file changed, 17 insertions(+), 18 deletions(-)

diff --git a/campaignion_m2t_send/src/SendMessagesCron.php b/campaignion_m2t_send/src/SendMessagesCron.php
index f4cf11c6..88641239 100644
--- a/campaignion_m2t_send/src/SendMessagesCron.php
+++ b/campaignion_m2t_send/src/SendMessagesCron.php
@@ -22,11 +22,11 @@ class SendMessagesCron {
   protected $enabledNodes;
 
   /**
-   * Number of messages sent for each target.
+   * Store a flag for each target, that shows if more messages should be sent or not.
    *
-   * @var int[]
+   * @var bool[]
    */
-  protected $sendHistory = [];
+  protected $stopSending = [];
 
   /**
    * Message statistics.
@@ -81,25 +81,24 @@ protected function replaceTarget($m, $target) {
   }
 
   /**
-   * Check whether to send an additional email to the target or not.
+   * Decide whether to send an additional email to the target or not.
+   *
+   * This should be somewhat random as to look natural. It also keeps the
+   * order of messages.
    */
   protected function rateLimit($target) {
     $email = $target['email'];
-    $count = $this->sendHistory[$email] ?? 0;
-    if ($count === FALSE) {
-      $this->messageStats['withheld'] += 1;
-      return FALSE;
-    }
-    if (rand(0, 1) === 1) {
-      $this->sendHistory[$email] = $count + 1;
-      $this->messageStats['sent'] += 1;
-      return TRUE;
-    }
-    else {
-      $this->sendHistory[$email] = FALSE;
-      $this->messageStats['withheld'] += 1;
-      return FALSE;
+    if (!($this->stopSending[$email] ?? FALSE)) {
+      if (rand(0, 1) === 1) { // 50:50 chance.
+        $this->messageStats['sent'] += 1;
+        return TRUE;
+      }
+      else {
+        $this->stopSending[$email] = TRUE;
+      }
     }
+    $this->messageStats['withheld'] += 1;
+    return FALSE;
   }
 
   /**

From 68a3f8db56f523f940780e8f5b71a71fac7a3003 Mon Sep 17 00:00:00 2001
From: Roman Zimmermann <roman@more-onion.com>
Date: Tue, 20 Aug 2024 15:32:59 +0200
Subject: [PATCH 13/13] feat(m2t_send): Limit functionality to M2T node types

Those node types can be identified by using the the EmailNoSend channel.
---
 .../campaignion_m2t_send.install              | 13 ++++++------
 .../campaignion_m2t_send.module               | 20 +++++++++++++++++++
 2 files changed, 27 insertions(+), 6 deletions(-)

diff --git a/campaignion_m2t_send/campaignion_m2t_send.install b/campaignion_m2t_send/campaignion_m2t_send.install
index 7c81971e..a4966c85 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.install
+++ b/campaignion_m2t_send/campaignion_m2t_send.install
@@ -8,19 +8,20 @@ use Drupal\campaignion_m2t_send\Submission;
  * Migrate flags stored in webform_submitted_data.
  */
 function campaignion_m2t_send_install() {
+  $types = _campaignion_m2t_send_content_types();
   $sql = <<<SQL
   INSERT INTO {campaignion_m2t_send}
   SELECT nid, sid, cid, no, NULL, NULL
-  FROM {webform_submissions} s
+  FROM {node} n
+    INNER JOIN {webform_submissions} s USING (nid)
     INNER JOIN {webform_component} c USING(nid)
     INNER JOIN {webform_submitted_data} d USING(nid, sid, cid)
-  WHERE c.type='e2t_selector'
+  WHERE n.type IN(:types) AND c.type='e2t_selector'
   SQL;
-  db_query($sql);
-  $nids = db_select('webform_component', 'c')
-    ->fields('c', ['nid'])
+  db_query($sql, [':types' => $types]);
+  $nids = db_select('campaignion_m2t_send', 'm')
+    ->fields('m', ['nid'])
     ->groupBy('nid')
-    ->condition('type', 'e2t_selector')
     ->execute()
     ->fetchCol();
   $nodes = entity_load('node', $nids);
diff --git a/campaignion_m2t_send/campaignion_m2t_send.module b/campaignion_m2t_send/campaignion_m2t_send.module
index c4a0cf6a..800d27b9 100644
--- a/campaignion_m2t_send/campaignion_m2t_send.module
+++ b/campaignion_m2t_send/campaignion_m2t_send.module
@@ -5,6 +5,7 @@
  * Hook and callback implementations for this module.
  */
 
+use Drupal\campaignion_email_to_target\Channel\EmailNoSend;
 use Drupal\campaignion_m2t_send\SendMessagesCron;
 use Drupal\little_helpers\Services\Container;
 use Drupal\little_helpers\Webform\Submission;
@@ -44,10 +45,29 @@ function _campaignion_m2t_send_run($service) {
   Container::get()->loadService($service)->run();
 }
 
+/**
+ * Helper function to get all M2T content types.
+ *
+ * @return str[] Array of node type machine names.
+ */
+function _campaignion_m2t_send_content_types() {
+  $types = &drupal_static(__FUNCTION__);
+  if ($types === NULL) {
+    $info = array_filter(module_invoke_all('campaignion_action_info'), function ($i) {
+      return ($i['channel'] ?? NULL) == EmailNoSend::class;
+    });
+    $types = array_keys($info);
+  }
+  return $types;
+}
+
 /**
  * Implements hook_webform_submission_insert().
  */
 function campaignion_m2t_send_webform_submission_insert($node, $submission) {
+  if (!in_array($node->type, _campaignion_m2t_send_content_types())) {
+    return;
+  }
   $s = new Submission($node, $submission);
   foreach ($s->webform->componentsByType('e2t_selector') as $component) {
     $values = $s->valuesByCid($component['cid']);