JAMES-2433 Implement EventStore for a JPA backend

event-sourcing/event-store-jpa/pom.xml

@@ -0,0 +1,128 @@
+<?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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.james</groupId>
+        <artifactId>event-sourcing</artifactId>
+        <version>3.9.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>event-sourcing-event-store-jpa</artifactId>
+
+    <name>Apache James :: Event sourcing :: Event Store :: JPA</name>
+    <description>JPA implementation for James Event Store</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>apache-james-backends-jpa</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>apache-james-backends-jpa</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>event-sourcing-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>event-sourcing-core</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>event-sourcing-event-store-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>event-sourcing-event-store-api</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>event-sourcing-pojo</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${james.groupId}</groupId>
+            <artifactId>testing-base</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.derby</groupId>
+            <artifactId>derby</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.openjpa</groupId>
+                <artifactId>openjpa-maven-plugin</artifactId>
+                <version>${apache.openjpa.version}</version>
+                <configuration>
+                    <includes>org/apache/james/eventsourcing/eventstore/jpa/model/JPAEvent.class</includes>
+                    <addDefaultConstructor>true</addDefaultConstructor>
+                    <enforcePropertyRestrictions>true</enforcePropertyRestrictions>
+                    <toolProperties>
+                        <property>
+                            <name>log</name>
+                            <value>TOOL=TRACE</value>
+                        </property>
+                        <property>
+                            <name>metaDataFactory</name>
+                            <value>jpa(Types=org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent)</value>
+                        </property>
+                    </toolProperties>
+                    <persistenceXmlFile>${basedir}/src/test/resources/persistence.xml</persistenceXmlFile>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>enhancer</id>
+                        <goals>
+                            <goal>enhance</goal>
+                        </goals>
+                        <phase>process-classes</phase>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/JPAEventStore.java

@@ -0,0 +1,118 @@
+/****************************************************************
+ * 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.james.eventsourcing.eventstore.jpa;
+
+
+import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY;
+import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY;
+
+import jakarta.inject.Inject;
+import jakarta.persistence.EntityManagerFactory;
+import jakarta.persistence.PersistenceUnit;
+
+import org.apache.james.backends.jpa.TransactionRunner;
+import org.apache.james.eventsourcing.AggregateId;
+import org.apache.james.eventsourcing.Event;
+import org.apache.james.eventsourcing.eventstore.EventStore;
+import org.apache.james.eventsourcing.eventstore.EventStoreFailedException;
+import org.apache.james.eventsourcing.eventstore.History;
+import org.apache.james.eventsourcing.eventstore.JsonEventSerializer;
+import org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent;
+import org.reactivestreams.Publisher;
+
+import com.github.fge.lambdas.Throwing;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+import reactor.core.publisher.Mono;
+import scala.collection.Iterable;
+import scala.collection.JavaConverters;
+
+@PersistenceUnit(unitName = "James")
+public class JPAEventStore implements EventStore {
+
+    /**
+     * The entity manager to access the database.
+     */
+    private EntityManagerFactory entityManagerFactory;
+
+    /**
+     * The JSON serializer to serialize the event data.
+     */
+    private JsonEventSerializer jsonEventSerializer;
+
+    /**
+     * Constructs a JPAEventStore.
+     */
+    @Inject
+    public JPAEventStore(EntityManagerFactory entityManagerFactory, JsonEventSerializer jsonEventSerializer) {
+        this.jsonEventSerializer = jsonEventSerializer;
+        this.entityManagerFactory = entityManagerFactory;
+    }
+
+    @Override
+    public Publisher<Void> appendAll(Iterable<Event> events) {
+        if (events.isEmpty()) {
+            return Mono.empty();
+        }
+        Preconditions.checkArgument(Event.belongsToSameAggregate(events));
+        AggregateId aggregateId = events.head().getAggregateId();
+        return Mono.fromRunnable(() -> new TransactionRunner(entityManagerFactory).runAndHandleException(
+            entityManager ->
+                JavaConverters.asJava(events).forEach(Throwing.consumer(e -> {
+                    JPAEvent jpaEvent = new JPAEvent(aggregateId, e.eventId(), jsonEventSerializer.serialize(e));
+                    entityManager.persist(jpaEvent);
+            })),
+            exception -> {
+                EventStoreFailedException esfe = new EventStoreFailedException("Unable to add events");
+                esfe.initCause(exception);
+                throw esfe;
+            }));
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public Publisher<History> getEventsOfAggregate(AggregateId aggregateId) {
+        Preconditions.checkNotNull(aggregateId);
+        return Mono.fromSupplier(() -> new TransactionRunner(entityManagerFactory).runAndRetrieveResult(
+            entityManager -> History.of(
+                (Event[]) entityManager.createNamedQuery(SELECT_AGGREGATE_QUERY)
+                .setParameter("aggregateId", aggregateId.asAggregateKey())
+                .getResultStream()
+                .map(Throwing.function(e -> jsonEventSerializer.deserialize(((JPAEvent) e).getEvent())))
+                .toArray(Event[]::new))));
+    }
+
+    @Override
+    public Publisher<Void> remove(AggregateId aggregateId) {
+        return Mono.fromSupplier(() -> new TransactionRunner(entityManagerFactory).runAndRetrieveResult(
+            entityManager -> {
+                entityManager.createNamedQuery(DELETE_AGGREGATE_QUERY)
+                    .setParameter("aggregateId", aggregateId.asAggregateKey())
+                    .executeUpdate();
+                return null;
+            }));
+    }
+
+    @VisibleForTesting
+    protected void removeAll() {
+        new TransactionRunner(entityManagerFactory).runAndRetrieveResult(
+            entityManager -> entityManager.createQuery("DELETE FROM JPAEvent").executeUpdate());
+    }
+}

