Postgre SQL Data Repositories

@Table
@ColumnMappingStrategy
public final class Account {

    @PrimaryKey
    @SequenceGenerator
    Long id;

    @Column(length = Column.UNLIMITED_LENGTH)
    @NotUpdatable
    String email;

    @Column(length = Column.UNLIMITED_LENGTH)
    String firstName;

    @Column(length = Column.UNLIMITED_LENGTH)
    String lastName;

    BigDecimal balance;

    Role role;
}

@Table
@ColumnMappingStrategy
public final class Order {

    @PrimaryKey
    Long id;

    Long idAccount;

    Integer idProduct;

    Integer count;

    @NotInsertable
    Instant created;
}

@Table
@ColumnMappingStrategy
public final class Product {

    @PrimaryKey(autoGenerated = false)
    Integer id;

    @Column(length = Column.UNLIMITED_LENGTH)
    String name;

    BigDecimal price;

    Integer count;
}

public enum Role {

    CEO,

    Lead_Engineer,

    Systems_Architect
}

(1)
@Testcontainers
(2)
@RxMicroComponentTest(DataRepository.class)
final class DataRepositoryTestTemplate1 {

    (3)
    @Container
    private static final GenericContainer<?> POSTGRESQL_TEST_DB =
            new GenericContainer<>("rxmicro/postgres-test-db")
                    .withExposedPorts(5432); (4)

    (5)
    @WithConfig
    private static final PostgreSQLConfig CONFIG = new PostgreSQLConfig()
            .setDatabase("rxmicro")
            .setUser("rxmicro")
            .setPassword("password"); (6)

    @BeforeAll
    static void beforeAll() {
        POSTGRESQL_TEST_DB.start(); (7)
        CONFIG
                .setHost(POSTGRESQL_TEST_DB.getHost()) (8)
                .setPort(POSTGRESQL_TEST_DB.getFirstMappedPort());
    }

    private DataRepository dataRepository; (9)

    // ... test methods must be here

    @AfterAll
    static void afterAll() {
        POSTGRESQL_TEST_DB.stop(); (10)
    }
}

(1)
@Testcontainers
(2)
@RxMicroComponentTest(DataRepository.class)
final class DataRepositoryTestTemplate2 {

    (3)
    @Container
    private final GenericContainer<?> postgresqlTestDb =
            new GenericContainer<>("rxmicro/postgres-test-db")
                    .withExposedPorts(5432); (4)

    (5)
    @WithConfig
    private final PostgreSQLConfig config = new PostgreSQLConfig()
            .setDatabase("rxmicro")
            .setUser("rxmicro")
            .setPassword("password"); (6)

    private DataRepository dataRepository; (7)

    @BeforeEach
    void beforeEach() {
        config
                .setHost(postgresqlTestDb.getHost()) (8)
                .setPort(postgresqlTestDb.getFirstMappedPort());
    }

    // ... test methods must be here
}

@Table
@ColumnMappingStrategy
public final class Account {

    @PrimaryKey
    @SequenceGenerator
    Long id;

    @Column(length = Column.UNLIMITED_LENGTH)
    @NotUpdatable
    String email;

    @Column(length = Column.UNLIMITED_LENGTH)
    String firstName;

    @Column(length = Column.UNLIMITED_LENGTH)
    String lastName;

    BigDecimal balance;

    Role role;
}

@PostgreSQLRepository
public interface DataRepository {

    @Insert
    CompletableFuture<Boolean> insert1(Account account);

    @Insert
    CompletableFuture<Account> insert2(Account account);

    @Insert("INSERT INTO ${table} VALUES(nextval('account_seq'),?,?,?,?,?)")
    (1)
    @VariableValues({
            "${table}", "account"
    })
    CompletableFuture<Long> insert3(
            String email, String firstName, String lastName, BigDecimal balance, Role role
    );

    @Insert("INSERT INTO ${table} VALUES(nextval('account_seq'),?,?,?,?,?) RETURNING *")
    CompletableFuture<Account> insert4(
            String email, String firstName, String lastName, BigDecimal balance, Role role
    );

