qbit-document-generation/foundation/src/test/java/com/kingsrook/qbits/documentgeneration/GenerationQueueJdbcTest.java

/*
 * 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.documentgeneration;

import java.nio.file.Files;
import java.nio.file.Path;
import java.sql.DriverManager;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.instances.QInstanceEnricher;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
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.permissions.PermissionLevel;
import com.kingsrook.qqq.backend.core.model.metadata.permissions.QPermissionRules;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.session.QUser;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
import com.kingsrook.qqq.backend.module.rdbms.jdbc.ConnectionManager;
import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData;
import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSTableBackendDetails;
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;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

/*******************************************************************************
 ** Real independent file-H2 transactions, restart and concurrency. Worker/storage
 ** ports are fixed trusted fixtures; no hostile code, containers or print devices.
 *******************************************************************************/
class GenerationQueueJdbcTest
{
   protected QInstance instance;
   protected DocumentGenerationQBitConfig config;
   protected GenerationQueue queue;
   protected Integer template;
   protected Integer stock;
   private String url;
   protected RDBMSBackendMetaData jdbcBackend;
   private final TestClock clock = new TestClock();
   private final AtomicInteger renders = new AtomicInteger();
   private final AtomicInteger retains = new AtomicInteger();
   private final Map<String, GenerationPorts.Receipt> receipts = new HashMap<>();
   private Runnable rendering = () -> {};
   private boolean failRetainedAudit;

