fix(attendees): chunk the bulk attendee email send and add a sent-proof for resume-safety

.github/workflows/push.yml

@@ -80,6 +80,8 @@ jobs:
                     # subdirectories - a file added there runs nowhere unless it is listed here.
                     - { name: "PresentationMediaUploads", filter: "tests/PresentationMediaUploadsTest.php tests/PresentationMediaUploadsVisibilityTest.php tests/PresentationSerializerCacheKeyTest.php tests/PresentationReopenModelTest.php tests/PresentationReopenApiTest.php tests/PresentationReopenAuthzTest.php tests/PresentationSubmissionReopenedEmailTest.php" }
                     - { name: "SpeakerSubmitterPublishedFilter", filter: "tests/ProcessSpeakersEmailRequestJobTest.php tests/ProcessSubmittersEmailRequestJobTest.php tests/SpeakerRepositoryTest.php tests/SubmitterRepositoryTest.php tests/SpeakerServiceOriginalFilterTest.php tests/SubmitterServiceOriginalFilterTest.php" }
+                    - { name: "SpeakerBulkEmailChunkResume", filter: "tests/SpeakerServiceBulkSendChunkingTest.php tests/SpeakerServiceResumeSendEmailsTest.php tests/ProcessSpeakersEmailRequestJobResumeTest.php tests/ProcessSpeakersEmailRequestJobFailedHookTest.php" }
+                    - { name: "AttendeeBulkEmailChunkResume", filter: "tests/AttendeeServiceBulkSendChunkingTest.php tests/AttendeeServiceResumeSendEmailsTest.php tests/ProcessAttendeesEmailRequestJobResumeTest.php tests/ProcessAttendeesEmailRequestJobFailedHookTest.php tests/SummitAttendeeAnnouncementEmailTest.php tests/DoctrineSummitAttendeeRepositoryTest.php tests/SummitExtraQuestionsMemoizationTest.php" }
                     - { name: "Repositories", filter: "tests/Repositories/" }
                     - { name: "Services", filter: "tests/Unit/Services/" }
                     - { name: "Integration", filter: "tests/Integration/" }

app/Http/Controllers/Apis/Protected/Summit/OAuth2SummitAttendeesApiController.php

@@ -48,6 +48,7 @@
 use utils\Filter;
 use utils\FilterElement;
 use Illuminate\Support\Facades\Validator;
+use services\model\IAttendeeEmailFilterFields;
 use utils\FilterParser;
 
 /**
@@ -1485,42 +1486,7 @@ public function send($summit_id)
             $filter = null;
 
             if (Request::has('filter')) {
-                $filter = FilterParser::parse(Request::input('filter'), [
-                    'id' => ['=='],
-                    'not_id' => ['=='],
-                    'first_name' => ['=@', '=='],
-                    'last_name' => ['=@', '=='],
-                    'full_name' => ['=@', '=='],
-                    'company' => ['=@', '=='],
-                    'has_company' => ['=='],
-                    'email' => ['=@', '=='],
-                    'external_order_id' => ['=@', '=='],
-                    'external_attendee_id' => ['=@', '=='],
-                    'member_id' => ['==', '>'],
-                    'ticket_type' => ['=@', '==', '@@'],
-                    'ticket_type_id' => ['=='],
-                    'badge_type' => ['=@', '==', '@@'],
-                    'badge_type_id' => ['=='],
-                    'features' => ['=@', '==', '@@'],
-                    'features_id' => ['=='],
-                    'access_levels' => ['=@', '==', '@@'],
-                    'access_levels_id' => ['=='],
-                    'status' => ['=@', '=='],
-                    'has_member' => ['=='],
-                    'has_tickets' => ['=='],
-                    'has_virtual_checkin' => ['=='],
-                    'has_checkin' => ['=='],
-                    'tickets_count' => ['==', '>=', '<=', '>', '<'],
-                    'presentation_votes_date' => ['==', '>=', '<=', '>', '<'],
-                    'presentation_votes_count' => ['==', '>=', '<=', '>', '<'],
-                    'presentation_votes_track_group_id' => ['=='],
-                    'summit_hall_checked_in_date' => ['==', '>=', '<=', '>', '<','[]'],
-                    'tags' => ['=@', '==', '@@'],
-                    'tags_id' => ['=='],
-                    'notes' => ['=@', '@@'],
-                    'has_notes' => ['=='],
-                    'has_manager' => ['=='],
-                ]);
+                $filter = FilterParser::parse(Request::input('filter'), IAttendeeEmailFilterFields::OPERATORS);
             }
 
             if (is_null($filter))

app/Jobs/Emails/PresentationSubmissions/ProcessSpeakersEmailRequestJob.php

@@ -210,7 +210,15 @@ public function failed(\Throwable $e): void
      */
     private function redactFilterFieldNames($filter): array
     {
-        if (empty($filter) || !is_array($filter)) return [];
-        return array_map(fn($condition) => preg_replace('/[=<>@!].*/', '', (string)$condition), $filter);
+        if (empty($filter)) return [];
+        // FiltersParams::getFilterParam() passes the raw request value through: filter[] arrives
+        // as an array, a bare filter= as a string. FilterParser::parse accepts both by wrapping the
+        // scalar, so the redaction accepts the same shape instead of dropping the field names.
+        if (!is_array($filter)) $filter = [$filter];
+        $conditions = array_filter($filter, 'is_scalar');
+        // Cut at the first character of ANY operator FilterParser::filterExpresion recognizes
+        // (==, =@, @@, <>, <, >, <=, >=, [] range, () set) - [ and ( included, or a range/set
+        // condition's operands are logged verbatim.
+        return array_values(array_map(fn($condition) => preg_replace('/[=<>@!\[(].*/', '', (string)$condition), $conditions));
     }
 }

