Once the table is big
Everything that was fine at a thousand rows and is not fine at ten million: deep OFFSET, unbounded FindAll, and a single pool doing all the reading.
Pagination, ordering and search
order_by becomes a real ORDER BY, so the allow-list is a security control and not a nicety. Then the part most services need eventually: a keyset walk, because OFFSET 100000 reads a hundred thousand rows in order to discard them.
go
package main
import (
"fmt"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/repository"
"gitlab.finema.co/finema/idin-core/v2/utils"
)
// --- Example 5: pagination, ordering and search -----------------------------
//
// Pagination runs the query twice — a COUNT and a LIMIT/OFFSET page — and
// returns both in one typed core.Page[T] that marshals straight to JSON. In an
// HTTP handler the options come from c.GetPageOptionsWithAllowed(...); here
// they are built by hand, which is exactly the case where the allow-list has to
// be applied yourself.
// userSortable is the allow-list. order_by becomes a real SQL ORDER BY — GORM
// hands the string to the driver — so the framework drops anything that is not
// a column identifier, and this list drops the real columns the endpoint did
// not mean to expose. Unknown names are ignored rather than rejected, so an old
// bookmark keeps working.
var userSortable = []string{"created_at", "name", "status"}
type UserResponse struct {
ID string `json:"id"`
Name string `json:"name"`
}
func paginationTour(ctx core.IContext) core.IError {
if err := seedPagedUsers(ctx, 45); err != nil {
return err
}
page, err := listUsers(ctx, "page-", "name asc", 2, 20)
if err != nil {
return err
}
ctx.Log().Info("page 2", "total", page.Total, "count", page.Count, "limit", page.Limit)
// Anything not on the allow-list is dropped silently, including an attempt
// to smuggle SQL through the parameter.
if _, err := listUsers(ctx, "page-", "credits_satang, id;DROP TABLE users", 1, 20); err != nil {
return err
}
return keysetTour(ctx)
}
func listUsers(ctx core.IContext, prefix, rawOrderBy string, page, limit int64) (*core.Page[UserResponse], core.IError) {
opts := &core.PageOptions{
Q: prefix,
Page: page, // 0 or negative becomes 1
Limit: limit, // 0 becomes 30, anything over 10000 is clamped
// Setting OrderBy by hand skips both of GetPageOptions' guards, so put
// the string through ParseOrderBy yourself whenever the order comes
// from somewhere other than the query string.
OrderBy: core.ParseOrderBy(rawOrderBy, userSortable),
}
repo := repository.New[User](ctx)
if opts.Q != "" {
// Nothing applies Q for you — what "search" means is the endpoint's
// decision. LIKE '%…%' cannot use a plain B-tree index: fine over a few
// thousand rows, the wrong tool over a few million, where a trigram
// index or a full-text column is the next step. (On postgres this would
// be ILIKE; sqlite's LIKE is already case-insensitive for ASCII.)
repo = repo.Where("email LIKE ?", opts.Q+"%")
}
if len(opts.OrderBy) == 0 {
// A fallback the caller cannot see, applied only when it asked for
// nothing. Ordering by something unique — or tie-broken by something
// unique — is what stops a row appearing on two pages or on none.
repo = repo.Order("created_at desc, id desc")
}
pageOfUsers, err := repo.Pagination(opts)
if err != nil {
return nil, err
}
// MapPage keeps the pagination metadata and swaps the item type, so the
// client never sees the model that was queried.
return core.MapPage(pageOfUsers, func(u User) UserResponse {
return UserResponse{ID: u.ID, Name: u.Name}
}), nil
}
type userCursor struct {
CreatedAt time.Time
ID string
}
// keysetTour walks the whole set without OFFSET.
//
// OFFSET 100000 makes the database walk a hundred thousand rows and throw them
// away, and the COUNT scans the whole matching set every page. For an admin
// table that is fine; for an endpoint under load or a deep-paged export it is
// not. A keyset is O(limit) at any depth and stable while rows are inserted —
// it gives up random access to page N, which most callers were not using.
func keysetTour(ctx core.IContext) core.IError {
var cursor *userCursor
seen := 0
for {
repo := repository.New[User](ctx).
Where("email LIKE ?", "page-%").
// The ORDER BY must match the cursor comparison exactly, or rows
// are skipped and repeated.
Order("created_at desc, id desc").
Limit(20)
if cursor != nil {
repo = repo.Where("(created_at, id) < (?, ?)", cursor.CreatedAt, cursor.ID)
}
batch, err := repo.FindAll()
if err != nil {
return err
}
if len(batch) == 0 {
ctx.Log().Info("keyset walk finished", "rows", seen)
return nil
}
seen += len(batch)
last := batch[len(batch)-1]
cursor = &userCursor{CreatedAt: utils.ToNonPointer(last.CreatedAt), ID: last.ID}
}
}
func seedPagedUsers(ctx core.IContext, n int) core.IError {
rows := make([]User, 0, n)
for i := 0; i < n; i++ {
rows = append(rows, User{
Email: fmt.Sprintf("page-%03d@example.com", i),
Name: fmt.Sprintf("Paged %03d", i),
Status: UserActive,
})
}
return repository.New[User](ctx).CreateInBatches(rows, 100)
}More rows than fit in memory
FindAll() with no bound is an OOM waiting for the table to grow. CreateInBatches and FindInBatches keep the working set to one batch, and a keyed loop updates a million rows without holding a lock across all of them.
go
package main
import (
"encoding/csv"
"fmt"
"io"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/repository"
"gorm.io/gorm"
)
// --- Example 6: reading and writing a lot of rows ---------------------------
//
// FindAll() with no bound is an OOM waiting for the table to grow: it builds a
// slice of every matching row before the first one is used. Everything on this
// page exists to keep the working set to one batch — a fixed amount of memory
// whatever the table does.
func batchTour(ctx core.IContext) core.IError {
if err := bulkInsert(ctx, 2_000); err != nil {
return err
}
n, err := exportUsers(ctx, io.Discard)
if err != nil {
return err
}
ctx.Log().Info("exported", "rows", n)
updated, err := markDormant(ctx, time.Now().Add(time.Hour))
if err != nil {
return err
}
ctx.Log().Info("marked dormant", "rows", updated)
return nil
}
// bulkInsert chunks the INSERT so one statement does not exceed the driver's
// bound-parameter limit (postgres stops at 65535, and every column of every row
// is one parameter). The batch size is per statement, not per transaction.
func bulkInsert(ctx core.IContext, n int) core.IError {
rows := make([]User, 0, n)
for i := 0; i < n; i++ {
rows = append(rows, User{
Email: fmt.Sprintf("batch-%05d@example.com", i),
Name: fmt.Sprintf("Batch %05d", i),
Status: UserTrial,
CreditsSatang: int64(i),
})
}
return repository.New[User](ctx).CreateInBatches(rows, 200)
}
// exportUsers streams the whole table through a fixed-size buffer.
//
// FindInBatches reuses dest for every batch, so memory is one batch rather than
// one table. The trade-off: the batches are separate queries against a moving
// table, so a row inserted while the export runs may or may not appear. When
// the export has to be a single point in time, run it inside one transaction.
func exportUsers(ctx core.IContext, w io.Writer) (int, core.IError) {
out := csv.NewWriter(w)
defer out.Flush()
written := 0
var batch []User
err := repository.New[User](ctx).
Where("email LIKE ?", "batch-%").
// Order by the primary key: batching without a stable order can visit a
// row twice and miss another.
Order("id").
FindInBatches(&batch, 500, func(tx *gorm.DB, _ int) error {
for _, u := range batch {
if err := out.Write([]string{u.ID, u.Email, string(u.Status)}); err != nil {
// Returning an error stops the walk — the remaining batches
// are never fetched.
return err
}
written++
}
out.Flush()
return out.Error()
})
if err != nil {
return written, err
}
return written, nil
}
// markDormant updates a large set without holding a lock across all of it.
//
// One UPDATE over a million rows blocks every writer of those rows until it
// finishes. Walking the primary key in pages makes each batch its own short
// transaction, and the loop is resumable: if it dies halfway, running it again
// picks up where it stopped instead of starting over.
func markDormant(ctx core.IContext, cutoff time.Time) (int, core.IError) {
last := ""
total := 0
for {
var ids []string
if err := repository.New[User](ctx).
Where("id > ? AND created_at < ? AND status = ?", last, cutoff, UserTrial).
Order("id").
Limit(500).
Pluck("id", &ids); err != nil {
return total, err
}
if len(ids) == 0 {
return total, nil
}
// Pluck first, then update by id: the second statement touches exactly
// the rows the first one saw, so a row that changes underneath the loop
// cannot make it run forever.
if err := repository.New[User](ctx).
Where("id IN ?", ids).
Updates(map[string]any{"status": UserDormant}); err != nil {
return total, err
}
total += len(ids)
last = ids[len(ids)-1]
}
}Named connections, replicas and the pool
Registering a second connection is two lines; knowing what may read from it is the whole problem. Replication lag is not a bug, so routing goes by whether the caller tolerates staleness — not by whether the statement is a SELECT.
go
package main
import (
"time"
"github.com/glebarez/sqlite"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/repository"
"gorm.io/gorm"
)
// --- Example 7: connections, replicas and the pool --------------------------
//
// A connection is opened once, at startup, and registered on the App. Nothing
// opens one per request — that was v1's bug, and it is why IContext.Close() no
// longer exists. "default" is what ctx.DB() returns; every other name is
// ctx.DBS(name).
// openDatabase gives this example something to run against.
//
// core.NewDatabase supports postgres, mysql, sqlserver and oracle — and on
// purpose not sqlite, because no schema you deploy is ever sqlite. So: use the
// configured database when DB_* is set, and otherwise fall back to an in-memory
// sqlite opened directly, which is what the tour runs on with no setup at all.
func openDatabase(env core.IENV) (*gorm.DB, core.IError) {
cfg := env.Config()
if cfg.DBConnectionString != "" || cfg.DBDriver != "" {
// Pool sizes are options here rather than environment keys: the right
// numbers depend on what the process *is* — an API serving 200
// concurrent requests and a cron worker running one query at a time do
// not want the same pool — not on which environment it runs in. The
// arithmetic worth doing once is MaxOpenConns × replicas staying well
// under the server's max_connections, with room for migrations and a
// human holding a psql session.
return core.NewDatabase(env,
core.WithMaxOpenConns(20),
core.WithMaxIdleConns(5),
core.WithConnMaxLifetime(30*time.Minute), // shorter than any proxy's idle timeout
)
}
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
// NewDatabase would set this for you. Without it, timestamps depend on
// the machine's timezone and two servers disagree about "now".
NowFunc: func() time.Time { return time.Now().UTC() },
})
if err != nil {
return nil, core.Wrap(err, "database example: open sqlite")
}
sqlDB, sqlErr := db.DB()
if sqlErr != nil {
return nil, core.Wrap(sqlErr, "database example: sql handle")
}
// Every connection to ":memory:" gets its *own* empty database, so a pool of
// more than one would see tables come and go. One connection, one database.
sqlDB.SetMaxOpenConns(1)
return db, nil
}
// registerConnections is the shape a service with a read replica uses. Reads
// that can tolerate being a little behind go to the replica; everything else
// stays on the primary.
//
// NewDatabase reads one set of DB_* keys, so a second connection is opened
// directly — env.String reads any key that is not part of ENVConfig, so a
// second DSN needs no change to the framework.
func registerConnections(env core.IENV, primary, replica *gorm.DB) (*core.App, core.IError) {
return core.NewApp(env,
core.WithSQL("default", primary),
core.WithSQL("readonly", replica),
)
// app.Shutdown closes both. Handing a *gorm.DB to WithSQL transfers that
// responsibility — do not also close it yourself.
}
// reportFromReplica is the good case for a replica: a heavy read whose answer
// being a second old changes nothing.
func reportFromReplica(ctx core.IContext) (int64, core.IError) {
// ctx.DBS on a name that was never registered returns nil, and nil panics at
// the first use. That is deliberate: a wiring mistake should be loud and
// immediate rather than a confusing error much later.
replica := ctx.DBS("readonly")
if replica == nil {
return 0, core.New(500, "INVALID_CONFIG", `no "readonly" connection registered`)
}
return repository.NewWithDB[User](ctx, replica).
Where("status = ?", UserActive).
Count()
}
// createThenRead is the trap. Reads sent to a replica arrive *behind* the
// primary, so a row written a millisecond ago may genuinely not be there yet —
// replication lag is not a bug, and no retry loop makes it correct.
//
// Route by whether the caller can tolerate staleness, not by whether the
// statement happens to be a SELECT.
func createThenRead(ctx core.IContext, email string) (*User, core.IError) {
users := repository.New[User](ctx) // primary
u := User{Email: email, Name: "Fresh", Status: UserActive}
if err := users.Create(&u); err != nil {
return nil, err
}
// ✅ read it back from the primary, not from ctx.DBS("readonly")
return users.FindOne("id = ?", u.ID)
}
func connectionsTour(ctx core.IContext) core.IError {
if err := pingDB(ctx); err != nil {
return err
}
active, err := reportFromReplica(ctx)
if err != nil {
return err
}
if _, err := createThenRead(ctx, "[email protected]"); err != nil {
return err
}
ctx.Log().Info("replica report", "active_users", active)
return nil
}
// pingDB belongs in a readiness probe, not a liveness probe: a database that is
// briefly unreachable should stop traffic being routed to the pod, not restart
// it.
func pingDB(ctx core.IContext) core.IError {
sqlDB, err := ctx.DB().DB()
if err != nil {
return core.Wrap(err, "database example: sql handle")
}
if err := sqlDB.PingContext(ctx); err != nil {
return core.Wrap(err, "database example: ping")
}
return nil
}