   /***************************************************************************
    ** Create a synchronous file-backed database, native source authority and observable queue effect ports.
    ***************************************************************************/
   @BeforeEach
   void setup() throws Exception
   {
      ConnectionManager.closeConnectionProviders();
      MemoryRecordStore.getInstance().reset();
      Path directory = Files.createDirectories(Path.of("target", "queue-databases", UUID.randomUUID().toString())).toAbsolutePath();
      url = "jdbc:h2:file:" + directory.resolve("queue").toString().replace('\\', '/') + ";DATABASE_TO_UPPER=FALSE;WRITE_DELAY=0;LOCK_TIMEOUT=5000";
      instance = new QInstance().withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("anonymous").withType(QAuthenticationType.FULLY_ANONYMOUS));
      jdbcBackend = database("queue", url);
      instance.addBackend(jdbcBackend);
      instance.addBackend(new QBackendMetaData().withName("memory").withBackendType(MemoryBackendModule.class));
      instance.setEnforcePermissionsInCoreActions(true);
      instance.addTable(new QTableMetaData().withName("orders").withBackendName("memory").withPrimaryKeyField("id")
         .withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.READ_WRITE_PERMISSIONS))
         .withField(new QFieldMetaData("id", QFieldType.INTEGER)).withField(new QFieldMetaData("name", QFieldType.STRING)));
      config = new DocumentGenerationQBitConfig().withBackendName("queue").withAudit(event ->
      {
         if(failRetainedAudit && "GENERATION_RETAINED".equals(event.outcome()))
         {
            throw new QException("Injected audit outage");
         }
         new GenerationAudit().append(event);
      }).withRenderer(new GenerationPorts.IsolatedRenderer()
      {
         /***************************************************************************
          ** Accept every revision; queue tests focus on durable state transitions, not renderer validation.
          ***************************************************************************/
         public void validate(String actor, GenerationPorts.Revision revision) {}

         /***************************************************************************
          ** Count the render, run the scenario's hook (for example a competing claim), and return fixed trusted bytes.
          ***************************************************************************/
         public GenerationPorts.Content render(GenerationPorts.RenderRequest request)
         {
            renders.incrementAndGet();
            rendering.run();
            return (new GenerationPorts.Content("trusted fixture".getBytes(java.nio.charset.StandardCharsets.UTF_8)));
         }
      }).withIntake(new GenerationPorts.RetainedOutputIntake()
      {
         /***************************************************************************
          ** Count the retention and remember the receipt, so recovery tests can assert exactly-once intake across claims.
          ***************************************************************************/
         public GenerationPorts.Receipt retain(GenerationPorts.IntakeRequest request)
         {
            retains.incrementAndGet();
            var result = new GenerationPorts.Receipt("file", "version-" + request.requestId(), request.sha256(), (long)request.content().bytes().length, true, true);
            receipts.put(result.versionId(), result);
            return (result);
         }

         /***************************************************************************
          ** Return the remembered receipt, modelling an owner that still authorizes the retained version.
          ***************************************************************************/
         public GenerationPorts.Receipt authorize(String actor, GenerationPorts.Source source, String version)
         {
            return (receipts.get(version));
         }
      }).withPrinting(request -> "job-" + request.requestId());
      // Trusted test authority: production installs the exact retained-owner adapter.
      config.withRetainedAccess(reference ->
      {
         GenerationSecurity.source(reference.source());
         return (reference);
      });
      new DocumentGenerationQBitProducer().withQBitConfig(config).produce(instance);
      GenerationJobs.register(instance);
      for(var table : instance.getTables().values())
      {
         if(table.getName().startsWith("document_generation_"))
         {
            table.setBackendDetails(new RDBMSTableBackendDetails().withTableName(table.getName()));
         }
      }
      new QInstanceEnricher(instance).enrich();
      sql(schema(config));
      login("alice", true);
      new InsertAction().execute(new InsertInput("orders").withRecords(List.of(new QRecord().withValue("id", 1).withValue("name", "first"), new QRecord().withValue("id", 2).withValue("name", "second"))));
      var definitions = new DocumentDefinitionService(config);
      int type = definitions.createType("packing");
      stock = definitions.createStock("stock", new GenerationPorts.Stock(101600,152400,0,0,0,0,203));
      definitions.bind(type,"orders",true,true);
      template = definitions.createTemplate("packing",type,stock,"orders");
      definitions.saveDraft(template,stock,new DocumentDefinitionService.Draft("first template",null,null,List.of()));
      definitions.publish(template);
      queue = new GenerationQueue(config,clock,1,30);
   }

   /***************************************************************************
    ** Clear native context, pooled connections and memory records after each database scenario.
    ***************************************************************************/
   @AfterEach
   void cleanup()
   {
      QContext.clear();
      ConnectionManager.closeConnectionProviders();
      MemoryRecordStore.getInstance().reset();
   }

   /***************************************************************************
    ** Install the requested actor with generation rights and optional source-read permission.
    ***************************************************************************/
   private void login(String actor, boolean read)
   {
      QSession session = new QSession().withUser(new QUser().withIdReference(actor)).withPermissions("document-generation.view","document-generation.configure","document-generation.publish","document-generation.generate","document-generation.print","orders.write");
      if(read)
      {
         session.withPermissions("orders.read");
      }
      QContext.init(instance,session);
   }

   /***************************************************************************
    ** Refuse output effects when the database rejects the resolved filename write.
    ***************************************************************************/
   @Test
   void databaseRejectingFilenamePreventsRenderingAndIntake() throws Exception
   {
      sql("ALTER TABLE document_generation_request ADD CONSTRAINT reject_name CHECK (resolved_file_name IS NULL)");
      Integer id = enqueue("failed-name-write");
      assertThat(queue.advance(id).state()).isEqualTo("RECONCILIATION_REQUIRED");
      assertThat(renders.get()).isZero();
      assertThat(retains.get()).isZero();
      assertThat(value("SELECT resolved_file_name FROM document_generation_request WHERE kind='GENERATE'")).isNull();
   }

   /***************************************************************************
    ** Verify filename commit through an independent connection before render despite later draft publication.
    ***************************************************************************/
   @Test
   void filenameIsCommittedBeforeRenderAndDraftChangesDoNotAlterAcceptedPolicy() throws Exception
   {
      var definitions = new DocumentDefinitionService(config);
      definitions.saveOutputNaming(template, "Accepted name", null, definitions.loadDraft(template).fingerprint());
      definitions.publish(template);
      Integer id = enqueue("named-queue");
      definitions.saveOutputNaming(template, "Future name", null, definitions.loadDraft(template).fingerprint());
      definitions.publish(template);
      rendering = () ->
      {
         try
         {
            assertThat(value("SELECT resolved_file_name FROM document_generation_request WHERE kind='GENERATE'")).isEqualTo("Accepted name.pdf");
         }
         catch(Exception failure)
         {
            throw new IllegalStateException(failure);
         }
      };
      assertThat(queue.advance(id).state()).isEqualTo("GENERATED");
      assertThat(renders.get()).isEqualTo(1);
   }

   /***************************************************************************
    ** Refuse a tampered pinned filename policy before rendering or retaining output.
    ***************************************************************************/
   @Test
   void alteredPinnedFilenamePolicyCannotReachRenderer() throws Exception
   {
      var definitions = new DocumentDefinitionService(config);
      definitions.saveOutputNaming(template, "Original", null, definitions.loadDraft(template).fingerprint());
      int revision = definitions.publish(template);
      Integer id = enqueue("name-pin");
      sql("UPDATE document_generation_revision SET output_base_name='Changed' WHERE id=" + revision);
      assertThat(queue.advance(id).state()).isEqualTo("RECONCILIATION_REQUIRED");
      assertThat(renders.get()).isZero();
      assertThat(retains.get()).isZero();
   }

   /***************************************************************************
    ** Actual additive DDL preserves legacy batch identity and unknown completion;
    ** exact-selection receipt replay then marks it complete without new effects.
    ***************************************************************************/
   @Test
   void batchCompletionMigrationPreservesLegacyRowsAndAllowsSafeReplay() throws Exception
   {
      var service = new DocumentGenerationService(config);
      var selected = List.of(new GenerationPorts.Source("orders", "1"));
      assertThat(service.generateBatch(template, selected, GenerationPorts.Format.PDF, "legacy-batch", null).complete()).isTrue();
      String revision = value("SELECT revision_id FROM document_generation_batch WHERE request_key='legacy-batch'");
      String selection = value("SELECT selection_hash FROM document_generation_batch WHERE request_key='legacy-batch'");
      sql("ALTER TABLE document_generation_batch DROP COLUMN completed");
      String vendor = List.of("mysql", "postgres").contains(jdbcBackend.getVendor()) ? jdbcBackend.getVendor() : "h2";
      for(String statement : Files.readString(Path.of("src/main/resources/db/migrations/20261031-warm-batch-completion-" + vendor + ".sql")).split(";"))
      {
         if(!statement.isBlank())
         {
            sql(statement);
         }
      }
      assertThat(value("SELECT completed FROM document_generation_batch WHERE request_key='legacy-batch'")).isNull();
      assertThat(value("SELECT revision_id FROM document_generation_batch WHERE request_key='legacy-batch'")).isEqualTo(revision);
      assertThat(value("SELECT selection_hash FROM document_generation_batch WHERE request_key='legacy-batch'")).isEqualTo(selection);
      int before = renders.get();
      assertThat(service.generateBatch(template, selected, GenerationPorts.Format.PDF, "legacy-batch", null).complete()).isTrue();
      assertThat(renders.get()).isEqualTo(before);
      assertThat(value("SELECT completed FROM document_generation_batch WHERE request_key='legacy-batch'")).isNotNull();
   }

   /***************************************************************************
    ** Apply filename migrations to existing queue rows while preserving pinned revision hashes.
    ***************************************************************************/
   @Test
   void explicitFilenameMigrationPreservesOldRowsAndQueueHashes() throws Exception
   {
      Integer id = enqueue("migration-name");
      String hash = value("SELECT revision_hash FROM document_generation_queue WHERE id=" + id);
      for(String table : List.of("document_generation_template", "document_generation_revision"))
      {
         sql("ALTER TABLE " + table + " DROP COLUMN output_base_name");
         sql("ALTER TABLE " + table + " DROP COLUMN output_name_context_key");
      }
      sql("ALTER TABLE document_generation_request DROP COLUMN resolved_file_name");
      String vendor = List.of("mysql", "postgres").contains(jdbcBackend.getVendor()) ? jdbcBackend.getVendor() : "h2";
      for(String statement : Files.readString(Path.of("src/main/resources/db/migrations/20261031-output-filenames-" + vendor + ".sql")).split(";"))
      {
         if(!statement.isBlank())
         {
            sql(statement);
         }
      }
      assertThat(value("SELECT revision_hash FROM document_generation_queue WHERE id=" + id)).isEqualTo(hash);
      assertThat(new DocumentDefinitionService(config).published(template).outputBaseName()).isNull();
      assertThat(queue.advance(id).state()).isEqualTo("GENERATED");
      assertThat(value("SELECT resolved_file_name FROM document_generation_request WHERE kind='GENERATE'")).startsWith("generated-revision-").endsWith(".pdf");
   }

   /***************************************************************************
    ** Resume an old four-field asset snapshot using compatible font defaults without changing its hash.
    ***************************************************************************/
   @Test
   void legacyAssetSnapshotStillAdvancesAfterFontMetadataUpgrade() throws Exception
   {
      var definitions = new DocumentDefinitionService(config);
      Integer image = definitions.addAsset(template, new GenerationPorts.Asset("logo", "image", "a".repeat(64), "image/png"));
      Integer font = definitions.addAsset(template, new GenerationPorts.Asset("Regular", "font", "b".repeat(64), "font/ttf"));
      definitions.saveDraft(template, stock, new DocumentDefinitionService.Draft("legacy", null, null, List.of(image, font)));
      Integer revision = definitions.publish(template);
      Integer id = queue.enqueue(template, List.of(new GenerationPorts.Source("orders", "1")), GenerationPorts.Format.PDF, "old-shape", null);
      var mapper = new com.fasterxml.jackson.databind.ObjectMapper().enable(com.fasterxml.jackson.databind.SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS);
      com.fasterxml.jackson.databind.node.ObjectNode legacy = mapper.valueToTree(definitions.revision(revision));
      legacy.remove(List.of("outputBaseName", "outputNameContextKey"));
      for(var asset : legacy.path("assets"))
      {
         ((com.fasterxml.jackson.databind.node.ObjectNode) asset).remove(List.of("fontFamily", "fontWeight", "fontStyle"));
      }
      String hash = java.util.HexFormat.of().formatHex(java.security.MessageDigest.getInstance("SHA-256").digest(mapper.writeValueAsBytes(legacy)));
      sql("UPDATE document_generation_queue SET revision_hash='" + hash + "' WHERE id=" + id);
      assertThat(queue.advance(id).state()).isEqualTo("GENERATED");
      assertThat(renders.get()).isEqualTo(1);
   }

   /***************************************************************************
    ** Explicit face semantics cannot be altered behind an accepted immutable pin.
    ***************************************************************************/
   @Test
   void explicitFontFaceRemainsInQueuedRevisionHash() throws Exception
   {
      var definitions = new DocumentDefinitionService(config);
      Integer font = definitions.addAsset(template, new GenerationPorts.Asset("Bold", "font", "b".repeat(64), "font/ttf", "Warehouse", 700, "italic"));
      definitions.saveDraft(template, stock, new DocumentDefinitionService.Draft("face", null, null, List.of(font)));
      definitions.publish(template);
      Integer id = queue.enqueue(template, List.of(new GenerationPorts.Source("orders", "1")), GenerationPorts.Format.PDF, "font-pin", null);
      sql("UPDATE document_generation_asset SET font_weight=800 WHERE id=" + font);
      assertThat(definitions.published(template).assets().getFirst().fontWeight()).isEqualTo(800);
      GenerationQueue.Item refused = queue.advance(id);
      assertThat(refused.state()).isEqualTo("RECONCILIATION_REQUIRED");
      assertThat(refused.errorCode()).isEqualTo("GENERATION_INCOMPLETE");
      assertThat(refused.stage()).isEqualTo("CLAIMED");
      assertThat(refused.requestId()).isNull();
      assertThat(queue.status(id, -1, 1).items()).containsExactly(refused);
      assertThat(queue.advance(id)).isNull();
      assertThat(renders.get()).isZero();
      assertThat(retains.get()).isZero();
   }

   /*******************************************************************************
    ** Accepted print work is terminal for generation and cannot be claimed again.
    *******************************************************************************/
   @Test
   void printQueuedItemIsNeverUntouchedQueuedWork() throws Exception
   {
      int id = queue.enqueue(template, List.of(new GenerationPorts.Source("orders", "1")), GenerationPorts.Format.PDF, "print-once", "printer");
      assertThat(queue.advance(id).state()).isEqualTo("PRINT_QUEUED");
      assertThat(queue.advance(id)).isNull();
      assertThat(renders).hasValue(1);
      assertThat(retains).hasValue(1);
   }

   /*******************************************************************************
    ** Queue/history recovery uses exact retained authority after source deletion;
    ** successful reconciliation neither renders nor stores the original again.
    *******************************************************************************/
   @Test
   void archivedQueueHistoryAndKnownVersionReconciliationUseOwnerAuthority() throws Exception
   {
      int id = queue.enqueue(template, List.of(new GenerationPorts.Source("orders", "1")), GenerationPorts.Format.PDF, "archive-queue", null);
      failRetainedAudit = true;
      var failed = queue.advance(id);
      assertThat(failed.state()).isEqualTo("RECONCILIATION_REQUIRED");
      config.withRetainedAccess(reference ->
      {
         var receipt = receipts.get(reference.versionId());
         assertThat(receipt).isNotNull();
         assertThat(reference.fileId()).isEqualTo(receipt.fileId());
         assertThat(reference.sha256()).isEqualTo(receipt.sha256());
         assertThat(reference.length()).isEqualTo(receipt.length());
         return (reference);
      });
      new com.kingsrook.qqq.backend.core.actions.tables.DeleteAction().execute(
         new com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput("orders").withPrimaryKeys(List.of(1)));
      assertThat(queue.status(id, -1, 50).items()).hasSize(1);
      assertThat(new com.kingsrook.qqq.backend.core.actions.tables.QueryAction().execute(
         new com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput(GenerationTables.ITEM)).getRecords()).hasSize(1);
      failRetainedAudit = false;
      assertThat(queue.reconcile(id, failed.id()).state()).isEqualTo("GENERATED");
      assertThat(renders).hasValue(1);
      assertThat(retains).hasValue(1);
   }

   /*******************************************************************************
    ** A core job step drains untouched items, preserving domain outcomes on rerun.
    *******************************************************************************/
   @Test
   void jobDrainUsesExistingQueueAndNeverRepeatsCompletedItems() throws Exception
   {
      queue = new GenerationQueue(config);
      int id = queue.enqueue(template, List.of(new GenerationPorts.Source("orders", "1"), new GenerationPorts.Source("orders", "2")), GenerationPorts.Format.PDF, "job-drain", null);
      var input = new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput();
      input.setProcessName(GenerationJobs.PROCESS);
      input.addValue("queueId", id);
      var output = new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput();
      new GenerationJobs.Drain().run(input, output);
      assertThat(output.getValueInteger("processedItems")).isEqualTo(2);
      new GenerationJobs.Drain().run(input, new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput());
      assertThat(renders).hasValue(2);
      assertThat(retains).hasValue(2);
      assertThatThrownBy(() -> GenerationJobs.launch(id)).hasMessageContaining("not configured");
      assertThatThrownBy(() -> GenerationJobs.register(instance)).hasMessageContaining("already exists");
   }

   /*******************************************************************************
    ** Cooperative cancellation after render prevents intake and leaves review state.
    ** The try-with-resources scope is never referenced on purpose: opening it
    ** installs the thread-bound cancellation check the queue must observe, so only
    ** this method suppresses javac's [try] lint.
    *******************************************************************************/
   @Test
   @SuppressWarnings("try")
   void jobCancellationFencesIntakeAndUntouchedItemsStayQueued() throws Exception
   {
      queue = new GenerationQueue(config);
      int id = queue.enqueue(template, List.of(new GenerationPorts.Source("orders", "1"), new GenerationPorts.Source("orders", "2")), GenerationPorts.Format.PDF, "job-cancel", null);
      java.util.concurrent.atomic.AtomicBoolean stopped = new java.util.concurrent.atomic.AtomicBoolean();
      rendering = () -> stopped.set(true);
      try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, () ->
      {
         if(stopped.get())
         {
            throw new com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.Reason.CANCEL);
         }
      }))
      {
         assertThatThrownBy(() -> queue.advance(id)).isInstanceOf(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.class);
      }
      assertThat(retains).hasValue(0);
      assertThat(queue.status(id, -1, 50).items()).extracting(GenerationQueue.Item::state).containsExactly("RECONCILIATION_REQUIRED", "QUEUED");
      assertThat(queue.advance(id).state()).isEqualTo("GENERATED");
      var input = new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput();
      input.setProcessName(GenerationJobs.PROCESS);
      input.addValue("queueId", id);
      assertThatThrownBy(() -> new GenerationJobs.Drain().run(input, new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput()))
         .hasMessageContaining("incomplete");
      assertThat(renders).hasValue(2);
   }

   /*******************************************************************************
    ** Launch uses public core persistence under the initiating identity. No worker
    ** runtime is started; this fixture does not claim production job-host coverage.
    *******************************************************************************/
   @Test
   void nativeBackgroundActionLaunchesThroughCoreKernel() throws Exception
   {
      queue = new GenerationQueue(config);
      int id = queue.enqueue(template, List.of(new GenerationPorts.Source("orders", "1")), GenerationPorts.Format.PDF, "job-launch", null);
      new com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaDataProvider().defineAll(instance, "memory", null);
      instance.withSupplementalMetaData(new com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaData());
      new QInstanceEnricher(instance).enrich();
      login("alice", true);
      var input = new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput();
      input.setProcessName(GenerationProcesses.name(GenerationTables.QUEUE, "background"));
      input.addValue("selectionJson", "[\"" + id + "\"]");
      var output = new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput();
      new GenerationProcesses.Execute().run(input, output);
      int jobId = Integer.parseInt(output.getValueString("resultJson"));
      QRecord job = com.kingsrook.qqq.backend.core.jobs.DurableJobs.status(jobId);
      assertThat(job.getValueString("processName")).isEqualTo(GenerationJobs.PROCESS);
      assertThat(job.getValueString("userId")).isEqualTo("alice");
      assertThat(job.getValueString("idempotency")).isEqualTo("NONE");
      assertThat(job.getValueString("paramsJson")).contains("queueId");
      assertThat(renders).hasValue(0);
      login("mallory", true);
      assertThatThrownBy(() -> GenerationJobs.launch(id)).isInstanceOf(QException.class);
      login("alice", false);
      assertThatThrownBy(() -> GenerationJobs.launch(id)).isInstanceOf(QException.class);
   }

   /***************************************************************************
    ** Enqueue the two fixture orders as one pinned PDF selection under the supplied idempotency key.
    ***************************************************************************/
   protected Integer enqueue(String key) throws QException
   {
      return (queue.enqueue(template,List.of(new GenerationPorts.Source("orders","1"),new GenerationPorts.Source("orders","2")),GenerationPorts.Format.PDF,key,null));
   }

   /***************************************************************************
    ** Persist selection and context snapshots across publication changes, retries and connection reopen.
    ***************************************************************************/
   @Test
   void durablePinnedSelectionAndPreparedSnapshot() throws Exception
   {
      int id=enqueue("same");
      int revision=queue.status(id,-1,1).revisionId();
      var definitions=new DocumentDefinitionService(config);
      definitions.saveDraft(template,stock,new DocumentDefinitionService.Draft("new template",null,null,List.of()));
      definitions.publish(template);
      assertThat(enqueue("same")).isEqualTo(id);
      assertThatThrownBy(() -> queue.enqueue(template,List.of(new GenerationPorts.Source("orders","2")),GenerationPorts.Format.PDF,"same",null)).hasMessageContaining("another selection");
      var first=queue.advance(id);
      assertThat(first.state()).isEqualTo("GENERATED");
      assertThat(first.contextHash()).matches("[a-f0-9]{64}");
      assertThat(value("SELECT context_json FROM document_generation_item WHERE id="+first.id())).contains("first");
      assertThat(value("SELECT revision_id FROM document_generation_request WHERE id="+first.requestId())).isEqualTo(String.valueOf(revision));
      ConnectionManager.closeConnectionProviders();
      queue=new GenerationQueue(config,clock,1,30);
      assertThat(queue.advance(id).ordinal()).isEqualTo(1);
      assertThat(queue.advance(id)).isNull();
      assertThat(renders.get()).isEqualTo(2);
      assertThat(queue.status(id,-1,1).nextCursor()).isEqualTo(0);
      assertThat(queue.status(id,0,1).nextCursor()).isNull();
   }

   /***************************************************************************
    ** Roll back the queue and every selection item when one item cannot be inserted.
    ***************************************************************************/
   @Test
   void selectionRollsBackCompletelyWhenAnItemCannotCommit() throws Exception
   {
      sql("ALTER TABLE document_generation_item ADD CONSTRAINT reject_second CHECK(record_id <> '2')");
      assertThatThrownBy(() -> enqueue("rollback")).isInstanceOf(QException.class);
      assertThat(value("SELECT COUNT(*) FROM document_generation_queue")).isEqualTo("0");
      assertThat(value("SELECT COUNT(*) FROM document_generation_item")).isEqualTo("0");
   }

   /***************************************************************************
    ** Require the accepting actor and current source access for status, cancellation and queue advancement.
    ***************************************************************************/
   @Test
   void currentSourcePermissionsAndActorAreRequired() throws Exception
   {
      int id=enqueue("permission");
      login("bob",true);
      assertThatThrownBy(() -> queue.status(id,-1,10)).isInstanceOf(QException.class);
      assertThatThrownBy(() -> queue.cancel(id)).isInstanceOf(QException.class);
      login("alice",false);
      assertThatThrownBy(() -> queue.status(id,-1,10)).isInstanceOf(QException.class);
      assertThat(queue.advance(id).state()).isEqualTo("RECONCILIATION_REQUIRED");
      assertThat(renders.get()).isZero();
   }

   /***************************************************************************
    ** Fence retention when cancellation occurs during rendering and cancel untouched selection items.
    ***************************************************************************/
   @Test
   void cancellationFencesRetentionAndCancelsUntouchedItems() throws Exception
   {
      int id=enqueue("cancel");
      rendering=() ->
      {
         try
         {
            queue.cancel(id);
         }
         catch(QException e)
         {
            throw new IllegalStateException(e);
         }
      };
      assertThat(queue.advance(id).state()).isEqualTo("RECONCILIATION_REQUIRED");
      assertThat(retains.get()).isZero();
      assertThat(queue.status(id,-1,10).items().get(1).state()).isEqualTo("CANCELLED");
      assertThat(queue.advance(id)).isNull();
   }

   /***************************************************************************
    ** Require explicit termination evidence after lease expiry and avoid repeating unknown intake effects.
    ***************************************************************************/
   @Test
   void expiredCrashRequiresTerminationEvidenceAndNeverRerendersUnknownIntake() throws Exception
   {
      int id=enqueue("crash");
      rendering=() ->
      {
         throw new AssertionError("Synthetic abrupt process death");
      };
      assertThatThrownBy(() -> queue.advance(id)).isInstanceOf(AssertionError.class);
      clock.now=clock.now.plusSeconds(31);
      var abandoned=queue.status(id,-1,10).items().getFirst();
      assertThat(abandoned.state()).isEqualTo("RECONCILIATION_REQUIRED");
      assertThatThrownBy(() -> queue.advance(id)).hasMessageContaining("admission");
      assertThatThrownBy(() -> queue.recoverTerminated(id,abandoned.id())).hasMessageContaining("termination evidence unavailable");
      config.withAttemptTermination((item,token) -> assertThat(token).matches("[a-f0-9-]{36}"));
      queue.recoverTerminated(id,abandoned.id());
      assertThat(queue.reconcile(id,abandoned.id()).state()).isEqualTo("RECONCILIATION_REQUIRED");
      rendering=() -> {};
      assertThat(queue.advance(id).ordinal()).isEqualTo(1);
      assertThat(renders.get()).isEqualTo(2);
      assertThat(retains.get()).isEqualTo(1);
   }

   /***************************************************************************
    ** Reconcile a known retained version after audit recovery without a second render or intake.
    ***************************************************************************/
   @Test
   void knownVersionReconcilesWithoutSecondRenderOrIntake() throws Exception
   {
      int id=enqueue("retained");
      failRetainedAudit=true;
      var failed=queue.advance(id);
      assertThat(failed.state()).isEqualTo("RECONCILIATION_REQUIRED");
      failRetainedAudit=false;
      assertThat(queue.reconcile(id,failed.id()).state()).isEqualTo("GENERATED");
      assertThat(renders.get()).isEqualTo(1);
      assertThat(retains.get()).isEqualTo(1);
   }

   /***************************************************************************
    ** Serialize concurrent idempotent submission and deny a second worker while admission is occupied.
    ***************************************************************************/
   @Test
   void crossThreadAdmissionAndIdempotentSubmissionAreSerialized() throws Exception
   {
      try(var executor=Executors.newFixedThreadPool(2))
      {
         var left=executor.submit(() ->
         {
            login("alice",true); try
            {
               return (enqueue("concurrent"));
            }
            finally
            {
               QContext.clear();
            }
         });
         var right=executor.submit(() ->
         {
            login("alice",true); try
            {
               return (enqueue("concurrent"));
            }
            finally
            {
               QContext.clear();
            }
         });
         int id=left.get(10,TimeUnit.SECONDS);
         assertThat(right.get(10,TimeUnit.SECONDS)).isEqualTo(id);
         CountDownLatch entered=new CountDownLatch(1);
         CountDownLatch release=new CountDownLatch(1);
         rendering=() ->
         {
            entered.countDown();
            try
            {
               if(!release.await(10,TimeUnit.SECONDS))
               {
                  throw new IllegalStateException("fixture timeout");
               }
            }
            catch(InterruptedException e)
            {
               Thread.currentThread().interrupt();
               throw new IllegalStateException(e);
            }
         };
         var work=executor.submit(() ->
         {
            login("alice",true); try
            {
               return (queue.advance(id));
            }
            finally
            {
               QContext.clear();
            }
         });
         assertThat(entered.await(10,TimeUnit.SECONDS)).isTrue();
         try
         {
            assertThatThrownBy(() -> queue.advance(id)).hasMessageContaining("admission");
         }
         finally
         {
            release.countDown();
         }
         assertThat(work.get(10,TimeUnit.SECONDS).state()).isEqualTo("GENERATED");
      }
   }

   /***************************************************************************
    ** Reject invalid page, lease and selection bounds plus conflicting persisted admission settings.
    ***************************************************************************/
   @Test
   void rejectsInvalidBoundsAndInconsistentOperatorAdmission() throws Exception
   {
      assertThatThrownBy(() -> new GenerationQueue(config,clock,0,30)).isInstanceOf(IllegalArgumentException.class);
      assertThatThrownBy(() -> new GenerationQueue(config,clock,1,601)).isInstanceOf(IllegalArgumentException.class);
      assertThatThrownBy(() -> queue.enqueue(template,List.of(),GenerationPorts.Format.PDF,"empty",null)).isInstanceOf(QException.class);
      int id=enqueue("bounds");
      assertThatThrownBy(() -> queue.status(id,-2,1)).isInstanceOf(QException.class);
      assertThatThrownBy(() -> queue.status(id,-1,51)).isInstanceOf(QException.class);
      queue.advance(id);
      assertThatThrownBy(() -> new GenerationQueue(config,clock,2,30).advance(id)).hasMessageContaining("persisted backend admission");
      assertThatThrownBy(() -> queue.reconcile(id,999)).isInstanceOf(QException.class);
      assertThatThrownBy(() -> queue.recoverTerminated(id,queue.status(id,-1,1).items().getFirst().id())).isInstanceOf(QException.class);
   }

   /***************************************************************************
    ** Execute fixture DDL or corruption setup through an independent database connection.
    ***************************************************************************/
   protected void sql(String sql) throws Exception
   {
      try(var connection=connection();var statement=connection.createStatement())
      {
         statement.execute(sql);
      }
   }

   /*******************************************************************************
    ** Global pending admission is serialized and bounded independently of workers.
    *******************************************************************************/
   @Test
   void boundsQueuedBacklogBeforeAnySelectionRowsAreInserted() throws Exception
   {
      sql("INSERT INTO document_generation_item (state) SELECT 'QUEUED' FROM SYSTEM_RANGE(1,10000)");
      assertThatThrownBy(() -> enqueue("full")).hasMessageContaining("backlog is full");
      assertThat(value("SELECT COUNT(*) FROM document_generation_queue")).isEqualTo("0");
   }

   /*******************************************************************************
    ** Read-only status remains useful while every semantic queue mutation denies.
    *******************************************************************************/
   @Test
   void queueJdbcCannotBypassCredentialMutationCeiling() throws Exception
   {
      int id=enqueue("readonly");
      QContext.getQSession().withCredentialScope(new com.kingsrook.qqq.backend.core.model.session.QCredentialScope().withReadOnly(true));
      assertThat(queue.status(id,-1,10).items()).hasSize(2);
      assertThatThrownBy(() -> queue.cancel(id)).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
      assertThatThrownBy(() -> queue.advance(id)).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
      assertThatThrownBy(() -> enqueue("blocked")).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
      assertThat(value("SELECT CASE WHEN cancel_requested THEN 'TRUE' ELSE 'FALSE' END FROM document_generation_queue WHERE id="+id)).isEqualTo("FALSE");
      assertThat(renders.get()).isZero();
   }

   /***************************************************************************
    ** Read one scalar through an independent connection to observe committed database state.
    ***************************************************************************/
   private String value(String sql) throws Exception
   {
      try(var connection=connection();var statement=connection.createStatement();var rows=statement.executeQuery(sql))
      {
         rows.next();
         return (rows.getString(1));
      }
   }

   /***************************************************************************
    ** Database fixture seam; inherited tests also run against actual MySQL.
    ***************************************************************************/
   protected RDBMSBackendMetaData database(String name, String defaultUrl) throws Exception
   {
      return (new RDBMSBackendMetaData().withName(name).withVendor("h2").withJdbcUrl(defaultUrl).withUsername("sa"));
   }

   /***************************************************************************
    ** Independent fixture connection; never production credentials or services.
    ***************************************************************************/
   protected java.sql.Connection connection() throws Exception
   {
      return (DriverManager.getConnection(jdbcBackend.getJdbcUrl(), jdbcBackend.getUsername(), jdbcBackend.getPassword()));
   }

   /***************************************************************************
    ** Bootstrap matches the selected actual database dialect.
    ***************************************************************************/
   protected String schema(DocumentGenerationQBitConfig configuration)
   {
      return (DocumentGenerationSchema.postgresScript(configuration).replace("TIMESTAMPTZ", "TIMESTAMP"));
   }

   /***************************************************************************
    ** Provide a deterministic UTC clock whose instant can advance to exercise lease expiry.
    ***************************************************************************/
   private static class TestClock extends Clock
   {
      private Instant now=Instant.parse("2026-10-30T00:00:00Z");

      /***************************************************************************
       ** Keep fixture timestamps in UTC so lease assertions do not depend on the host zone.
       ***************************************************************************/
      public ZoneId getZone()
      {
         return (ZoneId.of("UTC"));
      }

      /***************************************************************************
       ** Retain the same deterministic clock when a caller requests a zone view.
       ***************************************************************************/
      public Clock withZone(ZoneId zone)
      {
         return (this);
      }

      /***************************************************************************
       ** Return the test-controlled instant used for queue lease and expiry decisions.
       ***************************************************************************/
      public Instant instant()
      {
         return (now);
      }
   }
}