app/Jobs/Emails/Registration/Attendees/ProcessAttendeesEmailRequestJob.php

@@ -11,26 +11,33 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  **/
+use App\Jobs\Emails\Registration\Attendees\SummitAttendeeExcerptEmail;
+use App\Jobs\Emails\Traits\ResumableChunkJob;
+use App\Jobs\Utils\JobDispatcher;
 use App\Services\Model\IAttendeeService;
+use App\Services\utils\IEmailExcerptService;
 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\App;
+use Illuminate\Support\Facades\Config;
 use Illuminate\Support\Facades\Log;
+use models\summit\ISummitRepository;
 use models\summit\Summit;
+use services\model\IAttendeeEmailFilterFields;
 use utils\FilterParser;
 /**
  * Class ProcessAttendeesEmailRequestJob
  * @package App\Jobs\Emails
  */
 final class ProcessAttendeesEmailRequestJob implements ShouldQueue
 {
-    public $timeout = 0;
-
-    public $tries = 1;
-
-    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+    // $timeout/$tries/$backoff and the resume-on-retry mechanics come from ResumableChunkJob -
+    // see that trait's doc comment for why timeout must stay below every retry_after / worker
+    // --timeout this job can run under.
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels, ResumableChunkJob;
 
     private $summit_id;
 
@@ -63,44 +70,124 @@ public function handle(IAttendeeService $service){
             )
         );
 
-        $filter = !is_null($this->filter) ? FilterParser::parse($this->filter, [
-            'id' => ['=='],
-            'not_id' => ['=='],
-            'first_name' => ['=@', '=='],
-            'last_name' => ['=@', '=='],
-            'full_name' => ['=@', '=='],
-            'company' => ['=@', '=='],
-            'has_company' => ['=='],
-            'email' => ['=@', '=='],
-            'external_order_id' => ['=@', '=='],
-            'external_attendee_id' => ['=@', '=='],
-            'member_id' => ['==', '>'],
-            'ticket_type' => ['=@', '==', '@@'],
-            'ticket_type_id' => ['=='],
-            'badge_type' => ['=@', '==', '@@'],
-            'badge_type_id' => ['=='],
-            'features' => ['=@', '==', '@@'],
-            'features_id' => ['=='],
-            'access_levels' => ['=@', '==', '@@'],
-            'access_levels_id' => ['=='],
-            'status' => ['=@', '=='],
-            'has_member' => ['=='],
-            'has_tickets' => ['=='],
-            'has_virtual_checkin' => ['=='],
-            'has_checkin' => ['=='],
-            'tickets_count' => ['==', '>=', '<=', '>', '<'],
-            'presentation_votes_date' => ['==', '>=', '<=', '>', '<'],
-            'presentation_votes_count' => ['==', '>=', '<=', '>', '<'],
-            'presentation_votes_track_group_id' => ['=='],
-            'summit_hall_checked_in_date' => ['==', '>=', '<=', '>', '<','[]'],
-            'tags' => ['=@', '==', '@@'],
-            'tags_id' => ['=='],
-            'notes' => ['=@', '@@'],
-            'has_notes' => ['=='],
-            'has_manager' => ['==']
-        ]) : null;
+        // ResumableChunkJob::activateResumeIfRetrying(): resume, not resend. On a retry it sets
+        // resume_since = dispatched_at in $this->payload so AttendeeService::send skips only the
+        // attendees (or, for the ticket flow event, attendee+ticket pairs) whose proof for this
+        // run was written by THIS run.
+        $this->activateResumeIfRetrying();
+
+        $filter = !is_null($this->filter) ? FilterParser::parse($this->filter, IAttendeeEmailFilterFields::OPERATORS) : null;
 
         $service->send($this->summit_id, $this->payload, $filter);
     }
 
