diff --git a/charts/ccd-data-store-api/values.preview.template.yaml b/charts/ccd-data-store-api/values.preview.template.yaml index f8c1854b9d..c74fe2e0cc 100644 --- a/charts/ccd-data-store-api/values.preview.template.yaml +++ b/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 } } diff --git a/src/main/resources/db/migration/V20240130_4769__CCD-4769_create_case_data_logstash_queue.sql b/src/main/resources/db/migration/V20240130_4769__CCD-4769_create_case_data_logstash_queue.sql new file mode 100644 index 0000000000..c859cfefff --- /dev/null +++ b/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 +); + diff --git a/src/main/resources/db/migration/V20240617_4775__CCD-4775_modify_db_trigger_for_IU.sql b/src/main/resources/db/migration/V20240617_4775__CCD-4775_modify_db_trigger_for_IU.sql new file mode 100644 index 0000000000..04cf1b1d5d --- /dev/null +++ b/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(); + + + diff --git a/src/main/resources/db/migration/V20260702_0000__CCD-6936_skip_case_pointers_in_logstash_queue.sql b/src/main/resources/db/migration/V20260702_0000__CCD-6936_skip_case_pointers_in_logstash_queue.sql new file mode 100644 index 0000000000..fe51a4c9e8 --- /dev/null +++ b/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.'; diff --git a/src/test/java/uk/gov/hmcts/ccd/AbstractBaseIntegrationTest.java b/src/test/java/uk/gov/hmcts/ccd/AbstractBaseIntegrationTest.java index 2dd8266989..214540adca 100644 --- a/src/test/java/uk/gov/hmcts/ccd/AbstractBaseIntegrationTest.java +++ b/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,13 +209,35 @@ public void clearDownData() { List tables = determineTables(jdbcTemplate); List 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")); @@ -201,6 +245,30 @@ public void clearDownData() { 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 determineTables(JdbcTemplate jdbcTemplate) { return jdbcTemplate.queryForList("SELECT * FROM pg_catalog.pg_tables").stream() .filter(tableInfo -> tableInfo.get("schemaname").equals("public")) diff --git a/src/test/java/uk/gov/hmcts/ccd/LogstashPipelineConfigurationTest.java b/src/test/java/uk/gov/hmcts/ccd/LogstashPipelineConfigurationTest.java new file mode 100644 index 0000000000..4aa746a9bd --- /dev/null +++ b/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" + ) + ); + } +} diff --git a/src/test/java/uk/gov/hmcts/ccd/data/caselinking/CaseLinkRepositoryTest.java b/src/test/java/uk/gov/hmcts/ccd/data/caselinking/CaseLinkRepositoryTest.java index 71a207e21d..6ae60f8da4 100644 --- a/src/test/java/uk/gov/hmcts/ccd/data/caselinking/CaseLinkRepositoryTest.java +++ b/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() { diff --git a/src/test/java/uk/gov/hmcts/ccd/data/persistence/CasePointerRepositoryTest.java b/src/test/java/uk/gov/hmcts/ccd/data/persistence/CasePointerRepositoryTest.java index 1152534d68..314d74150f 100644 --- a/src/test/java/uk/gov/hmcts/ccd/data/persistence/CasePointerRepositoryTest.java +++ b/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)); } } diff --git a/src/test/java/uk/gov/hmcts/ccd/integrations/CaseRoleDefinitionCachingIT.java b/src/test/java/uk/gov/hmcts/ccd/integrations/CaseRoleDefinitionCachingIT.java index 4b6041fe0d..dd75c9660b 100644 --- a/src/test/java/uk/gov/hmcts/ccd/integrations/CaseRoleDefinitionCachingIT.java +++ b/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 diff --git a/src/test/java/uk/gov/hmcts/ccd/integrations/DefinitionsCachingIT.java b/src/test/java/uk/gov/hmcts/ccd/integrations/DefinitionsCachingIT.java index 077b7a64a8..b90620d656 100644 --- a/src/test/java/uk/gov/hmcts/ccd/integrations/DefinitionsCachingIT.java +++ b/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"; diff --git a/src/test/java/uk/gov/hmcts/ccd/integrations/IdamRepositoryCachingIT.java b/src/test/java/uk/gov/hmcts/ccd/integrations/IdamRepositoryCachingIT.java index 2c24a8fe75..e14b436e8d 100644 --- a/src/test/java/uk/gov/hmcts/ccd/integrations/IdamRepositoryCachingIT.java +++ b/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; diff --git a/src/test/resources/sql/insert_elasticsearch_cases.sql b/src/test/resources/sql/insert_elasticsearch_cases.sql index 65fd0cc498..4df73bc1fd 100644 --- a/src/test/resources/sql/insert_elasticsearch_cases.sql +++ b/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'); -