MODINVOSTO-216. Add old object for EDIT audit events for Invoice, Invoice Line

src/main/java/org/folio/dao/lines/InvoiceLinesDAO.java

@@ -11,6 +11,8 @@ public interface InvoiceLinesDAO {
 
   Future<List<InvoiceLine>> getInvoiceLines(Criterion criterion, Conn conn);
 
+  Future<InvoiceLine> getInvoiceLineByIdForUpdate(String id, Conn conn);
+
   Future<String> createInvoiceLine(InvoiceLine invoiceLine, Conn conn);
 
   Future<Void> updateInvoiceLine(String id, InvoiceLine invoiceLine, Conn conn);

src/main/java/org/folio/dao/lines/InvoiceLinesPostgresDAO.java

@@ -1,6 +1,7 @@
 package org.folio.dao.lines;
 
 import io.vertx.core.Future;
+import io.vertx.ext.web.handler.HttpException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.folio.dao.DbUtils;
@@ -11,6 +12,7 @@
 
 import java.util.List;
 
+import static javax.ws.rs.core.Response.Status.NOT_FOUND;
 import static org.folio.rest.impl.InvoiceStorageImpl.INVOICE_LINE_TABLE;
 import static org.folio.rest.utils.ResponseUtils.convertPgExceptionIfNeeded;
 
@@ -26,6 +28,18 @@ public Future<List<InvoiceLine>> getInvoiceLines(Criterion criterion, Conn conn)
       .onFailure(t -> log.error("Failed to get invoice lines with criterion: {}", criterion, t));
   }
 
