fix(outbound): the record-change listener only costs queries for configured source tables

OutboundEnqueue applied to every non-edi_ table, and per record opened a
system session and queried edi_agreement_document, so every insert and
update of audit, auditDetail, worm_audit_chain, jobs, alerts and the
rest paid for it (and UPDATEs fetched old records for it).

- OutboundSourceTables (per EdiRuntime) caches the source tables that
  have an active RECORD_CHANGE outbound document on an active agreement,
  loaded from committed data. appliesTo answers from it, so other tables
  do no work at all.
- Changes to edi_agreement / edi_agreement_document invalidate it, and
  again after their transaction commits (a reload that raced the change
  is discarded); until then the changing transaction bypasses the cache
  and sees its own definitions. A 60s TTL covers other processes.
- onRecordsChanged resolves definitions (documents + active agreements)
  once per event, not once per record. MANUAL and ESB_EVENT enqueue keep
  their semantics.

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

CHANGELOG.md

@@ -6,6 +6,8 @@ Following [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and [Semantic
 
 ### Changed
 
+- Derive qbit metadata version from the Maven artifact version, keeping release metadata and help bundle versions in sync.
+
 - Require QQQ 4.1.0-RC.7 for core help-bundle registration and coverage validation.
 
 - X12 routing and sending honor identity `isActive`, matching EDIFACT.

pom.xml

@@ -133,6 +133,7 @@
             <filtering>true</filtering>
             <includes>
                <include>help/*.yaml</include>
+               <include>qbit-edi.properties</include>
             </includes>
          </resource>
          <resource>
@@ -140,10 +141,21 @@
             <filtering>false</filtering>
             <excludes>
                <exclude>help/*.yaml</exclude>
+               <exclude>qbit-edi.properties</exclude>
             </excludes>
          </resource>
       </resources>
       <plugins>
+         <!-- the artifact version, for the help bundle version test -->
+         <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-surefire-plugin</artifactId>
+            <configuration>
+               <systemPropertyVariables combine.children="append">
+                  <qbit.artifactVersion>${project.version}</qbit.artifactVersion>
+               </systemPropertyVariables>
+            </configuration>
+         </plugin>
          <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-checkstyle-plugin</artifactId>

src/main/java/com/kingsrook/qbits/edi/EdiQBitProducer.java

@@ -21,6 +21,9 @@
 package com.kingsrook.qbits.edi;
 
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@@ -37,13 +40,41 @@ public class EdiQBitProducer implements QBitMetaDataProducer<EdiQBitConfig>
 {
    public static final String GROUP_ID    = "com.kingsrook.qbits";
    public static final String ARTIFACT_ID = "qbit-edi";
-   public static final String VERSION     = "0.1.0";
+   public static final String VERSION     = loadVersion();
 
    private EdiQBitConfig qBitConfig;
    private String        namespace;
 
 
 
+   /*******************************************************************************
+    ** Read the artifact version from the Maven-filtered resource.
+    *******************************************************************************/
+   private static String loadVersion()
+   {
+      try(InputStream inputStream = EdiQBitProducer.class.getClassLoader().getResourceAsStream("qbit-edi.properties"))
+      {
+         if(inputStream == null)
+         {
+            throw new IllegalStateException("Missing qbit-edi.properties; build with Maven to supply the artifact version");
+         }
+         Properties properties = new Properties();
+         properties.load(inputStream);
+         String version = properties.getProperty("version");
+         if(version == null || version.isBlank() || version.contains("${"))
+         {
+            throw new IllegalStateException("Missing or unfiltered version in qbit-edi.properties; build with Maven");
+         }
+         return (version);
+      }
+      catch(IOException e)
+      {
+         throw new IllegalStateException("Could not read the artifact version from qbit-edi.properties", e);
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Fluent configuration setter.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/edi/EdiRuntime.java

@@ -64,6 +64,7 @@ public final class EdiRuntime
    private final Ack999Builder ack999 = new Ack999Builder();
    private final TA1Builder ta1 = new TA1Builder();
    private final InboundAckReader acknowledgements = new InboundAckReader();
+   private final com.kingsrook.qbits.edi.flow.OutboundSourceTables outboundSources = new com.kingsrook.qbits.edi.flow.OutboundSourceTables();
 
    /*******************************************************************************
     ** Construct services without resolving keys until content is stored or opened.
@@ -95,6 +96,22 @@ public final class EdiRuntime
       return (Objects.requireNonNull(INSTANCES.get(QContext.getQInstance()), "EDI runtime has not been installed"));
    }
 
+   /*******************************************************************************
+    ** Services for the current QQQ context, or null when none are installed.
+    *******************************************************************************/
+   public static EdiRuntime find()
+   {
+      return (QContext.getQInstance() == null ? null : INSTANCES.get(QContext.getQInstance()));
+   }
+
+   /*******************************************************************************
+    ** Cached source tables that have active RECORD_CHANGE outbound definitions.
+    *******************************************************************************/
+   public com.kingsrook.qbits.edi.flow.OutboundSourceTables outboundSources()
+   {
+      return (outboundSources);
+   }
+
    /*******************************************************************************
     ** Call at host startup after DDL, with the service session established.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/edi/flow/OutboundEnqueue.java

@@ -20,7 +20,9 @@
 
 package com.kingsrook.qbits.edi.flow;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.kingsrook.qbits.edi.EdiRuntime;
 import com.kingsrook.qbits.edi.storage.EdiRecords;
@@ -30,36 +32,80 @@ import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeListe
 import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType;
 import com.kingsrook.qqq.backend.core.context.QContext;
 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.query.QCriteriaOperator;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
 
 /*******************************************************************************
  ** Transactional source listener and explicit enqueue boundary for MANUAL/ESB.
+ ** The listener applies only to tables with an active RECORD_CHANGE definition
+ ** (cached per runtime, see {@link OutboundSourceTables}), and to the definition
+ ** tables themselves, whose changes invalidate that cache.
  *******************************************************************************/
 public class OutboundEnqueue implements RecordChangeListenerInterface
 {
+   private static final QLogger LOG = QLogger.getLogger(OutboundEnqueue.class);
+
    /*******************************************************************************
-    ** Ignore internal tables and deletes; definitions are read at event time.
+    ** Definition changes (any type), and inserts/updates of configured sources.
     *******************************************************************************/
    public boolean appliesTo(String table, RecordChangeType type)
    {
-      return (!table.startsWith("edi_") && type != RecordChangeType.DELETE);
+      EdiRuntime runtime = EdiRuntime.find();
+      if(runtime == null)
+      {
+         return (false);
+      }
+      if(OutboundSourceTables.DEFINITION_TABLES.contains(table))
+      {
+         return (true);
+      }
+      if(table.startsWith("edi_") || type == RecordChangeType.DELETE)
+      {
+         return (false);
+      }
+      try
+      {
+         return (runtime.outboundSources().contains(table, QContext.getQBackendTransaction()));
+      }
+      catch(Exception e)
+      {
+         //////////////////////////////////////////////////////////////////////
+         // if the definitions can't be read, apply - so the event runs its  //
+         // own queries and reports the failure, rather than silently        //
+         // dropping outbound work.                                          //
+         //////////////////////////////////////////////////////////////////////
+         LOG.warn("Unable to read EDI outbound source tables", e, logPair("table", table));
+         return (true);
+      }
    }
 
    /*******************************************************************************
     ** Use the event transaction, even when the caller did not put it in QContext.
+    ** Definitions are read once per event, not once per record.
     *******************************************************************************/
    public void onRecordsChanged(RecordChangeEvent event) throws QException
    {
+      if(OutboundSourceTables.DEFINITION_TABLES.contains(event.getTableName()))
+      {
+         EdiRuntime.current().outboundSources().invalidate(event.getTransaction());
+         return;
+      }
       var previous = QContext.getQBackendTransaction();
       QContext.setTransaction(event.getTransaction());
       try
       {
-         for(QRecord record : event.getRecords())
+         SystemWrites.call(() ->
          {
-            enqueue(event.getTableName(), record, "RECORD_CHANGE", null);
-         }
+            List<QRecord> definitions = definitions(event.getTableName(), OutboundSourceTables.TRIGGER);
+            for(QRecord record : event.getRecords())
+            {
+               enqueue(event.getTableName(), record, null, definitions);
+            }
+            return (null);
+         });
       }
       finally
       {
@@ -72,45 +118,60 @@ public class OutboundEnqueue implements RecordChangeListenerInterface
     *******************************************************************************/
    public static Integer enqueue(String table, QRecord source, String trigger, String eventId) throws QException
    {
-      return (SystemWrites.call(() ->
+      return (SystemWrites.call(() -> enqueue(table, source, eventId, definitions(table, trigger))));
+   }
+
+   /*******************************************************************************
+    ** Active outbound documents for the table whose trigger matches (MANUAL
+    ** matches every trigger) and whose agreement is active.  Called inside a
+    ** {@link SystemWrites#call}.
+    *******************************************************************************/
+   static List<QRecord> definitions(String table, String trigger) throws QException
+   {
+      List<QRecord> matching = new ArrayList<>();
+      for(QRecord document : EdiRecords.query("edi_agreement_document", EdiRecords.equal("sourceTableName", table)
+         .withCriteria("direction", QCriteriaOperator.EQUALS, "OUTBOUND").withCriteria("isActive", QCriteriaOperator.EQUALS, true)))
       {
-         int count = 0;
-         var definitions = EdiRecords.query("edi_agreement_document", EdiRecords.equal("sourceTableName", table)
-            .withCriteria("direction", QCriteriaOperator.EQUALS, "OUTBOUND").withCriteria("isActive", QCriteriaOperator.EQUALS, true));
-         for(QRecord document : definitions)
+         if("MANUAL".equals(trigger) || trigger.equals(document.getValueString("outboundTrigger")))
          {
-            if(!"MANUAL".equals(trigger) && !trigger.equals(document.getValueString("outboundTrigger")))
-            {
-               continue;
-            }
-            QRecord agreement = OutboundWorker.row("edi_agreement", document.getValueInteger("agreementId"));
-            if(!Boolean.TRUE.equals(agreement.getValueBoolean("isActive")))
-            {
-               continue;
-            }
-            var metadata = QContext.getQInstance().getTable(table);
-            if(!metadata.getBackendName().equals(EdiRuntime.current().config().getBackendName()))
-            {
-               throw new QException("Outbound source and EDI must share the transactional backend");
-            }
-            String filter = document.getValueString("triggerFilter");
-            if(filter != null && !filter.isBlank() && EdiRecords.query(table, EdiRecords.equal(metadata.getPrimaryKeyField(), source.getValue(metadata.getPrimaryKeyField()))
-               .withSubFilter(new ObjectMapper().readValue(filter, QQueryFilter.class))).isEmpty())
-            {
-               continue;
-            }
-            if(eventId != null && !EdiRecords.query("edi_outbound_request", EdiRecords.equal("eventId", eventId)
-               .withCriteria("agreementDocumentId", QCriteriaOperator.EQUALS, document.getValue("id"))).isEmpty())
-            {
-               continue;
-            }
-            SystemWrites.insert("edi_outbound_request", new QRecord().withValue("agreementDocumentId", document.getValue("id"))
-               .withValue("sourceTableName", table).withValue("sourceRecordId", source.getValueString(metadata.getPrimaryKeyField()))
-               .withValue("eventId", eventId).withValue("status", "PENDING").withValue("attempts", 0)
-               .withValue("nextAttemptAt", EdiRuntime.current().config().getClock().instant()));
-            count++;
+            matching.add(document);
          }
-         return (count);
-      }));
+      }
+      Set<Integer> activeAgreements = OutboundSourceTables.activeAgreementIds(matching);
+      matching.removeIf(document -> !activeAgreements.contains(document.getValueInteger("agreementId")));
+      return (matching);
+   }
+
+   /*******************************************************************************
+    ** Enqueue one source record against already-resolved definitions.
+    *******************************************************************************/
+   private static int enqueue(String table, QRecord source, String eventId, List<QRecord> definitions) throws Exception
+   {
+      int count = 0;
+      for(QRecord document : definitions)
+      {
+         var metadata = QContext.getQInstance().getTable(table);
+         if(!metadata.getBackendName().equals(EdiRuntime.current().config().getBackendName()))
+         {
+            throw new QException("Outbound source and EDI must share the transactional backend");
+         }
+         String filter = document.getValueString("triggerFilter");
+         if(filter != null && !filter.isBlank() && EdiRecords.query(table, EdiRecords.equal(metadata.getPrimaryKeyField(), source.getValue(metadata.getPrimaryKeyField()))
+            .withSubFilter(new ObjectMapper().readValue(filter, QQueryFilter.class))).isEmpty())
+         {
+            continue;
+         }
+         if(eventId != null && !EdiRecords.query("edi_outbound_request", EdiRecords.equal("eventId", eventId)
+            .withCriteria("agreementDocumentId", QCriteriaOperator.EQUALS, document.getValue("id"))).isEmpty())
+         {
+            continue;
+         }
+         SystemWrites.insert("edi_outbound_request", new QRecord().withValue("agreementDocumentId", document.getValue("id"))
+            .withValue("sourceTableName", table).withValue("sourceRecordId", source.getValueString(metadata.getPrimaryKeyField()))
+            .withValue("eventId", eventId).withValue("status", "PENDING").withValue("attempts", 0)
+            .withValue("nextAttemptAt", EdiRuntime.current().config().getClock().instant()));
+         count++;
+      }
+      return (count);
    }
 }

src/main/java/com/kingsrook/qbits/edi/flow/OutboundSourceTables.java

@@ -0,0 +1,183 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  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.edi.flow;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.WeakHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import com.kingsrook.qbits.edi.storage.EdiRecords;
+import com.kingsrook.qbits.edi.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+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.tables.query.QCriteriaOperator;
+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;
+
+/*******************************************************************************
+ ** Per-runtime cache of the source tables that have an active RECORD_CHANGE
+ ** outbound definition on an active agreement.  The record-change listener
+ ** consults it so writes to every other table (audits, jobs, alerts...) cost
+ ** nothing.  It is loaded from committed data only (never in a writer's
+ ** transaction).  A change to an agreement or document invalidates it, and
+ ** again once that change commits (discarding any reload that raced it); until
+ ** then the changing transaction bypasses the cache, so it sees its own
+ ** definitions.  A short TTL covers changes made by another process.
+ *******************************************************************************/
+public final class OutboundSourceTables
+{
+   static final String TRIGGER = "RECORD_CHANGE";
+   static final Set<String> DEFINITION_TABLES = Set.of("edi_agreement", "edi_agreement_document");
+
+   private static final Duration TTL = Duration.ofSeconds(60);
+
+   private final AtomicLong generation = new AtomicLong();
+   private final AtomicLong loads = new AtomicLong();
+   private final Set<QBackendTransaction> changingTransactions = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>()));
+   private volatile Snapshot snapshot;
+
+   /*******************************************************************************
+    ** Loaded table names, with the generation and time they were loaded at.
+    *******************************************************************************/
+   private record Snapshot(Set<String> tables, long generation, long loadedAtNanos)
+   {
+   }
+
+   /*******************************************************************************
+    ** Whether record changes to this table, written in this transaction (may be
+    ** null), can enqueue outbound work.
+    *******************************************************************************/
+   public boolean contains(String table, QBackendTransaction transaction) throws QException
+   {
+      if(transaction != null && changingTransactions.contains(transaction))
+      {
+         return (true);
+      }
+      Snapshot current = snapshot;
+      if(current == null || current.generation() != generation.get() || System.nanoTime() - current.loadedAtNanos() > TTL.toNanos())
+      {
+         current = load();
+      }
+      return (current.tables().contains(table));
+   }
+
+   /*******************************************************************************
+    ** Drop the cache now, and again once the changing transaction commits.
+    *******************************************************************************/
+   public void invalidate(QBackendTransaction transaction)
+   {
+      generation.incrementAndGet();
+      if(transaction != null)
+      {
+         changingTransactions.add(transaction);
+         transaction.addAfterCommitCallback(() ->
+         {
+            changingTransactions.remove(transaction);
+            generation.incrementAndGet();
+         });
+      }
+   }
+
+   /*******************************************************************************
+    ** How many times the definitions have been loaded (for tests and diagnostics).
+    *******************************************************************************/
+   public long loadCount()
+   {
+      return (loads.get());
+   }
+
+   /*******************************************************************************
+    ** Read committed active definitions once.  A load that raced an invalidation
+    ** is not kept as current, because it carries the older generation.
+    *******************************************************************************/
+   private Snapshot load() throws QException
+   {
+      long loadGeneration = generation.get();
+      Set<String> tables = SystemWrites.call(() ->
+      {
+         List<QRecord> documents = committed("edi_agreement_document", new QQueryFilter()
+            .withCriteria("direction", QCriteriaOperator.EQUALS, "OUTBOUND").withCriteria("isActive", QCriteriaOperator.EQUALS, true)
+            .withCriteria("outboundTrigger", QCriteriaOperator.EQUALS, TRIGGER));
+         Set<Integer> agreementIds = agreementIds(documents);
+         Set<Integer> activeAgreements = new HashSet<>();
+         if(!agreementIds.isEmpty())
+         {
+            committed("edi_agreement", new QQueryFilter().withCriteria("isActive", QCriteriaOperator.EQUALS, true)
+               .withCriteria("id", QCriteriaOperator.IN, agreementIds)).forEach(agreement -> activeAgreements.add(agreement.getValueInteger("id")));
+         }
+         Set<String> result = new HashSet<>();
+         for(QRecord document : documents)
+         {
+            if(document.getValueString("sourceTableName") != null && activeAgreements.contains(document.getValueInteger("agreementId")))
+            {
+               result.add(document.getValueString("sourceTableName"));
+            }
+         }
+         return (Set.copyOf(result));
+      });
+      loads.incrementAndGet();
+      Snapshot loaded = new Snapshot(tables, loadGeneration, System.nanoTime());
+      snapshot = loaded;
+      return (loaded);
+   }
+
+   /*******************************************************************************
+    ** Query committed rows - deliberately without the caller's transaction.
+    *******************************************************************************/
+   private static List<QRecord> committed(String table, QQueryFilter filter) throws QException
+   {
+      return (new QueryAction().execute(new QueryInput(table).withFilter(filter)).getRecords());
+   }
+
+   /*******************************************************************************
+    ** The distinct non-null agreement ids the documents reference.
+    *******************************************************************************/
+   private static Set<Integer> agreementIds(List<QRecord> documents)
+   {
+      Set<Integer> ids = new HashSet<>();
+      documents.forEach(document -> ids.add(document.getValueInteger("agreementId")));
+      ids.remove(null);
+      return (ids);
+   }
+
+   /*******************************************************************************
+    ** Active agreement ids among those the documents reference, in one query in
+    ** the current transaction.
+    *******************************************************************************/
+   static Set<Integer> activeAgreementIds(List<QRecord> documents) throws QException
+   {
+      Set<Integer> ids = agreementIds(documents);
+      Set<Integer> active = new HashSet<>();
+      if(!ids.isEmpty())
+      {
+         for(QRecord agreement : EdiRecords.query("edi_agreement", EdiRecords.equal("isActive", true).withCriteria("id", QCriteriaOperator.IN, ids)))
+         {
+            active.add(agreement.getValueInteger("id"));
+         }
+      }
+      return (active);
+   }
+}

src/main/resources/qbit-edi.properties

@@ -0,0 +1 @@
+version=${project.version}

src/test/java/com/kingsrook/qbits/edi/EdiHelpBundleTest.java

@@ -49,7 +49,7 @@ class EdiHelpBundleTest
       assertThat(HelpBundles.of(instance).getBundles()).singleElement().satisfies(bundle ->
       {
          assertThat(bundle.getId()).isEqualTo("qbit-edi");
-         assertThat(bundle.getVersion()).isEqualTo("0.1.0-SNAPSHOT");
+         assertThat(bundle.getVersion()).as("the help bundle carries the artifact version").isEqualTo(System.getProperty("qbit.artifactVersion")).isEqualTo(EdiQBitProducer.VERSION);
       });
       HelpBundles.syncOnStartup(instance);
       HelpBundleAssertions.forInstance(instance)

src/test/java/com/kingsrook/qbits/edi/EdiQBitProducerTest.java

@@ -65,6 +65,8 @@ class EdiQBitProducerTest
       assertThat(producer.getQBitConfig()).isSameAs(config);
       assertThat(producer.getNamespace()).isEqualTo("ns");
       assertThat(producer.getQBitMetaData().getArtifactId()).isEqualTo("qbit-edi");
+      assertThat(System.getProperty("qbit.artifactVersion")).isNotBlank();
+      assertThat(producer.getQBitMetaData().getVersion()).isEqualTo(System.getProperty("qbit.artifactVersion")).isEqualTo(EdiQBitProducer.VERSION);
       assertThat(producer.getPackageNameForFindingMetaDataProducers()).isEqualTo("com.kingsrook.qbits.edi.metadata");
       assertThat(config.getAppName()).isEqualTo("edi");
       assertThat(config.getAppLabel()).isEqualTo("EDI");

src/test/java/com/kingsrook/qbits/edi/OutboundSourceTablesTest.java

@@ -0,0 +1,213 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  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.edi;
+
+import java.util.List;
+import com.kingsrook.qbits.edi.flow.OutboundEnqueue;
+import com.kingsrook.qbits.edi.flow.OutboundSourceTables;
+import com.kingsrook.qbits.edi.storage.EdiRecords;
+import com.kingsrook.qbits.edi.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/*******************************************************************************
+ ** The record-change listener only costs queries for configured source tables,
+ ** and its cache follows agreement and document changes.
+ *******************************************************************************/
+class OutboundSourceTablesTest
+{
+   private QRecord document;
+   private QRecord agreement;
+
+   /*******************************************************************************
+    ** One active RECORD_CHANGE agreement document, for sample_order.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      H2Support.create();
+      RetailSupport.tables();
+      EdiRuntime.current().initialize();
+      SystemWrites.call(() ->
+      {
+         QRecord partner = SystemWrites.insert("edi_partner", new QRecord().withValue("name", "Retail test").withValue("code", "RETAIL").withValue("status", "ACTIVE"));
+         QRecord local = SystemWrites.insert("edi_identity", new QRecord().withValue("standard", "X12").withValue("interchangeQualifier", "ZZ").withValue("interchangeId", "QQQTEST").withValue("applicationCode", "QQQTEST"));
+         QRecord remote = SystemWrites.insert("edi_identity", new QRecord().withValue("partnerId", partner.getValue("id")).withValue("standard", "X12").withValue("interchangeQualifier", "ZZ").withValue("interchangeId", "SYNTHETIC").withValue("applicationCode", "SYNTHETIC"));
+         QRecord transport = SystemWrites.insert("edi_transport", new QRecord().withValue("name", "retail-out").withValue("type", "GATEWAY").withValue("direction", "BOTH").withValue("gatewayEndpointName", "retail-outbox").withValue("isActive", true).withValue("maxAttempts", 3));
+         agreement = SystemWrites.insert("edi_agreement", new QRecord().withValue("partnerId", partner.getValue("id")).withValue("localIdentityId", local.getValue("id")).withValue("partnerIdentityId", remote.getValue("id"))
+            .withValue("standard", "X12").withValue("usageIndicator", "T").withValue("isActive", true).withValue("outboundTransportId", transport.getValue("id")));
+         QRecord map = RetailSupport.installMap("855", "004010");
+         document = SystemWrites.insert("edi_agreement_document", outboundDocument(map, "sample_order"));
+         return (null);
+      });
+   }
+
+   /*******************************************************************************
+    ** Reset context.
+    *******************************************************************************/
+   @AfterEach
+   void clear()
+   {
+      QContext.clear();
+   }
+
+   /*******************************************************************************
+    ** Writes to a table without a definition neither apply nor reload definitions.
+    *******************************************************************************/
+   @Test
+   void unconfiguredTableDoesNoWork() throws Exception
+   {
+      OutboundEnqueue listener = new OutboundEnqueue();
+      assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue();
+      long loads = sources().loadCount();
+
+      assertThat(listener.appliesTo("sample_invoice", RecordChangeType.INSERT)).isFalse();
+      assertThat(listener.appliesTo("sample_invoice", RecordChangeType.UPDATE)).isFalse();
+      assertThat(listener.appliesTo("sample_order", RecordChangeType.DELETE)).isFalse();
+      SystemWrites.call(() -> SystemWrites.transaction(() ->
+      {
+         for(int i = 0; i < 5; i++)
+         {
+            RetailSupport.saveSource("sample_invoice", new QRecord().withValue("invoiceNumber", "INV-" + i));
+         }
+         return (null);
+      }));
+
+      assertThat(sources().loadCount()).isEqualTo(loads);
+      assertThat(rows("edi_outbound_request")).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Configured tables still enqueue, once per record, from one cached load.
+    *******************************************************************************/
+   @Test
+   void configuredTableEnqueues() throws Exception
+   {
+      SystemWrites.call(() -> SystemWrites.transaction(() -> RetailSupport.saveSource("sample_order", RetailSupport.source())));
+      long loads = sources().loadCount();
+      SystemWrites.call(() -> SystemWrites.transaction(() -> RetailSupport.saveSource("sample_order", RetailSupport.source().withValue("poNumber", "PO-2"))));
+      assertThat(sources().loadCount()).isEqualTo(loads);
+      assertThat(rows("edi_outbound_request")).hasSize(2);
+   }
+
+   /*******************************************************************************
+    ** Document and agreement changes invalidate the cached tables.
+    *******************************************************************************/
+   @Test
+   void definitionChangesInvalidate() throws Exception
+   {
+      OutboundEnqueue listener = new OutboundEnqueue();
+      assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue();
+      assertThat(listener.appliesTo("edi_agreement_document", RecordChangeType.DELETE)).isTrue();
+
+      update("edi_agreement_document", document.getValue("id"), "isActive", false);
+      assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isFalse();
+
+      update("edi_agreement_document", document.getValue("id"), "isActive", true);
+      assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue();
+
+      update("edi_agreement", agreement.getValue("id"), "isActive", false);
+      assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isFalse();
+      SystemWrites.call(() -> SystemWrites.transaction(() -> RetailSupport.saveSource("sample_order", RetailSupport.source())));
+      assertThat(rows("edi_outbound_request")).isEmpty();
+
+      update("edi_agreement", agreement.getValue("id"), "isActive", true);
+      assertThat(listener.appliesTo("sample_order", RecordChangeType.INSERT)).isTrue();
+   }
+
+   /*******************************************************************************
+    ** A transaction that adds a definition sees it before commit; a rolled-back
+    ** one leaves the cache on committed data.
+    *******************************************************************************/
+   @Test
+   void changingTransactionSeesItsOwnDefinitions() throws Exception
+   {
+      QRecord map = SystemWrites.call(() -> EdiRecords.query("edi_map", new QQueryFilter()).getFirst());
+      SystemWrites.call(() -> SystemWrites.transaction(() ->
+      {
+         SystemWrites.insert("edi_agreement_document", outboundDocument(map, "sample_invoice").withValue("transactionSet", "810"));
+         RetailSupport.saveSource("sample_invoice", new QRecord().withValue("invoiceNumber", "INV-1"));
+         return (null);
+      }));
+      assertThat(rows("edi_outbound_request")).hasSize(1);
+      assertThat(new OutboundEnqueue().appliesTo("sample_invoice", RecordChangeType.INSERT)).isTrue();
+
+      try
+      {
+         SystemWrites.call(() -> SystemWrites.transaction(() ->
+         {
+            SystemWrites.insert("edi_agreement_document", outboundDocument(map, "sample_inventory").withValue("transactionSet", "846"));
+            throw new IllegalStateException("rollback synthetic write");
+         }));
+      }
+      catch(Exception e)
+      {
+         ////////////////////////////
+         // expected - rolled back //
+         ////////////////////////////
+      }
+      assertThat(new OutboundEnqueue().appliesTo("sample_inventory", RecordChangeType.INSERT)).isFalse();
+   }
+
+   /*******************************************************************************
+    ** An active outbound RECORD_CHANGE document for the source table.
+    *******************************************************************************/
+   private QRecord outboundDocument(QRecord map, String sourceTable)
+   {
+      return (new QRecord().withValue("agreementId", agreement.getValue("id")).withValue("direction", "OUTBOUND").withValue("transactionSet", "855")
+         .withValue("version", "004010").withValue("functionalIdCode", "PR").withValue("guideId", map.getValue("guideId")).withValue("mapId", map.getValue("id"))
+         .withValue("sourceTableName", sourceTable).withValue("outboundTrigger", "RECORD_CHANGE").withValue("batchMode", "IMMEDIATE").withValue("ackExpectation", "997").withValue("ackDueMinutes", 1).withValue("isActive", true));
+   }
+
+   /*******************************************************************************
+    ** Update one field through QQQ, outside any transaction.
+    *******************************************************************************/
+   private static void update(String table, Object id, String field, Object value) throws Exception
+   {
+      SystemWrites.call(() ->
+      {
+         SystemWrites.update(table, new QRecord().withValue("id", (java.io.Serializable) id).withValue(field, (java.io.Serializable) value));
+         return (null);
+      });
+   }
+
+   /*******************************************************************************
+    ** The runtime cache under test.
+    *******************************************************************************/
+   private static OutboundSourceTables sources()
+   {
+      return (EdiRuntime.current().outboundSources());
+   }
+
+   /*******************************************************************************
+    ** All rows.
+    *******************************************************************************/
+   private static List<QRecord> rows(String table) throws Exception
+   {
+      return (SystemWrites.call(() -> EdiRecords.query(table, new QQueryFilter())));
+   }
+}