event-sourcing/event-store-jpa/src/main/java/org/apache/james/eventsourcing/eventstore/jpa/model/JPAEvent.java

@@ -0,0 +1,124 @@
+/****************************************************************
+ * 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.james.eventsourcing.eventstore.jpa.model;
+
+import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.DELETE_AGGREGATE_QUERY;
+import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.JPAEventId;
+import static org.apache.james.eventsourcing.eventstore.jpa.model.JPAEvent.SELECT_AGGREGATE_QUERY;
+
+import java.io.Serializable;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.IdClass;
+import jakarta.persistence.Index;
+import jakarta.persistence.Lob;
+import jakarta.persistence.NamedQuery;
+import jakarta.persistence.Table;
+
+import org.apache.james.eventsourcing.AggregateId;
+import org.apache.james.eventsourcing.EventId;
+
+import com.google.common.base.Objects;
+
+
+
+/**
+ * JPAEvent class for the James Event Sourcing to be used for JPA persistence.
+ */
+@Entity(name = "JPAEvent")
+@Table(name = JPAEvent.JAMES_EVENTS, indexes = {
+    @Index(name = "AGGREGATE_ID_INDEX", columnList = "AGGREGATE_ID")
+})
+@NamedQuery(name = SELECT_AGGREGATE_QUERY, query = "SELECT e FROM JPAEvent e WHERE e.aggregateId=:aggregateId")
+@NamedQuery(name = DELETE_AGGREGATE_QUERY, query = "DELETE FROM JPAEvent e WHERE e.aggregateId=:aggregateId")
+@IdClass(JPAEventId.class)
+public class JPAEvent {
+    public static final String SELECT_AGGREGATE_QUERY = "selectAggregateEvents";
+    public static final String DELETE_AGGREGATE_QUERY = "deleteAggregateEvents";
+
+    public static final String JAMES_EVENTS = "JAMES_EVENTS";
+
+    public static class JPAEventId implements Serializable {
+
+        private static final long serialVersionUID = 1L;
+
+        private String aggregateId;
+
+        private int eventId;
+
+        public JPAEventId() {
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hashCode(aggregateId, eventId);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (this == obj) {
+                return true;
+            }
+            if (obj == null || getClass() != obj.getClass()) {
+                return false;
+            }
+            final JPAEventId other = (JPAEventId) obj;
+            return Objects.equal(this.aggregateId, other.aggregateId)
+                && Objects.equal(this.eventId, other.eventId);
+        }
+    }
+
+    @Id
+    @Column(name = "AGGREGATE_ID", nullable = false, length = 100)
+    private String aggregateId = "";
+
+    @Id
+    @Column(name = "EVENT_ID", nullable = false)
+    private int eventId;
+
+    @Lob
+    @Column(name = "EVENT", nullable = false, length = 1048576000)
+    private String event = "";
+
+    /**
+     * Default no-args constructor for JPA class enhancement.
+     * The constructor need to be public or protected to be used by JPA.
+     * See:  http://docs.oracle.com/javaee/6/tutorial/doc/bnbqa.html
+     * Do not us this constructor, it is for JPA only.
+     */
+    protected JPAEvent() {
+    }
+
+    public JPAEvent(AggregateId aggregateId, EventId eventId, String event) {
+        this.aggregateId = aggregateId.asAggregateKey();
+        this.eventId = eventId.serialize();
+        this.event = event;
+    }
+
+    public EventId getEventId() {
+        return EventId.fromSerialized(eventId);
+    }
+
+    public String getEvent() {
+        return event;
+    }
+
+}