+    /**
+     * Invoked by the queue worker once this job is marked failed - with ResumableChunkJob's
+     * tries = 2 and the resume check inside AttendeeService::send (Task 3), that means BOTH
+     * attempts failed: a chunk whose worker was merely killed mid-run (rolling deploy, OOM,
+     * scale-down) is re-served and automatically resumed once, skipping only the attendees this
+     * run already reached. This hook only fires when that automatic resume itself also failed to
+     * finish the chunk. Nothing else reports that loss - the outcome excerpt is only sent when
+     * send() runs to completion - so without this hook a dead chunk leaves no trace beyond a
+     * queue_failed_jobs row.
+     *
+     * Log the chunk's attendee ids at error, and when the operator asked for an outcome e-mail
+     * send one naming them, so the chunk can be re-sent by id. Mirrors
+     * ProcessSpeakersEmailRequestJob::failed(), simplified: attendees generate no promo codes on
+     * this path, so there is no should_resend/promo-code resend caveat to add.
+     *
+     * @param \Throwable $e
+     */
+    public function failed(\Throwable $e): void
+    {
+        $attendee_ids = $this->payload['attendees_ids'] ?? [];
+        $flow_event = $this->payload['email_flow_event'] ?? '';
+        $ids_list = implode(', ', $attendee_ids);
+
+        Log::error
+        (
+            sprintf
+            (
+                "ProcessAttendeesEmailRequestJob::failed summit %s flow_event %s: chunk of %s attendee(s) failed (%s: %s); up to %s of them may not have been processed. Attendee ids in the chunk: [%s] filter fields %s.",
+                $this->summit_id,
+                $flow_event,
+                count($attendee_ids),
+                get_class($e),
+                $e->getMessage(),
+                count($attendee_ids),
+                $ids_list,
+                json_encode($this->redactFilterFieldNames($this->filter))
+            )
+        );
+
+        $outcome_email_recipient = $this->payload['outcome_email_recipient'] ?? null;
+        if (empty($outcome_email_recipient)) return;
+
+        try {
+            $summit = App::make(ISummitRepository::class)->getById($this->summit_id);
+            if (!$summit instanceof Summit) {
+                Log::warning(sprintf("ProcessAttendeesEmailRequestJob::failed summit %s not found, outcome excerpt not sent", $this->summit_id));
+                return;
+            }
+
+            // Same line types AbstractExcerptEmailJob renders for a completed run, so the
+            // operator's inbox reads the same either way.
+            $report = [
+                [
+                    'type' => IEmailExcerptService::InfoType,
+                    'message' => sprintf("Processing EMAIL %s for Summit %s", $flow_event, $this->summit_id),
+                ],
+                [
+                    'type' => IEmailExcerptService::ErrorType,
+                    'message' => sprintf
+                    (
+                        "Chunk of %s attendee(s) failed (%s); up to %s of them may not have been processed. Attendee ids in the chunk: %s.",
+                        count($attendee_ids),
+                        $e->getMessage(),
+                        count($attendee_ids),
+                        $ids_list
+                    ),
+                ],
+                [
+                    'type' => IEmailExcerptService::InfoType,
+                    'message' => "TOTAL processed for this chunk is unknown, the job did not run to completion",
+                ],
+            ];
+
+            // Same failover route as the chunk itself (AttendeeService::triggerSend): a chunk
+            // runs on the database fallback worker precisely when the redis primary was down at
+            // dispatch time, so a bare ::dispatch() here would throw into the catch below and lose
+            // the report in the one scenario it exists for.
+            JobDispatcher::withDbFallback(
+                job: new SummitAttendeeExcerptEmail($summit, $outcome_email_recipient, $report),
+                logContext: ['summit_id' => $this->summit_id, 'attendee_count' => count($attendee_ids)],
+                primaryConnection: Config::get('queue.default')
+            );
+        }
+        catch (\Throwable $ex) {
+            Log::error($ex);
+        }
+    }
+
+    /**
+     * Reduces a raw filter (["email==foo@bar.com", "first_name==Jane"]) to just its field names
+     * (["email", "first_name"]) so error logs never carry filter values that may be PII - email
+     * and first_name are valid filter fields (IAttendeeEmailFilterFields::OPERATORS).
+     *
+     * @param mixed $filter
+     * @return string[]
+     */
+    private function redactFilterFieldNames($filter): array
+    {
+        if (empty($filter)) return [];
+        // FiltersParams::getFilterParam() passes the raw request value through: filter[] arrives
+        // as an array, a bare filter= as a string. FilterParser::parse accepts both by wrapping the
+        // scalar, so the redaction accepts the same shape instead of dropping the field names.
+        if (!is_array($filter)) $filter = [$filter];
+        $conditions = array_filter($filter, 'is_scalar');
+        // Cut at the first character of ANY operator FilterParser::filterExpresion recognizes
+        // (==, =@, @@, <>, <, >, <=, >=, [] range, () set) - [ and ( included, or a range/set
+        // condition's operands are logged verbatim.
+        return array_values(array_map(fn($condition) => preg_replace('/[=<>@!\[(].*/', '', (string)$condition), $conditions));
+    }
 }

