[MODAUD-297] Implement user event consumer

NEWS.md

@@ -5,6 +5,7 @@
 * [MODAUD-288](https://folio-org.atlassian.net/browse/MODAUD-288) - assertj: Upgrade from 3.27.3 to 3.27.7, change scope from compile to test
 * [MODAUD-296](https://folio-org.atlassian.net/browse/MODAUD-296) - Implement User Audit Enabled Setting
 * [MODAUD-298](https://folio-org.atlassian.net/browse/MODAUD-298) - Extract shared utilities and decouple DiffCalculator from inventory types
+* [MODAUD-297](https://folio-org.atlassian.net/browse/MODAUD-297) - Consume user domain events and store audit history
 
 ## 2.11.1 2025-04-15 
 * [MODAUD-250](https://folio-org.atlassian.net/browse/MODAUD-250) - Version history of "MARC" records is not tracked

PERSONAL_DATA_DISCLOSURE.md

@@ -12,24 +12,25 @@ For the purposes of this form, "store" includes the following:
 ## Personal Data Stored by This Module
 - [ ] This module does not store any personal data.
 - [ ] This module provides [custom fields](https://github.com/folio-org/folio-custom-fields).
-- [ ] This module stores fields with free-form text (tags, notes, descriptions, etc.)
+- [x] This module stores custom fields values
+- [x] This module stores fields with free-form text (tags, notes, descriptions, etc.)
 - [x] This module caches personal data
 ---
 - [x] First name
 - [x] Last name
-- [ ] Middle name
-- [ ] Pseudonym / Alias / Nickname / Username / User ID
+- [x] Middle name
+- [x] Pseudonym / Alias / Nickname / Username / User ID
 - [ ] Gender
-- [ ] Date of birth
+- [x] Date of birth
 - [ ] Place of birth
 - [ ] Racial or ethnic origin
-- [ ] Address
+- [x] Address
 - [ ] Location information
-- [ ] Phone numbers
+- [x] Phone numbers
 - [ ] Passport number / National identification numbers
 - [ ] Driver’s license number
 - [ ] Social security number
-- [ ] Email address
+- [x] Email address
 - [ ] Web cookies
 - [ ] IP address
 - [ ] Geolocation data

mod-audit-server/src/main/java/org/folio/dao/user/UserAuditEntity.java

@@ -0,0 +1,8 @@
+package org.folio.dao.user;
+
+import java.sql.Timestamp;
+import java.util.UUID;
+import org.folio.domain.diff.ChangeRecordDto;
+
+public record UserAuditEntity(UUID eventId, Timestamp eventDate, UUID userId,
+                               String action, UUID performedBy, ChangeRecordDto diff) { }

mod-audit-server/src/main/java/org/folio/dao/user/UserEventDao.java

@@ -0,0 +1,15 @@
+package org.folio.dao.user;
+
+import io.vertx.core.Future;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowSet;
+import java.util.UUID;
+
+public interface UserEventDao {
+
+  Future<RowSet<Row>> save(UserAuditEntity userAuditEntity, String tenantId);
+
+  Future<Void> deleteByUserId(UUID userId, String tenantId);
+
+  String tableName();
+}

mod-audit-server/src/main/java/org/folio/dao/user/impl/UserEventDaoImpl.java

@@ -0,0 +1,87 @@
+package org.folio.dao.user.impl;
+
+import static org.folio.util.DbUtils.formatDBTableName;
+
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import io.vertx.core.json.JsonObject;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowSet;
+import io.vertx.sqlclient.Tuple;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.UUID;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.folio.dao.user.UserAuditEntity;
+import org.folio.dao.user.UserEventDao;
+import org.folio.util.PostgresClientFactory;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public class UserEventDaoImpl implements UserEventDao {
+
+  private static final Logger LOGGER = LogManager.getLogger();
+  private static final String USER_AUDIT_TABLE = "user_audit";
+
+  private static final String INSERT_SQL = """
+    INSERT INTO %s (event_id, event_date, user_id, action, performed_by, diff)
+    VALUES ($1, $2, $3, $4, $5, $6)
+    """;
+
+  private static final String DELETE_BY_USER_ID_SQL = """
+    DELETE FROM %s
+      WHERE user_id = $1
+    """;
+
+  private final PostgresClientFactory pgClientFactory;
+
+  public UserEventDaoImpl(PostgresClientFactory pgClientFactory) {
+    this.pgClientFactory = pgClientFactory;
+  }
+
+  @Override
+  public Future<RowSet<Row>> save(UserAuditEntity event, String tenantId) {
+    LOGGER.debug("save:: Trying to save UserAuditEntity with [tenantId: {}, eventId: {}, userId: {}]",
+      tenantId, event.eventId(), event.userId());
+    var promise = Promise.<RowSet<Row>>promise();
+    var table = formatDBTableName(tenantId, tableName());
+    var query = INSERT_SQL.formatted(table);
+    makeSaveCall(promise, query, event, tenantId);
+    return promise.future();
+  }
+
+  @Override
+  public Future<Void> deleteByUserId(UUID userId, String tenantId) {
+    LOGGER.debug("deleteByUserId:: Deleting user audit records with [tenantId: {}, userId: {}]",
+      tenantId, userId);
+    var table = formatDBTableName(tenantId, tableName());
+    var query = DELETE_BY_USER_ID_SQL.formatted(table);
+    return pgClientFactory.createInstance(tenantId).execute(query, Tuple.of(userId))
+      .mapEmpty();
+  }
+
+  @Override
+  public String tableName() {
+    return USER_AUDIT_TABLE;
+  }
+
+  private void makeSaveCall(Promise<RowSet<Row>> promise, String query, UserAuditEntity event, String tenantId) {
+    LOGGER.debug("makeSaveCall:: Making save call with query : {} and tenant id : {}", query, tenantId);
+    try {
+      pgClientFactory.createInstance(tenantId).execute(query, Tuple.of(event.eventId(),
+          LocalDateTime.ofInstant(event.eventDate().toInstant(), ZoneId.systemDefault()),
+          event.userId(),
+          event.action(),
+          event.performedBy(),
+          event.diff() != null ? JsonObject.mapFrom(event.diff()) : null),
+        promise);
+      LOGGER.info("makeSaveCall:: Saving UserAuditEntity with [tenantId: {}, eventId:{}, userId:{}]",
+        tenantId, event.eventId(), event.userId());
+    } catch (Exception e) {
+      LOGGER.error("Failed to save record with [eventId:{}, userId:{}, tableName: {}]",
+        event.eventId(), event.userId(), tableName(), e);
+      promise.fail(e);
+    }
+  }
+}

mod-audit-server/src/main/java/org/folio/mapper/user/UserEventToEntityMapper.java

@@ -0,0 +1,43 @@
+package org.folio.mapper.user;
+
+import java.sql.Timestamp;
+import java.util.UUID;
+import java.util.function.Function;
+import org.folio.dao.user.UserAuditEntity;
+import org.folio.domain.diff.ChangeRecordDto;
+import org.folio.services.diff.user.UserDiffCalculator;
+import org.folio.util.user.UserEvent;
+import org.folio.util.user.UserEventType;
+import org.folio.util.user.UserUtils;
+import org.springframework.stereotype.Component;
+
+@Component
+public class UserEventToEntityMapper implements Function<UserEvent, UserAuditEntity> {
+
+  private final UserDiffCalculator userDiffCalculator;
+
+  public UserEventToEntityMapper(UserDiffCalculator userDiffCalculator) {
+    this.userDiffCalculator = userDiffCalculator;
+  }
+
+  @Override
+  public UserAuditEntity apply(UserEvent event) {
+    var performedByStr = UserUtils.extractPerformedBy(event);
+    var performedBy = performedByStr != null ? UUID.fromString(performedByStr) : null;
+    var diff = UserEventType.UPDATED.equals(event.getType())
+               ? getDiff(event)
+               : null;
+    return new UserAuditEntity(
+      UUID.fromString(event.getId()),
+      new Timestamp(event.getTimestamp()),
+      UUID.fromString(event.getUserId()),
+      event.getType().name(),
+      performedBy,
+      diff
+    );
+  }
+
+  private ChangeRecordDto getDiff(UserEvent event) {
+    return userDiffCalculator.calculateDiff(event.getOldValue(), event.getNewValue());
+  }
+}

mod-audit-server/src/main/java/org/folio/rest/impl/InitAPIs.java

@@ -26,6 +26,7 @@
 import org.folio.verticle.inventory.InstanceConsumersVerticle;
 import org.folio.verticle.inventory.ItemConsumersVerticle;
 import org.folio.verticle.marc.MarcRecordEventConsumersVerticle;
+import org.folio.verticle.user.UserConsumersVerticle;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.support.AbstractApplicationContext;
 
@@ -85,6 +86,11 @@ public class InitAPIs implements InitAPI {
   @Value("${src.source-records.kafka.consumer.pool.size:5}")
   private int srsSourceRecordsConsumerPoolSize;
 
+  @Value("${user.users.kafka.consumer.instancesNumber:1}")
+  private int userConsumerInstancesNumber;
+  @Value("${user.users.kafka.consumer.pool.size:5}")
+  private int userConsumerPoolSize;
+
   @Override
   public void init(Vertx vertx, Context context, Handler<AsyncResult<Boolean>> handler) {
     LOGGER.debug("init:: InitAPI starting...");
@@ -122,6 +128,7 @@ private Future<?> deployConsumersVerticles(Vertx vertx) {
     Promise<String> inventoryHoldingsConsumer = Promise.promise();
     Promise<String> inventoryItemConsumer = Promise.promise();
     Promise<String> sourceRecordsConsumer = Promise.promise();
+    Promise<String> userEventsConsumer = Promise.promise();
 
     deployVerticle(vertx, verticleFactory, OrderEventConsumersVerticle.class, acqOrderConsumerInstancesNumber, acqOrderConsumerPoolSize, orderEventsConsumer);
     deployVerticle(vertx, verticleFactory, OrderLineEventConsumersVerticle.class, acqOrderLineConsumerInstancesNumber, acqOrderLineConsumerPoolSize, orderLineEventsConsumer);
@@ -133,6 +140,7 @@ private Future<?> deployConsumersVerticles(Vertx vertx) {
     deployVerticle(vertx, verticleFactory, HoldingsConsumersVerticle.class, invHoldingsConsumerInstancesNumber, invHoldingsConsumerPoolSize, inventoryHoldingsConsumer);
     deployVerticle(vertx, verticleFactory, ItemConsumersVerticle.class, invItemConsumerInstancesNumber, invItemConsumerPoolSize, inventoryItemConsumer);
     deployVerticle(vertx, verticleFactory, MarcRecordEventConsumersVerticle.class, srsSourceRecordsConsumerInstancesNumber, srsSourceRecordsConsumerPoolSize, sourceRecordsConsumer);
+    deployVerticle(vertx, verticleFactory, UserConsumersVerticle.class, userConsumerInstancesNumber, userConsumerPoolSize, userEventsConsumer);
 
     LOGGER.info("deployConsumersVerticles:: All consumer verticles were successfully deployed");
     return GenericCompositeFuture.all(Arrays.asList(
@@ -145,7 +153,8 @@ private Future<?> deployConsumersVerticles(Vertx vertx) {
       inventoryInstanceConsumer.future(),
       inventoryHoldingsConsumer.future(),
       inventoryItemConsumer.future(),
-      sourceRecordsConsumer.future()
+      sourceRecordsConsumer.future(),
+      userEventsConsumer.future()
     ));
   }
 

mod-audit-server/src/main/java/org/folio/services/diff/DiffCalculator.java

@@ -26,6 +26,11 @@
 import org.javers.core.diff.changetype.container.ElementValueChange;
 import org.javers.core.diff.changetype.container.ValueAddOrRemove;
 import org.javers.core.diff.changetype.container.ValueAdded;
+import org.javers.core.diff.changetype.map.EntryAdded;
+import org.javers.core.diff.changetype.map.EntryChange;
+import org.javers.core.diff.changetype.map.EntryRemoved;
+import org.javers.core.diff.changetype.map.EntryValueChange;
+import org.javers.core.diff.changetype.map.MapChange;
 import org.javers.core.metamodel.object.ValueObjectId;
 
 public abstract class DiffCalculator<T> {
@@ -71,6 +76,10 @@ private ChangeRecordDto convert(Changes changes) {
         fieldChanges.add(processValueChange(valueChange));
       } else if (change instanceof CollectionChange<?> collectionChange) {
         collectionChanges.add(processCollectionChange(collectionChange, groupedChanges));
+      // MapChanges are skipped by default to filter out noise from JSON schema additionalProperties.
+      // Subclasses opt in for map properties that carry real data (see UserDiffCalculator).
+      } else if (change instanceof MapChange mapChange && shouldProcessMapChange(mapChange)) {
+        fieldChanges.addAll(processMapChange(mapChange));
       }
     }
 
@@ -129,6 +138,48 @@ private CollectionItemChangeDto processElementValueChange(ElementValueChange ele
     );
   }
 
+  /**
+   * Determines whether a {@link MapChange} should be processed and included in the diff output.
+   *
+   * <p>JSON schema-generated models include an {@code additionalProperties} catch-all map that
+   * Javers treats as a real property, producing spurious {@link MapChange} events for every entity.
+   * The default implementation returns {@code false} to suppress these.
+   *
+   * <p>Subclasses should override this when the entity has map-type properties that carry real
+   * semantic data (e.g., {@code CustomFields} in the User model, whose values are stored in the
+   * underlying {@code additionalProperties} map).
+   *
+   * @param mapChange the map change reported by Javers
+   * @return {@code true} to include this change in the diff, {@code false} to skip it
+   */
+  protected boolean shouldProcessMapChange(MapChange mapChange) {
+    return false;
+  }
+
+  private List<FieldChangeDto> processMapChange(MapChange mapChange) {
+    var result = new ArrayList<FieldChangeDto>();
+    String basePath = mapChange.getPropertyNameWithPath();
+    String propName = mapChange.getPropertyName();
+    String prefix = basePath.endsWith(propName)
+      ? basePath.substring(0, basePath.length() - propName.length())
+      : "";
+
+    for (var entry : mapChange.getEntryChanges()) {
+      EntryChange entryChange = (EntryChange) entry;
+      String key = String.valueOf(entryChange.getKey());
+      String fullPath = prefix + key;
+
+      if (entryChange instanceof EntryValueChange evc) {
+        result.add(FieldChangeDto.modified(key, fullPath, evc.getLeftValue(), evc.getRightValue()));
+      } else if (entryChange instanceof EntryAdded ea) {
+        result.add(FieldChangeDto.added(key, fullPath, ea.getValue()));
+      } else if (entryChange instanceof EntryRemoved er) {
+        result.add(FieldChangeDto.removed(key, fullPath, er.getValue()));
+      }
+    }
+    return result;
+  }
+
   private FieldChangeDto processValueChange(ValueChange valueChange) {
     return FieldChangeDto.of(
       valueChange.getPropertyName(),

mod-audit-server/src/main/java/org/folio/services/diff/user/UserDiffCalculator.java

@@ -0,0 +1,44 @@
+package org.folio.services.diff.user;
+
+import java.util.function.Supplier;
+import org.folio.rest.external.CustomFields;
+import org.javers.core.diff.changetype.map.MapChange;
+import org.folio.rest.external.Metadata;
+import org.folio.rest.external.Personal__1;
+import org.folio.rest.external.Tags__3;
+import org.folio.rest.external.User;
+import org.folio.services.diff.DiffCalculator;
+import org.springframework.stereotype.Component;
+
+@Component
+public class UserDiffCalculator extends DiffCalculator<User> {
+
+  @Override
+  protected Supplier<User> access(User value) {
+    return () -> {
+      if (value.getPersonal() == null) {
+        value.setPersonal(new Personal__1());
+      }
+      if (value.getMetadata() == null) {
+        value.setMetadata(new Metadata());
+      }
+      if (value.getTags() == null) {
+        value.setTags(new Tags__3());
+      }
+      if (value.getCustomFields() == null) {
+        value.setCustomFields(new CustomFields());
+      }
+      return value;
+    };
+  }
+
+  @Override
+  protected boolean shouldProcessMapChange(MapChange mapChange) {
+    return mapChange.getPropertyNameWithPath().startsWith("customFields.");
+  }
+
+  @Override
+  protected Class<User> getType() {
+    return User.class;
+  }
+}

mod-audit-server/src/main/java/org/folio/services/user/UserEventService.java

@@ -0,0 +1,9 @@
+package org.folio.services.user;
+
+import io.vertx.core.Future;
+import org.folio.util.user.UserEvent;
+
+public interface UserEventService {
+
+  Future<String> processEvent(UserEvent event, String tenantId);
+}