MongoDB 7+ Patterns

Polymorphic pattern

Documents de types différents dans la même collection :

// events collection
{ _id: "e1", type: "login", userId: "u1", at: ISODate() }
{ _id: "e2", type: "purchase", userId: "u1", items: [...], total: 99.99 }
{ _id: "e3", type: "signup", userId: "u2", referrer: "google" }

Un index sur type accélère le filtrage.

Attribute pattern

Pour des attributs variables sans créer N colonnes :

// Products with variable attributes
{
  _id: "p1",
  name: "Laptop",
  attributes: [
    { k: "ram", v: "16GB" },
    { k: "storage", v: "512GB SSD" },
    { k: "weight", v: "1.5kg" }
  ]
}

// Index on attributes.k
db.products.createIndex({ "attributes.k": 1, "attributes.v": 1 })

Bucket pattern

Pour des séries temporelles — grouper les events par bucket de temps :

// Au lieu de N documents (1 par mesure)
// 1 document par heure avec 3600 mesures
{
  _id: "sensor1_2026-04-08T10",
  sensorId: "sensor1",
  hour: ISODate("2026-04-08T10:00:00Z"),
  measurements: [
    { ts: ISODate("2026-04-08T10:00:00Z"), value: 22.5 },
    { ts: ISODate("2026-04-08T10:00:01Z"), value: 22.6 },
    // ... 3598 more
  ],
  count: 3600,
  avg: 22.55
}

Computed pattern

Pré-calculer les agrégations dans le document :

// Product with pre-computed rating
{
  _id: "p1",
  name: "Book",
  reviewCount: 150,
  ratingAvg: 4.3,
  ratingTotal: 645
}

Update atomique lors d'un nouvel avis :

await db.collection('products').updateOne(
  { _id: productId },
  {
    $inc: { reviewCount: 1, ratingTotal: newRating },
    $set: { ratingAvg: (ratingTotal + newRating) / (reviewCount + 1) }
  }
)

Subset pattern

Stocker un subset des sous-documents dans le parent pour éviter une jointure :

{
  _id: "p1",
  name: "Laptop",
  // 3 most recent reviews embedded
  recentReviews: [
    { _id: "r100", rating: 5, text: "..." },
    { _id: "r99", rating: 4, text: "..." },
    { _id: "r98", rating: 5, text: "..." }
  ],
  // All reviews in separate collection
  totalReviews: 1500
}

2. Indexes

// Single field
db.users.createIndex({ email: 1 }, { unique: true })

// Compound — order matters!
db.posts.createIndex({ authorId: 1, publishedAt: -1 })

// Multikey — pour les arrays
db.products.createIndex({ tags: 1 })

// Text search
db.articles.createIndex({ title: "text", body: "text" })

// Geospatial 2dsphere
db.places.createIndex({ location: "2dsphere" })

// Partial — seulement certains docs
db.orders.createIndex(
  { userId: 1, status: 1 },
  { partialFilterExpression: { status: "pending" } }
)

// TTL — auto-delete après N secondes
db.sessions.createIndex({ expiresAt: 1 }, { expireAfterSeconds: 0 })

// Wildcard — tous les champs
db.dynamic.createIndex({ "$**": 1 })

Règles d'index

• ESR rule — Equality, Sort, Range (ordre des champs dans un compound index)
• Covered queries — projection + index contient tous les champs = pas de fetch document
• Max 64 indexes / collection (limite dure)
• Index intersection — MongoDB peut combiner 2 indexes single-field, mais les compound sont toujours plus rapides

3. Aggregation pipeline

// Exemple complet : top 10 authors par nombre de posts publiés ce mois-ci
const result = await db.collection('posts').aggregate([
  {
    $match: {
      publishedAt: { $gte: startOfMonth, $lt: endOfMonth },
      status: 'published'
    }
  },
  {
    $group: {
      _id: '$authorId',
      postCount: { $sum: 1 },
      lastPost: { $max: '$publishedAt' }
    }
  },
  { $sort: { postCount: -1 } },
  { $limit: 10 },
  {
    $lookup: {
      from: 'users',
      localField: '_id',
      foreignField: '_id',
      as: 'author'
    }
  },
  { $unwind: '$author' },
  {
    $project: {
      authorName: '$author.name',
      postCount: 1,
      lastPost: 1,
      _id: 0
    }
  }
]).toArray()

Stages utiles

4. Transactions (multi-document ACID)

const session = client.startSession()

try {
  await session.withTransaction(async () => {
    await db.collection('accounts').updateOne(
      { _id: fromAccountId },
      { $inc: { balance: -amount } },
      { session }
    )
    await db.collection('accounts').updateOne(
      { _id: toAccountId },
      { $inc: { balance: amount } },
      { session }
    )
    await db.collection('transactions').insertOne(
      { from: fromAccountId, to: toAccountId, amount, at: new Date() },
      { session }
    )
  })
} finally {
  await session.endSession()
}

