feat(account): add repository gateway (#9)

infrastructure/build.gradle

@@ -2,6 +2,9 @@ buildscript {
     repositories {
         mavenCentral()
     }
+    dependencies {
+        classpath 'org.flywaydb:flyway-mysql:9.11.0'
+    }
 }
 
 plugins {
@@ -10,6 +13,7 @@ plugins {
     id 'jacoco-report-aggregation'
     id 'org.springframework.boot' version '3.3.2'
     id 'io.spring.dependency-management' version '1.1.6'
+    id 'org.flywaydb.flyway' version '9.11.0'
 }
 
 group 'io.github.gabrmsouza.subscription.infrastructure'
@@ -27,6 +31,8 @@ dependencies {
     implementation(project(":domain"))
     implementation(project(":application"))
 
+    implementation("com.mysql:mysql-connector-j")
+
     implementation('ch.qos.logback:logback-classic:1.4.12')
     implementation('net.logstash.logback:logstash-logback-encoder:7.0.1')
 
@@ -39,6 +45,7 @@ dependencies {
     implementation('org.springframework.boot:spring-boot-starter-security')
     implementation('org.springframework.boot:spring-boot-starter-oauth2-resource-server')
     implementation("org.springframework.boot:spring-boot-starter-validation")
+    implementation("org.springframework.boot:spring-boot-starter-data-jdbc")
 
     implementation("org.springdoc:springdoc-openapi-starter-webmvc-ui:$springdoc") {
         exclude group: 'org.springdoc', module: 'springdoc-openapi-ui'
@@ -51,6 +58,10 @@ dependencies {
     testImplementation("org.springframework.cloud:spring-cloud-contract-wiremock")
     testImplementation("org.springframework.boot:spring-boot-testcontainers")
 
+    testImplementation("com.h2database:h2")
+    testImplementation("org.flywaydb:flyway-core")
+    testImplementation("org.flywaydb:flyway-mysql")
+
     testImplementation('org.testcontainers:testcontainers:1.19.8')
     testImplementation('org.testcontainers:junit-jupiter:1.19.8')
 }
@@ -61,6 +72,12 @@ dependencyManagement {
     }
 }
 
+flyway {
+    url = System.getenv('FLYWAY_DB') ?: 'jdbc:mysql://localhost:3306/adm_videos'
+    user = System.getenv('FLYWAY_USER') ?: 'root'
+    password = System.getenv('FLYWAY_PASS') ?: '123456'
+}
+
 testCodeCoverageReport {
     reports {
         xml.required = true

infrastructure/src/main/java/io/github/gabrmsouza/subscription/infrastructure/configuration/JdbcConfiguration.java

@@ -0,0 +1,15 @@
+package io.github.gabrmsouza.subscription.infrastructure.configuration;
+
+import io.github.gabrmsouza.subscription.infrastructure.jdbc.DatabaseClient;
+import io.github.gabrmsouza.subscription.infrastructure.jdbc.JdbcClientAdapter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.simple.JdbcClient;
+
+@Configuration(proxyBeanMethods = false)
+public class JdbcConfiguration {
+    @Bean
+    DatabaseClient databaseClient(JdbcClient jdbcClient) {
+        return new JdbcClientAdapter(jdbcClient);
+    }
+}

infrastructure/src/main/java/io/github/gabrmsouza/subscription/infrastructure/gateway/repository/AccountInMemoryRepository.java

@@ -5,13 +5,15 @@
 import io.github.gabrmsouza.subscription.domain.account.AccountId;
 import io.github.gabrmsouza.subscription.domain.account.idp.UserId;
 import io.github.gabrmsouza.subscription.domain.utils.IDUtils;
+import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Component;
 
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
 @Component
+@Profile("dev")
 public class AccountInMemoryRepository implements AccountGateway {
 
     private Map<String, Account> db = new ConcurrentHashMap<>();

infrastructure/src/main/java/io/github/gabrmsouza/subscription/infrastructure/gateway/repository/AccountJdbcRepository.java

@@ -0,0 +1,144 @@
+package io.github.gabrmsouza.subscription.infrastructure.gateway.repository;
+
+import io.github.gabrmsouza.subscription.domain.account.Account;
+import io.github.gabrmsouza.subscription.domain.account.AccountGateway;
+import io.github.gabrmsouza.subscription.domain.account.AccountId;
+import io.github.gabrmsouza.subscription.domain.account.idp.UserId;
+import io.github.gabrmsouza.subscription.domain.person.Address;
+import io.github.gabrmsouza.subscription.domain.person.Document;
+import io.github.gabrmsouza.subscription.domain.person.Email;
+import io.github.gabrmsouza.subscription.domain.person.Name;
+import io.github.gabrmsouza.subscription.domain.utils.IDUtils;
+import io.github.gabrmsouza.subscription.infrastructure.jdbc.DatabaseClient;
+import io.github.gabrmsouza.subscription.infrastructure.jdbc.RowMap;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+@Repository
+public class AccountJdbcRepository implements AccountGateway {
+    private final DatabaseClient database;
+    private final EventJdbcRepository eventRepository;
+
+    public AccountJdbcRepository(final DatabaseClient databaseClient, final EventJdbcRepository eventRepository) {
+        this.database = Objects.requireNonNull(databaseClient);
+        this.eventRepository = Objects.requireNonNull(eventRepository);
+    }
+
+    @Override
+    public AccountId nextId() {
+        return new AccountId(IDUtils.uuid());
+    }
+
+    @Override
+    public Optional<Account> accountOfId(final AccountId anId) {
+        final var sql = """
+                SELECT
+                    id, version, idp_user_id, email, firstname, lastname, document_number, document_type, address_zip_code, address_number, address_complement, address_country
+                FROM accounts
+                WHERE id = :id
+                """;
+        return this.database.queryOne(sql, Map.of("id", anId.value()), accountMapper());
+    }
+
+    @Override
+    public Optional<Account> accountOfUserId(final UserId userId) {
+        final var sql = """
+                SELECT
+                    id, version, idp_user_id, email, firstname, lastname, document_number, document_type, address_zip_code, address_number, address_complement, address_country
+                FROM accounts
+                WHERE idp_user_id = :userId
+                """;
+        return this.database.queryOne(sql, Map.of("userId", userId.value()), accountMapper());
+    }
+
+    @Override
+    @Transactional(propagation = Propagation.REQUIRED)
+    public Account save(final Account anAccount) {
+        if (anAccount.version() == 0) {
+            create(anAccount);
+        } else {
+            update(anAccount);
+        }
+        this.eventRepository.saveAll(anAccount.domainEvents());
+        return anAccount;
+    }
+
+    private void create(final Account account) {
+        final var sql = """
+                INSERT INTO accounts (id, version, idp_user_id, email, firstname, lastname, document_number, document_type, address_zip_code, address_number, address_complement, address_country)
+                VALUES (:id, (:version + 1), :userId, :email, :firstname, :lastname, :documentNumber, :documentType, :addressZipCode, :addressNumber, :addressComplement, :addressCountry)
+                """;
+        executeUpdate(sql, account);
+    }
+
+    private void update(final Account account) {
+        final var sql = """
+                UPDATE accounts
+                SET
+                    version = :version + 1,
+                    idp_user_id = :userId,
+                    email = :email,
+                    firstname = :firstname,
+                    lastname = :lastname,
+                    document_number = :documentNumber,
+                    document_type = :documentType,
+                    address_zip_code = :addressZipCode,
+                    address_number = :addressNumber,
+                    address_complement = :addressComplement,
+                    address_country = :addressCountry
+                WHERE id = :id and version = :version
+                """;
+
+        if (executeUpdate(sql, account) == 0) {
+            throw new IllegalArgumentException("Account with id %s and version %s was not found".formatted(account.id().value(), account.version()));
+        }
+    }
+
+    private int executeUpdate(final String sql, final Account account) {
+        final var params = new HashMap<String, Object>();
+        params.put("version", account.version());
+        params.put("userId", account.userId().value());
+        params.put("email", account.email().value());
+        params.put("firstname", account.name().firstname());
+        params.put("lastname", account.name().lastname());
+        params.put("documentNumber", account.document().value());
+        params.put("documentType", account.document().type());
+
+        final var address = account.billingAddress();
+        params.put("addressZipCode", address != null ? address.zipcode() : "");
+        params.put("addressNumber", address != null ? address.number() : "");
+        params.put("addressComplement", address != null ? address.complement() : "");
+        params.put("addressCountry", address != null ? address.country() : "");
+        params.put("id", account.id().value());
+
+        return this.database.update(sql, params);
+    }
+
+    private RowMap<Account> accountMapper() {
+        return (rs) -> {
+            final var zipCode = rs.getString("address_zip_code");
+            return Account.with(
+                    new AccountId(rs.getString("id")),
+                    rs.getInt("version"),
+                    new UserId(rs.getString("idp_user_id")),
+                    new Email(rs.getString("email")),
+                    new Name(rs.getString("firstname"), rs.getString("lastname")),
+                    Document.create(rs.getString("document_number"), rs.getString("document_type")),
+                    zipCode != null && !zipCode.isBlank() ?
+                            new Address(
+                                    zipCode,
+                                    rs.getString("address_number"),
+                                    rs.getString("address_complement"),
+                                    rs.getString("address_country")
+                            ) :
+                            null
+            );
+        };
+    }
+}

infrastructure/src/main/java/io/github/gabrmsouza/subscription/infrastructure/gateway/repository/EventJdbcRepository.java

@@ -0,0 +1,99 @@
+package io.github.gabrmsouza.subscription.infrastructure.gateway.repository;
+
+import io.github.gabrmsouza.subscription.domain.DomainEvent;
+import io.github.gabrmsouza.subscription.domain.utils.InstantUtils;
+import io.github.gabrmsouza.subscription.infrastructure.jdbc.DatabaseClient;
+import io.github.gabrmsouza.subscription.infrastructure.jdbc.JdbcUtils;
+import io.github.gabrmsouza.subscription.infrastructure.jdbc.RowMap;
+import io.github.gabrmsouza.subscription.infrastructure.json.Json;
+import org.springframework.stereotype.Repository;
+
+import java.time.Instant;
+import java.util.*;
+
+@Repository
+public class EventJdbcRepository {
+
+    private final DatabaseClient database;
+
+    public EventJdbcRepository(final DatabaseClient databaseClient) {
+        this.database = Objects.requireNonNull(databaseClient);
+    }
+
+    public Optional<DomainEvent> eventOfIdAndUnprocessed(final Long eventId) {
+        final var sql = "SELECT event_id, processed, aggregate_id, aggregate_type, event_type, event_date, event_data FROM events WHERE event_id = :eventId AND processed = false";
+        final var params = Map.<String, Object>of("eventId", eventId);
+        return this.database.queryOne(sql, params, eventMapper())
+                .map(this::toDomainEvent);
+    }
+
+    public List<DomainEvent> allEventsOfAggregate(final String aggregateId, final String aggregateType) {
+        final var sql = "SELECT event_id, processed, aggregate_id, aggregate_type, event_type, event_date, event_data FROM events WHERE aggregate_id = :aggregateId and aggregate_type = :aggregateType";
+        final var params = Map.<String, Object>of("aggregateId", aggregateId, "aggregateType", aggregateType);
+        return this.database.query(sql, params, eventMapper()).stream()
+                .map(this::toDomainEvent)
+                .toList();
+    }
+
+    public void markAsProcessed(final Long eventId) {
+        final var sql = "UPDATE events SET processed = true WHERE event_id = :id";
+        if (this.database.update(sql, Map.of("id", eventId)) == 0) {
+            throw new IllegalArgumentException("Event with id %s was not found".formatted(eventId));
+        }
+    }
+
+    public void saveAll(final Collection<DomainEvent> events) {
+        for (var ev : events) {
+            this.insertEvent(Event.newEvent(ev.aggregateId(), ev.aggregateType(), ev.getClass().getCanonicalName(), Json.writeValueAsString(ev)));
+        }
+    }
+
+    private void insertEvent(final Event event) {
+        final var sql = "INSERT INTO events (processed, aggregate_id, aggregate_type, event_type, event_date, event_data) VALUES (:processed, :aggregateId, :aggregateType, :eventType, :eventDate, :eventData)";
+
+        final var params = new HashMap<String, Object>();
+        params.put("processed", event.processed());
+        params.put("aggregateId", event.aggregateId());
+        params.put("aggregateType", event.aggregateType());
+        params.put("eventType", event.eventType());
+        params.put("eventDate", event.eventDate());
+        params.put("eventData", event.eventData());
+
+        this.database.update(sql, params);
+    }
+
+    private DomainEvent toDomainEvent(final Event event) {
+        try {
+            return (DomainEvent) Json.readTree(event.eventData(), Class.forName(event.eventType()));
+        } catch (ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private RowMap<Event> eventMapper() {
+        return (rs) -> new Event(
+                rs.getLong("event_id"),
+                rs.getBoolean("processed"),
+                rs.getString("aggregate_id"),
+                rs.getString("aggregate_type"),
+                rs.getString("event_type"),
+                JdbcUtils.getInstant(rs, "event_date"),
+                rs.getString("event_data")
+        );
+    }
+
+    private record Event(
+            Long eventId,
+            boolean processed,
+            String aggregateId,
+            String aggregateType,
+            String eventType,
+            Instant eventDate,
+            String eventData
+    ) {
+
+        public static Event newEvent(String aggregateId, String aggregateType, String eventType, String data) {
+            return new Event(null, false, aggregateId, aggregateType, eventType, InstantUtils.now(), data);
+        }
+    }
+}

infrastructure/src/main/java/io/github/gabrmsouza/subscription/infrastructure/jdbc/DatabaseClient.java

@@ -0,0 +1,13 @@
+package io.github.gabrmsouza.subscription.infrastructure.jdbc;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public interface DatabaseClient {
+    <T> Optional<T> queryOne(String sql, Map<String, Object> params, RowMap<T> mapper);
+    <T> List<T> query(String sql, RowMap<T> mapper);
+    <T> List<T> query(String sql, Map<String, Object> params, RowMap<T> mapper);
+    int update(String sql, Map<String, Object> params);
+    Number insert(String sql, Map<String, Object> params);
+}