Schema, streams and consistency
The parts that depend on the server rather than on the code: what indexes exist, what a replica set makes possible, and what a transaction is actually for.
Indexes at boot, and change streams
Mongo indexes are declared in code and applied at startup, which is idempotent — but it also means a removed index is never dropped. A change stream is a live feed, not a queue.
go
package main
import (
"context"
"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 6: indexes and change streams ---------------------------------
//
// Unlike SQL, where migrations own the schema, Mongo indexes are declared in
// code and applied at boot. That works because creating one is idempotent and
// there is no schema to drift from — but it also means a *removed* MongoIndex
// is not dropped. Retiring an index is a deliberate DropIndex.
func indexesAndStreams(ctx core.IContext) core.IError {
if err := ensureIndexes(ctx); err != nil {
return err
}
if err := listIndexes(ctx); err != nil {
return err
}
// A change stream needs a replica set. On a standalone server this fails
// immediately with a message saying so, which is why main treats a failed
// step as information rather than as a reason to stop.
return watchUsers(ctx, 2*time.Second)
}
// ensureIndexes is safe to run on every boot.
func ensureIndexes(ctx core.IContext) core.IError {
if err := mongorepo.New[User](ctx).EnsureIndexes(
// A unique index is the only thing that actually enforces uniqueness —
// a check-then-insert in application code loses the race. Partial keeps
// it honest alongside soft deletes: a deleted user should not keep
// holding the email address forever.
core.MongoIndex{
Keys: []string{"email"},
Unique: true,
Partial: bson.M{"deleted_at": nil},
},
// Equality, then sort, then range. This serves Eq("status", …) and
// Eq("status", …).Sort("-joined"); it does not serve Sort("-joined")
// on its own, and no amount of hoping changes that.
core.MongoIndex{Keys: []string{"status", "-joined"}},
// A non-numeric direction goes after a colon. One array field per
// compound index — {tags, items.sku} is rejected, because the number of
// index entries would be the product of the two arrays.
core.MongoIndex{Keys: []string{"tags"}},
core.MongoIndex{Keys: []string{"name:text"}},
); err != nil {
return err
}
// Through the driver layer, for a collection with no document type of its
// own. TTL is the right answer for sessions, one-time tokens and rate-limit
// rows; it is the wrong answer for data with a retention *policy*, because
// it deletes silently and leaves no record that it did.
return ctx.DBMongo().EnsureIndexes("sessions",
core.MongoIndex{Keys: []string{"token"}, Unique: true},
core.MongoIndex{Keys: []string{"created_at"}, TTL: 24 * time.Hour},
)
}
// listIndexes is what to run before deleting one: dropping an index a query
// depends on turns that query into a collection scan, quietly, under production
// load.
func listIndexes(ctx core.IContext) core.IError {
list, err := ctx.DBMongo().ListIndexes("users")
if err != nil {
return err
}
names := make([]string, 0, len(list))
for _, index := range list {
if name, ok := index["name"].(string); ok {
names = append(names, name)
}
}
ctx.Log().Info("user indexes", "names", names)
return nil
}
// watchUsers is a live feed of the writes to a collection.
//
// The repository re-points its filter at fullDocument when it builds the
// pipeline, so Eq("status", "active").Watch() means what it reads like — the
// changes to active users, not the events whose top level has a status field.
//
// budget exists because an example has to end. A real subscriber runs until the
// process stops, and stores the resume token so it can reopen where it left off.
func watchUsers(ctx core.IContext, budget time.Duration) core.IError {
stream, err := mongorepo.New[User](ctx).Eq("status", "active").Watch()
if err != nil {
return err
}
defer func() { _ = stream.Close(ctx) }()
watchCtx, cancel := context.WithTimeout(ctx, budget)
defer cancel()
seen := 0
for stream.Next(watchCtx) {
var event struct {
OperationType string `bson:"operationType"`
FullDocument User `bson:"fullDocument"`
}
if decodeErr := stream.Decode(&event); decodeErr != nil {
return core.Wrap(decodeErr, "mongo: decode change event")
}
seen++
ctx.Log().Info("user changed",
"op", event.OperationType, "email", event.FullDocument.Email)
}
ctx.Log().Info("change stream closed", "events", seen)
// A change stream is a live feed, not a queue: no acknowledgement, no retry,
// no dead letter. A consumer that goes away loses whatever happened while it
// was gone. If losing an event would be a bug, write the fact down durably
// and let a job act on it.
//
// The deadline this example imposed on itself is not one of those failures,
// so it is not reported as one.
if streamErr := stream.Err(); streamErr != nil && !errors.Is(streamErr, context.DeadlineExceeded) {
return core.Wrap(streamErr, "mongo: change stream")
}
return nil
}Two collections that have to agree
Needs a replica set, and needs the tx handle: a write made through the outer one commits immediately and survives the abort. See Transactions.
go
package main
import (
"errors"
"strconv"
"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 7: transactions ------------------------------------------------
//
// Mongo only offers transactions on a replica set or a sharded cluster. A
// standalone server — including the single-node `docker run mongo` most people
// develop against — refuses, with a message saying so. See main.go for the
// one-liner that starts a single-node replica set instead; without it this
// whole example fails, which is exactly what a service would do in production.
//
// Returning nil from the callback commits; returning an error or panicking
// aborts.
func transactions(ctx core.IContext) core.IError {
owner, err := mongorepo.New[User](ctx).Eq("email", "[email protected]").FindOne()
if err != nil {
return err
}
order := Order{
UserID: owner.ID,
Status: "paid",
Total: 99,
Items: []OrderItem{{SKU: "SKU-9", Qty: 1, Price: 99}},
PlacedAt: utils.ToPointer(time.Now()),
}
if err := placeOrder(ctx, owner.ID, &order); err != nil {
return err
}
ctx.Log().Info("order committed", "id", order.ID.Hex())
// Act on a transaction only after it has committed. Publishing inside one
// lets a subscriber see an event for work that then aborts — and the
// subscriber has no way to find out.
// ctx.PubSub().Publish("order.created", order)
return rollbackIsRealRollback(ctx)
}
// placeOrder writes two collections that have to agree.
//
// The rule that causes every bug: use the `tx` handle inside. The outer handle
// — ctx.DBMongo(), or any repository built with New inside the closure — is not
// in the transaction, and a write made through it commits immediately and
// survives the abort.
func placeOrder(ctx core.IContext, userID bson.ObjectID, order *Order) core.IError {
return ctx.DBMongo().Transaction(func(tx core.IMongoDB) error {
orders := mongorepo.NewIn[Order](tx)
users := mongorepo.NewIn[User](tx)
if err := orders.Create(order); err != nil {
return err
}
_, err := users.ByID(userID.Hex()).Inc("order_count", 1)
return err
})
}
// rollbackIsRealRollback proves the abort actually rolls back, using a marker
// nothing else writes so a leftover document from a previous run cannot make
// the check pass by accident.
func rollbackIsRealRollback(ctx core.IContext) core.IError {
marker := "rollback-" + strconv.FormatInt(time.Now().UnixNano(), 10)
err := ctx.DBMongo().Transaction(func(tx core.IMongoDB) error {
if _, insertErr := tx.InsertOne("orders", Order{
Status: marker, PlacedAt: utils.ToPointer(time.Now()),
}); insertErr != nil {
return insertErr
}
return errors.New("deliberate failure")
})
if err == nil {
return core.New(500, "EXAMPLE_FAILED", "the transaction should have aborted")
}
left, countErr := mongorepo.New[Order](ctx).Eq("status", marker).Count()
if countErr != nil {
return countErr
}
if left != 0 {
return core.New(500, "EXAMPLE_FAILED", "the aborted insert should be gone")
}
ctx.Log().Info("abort rolled the insert back", "marker", marker)
return nil
}
// singleCollectionTransaction is the shorter spelling when everything happens in
// one collection: the repository binds itself to the session for you.
func singleCollectionTransaction(ctx core.IContext, id string) core.IError {
return mongorepo.New[User](ctx).Transaction(func(tx *mongorepo.Repo[User]) error {
if _, err := tx.ByID(id).Inc("logins", 1); err != nil {
return err
}
_, err := tx.ByID(id).Updates(bson.M{"status": "active"})
return err
})
}
// rawDriverInsideATransaction is the trap worth knowing about. Every helper on
// the tx handle already runs on the session's context; a *raw driver call* does
// not unless you give it one — and the version that lands outside the
// transaction succeeds, is never rolled back, and logs nothing to distinguish
// itself.
func rawDriverInsideATransaction(ctx core.IContext) core.IError {
return ctx.DBMongo().Transaction(func(tx core.IMongoDB) error {
collection := tx.Collection("orders")
// ✅ in the transaction
_, err := collection.InsertOne(tx.Context(), Order{Status: "raw", PlacedAt: utils.ToPointer(time.Now())})
// ❌ silently outside it — no error, no rollback:
// collection.InsertOne(context.Background(), order)
return err
})
}
// What belongs inside a transaction is only the writes that must succeed or
// fail together. A session holds locks on the documents it touches and has a
// 60-second server limit, so an HTTP call, an upload or a long loop inside one
// turns someone else's latency into your lock-hold time.
//
// And prefer designs that need no transaction at all: $inc and $addToSet are
// already atomic, FindOneAndUpdate claims a document in one operation, and a
// write to a single document never needed a transaction in the first place —
// which is the modelling advantage embedding buys you.