MODSOURMAN-1044 - Adjust logs during marc-to-marc matching on central tenant

NEWS.md

@@ -9,6 +9,7 @@
 * [MODSOURMAN-971](https://issues.folio.org/browse/MODSOURMAN-971) Adjust journal records population to create multiple journal records for each Holdings/Item
 * [MODSOURMAN-1014](https://issues.folio.org/browse/MODSOURMAN-1014) Upgrade folio-kafka-wrapper to 3.0.0 version
 * [MODDATAIMP-866](https://issues.folio.org/browse/MODDATAIMP-866) Add composite job types to support DI splitting workflow (bump interface `source-manager-job-executions` to version `3.3`)
+* [MODSOURMAN-1044](https://issues.folio.org/browse/MODSOURMAN-1044) Adjust logs during marc-to-marc matching on central tenant
 
 ## 2023-03-xo v3.6.1-SNAPSHOT
 * [MODSOURMAN-957](https://issues.folio.org/browse/MODSOURMAN-957) The '1' number of SRS MARC and Instance are displayed in cells in the row with the 'Updated' row header at the individual import job's log

mod-source-record-manager-server/pom.xml

@@ -196,7 +196,7 @@
     <dependency>
       <groupId>org.folio</groupId>
       <artifactId>folio-kafka-wrapper</artifactId>
-      <version>3.0.0-SNAPSHOT</version>
+      <version>3.0.0</version>
     </dependency>
     <dependency>
       <groupId>org.folio</groupId>
@@ -206,7 +206,7 @@
     <dependency>
       <groupId>org.folio</groupId>
       <artifactId>data-import-processing-core</artifactId>
-      <version>4.1.0-SNAPSHOT</version>
+      <version>4.1.0</version>
     </dependency>
     <dependency>
       <groupId>org.folio</groupId>

mod-source-record-manager-server/src/main/java/org/folio/dao/JournalRecordDaoImpl.java

@@ -68,6 +68,7 @@
 import static org.folio.dao.util.JournalRecordsColumns.INSTANCE_ENTITY_ERROR;
 import static org.folio.dao.util.JournalRecordsColumns.INSTANCE_ENTITY_HRID;
 import static org.folio.dao.util.JournalRecordsColumns.INSTANCE_ENTITY_ID;
+import static org.folio.dao.util.JournalRecordsColumns.INSTANCE_ENTITY_TENANT_ID;
 import static org.folio.dao.util.JournalRecordsColumns.INSTANCE_ID;
 import static org.folio.dao.util.JournalRecordsColumns.INVOICE_ACTION_STATUS;
 import static org.folio.dao.util.JournalRecordsColumns.INVOICE_ENTITY_ERROR;
@@ -88,6 +89,8 @@
 import static org.folio.dao.util.JournalRecordsColumns.SOURCE_ID;
 import static org.folio.dao.util.JournalRecordsColumns.SOURCE_RECORD_ACTION_STATUS;
 import static org.folio.dao.util.JournalRecordsColumns.SOURCE_RECORD_ORDER;
+import static org.folio.dao.util.JournalRecordsColumns.SOURCE_RECORD_TENANT_ID;
+import static org.folio.dao.util.JournalRecordsColumns.TENANT_ID;
 import static org.folio.dao.util.JournalRecordsColumns.TITLE;
 import static org.folio.dao.util.JournalRecordsColumns.TOTAL_AUTHORITIES_ERRORS;
 import static org.folio.dao.util.JournalRecordsColumns.TOTAL_COUNT;
@@ -139,7 +142,7 @@ public class JournalRecordDaoImpl implements JournalRecordDao {
     "instance_action_status", "holdings_action_status", "item_action_status", "order_action_status", "invoice_action_status", "error");
 
   private static final String JOURNAL_RECORDS_TABLE = "journal_records";
-  private static final String INSERT_SQL = "INSERT INTO %s.%s (id, job_execution_id, source_id, source_record_order, entity_type, entity_id, entity_hrid, action_type, action_status, error, action_date, title, instance_id, holdings_id, order_id, permanent_location_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)";
+  private static final String INSERT_SQL = "INSERT INTO %s.%s (id, job_execution_id, source_id, source_record_order, entity_type, entity_id, entity_hrid, action_type, action_status, error, action_date, title, instance_id, holdings_id, order_id, permanent_location_id, tenant_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)";
   private static final String SELECT_BY_JOB_EXECUTION_ID_QUERY = "SELECT * FROM %s.%s WHERE job_execution_id = $1";
   private static final String ORDER_BY_PATTERN = " ORDER BY %s %s";
   private static final String DELETE_BY_JOB_EXECUTION_ID_QUERY = "DELETE FROM %s.%s WHERE job_execution_id = $1";
@@ -200,7 +203,8 @@ private Tuple prepareInsertQueryParameters(JournalRecord journalRecord) {
       journalRecord.getInstanceId(),
       journalRecord.getHoldingsId(),
       journalRecord.getOrderId(),
-      journalRecord.getPermanentLocationId());
+      journalRecord.getPermanentLocationId(),
+      journalRecord.getTenantId());
   }
 
   @Override
@@ -281,11 +285,11 @@ public Future<Integer> updateErrorJournalRecordsByOrderIdAndJobExecution(String
 
   private List<JournalRecord> mapResultSetToJournalRecordsList(RowSet<Row> resultSet) {
     List<JournalRecord> journalRecords = new ArrayList<>();
-    resultSet.forEach(row -> journalRecords.add(mapRowJsonToJournalRecord(row)));
+    resultSet.forEach(row -> journalRecords.add(mapRowToJournalRecord(row)));
     return journalRecords;
   }
 
-  private JournalRecord mapRowJsonToJournalRecord(Row row) {
+  private JournalRecord mapRowToJournalRecord(Row row) {
     return new JournalRecord()
       .withId(row.getValue(ID).toString())
       .withJobExecutionId(row.getValue(JOB_EXECUTION_ID).toString())
@@ -301,7 +305,8 @@ private JournalRecord mapRowJsonToJournalRecord(Row row) {
       .withActionType(ActionType.valueOf(row.getString(ACTION_TYPE)))
       .withActionStatus(ActionStatus.valueOf(row.getString(ACTION_STATUS)))
       .withError(row.getString(ERROR))
-      .withActionDate(Date.from(LocalDateTime.parse(row.getValue(ACTION_DATE).toString()).toInstant(ZoneOffset.UTC)));
+      .withActionDate(Date.from(LocalDateTime.parse(row.getValue(ACTION_DATE).toString()).toInstant(ZoneOffset.UTC)))
+      .withTenantId(row.getString(TENANT_ID));
   }
 
   private String prepareSortingClause(String sortBy, String order) {
@@ -377,8 +382,8 @@ private RecordProcessingLogDto mapRowSetToRecordProcessingLogDto(RowSet<Row> res
         .withSourceRecordTitle(row.getString(TITLE))
         .withSourceRecordActionStatus(mapNameToEntityActionStatus(row.getString(SOURCE_RECORD_ACTION_STATUS)))
         .withError(row.getString(SOURCE_ENTITY_ERROR))
-        .withRelatedInstanceInfo(constructProcessedEntityWithSingleIdInfoBasedOnEntityType(row,
-          INSTANCE_ACTION_STATUS, INSTANCE_ENTITY_ID, INSTANCE_ENTITY_HRID, INSTANCE_ENTITY_ERROR))
+        .withSourceRecordTenantId(row.getString(SOURCE_RECORD_TENANT_ID))
+        .withRelatedInstanceInfo(constructInstanceProcessingInfo(row))
         .withRelatedAuthorityInfo(constructProcessedEntityWithSingleIdInfoBasedOnEntityType(row,
           AUTHORITY_ACTION_STATUS, AUTHORITY_ENTITY_ID, null, AUTHORITY_ENTITY_ERROR))
         .withRelatedPoLineInfo(new RelatedPoLineInfo()
@@ -417,6 +422,12 @@ private ProcessedEntityInfo constructProcessedEntityInfoBasedOnEntityType(Row ro
       .withError(row.getString(error));
   }
 
+  private ProcessedEntityInfo constructInstanceProcessingInfo(Row row) {
+    return constructProcessedEntityWithSingleIdInfoBasedOnEntityType(row, INSTANCE_ACTION_STATUS, INSTANCE_ENTITY_ID,
+      INSTANCE_ENTITY_HRID, INSTANCE_ENTITY_ERROR)
+      .withTenantId(row.getString(INSTANCE_ENTITY_TENANT_ID));
+  }
+
   private ProcessedEntityInfo constructProcessedEntityWithSingleIdInfoBasedOnEntityType(Row row, String actionStatus, String id, String hrid, String error) {
     return new ProcessedEntityInfo()
       .withActionStatus(mapNameToEntityActionStatus(row.getString(actionStatus)))

mod-source-record-manager-server/src/main/java/org/folio/dao/util/JournalRecordsColumns.java

@@ -18,6 +18,7 @@ public final class JournalRecordsColumns {
   public static final String ACTION_STATUS = "action_status";
   public static final String ACTION_DATE = "action_date";
   public static final String ERROR = "error";
+  public static final String TENANT_ID = "tenant_id";
 
   public static final String SOURCE_RECORD_ACTION_STATUS = "source_record_action_status";
   public static final String INSTANCE_ACTION_STATUS = "instance_action_status";
@@ -31,9 +32,11 @@ public final class JournalRecordsColumns {
   public static final String TOTAL_COUNT = "total_count";
 
   public static final String SOURCE_ENTITY_ERROR = "source_entity_error";
+  public static final String SOURCE_RECORD_TENANT_ID = "source_record_tenant_id";
   public static final String INSTANCE_ENTITY_ID = "instance_entity_id";
   public static final String INSTANCE_ENTITY_HRID = "instance_entity_hrid";
   public static final String INSTANCE_ENTITY_ERROR = "instance_entity_error";
+  public static final String INSTANCE_ENTITY_TENANT_ID = "instance_entity_tenant_id";
   public static final String HOLDINGS_ENTITY_ID = "holdings_entity_id";
   public static final String HOLDINGS_ENTITY_HRID = "holdings_entity_hrid";
   public static final String HOLDINGS_PERMANENT_LOCATION_ID = "holdings_permanent_location_id";

mod-source-record-manager-server/src/main/java/org/folio/services/journal/JournalUtil.java

@@ -49,6 +49,7 @@ public class JournalUtil {
   public static final String HRID_KEY = "hrid";
   private static final String NOT_MATCHED_NUMBER = "NOT_MATCHED_NUMBER";
   public static final String PERMANENT_LOCATION_ID_KEY = "permanentLocationId";
+  private static final String CENTRAL_TENANT_ID_KEY = "CENTRAL_TENANT_ID";
 
   private JournalUtil() {
 
@@ -86,7 +87,10 @@ record = new ObjectMapper().readValue(recordAsString, Record.class);
         .withEntityType(entityType)
         .withActionType(actionType)
         .withActionDate(new Date())
-        .withActionStatus(actionStatus);
+        .withActionStatus(actionStatus)
+        // tenantId field is filled in only for the case when record/entity has been changed on central tenant
+        // by data import initiated on a member tenant
+        .withTenantId(eventPayload.getContext().get(CENTRAL_TENANT_ID_KEY));
 
       if (DI_ERROR == DataImportEventTypes.fromValue(eventPayload.getEventType())) {
         journalRecord.setError(eventPayloadContext.get(ERROR_KEY));

mod-source-record-manager-server/src/main/resources/templates/db_scripts/create_get_record_processing_log_function.sql

@@ -1,18 +1,18 @@
 DROP FUNCTION IF EXISTS get_record_processing_log(uuid, uuid);
 
 CREATE OR REPLACE FUNCTION get_record_processing_log(jobExecutionId uuid, recordId uuid)
-    RETURNS TABLE(job_execution_id uuid, source_id uuid, source_record_order integer, title text, source_record_action_status text, source_entity_error text, instance_action_status text, instance_entity_id text, instance_entity_hrid text, instance_entity_error text, holdings_action_status text, holdings_entity_hrid text, holdings_entity_id text, holdings_permanent_location_id text, holdings_entity_error text, item_action_status text, item_entity_hrid text, item_entity_id text, item_entity_error text, authority_action_status text, authority_entity_id text, authority_entity_error text, po_line_action_status text, po_line_entity_id text, po_line_entity_hrid text, po_line_entity_error text, order_entity_id text, invoice_action_status text, invoice_entity_id text[], invoice_entity_hrid text[], invoice_entity_error text, invoice_line_action_status text, invoice_line_entity_id text, invoice_line_entity_hrid text, invoice_line_entity_error text)
+    RETURNS TABLE(job_execution_id uuid, source_id uuid, source_record_order integer, title text, source_record_action_status text, source_entity_error text, source_record_tenant_id text, instance_action_status text, instance_entity_id text, instance_entity_hrid text, instance_entity_error text, instance_entity_tenant_id text, holdings_action_status text, holdings_entity_hrid text, holdings_entity_id text, holdings_permanent_location_id text, holdings_entity_error text, item_action_status text, item_entity_hrid text, item_entity_id text, item_entity_error text, authority_action_status text, authority_entity_id text, authority_entity_error text, po_line_action_status text, po_line_entity_id text, po_line_entity_hrid text, po_line_entity_error text, order_entity_id text, invoice_action_status text, invoice_entity_id text[], invoice_entity_hrid text[], invoice_entity_error text, invoice_line_action_status text, invoice_line_entity_id text, invoice_line_entity_hrid text, invoice_line_entity_error text)
 AS $$
 BEGIN
     RETURN QUERY
         WITH temp_result AS (SELECT id, journal_records.job_execution_id, journal_records.source_id, journal_records.entity_type, journal_records.entity_id, journal_records.entity_hrid,
     					 CASE WHEN error_max != ''  OR action_type = 'NON_MATCH'
                     		THEN 'DISCARDED'
-                  		WHEN action_type = 'CREATE'
+                    WHEN action_type = 'CREATE'
                     		THEN 'CREATED'
-                  		WHEN action_type IN ('UPDATE', 'MODIFY')
+                    WHEN action_type IN ('UPDATE', 'MODIFY')
                     		THEN 'UPDATED'
-                		END AS action_type, journal_records.action_status, journal_records.action_date, journal_records.source_record_order, journal_records.error, journal_records.title, journal_records.instance_id, journal_records.holdings_id, journal_records.order_id, journal_records.permanent_location_id
+               END AS action_type, journal_records.action_status, journal_records.action_date, journal_records.source_record_order, journal_records.error, journal_records.title, journal_records.tenant_id, journal_records.instance_id, journal_records.holdings_id, journal_records.order_id, journal_records.permanent_location_id
     		FROM journal_records
     		INNER JOIN
     		(SELECT entity_type as entity_type_max, entity_id as entity_id_max,action_status as action_status_max, max(error) AS error_max,(array_agg(id ORDER BY array_position(array['CREATE', 'UPDATE', 'MODIFY', 'NON_MATCH'], action_type)))[1] AS id_max
@@ -27,11 +27,13 @@ BEGIN
     	      COALESCE(marc.title,instances.title,holdings.title,items.title) AS title,
     	      marc.action_type AS source_record_action_status,
     	      marc.error AS source_entity_error,
+            marc.tenant_id AS source_record_tenant_id,
 
     	      instances.action_type AS instance_action_status,
     	      COALESCE(instances.entity_id,holdings.instance_id,items.instance_id) AS instance_entity_id,
     	      instances.entity_hrid AS instance_entity_hrid,
     	      instances.error AS instance_entity_error,
+    	      instances.tenant_id AS instance_entity_tenant_id,
 
     	      holdings.action_type AS holdings_action_status,
 			      holdings.entity_hrid AS holdings_entity_hrid,
@@ -63,10 +65,10 @@ BEGIN
             null AS invoice_line_entity_hrid,
             null AS invoice_line_entity_error
       FROM
-    	    (SELECT temp_result.job_execution_id, entity_id, temp_result.title, temp_result.source_record_order, action_type, error, temp_result.source_id
+    	    (SELECT temp_result.job_execution_id, entity_id, temp_result.title, temp_result.source_record_order, action_type, error, temp_result.source_id, temp_result.tenant_id
     	    FROM temp_result WHERE entity_type IN ('MARC_BIBLIOGRAPHIC', 'MARC_HOLDINGS', 'MARC_AUTHORITY')) AS marc
     	LEFT JOIN
-    	    (SELECT action_type, entity_id, temp_result.source_id, entity_hrid, error, temp_result.job_execution_id, temp_result.title, temp_result.source_record_order
+    	    (SELECT action_type, entity_id, temp_result.source_id, entity_hrid, error, temp_result.job_execution_id, temp_result.title, temp_result.source_record_order, temp_result.tenant_id
     	    FROM temp_result WHERE entity_type = 'INSTANCE') AS instances
     	ON marc.source_id = instances.source_id
     	LEFT JOIN
@@ -95,10 +97,12 @@ BEGIN
                   WHEN edifact_actions[array_length(edifact_actions, 1)] = 'CREATE' THEN 'CREATED'
                 END AS source_record_action_status,
                 records_actions.source_record_error[1],
+                records_actions.source_record_tenant_id,
                 null AS instance_action_status,
                 null AS instance_entity_id,
                 null AS instance_entity_hrid,
                 null AS instance_entity_error,
+                null AS instance_entity_tenant_id,
                 null AS holdings_action_status,
                 null AS holdings_entity_hrid,
                 null AS holdings_entity_id,
@@ -144,6 +148,7 @@ BEGIN
                        array_agg(action_type) FILTER (WHERE entity_type = 'EDIFACT') AS edifact_actions,
                        count(journal_records.source_id) FILTER (WHERE entity_type = 'EDIFACT' AND journal_records.error != '') AS edifact_errors_number,
                        array_agg(error) FILTER (WHERE entity_type = 'EDIFACT') AS source_record_error,
+                       journal_records.tenant_id AS source_record_tenant_id,
 
                        array_agg(action_type) FILTER (WHERE entity_type = 'INVOICE' AND journal_records.title = 'INVOICE') AS invoice_actions,
                        count(journal_records.source_id) FILTER (WHERE entity_type = 'INVOICE' AND journal_records.title = 'INVOICE' AND journal_records.error != '') AS invoice_errors_number,
@@ -152,7 +157,7 @@ BEGIN
                        array_agg(error) FILTER (WHERE entity_type = 'INVOICE' AND journal_records.title = 'INVOICE') AS invoice_entity_error
                 FROM journal_records
                 WHERE journal_records.source_id = invoice_line_info.source_id AND (entity_type = 'EDIFACT' OR journal_records.title = 'INVOICE')
-                GROUP BY journal_records.source_id, journal_records.job_execution_id, journal_records.source_record_order
+                GROUP BY journal_records.source_id, journal_records.job_execution_id,journal_records.source_record_order, journal_records.tenant_id
             ) AS records_actions ON TRUE;
 END;
 $$ LANGUAGE plpgsql;

mod-source-record-manager-server/src/main/resources/templates/db_scripts/schema.json

@@ -271,6 +271,11 @@
       "run": "after",
       "snippetPath": "alter_job_executions_table_for_total_records.sql",
       "fromModuleVersion": "mod-source-record-manager-3.7.0"
+    },
+    {
+      "run": "after",
+      "snippet": "ALTER TABLE journal_records ADD COLUMN IF NOT EXISTS tenant_id text;",
+      "fromModuleVersion": "mod-source-record-manager-3.7.0"
     }
   ]
 }