FINERACT-2452: New command processing - audit

Co-Authored-By: Adam Saghy <adam.saghy@dpc.hu>
Co-Authored-By: Yu Wati Nyi <nyi22y@mtholyoke.edu>

fineract-command/src/main/java/org/apache/fineract/command/core/Command.java

@@ -20,8 +20,7 @@ package org.apache.fineract.command.core;
 
 import java.io.Serial;
 import java.io.Serializable;
-import java.time.OffsetDateTime;
-import java.util.UUID;
+import java.time.Instant;
 import lombok.Data;
 import lombok.experimental.FieldNameConstants;
 
@@ -32,15 +31,31 @@ public class Command<T> implements Serializable {
     @Serial
     private static final long serialVersionUID = 1L;
 
-    private UUID id;
+    private Long commandId;
 
     private String idempotencyKey;
 
-    private OffsetDateTime createdAt;
+    private String ipAddress;
 
-    private String tenantId;
+    private Instant createdAt;
 
-    private String username;
+    private Instant updatedAt;
+
+    private Instant executedAt;
+
+    private Instant approvedAt;
+
+    private Instant rejectedAt;
+
+    private String initiatedByUsername;
+
+    private String executedByUsername;
+
+    private String approvedByUsername;
+
+    private String rejectedByUsername;
+
+    private String error;
 
     private T payload;
 }

fineract-command/src/main/java/org/apache/fineract/command/core/CommandAuditor.java

@@ -0,0 +1,30 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.core;
+
+public interface CommandAuditor {
+
+    Object getResponseByIdempotencyKey(String key);
+
+    void processing(Command<?> command);
+
+    void processed(Command<?> command, Object response);
+
+    void error(Command<?> command);
+}

fineract-command/src/main/java/org/apache/fineract/command/core/CommandConstants.java

@@ -23,5 +23,8 @@ import lombok.experimental.UtilityClass;
 @UtilityClass
 public final class CommandConstants {
 
-    public static final String COMMAND_REQUEST_ID = "x-fineract-request-id";
+    public static final String COMMAND_JSON_CLASS_ATTRIBUTE = "@class";
+    public static final String COMMAND_HTTP_HEADER_REQUEST_ID = "x-fineract-request-id";
+    public static final String COMMAND_HTTP_HEADER_TENANT_ID = "Fineract-Platform-TenantId";
+    public static final String COMMAND_HTTP_HEADER_IP = "IP";
 }

fineract-command/src/main/java/org/apache/fineract/command/core/CommandHandler.java

@@ -19,11 +19,19 @@
 package org.apache.fineract.command.core;
 
 import com.google.common.reflect.TypeToken;
