package com.cygnus.cloud.database;

import io.vertx.sqlclient.Pool;
import io.vertx.sqlclient.Row;
import io.vertx.sqlclient.RowSet;
import io.vertx.sqlclient.SqlConnection;
import io.vertx.sqlclient.Tuple;
import java.util.function.Function;
import org.springframework.stereotype.Service;

import reactor.core.publisher.Mono;

/**
 * Reactor boundary around the Vert.x PostgreSQL pool. Feature repositories
 * should use parameterized SQL and pass values through {@link Tuple}.
 */
@Service
public class ReactiveDatabaseClient {

    private final Pool pool;

    public ReactiveDatabaseClient(Pool pool) {
        this.pool = pool;
    }

    public Mono<RowSet<Row>> query(String sql) {
        return Mono.fromCompletionStage(() -> pool.query(sql).execute().toCompletionStage());
    }

    public Mono<RowSet<Row>> preparedQuery(String sql, Tuple parameters) {
        return Mono.fromCompletionStage(
                () -> pool.preparedQuery(sql).execute(parameters).toCompletionStage());
    }

    public Mono<Integer> preparedUpdate(String sql, Tuple parameters) {
        return preparedQuery(sql, parameters).map(RowSet::rowCount);
    }

    public <T> Mono<T> inTransaction(Function<SqlConnection, Mono<T>> work) {
        return Mono.usingWhen(
                Mono.fromCompletionStage(() -> pool.getConnection().toCompletionStage()),
                connection -> Mono.fromCompletionStage(
                                () -> connection.begin().toCompletionStage())
                        .flatMap(transaction -> work.apply(connection)
                                .flatMap(result -> Mono.fromCompletionStage(
                                                () -> transaction.commit().toCompletionStage())
                                        .thenReturn(result))
                                .onErrorResume(error -> Mono.fromCompletionStage(
                                                () -> transaction.rollback().toCompletionStage())
                                        .onErrorResume(ignored -> Mono.empty())
                                        .then(Mono.error(error)))),
                connection -> Mono.fromCompletionStage(
                        () -> connection.close().toCompletionStage()),
                (connection, error) -> Mono.fromCompletionStage(
                        () -> connection.close().toCompletionStage()),
                connection -> Mono.fromCompletionStage(
                        () -> connection.close().toCompletionStage()));
    }

    public Mono<RowSet<Row>> preparedQuery(
            SqlConnection connection, String sql, Tuple parameters) {
        return Mono.fromCompletionStage(
                () -> connection.preparedQuery(sql)
                        .execute(parameters)
                        .toCompletionStage());
    }
}