Ccd 4262 4769 create index table

charts/ccd-data-store-api/values.preview.template.yaml

@@ -222,7 +222,7 @@ logstash:
           jdbc_driver_library => "/usr/share/logstash/ccd/postgresql.jar"
           jdbc_driver_class => "org.postgresql.Driver"
           jdbc_default_timezone => "UTC"
-          statement => "UPDATE case_data SET marked_by_logstash = true WHERE marked_by_logstash = false RETURNING id, created_date, last_modified, jurisdiction, case_type_id, state, last_state_modified_date, data::TEXT as json_data, data_classification::TEXT as json_data_classification, reference, security_classification, supplementary_data::TEXT as json_supplementary_data"
+          statement => "DELETE FROM case_data_logstash_queue USING case_data WHERE case_data_logstash_queue.case_data_id = case_data.id RETURNING case_data.id, created_date, last_modified, jurisdiction, case_type_id, state, last_state_modified_date, data::TEXT as json_data, data_classification::TEXT as json_data_classification, reference, security_classification, supplementary_data::TEXT as json_supplementary_data, version"
           clean_run => false
           schedule => "* * * * * *"
         }
@@ -335,5 +335,6 @@ logstash:
               document_type => "_doc"
               document_id => "%{id}"
               timeout => 60
+              retry_on_conflict => 0
           }
       }

src/main/resources/db/migration/V20240130_4769__CCD-4769_create_case_data_logstash_queue.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS public.case_data_logstash_queue (
+    id serial PRIMARY KEY,
+    case_data_id bigint NOT NULL,
+    CONSTRAINT case_data_logstash_queue_case_data_id_fk FOREIGN KEY (case_data_id) REFERENCES public.case_data (id) ON DELETE CASCADE
+    DEFERRABLE INITIALLY DEFERRED
+);
+

src/main/resources/db/migration/V20240617_4775__CCD-4775_modify_db_trigger_for_IU.sql

@@ -0,0 +1,22 @@
+DROP TRIGGER IF EXISTS trg_case_data_updated ON public.case_data;
+
+DROP FUNCTION IF exists insert_update_logstash_queue;
+
+CREATE OR REPLACE FUNCTION public.insert_update_logstash_queue () RETURNS TRIGGER
+    LANGUAGE plpgsql
+    AS $$
+    BEGIN
+        INSERT INTO public.case_data_logstash_queue ("case_data_id") VALUES (NEW.id);
+        RETURN NEW;
+    END;
+    $$;
+
+DROP TRIGGER IF EXISTS trg_case_data_updated_queue ON public.case_data;
+
+CREATE TRIGGER trg_case_data_updated_queue
+BEFORE INSERT OR UPDATE OF data, data_classification, last_modified, last_state_modified_date, security_classification, state, supplementary_data
+ON public.case_data
+FOR EACH ROW EXECUTE FUNCTION public.insert_update_logstash_queue();
+
+
+

src/main/resources/db/migration/V20260702_0000__CCD-6936_skip_case_pointers_in_logstash_queue.sql

@@ -0,0 +1,19 @@
+-- Limit how long this migration waits for locks to ensure we do not hang if long queries are active.
+SET LOCAL lock_timeout = '15s';
+
+ALTER TABLE public.case_data
+    DROP CONSTRAINT IF EXISTS case_pointer_always_marked_by_logstash;
+
+CREATE OR REPLACE FUNCTION public.insert_update_logstash_queue () RETURNS TRIGGER
+    LANGUAGE plpgsql
+AS $$
+BEGIN
+    IF NOT (NEW.data = '{}'::jsonb AND NEW.state = '') THEN
+        INSERT INTO public.case_data_logstash_queue ("case_data_id") VALUES (NEW.id);
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+COMMENT ON FUNCTION public.insert_update_logstash_queue() IS
+    'Adds changed case rows to the Logstash queue, excluding case pointer rows.';

src/test/java/uk/gov/hmcts/ccd/AbstractBaseIntegrationTest.java

@@ -10,6 +10,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cache.CacheManager;
@@ -19,6 +20,7 @@
 import org.springframework.security.core.Authentication;
 import org.springframework.security.core.context.SecurityContext;
 import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
@@ -79,10 +81,30 @@
 @ExtendWith(SpringExtension.class)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 @TestPropertySource(locations = "classpath:test.properties")