    @Insert(
            value = "INSERT INTO ${table} VALUES(nextval('account_seq'),?,?,?,?,?)",
            entityClass = Account.class
    )
    CompletableFuture<Long> insert5(
            String email, String firstName, String lastName, BigDecimal balance, Role role
    );

    @Insert(
            value = "INSERT INTO ${table} VALUES(nextval('account_seq'),?,?,?,?,?) RETURNING *",
            entityClass = Account.class
    )
    CompletableFuture<EntityFieldMap> insert6(
            String email, String firstName, String lastName, BigDecimal balance, Role role
    );

    @Insert("INSERT INTO ${table}(${inserted-columns}) VALUES(${values}) " +
            "RETURNING ${returning-columns}")
    CompletableFuture<AccountResult> insert7(Account account);

    @Insert("INSERT INTO ${table}(${inserted-columns}) VALUES(${values}) " +
            "ON CONFLICT (${id-columns}) DO UPDATE SET ${on-conflict-update-inserted-columns}" +
            "RETURNING ${returning-columns}")
    CompletableFuture<AccountResult> insert8(Account account);

    @Insert("INSERT INTO ${table}(${inserted-columns}) VALUES(${values}) " +
            "ON CONFLICT (${id-columns}) DO UPDATE SET ${on-conflict-update-inserted-columns}")
    CompletableFuture<Void> insert9(Account account);

    @Insert("INSERT INTO ${table}(${inserted-columns}) VALUES(${values}) " +
            "ON CONFLICT (${id-columns}) DO NOTHING")
    CompletableFuture<Void> insert10(Account account);

    @Insert("INSERT INTO ${table} SELECT * FROM dump RETURNING *")
    CompletableFuture<List<Account>> insertMany1();

    @Insert("INSERT INTO account SELECT * FROM dump")
    CompletableFuture<Long> insertMany2();
}

@PostgreSQLRepository
public interface DataRepository {

    @Update
    CompletableFuture<Boolean> update1(Account account);

    @Update("UPDATE ${table} SET first_name=?, last_name=? WHERE id=?")
    (1)
    @VariableValues({
            "${table}", "account"
    })
    CompletableFuture<Long> update2(String firstName, String lastName, Long id);

    @Update("UPDATE ${table} SET first_name=?, last_name=? WHERE ${by-id-filter} RETURNING *")
    CompletableFuture<Account> update3(String firstName, String lastName, Long id);

    @Update(
            value = "UPDATE ${table} SET first_name=?, last_name=? " +
                    "WHERE id = ?",
            entityClass = Account.class
    )
    CompletableFuture<Long> update4(String firstName, String lastName, Long id);

    @Update(
            value = "UPDATE ${table} SET first_name=?, last_name=? " +
                    "WHERE ${by-id-filter} RETURNING *",
            entityClass = Account.class
    )
    CompletableFuture<EntityFieldMap> update5(String firstName, String lastName, Long id);
}

@PostgreSQLRepository
public interface DataRepository {

    @Delete
    CompletableFuture<Boolean> delete1(Account account);

    @Delete("DELETE FROM ${table} WHERE balance < ?")
    (1)
    @VariableValues({
            "${table}", "account"
    })
    CompletableFuture<Long> delete2(BigDecimal minRequiredBalance);

    @Delete("DELETE FROM ${table} WHERE ${by-id-filter} RETURNING *")
    CompletableFuture<Account> delete3(Long id);

    @Delete(entityClass = Account.class)
    CompletableFuture<Long> delete4(Long id);

    @Delete(
            value = "DELETE FROM ${table} WHERE ${by-id-filter} RETURNING *",
            entityClass = Account.class
    )
    CompletableFuture<EntityFieldMap> delete5(Long id);
}

public interface Transaction {

    ReactiveType commit();

    ReactiveType rollback();

    ReactiveType create(SavePoint savePoint);

    ReactiveType release(SavePoint savePoint);

    ReactiveType rollback(SavePoint savePoint);

    IsolationLevel getIsolationLevel();

    ReactiveType setIsolationLevel(IsolationLevel isolationLevel);
}

import io.rxmicro.data.sql.model.reactor.Transaction;

@PostgreSQLRepository
public interface BeginReactorTransactionRepository {

