Trying to write a wrapper around R2DBC using Effec...

# arrow

[Kristian Nedrevold]

Trying to write a wrapper around R2DBC using Effect and context receivers. select returns an Effect<Err, Flow<T> and execute returns Effect<Err, Long> for rows affected. This is my current API. Would really appreciate it if anyone could take a look and see if it looks sane 🙂

sealed interface R2DBCError {
    object Error: R2DBCError
    object TransactionError : R2DBCError
}

context (ConnectionFactory)
object R2DBC {
    inline fun <T>transaction(crossinline handler: context(Connection) (Handle) -> Effect<R2DBCError, T>): Effect<R2DBCError, T> = effect {
        val connection = create().awaitFirst()
        val handle = Handle
        try {
            val value = handler(connection, handle)
            connection.commitTransaction()
            value.bind()
        } catch (e: Exception) {
            connection.rollbackTransaction()
            if (e is kotlinx.coroutines.CancellationException) {
                throw e
            }
            shift(R2DBCError.TransactionError)
        }
    }
}

object Handle {

    /**
     * [execute] executes an insert/update/delete and returns the amount of affected rows
     */
    context (Connection)
    fun execute(sql: String, vararg params: Any) = effect<R2DBCError, Long> {
        val stmt = createStatement(sql)
        params.forEachIndexed { idx, param ->
            stmt.bind(idx, param)
        }

        stmt.add().execute().awaitFirst()
            .rowsUpdated
            .awaitFirst()
    }

    /**
     * [select] Executes a select statement and maps the result into a Flow<T> with a given mapper
     */
    context (Connection)
    inline fun <reified T: Any>select(sql: String, vararg params: Any, noinline r2dbcMapper: R2DBCMapper<T>) = effect<R2DBCError, Flow<T>> {
        val stmt = createStatement(sql)

        params.forEachIndexed { idx, param ->
            stmt.bind(idx, param)
        }

        stmt.add().execute().awaitFirst().map(r2dbcMapper).asFlow()
    }
}

[Kristian Nedrevold]

I think that the Handle object probably is redundant and I could just change it to top level functions and

inline fun <T>transaction(crossinline handler: context(Connection) (Handle) -> Effect<R2DBCError, T>)

becomes

inline fun <T>transaction(crossinline handler: context(Connection) () -> Effect<R2DBCError, T>)

[Kristian Nedrevold]

I think the design is quite poor right now because the caller has to care about the R2DBCError as it is now, while not really being able to get any information from it.

[simon.vergauwen]

I would define the errors as:

sealed interface R2DBCError {
    object Error: R2DBCError
    data class TransactionError(val cause: Exception) : R2DBCError
}

You're not using R2DBCError.Error anywhere? 🤔

[simon.vergauwen]

Also I think you can design this DSL without context receivers if you want.

[simon.vergauwen]

Untitled.cpp

Untitled.cpp

[Kristian Nedrevold]

That seems pretty smart yes 🙂 I think what I am having trouble with is deciding if transaction errors are something that I want to be typed out to the caller or if I just want it to fail with some exception e.

[simon.vergauwen]

In Scala, and I think also Haskell, they always just use an exception in IO. Even ZIO-sql just uses Exception and Throwable in their error channel..

I've been thinking about writing a super small JDBC wrapper (for Postgres) that translates SQLState into an typed error but even then. No1 is probably going to when over 100 different errors.

[Kristian Nedrevold]

If you have compile time checks on your queries then exceptions should be fine right?

[Kristian Nedrevold]

If I want to type it out I'm guessing I have to map over every exception that R2DBC can throw and map it into either exceptions in the case that there is something that shouldn't happen like an incorrect SQL query or some sort of typed error if it's because of something we can recover from

[Kristian Nedrevold]

I was thinking that something like this might be an okay compromise

object R2DBC {
    inline fun <E, T>ConnectionFactory.transaction(
        crossinline handler: ConnectionHandle<E>.() -> Effect<E, T>): Effect<E, T> = effect {
        val connection = create().awaitFirst()
        try {
            val value = handler(ConnectionHandle(connection, this))
            connection.commitTransaction()
            value.fold({
                connection.rollbackTransaction()
                shift(it)
            }, { it })
        } catch (e: Exception) {
            connection.rollbackTransaction()
            throw e
        }
    }
}

class ConnectionHandle<E>(connection: Connection, raise: EffectScope<E>): Connection by connection, EffectScope<E> by raise {
    /**
     * [select] Executes a select statement and maps the result into a Flow<T> with a given mapper
     */
    inline fun <T: Any>select(
        noinline r2dbcMapper: R2DBCMapper<T>,
        sql: String, 
        vararg params: Any,
        crossinline failWith: (Exception) -> E,
    ) = effect<E, Flow<T>> {
        try {
            val stmt = createStatement(sql)
            params.forEachIndexed { idx, param ->
                stmt.bind(idx, param)
            }
            stmt.add().execute().awaitFirst()
                .map(r2dbcMapper)
                .asFlow()
        } catch (e: Exception) {
            e.nonFatalOrThrow()
            shift(failWith(e))
        }
    }

    /**
     * [execute] executes an insert/update/delete and returns the amount of affected rows
     */
    inline fun <E>execute(
        sql: String,
        vararg params: Any,
        crossinline failWith: (Exception) -> E,
    ) = effect<E, Long> {
        try {

            val stmt = createStatement(sql)
            params.forEachIndexed { idx, param ->
                stmt.bind(idx, param)
            }

            stmt.add().execute().awaitFirst()
                .rowsUpdated
                .awaitFirst()
        } catch (e: Exception) {
            e.nonFatalOrThrow()
            shift(failWith(e))
        }
    }
}

Now I can just add some error handling in the failWith lambda in the caller

[Kristian Nedrevold]

Thanks for the feedback 🙂

[simon.vergauwen]

Adding such an error handler is always useful, and you can still provide a translation to RDBC2Error if they don't provide the handler. That way you can enforce error handling, and it exposes best of both worlds without code duplication even 😉

[Kristian Nedrevold]

Smart 🙂 Thanks 👍