+@DirtiesContext
 public abstract class AbstractBaseIntegrationTest {
     protected static final ObjectMapper mapper = JacksonUtils.MAPPER;
     protected static final Slf4jNotifier slf4jNotifier = new Slf4jNotifier(true);
 
+    private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(AbstractBaseIntegrationTest.class);
+
+    protected static final String dropTableFKConstraint =
+        "SELECT 'ALTER TABLE \"'||nspname||'\".\"'||relname||'\" DROP CONSTRAINT \"'||conname||'\" %s;'" 
+        + "FROM pg_constraint " 
+        + "INNER JOIN pg_class ON conrelid=pg_class.oid " 
+        + "INNER JOIN pg_namespace ON pg_namespace.oid=pg_class.relnamespace " 
+        + "WHERE relname = '%s'" 
+        + "ORDER BY CASE WHEN contype='f' THEN 0 ELSE 1 END,contype,nspname,relname,conname;";
+
+    protected static final String recreateTableFKConstraint =
+        "SELECT 'ALTER TABLE \"'||nspname||'\".\"'||relname||'\" ADD CONSTRAINT \"'||conname||'\" '|| " 
+        + "pg_get_constraintdef(pg_constraint.oid)||';' " 
+        + "FROM pg_constraint " 
+        + "INNER JOIN pg_class ON conrelid=pg_class.oid " 
+        + "INNER JOIN pg_namespace ON pg_namespace.oid=pg_class.relnamespace " 
+        + "WHERE relname = '%s'" 
+        + "ORDER BY CASE WHEN contype='f' THEN 0 ELSE 1 END DESC,contype DESC,nspname DESC,relname DESC,conname DESC;";
+
     protected static final MediaType JSON_CONTENT_TYPE = new MediaType(
         MediaType.APPLICATION_JSON.getType(),
         MediaType.APPLICATION_JSON.getSubtype(),
@@ -187,20 +209,66 @@ public void clearDownData() {
         List<String> tables = determineTables(jdbcTemplate);
         List<String> sequences = determineSequences(jdbcTemplate);
 
-        String truncateTablesQuery =
-            "START TRANSACTION;\n"
-                + tables.stream()
-                    .map(record -> String.format("TRUNCATE TABLE %s CASCADE;", record))
-                    .collect(Collectors.joining("\n"))
-                + "\nCOMMIT;";
-        jdbcTemplate.execute(truncateTablesQuery);
+        String dropTableFKConstraintString = null;
+        String recreateTableFKConstraintString = null;
+        String casecadeString = "";
+
+        for (String tableName : tables) {
+            casecadeString = "";
+            if (!tableName.equals("case_users")) {
+                casecadeString = "CASCADE";
+            }
+            dropTableFKConstraintString = getFKConstraintToDrop(casecadeString, tableName);
+            recreateTableFKConstraintString = getFKConstraintToRecreate(tableName);
+            String truncateTablesQuery =
+                String.format(
+                    "START TRANSACTION;\n"
+                        + dropTableFKConstraintString
+                        + "\nCOMMIT;\n"
+                        + "TRUNCATE TABLE %s CASCADE;\n"
+                        + "ALTER TABLE %s DISABLE TRIGGER ALL;\n"
+                        + "COMMIT;\n"
+                        + recreateTableFKConstraintString
+                        + "\nCOMMIT;\n "
+                        + "ALTER TABLE %s ENABLE TRIGGER ALL;\n"
+                        + "END TRANSACTION;\n",
+                    tableName, tableName, tableName);
+
+            LOG.info("Executing :" + truncateTablesQuery + "\n");
+
+            jdbcTemplate.execute(truncateTablesQuery);
+        }
 
         sequences.forEach(sequence -> jdbcTemplate.execute("ALTER SEQUENCE " + sequence + " RESTART WITH 1"));
 
         cacheManager.getCacheNames().forEach(
             cacheName -> Objects.requireNonNull(cacheManager.getCache(cacheName)).clear());
     }
 
+    private  String getFKConstraintToDrop(String tableName, String cascade) {
+        JdbcTemplate jdbcTemplate = new JdbcTemplate(db);
+
+        String formatedString = String.format(dropTableFKConstraint, tableName, cascade);
+
+        List result = jdbcTemplate.queryForList(formatedString, String.class);
+        String commaseparatedlist = result.toString().replace("[", "")
+            .replace("]", "")
+            .replace(";,", ";");
+        return commaseparatedlist;
+
+    }
+
+    private String getFKConstraintToRecreate(String tableName) {
+        JdbcTemplate jdbcTemplate = new JdbcTemplate(db);
+        String formatedString = String.format(recreateTableFKConstraint, tableName);
+        List result = jdbcTemplate.queryForList(formatedString, String.class);
+        String commaseparatedlist = result.toString().replace("[", "")
+            .replace("]", "")
+            .replace(";,", ";");
+        return commaseparatedlist;
+
+    }
+
     private List<String> determineTables(JdbcTemplate jdbcTemplate) {
         return jdbcTemplate.queryForList("SELECT * FROM pg_catalog.pg_tables").stream()
             .filter(tableInfo -> tableInfo.get("schemaname").equals("public"))

src/test/java/uk/gov/hmcts/ccd/LogstashPipelineConfigurationTest.java

@@ -0,0 +1,40 @@
+package uk.gov.hmcts.ccd;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class LogstashPipelineConfigurationTest {
+
+    private static final Path PREVIEW_VALUES =
+        Path.of("charts/ccd-data-store-api/values.preview.template.yaml");
+
+    @Test
+    void previewLogstashPipelineShouldUseElasticsearchManagedDocumentVersion() throws IOException {
+        String previewValues = Files.readString(PREVIEW_VALUES);
+
+        assertAll(
+            () -> assertTrue(
+                previewValues.contains("document_id => \"%{id}\""),
+                "Logstash output must keep stable document ids"
+            ),
+            () -> assertTrue(
+                !previewValues.contains("version_type => \"external\""),
+                "Preview Logstash output must let Elasticsearch manage document versions"
+            ),
+            () -> assertTrue(
+                !previewValues.contains("version => \"%{version}\""),
+                "Preview Logstash output must not use case_data.version as an external version"
+            ),
+            () -> assertTrue(
+                !previewValues.contains("version => \"%{[@metadata][version]}\""),
+                "Preview Logstash output must not use queue metadata as an external version"
+            )
+        );
+    }
+}

src/test/java/uk/gov/hmcts/ccd/data/caselinking/CaseLinkRepositoryTest.java

@@ -3,6 +3,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.jdbc.Sql;
 import uk.gov.hmcts.ccd.WireMockBaseTest;
 
@@ -22,6 +23,7 @@
 import static uk.gov.hmcts.ccd.data.caselinking.CaseLinkEntity.NON_STANDARD_LINK;
 import static uk.gov.hmcts.ccd.data.caselinking.CaseLinkEntity.STANDARD_LINK;
 
+@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_CLASS)
 class CaseLinkRepositoryTest extends WireMockBaseTest {
 
     private static final String TEST_ADDRESS_BOOK_CASE = "TestAddressBookCase";
@@ -42,6 +44,7 @@ void setup() {
         caseLinkIdToReferenceMap.put(CASE_04_ID, parseLong(CASE_04_REFERENCE));
         caseLinkIdToReferenceMap.put(CASE_13_ID, parseLong(CASE_13_REFERENCE));
         caseLinkIdToReferenceMap.put(CASE_14_ID, parseLong(CASE_14_REFERENCE));
+
     }
 
     @Test
@@ -114,6 +117,7 @@ void testFindCaseLinkEntityById() {
     }
 
     @Test
+    @DirtiesContext(methodMode = DirtiesContext.MethodMode.BEFORE_METHOD)
     @Sql(executionPhase = Sql.ExecutionPhase.BEFORE_TEST_METHOD, scripts = {"classpath:sql/insert_cases.sql"})
     void testSaveFailsIfCaseIdDoesNotExist() {
 
@@ -125,6 +129,7 @@ void testSaveFailsIfCaseIdDoesNotExist() {
     }
 
     @Test
+    @DirtiesContext(methodMode = DirtiesContext.MethodMode.BEFORE_METHOD)
     @Sql(executionPhase = Sql.ExecutionPhase.BEFORE_TEST_METHOD, scripts = {"classpath:sql/insert_cases.sql"})
     void testSaveFailsIfLinkedCaseIdDoesNotExist() {
 

src/test/java/uk/gov/hmcts/ccd/data/persistence/CasePointerRepositoryTest.java

@@ -1,13 +1,10 @@
 package uk.gov.hmcts.ccd.data.persistence;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.transaction.support.TransactionTemplate;
 import uk.gov.hmcts.ccd.WireMockBaseTest;
 import uk.gov.hmcts.ccd.data.casedetails.CaseDetailsRepository;
 import uk.gov.hmcts.ccd.data.casedetails.DefaultCaseDetailsRepository;
@@ -45,10 +42,8 @@ public class CasePointerRepositoryTest extends WireMockBaseTest {
 
     private CaseDetails originalCaseDetails;
     private Long currentCaseReference;
-    @Inject
-    private PlatformTransactionManager transactionManager;
 
-    @Before
+    @BeforeEach
     public void setUp() {
         originalCaseDetails = createOriginalCaseDetails();
     }
@@ -132,59 +127,31 @@ public void persistCasePointer_shouldRespectExistingResolvedTtl() {
     }
 
     @Test
-    public void persistCasePointer_shouldKeepMarkedByLogstashFlagTrue() {
+    public void persistCasePointer_shouldNotQueueCasePointerForLogstash() {
         casePointerRepository.persistCasePointerAndInitId(originalCaseDetails);
 
         JdbcTemplate jdbcTemplate = new JdbcTemplate(db);
-        Boolean markedByLogstash = jdbcTemplate.queryForObject(
-            "SELECT marked_by_logstash FROM case_data WHERE reference = ?",
-            Boolean.class,
-            currentCaseReference
+        Integer queueEntries = jdbcTemplate.queryForObject(
+            "SELECT count(*) FROM case_data_logstash_queue WHERE case_data_id = ?",
+            Integer.class,
+            Long.valueOf(originalCaseDetails.getId())
         );
 
-        assertThat(markedByLogstash, is(true));
+        assertThat(queueEntries, is(0));
     }
 
     @Test
-    public void persistRegularCase_shouldUnsetMarkedByLogstashFlag() {
+    public void persistRegularCase_shouldQueueForLogstash() {
         // Simulate a standard case creation by reusing the original details directly
         JdbcTemplate jdbcTemplate = new JdbcTemplate(db);
         var persisted = caseDetailsRepository.set(originalCaseDetails);
 
-        Boolean markedByLogstash = jdbcTemplate.queryForObject(
-            "SELECT marked_by_logstash FROM case_data WHERE id = ?",
-            Boolean.class,
+        Integer queueEntries = jdbcTemplate.queryForObject(
+            "SELECT count(*) FROM case_data_logstash_queue WHERE case_data_id = ?",
+            Integer.class,
             persisted.getId()
         );
 
-        assertThat(markedByLogstash, is(false));
-    }
-
-    @Test
-    public void constraintShouldPreventUnmarkingCasePointer() {
-        casePointerRepository.persistCasePointerAndInitId(originalCaseDetails);
-
-        JdbcTemplate jdbcTemplate = new JdbcTemplate(db);
-        Long pointerId = Long.valueOf(originalCaseDetails.getId());
-
-        // Run the raw update in its own transaction so an expected constraint failure does not poison
-        // the outer test transaction that the tests run in.
-        TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
-        transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
-
-        boolean constraintViolated = false;
-        try {
-            transactionTemplate.execute(status -> {
-                jdbcTemplate.update(
-                    "UPDATE case_data SET marked_by_logstash = false WHERE id = ?",
-                    pointerId
-                );
-                return null;
-            });
-        } catch (RuntimeException expected) {
-            constraintViolated = true;
-        }
-
-        assertThat("Constraint should prevent unmarking a case pointer", constraintViolated, is(true));
+        assertThat(queueEntries, is(1));
     }
 }

src/test/java/uk/gov/hmcts/ccd/integrations/CaseRoleDefinitionCachingIT.java

@@ -16,7 +16,6 @@
 import static org.mockito.Mockito.times;
 
 import static java.util.Collections.singletonList;
-
 import uk.gov.hmcts.ccd.WireMockBaseTest;
 
 @DirtiesContext

src/test/java/uk/gov/hmcts/ccd/integrations/DefinitionsCachingIT.java

@@ -51,6 +51,7 @@
 @SpringBootTest
 @AutoConfigureWireMock(port = 0)
 @TestPropertySource(locations = "classpath:test.properties")
+@DirtiesContext
 public class DefinitionsCachingIT {
 
     private static final String ID_1 = "case1";

src/test/java/uk/gov/hmcts/ccd/integrations/IdamRepositoryCachingIT.java

@@ -10,7 +10,6 @@
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.bean.override.mockito.MockitoBean;
 import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
-
 import uk.gov.hmcts.ccd.ApplicationParams;
 import uk.gov.hmcts.ccd.WireMockBaseTest;
 import uk.gov.hmcts.ccd.security.idam.IdamRepository;

src/test/resources/sql/insert_elasticsearch_cases.sql

@@ -26,4 +26,3 @@ VALUES (6, 'AAT', 'AUTOTEST1', 'IN_PROGRESS', 'PUBLIC', '{}', '{}', '15894600562
 
 INSERT INTO case_data (id, case_type_id, jurisdiction, state, security_classification, data, data_classification, reference)
 VALUES (7, 'MAPPER', 'AUTOTEST1', 'TODO', 'PUBLIC', '{}', '{}', '1588870615652827');
-