The typed repository
mongorepo.Repo[D] is the layer most code should use: typed operators, copy-on-write chains, and a pipeline builder that composes with them.
Conditions, projections and paging
The operators that are easy to get subtly wrong by hand — Between, ElemMatch, Search — plus the two ways to read a lot of documents without holding them all. See Repository: querying.
go
package main
import (
"errors"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/mongorepo"
"go.mongodb.org/mongo-driver/v2/bson"
)
// --- Example 3: the typed repository ---------------------------------------
//
// mongorepo.Repo[D] is the same shape as the SQL repository: the context is
// bound once at New, the finishers take no ctx, and every chaining method
// clones — so a base scope can be branched twice without the first branch
// leaking into the second.
//
// It is a layer over core.IMongoDB, not a replacement: DB() gives the handle
// back and Collection() gives the driver's, so nothing is out of reach.
func repoQueries(ctx core.IContext) core.IError {
users := mongorepo.New[User](ctx)
// One base scope, reused below. Because the chain is copy-on-write this is a
// value, not a builder someone else can spoil.
active := users.Eq("status", "active").HasField("deleted_at", false)
recent, err := active.
Between("joined", time.Now().AddDate(0, 0, -30), time.Now()).
Sort("-joined").
Limit(20).
Omit("tags"). // projection: keep the bulky fields out of a list read
FindAll()
if err != nil {
return err
}
// In with no values matches nothing, deliberately: that is what the caller
// asked for, so it is not silently dropped into "match everything".
invited, err := users.In("status", "invited", "pending").Count()
if err != nil {
return err
}
// A miss is a 404 wrapping core.ErrDocumentNotFound, not a nil document —
// so the "not there" branch is explicit rather than a nil check that
// someone forgets.
one, err := active.Sort("-joined").FindOne()
switch {
case errors.Is(err, core.ErrDocumentNotFound):
ctx.Log().Info("no active user yet")
case err != nil:
return err
default:
ctx.Log().Info("newest active user", "name", one.Name, "id", one.ID.Hex())
}
ctx.Log().Info("repo reads", "recent", len(recent), "invited", invited)
if err := repoNested(ctx); err != nil {
return err
}
page, err := listUsers(ctx, &core.PageOptions{Page: 1, Limit: 20, Q: "example"})
if err != nil {
return err
}
exported, err := exportUsers(ctx)
if err != nil {
return err
}
ctx.Log().Info("repo paging", "total", page.Total, "on_page", page.Count, "exported", exported)
return nil
}
// repoNested is where the typed operators earn their keep: the conditions that
// are easy to get subtly wrong when written as bson.M by hand.
func repoNested(ctx core.IContext) core.IError {
orders := mongorepo.New[Order](ctx)
// ElemMatch is the difference a nested query usually turns on. A plain
// dotted filter on two fields of the same array can be satisfied by two
// *different* elements:
//
// Eq("items.sku", "A").Gte("items.qty", 2) → element 1 is A, element 2 has qty 2
// ElemMatch("items", …) → one element is both
bulk, err := orders.
ElemMatch("items", bson.M{"sku": "SKU-1", "qty": bson.M{"$gte": 2}}).
FindAll()
if err != nil {
return err
}
// Search is the "q" of a list endpoint: a case-insensitive substring ORed
// across the fields, with the term escaped so a user cannot turn it into a
// pattern of their own choosing. Only a prefix-anchored pattern can use an
// index, so this is a convenience for small collections, not a search engine.
found, err := mongorepo.New[User](ctx).
Eq("status", "active").
Search("exam", "name", "email").
Count()
if err != nil {
return err
}
// Pluck reads one field of every match without decoding whole documents —
// how you gather ids to hand to the next query.
var emails []string
if err := mongorepo.New[User](ctx).Eq("status", "active").
Limit(100).Pluck("email", &emails); err != nil {
return err
}
ctx.Log().Info("repo nested", "bulk_orders", len(bulk), "matched_q", found, "emails", len(emails))
return nil
}
// listUsers is the shape an HTTP handler actually has: options straight from
// the request, one call, a *core.Page ready to return.
//
// GetPageOptionsWithAllowed is what keeps order_by from being whatever the
// caller typed. A Sort on the chain is only the *default* — an explicit OrderBy
// in the options wins, which is what makes the endpoint sortable at all.
func listUsers(ctx core.IContext, opts *core.PageOptions) (*core.Page[User], core.IError) {
return mongorepo.New[User](ctx).
Eq("status", "active").
Search(opts.Q, "name", "email").
Sort("-joined").
Pagination(opts)
}
// exportUsers streams instead of collecting. FindAll without a Limit reads the
// whole result set into memory; Each never holds more than a batch, which is
// the difference between an export that runs and one that gets OOM-killed at
// three in the morning.
func exportUsers(ctx core.IContext) (int, core.IError) {
count := 0
err := mongorepo.New[User](ctx).
Eq("status", "active").
Select("_id", "email"). // read only what the export writes
Each(func(u User) error {
count++
return nil // returning an error here stops the iteration
})
return count, err
}Create, upsert and the atomic operators
$inc, $addToSet and FindOneAndUpdate need no transaction at all — a write to one document is already atomic. Also why an unscoped Delete is refused.
go
package main
import (
"errors"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/mongorepo"
"gitlab.finema.co/finema/idin-core/v2/utils"
"go.mongodb.org/mongo-driver/v2/bson"
)
// --- Example 4: writing through the repository -----------------------------
//
// Every write reports what it actually did (Matched, Modified, UpsertedID)
// rather than only whether it failed: "no such document" and "found it, nothing
// to change" are different answers, and only the caller knows which matters.
func repoWrites(ctx core.IContext) core.IError {
users := mongorepo.New[User](ctx)
u := User{
Email: "[email protected]",
Name: "Repo Example",
Status: "active",
Age: 41,
Tags: []string{"seed"},
Joined: utils.ToPointer(time.Now()),
}
// Create fills the generated id back into the document, the way GORM fills a
// primary key — so the caller can use it without a second read.
err := users.Create(&u)
if errors.Is(err, core.ErrDuplicateKey) {
// The unique index is the only thing that actually enforces uniqueness;
// a check-then-insert loses the race that matters.
found, findErr := users.Eq("email", u.Email).FindOne()
if findErr != nil {
return findErr
}
u = *found
} else if err != nil {
return err
}
ctx.Log().Info("created", "id", u.ID.Hex())
if err := repoAtomicWrites(ctx, u.ID.Hex()); err != nil {
return err
}
return repoUpsertAndDelete(ctx)
}
// repoAtomicWrites is the set of updates that need no transaction at all: each
// is a single-document operation, and a write to one document is already
// atomic. Reaching for a transaction here is a habit carried over from SQL.
func repoAtomicWrites(ctx core.IContext, id string) core.IError {
user := mongorepo.New[User](ctx).ByID(id)
// $inc, not read-modify-write: two requests can increment at once without
// one of them losing its update.
if _, err := user.Inc("logins", 1); err != nil {
return err
}
// AddToSet is Push that skips values already there.
if _, err := user.AddToSet("tags", "returning", "seed"); err != nil {
return err
}
if _, err := user.Pull("tags", "stale"); err != nil {
return err
}
// Unset removes the field rather than zeroing it — which is what "not
// deleted" has to look like for the partial unique index in Example 6.
if _, err := user.Unset("deleted_at"); err != nil {
return err
}
// A plain map is wrapped in $set; a map that already speaks in operators is
// sent as written. Both spellings work, so nothing has to be un-learned.
res, err := user.Updates(bson.M{"status": "active", "name": "Repo Example"})
if err != nil {
return err
}
ctx.Log().Info("updated", "matched", res.Matched, "modified", res.Modified)
// Claiming a document: one operation, so exactly one worker wins.
claimed, err := mongorepo.New[User](ctx).
Eq("status", "active").
Sort("joined").
FindOneAndUpdate(bson.M{"$inc": bson.M{"logins": 1}})
if errors.Is(err, core.ErrDocumentNotFound) {
return nil // nothing to claim is not a failure
}
if err != nil {
return err
}
ctx.Log().Info("claimed", "user", claimed.Name, "logins", claimed.Logins)
return nil
}
func repoUpsertAndDelete(ctx core.IContext) core.IError {
users := mongorepo.New[User](ctx)
// Upsert is the idempotent write: insert when the filter matches nothing,
// update when it does. The filter's own fields land on the inserted
// document, which is why the email is not repeated in the values.
res, err := users.Eq("email", "[email protected]").Upsert(bson.M{
"name": "Upsert Example",
"status": "invited",
"joined": time.Now(),
})
if err != nil {
return err
}
ctx.Log().Info("upserted", "matched", res.Matched, "upserted_id", res.UpsertedID)
// Save replaces the whole document: every field not on the struct is gone.
invited, err := users.Eq("email", "[email protected]").FindOne()
if err != nil {
return err
}
invited.Status = "active"
if err := users.ByID(invited.ID.Hex()).Save(invited); err != nil {
return err
}
// A scope-less Delete would empty the collection, so it is refused: that is
// almost always a filter that was forgotten. DeleteAll is the deliberate
// version.
if _, err := users.Delete(); err == nil {
return core.New(500, "EXAMPLE_FAILED", "an unscoped Delete must be refused")
}
n, err := users.Eq("status", "expired").Delete()
if err != nil {
return err
}
ctx.Log().Info("deleted", "count", n)
return nil
}
// seedOrders gives the aggregation example something to read. CreateMany is one
// round trip for many documents, and it returns the generated ids in order.
func seedOrders(ctx core.IContext, userID bson.ObjectID) ([]string, core.IError) {
return mongorepo.New[Order](ctx).CreateMany([]Order{
{UserID: userID, Status: "paid", Total: 250, PlacedAt: utils.ToPointer(time.Now()),
Items: []OrderItem{{SKU: "SKU-1", Qty: 2, Price: 125}}},
{UserID: userID, Status: "paid", Total: 80, PlacedAt: utils.ToPointer(time.Now()),
Items: []OrderItem{{SKU: "SKU-2", Qty: 1, Price: 80}}},
{UserID: userID, Status: "cancelled", Total: 40, PlacedAt: utils.ToPointer(time.Now()),
Items: []OrderItem{{SKU: "SKU-3", Qty: 1, Price: 40}}},
})
}Typed aggregation pipelines
The repository chain becomes the leading $match, so a scope and a pipeline compose instead of being two ways to say the same thing — and paging the output of a pipeline is one call.
go
package main
import (
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/mongorepo"
"go.mongodb.org/mongo-driver/v2/bson"
)
// --- Example 5: typed aggregation pipelines --------------------------------
//
// mongorepo.Aggregate[Row](repo) is typed twice over: D is the document it
// reads, Row is what each result decodes into. The repository's own chain
// becomes the leading $match, so a scope and a pipeline compose instead of
// being two different ways to say "active users".
//
// Nothing is hidden — Stage(bson.M{…}) appends anything the builder does not
// name ($setWindowFields, Atlas $search), and Stages() hands the pipeline back.
type revenueRow struct {
Status string `bson:"_id"`
Total float64 `bson:"total"`
Orders int64 `bson:"orders"`
}
type spenderRow struct {
ID bson.ObjectID `bson:"_id"`
Email string `bson:"email"`
Spent float64 `bson:"spent"`
}
type orderRow struct {
ID bson.ObjectID `bson:"_id"`
Status string `bson:"status"`
Total float64 `bson:"total"`
UserEmail string `bson:"user_email"`
}
func aggregations(ctx core.IContext) core.IError {
if err := ensureOrders(ctx); err != nil {
return err
}
revenue, err := revenueByStatus(ctx, time.Now().AddDate(0, 0, -30))
if err != nil {
return err
}
spenders, err := topSpenders(ctx, 5)
if err != nil {
return err
}
page, err := orderPage(ctx, &core.PageOptions{Page: 1, Limit: 20, OrderBy: []string{"-total"}})
if err != nil {
return err
}
ctx.Log().Info("aggregations",
"revenue_groups", len(revenue), "spenders", len(spenders),
"orders_total", page.Total, "orders_on_page", page.Count)
return nil
}
// revenueByStatus is the plain grouping case.
//
// The $match comes from the repository chain and therefore runs *first*, which
// is the whole performance story of an aggregation: a $match before a $group is
// an index read, the same $match after it is a collection scan.
func revenueByStatus(ctx core.IContext, since time.Time) ([]revenueRow, core.IError) {
return mongorepo.Aggregate[revenueRow](
mongorepo.New[Order](ctx).Gte("placed_at", since),
).
Group("$status", bson.M{
"total": bson.M{"$sum": "$total"},
"orders": bson.M{"$sum": 1},
}).
Sort("-total").
// $group and $sort get 100MB of memory and then fail. AllowDiskUse is
// the difference between a report that keeps working as the data grows
// and one that starts failing with "Sort exceeded memory limit".
AllowDiskUse().
// The comment shows up in currentOp and the profiler — how a slow
// pipeline is identified in production without guessing.
Comment("revenue-by-status").
All()
}
// topSpenders joins. Unwind's preserveEmpty is the flag that decides whether
// users with no orders disappear: without it the join silently drops them and
// the report under-counts in a way nobody notices until someone complains.
func topSpenders(ctx core.IContext, n int64) ([]spenderRow, core.IError) {
return mongorepo.Aggregate[spenderRow](
mongorepo.New[User](ctx).Eq("status", "active"),
).
Lookup(mongorepo.Lookup{
From: "orders", LocalField: "_id", ForeignField: "user_id", As: "orders",
}).
Unwind("$orders", true).
Group("$_id", bson.M{
"email": bson.M{"$first": "$email"},
"spent": bson.M{"$sum": "$orders.total"},
}).
Sort("-spent").
Limit(n).
All()
}
// orderPage pages the *output of a pipeline* rather than of a filter — a list
// with a joined column, which no repository chain can express.
//
// Build it without $sort/$skip/$limit: Page appends them from the options. A
// page that fits comes back in one pass via $facet, so the count cannot drift
// from the items; a page too large for the single document $facet builds costs
// a second round trip instead.
func orderPage(ctx core.IContext, opts *core.PageOptions) (*core.Page[orderRow], core.IError) {
return mongorepo.Aggregate[orderRow](mongorepo.New[Order](ctx)).
// LookupOne is Lookup plus the unwind that turns a one-element array
// into an embedded document — the shape a belongs-to join is wanted in.
LookupOne(mongorepo.Lookup{
From: "users", LocalField: "user_id", ForeignField: "_id", As: "user",
}).
Project(bson.M{
"status": 1,
"total": 1,
"user_email": "$user.email",
}).
Page(opts)
}
// ensureOrders seeds the collection the first time this example runs, so the
// pipelines above have something to aggregate.
func ensureOrders(ctx core.IContext) core.IError {
n, err := mongorepo.New[Order](ctx).Count()
if err != nil || n > 0 {
return err
}
owner, err := mongorepo.New[User](ctx).Eq("email", "[email protected]").FindOne()
if err != nil {
return err
}
ids, err := seedOrders(ctx, owner.ID)
if err != nil {
return err
}
ctx.Log().Info("seeded orders", "count", len(ids))
return nil
}