+  @Override
+  public Future<InvoiceLine> getInvoiceLineByIdForUpdate(String id, Conn conn) {
+    return conn.getByIdForUpdate(INVOICE_LINE_TABLE, id, InvoiceLine.class)
+      .map(invoiceLine -> {
+        if (invoiceLine == null) {
+          throw new HttpException(NOT_FOUND.getStatusCode(), NOT_FOUND.getReasonPhrase());
+        }
+        return invoiceLine;
+      })
+      .onFailure(t -> log.error("getInvoiceLineByIdForUpdate failed for invoice line with id {}", id, t));
+  }
+
   @Override
   public Future<String> createInvoiceLine(InvoiceLine invoiceLine, Conn conn) {
     log.trace("createInvoiceLine:: Creating invoice line: {}", invoiceLine);

src/main/java/org/folio/service/InvoiceLineStorageService.java

@@ -55,8 +55,9 @@ public void updateInvoiceLine(String id, InvoiceLine invoiceLine, Map<String, St
       asyncResultHandler.handle(buildBadRequestResponse("Invoice line id is required"));
     }
     new DBClient(vertxContext, headers).getPgClient()
-      .withTrans(conn -> invoiceLinesDAO.updateInvoiceLine(id, invoiceLine, conn)
-        .compose(invoiceLineId -> auditOutboxService.saveInvoiceLineOutboxLog(conn, invoiceLine, InvoiceLineAuditEvent.Action.EDIT, headers)))
+      .withTrans(conn -> invoiceLinesDAO.getInvoiceLineByIdForUpdate(id, conn)
+        .compose(original -> invoiceLinesDAO.updateInvoiceLine(id, invoiceLine, conn)
+          .compose(v -> auditOutboxService.saveInvoiceLineOutboxLog(conn, invoiceLine, original, InvoiceLineAuditEvent.Action.EDIT, headers))))
       .onSuccess(s -> {
         log.info("updateInvoiceLine:: Successfully updated invoice line with id: {}", id);
         auditOutboxService.processOutboxEventLogs(headers, vertxContext);

src/main/java/org/folio/service/InvoiceStorageService.java

@@ -68,8 +68,9 @@ public void putInvoiceStorageInvoicesById(String id, Invoice invoice, Map<String
       asyncResultHandler.handle(buildBadRequestResponse("Invoice id is required"));
     }
     new DBClient(vertxContext, headers).getPgClient()
-      .withTrans(conn -> invoiceDAO.updateInvoice(id, invoice, conn)
-        .compose(invoiceId -> auditOutboxService.saveInvoiceOutboxLog(conn, invoice, InvoiceAuditEvent.Action.EDIT, headers)))
+      .withTrans(conn -> invoiceDAO.getInvoiceByIdForUpdate(id, conn)
+        .compose(original -> invoiceDAO.updateInvoice(id, invoice, conn)
+          .compose(v -> auditOutboxService.saveInvoiceOutboxLog(conn, invoice, original, InvoiceAuditEvent.Action.EDIT, headers))))
       .onSuccess(s -> {
         log.info("putInvoiceStorageInvoicesById:: Successfully updated invoice with id: {}", id);
         auditOutboxService.processOutboxEventLogs(headers, vertxContext);

src/main/java/org/folio/service/audit/AuditEntityWrapper.java

@@ -0,0 +1,18 @@
+package org.folio.service.audit;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class AuditEntityWrapper<T> {
+
+  private T entity;
+  private T originalEntity;
+
+  public static <T> AuditEntityWrapper<T> of(T entity, T originalEntity) {
+    return new AuditEntityWrapper<>(entity, originalEntity);
+  }
+}

src/main/java/org/folio/service/audit/AuditEventProducer.java

@@ -32,13 +32,14 @@ public class AuditEventProducer {
    * Sends event for invoice change(Create, Edit) to kafka.
    * InvoiceId is used as partition key to send all events for particular invoice to the same partition.
    *
-   * @param invoice      the event payload
-   * @param eventAction  the event action
-   * @param okapiHeaders the okapi headers
+   * @param invoice         the event payload (post-edit state)
+   * @param originalInvoice the pre-edit invoice state; null for Create
+   * @param eventAction     the event action
+   * @param okapiHeaders    the okapi headers
    * @return future with true if sending was success or failed future in another case
    */
-  public Future<Void> sendInvoiceEvent(Invoice invoice, InvoiceAuditEvent.Action eventAction, Map<String, String> okapiHeaders) {
-    var event = getAuditEvent(invoice, eventAction);
+  public Future<Void> sendInvoiceEvent(Invoice invoice, Invoice originalInvoice, InvoiceAuditEvent.Action eventAction, Map<String, String> okapiHeaders) {
+    var event = getAuditEvent(invoice, originalInvoice, eventAction);
     log.info("sendInvoiceEvent:: Sending event with id: {} and invoiceId: {} to Kafka", event.getId(), invoice.getId());
     return sendToKafka(EventTopic.ACQ_INVOICE_CHANGED, event.getInvoiceId(), event, okapiHeaders)
       .onFailure(t -> log.warn("sendInvoiceEvent:: Failed to send event with id: {} and invoiceId: {} to Kafka", event.getId(), invoice.getId(), t));
@@ -48,31 +49,36 @@ public Future<Void> sendInvoiceEvent(Invoice invoice, InvoiceAuditEvent.Action e
    * Sends event for invoice line change(Create, Edit) to kafka.
    * InvoiceLineId is used as partition key to send all events for particular invoice line to the same partition.
    *
-   * @param invoiceLine  the event payload
-   * @param eventAction  the event action
-   * @param okapiHeaders the okapi headers
+   * @param invoiceLine         the event payload (post-edit state)
+   * @param originalInvoiceLine the pre-edit invoice line state; null for Create
+   * @param eventAction         the event action
+   * @param okapiHeaders        the okapi headers
    * @return future with true if sending was success or failed future in another case
    */
-  public Future<Void> sendInvoiceLineEvent(InvoiceLine invoiceLine, InvoiceLineAuditEvent.Action eventAction, Map<String, String> okapiHeaders) {
-    var event = getAuditEvent(invoiceLine, eventAction);
+  public Future<Void> sendInvoiceLineEvent(InvoiceLine invoiceLine, InvoiceLine originalInvoiceLine, InvoiceLineAuditEvent.Action eventAction, Map<String, String> okapiHeaders) {
+    var event = getAuditEvent(invoiceLine, originalInvoiceLine, eventAction);
     log.info("sendInvoiceLineEvent:: Sending event with id: {} and invoiceLineId: {} to Kafka", event.getId(), invoiceLine.getId());
     return sendToKafka(EventTopic.ACQ_INVOICE_LINE_CHANGED, event.getInvoiceLineId(), event, okapiHeaders)
       .onFailure(t -> log.error("sendInvoiceLineEvent:: Failed to send event with id: {} and invoiceLineId: {} to Kafka", event.getId(), invoiceLine.getId(), t));
   }
 
-  private InvoiceAuditEvent getAuditEvent(Invoice invoice, InvoiceAuditEvent.Action eventAction) {
-    return new InvoiceAuditEvent()
+  InvoiceAuditEvent getAuditEvent(Invoice invoice, Invoice originalInvoice, InvoiceAuditEvent.Action eventAction) {
+    var event = new InvoiceAuditEvent()
       .withId(UUID.randomUUID().toString())
       .withAction(eventAction)
       .withInvoiceId(invoice.getId())
       .withEventDate(new Date())
       .withActionDate(invoice.getMetadata().getUpdatedDate())
       .withUserId(invoice.getMetadata().getUpdatedByUserId())
       .withInvoiceSnapshot(invoice.withMetadata(null));
+    if (originalInvoice != null) {
+      event.setOriginalInvoiceSnapshot(originalInvoice.withMetadata(null));
+    }
+    return event;
   }
 
-  private InvoiceLineAuditEvent getAuditEvent(InvoiceLine invoiceLine, InvoiceLineAuditEvent.Action eventAction) {
-    return new InvoiceLineAuditEvent()
+  InvoiceLineAuditEvent getAuditEvent(InvoiceLine invoiceLine, InvoiceLine originalInvoiceLine, InvoiceLineAuditEvent.Action eventAction) {
+    var event = new InvoiceLineAuditEvent()
       .withId(UUID.randomUUID().toString())
       .withAction(eventAction)
       .withInvoiceId(invoiceLine.getInvoiceId())
@@ -81,6 +87,10 @@ private InvoiceLineAuditEvent getAuditEvent(InvoiceLine invoiceLine, InvoiceLine
       .withActionDate(invoiceLine.getMetadata().getUpdatedDate())
       .withUserId(invoiceLine.getMetadata().getUpdatedByUserId())
       .withInvoiceLineSnapshot(invoiceLine.withMetadata(null));
+    if (originalInvoiceLine != null) {
+      event.setOriginalInvoiceLineSnapshot(originalInvoiceLine.withMetadata(null));
+    }
+    return event;
   }
 
   private Future<Void> sendToKafka(EventTopic eventTopic, String key, Object eventPayload, Map<String, String> okapiHeaders) {

src/main/java/org/folio/service/audit/AuditOutboxService.java

@@ -19,6 +19,7 @@
 import io.vertx.core.Context;
 import io.vertx.core.Future;
 import io.vertx.core.json.Json;
+import io.vertx.core.json.jackson.DatabindCodec;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 
@@ -60,18 +61,38 @@ private List<Future<Void>> sendEventLogsToKafka(List<OutboxEventLog> eventLogs,
     return eventLogs.stream().map(eventLog ->
       switch (eventLog.getEntityType()) {
         case INVOICE -> {
-          var invoice = Json.decodeValue(eventLog.getPayload(), Invoice.class);
+          var wrapper = decodePayload(eventLog.getPayload(), Invoice.class);
           var action = InvoiceAuditEvent.Action.fromValue(eventLog.getAction());
-          yield producer.sendInvoiceEvent(invoice, action, okapiHeaders);
+          yield producer.sendInvoiceEvent(wrapper.getEntity(), wrapper.getOriginalEntity(), action, okapiHeaders);
         }
         case INVOICE_LINE -> {
-          var invoiceLine = Json.decodeValue(eventLog.getPayload(), InvoiceLine.class);
+          var wrapper = decodePayload(eventLog.getPayload(), InvoiceLine.class);
           var action = InvoiceLineAuditEvent.Action.fromValue(eventLog.getAction());
-          yield producer.sendInvoiceLineEvent(invoiceLine, action, okapiHeaders);
+          yield producer.sendInvoiceLineEvent(wrapper.getEntity(), wrapper.getOriginalEntity(), action, okapiHeaders);
         }
       }).toList();
   }
 
+  /**
+   * Decode an outbox payload into a wrapper. Falls back to decoding the payload as a bare entity
+   * for backwards compatibility with rows written before the wrapper format was introduced.
+   */
+  <T> AuditEntityWrapper<T> decodePayload(String payload, Class<T> entityClass) {
+    var mapper = DatabindCodec.mapper();
+    try {
+      var wrapperType = mapper.getTypeFactory().constructParametricType(AuditEntityWrapper.class, entityClass);
+      AuditEntityWrapper<T> wrapper = mapper.readValue(payload, wrapperType);
+      if (wrapper.getEntity() != null) {
+        return wrapper;
+      }
+    } catch (Exception ignored) {
+      // fall through to legacy decoding
+    }
+    log.warn("decodePayload:: Falling back to legacy (bare-entity) outbox payload decoding");
+    T entity = Json.decodeValue(payload, entityClass);
+    return AuditEntityWrapper.of(entity, null);
+  }
+
   /**
    * Saves invoice outbox log.
    *
@@ -82,7 +103,20 @@ private List<Future<Void>> sendEventLogsToKafka(List<OutboxEventLog> eventLogs,
    * @return future with saved outbox log id in the same transaction
    */
   public Future<Void> saveInvoiceOutboxLog(Conn conn, Invoice entity, InvoiceAuditEvent.Action action, Map<String, String> okapiHeaders) {
-    return saveOutboxLog(conn, okapiHeaders, action.value(), EntityType.INVOICE, entity.getId(), entity);
+    return saveInvoiceOutboxLog(conn, entity, null, action, okapiHeaders);
+  }
+
+  /**
+   * Saves invoice outbox log capturing the pre-edit state.
+   *
+   * @param conn         connection in transaction
+   * @param entity       the invoice (post-edit state)
+   * @param original     the invoice before the edit; null for Create
+   * @param action       the event action
+   * @param okapiHeaders okapi headers
+   */
+  public Future<Void> saveInvoiceOutboxLog(Conn conn, Invoice entity, Invoice original, InvoiceAuditEvent.Action action, Map<String, String> okapiHeaders) {
+    return saveOutboxLog(conn, okapiHeaders, action.value(), EntityType.INVOICE, entity.getId(), AuditEntityWrapper.of(entity, original));
   }
 
   /**
@@ -95,16 +129,29 @@ public Future<Void> saveInvoiceOutboxLog(Conn conn, Invoice entity, InvoiceAudit
    * @return future with saved outbox log id in the same transaction
    */
   public Future<Void> saveInvoiceLineOutboxLog(Conn conn, InvoiceLine entity, InvoiceLineAuditEvent.Action action, Map<String, String> okapiHeaders) {
-    return saveOutboxLog(conn, okapiHeaders, action.value(), EntityType.INVOICE_LINE, entity.getId(), entity);
+    return saveInvoiceLineOutboxLog(conn, entity, null, action, okapiHeaders);
+  }
+
+  /**
+   * Saves invoice line outbox log capturing the pre-edit state.
+   *
+   * @param conn         connection in transaction
+   * @param entity       the invoice line (post-edit state)
+   * @param original     the invoice line before the edit; null for Create
+   * @param action       the event action
+   * @param okapiHeaders okapi headers
+   */
+  public Future<Void> saveInvoiceLineOutboxLog(Conn conn, InvoiceLine entity, InvoiceLine original, InvoiceLineAuditEvent.Action action, Map<String, String> okapiHeaders) {
+    return saveOutboxLog(conn, okapiHeaders, action.value(), EntityType.INVOICE_LINE, entity.getId(), AuditEntityWrapper.of(entity, original));
   }
 
-  private Future<Void> saveOutboxLog(Conn conn, Map<String, String> okapiHeaders, String action, EntityType entityType, String entityId, Object entity) {
+  private Future<Void> saveOutboxLog(Conn conn, Map<String, String> okapiHeaders, String action, EntityType entityType, String entityId, AuditEntityWrapper<?> wrapper) {
     log.debug("saveOutboxLog:: Saving outbox log for {} with id: {}", entityType, entityId);
     var eventLog = new OutboxEventLog()
       .withEventId(UUID.randomUUID().toString())
       .withAction(action)
       .withEntityType(entityType)
-      .withPayload(Json.encode(entity));
+      .withPayload(Json.encode(wrapper));
     return outboxEventLogDAO.saveEventLog(conn, eventLog, TenantTool.tenantId(okapiHeaders))
       .onSuccess(reply -> log.info("saveOutboxLog:: Outbox log has been saved for {} with id: {}", entityType, entityId))
       .onFailure(e -> log.warn("saveOutboxLog:: Could not save outbox audit log for {} with id: {}", entityType, entityId, e));

src/test/java/org/folio/service/audit/AuditEventProducerTest.java

@@ -0,0 +1,95 @@
+package org.folio.service.audit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.Date;
+import java.util.UUID;
+
+import org.folio.rest.jaxrs.model.Invoice;
+import org.folio.rest.jaxrs.model.InvoiceAuditEvent;
+import org.folio.rest.jaxrs.model.InvoiceLine;
+import org.folio.rest.jaxrs.model.InvoiceLineAuditEvent;
+import org.folio.rest.jaxrs.model.Metadata;
+import org.junit.jupiter.api.Test;
+
+class AuditEventProducerTest {
+
+  private final AuditEventProducer producer = new AuditEventProducer(null);
+
+  @Test
+  void invoiceEditEventCarriesOriginalSnapshot() {
+    var original = getInvoice("vendor-old", "Open");
+    var updated = getInvoice("vendor-new", "Reviewed");
+
+    InvoiceAuditEvent event = producer.getAuditEvent(updated, original, InvoiceAuditEvent.Action.EDIT);
+
+    assertEquals(InvoiceAuditEvent.Action.EDIT, event.getAction());
+    assertNotNull(event.getInvoiceSnapshot());
+    assertEquals("vendor-new", event.getInvoiceSnapshot().getVendorInvoiceNo());
+    assertEquals(Invoice.Status.REVIEWED, event.getInvoiceSnapshot().getStatus());
+
+    assertNotNull(event.getOriginalInvoiceSnapshot());
+    assertEquals("vendor-old", event.getOriginalInvoiceSnapshot().getVendorInvoiceNo());
+    assertEquals(Invoice.Status.OPEN, event.getOriginalInvoiceSnapshot().getStatus());
+  }
+
+  @Test
+  void invoiceCreateEventOmitsOriginalSnapshot() {
+    InvoiceAuditEvent event = producer.getAuditEvent(getInvoice("v", "Open"), null, InvoiceAuditEvent.Action.CREATE);
+
+    assertEquals(InvoiceAuditEvent.Action.CREATE, event.getAction());
+    assertNotNull(event.getInvoiceSnapshot());
+    assertNull(event.getOriginalInvoiceSnapshot());
+  }
+
+  @Test
+  void invoiceLineEditEventCarriesOriginalSnapshot() {
+    var original = getInvoiceLine("desc-old", 10d);
+    var updated = getInvoiceLine("desc-new", 25d);
+
+    InvoiceLineAuditEvent event = producer.getAuditEvent(updated, original, InvoiceLineAuditEvent.Action.EDIT);
+
+    assertEquals(InvoiceLineAuditEvent.Action.EDIT, event.getAction());
+    assertNotNull(event.getInvoiceLineSnapshot());
+    assertEquals("desc-new", event.getInvoiceLineSnapshot().getDescription());
+    assertEquals(25d, event.getInvoiceLineSnapshot().getSubTotal());
+
+    assertNotNull(event.getOriginalInvoiceLineSnapshot());
+    assertEquals("desc-old", event.getOriginalInvoiceLineSnapshot().getDescription());
+    assertEquals(10d, event.getOriginalInvoiceLineSnapshot().getSubTotal());
+  }
+
+  @Test
+  void invoiceLineCreateEventOmitsOriginalSnapshot() {
+    InvoiceLineAuditEvent event = producer.getAuditEvent(getInvoiceLine("d", 1d), null, InvoiceLineAuditEvent.Action.CREATE);
+
+    assertEquals(InvoiceLineAuditEvent.Action.CREATE, event.getAction());
+    assertNotNull(event.getInvoiceLineSnapshot());
+    assertNull(event.getOriginalInvoiceLineSnapshot());
+  }
+
+  private Invoice getInvoice(String vendorInvoiceNo, String status) {
+    return new Invoice()
+      .withId(UUID.randomUUID().toString())
+      .withVendorInvoiceNo(vendorInvoiceNo)
+      .withStatus(Invoice.Status.fromValue(status))
+      .withMetadata(getMetadata());
+  }
+
+  private InvoiceLine getInvoiceLine(String description, double subTotal) {
+    return new InvoiceLine()
+      .withId(UUID.randomUUID().toString())
+      .withInvoiceId(UUID.randomUUID().toString())
+      .withDescription(description)
+      .withSubTotal(subTotal)
+      .withMetadata(getMetadata());
+  }
+
+  private Metadata getMetadata() {
+    return new Metadata()
+      .withUpdatedDate(new Date())
+      .withUpdatedByUserId(UUID.randomUUID().toString());
+  }
+}