pagination

typo-dsl-zio-jdbc/src/scala/typo/dsl/SqlExpr.scala

@@ -142,6 +142,17 @@ object SqlExpr {
     override def render(counter: AtomicInteger): SqlFragment = sql"${E.encode(value)}" ++ s"::${P.sqlType}"
   }
 
+  object Const {
+    trait As[T, N[_], R] {
+      def apply(value: N[T]): SqlExpr.Const[T, N, R]
+    }
+
+    object As {
+      implicit def as[T, N[_], R](implicit E: JdbcEncoder[N[T]], P: ParameterMetaData[T]): As[T, N, R] =
+        (value: N[T]) => SqlExpr.Const(value, E, P)
+    }
+  }
+
   final case class ArrayIndex[T, N1[_], N2[_], R](arr: SqlExpr[Array[T], N1, R], idx: SqlExpr[Int, N2, R], N: Nullability2[N1, N2, Option]) extends SqlExpr[T, Option, R] {
     override def eval(row: R): Option[T] = {
       N.mapN(arr.eval(row), idx.eval(row)) { (arr, idx) =>

typo-dsl-zio-jdbc/src/scala/typo/dsl/pagination/AbstractJsonCodec.scala

@@ -0,0 +1,7 @@
+package typo.dsl.pagination
+
+/** abstract over json libraries */
+class AbstractJsonCodec[T, E](
+    val encode: T => E,
+    val decode: E => Either[String, T]
+)

typo-dsl-zio-jdbc/src/scala/typo/dsl/pagination/ClientCursor.scala

@@ -0,0 +1,17 @@
+package typo.dsl.pagination
+
+/** This will typically be JSON encoded and passed to clients.
+  *
+  * It represents a cursor that can be used to fetch the next page of results.
+  *
+  * The position is a given row for a set of [[order by]] expressions.
+  *
+  * The [[ClientCursor]] itself is a glorified [[Map]], with pairs of stringified `order by` expressions and a JSON representation of the corresponding value from the current row
+  *
+  * The design has a few interesting tradeoffs:
+  *   - it leaks database column names to the client, so you may want to obfuscate/encrypt it
+  *   - it can be re-used for "similar" queries, not just the exact same query. Fewer `order by` expressions or different ordering of `order by` expressions is fine.
+  *   - [[SortOrderRepr]] does not encode ascending/descending or nulls first/last, but you're still anchored to a specific row for a set of orderings. If you want, you can change your query to go
+  *     both ways from a given cursor.
+  */
+case class ClientCursor[E](parts: Map[SortOrderRepr, E])

typo-dsl-zio-jdbc/src/scala/typo/dsl/pagination/PaginationQuery.scala

@@ -0,0 +1,59 @@
+package typo.dsl.pagination
+
+import typo.dsl.*
+import zio.jdbc.ZConnection
+import zio.{Chunk, ZIO}
+
+import java.util.concurrent.atomic.AtomicInteger
+
+/** You'll typically use this behind a facade which accounts for your json library of choice.
+  *
+  * If you don't, you'll just have to fill in [[AbstractJsonCodec]] yourself.
+  */
+class PaginationQuery[Fields[_], Row, E](
+    query: SelectBuilder[Fields, Row],
+    part1s: List[ServerCursor.Part1NoHkt[Fields, Row, ?, E]]
+) {
+  def andOn[T, N[_]](v: Fields[Row] => SortOrder[T, N, Row])(codec: AbstractJsonCodec[N[T], E])(implicit asConst: SqlExpr.Const.As[T, N, Row]): PaginationQuery[Fields, Row, E] =
+    new PaginationQuery(query, part1s :+ new ServerCursor.Part1(v, codec, asConst))
+
+  def done(limit: Int, continueFrom: Option[ClientCursor[E]]): Either[String, (SelectBuilder[Fields, Row], ServerCursor[Fields, Row, E])] = {
+    val initialCursor = {
+      // hack: improve DSL to avoid  instantiating structure twice
+      val structure: Structure[Fields, Row] = query match {
+        case mock: SelectBuilderMock[Fields, Row] => mock.structure
+        case sql: SelectBuilderSql[Fields, Row]   => sql.instantiate(new AtomicInteger(0)).structure
+        case _                                    => sys.error("unsupported query type")
+      }
+      new ServerCursor.Initial(part1s.map(_.asPart2(structure.fields)), limit)
+    }
+
+    continueFrom match {
+      case None =>
+        val newQuery = initialCursor.part2s
+          .foldLeft(query) { case (q, part2: ServerCursor.Part2[Fields, Row, t, n, E] @unchecked /* for 2.13 */ ) =>
+            q.orderBy(part2.part1.v.asInstanceOf[Fields[Hidden] => SortOrder[t, n, Hidden]])
+          }
+          .limit(limit)
+        Right((newQuery, initialCursor))
+
+      case Some(clientCursor) =>
+        initialCursor.withTupleFrom(clientCursor).map { cursor =>
+          val newQuery = cursor.part3s
+            .foldLeft(query) { case (q, part3: ServerCursor.Part3[Fields, Row, _, _, E] @unchecked /* for 2.13 */ ) =>
+              q.seek(part3.part2.part1.v)(part3.value)
+            }
+            .limit(limit)
+          (newQuery, cursor)
+        }
+    }
+  }
+
+  def toChunk(limit: Int, continueFrom: Option[ClientCursor[E]]): ZIO[ZConnection, Throwable, (Chunk[Row], Option[ClientCursor[E]])] =
+    done(limit, continueFrom) match {
+      case Left(msg) =>
+        ZIO.fail(new IllegalArgumentException(msg))
+      case Right((newQuery, cursor)) =>
+        newQuery.toChunk.map(rows => (rows, cursor.withNewResults(rows).map(_.clientCursor)))
+    }
+}

typo-dsl-zio-jdbc/src/scala/typo/dsl/pagination/ServerCursor.scala

@@ -0,0 +1,88 @@
+package typo.dsl.pagination
+
+import typo.dsl.{SortOrder, SqlExpr}
+import typo.dsl.pagination.internal.ListOps
+
+sealed trait ServerCursor[Fields[_], Row, E] {
+  def part2s: List[ServerCursor.Part2NoHkt[Fields, Row, ?, E]]
+  val limit: Int
+
+  final def withNewResults(rows: Seq[Row]): Option[ServerCursor.AtTuple[Fields, Row, E]] =
+    if (rows.length < limit) None // no new cursor if we know we reached end
+    else rows.lastOption.map(lastRow => new ServerCursor.AtTuple(part2s.map(_.asPart3(lastRow)), limit))
+
+  final def withTupleFrom(clientCursor: ClientCursor[E]): Either[String, ServerCursor.AtTuple[Fields, Row, E]] =
+    part2s.traverse(_.decodeDataFrom(clientCursor)).map(parts => new ServerCursor.AtTuple(parts, limit))
+}
+
+object ServerCursor {
+  class Initial[Fields[_], Row, E](val part2s: List[Part2NoHkt[Fields, Row, ?, E]], val limit: Int) extends ServerCursor[Fields, Row, E]
+
+  class AtTuple[Fields[_], Row, E](val part3s: List[Part3NoHkt[Fields, Row, ?, E]], val limit: Int) extends ServerCursor[Fields, Row, E] {
+    def part2s: List[Part2NoHkt[Fields, Row, ?, E]] = part3s.map(_.part2)
+    def clientCursor: ClientCursor[E] = ClientCursor(part3s.map(_.encoded).toMap)
+  }
+
+  /** A constituent in a server-side cursor.
+    *
+    * At this point we have a function from `Fields` to a [[SortOrder]], making it dependant on the *shape* of the query, but not an *instance* of a query.
+    *
+    * You very likely don't have to worry about this, as it is used internally.
+    */
+  class Part1[Fields[_], Row, T, N[_], E](
+      val v: Fields[Row] => SortOrder[T, N, Row],
+      val codec: AbstractJsonCodec[N[T], E],
+      val asConst: SqlExpr.Const.As[T, N, Row]
+  ) extends Part1NoHkt[Fields, Row, N[T], E] {
+    def asPart2(fields: Fields[Row]): ServerCursor.Part2[Fields, Row, T, N, E] =
+      new ServerCursor.Part2(this, v(fields))
+  }
+  sealed trait Part1NoHkt[Fields[_], Row, NT, E] { // for scala 2.13
+    def asPart2(fields: Fields[Row]): ServerCursor.Part2NoHkt[Fields, Row, NT, E]
+  }
+
+  /** A constituent in a server-side cursor.
+    *
+    * At this point we have extracted the [[SortOrder]] from a query, making the value somewhat depending on it. That dependency affects the rendering of columns only
+    *
+    * The benefit is that we can now render it to a [[SortOrderRepr]], so we can encode it in a [[ClientCursor]]
+    */
+  class Part2[Fields[_], Row, T, N[_], E](val part1: Part1[Fields, Row, T, N, E], val sortOrder: SortOrder[T, N, Row]) extends Part2NoHkt[Fields, Row, N[T], E] {
+    val sortOrderRepr: SortOrderRepr = SortOrderRepr.from(sortOrder)
+
+    override def asPart3(row: Row): Part3[Fields, Row, T, N, E] =
+      new ServerCursor.Part3[Fields, Row, T, N, E](this, part1.asConst(sortOrder.expr.eval(row)))
+
+    override def decodeDataFrom(clientCursor: ClientCursor[E]): Either[String, Part3[Fields, Row, T, N, E]] =
+      clientCursor.parts.get(sortOrderRepr) match {
+        case None => Left(s"Cursor is missing value for column '${sortOrderRepr.expr}'")
+        case Some(value) =>
+          part1.codec.decode(value) match {
+            case Left(msg) =>
+              Left(s"Cursor had invalid value '$value' for column '${sortOrderRepr.expr}': $msg")
+            case Right(nt) =>
+              Right(new ServerCursor.Part3[Fields, Row, T, N, E](this, part1.asConst(nt)))
+          }
+      }
+  }
+  sealed trait Part2NoHkt[Fields[_], Row, NT, E] { // for scala 2.13
+    def asPart3(row: Row): Part3NoHkt[Fields, Row, NT, E]
+    def decodeDataFrom(clientCursor: ClientCursor[E]): Either[String, Part3NoHkt[Fields, Row, NT, E]]
+  }
+
+  /** A constituent in a server-side cursor.
+    *
+    * At this point there is also data, essentially a column reference plus a value. This makes it an actual cursor.
+    *
+    * The data is wrapped in [[SqlExpr.Const]] so it's ready to embed into a query without threading through the type class instances
+    */
+  class Part3[Fields[_], Row, T, N[_], E](val part2: Part2[Fields, Row, T, N, E], val value: SqlExpr.Const[T, N, Row]) extends Part3NoHkt[Fields, Row, N[T], E] {
+    override def encoded: (SortOrderRepr, E) =
+      (part2.sortOrderRepr, part2.part1.codec.encode(value.value))
+  }
+
+  sealed trait Part3NoHkt[Fields[_], Row, NT, E] { // for scala 2.13
+    val part2: Part2NoHkt[Fields, Row, ?, E]
+    def encoded: (SortOrderRepr, E)
+  }
+}

typo-dsl-zio-jdbc/src/scala/typo/dsl/pagination/SortOrderRepr.scala

@@ -0,0 +1,32 @@
+package typo.dsl.pagination
+
+import typo.dsl.*
+import zio.jdbc.SqlFragment
+
+import java.util.concurrent.atomic.AtomicInteger
+
+/** A client cursor is inherently tied to a set of sort orderings.
+  *
+  * As such we encode them in the cursor itself, and verify them on the way in.
+  */
+case class SortOrderRepr(expr: String) extends AnyVal
+
+object SortOrderRepr {
+  def from[NT, R](x: SortOrderNoHkt[NT, R]): SortOrderRepr = {
+    val sql = new StringBuilder()
+
+    // no usable way to just get the sql without parameters in zio-jdbc :(
+    def recAdd(x: SqlFragment): Unit =
+      x.segments.foreach {
+        case SqlFragment.Segment.Empty           => ()
+        case SqlFragment.Segment.Syntax(value)   => sql.append(value)
+        case SqlFragment.Segment.Param(value, _) => sql.append(value)
+        case SqlFragment.Segment.Nested(sql)     => recAdd(sql)
+      }
+
+    // note `x.expr`! the value is independent of ascending/descending and nulls first/last
+    recAdd(x.expr.render(new AtomicInteger(0)))
+
+    SortOrderRepr(sql.result().trim)
+  }
+}

typo-dsl-zio-jdbc/src/scala/typo/dsl/pagination/internal.scala

@@ -0,0 +1,18 @@
+package typo.dsl.pagination
+
+object internal {
+  // avoid cats dependency
+  implicit class ListOps[T](private val list: List[T]) extends AnyVal {
+    def traverse[U](f: T => Either[String, U]): Either[String, List[U]] = {
+      val it = list.iterator
+      val result = List.newBuilder[U]
+      while (it.hasNext) {
+        f(it.next()) match {
+          case Left(e)  => return Left(e)
+          case Right(u) => result += u
+        }
+      }
+      Right(result.result())
+    }
+  }
+}

typo-tester-zio-jdbc/src/scala/adventureworks/PaginationQueryZioJson.scala

@@ -0,0 +1,42 @@
+package adventureworks
+
+import typo.dsl.pagination.*
+import typo.dsl.{SelectBuilder, SortOrder, SqlExpr}
+import zio.jdbc.ZConnection
+import zio.json.ast.Json
+import zio.json.{JsonDecoder, JsonEncoder}
+import zio.{Chunk, ZIO}
+
+class PaginationQueryZioJson[Fields[_], Row](underlying: PaginationQuery[Fields, Row, Json]) {
+  def andOn[T, N[_]](v: Fields[Row] => SortOrder[T, N, Row])(implicit
+      e: JsonEncoder[N[T]],
+      d: JsonDecoder[N[T]],
+      asConst: SqlExpr.Const.As[T, N, Row]
+  ): PaginationQueryZioJson[Fields, Row] =
+    new PaginationQueryZioJson(underlying.andOn(v)(PaginationQueryZioJson.abstractCodec)(asConst))
+
+  def done(limit: Int, continueFrom: Option[ClientCursor[Json]]): Either[String, (SelectBuilder[Fields, Row], ServerCursor[Fields, Row, Json])] =
+    underlying.done(limit, continueFrom)
+
+  def toChunk(limit: Int, continueFrom: Option[ClientCursor[Json]]): ZIO[ZConnection, Throwable, (Chunk[Row], Option[ClientCursor[Json]])] =
+    underlying.toChunk(limit, continueFrom)
+}
+
+object PaginationQueryZioJson {
+  implicit val clientCursorEncoder: JsonEncoder[ClientCursor[Json]] =
+    JsonEncoder[Map[String, Json]].contramap(_.parts.map { case (k, v) => (k.expr, v) }.toMap)
+  implicit val clientCursorDecoder: JsonDecoder[ClientCursor[Json]] =
+    JsonDecoder[Map[String, Json]].map(parts => ClientCursor(parts.map { case (k, v) => (SortOrderRepr(k), v) }))
+
+  implicit class PaginationQuerySyntax[Fields[_], Row](private val query: SelectBuilder[Fields, Row]) extends AnyVal {
+    def seekPaginationOn[T, N[_]](v: Fields[Row] => SortOrder[T, N, Row])(implicit
+        e: JsonEncoder[N[T]],
+        d: JsonDecoder[N[T]],
+        asConst: SqlExpr.Const.As[T, N, Row]
+    ): PaginationQueryZioJson[Fields, Row] =
+      new PaginationQueryZioJson(new PaginationQuery(query, Nil).andOn(v)(PaginationQueryZioJson.abstractCodec))
+  }
+
+  def abstractCodec[N[_], T](implicit e: JsonEncoder[N[T]], d: JsonDecoder[N[T]]): AbstractJsonCodec[N[T], Json] =
+    new AbstractJsonCodec[N[T], Json](e.toJsonAST(_).getOrElse(???), d.fromJsonAST)
+}