Merge remote-tracking branch 'origin/develop' into feature/platform-reuse

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/ddl/postgres.sql

@@ -1,6 +1,6 @@
 -- Copyright 2026 Kingsrook, LLC. Licensed under the Apache License, Version 2.0.
 -- https://www.apache.org/licenses/LICENSE-2.0
--- QBit CRM 0.2.0 P1 PostgreSQL reference schema (33 tables).
+-- QBit CRM 0.2.0 P1 PostgreSQL reference schema (34 tables).
 -- QQQ table names equal SQL names; Java field names map to snake_case columns.
 -- Unique constraints match QQQ metadata. Relationship columns are indexed;
 -- foreign-key constraints are deliberately not imposed (the QQQ/host layer owns
@@ -189,7 +189,7 @@ CREATE TABLE IF NOT EXISTS crm_contact (
    company_id INTEGER,
    lifecycle_stage_id INTEGER,
    lead_source_id INTEGER,
-   owner_user_id TEXT NOT NULL,
+   owner_user_id TEXT,
    address_line1 VARCHAR(250),
    address_line2 VARCHAR(250),
    city VARCHAR(100),
@@ -252,7 +252,7 @@ CREATE TABLE IF NOT EXISTS crm_deal (
    weighted_amount DECIMAL(20,6),
    company_id INTEGER,
    primary_contact_id INTEGER,
-   owner_user_id TEXT NOT NULL,
+   owner_user_id TEXT,
    lead_source_id INTEGER,
    expected_close_date DATE,
    actual_close_date DATE,
@@ -364,9 +364,13 @@ CREATE INDEX IF NOT EXISTS ix_crm_email_list_member_form_submission_id ON crm_em
 
 CREATE TABLE IF NOT EXISTS crm_email_message (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   idempotency_key VARCHAR(250) UNIQUE,
+   email_message_id VARCHAR(100),
+   delivered_date TIMESTAMP WITHOUT TIME ZONE,
+   bounced_date TIMESTAMP WITHOUT TIME ZONE,
    activity_id INTEGER,
    contact_id INTEGER,
-   from_address VARCHAR(250) NOT NULL,
+   from_address VARCHAR(250),
    from_name VARCHAR(100),
    to_addresses TEXT NOT NULL,
    cc_addresses TEXT,
@@ -392,6 +396,7 @@ CREATE TABLE IF NOT EXISTS crm_email_message (
    modify_date TIMESTAMP WITHOUT TIME ZONE
 );
 
+CREATE INDEX IF NOT EXISTS ix_crm_email_message_email_message_id ON crm_email_message (email_message_id);
 CREATE INDEX IF NOT EXISTS ix_crm_email_message_contact_id ON crm_email_message (contact_id);
 CREATE INDEX IF NOT EXISTS ix_crm_email_message_activity_id ON crm_email_message (activity_id);
 CREATE INDEX IF NOT EXISTS ix_crm_email_message_thread_id ON crm_email_message (thread_id);
@@ -640,3 +645,10 @@ CREATE TABLE IF NOT EXISTS crm_win_loss_reason (
 );
 
 CREATE INDEX IF NOT EXISTS ix_crm_consent_record_evidence_form_submission_id ON crm_consent_record (evidence_form_submission_id);
+
+CREATE TABLE IF NOT EXISTS crm_job_state (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   name VARCHAR(250) NOT NULL UNIQUE,
+   last_success TIMESTAMP WITHOUT TIME ZONE,
+   last_full_pass TIMESTAMP WITHOUT TIME ZONE
+);

src/main/java/com/kingsrook/qbits/crm/CrmQBitConfig.java

@@ -43,6 +43,9 @@ public class CrmQBitConfig implements QBitConfig
 {
    public static final CrmQBitConfig DEFAULT_CONFIG = new CrmQBitConfig();
 
+   private String schedulerName;
+   private Boolean schedulesEnabled = true;
+
    private String messagingProviderName;
 
    private String publicBaseUrl;
@@ -271,4 +274,38 @@ public class CrmQBitConfig implements QBitConfig
       return (this);
    }
 
+   /*******************************************************************************
+    ** Scheduler configuration: schedulerName.
+    *******************************************************************************/
+   public String getSchedulerName()
+   {
+      return (schedulerName);
+   }
+
+   /*******************************************************************************
+    ** Configure schedulerName.
+    *******************************************************************************/
+   public CrmQBitConfig withSchedulerName(String schedulerName)
+   {
+      this.schedulerName = schedulerName;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Scheduler configuration: schedulesEnabled.
+    *******************************************************************************/
+   public Boolean getSchedulesEnabled()
+   {
+      return (schedulesEnabled);
+   }
+
+   /*******************************************************************************
+    ** Configure schedulesEnabled.
+    *******************************************************************************/
+   public CrmQBitConfig withSchedulesEnabled(Boolean schedulesEnabled)
+   {
+      this.schedulesEnabled = schedulesEnabled;
+      return (this);
+   }
+
 }

src/main/java/com/kingsrook/qbits/crm/CrmQBitProducer.java

@@ -28,6 +28,7 @@ package com.kingsrook.qbits.crm;
 
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import com.kingsrook.qbits.crm.activities.customizers.ActivityLastDateCustomizer;
 import com.kingsrook.qbits.crm.activities.model.Activity;
 import com.kingsrook.qbits.crm.core.customizers.FormSubmissionAutoLinkCustomizer;
@@ -39,6 +40,8 @@ import com.kingsrook.qbits.crm.deals.customizers.DealProductRecalculationCustomi
 import com.kingsrook.qbits.crm.deals.customizers.DealStageChangeCustomizer;
 import com.kingsrook.qbits.crm.deals.model.Deal;
 import com.kingsrook.qbits.crm.deals.model.DealProduct;
+import com.kingsrook.qbits.crm.email.CrmEmailProcessIdentity;
+import com.kingsrook.qbits.crm.scoring.AutoAssignCustomizer;
 import com.kingsrook.qbits.crm.security.CrmPermissions;
 import com.kingsrook.qbits.crm.widgets.CrmWidgetMetaData;
 import com.kingsrook.qqq.backend.core.actions.customizers.MultiCustomizer;
@@ -54,6 +57,9 @@ import com.kingsrook.qqq.backend.core.model.metadata.audits.AuditLevel;
 import com.kingsrook.qqq.backend.core.model.metadata.audits.QAuditRules;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReferenceWithProperties;
+import com.kingsrook.qqq.backend.core.model.metadata.collaboration.RecordFeature;
+import com.kingsrook.qqq.backend.core.model.metadata.collaboration.RecordFeaturesMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.collaboration.RecordOwnership;
 import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.joins.QJoinMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppSection;
@@ -61,6 +67,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
 import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaDataProducer;
+import com.kingsrook.qqq.backend.core.model.metadata.scheduleing.QScheduleMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QFieldSection;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.Tier;
@@ -139,6 +146,12 @@ public class CrmQBitProducer implements QBitMetaDataProducer<CrmQBitConfig>
       {
          process.setPermissionRules(CrmPermissions.processRules(process.getName()));
          process.setSourceQBitName(getQBitMetaData().getName());
+         Integer seconds = Map.of("crm_process_sequence_steps", 300, "crm_calculate_lead_scores", 3600,
+            "crm_sync_email_delivery", 900, "crm_expire_consents", 86400).get(process.getName());
+         if(seconds != null && qBitConfig.getSchedulerName() != null && !Boolean.FALSE.equals(qBitConfig.getSchedulesEnabled()))
+         {
+            process.withSchedule(new QScheduleMetaData().withSchedulerName(qBitConfig.getSchedulerName()).withRepeatSeconds(seconds));
+         }
       }
 
       /////////////////////////////////////////////////////////////////////////
@@ -171,11 +184,26 @@ public class CrmQBitProducer implements QBitMetaDataProducer<CrmQBitConfig>
             || tableName.equals(Deal.TABLE_NAME) || tableName.equals(Activity.TABLE_NAME))
          {
             table.withAuditRules(new QAuditRules().withAuditLevel(AuditLevel.FIELD));
+            RecordFeaturesMetaData features = RecordFeaturesMetaData.ofOrWithNew(table)
+               .withOwnership(new RecordOwnership().withOwnerFieldName("ownerUserId"));
+            for(RecordFeature feature : List.of(RecordFeature.TAGS, RecordFeature.COMMENTS, RecordFeature.ATTACHMENTS,
+               RecordFeature.FOLLOW, RecordFeature.TIMELINE, RecordFeature.RECENTS, RecordFeature.FAVORITES))
+            {
+               features.withEnabled(feature, true);
+            }
+            if(!tableName.equals(Activity.TABLE_NAME))
+            {
+               features.withEnabled(RecordFeature.TASKS, true).withEnabled(RecordFeature.ASSIGNMENT, true);
+            }
          }
 
          ///////////////////////////////////////////////////////////////
          // wire activity timestamp propagation                       //
          ///////////////////////////////////////////////////////////////
+         if(tableName.equals(Contact.TABLE_NAME) || tableName.equals(Deal.TABLE_NAME))
+         {
+            addOrComposeCustomizer(table, TableCustomizers.POST_INSERT_RECORD, new QCodeReference(AutoAssignCustomizer.class));
+         }
          if(tableName.equals(Activity.TABLE_NAME))
          {
             addOrComposeCustomizer(table, TableCustomizers.POST_INSERT_RECORD, new QCodeReference(ActivityLastDateCustomizer.class));
@@ -324,6 +352,12 @@ public class CrmQBitProducer implements QBitMetaDataProducer<CrmQBitConfig>
          .withSourceQBitName(getQBitMetaData().getName()));
       output.add(new QAlertTypeMetaData().withName("crm.email.unavailable").withLabel("Email unavailable")
          .withSourceQBitName(getQBitMetaData().getName()));
+      for(String notification : List.of("crm.lead.assigned", "crm.deal.won", "crm.deal.rotting", "crm.task.due"))
+      {
+         output.add(new QAlertTypeMetaData().withName(notification).withLabel(notification)
+            .withDefaultSeverity(AlertSeverity.INFO).withSourceQBitName(getQBitMetaData().getName()));
+      }
+      qInstance.withProcessLifecycleListener(new QCodeReference(CrmEmailProcessIdentity.class));
       registerHooks(qInstance);
       return (output);
    }

src/main/java/com/kingsrook/qbits/crm/actions/CrmAction.java

@@ -32,6 +32,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import com.kingsrook.qbits.crm.activities.processes.LogActivityProcess;
+import com.kingsrook.qbits.crm.api.CrmApi;
 import com.kingsrook.qbits.crm.core.model.enums.CrmEnrollmentStatus;
 import com.kingsrook.qbits.crm.core.processes.ConvertFormToContactProcess;
 import com.kingsrook.qbits.crm.core.processes.GdprAnonymizeContactProcess;
@@ -42,6 +43,8 @@ import com.kingsrook.qbits.crm.core.processes.UnsubscribeFromListProcess;
 import com.kingsrook.qbits.crm.deals.processes.CloneDealProcess;
 import com.kingsrook.qbits.crm.deals.processes.ConvertLeadProcess;
 import com.kingsrook.qbits.crm.deals.processes.ManageContactOnDealProcess;
+import com.kingsrook.qbits.crm.email.CrmEmailTransport;
+import com.kingsrook.qbits.crm.email.CrmEmails;
 import com.kingsrook.qbits.crm.email.processes.EnrollInSequenceProcess;
 import com.kingsrook.qbits.crm.email.processes.UnenrollFromSequenceProcess;
 import com.kingsrook.qbits.crm.scoring.processes.AutoAssignProcess;
@@ -84,7 +87,7 @@ public final class CrmAction
 {
    private static final QLogger LOG = QLogger.getLogger(CrmAction.class);
 
-   public static final String EMAIL_WARNING = "Email sending arrives with the email transport";
+
 
    /*******************************************************************************
     ** Utility holder.
@@ -162,6 +165,14 @@ public final class CrmAction
     *******************************************************************************/
    public static class Transform extends AbstractTransformStep implements ProcessSummaryProviderInterface
    {
+      /*******************************************************************************
+       ** Resolve the host transport; tests may supply a fake.
+       *******************************************************************************/
+      protected CrmEmailTransport emailTransport() throws QException
+      {
+         return (new CrmEmailTransport(CrmApi.config()));
+      }
+
       private final Map<String, ProcessSummaryLine> lines = new LinkedHashMap<>();
       private final Map<String, List<Serializable>> duplicateGroups = new LinkedHashMap<>();
       private List<QRecord> pair;
@@ -190,7 +201,7 @@ public final class CrmAction
             try
             {
                QRecord row = requireRow(action, input, id);
-               String reason = check(action, input, row);
+               String reason = check(action, input, row, action.operation().equals("email") ? emailTransport() : null);
                if(reason != null)
                {
                   add(lines, Status.WARNING, reason, reason, id);
@@ -253,6 +264,14 @@ public final class CrmAction
     *******************************************************************************/
    public static class Load extends AbstractLoadStep implements ProcessSummaryProviderInterface
    {
+      /*******************************************************************************
+       ** Resolve the host transport; tests may supply a fake.
+       *******************************************************************************/
+      protected CrmEmailTransport emailTransport() throws QException
+      {
+         return (new CrmEmailTransport(CrmApi.config()));
+      }
+
       private final Map<String, ProcessSummaryLine> lines = new LinkedHashMap<>();
       private boolean duplicates;
 
@@ -267,14 +286,14 @@ public final class CrmAction
             try
             {
                QRecord row = requireRow(action, input, id);
-               String reason = check(action, input, row);
+               String reason = check(action, input, row, action.operation().equals("email") ? emailTransport() : null);
                if(reason != null)
                {
                   add(lines, Status.WARNING, reason, reason, id);
                   continue;
                }
                row.setValue("mergeSecondaryId", reviewed.getValue("mergeSecondaryId"));
-               String skipped = perform(action, input, row);
+               String skipped = perform(action, input, row, action.operation().equals("email") ? emailTransport() : null);
                if(skipped == null)
                {
                   add(lines, Status.OK, "will be " + action.past(), "was " + action.past(), id);
@@ -345,7 +364,7 @@ public final class CrmAction
    /*******************************************************************************
     ** Shared read-only checks for Review and Result.
     *******************************************************************************/
-   private static String check(CrmActions.Action action, RunBackendStepInput input, QRecord row) throws QException
+   private static String check(CrmActions.Action action, RunBackendStepInput input, QRecord row, CrmEmailTransport transport) throws QException
    {
       for(QFieldMetaData field : CrmActions.inputFields(action))
       {
@@ -365,15 +384,7 @@ public final class CrmAction
       switch(action.operation())
       {
          case "email":
-            try
-            {
-               Class.forName("com.kingsrook.qbits.crm.email.CrmEmailTransport");
-               return ("Email transport is present; sending integration is not configured for this action yet");
-            }
-            catch(ClassNotFoundException expected)
-            {
-               return (EMAIL_WARNING);
-            }
+            return (CrmEmails.skipReason(row, transport));
          case "subscribe", "unsubscribe":
             if(!StringUtils.hasContent(row.getValueString("email")))
             {
@@ -479,7 +490,7 @@ public final class CrmAction
    /*******************************************************************************
     ** Invoke the existing business operations only at Load time.
     *******************************************************************************/
-   private static String perform(CrmActions.Action action, RunBackendStepInput input, QRecord row) throws QException
+   private static String perform(CrmActions.Action action, RunBackendStepInput input, QRecord row, CrmEmailTransport transport) throws QException
    {
       RunBackendStepInput call = new RunBackendStepInput().withProcessName(input.getProcessName()).withValues(new HashMap<>());
       for(QFieldMetaData field : CrmActions.inputFields(action))
@@ -494,6 +505,16 @@ public final class CrmAction
       RunBackendStepOutput result = new RunBackendStepOutput();
       switch(action.operation())
       {
+         case "email" ->
+         {
+            String processUUID = input.getValueString("crmEmailProcessUUID");
+            if(processUUID == null)
+            {
+               throw (new QUserFacingException("Email process identity is missing; start the Send Email action again"));
+            }
+            CrmEmails.send(transport, row, input.getValueInteger("emailTemplateId"), input.getValueString("subject"), input.getValueString("body"),
+               "email", id + ":" + processUUID, null);
+         }
          case "assign" -> update(action.table(), new QRecord().withValue("id", id).withValue("ownerUserId", input.getValue("newOwnerUserId")));
          case "auto" ->
          {

src/main/java/com/kingsrook/qbits/crm/actions/CrmActions.java

@@ -148,6 +148,12 @@ public class CrmActions implements MetaDataProducerInterface<MetaDataProducerMul
       List<QFieldMetaData> fields = new ArrayList<>();
       switch(action.operation())
       {
+         case "email" ->
+         {
+            fields.add(field("emailTemplateId", "Template", QFieldType.INTEGER, "crm_email_template").withIsRequired(false));
+            fields.add(field("subject", "Subject", QFieldType.STRING, null).withIsRequired(false));
+            fields.add(field("body", "Body", QFieldType.TEXT, null).withIsRequired(false));
+         }
          case "assign" -> fields.add(field("newOwnerUserId", "Owner", QFieldType.STRING, null));
          case "subscribe", "unsubscribe" -> fields.add(field("emailListId", "List", QFieldType.INTEGER, "crm_email_list"));
          case "enroll", "unenroll" -> fields.add(field("sequenceId", "Sequence", QFieldType.INTEGER, "crm_email_sequence"));

src/main/java/com/kingsrook/qbits/crm/api/CrmIntakeApi.java

@@ -323,7 +323,7 @@ public class CrmIntakeApi
       }
       update.withValue("lastTouchDate", Instant.now());
       store.update("crm_contact", update);
-      if(contacts.isEmpty() && isUnassigned(contact.getValueString("ownerUserId")))
+      if(contacts.isEmpty() && isUnassigned(store.get("crm_contact", contact.getValueInteger("id")).getValueString("ownerUserId")))
       {
          CrmAutoAssign.assign(CrmAssignableEntity.CONTACT, contact.getValueInteger("id"));
       }

src/main/java/com/kingsrook/qbits/crm/core/CrmRecords.java

@@ -0,0 +1,72 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.core;
+
+
+import java.util.List;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Checked writes for background work: record errors must not advance checkpoints.
+ *******************************************************************************/
+public final class CrmRecords
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private CrmRecords()
+   {
+   }
+
+   /*******************************************************************************
+    ** Insert and require a successful persisted row.
+    *******************************************************************************/
+   public static QRecord insert(String table, QRecord row) throws QException
+   {
+      return (check(new InsertAction().execute(new InsertInput(table).withRecord(row)).getRecords()));
+   }
+
+   /*******************************************************************************
+    ** Update and require a successful persisted row.
+    *******************************************************************************/
+   public static QRecord update(String table, QRecord row) throws QException
+   {
+      return (check(new UpdateAction().execute(new UpdateInput(table).withRecord(row)).getRecords()));
+   }
+
+   /*******************************************************************************
+    ** DML errors are failed work, never successful job progress.
+    *******************************************************************************/
+   private static QRecord check(List<QRecord> rows) throws QException
+   {
+      if(rows.isEmpty() || rows.stream().anyMatch(row -> row.getErrors() != null && !row.getErrors().isEmpty()))
+      {
+         throw (new QException("CRM write was rejected"));
+      }
+      return (rows.getFirst());
+   }
+}

src/main/java/com/kingsrook/qbits/crm/core/model/Contact.java

@@ -241,7 +241,7 @@ public class Contact extends QRecordEntity
    @QField(possibleValueSourceName = LeadSource.TABLE_NAME)
    private Integer leadSourceId;
 
-   @QField(isRequired = true)
+   @QField()
    private String ownerUserId;
 
    @QField(maxLength = 250, valueTooLongBehavior = ValueTooLongBehavior.ERROR)

src/main/java/com/kingsrook/qbits/crm/core/model/enums/CrmEmailStatus.java

@@ -40,7 +40,13 @@ public enum CrmEmailStatus implements PossibleValueEnum<Integer>
    DRAFT(1, "Draft"),
    SENT(2, "Sent"),
    DELIVERED(3, "Delivered"),
-   FAILED(4, "Failed");
+   FAILED(4, "Failed"),
+   QUEUED(5, "Queued"),
+   BOUNCED(6, "Bounced"),
+   COMPLAINED(7, "Complained"),
+   OPENED(8, "Opened"),
+   CLICKED(9, "Clicked"),
+   SUPPRESSED(10, "Suppressed");
 
    private final Integer id;
    private final String  label;

src/main/java/com/kingsrook/qbits/crm/deals/CrmDealNotifications.java

@@ -0,0 +1,108 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.deals;
+
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.crm.core.CrmRecords;
+import com.kingsrook.qbits.crm.scoring.JobState;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.alerts.NotificationAudience;
+import com.kingsrook.qqq.backend.core.alerts.NotificationSignal;
+import com.kingsrook.qqq.backend.core.alerts.QAlerts;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Deal age notifications with a persistent once-per-deal-and-stage checkpoint.
+ *******************************************************************************/
+public final class CrmDealNotifications
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private CrmDealNotifications()
+   {
+   }
+
+   /*******************************************************************************
+    ** Scan open deals in keyset pages, resolving stage metadata once per run.
+    *******************************************************************************/
+   public static void checkRotting() throws QException
+   {
+      Map<Integer, QRecord> stages = new HashMap<>();
+      for(QRecord stage : new QueryAction().execute(new QueryInput("crm_pipeline_stage")).getRecords())
+      {
+         stages.put(stage.getValueInteger("id"), stage);
+      }
+      Integer lastId = null;
+      Instant now = Instant.now();
+      while(true)
+      {
+         QQueryFilter filter = new QQueryFilter().withOrderBy(new QFilterOrderBy("id")).withLimit(200);
+         if(lastId != null)
+         {
+            filter.withCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, lastId));
+         }
+         List<QRecord> page = new QueryAction().execute(new QueryInput("crm_deal").withFilter(filter)).getRecords();
+         for(QRecord deal : page)
+         {
+            lastId = deal.getValueInteger("id");
+            QRecord stage = stages.get(deal.getValueInteger("pipelineStageId"));
+            Instant entered = deal.getValueInstant("stageEnteredDate");
+            if(stage == null || entered == null || deal.getValueString("ownerUserId") == null
+               || Boolean.TRUE.equals(stage.getValueBoolean("isClosedWon")) || Boolean.TRUE.equals(stage.getValueBoolean("isClosedLost")))
+            {
+               continue;
+            }
+            Integer days = stage.getValueInteger("rotDays");
+            if(days == null || days <= 0 || !entered.isBefore(now.minus(days, ChronoUnit.DAYS)))
+            {
+               continue;
+            }
+            String key = "deal:" + lastId + ":stage:" + stage.getValue("id");
+            String checkpoint = "crm.deal.rotting:" + key;
+            if(new QueryAction().execute(new QueryInput(JobState.TABLE_NAME)
+               .withFilter(new QQueryFilter(new QFilterCriteria("name", QCriteriaOperator.EQUALS, checkpoint)))).getRecords().isEmpty())
+            {
+               QAlerts.notify(new NotificationSignal("Deal needs attention").withSourceName("crm.deal.rotting").withSourceKey(key)
+                  .withSourceTableName("crm_deal").withSourceRecordId(lastId.toString())
+                  .withAudience(NotificationAudience.user(deal.getValueString("ownerUserId"))));
+               CrmRecords.insert(JobState.TABLE_NAME, new QRecord().withValue("name", checkpoint).withValue("lastSuccess", now));
+            }
+         }
+         if(page.size() < 200)
+         {
+            break;
+         }
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/crm/deals/customizers/DealStageChangeCustomizer.java

@@ -51,6 +51,9 @@ import com.kingsrook.qbits.crm.sync.CrmCurrencyUtils;
 import com.kingsrook.qqq.backend.core.actions.customizers.AbstractPreUpdateCustomizer;
 import com.kingsrook.qqq.backend.core.actions.tables.GetAction;
 import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.alerts.NotificationAudience;
+import com.kingsrook.qqq.backend.core.alerts.NotificationSignal;
+import com.kingsrook.qqq.backend.core.alerts.QAlerts;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.actions.tables.get.GetInput;
@@ -192,6 +195,14 @@ public class DealStageChangeCustomizer extends AbstractPreUpdateCustomizer
       // insert DealStageHistory row           //
       ///////////////////////////////////////////
       insertStageHistoryRow(record, oldRecord, oldStageId, newStageId, now);
+      String owner = record.getValues().containsKey("ownerUserId") ? record.getValueString("ownerUserId") : oldRecord.getValueString("ownerUserId");
+      if(Boolean.TRUE.equals(newStage.getIsClosedWon()) && owner != null)
+      {
+         QAlerts.notify(new NotificationSignal("Deal won").withSourceName("crm.deal.won")
+            .withSourceKey("deal:" + record.getValue("id") + ":stage:" + newStageId + ":" + now)
+            .withSourceTableName(Deal.TABLE_NAME).withSourceRecordId(record.getValueString("id"))
+            .withAudience(NotificationAudience.user(owner)), updateInput == null ? null : updateInput.getTransaction());
+      }
    }
 
 

src/main/java/com/kingsrook/qbits/crm/deals/model/Deal.java

@@ -195,7 +195,7 @@ public class Deal extends QRecordEntity
    @QField(possibleValueSourceName = Contact.TABLE_NAME)
    private Integer primaryContactId;
 
-   @QField(isRequired = true)
+   @QField()
    private String ownerUserId;
 
    @QField(possibleValueSourceName = LeadSource.TABLE_NAME)

src/main/java/com/kingsrook/qbits/crm/email/CrmEmailProcessIdentity.java

@@ -0,0 +1,55 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.email;
+
+
+import com.kingsrook.qqq.backend.core.actions.processes.listeners.ProcessLifecycleListenerInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
+
+
+/*******************************************************************************
+ ** Carry the core process UUID across Review/Result and retries in process state.
+ *******************************************************************************/
+public class CrmEmailProcessIdentity implements ProcessLifecycleListenerInterface
+{
+   @Override
+   public boolean appliesTo(String processName)
+   {
+      return ("crm_send_email".equals(processName));
+   }
+
+   @Override
+   public void onProcessStarted(RunProcessInput input)
+   {
+      input.getProcessState().getValues().put("crmEmailProcessUUID", input.getProcessUUID());
+   }
+
+   @Override
+   public void onProcessCompleted(RunProcessInput input, RunProcessOutput output)
+   {
+   }
+
+   @Override
+   public void onProcessFailed(RunProcessInput input, Exception exception)
+   {
+   }
+}

src/main/java/com/kingsrook/qbits/crm/email/CrmEmailTransport.java

@@ -21,8 +21,11 @@
 package com.kingsrook.qbits.crm.email;
 
 
+import java.time.Instant;
+import java.util.Optional;
 import com.kingsrook.qbits.crm.CrmQBitConfig;
 import com.kingsrook.qbits.crm.api.CrmReadiness;
+import com.kingsrook.qbits.crm.core.model.enums.CrmEmailStatus;
 import com.kingsrook.qbits.email.api.EmailApi;
 import com.kingsrook.qbits.email.api.EmailApiCategory;
 import com.kingsrook.qbits.email.api.EmailApiEffectOutcome;
@@ -87,6 +90,14 @@ public class CrmEmailTransport
     ** Deliver one message, using a stable CRM purpose/id key for the durable email outbox.
     *******************************************************************************/
    public void send(String purpose, String id, Integer contactId, String address, String subject, String body, boolean marketing) throws QException
+   {
+      deliver(purpose, id, contactId, address, subject, body, marketing);
+   }
+
+   /*******************************************************************************
+    ** Return the transport result without guessing that queued email was sent.
+    *******************************************************************************/
+   private Delivery deliver(String purpose, String id, Integer contactId, String address, String subject, String body, boolean marketing) throws QException
    {
       if(emailInstalled())
       {
@@ -98,7 +109,9 @@ public class CrmEmailTransport
          {
             throw (new QUserFacingException("The email service could not accept this message."));
          }
-         return;
+         var message = result.message();
+         return (message == null ? new Delivery(null, CrmEmailStatus.QUEUED, null, null, null, null, null)
+            : new Delivery(String.valueOf(message.id()), status(message.status()), message.sentDate(), message.deliveredDate(), null, null, null));
       }
       if(readiness() != CrmReadiness.CORE_MESSAGING)
       {
@@ -107,5 +120,52 @@ public class CrmEmailTransport
       new SendMessageAction().execute(new SendMessageInput().withMessagingProviderName(config.getMessagingProviderName())
          .withTo(new Party().withAddress(address).withRole(EmailPartyRole.TO)).withSubject(subject)
          .withContent(new Content().withBody(body).withContentRole(EmailContentRole.TEXT)));
+      return (new Delivery(null, CrmEmailStatus.SENT, Instant.now(), null, null, null, null));
+   }
+
+   /*******************************************************************************
+    ** CRM-owned delivery snapshot; optional email types never escape this adapter.
+    *******************************************************************************/
+   public record Delivery(String messageId, CrmEmailStatus status, Instant sentDate, Instant deliveredDate,
+                          Instant openedDate, Instant clickedDate, Instant bouncedDate)
+   {
+   }
+
+   /*******************************************************************************
+    ** Send and return the actual outbox state, or the successful core send state.
+    *******************************************************************************/
+   public Delivery sendMessage(String purpose, String id, Integer contactId, String address, String subject, String body, boolean marketing) throws QException
+   {
+      send(purpose, id, contactId, address, subject, body, marketing);
+      return (lookup("crm:" + purpose + ":" + id).orElseGet(() ->
+         new Delivery(null, CrmEmailStatus.SENT, Instant.now(), null, null, null, null)));
+   }
+
+   /*******************************************************************************
+    ** Read delivery through the published API only. RC.3 exposes sent/delivered dates.
+    *******************************************************************************/
+   public Optional<Delivery> lookup(String key) throws QException
+   {
+      if(!emailInstalled() || key == null)
+      {
+         return (Optional.empty());
+      }
+      return (EmailApi.getMessageByIdempotencyKey(key).map(message -> new Delivery(String.valueOf(message.id()),
+         status(message.status()), message.sentDate(), message.deliveredDate(), null, null, null)));
+   }
+
+   /*******************************************************************************
+    ** Preserve known delivery states and leave unknown/pending states eligible for sync.
+    *******************************************************************************/
+   private CrmEmailStatus status(String status)
+   {
+      try
+      {
+         return (CrmEmailStatus.valueOf(status));
+      }
+      catch(IllegalArgumentException | NullPointerException e)
+      {
+         return (CrmEmailStatus.QUEUED);
+      }
    }
 }

src/main/java/com/kingsrook/qbits/crm/email/CrmEmails.java

@@ -0,0 +1,167 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.email;
+
+
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.crm.api.ActivityKind;
+import com.kingsrook.qbits.crm.api.CrmMarketingApi;
+import com.kingsrook.qbits.crm.api.CrmReadiness;
+import com.kingsrook.qbits.crm.core.CrmRecords;
+import com.kingsrook.qbits.crm.core.model.enums.CrmDirection;
+import com.kingsrook.qqq.backend.core.actions.tables.GetAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.exceptions.QUserFacingException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Shared CRM template rendering, transport delivery and timeline recording.
+ *******************************************************************************/
+public final class CrmEmails
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private CrmEmails()
+   {
+   }
+
+   /*******************************************************************************
+    ** A skip reason suitable for standard WARNING summary lines.
+    *******************************************************************************/
+   public static String skipReason(QRecord contact, CrmEmailTransport transport) throws QException
+   {
+      if(Boolean.TRUE.equals(contact.getValueBoolean("doNotEmail")))
+      {
+         return ("Contact has opted out of email");
+      }
+      if(contact.getValueString("email") == null || contact.getValueString("email").isBlank())
+      {
+         return ("Contact has no email address");
+      }
+      return (transport.readiness() == CrmReadiness.NOT_CONFIGURED ? "CRM email is not configured" : null);
+   }
+
+   /*******************************************************************************
+    ** Send once using the caller's stable purpose/id and record the transport outcome.
+    *******************************************************************************/
+   public static void send(CrmEmailTransport transport, QRecord contact, Integer templateId, String subject, String body,
+                           String purpose, String id, Integer enrollmentId) throws QException
+   {
+      String key = "crm:" + purpose + ":" + id;
+      List<QRecord> existing = new QueryAction().execute(new QueryInput("crm_email_message").withFilter(
+         new QQueryFilter(new QFilterCriteria("idempotencyKey", QCriteriaOperator.EQUALS, key)))).getRecords();
+      if(!existing.isEmpty())
+      {
+         recordActivity(existing.getFirst());
+         return;
+      }
+      if(templateId != null)
+      {
+         QRecord template = GetAction.execute("crm_email_template", templateId);
+         if(template == null)
+         {
+            throw (new QUserFacingException("Email template is unavailable"));
+         }
+         subject = subject == null || subject.isBlank() ? template.getValueString("subject") : subject;
+         body = body == null || body.isBlank() ? template.getValueString("bodyText") : body;
+         if(body == null)
+         {
+            body = template.getValueString("bodyHtml");
+         }
+      }
+      if(subject == null || subject.isBlank() || body == null || body.isBlank())
+      {
+         throw (new QUserFacingException("Email subject and body are required"));
+      }
+      subject = render(subject, contact);
+      body = render(body, contact);
+      CrmEmailTransport.Delivery delivery = transport.sendMessage(purpose, id, contact.getValueInteger("id"),
+         contact.getValueString("email"), subject, body, enrollmentId != null);
+      QRecord message = CrmRecords.insert("crm_email_message", new QRecord().withValue("contactId", contact.getValue("id"))
+         .withValue("emailTemplateId", templateId).withValue("sequenceEnrollmentId", enrollmentId)
+         .withValue("toAddresses", contact.getValue("email")).withValue("subject", subject)
+         .withValue("direction", CrmDirection.OUTBOUND.getId()).withValue("status", delivery.status().getId())
+         .withValue("idempotencyKey", key).withValue("emailMessageId", delivery.messageId())
+         .withValue("sentDate", delivery.sentDate()).withValue("deliveredDate", delivery.deliveredDate()));
+      recordActivity(message);
+   }
+
+   /*******************************************************************************
+    ** Repair a partially recorded send without sending again or duplicating its activity.
+    *******************************************************************************/
+   private static void recordActivity(QRecord message) throws QException
+   {
+      if(message.getValue("activityId") != null)
+      {
+         return;
+      }
+      String description = "Email: " + message.getValueString("idempotencyKey");
+      QQueryFilter filter = new QQueryFilter(new QFilterCriteria("contactId", QCriteriaOperator.EQUALS, message.getValue("contactId")),
+         new QFilterCriteria("description", QCriteriaOperator.EQUALS, description));
+      List<QRecord> activities = new QueryAction().execute(new QueryInput("crm_activity").withFilter(filter)).getRecords();
+      if(activities.isEmpty())
+      {
+         new CrmMarketingApi().recordActivity(message.getValueInteger("contactId"),
+            message.getValue("sequenceEnrollmentId") == null ? ActivityKind.EMAIL : ActivityKind.SEQUENCE_STEP,
+            message.getValueString("subject"), description);
+         activities = new QueryAction().execute(new QueryInput("crm_activity").withFilter(filter)).getRecords();
+      }
+      CrmRecords.update("crm_email_message", new QRecord().withValue("id", message.getValue("id"))
+         .withValue("activityId", activities.getFirst().getValue("id")));
+   }
+
+   /*******************************************************************************
+    ** Literal replacements avoid treating customer data as a template program.
+    *******************************************************************************/
+   public static String render(String text, QRecord contact) throws QException
+   {
+      QRecord company = contact.getValue("companyId") == null ? null : GetAction.execute("crm_company", contact.getValue("companyId"));
+      Map<String, String> fields = Map.of("contact.firstName", value(contact, "firstName"), "contact.lastName", value(contact, "lastName"),
+         "company.name", value(company, "name"), "owner.name", value(contact, "ownerUserId"));
+      // Replace each token in the original text only, so inserted values cannot introduce merge fields.
+      java.util.regex.Matcher matcher = java.util.regex.Pattern.compile("\\{\\{(contact.firstName|contact.lastName|company.name|owner.name|unsubscribeUrl)\\}\\}").matcher(text);
+      StringBuilder rendered = new StringBuilder();
+      while(matcher.find())
+      {
+         String replacement = "unsubscribeUrl".equals(matcher.group(1))
+            ? new CrmMarketingApi().buildUnsubscribeUrl(contact.getValueInteger("id"), null) : fields.get(matcher.group(1));
+         matcher.appendReplacement(rendered, java.util.regex.Matcher.quoteReplacement(replacement));
+      }
+      matcher.appendTail(rendered);
+      return (rendered.toString());
+   }
+
+   /*******************************************************************************
+    ** Missing merge values render as empty strings.
+    *******************************************************************************/
+   private static String value(QRecord row, String field)
+   {
+      return (row == null || row.getValueString(field) == null ? "" : row.getValueString(field));
+   }
+}

src/main/java/com/kingsrook/qbits/crm/email/model/EmailMessage.java

@@ -46,6 +46,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.tables.QFieldSection;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.SectionFactory;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.Tier;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.UniqueKey;
 
 
 /*******************************************************************************
@@ -75,6 +76,7 @@ public class EmailMessage extends QRecordEntity
       public QTableMetaData customizeMetaData(QInstance qInstance, QTableMetaData table) throws QException
       {
          table
+            .withUniqueKey(new UniqueKey("idempotencyKey"))
             .withIcon(new QIcon().withName("email"))
             .withRecordLabelFormat("%s")
             .withRecordLabelFields("subject")
@@ -85,10 +87,10 @@ public class EmailMessage extends QRecordEntity
                List.of("bodyHtml", "bodyText")))
             .withSection(new QFieldSection("tracking", new QIcon("visibility"), Tier.T2,
                List.of("isTracked", "openCount", "clickCount", "firstOpenedDate", "firstClickedDate",
-                  "bounceType", "isSpamComplaint")))
+                  "bounceType", "isSpamComplaint", "deliveredDate", "bouncedDate")))
             .withSection(new QFieldSection("references", new QIcon("link"), Tier.T2,
                List.of("contactId", "activityId", "emailTemplateId", "sequenceEnrollmentId",
-                  "threadId", "replyToEmailId")))
+                  "threadId", "replyToEmailId", "idempotencyKey", "emailMessageId")))
             .withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
 
          return (table);
@@ -106,7 +108,7 @@ public class EmailMessage extends QRecordEntity
    @QField(possibleValueSourceName = Contact.TABLE_NAME)
    private Integer contactId;
 
-   @QField(isRequired = true, maxLength = 250, valueTooLongBehavior = ValueTooLongBehavior.ERROR)
+   @QField(maxLength = 250, valueTooLongBehavior = ValueTooLongBehavior.ERROR)
    private String fromAddress;
 
    @QField(maxLength = 100, valueTooLongBehavior = ValueTooLongBehavior.ERROR)
@@ -172,6 +174,18 @@ public class EmailMessage extends QRecordEntity
    @QField()
    private Integer replyToEmailId;
 
+   @QField(maxLength = 250)
+   private String idempotencyKey;
+
+   @QField(maxLength = 100)
+   private String emailMessageId;
+
+   @QField()
+   private Instant deliveredDate;
+
+   @QField()
+   private Instant bouncedDate;
+
    @QField(isEditable = false)
    private Instant createDate;
 
@@ -1031,4 +1045,103 @@ public class EmailMessage extends QRecordEntity
       return (this);
    }
 
+   /*******************************************************************************
+    ** Get idempotencyKey.
+    *******************************************************************************/
+   public String getIdempotencyKey()
+   {
+      return (idempotencyKey);
+   }
+
+   /*******************************************************************************
+    ** Set idempotencyKey.
+    *******************************************************************************/
+   public EmailMessage withIdempotencyKey(String idempotencyKey)
+   {
+      this.idempotencyKey = idempotencyKey;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Get emailMessageId.
+    *******************************************************************************/
+   public String getEmailMessageId()
+   {
+      return (emailMessageId);
+   }
+
+   /*******************************************************************************
+    ** Set emailMessageId.
+    *******************************************************************************/
+   public EmailMessage withEmailMessageId(String emailMessageId)
+   {
+      this.emailMessageId = emailMessageId;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Get deliveredDate.
+    *******************************************************************************/
+   public Instant getDeliveredDate()
+   {
+      return (deliveredDate);
+   }
+
+   /*******************************************************************************
+    ** Set deliveredDate.
+    *******************************************************************************/
+   public EmailMessage withDeliveredDate(Instant deliveredDate)
+   {
+      this.deliveredDate = deliveredDate;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Get bouncedDate.
+    *******************************************************************************/
+   public Instant getBouncedDate()
+   {
+      return (bouncedDate);
+   }
+
+   /*******************************************************************************
+    ** Set bouncedDate.
+    *******************************************************************************/
+   public EmailMessage withBouncedDate(Instant bouncedDate)
+   {
+      this.bouncedDate = bouncedDate;
+      return (this);
+   }
+
+   /*******************************************************************************
+    ** Set idempotencyKey for entity mapping.
+    *******************************************************************************/
+   public void setIdempotencyKey(String idempotencyKey)
+   {
+      this.idempotencyKey = idempotencyKey;
+   }
+
+   /*******************************************************************************
+    ** Set emailMessageId for entity mapping.
+    *******************************************************************************/
+   public void setEmailMessageId(String emailMessageId)
+   {
+      this.emailMessageId = emailMessageId;
+   }
+
+   /*******************************************************************************
+    ** Set deliveredDate for entity mapping.
+    *******************************************************************************/
+   public void setDeliveredDate(Instant deliveredDate)
+   {
+      this.deliveredDate = deliveredDate;
+   }
+
+   /*******************************************************************************
+    ** Set bouncedDate for entity mapping.
+    *******************************************************************************/
+   public void setBouncedDate(Instant bouncedDate)
+   {
+      this.bouncedDate = bouncedDate;
+   }
 }

src/main/java/com/kingsrook/qbits/crm/email/processes/ExpireConsentsProcess.java

@@ -0,0 +1,104 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.email.processes;
+
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import com.kingsrook.qbits.crm.core.CrmRecords;
+import com.kingsrook.qbits.crm.core.model.enums.CrmListMemberStatus;
+import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
+import com.kingsrook.qqq.backend.core.actions.tables.CountAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.count.CountInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerInterface;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+
+
+/*******************************************************************************
+ ** Scheduled CRM maintenance: crm_expire_consents.
+ *******************************************************************************/
+public class ExpireConsentsProcess implements BackendStep, MetaDataProducerInterface<QProcessMetaData>
+{
+   public static final String NAME = "crm_expire_consents";
+
+   @Override
+   public QProcessMetaData produce(QInstance instance)
+   {
+      return (new QProcessMetaData().withName(NAME).withLabel("ExpireConsents").withIsHidden(true)
+         .withStep(new QBackendStepMetaData().withName("execute").withCode(new QCodeReference(ExpireConsentsProcess.class))));
+   }
+
+   @Override
+   public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+   {
+      Instant now = Instant.now();
+      Instant cutoff = now.minus(14, ChronoUnit.DAYS);
+      Integer lastId = null;
+      int expired = 0;
+      while(true)
+      {
+         QQueryFilter filter = new QQueryFilter(new QFilterCriteria("status", QCriteriaOperator.EQUALS, CrmListMemberStatus.PENDING_CONFIRMATION.getId()))
+            .withOrderBy(new QFilterOrderBy("id")).withLimit(200);
+         if(lastId != null)
+         {
+            filter.addCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, lastId));
+         }
+         List<QRecord> page = new QueryAction().execute(new QueryInput("crm_email_list_member").withFilter(filter)).getRecords();
+         for(QRecord member : page)
+         {
+            lastId = member.getValueInteger("id");
+            Instant pendingSince = member.getValueInstant("confirmationSentDate");
+            if(pendingSince == null)
+            {
+               pendingSince = member.getValueInstant("createDate");
+            }
+            if(pendingSince != null && !pendingSince.isAfter(cutoff))
+            {
+               CrmRecords.update("crm_email_list_member", new QRecord().withValue("id", lastId)
+                  .withValue("status", CrmListMemberStatus.UNSUBSCRIBED.getId()).withValue("confirmationTokenHash", null)
+                  .withValue("unsubscribedDate", now));
+               expired++;
+            }
+         }
+         if(page.size() < 200)
+         {
+            break;
+         }
+      }
+      output.addValue("expiredPendingCount", expired);
+      output.addValue("expiredConsentCount", new CountAction().execute(new CountInput("crm_consent_record")
+         .withFilter(new QQueryFilter(new QFilterCriteria("expiryDate", QCriteriaOperator.LESS_THAN_OR_EQUALS, now)))).getCount());
+   }
+}

src/main/java/com/kingsrook/qbits/crm/email/processes/ProcessSequenceStepsProcess.java

@@ -34,9 +34,15 @@ import java.time.temporal.ChronoUnit;
 import java.util.List;
 import com.kingsrook.qbits.crm.activities.model.Activity;
 import com.kingsrook.qbits.crm.activities.model.ActivityType;
+import com.kingsrook.qbits.crm.api.CrmApi;
+import com.kingsrook.qbits.crm.api.CrmMarketingApi;
+import com.kingsrook.qbits.crm.api.CrmReadiness;
+import com.kingsrook.qbits.crm.core.CrmRecords;
 import com.kingsrook.qbits.crm.core.model.Contact;
 import com.kingsrook.qbits.crm.core.model.enums.CrmEnrollmentStatus;
 import com.kingsrook.qbits.crm.core.model.enums.CrmSequenceStepType;
+import com.kingsrook.qbits.crm.email.CrmEmailTransport;
+import com.kingsrook.qbits.crm.email.CrmEmails;
 import com.kingsrook.qbits.crm.email.model.EmailSequence;
 import com.kingsrook.qbits.crm.email.model.SequenceEnrollment;
 import com.kingsrook.qbits.crm.email.model.SequenceStep;
@@ -45,6 +51,8 @@ import com.kingsrook.qqq.backend.core.actions.tables.GetAction;
 import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
 import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
 import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.alerts.AlertSignal;
+import com.kingsrook.qqq.backend.core.alerts.QAlerts;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
@@ -79,6 +87,15 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
 
    private static final int MAX_FAILURES = 3;
 
+   /*******************************************************************************
+    ** Resolve the host transport; tests may supply a fake.
+    *******************************************************************************/
+   protected CrmEmailTransport emailTransport() throws QException
+   {
+      return (new CrmEmailTransport(CrmApi.config()));
+   }
+
+
 
 
    /*******************************************************************************
@@ -92,7 +109,6 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
          .withIsHidden(true)
          .withLabel("Process Sequence Steps")
          .withIcon(new QIcon().withName("play_circle"))
-         // Scheduling is configured by the host application via ScheduledJob records
          .withStepList(List.of(
             new QBackendStepMetaData()
                .withName("execute")
@@ -111,6 +127,18 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
       ///////////////////////////////////////////////////////////
       // query enrollments where status=ACTIVE and due now     //
       ///////////////////////////////////////////////////////////
+      CrmEmailTransport transport = emailTransport();
+      boolean ready = transport.readiness() != CrmReadiness.NOT_CONFIGURED;
+      boolean activeSequences = !new QueryAction().execute(new QueryInput(EmailSequence.TABLE_NAME)
+         .withFilter(new QQueryFilter(new QFilterCriteria("isActive", QCriteriaOperator.EQUALS, true)).withLimit(1))).getRecords().isEmpty();
+      if(!ready && activeSequences)
+      {
+         QAlerts.raise(new AlertSignal("crm.email.unavailable").withDedupKey("transport").withTitle("CRM email is unavailable"));
+      }
+      else if(ready)
+      {
+         QAlerts.resolve("crm.email.unavailable", "transport", "Email transport is ready");
+      }
       int processedCount = 0;
       Integer lastId = null;
       Instant dueThrough = Instant.now();
@@ -130,7 +158,7 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
             SequenceEnrollment enrollment = new SequenceEnrollment(enrollmentRecord);
             try
             {
-               processEnrollment(enrollment);
+               processEnrollment(enrollment, transport);
                processedCount++;
             }
             catch(Exception e)
@@ -154,7 +182,7 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
     ** Process a single enrollment: load the next step and execute it.
     ** Respects businessDaysOnly and sendWindow settings on the sequence.
     ***************************************************************************/
-   private void processEnrollment(SequenceEnrollment enrollment) throws QException
+   private void processEnrollment(SequenceEnrollment enrollment, CrmEmailTransport transport) throws QException
    {
       Integer nextStepNumber = enrollment.getCurrentStepNumber() + 1;
 
@@ -167,6 +195,11 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
       EmailSequence sequence = sequenceGet.getRecord() != null
          ? new EmailSequence(sequenceGet.getRecord()) : null;
 
+      if(sequence == null || !Boolean.TRUE.equals(sequence.getIsActive()))
+      {
+         return;
+      }
+
       /////////////////////////////////////////////
       // load the contact for timezone           //
       /////////////////////////////////////////////
@@ -237,7 +270,27 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
       {
          switch(stepType)
          {
-            case EMAIL -> executeEmailStep(enrollment, step);
+            case EMAIL ->
+            {
+               var eligibility = new CrmMarketingApi().checkEmailEligibility(enrollment.getContactId());
+               if(!eligibility.eligible())
+               {
+                  CrmRecords.update(SequenceEnrollment.TABLE_NAME, new QRecord().withValue("id", enrollment.getId())
+                     .withValue("status", CrmEnrollmentStatus.PAUSED.getId()).withValue("lastFailureMessage", eligibility.reason())
+                     .withValue("nextStepDate", null));
+                  return;
+               }
+               if(transport.readiness() == CrmReadiness.NOT_CONFIGURED)
+               {
+                  return;
+               }
+               if(step.getEmailTemplateId() == null)
+               {
+                  throw (new QException("Sequence email template is required"));
+               }
+               CrmEmails.send(transport, contactGet.getRecord(), step.getEmailTemplateId(), null, null,
+                  "seq", enrollment.getId() + ":" + step.getId(), enrollment.getId());
+            }
             case CALL -> executeCallStep(enrollment, step);
             case TASK -> executeTaskStep(enrollment, step);
             case WAIT ->
@@ -257,50 +310,6 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
 
 
 
-   /***************************************************************************
-    ** Execute an EMAIL step: verify doNotEmail, create an activity record.
-    ***************************************************************************/
-   private void executeEmailStep(SequenceEnrollment enrollment, SequenceStep step) throws QException
-   {
-      ///////////////////////////////////////////////
-      // check doNotEmail                          //
-      ///////////////////////////////////////////////
-      GetOutput contactGet = new GetAction().execute(
-         new GetInput(Contact.TABLE_NAME).withPrimaryKey(enrollment.getContactId()));
-
-      if(contactGet.getRecord() != null)
-      {
-         Contact contact = new Contact(contactGet.getRecord());
-         if(Boolean.TRUE.equals(contact.getDoNotEmail()))
-         {
-            throw (new QException("Contact has opted out of email"));
-         }
-      }
-
-      ///////////////////////////////////////////////
-      // create activity for the email             //
-      ///////////////////////////////////////////////
-      String subject = "Sequence email: step " + step.getStepNumber();
-      if(step.getEmailTemplateId() != null)
-      {
-         subject += " (template #" + step.getEmailTemplateId() + ")";
-      }
-
-      Integer activityTypeId = lookupActivityTypeByName("Email");
-
-      new InsertAction().execute(
-         new InsertInput(Activity.TABLE_NAME).withRecordEntity(
-            new Activity()
-               .withActivityTypeId(activityTypeId)
-               .withContactId(enrollment.getContactId())
-               .withDealId(enrollment.getDealId())
-               .withSubject(subject)
-               .withOwnerUserId("system")
-               .withIsCompleted(true)));
-   }
-
-
-
    /***************************************************************************
     ** Execute a CALL step: create a task activity assigned to the contact owner.
     ***************************************************************************/
@@ -524,6 +533,8 @@ public class ProcessSequenceStepsProcess implements BackendStep, MetaDataProduce
       {
          updateRecord.withValue("status", CrmEnrollmentStatus.FAILED.getPossibleValueId());
          updateRecord.withValue("nextStepDate", null);
+         QAlerts.raise(new AlertSignal("crm.sequence.failing").withDedupKey("enrollment:" + enrollment.getId())
+            .withTitle("Sequence enrollment failed three times"));
       }
 
       new UpdateAction().execute(

src/main/java/com/kingsrook/qbits/crm/email/processes/SyncEmailDeliveryProcess.java

@@ -0,0 +1,153 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.email.processes;
+
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import com.kingsrook.qbits.crm.api.CrmApi;
+import com.kingsrook.qbits.crm.core.CrmRecords;
+import com.kingsrook.qbits.crm.core.model.enums.CrmConsentStatus;
+import com.kingsrook.qbits.crm.core.model.enums.CrmConsentType;
+import com.kingsrook.qbits.crm.core.model.enums.CrmEmailStatus;
+import com.kingsrook.qbits.crm.email.CrmEmailTransport;
+import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerInterface;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+
+
+/*******************************************************************************
+ ** Scheduled CRM maintenance: crm_sync_email_delivery.
+ *******************************************************************************/
+public class SyncEmailDeliveryProcess implements BackendStep, MetaDataProducerInterface<QProcessMetaData>
+{
+   public static final String NAME = "crm_sync_email_delivery";
+
+   @Override
+   public QProcessMetaData produce(QInstance instance)
+   {
+      return (new QProcessMetaData().withName(NAME).withLabel("SyncEmailDelivery").withIsHidden(true)
+         .withStep(new QBackendStepMetaData().withName("execute").withCode(new QCodeReference(SyncEmailDeliveryProcess.class))));
+   }
+
+   /*******************************************************************************
+    ** Resolve the host transport; an absent qbit-email yields empty lookups.
+    *******************************************************************************/
+   protected CrmEmailTransport emailTransport() throws QException
+   {
+      return (new CrmEmailTransport(CrmApi.config()));
+   }
+
+   @Override
+   public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+   {
+      CrmEmailTransport transport = emailTransport();
+      Integer lastId = null;
+      int updated = 0;
+      Instant since = Instant.now().minus(30, ChronoUnit.DAYS);
+      while(true)
+      {
+         QQueryFilter filter = new QQueryFilter(new QFilterCriteria("createDate", QCriteriaOperator.GREATER_THAN_OR_EQUALS, since))
+            .withOrderBy(new QFilterOrderBy("id")).withLimit(200);
+         if(lastId != null)
+         {
+            filter.addCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, lastId));
+         }
+         List<QRecord> page = new QueryAction().execute(new QueryInput("crm_email_message").withFilter(filter)).getRecords();
+         for(QRecord message : page)
+         {
+            lastId = message.getValueInteger("id");
+            if(message.getValueInteger("status") != null && List.of(CrmEmailStatus.BOUNCED.getId(), CrmEmailStatus.COMPLAINED.getId(), CrmEmailStatus.FAILED.getId(),
+               CrmEmailStatus.SUPPRESSED.getId()).contains(message.getValueInteger("status")))
+            {
+               continue;
+            }
+            var delivery = transport.lookup(message.getValueString("idempotencyKey"));
+            if(delivery.isEmpty())
+            {
+               continue;
+            }
+            CrmEmailTransport.Delivery state = delivery.get();
+            QRecord change = new QRecord().withValue("id", lastId).withValue("status", state.status().getId());
+            date(change, "sentDate", state.sentDate());
+            date(change, "deliveredDate", state.deliveredDate());
+            date(change, "firstOpenedDate", state.openedDate());
+            date(change, "firstClickedDate", state.clickedDate());
+            date(change, "bouncedDate", state.bouncedDate());
+            boolean complaint = state.status() == CrmEmailStatus.COMPLAINED;
+            if(complaint || state.status() == CrmEmailStatus.BOUNCED)
+            {
+               Integer contactId = message.getValueInteger("contactId");
+               if(contactId != null)
+               {
+                  CrmRecords.update("crm_contact", new QRecord().withValue("id", contactId).withValue("doNotEmail", true));
+                  String source = complaint ? "Complaint" : "Hard bounce";
+                  // The terminal message checkpoint is written last. A retry must not duplicate consent evidence.
+                  String evidence = "Email: " + message.getValueString("idempotencyKey");
+                  List<QRecord> recorded = new QueryAction().execute(new QueryInput("crm_consent_record").withFilter(new QQueryFilter(
+                     new QFilterCriteria("contactId", QCriteriaOperator.EQUALS, contactId),
+                     new QFilterCriteria("source", QCriteriaOperator.EQUALS, source),
+                     new QFilterCriteria("consentText", QCriteriaOperator.EQUALS, evidence)))).getRecords();
+                  if(recorded.isEmpty())
+                  {
+                     CrmRecords.insert("crm_consent_record", new QRecord().withValue("contactId", contactId)
+                        .withValue("consentType", CrmConsentType.EMAIL.getId()).withValue("status", CrmConsentStatus.WITHDRAWN.getId())
+                        .withValue("consentDate", Instant.now()).withValue("source", source).withValue("consentText", evidence));
+                  }
+               }
+               change.withValue("isSpamComplaint", complaint);
+            }
+            CrmRecords.update("crm_email_message", change);
+            updated++;
+         }
+         if(page.size() < 200)
+         {
+            break;
+         }
+      }
+      output.addValue("updatedCount", updated);
+   }
+
+   /*******************************************************************************
+    ** Never clear a timestamp when the provider has not exposed it.
+    *******************************************************************************/
+   private void date(QRecord row, String field, Instant value)
+   {
+      if(value != null)
+      {
+         row.setValue(field, value);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/crm/scoring/AutoAssignCustomizer.java

@@ -0,0 +1,65 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.scoring;
+
+
+import java.util.List;
+import com.kingsrook.qbits.crm.core.model.enums.CrmAssignableEntity;
+import com.kingsrook.qqq.backend.core.actions.customizers.AbstractPostInsertCustomizer;
+import com.kingsrook.qqq.backend.core.alerts.NotificationAudience;
+import com.kingsrook.qqq.backend.core.alerts.NotificationSignal;
+import com.kingsrook.qqq.backend.core.alerts.QAlerts;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Apply assignment rules to new contacts and deals without replacing explicit owners.
+ *******************************************************************************/
+public class AutoAssignCustomizer extends AbstractPostInsertCustomizer
+{
+   @Override
+   public List<QRecord> apply(List<QRecord> records) throws QException
+   {
+      for(QRecord row : records)
+      {
+         if(row.getErrors() != null && !row.getErrors().isEmpty())
+         {
+            continue;
+         }
+         String owner = row.getValueString("ownerUserId");
+         if(owner == null || "system".equals(owner))
+         {
+            CrmAutoAssign.Result result = CrmAutoAssign.assign("crm_contact".equals(insertInput.getTableName())
+               ? CrmAssignableEntity.CONTACT : CrmAssignableEntity.DEAL, row.getValueInteger("id"));
+            if("ASSIGNED".equals(result.action()))
+            {
+               row.setValue("ownerUserId", result.assignedUserId());
+               QAlerts.notify(new NotificationSignal("CRM lead assigned").withSourceName("crm.lead.assigned")
+                  .withSourceKey(insertInput.getTableName() + ":" + row.getValue("id"))
+                  .withSourceTableName(insertInput.getTableName()).withSourceRecordId(row.getValueString("id"))
+                  .withAudience(NotificationAudience.user(result.assignedUserId())), insertInput.getTransaction());
+            }
+         }
+      }
+      return (records);
+   }
+}

src/main/java/com/kingsrook/qbits/crm/scoring/JobState.java

@@ -0,0 +1,49 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm.scoring;
+
+
+import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerInterface;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.UniqueKey;
+
+
+/*******************************************************************************
+ ** Small persistent maintenance checkpoints, including the last full score pass.
+ *******************************************************************************/
+public class JobState implements MetaDataProducerInterface<QTableMetaData>
+{
+   public static final String TABLE_NAME = "crm_job_state";
+
+   @Override
+   public QTableMetaData produce(QInstance instance)
+   {
+      return (new QTableMetaData().withName(TABLE_NAME).withLabel("Job State").withPrimaryKeyField("id")
+         .withField(new QFieldMetaData("id", QFieldType.INTEGER).withIsEditable(false))
+         .withField(new QFieldMetaData("name", QFieldType.STRING).withIsRequired(true).withMaxLength(250))
+         .withField(new QFieldMetaData("lastSuccess", QFieldType.DATE_TIME))
+         .withField(new QFieldMetaData("lastFullPass", QFieldType.DATE_TIME))
+         .withUniqueKey(new UniqueKey("name")));
+   }
+}

src/main/java/com/kingsrook/qbits/crm/scoring/processes/CalculateLeadScoresProcess.java

@@ -26,22 +26,29 @@
 package com.kingsrook.qbits.crm.scoring.processes;
 
 
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
+import com.kingsrook.qbits.crm.core.CrmRecords;
 import com.kingsrook.qbits.crm.core.model.Company;
 import com.kingsrook.qbits.crm.core.model.Contact;
 import com.kingsrook.qbits.crm.core.model.enums.CrmScoreOperator;
+import com.kingsrook.qbits.crm.deals.CrmDealNotifications;
+import com.kingsrook.qbits.crm.scoring.JobState;
 import com.kingsrook.qbits.crm.scoring.model.LeadScoreRule;
 import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
-import com.kingsrook.qqq.backend.core.actions.tables.GetAction;
 import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
 import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
-import com.kingsrook.qqq.backend.core.model.actions.tables.get.GetInput;
-import com.kingsrook.qqq.backend.core.model.actions.tables.get.GetOutput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
@@ -83,7 +90,6 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
          .withIsHidden(true)
          .withLabel("Calculate Lead Scores")
          .withIcon(new QIcon().withName("score"))
-         // Scheduling is configured by the host application via ScheduledJob records
          .withStepList(List.of(
             new QBackendStepMetaData()
                .withName("execute")
@@ -99,6 +105,14 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
    @Override
    public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
    {
+      Instant startedAt = Instant.now();
+      List<QRecord> states = new QueryAction().execute(new QueryInput(JobState.TABLE_NAME)
+         .withFilter(new QQueryFilter(new QFilterCriteria("name", QCriteriaOperator.EQUALS, NAME)))).getRecords();
+      QRecord state = states.isEmpty() ? new QRecord().withValue("name", NAME) : states.getFirst();
+      Instant lastSuccess = state.getValueInstant("lastSuccess");
+      Instant lastFullPass = state.getValueInstant("lastFullPass");
+      boolean full = Boolean.TRUE.equals(input.getValueBoolean("fullRecalc")) || lastSuccess == null || lastFullPass == null
+         || !lastFullPass.isAfter(startedAt.minus(24, ChronoUnit.HOURS));
       ///////////////////////////////////////////////
       // load all active rules ordered by sort     //
       ///////////////////////////////////////////////
@@ -131,6 +145,10 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
             filter.withCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, lastId));
          }
 
+         if(!full)
+         {
+            filter.withCriteria(new QFilterCriteria("modifyDate", QCriteriaOperator.GREATER_THAN_OR_EQUALS, lastSuccess));
+         }
          QueryOutput contactsQuery = new QueryAction().execute(
             new QueryInput(Contact.TABLE_NAME).withFilter(filter));
 
@@ -139,15 +157,20 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
             break;
          }
 
+         Map<Integer, QRecord> companies = loadCompanies(contactsQuery.getRecords());
+         List<QRecord> changes = new ArrayList<>();
          for(QRecord contactRecord : contactsQuery.getRecords())
          {
             totalContacts++;
-            if(recalculate(contactRecord, rules))
+            QRecord change = calculateChange(contactRecord, rules, companies);
+            if(change != null)
             {
-               updatedCount++;
+               changes.add(change);
             }
             lastId = contactRecord.getValueInteger("id");
          }
+         writeScores(changes);
+         updatedCount += changes.size();
 
          /////////////////////////////////////////////
          // if page was not full, we are done       //
@@ -158,6 +181,21 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
          }
       }
 
+      CrmDealNotifications.checkRotting();
+      state.withValue("lastSuccess", startedAt);
+      if(full)
+      {
+         state.withValue("lastFullPass", startedAt);
+      }
+      if(states.isEmpty())
+      {
+         CrmRecords.insert(JobState.TABLE_NAME, state);
+      }
+      else
+      {
+         CrmRecords.update(JobState.TABLE_NAME, state);
+      }
+      output.addValue("fullRecalc", full);
       output.addValue("updatedCount", updatedCount);
       output.addValue("totalContacts", totalContacts);
       output.addValue("totalRules", rules.size());
@@ -173,18 +211,19 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
       List<LeadScoreRule> rules = new QueryAction().execute(new QueryInput(LeadScoreRule.TABLE_NAME)
          .withFilter(new QQueryFilter(new QFilterCriteria("isActive", QCriteriaOperator.EQUALS, true))
             .withOrderBy(new QFilterOrderBy("sortOrder")))).getRecords().stream().map(LeadScoreRule::new).toList();
-      recalculate(contact, rules);
+      QRecord change = calculateChange(contact, rules, loadCompanies(List.of(contact)));
+      writeScores(change == null ? List.of() : List.of(change));
    }
 
    /*******************************************************************************
-    ** Write only changed scores; return whether a write was needed.
+    ** Compute a score update only when the result changed.
     *******************************************************************************/
-   private boolean recalculate(QRecord contact, List<LeadScoreRule> rules) throws QException
+   private QRecord calculateChange(QRecord contact, List<LeadScoreRule> rules, Map<Integer, QRecord> companies) throws QException
    {
       int score = 0;
       for(LeadScoreRule rule : rules)
       {
-         if(evaluateRule(rule, contact))
+         if(evaluateRule(rule, contact, companies))
          {
             score += rule.getScoreAdjustment();
          }
@@ -192,11 +231,27 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
       Integer current = contact.getValueInteger("leadScore");
       if(Objects.equals(current == null ? 0 : current, score))
       {
-         return (false);
+         return (null);
+      }
+      return (new QRecord().withValue("id", contact.getValue("id")).withValue("leadScore", score));
+   }
+
+   /*******************************************************************************
+    ** Batch score writes so core audit label resolution also operates per page.
+    *******************************************************************************/
+   private void writeScores(List<QRecord> changes) throws QException
+   {
+      if(changes.isEmpty())
+      {
+         return;
+      }
+      UpdateInput update = new UpdateInput(Contact.TABLE_NAME).withOmitModifyDateUpdate(true);
+      update.setRecords(changes);
+      var updated = new UpdateAction().execute(update).getRecords();
+      if(updated.isEmpty() || updated.stream().anyMatch(row -> row.getErrors() != null && !row.getErrors().isEmpty()))
+      {
+         throw (new QException("Lead score update was rejected"));
       }
-      new UpdateAction().execute(new UpdateInput(Contact.TABLE_NAME)
-         .withRecord(new QRecord().withValue("id", contact.getValue("id")).withValue("leadScore", score)));
-      return (true);
    }
 
 
@@ -204,7 +259,7 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
     ** Evaluate a single rule against a contact record. Supports dot-notation
     ** field paths for related entities (e.g., "company.industry").
     ***************************************************************************/
-   private boolean evaluateRule(LeadScoreRule rule, QRecord contactRecord)
+   private boolean evaluateRule(LeadScoreRule rule, QRecord contactRecord, Map<Integer, QRecord> companies)
    {
       CrmScoreOperator operator = CrmScoreOperator.getById(rule.getOperator());
       if(operator == null)
@@ -213,7 +268,7 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
       }
 
       String fieldPath = rule.getFieldPath();
-      String actualValue = resolveFieldValue(fieldPath, contactRecord);
+      String actualValue = resolveFieldValue(fieldPath, contactRecord, companies);
       String expectedValue = rule.getFieldValue();
 
       switch(operator)
@@ -233,6 +288,10 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
          case LT:
             return (compareNumeric(actualValue, expectedValue) < 0);
 
+         case DAYS_AGO_GT:
+         case DAYS_AGO_LT:
+            return (compareAge(actualValue, expectedValue, operator));
+
          case IS_SET:
             return (StringUtils.hasContent(actualValue));
 
@@ -272,48 +331,54 @@ public class CalculateLeadScoresProcess implements BackendStep, MetaDataProducer
     ** for related entity fields (e.g., "company.industry" loads the contact's
     ** company record and returns its "industry" field).
     ***************************************************************************/
-   private String resolveFieldValue(String fieldPath, QRecord contactRecord)
+   private String resolveFieldValue(String fieldPath, QRecord contactRecord, Map<Integer, QRecord> companies)
    {
       if(fieldPath == null || !fieldPath.contains("."))
       {
          return (contactRecord.getValueString(fieldPath));
       }
-
       String[] parts = fieldPath.split("\\.", 2);
-      String prefix    = parts[0];
-      String fieldName = parts[1];
-
-      try
+      if("company".equalsIgnoreCase(parts[0]))
       {
-         if("company".equalsIgnoreCase(prefix))
-         {
-            Integer companyId = contactRecord.getValueInteger("companyId");
-            if(companyId == null)
-            {
-               return (null);
-            }
-
-            GetOutput companyOutput = new GetAction().execute(
-               new GetInput(Company.TABLE_NAME).withPrimaryKey(companyId));
-
-            if(companyOutput.getRecord() == null)
-            {
-               return (null);
-            }
-
-            return (companyOutput.getRecord().getValueString(fieldName));
-         }
-         else
-         {
-            LOG.warn("Unsupported dot-notation prefix in lead score rule fieldPath: " + prefix);
-            return (null);
-         }
-      }
-      catch(Exception e)
-      {
-         LOG.warn("Error resolving dot-notation field [" + fieldPath + "] for contact", e);
-         return (null);
+         QRecord company = companies.get(contactRecord.getValueInteger("companyId"));
+         return (company == null ? null : company.getValueString(parts[1]));
       }
+      LOG.warn("Unsupported dot-notation prefix in lead score rule fieldPath: " + parts[0]);
+      return (null);
    }
 
+   /*******************************************************************************
+    ** One company query per page, shared by every rule and contact in that page.
+    *******************************************************************************/
+   private Map<Integer, QRecord> loadCompanies(List<QRecord> contacts) throws QException
+   {
+      List<Integer> ids = contacts.stream().map(row -> row.getValueInteger("companyId")).filter(Objects::nonNull).distinct().toList();
+      Map<Integer, QRecord> companies = new HashMap<>();
+      if(!ids.isEmpty())
+      {
+         for(QRecord company : new QueryAction().execute(new QueryInput(Company.TABLE_NAME)
+            .withFilter(new QQueryFilter(new QFilterCriteria("id", QCriteriaOperator.IN, ids)))).getRecords())
+         {
+            companies.put(company.getValueInteger("id"), company);
+         }
+      }
+      return (companies);
+   }
+
+   /*******************************************************************************
+    ** Compare instants or UTC calendar dates with the strict N-days-ago boundary.
+    *******************************************************************************/
+   private boolean compareAge(String actual, String expected, CrmScoreOperator operator)
+   {
+      try
+      {
+         Instant date = actual.length() == 10 ? LocalDate.parse(actual).atStartOfDay(ZoneOffset.UTC).toInstant() : Instant.parse(actual);
+         Instant threshold = Instant.now().minus(Long.parseLong(expected), ChronoUnit.DAYS);
+         return (operator == CrmScoreOperator.DAYS_AGO_GT ? date.isBefore(threshold) : date.isAfter(threshold));
+      }
+      catch(RuntimeException e)
+      {
+         return (false);
+      }
+   }
 }

src/main/java/com/kingsrook/qbits/crm/security/CrmPermissions.java

@@ -49,7 +49,7 @@ public final class CrmPermissions
    public static final List<String> ALL = List.of(APP_ACCESS, READ, WRITE, MANAGE, MANAGE_READ, MANAGE_WRITE, ADMIN, ADMIN_READ, ADMIN_WRITE);
 
    private static final Set<String> SETTINGS_TABLES = Set.of(
-      "crm_pipeline", "crm_product", "crm_email_template", "crm_email_sequence",
+      "crm_job_state", "crm_pipeline", "crm_product", "crm_email_template", "crm_email_sequence",
       "crm_lead_score_rule", "crm_assignment_rule", "crm_sales_goal", "crm_lifecycle_stage",
       "crm_lead_source", "crm_industry", "crm_activity_type", "crm_activity_outcome",
       "crm_contact_role", "crm_win_loss_reason");
@@ -58,7 +58,7 @@ public final class CrmPermissions
       "crm_merge_contacts", "crm_merge_companies", "crm_bulk_assign_owner",
       "crm_subscribe_to_list", "crm_unsubscribe_from_list", "crm_enroll_in_sequence",
       "crm_unenroll_from_sequence", "crm_process_sequence_steps", "crm_assign_company_owner",
-      "crm_transfer_ownership", "crm_activate_sequence", "crm_pause_sequence");
+      "crm_transfer_ownership", "crm_activate_sequence", "crm_pause_sequence", "crm_sync_email_delivery", "crm_expire_consents", "crm_calculate_lead_scores");
 
 
 

src/main/resources/db/liquibase/qbit-crm/crm.changelog.sql

@@ -1,7 +1,7 @@
 --liquibase formatted sql
 -- Copyright 2026 Kingsrook, LLC. Licensed under the Apache License, Version 2.0.
 -- https://www.apache.org/licenses/LICENSE-2.0
--- QBit CRM 0.2.0 P1 PostgreSQL reference schema (33 tables).
+-- QBit CRM 0.2.0 P1 PostgreSQL reference schema (34 tables).
 -- QQQ table names equal SQL names; Java field names map to snake_case columns.
 -- Unique constraints match QQQ metadata. Relationship columns are indexed;
 -- foreign-key constraints are deliberately not imposed (the QQQ/host layer owns
@@ -199,7 +199,7 @@ CREATE TABLE IF NOT EXISTS crm_contact (
    company_id INTEGER,
    lifecycle_stage_id INTEGER,
    lead_source_id INTEGER,
-   owner_user_id TEXT NOT NULL,
+   owner_user_id TEXT,
    address_line1 VARCHAR(250),
    address_line2 VARCHAR(250),
    city VARCHAR(100),
@@ -265,7 +265,7 @@ CREATE TABLE IF NOT EXISTS crm_deal (
    weighted_amount DECIMAL(20,6),
    company_id INTEGER,
    primary_contact_id INTEGER,
-   owner_user_id TEXT NOT NULL,
+   owner_user_id TEXT,
    lead_source_id INTEGER,
    expected_close_date DATE,
    actual_close_date DATE,
@@ -383,9 +383,13 @@ CREATE INDEX IF NOT EXISTS ix_crm_email_list_member_form_submission_id ON crm_em
 --changeset qbit-crm:020-crm_email_message
 CREATE TABLE IF NOT EXISTS crm_email_message (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   idempotency_key VARCHAR(250) UNIQUE,
+   email_message_id VARCHAR(100),
+   delivered_date TIMESTAMP WITHOUT TIME ZONE,
+   bounced_date TIMESTAMP WITHOUT TIME ZONE,
    activity_id INTEGER,
    contact_id INTEGER,
-   from_address VARCHAR(250) NOT NULL,
+   from_address VARCHAR(250),
    from_name VARCHAR(100),
    to_addresses TEXT NOT NULL,
    cc_addresses TEXT,
@@ -411,6 +415,7 @@ CREATE TABLE IF NOT EXISTS crm_email_message (
    modify_date TIMESTAMP WITHOUT TIME ZONE
 );
 
+CREATE INDEX IF NOT EXISTS ix_crm_email_message_email_message_id ON crm_email_message (email_message_id);
 CREATE INDEX IF NOT EXISTS ix_crm_email_message_contact_id ON crm_email_message (contact_id);
 CREATE INDEX IF NOT EXISTS ix_crm_email_message_activity_id ON crm_email_message (activity_id);
 CREATE INDEX IF NOT EXISTS ix_crm_email_message_thread_id ON crm_email_message (thread_id);
@@ -674,3 +679,11 @@ CREATE TABLE IF NOT EXISTS crm_win_loss_reason (
 );
 
 CREATE INDEX IF NOT EXISTS ix_crm_consent_record_evidence_form_submission_id ON crm_consent_record (evidence_form_submission_id);
+
+--changeset qbit-crm:crm_job_state
+CREATE TABLE IF NOT EXISTS crm_job_state (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   name VARCHAR(250) NOT NULL UNIQUE,
+   last_success TIMESTAMP WITHOUT TIME ZONE,
+   last_full_pass TIMESTAMP WITHOUT TIME ZONE
+);

src/test/java/com/kingsrook/qbits/crm/CrmAppMetaDataTest.java

@@ -90,7 +90,7 @@ class CrmAppMetaDataTest extends BaseTest
          new CrmQBitProducer().withQBitConfig(new CrmQBitConfig().withDefaultBackendNameForTables(BACKEND_NAME).withAppEnabled(enabled))
             .produce(instance).addSelfToInstance(instance);
          assertThat(instance.getApps()).hasSize(Boolean.FALSE.equals(enabled) ? 0 : 1);
-         assertThat(instance.getTables()).hasSize(33);
+         assertThat(instance.getTables()).hasSize(34);
       }
    }
 

src/test/java/com/kingsrook/qbits/crm/CrmQBitProducerTest.java

@@ -142,7 +142,7 @@ class CrmQBitProducerTest extends BaseTest
       {
          assertThat(QContext.getQInstance().getTable(tableName).getAuditRules().getAuditLevel()).isEqualTo(AuditLevel.FIELD);
       }
-      assertThat(QContext.getQInstance().getTables()).hasSize(33)
+      assertThat(QContext.getQInstance().getTables()).hasSize(34)
          .doesNotContainKeys("crm_tag", "crm_contact_tag", "crm_company_tag", "crm_deal_tag", "crm_attachment", "crm_audit_log");
       assertThat(QContext.getQInstance().getProcesses()).doesNotContainKeys("crm_bulk_add_tag", "crm_bulk_remove_tag");
    }

src/test/java/com/kingsrook/qbits/crm/CrmRdbmsTest.java

@@ -226,7 +226,7 @@ class CrmRdbmsTest
    @Test
    void testSchemaMatchesMetadata() throws Exception
    {
-      assertThat(qInstance.getTables()).hasSize(33);
+      assertThat(qInstance.getTables()).hasSize(34);
       for(QTableMetaData table : qInstance.getTables().values())
       {
          Set<String> columns = new HashSet<>();
@@ -278,7 +278,7 @@ class CrmRdbmsTest
       String ddl = Files.readString(Path.of("docs/ddl/postgres.sql"));
       String changelog = Files.readString(Path.of("src/main/resources/db/liquibase/qbit-crm/crm.changelog.sql"));
       assertThat(changelog).startsWith("--liquibase formatted sql");
-      assertThat(changelog.lines().filter(line -> line.startsWith("--changeset qbit-crm:")).count()).isEqualTo(33);
+      assertThat(changelog.lines().filter(line -> line.startsWith("--changeset qbit-crm:")).count()).isEqualTo(34);
       assertThat(stripComments(changelog)).isEqualTo(stripComments(ddl));
    }
 

src/test/java/com/kingsrook/qbits/crm/PlatformReuseTest.java

@@ -0,0 +1,790 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.crm;
+
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import com.kingsrook.qbits.crm.actions.CrmAction;
+import com.kingsrook.qbits.crm.actions.CrmActions;
+import com.kingsrook.qbits.crm.api.CrmApi;
+import com.kingsrook.qbits.crm.api.CrmReadiness;
+import com.kingsrook.qbits.crm.core.CrmRecords;
+import com.kingsrook.qbits.crm.core.model.enums.CrmAssignStrategy;
+import com.kingsrook.qbits.crm.core.model.enums.CrmAssignableEntity;
+import com.kingsrook.qbits.crm.core.model.enums.CrmConsentStatus;
+import com.kingsrook.qbits.crm.core.model.enums.CrmConsentType;
+import com.kingsrook.qbits.crm.core.model.enums.CrmEmailStatus;
+import com.kingsrook.qbits.crm.core.model.enums.CrmEnrollmentStatus;
+import com.kingsrook.qbits.crm.core.model.enums.CrmListMemberStatus;
+import com.kingsrook.qbits.crm.core.model.enums.CrmScoreOperator;
+import com.kingsrook.qbits.crm.core.model.enums.CrmSequenceStepType;
+import com.kingsrook.qbits.crm.deals.CrmDealNotifications;
+import com.kingsrook.qbits.crm.email.CrmEmailProcessIdentity;
+import com.kingsrook.qbits.crm.email.CrmEmailTransport;
+import com.kingsrook.qbits.crm.email.CrmEmails;
+import com.kingsrook.qbits.crm.email.processes.ExpireConsentsProcess;
+import com.kingsrook.qbits.crm.email.processes.ProcessSequenceStepsProcess;
+import com.kingsrook.qbits.crm.email.processes.SyncEmailDeliveryProcess;
+import com.kingsrook.qbits.crm.scoring.processes.CalculateLeadScoresProcess;
+import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
+import com.kingsrook.qqq.backend.core.actions.tables.GetAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.alerts.AlertResolution;
+import com.kingsrook.qqq.backend.core.alerts.AlertSignal;
+import com.kingsrook.qqq.backend.core.alerts.AlertsInstanceMetaData;
+import com.kingsrook.qqq.backend.core.alerts.LoggingAlertSink;
+import com.kingsrook.qqq.backend.core.alerts.NotificationAudience;
+import com.kingsrook.qqq.backend.core.alerts.NotificationSignal;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.collaboration.RecordFeature;
+import com.kingsrook.qqq.backend.core.model.metadata.collaboration.RecordFeaturesMetaData;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+
+/*******************************************************************************
+ ** P4 contracts exercised through memory DML, fake transport and real QAlerts dispatch.
+ *******************************************************************************/
+class PlatformReuseTest extends BaseTest
+{
+   private FakeTransport transport;
+
+   /*******************************************************************************
+    ** Keep effects local to this test.
+    *******************************************************************************/
+   @BeforeEach
+   void configure() throws Exception
+   {
+      transport = new FakeTransport();
+      CrmApi.config().withPublicBaseUrl("https://crm.example.test").withTokenSecret("test-only-signing-key");
+      Sink.notifications.clear();
+      Sink.alerts.clear();
+      Sink.resolutions.clear();
+      AlertsInstanceMetaData.ofOrWithNew(QContext.getQInstance()).withSink(new QCodeReference(Sink.class));
+   }
+
+   /*******************************************************************************
+    ** Send action renders and records transactional delivery.
+    *******************************************************************************/
+   @Test
+   void sendActionRendersAndRecordsTransactionalDelivery() throws Exception
+   {
+      QRecord contact = contact(1).withValue("companyId", 1);
+      put("crm_company", new QRecord().withValue("id", 1).withValue("name", "Example"));
+      put("crm_contact", contact);
+      put("crm_email_template", new QRecord().withValue("id", 1).withValue("subject", "Hello {{contact.firstName}}")
+         .withValue("bodyText", "{{contact.lastName}} at {{company.name}} / {{owner.name}} / {{unsubscribeUrl}}"));
+      CrmAction.Load load = load();
+      RunBackendStepInput input = new RunBackendStepInput().withProcessName("crm_send_email").withRecords(List.of(contact))
+         .addValue("emailTemplateId", 1).addValue("crmEmailProcessUUID", "run-123");
+      load.runOnePage(input, new RunBackendStepOutput());
+      assertThat(transport.key).isEqualTo("crm:email:1:run-123");
+      assertThat(transport.marketing).isFalse();
+      assertThat(transport.subject).isEqualTo("Hello Alice");
+      assertThat(transport.body).startsWith("Smith at Example / owner / https://crm.example.test/qqq/hooks/crm/unsubscribe/");
+      assertThat(rows("crm_email_message")).singleElement().satisfies(row ->
+      {
+         assertThat(row.getValueInteger("status")).isEqualTo(CrmEmailStatus.QUEUED.getId());
+         assertThat(row.getValueString("emailMessageId")).isEqualTo("provider-1");
+         assertThat(row.getValueString("idempotencyKey")).isEqualTo(transport.key);
+      });
+      assertActivity("EMAIL");
+      load.runOnePage(input, new RunBackendStepOutput());
+      assertThat(transport.sends).isEqualTo(1);
+      assertThat(rows("crm_activity")).hasSize(1);
+   }
+
+   /*******************************************************************************
+    ** Overrides and missing templates and literal merge values.
+    *******************************************************************************/
+   @Test
+   void overridesAndMissingTemplatesAndLiteralMergeValues() throws Exception
+   {
+      QRecord contact = contact(1).withValue("firstName", "{{owner.name}}");
+      assertThat(CrmEmails.render("{{contact.firstName}} {{contact.lastName}}", contact)).isEqualTo("{{owner.name}} Smith");
+      assertThrows(QException.class, () -> CrmEmails.send(transport, contact, 999, null, null, "email", "missing", null));
+      assertThrows(QException.class, () -> CrmEmails.send(transport, contact, null, "subject", "", "email", "missing", null));
+      put("crm_email_template", new QRecord().withValue("id", 1).withValue("subject", "template").withValue("bodyText", "template"));
+      CrmEmails.send(transport, contact, 1, "override", "body", "email", "1:run", null);
+      assertThat(transport.subject).isEqualTo("override");
+      assertThat(transport.body).isEqualTo("body");
+   }
+
+   /*******************************************************************************
+    ** Skips opt out missing address and unavailable transport.
+    *******************************************************************************/
+   @Test
+   void skipsOptOutMissingAddressAndUnavailableTransport() throws Exception
+   {
+      List<QRecord> contacts = List.of(contact(1).withValue("doNotEmail", true), contact(2).withValue("email", null), contact(3));
+      for(QRecord contact : contacts)
+      {
+         put("crm_contact", contact);
+      }
+      transport.ready = false;
+      CrmAction.Load load = load();
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      load.runOnePage(new RunBackendStepInput().withProcessName("crm_send_email").withRecords(contacts), output);
+      assertThat(load.getProcessSummary(output, true)).hasSize(3);
+      assertThat(transport.sends).isZero();
+      assertThat(rows("crm_email_message")).isEmpty();
+      assertThat(rows("crm_activity")).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Process identity uses core uuid.
+    *******************************************************************************/
+   @Test
+   void processIdentityUsesCoreUuid() throws Exception
+   {
+      var output = new RunProcessAction().execute(new RunProcessInput().withProcessName("crm_send_email"));
+      assertThat(output.getException()).isEmpty();
+      assertThat(output.getValues().get("crmEmailProcessUUID")).isEqualTo(output.getProcessUUID());
+      var listener = new CrmEmailProcessIdentity();
+      assertThat(listener.appliesTo("crm_send_email")).isTrue();
+      assertThat(listener.appliesTo("crm_other")).isFalse();
+      listener.onProcessCompleted(new RunProcessInput(), output);
+      listener.onProcessFailed(new RunProcessInput(), new Exception("test"));
+      assertThat(CrmActions.inputFields(CrmActions.action("crm_send_email"))).extracting(field -> field.getName())
+         .containsExactly("emailTemplateId", "subject", "body");
+   }
+
+   /*******************************************************************************
+    ** Active sequence sends marketing and advances once.
+    *******************************************************************************/
+   @Test
+   void activeSequenceSendsMarketingAndAdvancesOnce() throws Exception
+   {
+      seedSequence(true, true);
+      sequence();
+      assertThat(transport.key).isEqualTo("crm:seq:1:1");
+      assertThat(transport.marketing).isTrue();
+      assertThat(rows("crm_email_message")).hasSize(1);
+      assertActivity("SEQUENCE_STEP");
+      assertThat(GetAction.execute("crm_sequence_enrollment", 1).getValueInteger("status")).isEqualTo(CrmEnrollmentStatus.COMPLETED.getId());
+      sequence();
+      assertThat(transport.sends).isEqualTo(1);
+   }
+
+   /*******************************************************************************
+    ** Paused sequence does not advance or send.
+    *******************************************************************************/
+   @Test
+   void pausedSequenceDoesNotAdvanceOrSend() throws Exception
+   {
+      seedSequence(false, true);
+      sequence();
+      assertThat(transport.sends).isZero();
+      assertThat(GetAction.execute("crm_sequence_enrollment", 1).getValueInteger("currentStepNumber")).isZero();
+      assertThat(Sink.alerts).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Missing consent pauses enrollment with reason.
+    *******************************************************************************/
+   @Test
+   void missingConsentPausesEnrollmentWithReason() throws Exception
+   {
+      seedSequence(true, false);
+      sequence();
+      QRecord enrollment = GetAction.execute("crm_sequence_enrollment", 1);
+      assertThat(enrollment.getValueInteger("status")).isEqualTo(CrmEnrollmentStatus.PAUSED.getId());
+      assertThat(enrollment.getValueString("lastFailureMessage")).isEqualTo("no_email_consent");
+      assertThat(enrollment.getValueInteger("failureCount")).isZero();
+      assertThat(transport.sends).isZero();
+   }
+
+   /*******************************************************************************
+    ** Transport failures raise on third attempt.
+    *******************************************************************************/
+   @Test
+   void transportFailuresRaiseOnThirdAttempt() throws Exception
+   {
+      seedSequence(true, true);
+      transport.fail = true;
+      sequence();
+      sequence();
+      sequence();
+      assertThat(GetAction.execute("crm_sequence_enrollment", 1).getValueInteger("failureCount")).isEqualTo(3);
+      assertThat(GetAction.execute("crm_sequence_enrollment", 1).getValueInteger("status")).isEqualTo(CrmEnrollmentStatus.FAILED.getId());
+      assertThat(Sink.alerts).singleElement().satisfies(signal ->
+      {
+         assertThat(signal.getAlertTypeName()).isEqualTo("crm.sequence.failing");
+         assertThat(signal.getDedupKey()).isEqualTo("enrollment:1");
+      });
+      assertThat(rows("crm_email_message")).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Unavailable transport raises then resolves without failing enrollment.
+    *******************************************************************************/
+   @Test
+   void unavailableTransportRaisesThenResolvesWithoutFailingEnrollment() throws Exception
+   {
+      seedSequence(true, true);
+      transport.ready = false;
+      sequence();
+      assertThat(Sink.alerts).singleElement().satisfies(signal ->
+      {
+         assertThat(signal.getAlertTypeName()).isEqualTo("crm.email.unavailable");
+         assertThat(signal.getDedupKey()).isEqualTo("transport");
+      });
+      assertThat(GetAction.execute("crm_sequence_enrollment", 1).getValueInteger("failureCount")).isZero();
+      transport.ready = true;
+      sequence();
+      assertThat(Sink.resolutions).hasSize(1);
+      assertThat(transport.sends).isEqualTo(1);
+   }
+
+   /*******************************************************************************
+    ** Delivery sync uses recent nonfinal messages and preserves dates.
+    *******************************************************************************/
+   @Test
+   void deliverySyncUsesRecentNonfinalMessagesAndPreservesDates() throws Exception
+   {
+      contact(1);
+      message(1, CrmEmailStatus.SENT, Instant.now());
+      message(2, CrmEmailStatus.SENT, Instant.now().minus(31, ChronoUnit.DAYS));
+      message(3, CrmEmailStatus.FAILED, Instant.now());
+      Instant now = Instant.now();
+      transport.deliveries.put("key:1", new CrmEmailTransport.Delivery("provider", CrmEmailStatus.CLICKED, now, now, now, now, null));
+      sync();
+      assertThat(transport.lookups).containsExactly("key:1");
+      QRecord message = GetAction.execute("crm_email_message", 1);
+      assertThat(message.getValueInstant("deliveredDate")).isEqualTo(now);
+      assertThat(message.getValueInstant("firstOpenedDate")).isEqualTo(now);
+      assertThat(message.getValueInstant("firstClickedDate")).isEqualTo(now);
+      transport.deliveries.put("key:1", new CrmEmailTransport.Delivery("provider", CrmEmailStatus.CLICKED, null, null, null, null, null));
+      sync();
+      assertThat(GetAction.execute("crm_email_message", 1).getValueInstant("firstOpenedDate")).isEqualTo(now);
+   }
+
+   /*******************************************************************************
+    ** Bounce and complaint withdraw once and suppress contact.
+    *******************************************************************************/
+   @Test
+   void bounceAndComplaintWithdrawOnceAndSuppressContact() throws Exception
+   {
+      for(int id = 1; id <= 2; id++)
+      {
+         contact(id);
+         message(id, CrmEmailStatus.DELIVERED, Instant.now());
+         CrmEmailStatus status = id == 1 ? CrmEmailStatus.BOUNCED : CrmEmailStatus.COMPLAINED;
+         transport.deliveries.put("key:" + id, new CrmEmailTransport.Delivery("provider", status, null, null, null, null, Instant.now()));
+      }
+      sync();
+      sync();
+      assertThat(rows("crm_contact")).allSatisfy(row -> assertThat(row.getValueBoolean("doNotEmail")).isTrue());
+      assertThat(rows("crm_consent_record")).hasSize(2).allSatisfy(row ->
+      {
+         assertThat(row.getValueInteger("consentType")).isEqualTo(CrmConsentType.EMAIL.getId());
+         assertThat(row.getValueInteger("status")).isEqualTo(CrmConsentStatus.WITHDRAWN.getId());
+      });
+      assertThat(rows("crm_consent_record")).extracting(row -> row.getValueString("source")).containsExactlyInAnyOrder("Hard bounce", "Complaint");
+   }
+
+   /*******************************************************************************
+    ** Absent email qbit is no op.
+    *******************************************************************************/
+   @Test
+   void absentEmailQbitIsNoOp() throws Exception
+   {
+      message(1, CrmEmailStatus.SENT, Instant.now());
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      new SyncEmailDeliveryProcess().run(new RunBackendStepInput(), output);
+      assertThat(output.getValueInteger("updatedCount")).isZero();
+      assertThat(new CrmEmailTransport(CrmApi.config()).lookup("key:1")).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Stale pending members expire without new consent rows.
+    *******************************************************************************/
+   @Test
+   void stalePendingMembersExpireWithoutNewConsentRows() throws Exception
+   {
+      for(int id = 1; id <= 3; id++)
+      {
+         put("crm_email_list_member", new QRecord().withValue("id", id).withValue("emailListId", 1).withValue("email", "a" + id + "@example.test")
+            .withValue("status", id == 3 ? CrmListMemberStatus.SUBSCRIBED.getId() : CrmListMemberStatus.PENDING_CONFIRMATION.getId())
+            .withValue("confirmationSentDate", Instant.now().minus(id == 2 ? 13 : 15, ChronoUnit.DAYS)).withValue("confirmationTokenHash", "hash"));
+      }
+      put("crm_consent_record", new QRecord().withValue("id", 1).withValue("expiryDate", Instant.now().minusSeconds(1)));
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      new ExpireConsentsProcess().run(new RunBackendStepInput(), output);
+      assertThat(output.getValueInteger("expiredPendingCount")).isEqualTo(1);
+      assertThat(output.getValueInteger("expiredConsentCount")).isEqualTo(1);
+      assertThat(GetAction.execute("crm_email_list_member", 1).getValue("confirmationTokenHash")).isNull();
+      assertThat(GetAction.execute("crm_email_list_member", 1).getValueInteger("status")).isEqualTo(CrmListMemberStatus.UNSUBSCRIBED.getId());
+      assertThat(GetAction.execute("crm_email_list_member", 2).getValueString("confirmationTokenHash")).isEqualTo("hash");
+      assertThat(rows("crm_consent_record")).hasSize(1);
+   }
+
+   /*******************************************************************************
+    ** Schedules require scheduler and can be disabled.
+    *******************************************************************************/
+   @Test
+   void schedulesRequireSchedulerAndCanBeDisabled() throws Exception
+   {
+      Map<String, Integer> intervals = Map.of("crm_process_sequence_steps", 300, "crm_calculate_lead_scores", 3600,
+         "crm_sync_email_delivery", 900, "crm_expire_consents", 86400);
+      for(String name : intervals.keySet())
+      {
+         assertThat(QContext.getQInstance().getProcess(name).getSchedule()).isNull();
+      }
+      CrmQBitConfig config = new CrmQBitConfig().withDefaultBackendNameForTables(BACKEND_NAME).withSchedulerName("jobs");
+      var output = new CrmQBitProducer().withQBitConfig(config).produce(new com.kingsrook.qqq.backend.core.model.metadata.QInstance());
+      for(var process : output.getEach(com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData.class))
+      {
+         if(intervals.containsKey(process.getName()))
+         {
+            assertThat(process.getSchedule().getRepeatSeconds()).isEqualTo(intervals.get(process.getName()));
+            assertThat(process.getSchedule().getSchedulerName()).isEqualTo("jobs");
+         }
+      }
+      assertThat(config.getSchedulesEnabled()).isTrue();
+      output = new CrmQBitProducer().withQBitConfig(config.withSchedulesEnabled(false)).produce(new com.kingsrook.qqq.backend.core.model.metadata.QInstance());
+      assertThat(output.getEach(com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData.class))
+         .allSatisfy(process -> assertThat(process.getSchedule()).isNull());
+   }
+
+   /*******************************************************************************
+    ** Collaboration and alerts use core metadata.
+    *******************************************************************************/
+   @Test
+   void collaborationAndAlertsUseCoreMetadata() throws Exception
+   {
+      for(String table : List.of("crm_contact", "crm_company", "crm_deal", "crm_activity"))
+      {
+         var features = RecordFeaturesMetaData.of(QContext.getQInstance().getTable(table));
+         for(RecordFeature feature : List.of(RecordFeature.TAGS, RecordFeature.COMMENTS, RecordFeature.ATTACHMENTS,
+            RecordFeature.FOLLOW, RecordFeature.TIMELINE, RecordFeature.RECENTS, RecordFeature.FAVORITES))
+         {
+            assertThat(features.getEnabled().get(feature)).isTrue();
+         }
+         assertThat(features.getOwnership().getOwnerFieldName()).isEqualTo("ownerUserId");
+         assertThat(features.getEnabled().get(RecordFeature.TASKS)).isEqualTo("crm_activity".equals(table) ? null : true);
+         assertThat(features.getEnabled().get(RecordFeature.ASSIGNMENT)).isEqualTo("crm_activity".equals(table) ? null : true);
+      }
+      assertThat(AlertsInstanceMetaData.of(QContext.getQInstance()).getAlertTypes()).containsKeys("crm.sequence.failing", "crm.email.unavailable",
+         "crm.lead.assigned", "crm.deal.won", "crm.deal.rotting", "crm.task.due");
+   }
+
+   /*******************************************************************************
+    ** Scoring uses dates incremental changes and daily full pass.
+    *******************************************************************************/
+   @Test
+   void scoringUsesDatesIncrementalChangesAndDailyFullPass() throws Exception
+   {
+      put("crm_contact", contact(1).withValue("lastActivityDate", Instant.now().minus(20, ChronoUnit.DAYS)).withValue("modifyDate", Instant.now().minusSeconds(300)));
+      put("crm_contact", contact(2).withValue("lastActivityDate", Instant.now().minus(2, ChronoUnit.DAYS)).withValue("modifyDate", Instant.now().minusSeconds(300)));
+      rule(1, "lastActivityDate", CrmScoreOperator.DAYS_AGO_GT, "10", -5);
+      rule(2, "lastActivityDate", CrmScoreOperator.DAYS_AGO_LT, "10", 10);
+      assertThat(scores(false).getValueInteger("totalContacts")).isEqualTo(2);
+      assertThat(GetAction.execute("crm_contact", 1).getValueInteger("leadScore")).isEqualTo(-5);
+      assertThat(GetAction.execute("crm_contact", 2).getValueInteger("leadScore")).isEqualTo(10);
+      assertThat(scores(false).getValueInteger("totalContacts")).isZero();
+      CrmRecords.update("crm_contact", new QRecord().withValue("id", 2).withValue("firstName", "Changed"));
+      assertThat(scores(false).getValueInteger("totalContacts")).isEqualTo(1);
+      assertThat(scores(true).getValueInteger("totalContacts")).isEqualTo(2);
+      QRecord state = rows("crm_job_state").getFirst();
+      CrmRecords.update("crm_job_state", state.withValue("lastFullPass", Instant.now().minus(25, ChronoUnit.HOURS)));
+      assertThat(scores(false).getValueInteger("totalContacts")).isEqualTo(2);
+   }
+
+   /*******************************************************************************
+    ** Scoring handles calendar dates and malformed rules.
+    *******************************************************************************/
+   @Test
+   void scoringHandlesCalendarDatesAndMalformedRules() throws Exception
+   {
+      put("crm_contact", contact(1).withValue("lastName", LocalDate.now().minusDays(30).toString()));
+      rule(1, "lastName", CrmScoreOperator.DAYS_AGO_GT, "10", 7);
+      rule(2, "firstName", CrmScoreOperator.DAYS_AGO_LT, "bad", 100);
+      rule(3, "email", CrmScoreOperator.DAYS_AGO_GT, "10", 100);
+      scores(true);
+      assertThat(GetAction.execute("crm_contact", 1).getValueInteger("leadScore")).isEqualTo(7);
+   }
+
+   /*******************************************************************************
+    ** Insert assignment preserves owners and notifies.
+    *******************************************************************************/
+   @Test
+   void insertAssignmentPreservesOwnersAndNotifies() throws Exception
+   {
+      for(CrmAssignableEntity entity : CrmAssignableEntity.values())
+      {
+         put("crm_assignment_rule", new QRecord().withValue("id", entity.getId()).withValue("name", "Assign")
+            .withValue("entityType", entity.getId()).withValue("assignStrategy", CrmAssignStrategy.SPECIFIC_USER.getId())
+            .withValue("assignToUserId", "assigned").withValue("isActive", true).withValue("sortOrder", 1));
+      }
+      QRecord inserted = CrmRecords.insert("crm_contact", new QRecord().withValue("firstName", "New").withValue("lastName", "Contact"));
+      assertThat(GetAction.execute("crm_contact", inserted.getValue("id")).getValueString("ownerUserId")).isEqualTo("assigned");
+      CrmRecords.insert("crm_contact", new QRecord().withValue("firstName", "Already").withValue("lastName", "Owned").withValue("ownerUserId", "explicit"));
+      put("crm_pipeline", new QRecord().withValue("id", 1).withValue("name", "Sales"));
+      put("crm_pipeline_stage", new QRecord().withValue("id", 1).withValue("pipelineId", 1).withValue("name", "Open"));
+      QRecord deal = CrmRecords.insert("crm_deal", new QRecord().withValue("name", "New deal").withValue("pipelineId", 1)
+         .withValue("pipelineStageId", 1).withValue("ownerUserId", "system").withValue("currencyCode", "USD"));
+      assertThat(GetAction.execute("crm_deal", deal.getValue("id")).getValueString("ownerUserId")).isEqualTo("assigned");
+      assertThat(Sink.notifications).hasSize(2).allSatisfy(signal ->
+      {
+         assertThat(signal.getSourceName()).isEqualTo("crm.lead.assigned");
+         assertThat(signal.getAudience()).isEqualTo(NotificationAudience.user("assigned"));
+      });
+   }
+
+   /*******************************************************************************
+    ** Won and rotting notifications target owner and deduplicate stage.
+    *******************************************************************************/
+   @Test
+   void wonAndRottingNotificationsTargetOwnerAndDeduplicateStage() throws Exception
+   {
+      put("crm_pipeline_stage", new QRecord().withValue("id", 1).withValue("name", "Open").withValue("pipelineId", 1).withValue("rotDays", 5));
+      put("crm_pipeline_stage", new QRecord().withValue("id", 2).withValue("name", "Won").withValue("pipelineId", 1).withValue("isClosedWon", true));
+      put("crm_deal", new QRecord().withValue("id", 1).withValue("name", "Old deal").withValue("pipelineId", 1)
+         .withValue("pipelineStageId", 1).withValue("ownerUserId", "owner").withValue("stageEnteredDate", Instant.now().minus(10, ChronoUnit.DAYS)));
+      CrmDealNotifications.checkRotting();
+      CrmDealNotifications.checkRotting();
+      assertThat(Sink.notifications).singleElement().satisfies(signal -> assertThat(signal.getSourceKey()).isEqualTo("deal:1:stage:1"));
+      CrmRecords.update("crm_deal", new QRecord().withValue("id", 1).withValue("pipelineStageId", 2));
+      assertThat(Sink.notifications).hasSize(2);
+      assertThat(Sink.notifications.getLast().getSourceName()).isEqualTo("crm.deal.won");
+      assertThat(Sink.notifications.getLast().getAudience()).isEqualTo(NotificationAudience.user("owner"));
+      CrmRecords.update("crm_deal", new QRecord().withValue("id", 1).withValue("pipelineStageId", 2));
+      CrmDealNotifications.checkRotting();
+      assertThat(Sink.notifications).hasSize(2);
+   }
+
+
+   /*******************************************************************************
+    ** Company loading scales with pages, not contacts or rules.
+    *******************************************************************************/
+   @Test
+   void scoringLoadsCompaniesOncePerPage() throws Exception
+   {
+      put("crm_company", new QRecord().withValue("id", 1).withValue("name", "Example"));
+      for(int id = 1; id <= 201; id++)
+      {
+         put("crm_contact", contact(id).withValue("companyId", 1));
+      }
+      rule(1, "company.name", CrmScoreOperator.EQUALS, "Example", 3);
+      rule(2, "company.name", CrmScoreOperator.IS_SET, null, 2);
+      CompanyReads.count = 0;
+      QContext.getQInstance().getTable("crm_company").withCustomizer(
+         com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers.POST_QUERY_RECORD, new QCodeReference(CompanyReads.class));
+      assertThat(scores(true).getValueInteger("updatedCount")).isEqualTo(201);
+      // Two explicit page loads, plus at most two core audit-label translations.
+      assertThat(CompanyReads.count).isBetween(2, 4);
+      assertThat(GetAction.execute("crm_contact", 201).getValueInteger("leadScore")).isEqualTo(5);
+   }
+
+   /*******************************************************************************
+    ** Failed scoring must leave the successful checkpoint unchanged.
+    *******************************************************************************/
+   @Test
+   void scoringFailureDoesNotAdvanceCheckpoint() throws Exception
+   {
+      contact(1);
+      scores(true);
+      Instant checkpoint = rows("crm_job_state").getFirst().getValueInstant("lastSuccess");
+      rule(1, "firstName", CrmScoreOperator.IS_SET, null, 1);
+      QContext.getQInstance().getTable("crm_contact").withCustomizer(
+         com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers.PRE_UPDATE_RECORD, new QCodeReference(RejectScore.class));
+      assertThrows(QException.class, () -> scores(true));
+      assertThat(rows("crm_job_state").getFirst().getValueInstant("lastSuccess")).isEqualTo(checkpoint);
+   }
+
+   /*******************************************************************************
+    ** Retry repairs a missing activity without sending the persisted message again.
+    *******************************************************************************/
+   @Test
+   void retryRepairsActivityAfterPartialRecording() throws Exception
+   {
+      QRecord contact = contact(1);
+      put("crm_email_message", new QRecord().withValue("id", 1).withValue("contactId", 1).withValue("subject", "Already sent")
+         .withValue("idempotencyKey", "crm:email:1:retry").withValue("status", CrmEmailStatus.SENT.getId()));
+      CrmEmails.send(transport, contact, null, "Already sent", "body", "email", "1:retry", null);
+      assertThat(transport.sends).isZero();
+      assertActivity("EMAIL");
+      assertThat(GetAction.execute("crm_email_message", 1).getValue("activityId")).isNotNull();
+   }
+
+   /*******************************************************************************
+    ** Count actual backend company queries.
+    *******************************************************************************/
+   public static class CompanyReads extends com.kingsrook.qqq.backend.core.actions.customizers.AbstractPostQueryCustomizer
+   {
+      private static int count;
+
+      @Override
+      public List<QRecord> apply(List<QRecord> records)
+      {
+         count++;
+         return (records);
+      }
+   }
+
+   /*******************************************************************************
+    ** Return a record-level error to exercise checked background writes.
+    *******************************************************************************/
+   public static class RejectScore extends com.kingsrook.qqq.backend.core.actions.customizers.AbstractPreUpdateCustomizer
+   {
+      @Override
+      public List<QRecord> apply(List<QRecord> records)
+      {
+         records.forEach(row -> row.addError(new com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage("Rejected score")));
+         return (records);
+      }
+   }
+
+   /*******************************************************************************
+    ** Send action with a local fake.
+    *******************************************************************************/
+   private CrmAction.Load load()
+   {
+      return (new CrmAction.Load()
+      {
+         @Override
+         protected CrmEmailTransport emailTransport()
+         {
+            return (transport);
+         }
+      });
+   }
+
+   /*******************************************************************************
+    ** Sequence runner with a local fake.
+    *******************************************************************************/
+   private void sequence() throws QException
+   {
+      new ProcessSequenceStepsProcess()
+      {
+         @Override
+         protected CrmEmailTransport emailTransport()
+         {
+            return (transport);
+         }
+      }.run(new RunBackendStepInput(), new RunBackendStepOutput());
+   }
+
+   /*******************************************************************************
+    ** Sync with a local fake.
+    *******************************************************************************/
+   private void sync() throws QException
+   {
+      new SyncEmailDeliveryProcess()
+      {
+         @Override
+         protected CrmEmailTransport emailTransport()
+         {
+            return (transport);
+         }
+      }.run(new RunBackendStepInput(), new RunBackendStepOutput());
+   }
+
+   /*******************************************************************************
+    ** Run score work using normal QQQ actions.
+    *******************************************************************************/
+   private RunBackendStepOutput scores(boolean full) throws QException
+   {
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      new CalculateLeadScoresProcess().run(new RunBackendStepInput().addValue("fullRecalc", full), output);
+      return (output);
+   }
+
+   /*******************************************************************************
+    ** Contact fixture.
+    *******************************************************************************/
+   private QRecord contact(int id) throws QException
+   {
+      QRecord row = new QRecord().withValue("id", id).withValue("firstName", "Alice").withValue("lastName", "Smith")
+         .withValue("email", "alice" + id + "@example.test").withValue("ownerUserId", "owner");
+      put("crm_contact", row);
+      return (row);
+   }
+
+   /*******************************************************************************
+    ** Sequence fixtures use enum identifiers.
+    *******************************************************************************/
+   private void seedSequence(boolean active, boolean eligible) throws QException
+   {
+      contact(1);
+      put("crm_email_sequence", new QRecord().withValue("id", 1).withValue("name", "Sequence").withValue("isActive", active).withValue("totalSteps", 1));
+      put("crm_email_template", new QRecord().withValue("id", 1).withValue("subject", "Hello {{contact.firstName}}").withValue("bodyText", "Welcome"));
+      put("crm_sequence_step", new QRecord().withValue("id", 1).withValue("sequenceId", 1).withValue("stepNumber", 1)
+         .withValue("stepType", CrmSequenceStepType.EMAIL.getId()).withValue("emailTemplateId", 1));
+      put("crm_sequence_enrollment", new QRecord().withValue("id", 1).withValue("sequenceId", 1).withValue("contactId", 1)
+         .withValue("status", CrmEnrollmentStatus.ACTIVE.getId()).withValue("currentStepNumber", 0).withValue("failureCount", 0)
+         .withValue("nextStepDate", Instant.now().minusSeconds(60)));
+      if(eligible)
+      {
+         put("crm_consent_record", new QRecord().withValue("id", 1).withValue("contactId", 1).withValue("consentType", CrmConsentType.EMAIL.getId())
+            .withValue("status", CrmConsentStatus.GRANTED.getId()).withValue("consentDate", Instant.now()));
+      }
+   }
+
+   /*******************************************************************************
+    ** Transport-linked message fixture.
+    *******************************************************************************/
+   private void message(int id, CrmEmailStatus status, Instant created) throws QException
+   {
+      put("crm_email_message", new QRecord().withValue("id", id).withValue("contactId", id).withValue("subject", "Test")
+         .withValue("idempotencyKey", "key:" + id).withValue("status", status.getId()).withValue("createDate", created));
+   }
+
+   /*******************************************************************************
+    ** Score rule fixture.
+    *******************************************************************************/
+   private void rule(int id, String field, CrmScoreOperator operator, String expected, int points) throws QException
+   {
+      put("crm_lead_score_rule", new QRecord().withValue("id", id).withValue("name", "Rule " + id).withValue("fieldPath", field)
+         .withValue("operator", operator.getId()).withValue("fieldValue", expected).withValue("scoreAdjustment", points).withValue("isActive", true));
+   }
+
+   /*******************************************************************************
+    ** Assert the activity's stable code.
+    *******************************************************************************/
+   private void assertActivity(String code) throws QException
+   {
+      QRecord activity = rows("crm_activity").getFirst();
+      assertThat(GetAction.execute("crm_activity_type", activity.getValue("activityTypeId")).getValueString("code")).isEqualTo(code);
+   }
+
+   /*******************************************************************************
+    ** Insert isolated fixture data.
+    *******************************************************************************/
+   private void put(String table, QRecord row) throws QException
+   {
+      if(row.getValue("id") != null && GetAction.execute(table, row.getValue("id")) != null)
+      {
+         MemoryRecordStore.getInstance().update(new com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput(table).withRecord(row), true);
+      }
+      else
+      {
+         MemoryRecordStore.getInstance().insert(new InsertInput(table).withRecord(row), true);
+      }
+   }
+
+   /*******************************************************************************
+    ** Query in the current session.
+    *******************************************************************************/
+   private List<QRecord> rows(String table) throws QException
+   {
+      return (new QueryAction().execute(new QueryInput(table)).getRecords());
+   }
+
+   /*******************************************************************************
+    ** Fake transport with independent send and delivery controls.
+    *******************************************************************************/
+   private static class FakeTransport extends CrmEmailTransport
+   {
+      private boolean ready = true;
+      private boolean fail;
+      private boolean marketing;
+      private int sends;
+      private String key;
+      private String subject;
+      private String body;
+      private final Map<String, Delivery> deliveries = new HashMap<>();
+      private final List<String> lookups = new ArrayList<>();
+
+      /*******************************************************************************
+       ** No external provider is installed.
+       *******************************************************************************/
+      FakeTransport()
+      {
+         super(new CrmQBitConfig());
+      }
+
+      @Override
+      public CrmReadiness readiness()
+      {
+         return (ready ? CrmReadiness.QBIT_EMAIL : CrmReadiness.NOT_CONFIGURED);
+      }
+
+      @Override
+      public Delivery sendMessage(String purpose, String id, Integer contactId, String address, String subject, String body, boolean marketing) throws QException
+      {
+         if(fail)
+         {
+            throw (new QException("Fake transport failure"));
+         }
+         sends++;
+         key = "crm:" + purpose + ":" + id;
+         this.subject = subject;
+         this.body = body;
+         this.marketing = marketing;
+         return (new Delivery("provider-1", CrmEmailStatus.QUEUED, null, null, null, null, null));
+      }
+
+      @Override
+      public Optional<Delivery> lookup(String key)
+      {
+         lookups.add(key);
+         return (Optional.ofNullable(deliveries.get(key)));
+      }
+   }
+
+   /*******************************************************************************
+    ** Capture the core dispatch without an alerts backend.
+    *******************************************************************************/
+   public static class Sink extends LoggingAlertSink
+   {
+      private static final List<NotificationSignal> notifications = new ArrayList<>();
+      private static final List<AlertSignal> alerts = new ArrayList<>();
+      private static final List<AlertResolution> resolutions = new ArrayList<>();
+
+      @Override
+      public void notify(NotificationSignal signal)
+      {
+         notifications.add(signal);
+      }
+
+      @Override
+      public void raise(AlertSignal signal)
+      {
+         alerts.add(signal);
+      }
+
+      @Override
+      public void resolve(AlertResolution resolution)
+      {
+         resolutions.add(resolution);
+      }
+   }
+}

src/test/java/com/kingsrook/qbits/crm/actions/CrmActionsTest.java

@@ -133,7 +133,7 @@ class CrmActionsTest extends BaseTest
    void metadataIsComplete()
    {
       assertThat(CrmActions.ACTIONS).hasSize(28);
-      assertThat(QContext.getQInstance().getProcesses().keySet().stream().filter(name -> !name.contains("."))).hasSize(30);
+      assertThat(QContext.getQInstance().getProcesses().keySet().stream().filter(name -> !name.contains("."))).hasSize(32);
       for(CrmActions.Action action : CrmActions.ACTIONS)
       {
          QProcessMetaData process = QContext.getQInstance().getProcess(action.name());
@@ -154,7 +154,7 @@ class CrmActionsTest extends BaseTest
             assertThat(process.getMaxInputRecords()).isEqualTo(2);
          }
       }
-      for(String name : List.of("crm_process_sequence_steps", "crm_calculate_lead_scores"))
+      for(String name : List.of("crm_process_sequence_steps", "crm_calculate_lead_scores", "crm_sync_email_delivery", "crm_expire_consents"))
       {
          assertThat(QContext.getQInstance().getProcess(name).getTableName()).isNull();
       }

src/test/java/com/kingsrook/qbits/crm/email/processes/ProcessSequenceStepsProcessTest.java

@@ -60,10 +60,10 @@ class ProcessSequenceStepsProcessTest extends BaseTest
 {
 
    /*******************************************************************************
-    ** Email step should create an activity record.
+    ** Email without recorded consent pauses without creating a fake activity.
     *******************************************************************************/
    @Test
-   void testEmailStepCreatesActivity() throws QException
+   void testEmailStepRequiresConsent() throws QException
    {
       insertActivityType("Email");
       insertActivityType("Task");
@@ -91,8 +91,7 @@ class ProcessSequenceStepsProcessTest extends BaseTest
       QueryOutput activityQuery = new QueryAction().execute(
          new QueryInput(Activity.TABLE_NAME));
 
-      assertThat(activityQuery.getRecords()).isNotEmpty();
-      assertThat(activityQuery.getRecords().get(0).getValueString("subject")).contains("Sequence email");
+      assertThat(activityQuery.getRecords()).isEmpty();
 
       /////////////////////////////////////////////
       // verify enrollment was advanced          //
@@ -101,7 +100,8 @@ class ProcessSequenceStepsProcessTest extends BaseTest
          new QueryInput(SequenceEnrollment.TABLE_NAME));
 
       SequenceEnrollment enrollment = new SequenceEnrollment(enrollmentQuery.getRecords().get(0));
-      assertEquals(1, enrollment.getCurrentStepNumber());
+      assertEquals(0, enrollment.getCurrentStepNumber());
+      assertEquals(CrmEnrollmentStatus.PAUSED.getId(), enrollment.getStatus());
    }
 
 
@@ -142,10 +142,10 @@ class ProcessSequenceStepsProcessTest extends BaseTest
 
 
    /*******************************************************************************
-    ** After 3 failures, enrollment should be marked as FAILED.
+    ** Opt-out pauses an enrollment without counting it as a delivery failure.
     *******************************************************************************/
    @Test
-   void testFailureCounting() throws QException
+   void testOptOutPausesWithoutCountingFailure() throws QException
    {
       Integer sequenceId = insertSequence("Fail Seq", true, 2, false);
       Integer contactId = insertContact("Carol", "Fail", true); // doNotEmail = true
@@ -181,8 +181,8 @@ class ProcessSequenceStepsProcessTest extends BaseTest
          new QueryInput(SequenceEnrollment.TABLE_NAME));
 
       SequenceEnrollment enrollment = new SequenceEnrollment(enrollmentQuery.getRecords().get(0));
-      assertEquals(CrmEnrollmentStatus.FAILED.getId(), enrollment.getStatus());
-      assertEquals(3, enrollment.getFailureCount());
+      assertEquals(CrmEnrollmentStatus.PAUSED.getId(), enrollment.getStatus());
+      assertEquals(2, enrollment.getFailureCount());
       assertThat(enrollment.getLastFailureMessage()).isNotNull();
    }