feat(page): add typed keyset pagination for Bob (#15)
Add transport-neutral keyset pagination through native Bob query mods. Preserve supported cursor value types in a compact versioned format, with bidirectional pages, explicit row keys, and a separate opt-in count. Include SQLite tests, PostgreSQL service CI, and the cursor-format ADR. Closes #6
This commit was merged in pull request #15.
This commit is contained in:
end committed
1 parent
e5298cff37
commit
c45e8abe08
11 files changed
+1225
-1
No files matched your search
@@ -0,0 +1,189 @@
|
||||
package page_test
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"os"
|
||||
"reflect"
|
||||
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
_ "github.com/lib/pq"
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
"github.com/stephenafamo/bob"
|
||||
"github.com/stephenafamo/bob/dialect/mysql"
|
||||
md "github.com/stephenafamo/bob/dialect/mysql/dialect"
|
||||
mm "github.com/stephenafamo/bob/dialect/mysql/sm"
|
||||
"github.com/stephenafamo/bob/dialect/psql"
|
||||
pd "github.com/stephenafamo/bob/dialect/psql/dialect"
|
||||
pm "github.com/stephenafamo/bob/dialect/psql/sm"
|
||||
"github.com/stephenafamo/bob/dialect/sqlite"
|
||||
sd "github.com/stephenafamo/bob/dialect/sqlite/dialect"
|
||||
"github.com/stephenafamo/bob/dialect/sqlite/sm"
|
||||
"github.com/stephenafamo/bob/expr"
|
||||
"github.com/stephenafamo/scan"
|
||||
"go.auvem.com/corestack/page"
|
||||
)
|
||||
|
||||
type item struct {
|
||||
ID int64
|
||||
Score int64
|
||||
}
|
||||
|
||||
func itemSort() page.Sort {
|
||||
return page.Sort{ID: "items/score/v1", Columns: []page.Column{
|
||||
{Expression: expr.Quote("score"), Direction: page.Desc},
|
||||
{Expression: expr.Quote("id"), Direction: page.Asc, Unique: true},
|
||||
}}
|
||||
}
|
||||
|
||||
func itemKey(i item) []any { return []any{i.Score, i.ID} }
|
||||
|
||||
func TestSQLiteFlow(t *testing.T) {
|
||||
db, err := bob.Open("sqlite3", ":memory:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
databaseFlow(t, db, false)
|
||||
}
|
||||
|
||||
func TestPostgresFlow(t *testing.T) {
|
||||
dsn := os.Getenv("PAGE_POSTGRES_DSN")
|
||||
if dsn == "" {
|
||||
if os.Getenv("PAGE_REQUIRE_POSTGRES") != "" {
|
||||
t.Fatal("PAGE_POSTGRES_DSN is required")
|
||||
}
|
||||
t.Skip("set PAGE_POSTGRES_DSN to run PostgreSQL integration")
|
||||
}
|
||||
db, err := bob.Open("postgres", dsn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
databaseFlow(t, db, true)
|
||||
}
|
||||
|
||||
func databaseFlow(t *testing.T, db bob.DB, postgres bool) {
|
||||
t.Helper()
|
||||
db.SetMaxOpenConns(1)
|
||||
for _, statement := range []string{
|
||||
"CREATE TEMP TABLE items (id BIGINT PRIMARY KEY, score BIGINT NOT NULL, tenant BIGINT NOT NULL)",
|
||||
"INSERT INTO items VALUES (1,30,1),(2,30,1),(3,20,1),(4,10,1),(5,5,2),(6,30,2)",
|
||||
} {
|
||||
if _, err := db.ExecContext(t.Context(), statement); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
fetch := func(direction page.Direction, size int, cursor string) page.Page[item] {
|
||||
t.Helper()
|
||||
p, err := page.Prepare(itemSort(), itemKey, page.Request{Direction: direction, Size: size, Cursor: cursor}, page.Limits{Default: 2, Maximum: 3})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var q bob.Query
|
||||
if postgres {
|
||||
query := psql.Select(pm.Columns("id", "score"), pm.From("items"), pm.Where(psql.Quote("tenant").EQ(psql.Arg(int64(1)))))
|
||||
query.Apply(page.Mod[*pd.SelectQuery](p))
|
||||
query.Apply(pm.Where(psql.Quote("score").GT(psql.Arg(int64(0)))))
|
||||
q = query
|
||||
} else {
|
||||
query := sqlite.Select(sm.Columns("id", "score"), sm.From("items"), sm.Where(sqlite.Quote("tenant").EQ(sqlite.Arg(int64(1)))))
|
||||
query.Apply(page.Mod[*sd.SelectQuery](p))
|
||||
query.Apply(sm.Where(sqlite.Quote("score").GT(sqlite.Arg(int64(0)))))
|
||||
q = query
|
||||
}
|
||||
rows, err := bob.All(t.Context(), db, q, scan.StructMapper[item]())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := append([]item(nil), rows...)
|
||||
result, err := p.Page(rows)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(before, rows) {
|
||||
t.Fatal("Page mutated fetched slice")
|
||||
}
|
||||
return result
|
||||
}
|
||||
check := func(got page.Page[item], ids []int64, next, previous bool) {
|
||||
t.Helper()
|
||||
actual := make([]int64, len(got.Items))
|
||||
for i, row := range got.Items {
|
||||
actual[i] = row.ID
|
||||
}
|
||||
if !reflect.DeepEqual(actual, ids) || got.HasNext != next || got.HasPrevious != previous {
|
||||
t.Fatalf("got IDs %v next=%v previous=%v; want %v %v %v", actual, got.HasNext, got.HasPrevious, ids, next, previous)
|
||||
}
|
||||
if len(ids) == 0 && (got.StartCursor != "" || got.EndCursor != "") {
|
||||
t.Fatal("empty page has cursors")
|
||||
}
|
||||
if len(ids) > 0 && (got.StartCursor == "" || got.EndCursor == "") {
|
||||
t.Fatal("missing cursors")
|
||||
}
|
||||
}
|
||||
first := fetch(page.Forward, 2, "")
|
||||
check(first, []int64{1, 2}, true, false)
|
||||
second := fetch(page.Forward, 2, first.EndCursor)
|
||||
check(second, []int64{3, 4}, false, true)
|
||||
check(fetch(page.Forward, 2, second.EndCursor), []int64{}, false, true)
|
||||
last := fetch(page.Backward, 2, "")
|
||||
check(last, []int64{3, 4}, false, true)
|
||||
previous := fetch(page.Backward, 2, last.StartCursor)
|
||||
check(previous, []int64{1, 2}, true, false)
|
||||
check(fetch(page.Backward, 2, previous.StartCursor), []int64{}, true, false)
|
||||
check(fetch(page.Forward, 0, ""), []int64{1, 2}, true, false)
|
||||
check(fetch(page.Forward, 1000, ""), []int64{1, 2, 3}, true, false)
|
||||
// A cursor describes a value boundary even after its row is deleted.
|
||||
if _, err := db.ExecContext(t.Context(), "DELETE FROM items WHERE id=2"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
check(fetch(page.Forward, 2, first.EndCursor), []int64{3, 4}, false, true)
|
||||
var countQuery, emptyQuery bob.Query
|
||||
if postgres {
|
||||
countQuery = psql.Select(pm.Columns("count(*)"), pm.From("items"), pm.Where(psql.Quote("tenant").EQ(psql.Arg(1))))
|
||||
emptyQuery = psql.Select(pm.Columns("id"), pm.From("items"), pm.Where(psql.Quote("id").LT(psql.Arg(0))))
|
||||
} else {
|
||||
countQuery = sqlite.Select(sm.Columns("count(*)"), sm.From("items"), sm.Where(sqlite.Quote("tenant").EQ(sqlite.Arg(1))))
|
||||
emptyQuery = sqlite.Select(sm.Columns("id"), sm.From("items"), sm.Where(sqlite.Quote("id").LT(sqlite.Arg(0))))
|
||||
}
|
||||
count, err := page.Count(t.Context(), db, countQuery)
|
||||
if err != nil || count != 3 {
|
||||
t.Fatalf("count %d: %v", count, err)
|
||||
}
|
||||
if _, err := page.Count(t.Context(), db, emptyQuery); !errors.Is(err, sql.ErrNoRows) {
|
||||
t.Fatalf("count error lost cause: %v", err)
|
||||
}
|
||||
_, err = page.Prepare(itemSort(), itemKey, page.Request{Cursor: "not!base64"}, page.Limits{Default: 2, Maximum: 3})
|
||||
var cursorError *page.CursorError
|
||||
if !errors.As(err, &cursorError) {
|
||||
t.Fatalf("invalid cursor: %v", err)
|
||||
}
|
||||
bad, err := page.Prepare(itemSort(), func(item) []any { return []any{int64(1)} }, page.Request{}, page.Limits{Default: 2, Maximum: 3})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result, err := bad.Page([]item{{ID: 1}})
|
||||
var keyError *page.KeyError
|
||||
if !errors.As(err, &keyError) || !reflect.DeepEqual(result, page.Page[item]{}) {
|
||||
t.Fatalf("key failure: %+v %v", result, err)
|
||||
}
|
||||
if _, err := db.ExecContext(t.Context(), "DELETE FROM items"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
check(fetch(page.Forward, 2, ""), []int64{}, false, false)
|
||||
check(fetch(page.Backward, 2, ""), []int64{}, false, false)
|
||||
check(fetch(page.Forward, 2, first.EndCursor), []int64{}, false, true)
|
||||
check(fetch(page.Backward, 2, last.StartCursor), []int64{}, true, false)
|
||||
}
|
||||
|
||||
func TestMySQLModComposition(t *testing.T) {
|
||||
p, err := page.Prepare(itemSort(), itemKey, page.Request{}, page.Limits{Default: 2, Maximum: 3})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
q := mysql.Select(mm.Columns("id", "score"), mm.From("items"), page.Mod[*md.SelectQuery](p))
|
||||
q.Apply(mm.Where(mysql.Quote("tenant").EQ(mysql.Arg(int64(1)))))
|
||||
// Compile-time composition only; MySQL runtime behavior needs a database test.
|
||||
}
|
||||
Reference in new issue
Block a user