    Mono<Transaction> beginTransaction();

    Mono<Transaction> beginTransaction(IsolationLevel isolationLevel);
}

import io.rxmicro.data.sql.model.rxjava3.Transaction;

@PostgreSQLRepository
public interface BeginRxJava3TransactionRepository {

    Single<Transaction> beginTransaction();

    Single<Transaction> beginTransaction(IsolationLevel isolationLevel);
}

import io.rxmicro.data.sql.model.completablefuture.Transaction;

@SuppressWarnings("unused")
@PostgreSQLRepository
public interface BeginCompletableFutureTransactionRepository {

    CompletionStage<Transaction> beginTransaction1();

    CompletionStage<Transaction> beginTransaction1(IsolationLevel isolationLevel);

    CompletableFuture<Transaction> beginTransaction2();

    CompletableFuture<Transaction> beginTransaction2(IsolationLevel isolationLevel);
}

@PostgreSQLRepository
public interface ConcurrentRepository {

    Mono<Transaction> beginTransaction();

    @Select("SELECT * FROM ${table} WHERE id=? FOR UPDATE")
    Mono<Account> findAccountById(Transaction transaction, long id);

    @Select("SELECT * FROM ${table} WHERE id=? FOR UPDATE")
    Mono<Product> findProductById(Transaction transaction, int id);

    @Update(value = "UPDATE ${table} SET balance=? WHERE id=?", entityClass = Account.class)
    Mono<Void> updateAccountBalance(Transaction transaction, BigDecimal balance, long id);

    @Update(value = "UPDATE ${table} SET count=? WHERE id=?", entityClass = Product.class)
    Mono<Void> updateProductCount(Transaction transaction, int count, long id);

    @Insert
    Mono<Order> createOrder(Transaction transaction, Order order);
}

public final class ConcurrentBusinessService {

    private final ConcurrentRepository repository = getRepository(ConcurrentRepository.class);

    /**
     * @return order id if purchase is successful or
     *         error signal if:
     *                  - account not found or
     *                  - product not found or
     *                  - products ran out or
     *                  - money ran out
     */
    public Mono<Long> tryToBuy(final long idAccount,
                               final int idProduct,
                               final int count) {
        return repository.beginTransaction()
                .flatMap(transaction -> repository.findAccountById(transaction, idAccount)
                        .flatMap(account -> repository.findProductById(transaction, idProduct)
                                .flatMap(product ->
                                        tryToBuy(transaction, account, product, count))
                                .switchIfEmpty(Mono.error(() ->
                                        // product not found
                                        new ProductNotFoundException(idProduct))))
                        // account not found
                        .switchIfEmpty(Mono.error(() -> new AccountNotFoundException(idAccount)))
                        .onErrorResume(transaction.createRollbackThenReturnErrorFallback())
                );
    }

    private Mono<Long> tryToBuy(final Transaction transaction,
                                final Account account,
                                final Product product,
                                final int count) {
        if (count <= product.getCount()) {
            final BigDecimal cost = product.getPrice().multiply(BigDecimal.valueOf(count));
            if (cost.compareTo(account.getBalance()) <= 0) {
                return buy(transaction, account, product, count, cost);
            } else {
                // money ran out
                return Mono.error(new NotEnoughFundsException(cost, account.getBalance()));
            }
        } else {
            // products ran out
            return Mono.error(new NotEnoughProductCountException(count, product.getCount()));
        }
    }

    // purchase is successful, returns order id
    private Mono<Long> buy(final Transaction transaction,
                           final Account account,
                           final Product product,
                           final int count,
                           final BigDecimal cost) {
        final int newProductCount = product.getCount() - count;
        final BigDecimal newBalance = account.getBalance().subtract(cost);
        final Order order = new Order(account.getId(), product.getId(), count);

        return repository.updateProductCount(transaction, newProductCount, product.getId())
                .then(repository.updateAccountBalance(transaction, newBalance, account.getId())
                        .then(repository.createOrder(transaction, order)
                                .map(Order::getId)
                                .flatMap(id -> transaction.commit()
                                        .thenReturn(id))
                        )
                );
    }
}