+import lombok.SneakyThrows;
 
 public interface CommandHandler<REQ, RES> {
 
     RES handle(Command<REQ> command);
 
+    @SneakyThrows
+    default RES fallback(Command<REQ> command, Throwable t) {
+        // NOTE: I think this should be enough for now; the errors bubble up and will eventually be handled the global
+        // error handler. Any command handler can override this default to implement more specialized fallbacks.
+        throw t;
+    }
+
     default boolean matches(Command<REQ> command) {
         TypeToken<REQ> handlerType = new TypeToken<>(getClass()) {};
 

fineract-command/src/main/java/org/apache/fineract/command/core/CommandProperties.java

@@ -51,6 +51,18 @@ public final class CommandProperties implements Serializable {
     @Builder.Default
     private ProducerType producerType = ProducerType.SINGLE;
 
+    @Builder.Default
+    private Boolean auditable = true;
+
+    @Builder.Default
+    private Boolean fileDeadLetterQueueEnabled = false;
+
+    @Builder.Default
+    private String fileDeadLetterQueuePath = "/tmp/fineract-command-audit";
+
+    @Builder.Default
+    private String idemPotencyKeyHeaderName = "Idempotency-Key";
+
     public enum CommandExecutorType {
         sync, //
         async, //

fineract-command/src/main/java/org/apache/fineract/command/implementation/AsynchronousCommandExecutor.java

@@ -18,41 +18,57 @@
  */
 package org.apache.fineract.command.implementation;
 
-import java.util.List;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Supplier;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.command.core.Command;
-import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandAuditor;
 import org.apache.fineract.command.core.CommandHandler;
-import org.apache.fineract.command.core.CommandMiddleware;
 import org.apache.fineract.command.core.CommandRouter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.stereotype.Component;
 
 @Slf4j
-@RequiredArgsConstructor
-@Component
-@ConditionalOnProperty(value = "fineract.command.executor", havingValue = "async")
-public class AsynchronousCommandExecutor implements CommandExecutor {
+// TODO: not ready yet for prime time
+// @Component
+// @ConditionalOnProperty(value = "fineract.command.executor", havingValue = "async")
+public class AsynchronousCommandExecutor extends BaseCommandExecutor {
 
-    private final List<CommandMiddleware> middlewares;
-
-    private final CommandRouter router;
+    public AsynchronousCommandExecutor(CommandRouter router, CommandAuditor auditor) {
+        super(router, auditor);
+    }
 
     @Override
-    public <REQ, RES> Supplier<RES> execute(Command<REQ> command) {
+    public <REQ, RES> Supplier<RES> execute(final Command<REQ> command) {
         CompletableFuture<RES> future = CompletableFuture.supplyAsync(() -> {
-            for (CommandMiddleware middleware : middlewares) {
-                middleware.invoke(command);
-            }
+            auditor.processing(command);
 
             CommandHandler<REQ, RES> handler = router.route(command);
 
             return handler.handle(command);
+        }).whenComplete((response, t) -> {
+            if (t == null) {
+                auditor.processed(command, response);
+            } else {
+                command.setError(t.getMessage());
+
+                auditor.error(command);
+            }
         });
 
-        return future::join;
+        return () -> {
+            try {
+                // TODO: make this configurable
+                return future.get(3, SECONDS);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            } catch (ExecutionException e) {
+                throw new RuntimeException(e);
+            } catch (TimeoutException e) {
+                throw new RuntimeException(e);
+            }
+        };
     }
 }

fineract-command/src/main/java/org/apache/fineract/command/implementation/BaseCommandExecutor.java

@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import org.apache.fineract.command.core.CommandAuditor;
+import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandRouter;
+
+public abstract class BaseCommandExecutor implements CommandExecutor {
+
+    protected final CommandRouter router;
+    protected final CommandAuditor auditor;
+
+    protected BaseCommandExecutor(CommandRouter router, CommandAuditor auditor) {
+        this.router = router;
+        this.auditor = auditor;
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/BaseCommandPipeline.java

@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import static org.apache.fineract.command.core.CommandConstants.COMMAND_HTTP_HEADER_IP;
+
+import java.time.Instant;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandPipeline;
+import org.apache.fineract.command.core.CommandProperties;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.web.context.request.RequestContextHolder;
+import org.springframework.web.context.request.ServletRequestAttributes;
+
+@Slf4j
+@RequiredArgsConstructor
+public abstract class BaseCommandPipeline implements CommandPipeline {
+
+    protected final CommandExecutor executor;
+    protected final CommandProperties properties;
+
+    protected void setAttributes(Command<?> command) {
+        if (StringUtils.isEmpty(command.getIpAddress())) {
+            command.setIpAddress(getHeader(COMMAND_HTTP_HEADER_IP));
+        }
+        if (StringUtils.isEmpty(command.getIdempotencyKey())) {
+            command.setIdempotencyKey(getHeader(properties.getIdemPotencyKeyHeaderName()));
+        }
+        if (command.getCreatedAt() == null) {
+            command.setCreatedAt(Instant.now());
+        }
+        if (StringUtils.isEmpty(command.getInitiatedByUsername())) {
+            command.setInitiatedByUsername(getUsername());
+        }
+    }
+
+    protected String getHeader(String name) {
+        return getHeader(name, false);
+    }
+
+    protected String getHeader(String name, boolean searchParameter) {
+        var attributes = RequestContextHolder.getRequestAttributes();
+
+        if (attributes instanceof ServletRequestAttributes servletAttributes) {
+            var value = servletAttributes.getRequest().getHeader(name.toLowerCase());
+
+            if (searchParameter && StringUtils.isEmpty(value)) {
+                value = servletAttributes.getRequest().getParameter(name.toLowerCase());
+            }
+
+            return value;
+        }
+
+        return null;
+    }
+
+    protected String getUsername() {
+        final var context = SecurityContextHolder.getContext();
+
+        if (context != null) {
+            final var auth = context.getAuthentication();
+
+            if (auth != null) {
+                return auth.getName();
+            }
+        }
+
+        return null;
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/DefaultCommandAuditor.java

@@ -0,0 +1,168 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.implementation;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.fineract.command.core.CommandConstants.COMMAND_JSON_CLASS_ATTRIBUTE;
+import static org.apache.fineract.command.persistence.domain.CommandEntityState.ERROR;
+import static org.apache.fineract.command.persistence.domain.CommandEntityState.PROCESSED;
+import static org.apache.fineract.command.persistence.domain.CommandEntityState.UNDER_PROCESSING;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.resilience4j.retry.annotation.Retry;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Instant;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.io.FileUtils;
+import org.apache.fineract.command.core.Command;
+import org.apache.fineract.command.core.CommandAuditor;
+import org.apache.fineract.command.core.CommandProperties;
+import org.apache.fineract.command.persistence.domain.CommandEntity;
+import org.apache.fineract.command.persistence.domain.CommandRepository;
+import org.apache.fineract.command.persistence.mapping.CommandMapper;
+import org.springframework.boot.context.event.ApplicationStartedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@RequiredArgsConstructor
+@Component
+public class DefaultCommandAuditor implements CommandAuditor {
+
+    private final CommandMapper mapper;
+    private final CommandRepository repository;
+    private final CommandProperties properties;
+    private final ObjectMapper objectMapper;
+
+    @SneakyThrows
+    @Override
+    public Object getResponseByIdempotencyKey(String key) {
+        final var result = repository.findOneByIdempotencyKey(key);
+        final var response = result.map(CommandEntity::getResponse).orElse(null);
+
+        if (response != null) {
+            final var clazz = response.get(COMMAND_JSON_CLASS_ATTRIBUTE).asText();
+
+            return objectMapper.convertValue(response, Class.forName(clazz));
+        }
+
+        return null;
+    }
+
+    @Retry(name = "commandAuditProcessing", fallbackMethod = "fallback")
+    @Override
+    public void processing(Command<?> command) {
+        if (Boolean.TRUE.equals(properties.getAuditable())) {
+            final var now = Instant.now();
+            final var commandEntity = mapper.map(command);
+
+            commandEntity.setCreatedAt(now);
+            commandEntity.setUpdatedAt(now);
+            commandEntity.setState(UNDER_PROCESSING);
+
+            repository.save(commandEntity);
+
+            command.setCreatedAt(now);
+            command.setUpdatedAt(now);
+            command.setCommandId(commandEntity.getId());
+        }
+    }
+
+    @Retry(name = "commandAuditProcessed", fallbackMethod = "fallback")
+    @Override
+    public void processed(Command<?> command, Object response) {
+        if (Boolean.TRUE.equals(properties.getAuditable())) {
+            final var now = Instant.now();
+            final var commandEntity = mapper.map(command, response);
+
+            // TODO: improve when maker-checker functionality lands
+            commandEntity.setExecutedByUsername(command.getInitiatedByUsername());
+            commandEntity.setUpdatedAt(now);
+            commandEntity.setExecutedAt(now);
+            commandEntity.setState(PROCESSED);
+
+            repository.save(commandEntity);
+
+            command.setUpdatedAt(now);
+            command.setExecutedAt(now);
+        }
+    }
+
+    @Retry(name = "commandAuditError", fallbackMethod = "fallback")
+    @Override
+    public void error(Command<?> command) {
+        if (Boolean.TRUE.equals(properties.getAuditable())) {
+            final var now = Instant.now();
+            final var commandEntity = mapper.map(command);
+
+            commandEntity.setUpdatedAt(now);
+            commandEntity.setState(ERROR);
+
+            repository.save(commandEntity);
+
+            command.setUpdatedAt(now);
+            command.setExecutedAt(now);
+        }
+    }
+
+    @EventListener(ApplicationStartedEvent.class)
+    void onStartup() {
+        if (Boolean.TRUE.equals(properties.getFileDeadLetterQueueEnabled())) {
+            try {
+                var created = Path.of(properties.getFileDeadLetterQueuePath()).toFile().mkdirs();
+                log.info("Created command dead-letter queue: {} ({})", properties.getFileDeadLetterQueuePath(), created);
+            } catch (Exception e) {
+                log.error("Unable to initialize command dead-letter queue:", e);
+            }
+        }
+    }
+
+    @Retry(name = "commandAuditFallback", fallbackMethod = "fatal")
+    @SuppressWarnings("UnusedMethod")
+    private void fallback(Command<?> command, Throwable t) throws Exception {
+        if (Boolean.TRUE.equals(properties.getFileDeadLetterQueueEnabled())) {
+            setError(command, t);
+            write(command);
+        } else {
+            fatal(command, t);
+        }
+    }
+
+    @SuppressWarnings("UnusedMethod")
+    private void fatal(Command<?> command, Throwable t) {
+        // note: last line of defense; if this fails then all is lost
+        setError(command, t);
+        log.error("Command audit error: {}", command, t);
+    }
+
+    private void setError(Command<?> command, Throwable t) {
+        if (command != null && t != null) {
+            command.setUpdatedAt(Instant.now());
+            command.setError(t.getMessage());
+        }
+    }
+
+    private void write(Command<?> command) throws IOException {
+        var file = Path.of(properties.getFileDeadLetterQueuePath(), command.getCreatedAt().toEpochMilli() + ".json").toFile();
+        FileUtils.write(file, objectMapper.writeValueAsString(command), UTF_8);
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/implementation/DefaultCommandPipeline.java

@@ -21,26 +21,29 @@ package org.apache.fineract.command.implementation;
 import static java.util.Objects.requireNonNull;
 
 import java.util.function.Supplier;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.command.core.Command;
 import org.apache.fineract.command.core.CommandExecutor;
 import org.apache.fineract.command.core.CommandPipeline;
+import org.apache.fineract.command.core.CommandProperties;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.stereotype.Component;
 
 @Slf4j
-@RequiredArgsConstructor
 @Component
 @ConditionalOnBean(CommandPipeline.class)
-public class DefaultCommandPipeline implements CommandPipeline {
+public class DefaultCommandPipeline extends BaseCommandPipeline {
 
-    private final CommandExecutor executor;
+    public DefaultCommandPipeline(CommandExecutor executor, CommandProperties properties) {
+        super(executor, properties);
+    }
 
     @Override
     public <REQ, RES> Supplier<RES> send(final Command<REQ> command) {
         requireNonNull(command, "Command must not be null");
 
+        setAttributes(command);
+
         return executor.execute(command);
     }
 }

fineract-command/src/main/java/org/apache/fineract/command/implementation/DisruptorCommandExecutor.java

@@ -22,7 +22,6 @@ import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.dsl.Disruptor;
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Supplier;
 import lombok.Getter;
@@ -30,26 +29,38 @@ import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.command.core.Command;
-import org.apache.fineract.command.core.CommandExecutor;
-import org.apache.fineract.command.core.CommandMiddleware;
+import org.apache.fineract.command.core.CommandAuditor;
 import org.apache.fineract.command.core.CommandRouter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.stereotype.Component;
 
 @Slf4j
-@RequiredArgsConstructor
-@Component
-@ConditionalOnProperty(value = "fineract.command.executor", havingValue = "disruptor")
+// TODO: not ready yet for prime time
+// @Component
+// @ConditionalOnProperty(value = "fineract.command.executor", havingValue = "disruptor")
 @SuppressWarnings({ "unchecked", "rawtypes" })
-public class DisruptorCommandExecutor implements CommandExecutor, Closeable {
+public class DisruptorCommandExecutor extends BaseCommandExecutor implements Closeable {
 
     private final Disruptor<CommandEvent> disruptor;
 
+    public DisruptorCommandExecutor(Disruptor<CommandEvent> disruptor, CommandRouter router, CommandAuditor auditor) {
+        super(router, auditor);
+        this.disruptor = disruptor;
+    }
+
     @Override
     public <REQ, RES> Supplier<RES> execute(Command<REQ> command) {
         CommandEvent<REQ, RES> processedEvent = next(command);
 
-        return processedEvent.getFuture()::join;
+        var future = processedEvent.getFuture().whenComplete((response, t) -> {
+            if (t == null) {
+                auditor.processed(command, response);
+            } else {
+                command.setError(t.getMessage());
+
+                auditor.error(command);
+            }
+        });
+
+        return future::join;
     }
 
     @Override
@@ -82,17 +93,11 @@ public class DisruptorCommandExecutor implements CommandExecutor, Closeable {
     @SuppressWarnings({ "unchecked", "rawtypes" })
     public static class CompleteableCommandEventHandler implements EventHandler<CommandEvent> {
 
-        private final List<CommandMiddleware> middlewares;
-
         private final CommandRouter router;
 
         @Override
-        public void onEvent(CommandEvent event, long sequence, boolean endOfBatch) throws Exception {
+        public void onEvent(CommandEvent event, long sequence, boolean endOfBatch) {
             try {
-                for (CommandMiddleware middleware : middlewares) {
-                    middleware.invoke(event.getCommand());
-                }
-
                 var handler = router.route(event.getCommand());
 
                 event.getFuture().complete(handler.handle(event.getCommand()));

fineract-command/src/main/java/org/apache/fineract/command/implementation/SynchronousCommandExecutor.java

@@ -18,37 +18,45 @@
  */
 package org.apache.fineract.command.implementation;
 
-import java.util.List;
 import java.util.function.Supplier;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.command.core.Command;
-import org.apache.fineract.command.core.CommandExecutor;
+import org.apache.fineract.command.core.CommandAuditor;
 import org.apache.fineract.command.core.CommandHandler;
-import org.apache.fineract.command.core.CommandMiddleware;
 import org.apache.fineract.command.core.CommandRouter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
 
 @Slf4j
-@RequiredArgsConstructor
 @Component
 @ConditionalOnProperty(value = "fineract.command.executor", havingValue = "sync")
 @SuppressWarnings({ "unchecked" })
-public class SynchronousCommandExecutor implements CommandExecutor {
+public class SynchronousCommandExecutor extends BaseCommandExecutor {
 
-    private final List<CommandMiddleware> middlewares;
-
-    private final CommandRouter router;
+    public SynchronousCommandExecutor(CommandRouter router, CommandAuditor auditor) {
+        super(router, auditor);
+    }
 
     @Override
-    public <REQ, RES> Supplier<RES> execute(Command<REQ> command) {
-        for (CommandMiddleware middleware : middlewares) {
-            middleware.invoke(command);
-        }
-
+    public <REQ, RES> Supplier<RES> execute(final Command<REQ> command) {
         CommandHandler<REQ, RES> handler = router.route(command);
 
-        return () -> handler.handle(command);
+        return () -> {
+            try {
+                auditor.processing(command);
+
+                RES response = handler.handle(command);
+
+                auditor.processed(command, response);
+
+                return response;
+            } catch (Exception e) {
+                command.setError(e.getMessage());
+
+                auditor.error(command);
+
+                throw e;
+            }
+        };
     }
 }

fineract-command/src/main/java/org/apache/fineract/command/persistence/converter/JsonNodeReadingConverter.java

@@ -20,40 +20,22 @@ package org.apache.fineract.command.persistence.converter;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import jakarta.persistence.AttributeConverter;
-import jakarta.persistence.Converter;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.ReadingConverter;
+import org.springframework.stereotype.Component;
 
-@Slf4j
+@Component
 @RequiredArgsConstructor
-@Converter
-public class JsonAttributeConverter implements AttributeConverter<JsonNode, String> {
-    // TODO: it would be nicer to use a native JSON type on the database side, but not every system supports this;
-    // string/text are the lowest common denominator that should work on every database
+@ReadingConverter
+public class JsonNodeReadingConverter implements Converter<String, JsonNode> {
 
     private final ObjectMapper mapper;
 
-    @Override
     @SneakyThrows
-    public String convertToDatabaseColumn(JsonNode source) {
-        if (source != null) {
-            return mapper.writeValueAsString(source);
-        }
-
-        // TODO: throw exception?
-        return null;
-    }
-
     @Override
-    @SneakyThrows
-    public JsonNode convertToEntityAttribute(String source) {
-        if (source != null) {
-            return mapper.readTree(source);
-        }
-
-        // TODO: throw exception?
-        return null;
+    public JsonNode convert(String source) {
+        return mapper.readTree(source);
     }
 }

fineract-command/src/main/java/org/apache/fineract/command/persistence/converter/JsonNodeWritingConverter.java

@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.fineract.command.persistence.converter;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.WritingConverter;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+@WritingConverter
+public class JsonNodeWritingConverter implements Converter<JsonNode, String> {
+
+    private final ObjectMapper mapper;
+
+    @SneakyThrows
+    @Override
+    public String convert(JsonNode source) {
+        return mapper.writeValueAsString(source);
+    }
+}

fineract-command/src/main/java/org/apache/fineract/command/persistence/domain/CommandEntity.java

@@ -19,65 +19,73 @@
 package org.apache.fineract.command.persistence.domain;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import jakarta.persistence.Column;
-import jakarta.persistence.Convert;
-import jakarta.persistence.Entity;
-import jakarta.persistence.GeneratedValue;
-import jakarta.persistence.GenerationType;
-import jakarta.persistence.PostLoad;
-import jakarta.persistence.PrePersist;
-import jakarta.persistence.Table;
-import jakarta.persistence.Transient;
 import java.io.Serial;
 import java.io.Serializable;
-import java.time.OffsetDateTime;
-import java.util.UUID;
-import lombok.AccessLevel;
+import java.time.Instant;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.ToString;
 import lombok.experimental.FieldNameConstants;
-import org.apache.fineract.command.persistence.converter.JsonAttributeConverter;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.Table;
 
 @Getter
 @Setter
 @ToString
 @FieldNameConstants
-@Entity
-@Table(name = "m_command")
+@Table("m_command")
 public class CommandEntity implements Serializable {
 
     @Serial
     private static final long serialVersionUID = 1L;
 
-    @jakarta.persistence.Id
-    @GeneratedValue(strategy = GenerationType.IDENTITY)
-    @Column(name = "id")
+    @Id
+    @Column("id")
     private Long id;
 
-    @Transient
-    @Setter(value = AccessLevel.NONE)
-    private boolean isNew = true;
+    @Column("created_at")
+    private Instant createdAt;
 
-    @Column(name = "command_id")
-    private UUID commandId;
+    @Column("updated_at")
+    private Instant updatedAt;
 
-    @Column(name = "created_at")
-    private OffsetDateTime createdAt;
+    @Column("executed_at")
+    private Instant executedAt;
 
-    @Column(name = "tenant_id")
-    private String tenantId;
+    @Column("approved_at")
+    private Instant approvedAt;
 
-    @Column(name = "username")
-    private String username;
+    @Column("rejected_at")
+    private Instant rejectedAt;
 
-    @Column(name = "payload")
-    @Convert(converter = JsonAttributeConverter.class)
-    private JsonNode payload;
+    @Column("initiated_by_username")
+    private String initiatedByUsername;
 
-    @PrePersist
-    @PostLoad
-    void markNotNew() {
-        this.isNew = false;
-    }
+    @Column("executed_by_username")
+    private String executedByUsername;
+
+    @Column("approved_by_username")
+    private String approvedByUsername;
+
+    @Column("rejected_by_username")
+    private String rejectedByUsername;
+
+    @Column("idempotency_key")
+    private String idempotencyKey;
+
+    @Column("state")
+    private CommandEntityState state;
+
+    @Column("error")
+    private String error;
+
+    @Column("ip_address")
+    private String ipAddress;
+
+    @Column("request")
+    private JsonNode request;
+
+    @Column("response")
+    private JsonNode response;
 }

fineract-command/src/main/java/org/apache/fineract/command/persistence/domain/CommandEntityState.java

@@ -16,10 +16,8 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.command.core;
+package org.apache.fineract.command.persistence.domain;
 
-@FunctionalInterface
-public interface CommandMiddleware {
-
-    void invoke(Command<?> command);
+public enum CommandEntityState {
+    INVALID, PROCESSED, AWAITING_APPROVAL, REJECTED, UNDER_PROCESSING, ERROR
 }

fineract-command/src/main/java/org/apache/fineract/command/persistence/domain/CommandRepository.java

@@ -18,7 +18,11 @@
  */
 package org.apache.fineract.command.persistence.domain;
 
-import org.springframework.data.jpa.repository.JpaRepository;
-import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+import java.util.Optional;
+import org.springframework.data.repository.ListCrudRepository;
+import org.springframework.data.repository.query.QueryByExampleExecutor;
 
-public interface CommandRepository extends JpaRepository<CommandEntity, Long>, JpaSpecificationExecutor<CommandEntity> {}
+public interface CommandRepository extends ListCrudRepository<CommandEntity, Long>, QueryByExampleExecutor<CommandEntity> {
+
+    Optional<CommandEntity> findOneByIdempotencyKey(String idempotencyKey);
+}

fineract-command/src/main/java/org/apache/fineract/command/persistence/mapping/CommandJsonMapper.java

@@ -18,6 +18,8 @@
  */
 package org.apache.fineract.command.persistence.mapping;
 
+import static org.apache.fineract.command.core.CommandConstants.COMMAND_JSON_CLASS_ATTRIBUTE;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -29,9 +31,8 @@ import org.springframework.stereotype.Component;
 @Slf4j
 @RequiredArgsConstructor
 @Component
-final class CommandJsonMapper {
+public final class CommandJsonMapper {
 
-    private static final String CLASS_ATTRIBUTE = "@class";
     private final ObjectMapper mapper;
 
     @SuppressWarnings("unchecked")
@@ -40,7 +41,7 @@ final class CommandJsonMapper {
             return null;
         }
 
-        var canonicalName = source.get(CLASS_ATTRIBUTE).asText();
+        var canonicalName = source.get(COMMAND_JSON_CLASS_ATTRIBUTE).asText();
 
         try {
             return (T) mapper.convertValue(source, Class.forName(canonicalName));
@@ -58,7 +59,7 @@ final class CommandJsonMapper {
 
         var json = mapper.convertValue(source, ObjectNode.class);
 
-        json.set(CLASS_ATTRIBUTE, new TextNode(source.getClass().getCanonicalName()));
+        json.set(COMMAND_JSON_CLASS_ATTRIBUTE, new TextNode(source.getClass().getCanonicalName()));
 
         return json;
     }

fineract-command/src/main/java/org/apache/fineract/command/persistence/mapping/CommandMapper.java

@@ -30,14 +30,42 @@ import org.mapstruct.Mapping;
 @Mapper(componentModel = SPRING, injectionStrategy = CONSTRUCTOR, uses = { CommandJsonMapper.class })
 public interface CommandMapper {
 
-    @Mapping(ignore = true, target = "id")
-    @Mapping(source = "id", target = "commandId")
+    @Mapping(source = "commandId", target = "id")
     @Mapping(source = "createdAt", target = "createdAt")
-    @Mapping(source = "tenantId", target = "tenantId")
-    @Mapping(source = "username", target = "username")
-    @Mapping(source = "payload", target = "payload")
+    @Mapping(source = "updatedAt", target = "updatedAt")
+    @Mapping(source = "executedAt", target = "executedAt")
+    @Mapping(source = "approvedAt", target = "approvedAt")
+    @Mapping(source = "rejectedAt", target = "rejectedAt")
+    @Mapping(source = "initiatedByUsername", target = "initiatedByUsername")
+    @Mapping(source = "executedByUsername", target = "executedByUsername")
+    @Mapping(source = "approvedByUsername", target = "approvedByUsername")
+    @Mapping(source = "rejectedByUsername", target = "rejectedByUsername")
+    @Mapping(source = "payload", target = "request")
+    @Mapping(source = "ipAddress", target = "ipAddress")
+    @Mapping(source = "idempotencyKey", target = "idempotencyKey")
+    @Mapping(source = "error", target = "error")
+    @Mapping(ignore = true, target = "state")
+    @Mapping(ignore = true, target = "response")
     CommandEntity map(Command source);
 
+    @Mapping(source = "source.commandId", target = "id")
+    @Mapping(source = "source.createdAt", target = "createdAt")
+    @Mapping(source = "source.updatedAt", target = "updatedAt")
+    @Mapping(source = "source.executedAt", target = "executedAt")
+    @Mapping(source = "source.approvedAt", target = "approvedAt")
+    @Mapping(source = "source.rejectedAt", target = "rejectedAt")
+    @Mapping(source = "source.initiatedByUsername", target = "initiatedByUsername")
+    @Mapping(source = "source.executedByUsername", target = "executedByUsername")
+    @Mapping(source = "source.approvedByUsername", target = "approvedByUsername")
+    @Mapping(source = "source.rejectedByUsername", target = "rejectedByUsername")
+    @Mapping(source = "source.payload", target = "request")
+    @Mapping(source = "source.ipAddress", target = "ipAddress")
+    @Mapping(source = "source.idempotencyKey", target = "idempotencyKey")
+    @Mapping(source = "source.error", target = "error")
+    @Mapping(source = "response", target = "response")
+    @Mapping(ignore = true, target = "state")
+    CommandEntity map(Command source, Object response);
+
     @InheritInverseConfiguration
     Command map(CommandEntity source);
 }

fineract-command/src/main/java/org/apache/fineract/command/starter/CommandConfiguration.java

@@ -23,8 +23,6 @@ import com.lmax.disruptor.WaitStrategy;
 import com.lmax.disruptor.YieldingWaitStrategy;
 import com.lmax.disruptor.dsl.Disruptor;
 import com.lmax.disruptor.util.DaemonThreadFactory;
-import java.util.List;
-import org.apache.fineract.command.core.CommandMiddleware;
 import org.apache.fineract.command.core.CommandProperties;
 import org.apache.fineract.command.core.CommandRouter;
 import org.apache.fineract.command.implementation.DisruptorCommandExecutor;
@@ -50,15 +48,14 @@ class CommandConfiguration {
 
     @Bean
     @ConditionalOnProperty(value = "fineract.command.executor", havingValue = "disruptor")
-    Disruptor<?> disruptor(CommandProperties properties, WaitStrategy waitStrategy, List<CommandMiddleware> middlewares,
-            CommandRouter router) {
+    Disruptor<?> disruptor(CommandProperties properties, WaitStrategy waitStrategy, CommandRouter router) {
         // TODO: make this more configurable
 
         // Create the disruptor
         Disruptor<DisruptorCommandExecutor.CommandEvent> disruptor = new Disruptor<>(DisruptorCommandExecutor.CommandEvent::new,
                 properties.getRingBufferSize(), DaemonThreadFactory.INSTANCE, properties.getProducerType(), waitStrategy);
 
-        disruptor.handleEventsWith(new DisruptorCommandExecutor.CompleteableCommandEventHandler(middlewares, router));
+        disruptor.handleEventsWith(new DisruptorCommandExecutor.CompleteableCommandEventHandler(router));
         disruptor.setDefaultExceptionHandler(new IgnoreExceptionHandler());
 
         // Start the disruptor

fineract-command/src/main/java/org/apache/fineract/command/starter/CommandPersistenceConfiguration.java

@@ -18,9 +18,23 @@
  */
 package org.apache.fineract.command.starter;
 
+import java.util.List;
+import org.apache.fineract.command.persistence.converter.JsonNodeReadingConverter;
+import org.apache.fineract.command.persistence.converter.JsonNodeWritingConverter;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.jdbc.core.convert.JdbcCustomConversions;
+import org.springframework.data.jdbc.repository.config.EnableJdbcRepositories;
 
 @Configuration
+@EnableJdbcRepositories(basePackages = { "org.apache.fineract.command.persistence" })
 @ComponentScan("org.apache.fineract.command.persistence")
-class CommandPersistenceConfiguration {}
+class CommandPersistenceConfiguration {
+
+    @Bean
+    JdbcCustomConversions customConversions(JsonNodeReadingConverter jsonNodeReadingConverter,
+            JsonNodeWritingConverter jsonNodeWritingConverter) {
+        return new JdbcCustomConversions(List.of(jsonNodeReadingConverter, jsonNodeWritingConverter));
+    }
+}

fineract-command/src/main/resources/db/changelog/tenant/module/command/module-changelog-master.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
+    <include relativeToChangelogFile="true" file="parts/0001_command_init.xml"/>
+    <include relativeToChangelogFile="true" file="parts/0002_command_fix_structure.xml"/>
+</databaseChangeLog>

fineract-command/src/main/resources/db/changelog/tenant/module/command/parts/0001_command_init.xml

@@ -23,6 +23,11 @@
                    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                    xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
     <changeSet author="fineract" id="command-00001" context="postgresql">
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <tableExists tableName="m_command"/>
+            </not>
+        </preConditions>
         <createTable tableName="m_command">
             <column autoIncrement="true" name="id" type="bigint" remarks="Internal ID">
                 <constraints nullable="false" primaryKey="true"/>
@@ -35,6 +40,11 @@
         </createTable>
     </changeSet>
     <changeSet author="fineract" id="command-00002" context="mysql">
+        <preConditions onFail="MARK_RAN">
+            <not>
+                <tableExists tableName="command"/>
+            </not>
+        </preConditions>
         <createTable tableName="command">
             <column autoIncrement="true" name="id" type="bigint" remarks="Internal ID">
                 <constraints nullable="false" primaryKey="true"/>

fineract-command/src/main/resources/db/changelog/tenant/module/command/parts/0002_command_fix_structure.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
+    <changeSet author="fineract" id="command-00002" context="mysql">
+        <renameTable oldTableName="command" newTableName="m_command" />
+        <addColumn tableName="m_command">
+            <column name="updated_at" type="datetime" />
+            <column name="executed_at" type="datetime" />
+            <column name="approved_at" type="datetime" />
+            <column name="rejected_at" type="datetime" />
+        </addColumn>
+    </changeSet>
+    <changeSet author="fineract" id="command-00003" context="postgresql">
+        <addColumn tableName="m_command">
+            <column name="updated_at" type="timestamp with time zone" />
+            <column name="executed_at" type="timestamp with time zone" />
+            <column name="approved_at" type="timestamp with time zone" />
+            <column name="rejected_at" type="timestamp with time zone" />
+        </addColumn>
+    </changeSet>
+    <changeSet author="fineract" id="command-00004" context="postgresql,mysql">
+        <renameColumn tableName="m_command" oldColumnName="username" newColumnName="initiated_by_username" columnDataType="varchar(255)" />
+        <renameColumn tableName="m_command" oldColumnName="payload" newColumnName="request" columnDataType="text" />
+
+        <addColumn tableName="m_command">
+            <column name="idempotency_key" type="varchar(255)" />
+            <column name="executed_by_username" type="varchar(255)" />
+            <column name="approved_by_username" type="varchar(255)" />
+            <column name="rejected_by_username" type="varchar(255)" />
+            <column name="ip_address" type="varchar(255)" />
+            <column name="state" type="varchar(255)" />
+            <column name="response" type="text" />
+            <column name="error" type="text" />
+        </addColumn>
+        <createIndex tableName="m_command" indexName="m_command_idempotency_key_index">
+            <column name="idempotency_key" />
+        </createIndex>
+    </changeSet>
+</databaseChangeLog>

fineract-command/src/main/resources/jpa/static-weaving/module/fineract-command/persistence.xml

@@ -1,43 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-
-    Licensed to the Apache Software Foundation (ASF) under one
-    or more contributor license agreements. See the NOTICE file
-    distributed with this work for additional information
-    regarding copyright ownership. The ASF licenses this file
-    to you under the Apache License, Version 2.0 (the
-    "License"); you may not use this file except in compliance
-    with the License. You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-    Unless required by applicable law or agreed to in writing,
-    software distributed under the License is distributed on an
-    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-    KIND, either express or implied. See the License for the
-    specific language governing permissions and limitations
-    under the License.
-
--->
-
-
-<persistence version="2.0"
-             xmlns="http://java.sun.com/xml/ns/persistence"
-             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-             xsi:schemaLocation="http://java.sun.com/xml/ns/persistence http://java.sun.com/xml/ns/persistence/persistence_2_0.xsd">
-
-    <!--  This file is only used for static weaving, nothing more.  -->
-    <!--  You can find the runtime configuration in the JPAConfig class  -->
-    <persistence-unit name="jpa-pu" transaction-type="RESOURCE_LOCAL">
-        <provider>org.eclipse.persistence.jpa.PersistenceProvider</provider>
-
-        <!-- Command Module Entities -->
-        <class>org.apache.fineract.command.persistence.domain.CommandEntity</class>
-        <class>org.apache.fineract.command.persistence.converter.JsonAttributeConverter</class>
-
-        <exclude-unlisted-classes>false</exclude-unlisted-classes>
-        <properties>
-            <property name="eclipselink.weaving" value="static" />
-        </properties>
-    </persistence-unit>
-</persistence>