Limitations :

• Transactions sur replica set (pas standalone)
• Timeout par défaut 60s
• Coût overhead — à éviter pour du single-document, utiliser les opérations atomiques single-doc à la place

5. Change Streams (CDC)

const changeStream = db.collection('orders').watch([
  { $match: { 'fullDocument.status': 'paid' } }
], { fullDocument: 'updateLookup' })

changeStream.on('change', async (change) => {
  if (change.operationType === 'update') {
    await sendOrderConfirmationEmail(change.fullDocument)
  }
})

Use cases : envoi d'email, sync vers Elasticsearch, audit log, webhooks sortants.

6. Node.js drivers

Driver officiel (recommandé pour perf)

npm install mongodb

import { MongoClient } from 'mongodb'

const client = new MongoClient(process.env.MONGO_URI!, {
  serverSelectionTimeoutMS: 5000,
})

await client.connect()
const db = client.db('myapp')

const users = db.collection<User>('users')
const user = await users.findOne({ email: '[email protected]' })

Mongoose ODM (schema validation + hooks)

npm install mongoose

import mongoose from 'mongoose'

const UserSchema = new mongoose.Schema({
  email: { type: String, required: true, unique: true, lowercase: true, trim: true },
  name: { type: String, required: true },
  createdAt: { type: Date, default: Date.now }
})

UserSchema.index({ email: 1 })

UserSchema.pre('save', function (next) {
  if (this.isModified('email')) this.email = this.email.toLowerCase()
  next()
})

export const User = mongoose.model('User', UserSchema)

7. Python drivers

motor (async)

pip install motor

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client.myapp

async def find_user(email: str):
    return await db.users.find_one({"email": email})

Beanie ODM (Pydantic-based, async)

pip install beanie

from beanie import Document, Indexed
from pydantic import EmailStr

class User(Document):
    email: Indexed(EmailStr, unique=True)
    name: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Settings:
        name = "users"

# Query
user = await User.find_one(User.email == "[email protected]")

8. Atlas Search + Vector Search

Full-text search

// Créer l'index dans Atlas UI puis :
const results = await db.collection('articles').aggregate([
  {
    $search: {
      index: 'default',
      text: {
        query: 'machine learning',
        path: ['title', 'body'],
        fuzzy: {}
      }
    }
  },
  { $limit: 10 }
]).toArray()

Vector Search (Atlas 2024+)

const embedding = await getEmbedding('query text')

const results = await db.collection('documents').aggregate([
  {
    $vectorSearch: {
      index: 'vector_index',
      path: 'embedding',
      queryVector: embedding,
      numCandidates: 100,
      limit: 10
    }
  }
]).toArray()

9. Sharding key selection

Règles :

• Haute cardinalité (beaucoup de valeurs uniques)
• Bonne distribution des writes (pas de hotspot)
• Query isolation — les queries communes peuvent cibler un seul shard

Bons candidats : { userId: 1, createdAt: 1 }, { tenantId: 1, _id: 1 } Mauvais candidats : _id seul (hotspot sur le last shard), timestamp seul (hotspot écritures récentes)

Anti-patterns

1. Unbounded arrays — orders.items: [...] qui grandit à l'infini → document grows > 16MB limit
2. Pas d'index sur les queries communes → full collection scan
3. $regex non ancré — $regex: "foo" est slow, $regex: "^foo" utilise l'index
4. Many $lookup — MongoDB n'est pas relationnel, au bout de 3 joins → repenser le schema
5. Transactions pour du single-doc — utilise les operations atomiques natives ($inc, $push, $set)
6. Write concern w: 0 — pas d'ACK, perte de data garantie
7. Pas de replica set en prod — single node = SPOF
8. findOneAndUpdate avec upsert: true sans index unique → race condition
9. skip pour la pagination deep — O(N) sur le skip, utiliser _id > lastSeenId à la place

Checklist production

• Replica set ≥ 3 nodes (HA)
• Backups — Atlas PITR ou mongodump cron
• Indexes sur toutes les queries fréquentes + uniqueness constraints
• Connection pool sized (default 100 is fine)
• Write concern w: "majority" pour les writes critiques
• Read preference primary par défaut, secondaryPreferred pour reporting
• TLS obligatoire en prod
• Auth — users dédiés par service, least privilege
• Monitoring — Atlas Metrics ou Prometheus mongo_exporter
• Slow query log activé (> 100ms)

Quand déléguer

• PostgreSQL relationnel → skill postgres-patterns (dans ce plugin)
• Redis cache → skill redis-patterns (dans ce plugin)
• Review sécurité → database-reviewer
• Migration SQL → Mongo → migration-expert

Ressources

• https://www.mongodb.com/docs/manual/
• https://www.mongodb.com/docs/manual/tutorial/use-schema-design-patterns/
• https://www.mongodb.com/docs/atlas/atlas-search/
• https://mongoosejs.com/docs/
• https://beanie-odm.dev/