app/Models/Foundation/Summit/Registration/Attendees/SummitAttendee.php

@@ -216,6 +216,12 @@ public function getSummitHallCheckedIn()
     #[ORM\ManyToMany(targetEntity: \models\main\Tag::class, cascade: ['persist'], fetch: 'EXTRA_LAZY')]
     private $tags;
 
+    /**
+     * @var \Doctrine\Common\Collections\Collection<int, SummitAttendeeAnnouncementEmail>
+     */
+    #[ORM\OneToMany(targetEntity: \models\summit\SummitAttendeeAnnouncementEmail::class, mappedBy: 'attendee', cascade: ['persist', 'remove'], orphanRemoval: true, fetch: 'EXTRA_LAZY')]
+    private $announcement_emails;
+
     /**
      * @param bool $summit_hall_checked_in
      */
@@ -297,6 +303,51 @@ public function getFirstTicket():?SummitAttendeeTicket{
         return $this->tickets->first();
     }
 
+    public function addAnnouncementEmail(SummitAttendeeAnnouncementEmail $announcementEmail)
+    {
+        if ($this->announcement_emails->contains($announcementEmail)) return;
+        $this->announcement_emails->add($announcementEmail);
+        $announcementEmail->setAttendee($this);
+    }
+
+    public function removeAnnouncementEmail(SummitAttendeeAnnouncementEmail $announcementEmail)
+    {
+        if (!$this->announcement_emails->contains($announcementEmail)) return;
+        $this->announcement_emails->removeElement($announcementEmail);
+        $announcementEmail->clearAttendee();
+    }
+
+    /**
+     * Per-recipient, per-email-type, timestamped proof check backing the bulk-send resume
+     * predicate (AbstractEmailAction::alreadySentSince). Bounded matching() query on the
+     * EXTRA_LAZY announcement_emails collection - does not hydrate it. $ticket distinguishes
+     * SummitAttendeeTicketEmailStrategy's per-ticket proofs (one email per active paid ticket)
+     * from every other strategy's single per-attendee proof (ticket left null).
+     *
+     * @param Summit $summit
+     * @param string $type
+     * @param \DateTime $since
+     * @param SummitAttendeeTicket|null $ticket
+     * @return bool
+     */
+    public function hasAnnouncementEmailTypeSentSince(Summit $summit, string $type, \DateTime $since, ?SummitAttendeeTicket $ticket = null): bool
+    {
+        $criteria = Criteria::create();
+
+        $criteria
+            ->where(Criteria::expr()->eq('summit', $summit))
+            ->andWhere(Criteria::expr()->eq('type', $type))
+            ->andWhere(Criteria::expr()->gte('send_date', $since));
+
+        if (!is_null($ticket)) {
+            $criteria->andWhere(Criteria::expr()->eq('ticket', $ticket));
+        } else {
+            $criteria->andWhere(Criteria::expr()->isNull('ticket'));
+        }
+
+        return $this->announcement_emails->matching($criteria)->count() > 0;
+    }
+
     /**
      * @param SummitAttendeeTicket $ticket
      */
@@ -349,6 +400,7 @@ public function __construct()
         $this->managed_attendees = new ArrayCollection();
         $this->manager = null;
         $this->rsvp_invitations = new ArrayCollection();
+        $this->announcement_emails = new ArrayCollection();
     }
 
     public function isVirtualCheckedIn(): bool