mongoDB - toàn tập

{
 "_id": ObjectId("507f1f77bcf86cd799439011"),
 "name": "John Doe",
 "email": "john@example.com",
 "age": 30,
 "address": {
   "street": "123 Main St",
   "city": "Ho Chi Minh",
   "country": "VN"
 },
 "tags": ["developer", "java"],
 "orders": [
   { "orderId": "ORD-001", "total": 500000, "date": ISODate("2024-01-15") }
 ],
 "createdAt": ISODate("2024-01-01T00:00:00Z"),
 "metadata": { "source": "web", "version": 2 }
}

// User với addresses embedded
{
  "_id": 1,
  "name": "John",
  "addresses": [
    { "type": "home", "city": "HCM" },
    { "type": "work", "city": "HN" }
  ]
}

// Order references user
{ "_id": "ORD-001", "userId": ObjectId("..."), "total": 500000 }

// Insert
db.users.insertOne({ name: 'John', email: 'john@x.com' })
db.users.insertMany([{ name: 'A' }, { name: 'B' }])
 
// Find
db.users.find({ age: { $gte: 18, $lte: 65 } })
db.users.find({ tags: { $in: ['java', 'spring'] } })
db.users.find({ 'address.city': 'HCM' }) // nested field
db.users.findOne({ email: 'john@x.com' })
 
// Update
db.users.updateOne({ _id: ObjectId('...') }, { $set: { name: 'Bob' }, $inc: { loginCount: 1 } })
db.users.updateMany({ status: 'inactive' }, { $set: { archived: true } })
 
// Delete
db.users.deleteOne({ _id: ObjectId('...') })
db.users.deleteMany({ createdAt: { $lt: ISODate('2023-01-01') } })

// Single & Unique Index
db.users.createIndex({ email: 1 }, { unique: true })
 
// Compound Index: Sức mạnh thực sự
db.orders.createIndex({ userId: 1, createdAt: -1 })
 
// TTL Index: Tự động dọn dẹp dữ liệu (thích hợp cho session, OTP)
db.sessions.createIndex({ createdAt: 1 }, { expireAfterSeconds: 3600 })
 
// Partial Index: Chỉ index những document thỏa mãn điều kiện, tiết kiệm dung lượng RAM
db.orders.createIndex({ createdAt: 1 }, { partialFilterExpression: { status: 'PENDING' } })
 
// Wildcard (dynamic fields)
db.products.createIndex({ 'metadata.$**': 1 })

db.orders.aggregate([
  // Stage 1: Filter
  {
    $match: { status: 'COMPLETED', createdAt: { $gte: ISODate('2024-01-01') } },
  },
 
  // Stage 2: Lookup (JOIN)
  {
    $lookup: {
      from: 'users',
      localField: 'userId',
      foreignField: '_id',
      as: 'user',
    },
  },
  { $unwind: '$user' },
 
  // Stage 3: Group
  {
    $group: {
      _id: '$user.city',
      totalRevenue: { $sum: '$total' },
      orderCount: { $count: {} },
      avgOrderValue: { $avg: '$total' },
    },
  },
 
  // Stage 4: Sort
  { $sort: { totalRevenue: -1 } },
 
  // Stage 5: Limit
  { $limit: 10 },
 
  // Stage 6: Project (reshape output)
  {
    $project: {
      city: '$_id',
      totalRevenue: 1,
      orderCount: 1,
      avgOrderValue: { $round: ['$avgOrderValue', 2] },
    },
  },
])

// Spring Data MongoDB
@ReadPreference("secondaryPreferred")
List<User> findByCity(String city);

// Đảm bảo write replicated trước khi confirm
mongoTemplate.setWriteConcern(WriteConcern.MAJORITY); // wait for majority ack

@Document(collection = "process_definitions")
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ProcessDefinition {
 
   @Id
   private String id;
 
   @Indexed(unique = true)
   private String key;
 
   private String name;
   private String workspaceId;
   private int version;
 
   @Field("created_at")
   private Instant createdAt;
 
   private Map<String, Object> metadata;
   private List<String> tags;
}

public interface ProcessDefinitionRepository extends MongoRepository<ProcessDefinition, String> {
 
   List<ProcessDefinition> findByWorkspaceIdAndVersion(String workspaceId, int version);
 
   @Query("{ 'workspaceId': ?0, 'tags': { $in: ?1 } }")
   List<ProcessDefinition> findByWorkspaceAndTags(String workspaceId, List<String> tags);
 
   @Aggregation(pipeline = {
       "{ $match: { workspaceId: ?0 } }",
       "{ $group: { _id: '$key', latestVersion: { $max: '$version' } } }"
   })
   List<Document> findLatestVersions(String workspaceId);
}

@Service
public class ProcessDefinitionService {
 
   private final MongoTemplate mongoTemplate;
 
   public List<ProcessDefinition> search(String workspace, String keyword, Pageable pageable) {
       Criteria criteria = new Criteria().andOperator(
           Criteria.where("workspaceId").is(workspace),
           new Criteria().orOperator(
               Criteria.where("name").regex(keyword, "i"),
               Criteria.where("key").regex(keyword, "i")
           )
       );
 
       Query query = new Query(criteria)
           .with(pageable)
           .with(Sort.by(Sort.Direction.DESC, "created_at"));
 
       return mongoTemplate.find(query, ProcessDefinition.class);
   }
}

@Component
@Slf4j
@RequiredArgsConstructor
public class OrderChangeStreamListener {
 
   private final MongoTemplate mongoTemplate;
   private final KafkaTemplate<String, Object> kafkaTemplate;
 
   @PostConstruct
   public void watchOrderChanges() {
       mongoTemplate.getCollection("orders")
           .watch(List.of(
               Aggregates.match(Filters.in("operationType", "insert", "update"))
           ))
           .forEach(event -> {
               log.info("Document thay đổi: {} | Thao tác: {}",
                   event.getDocumentKey(), event.getOperationType());
 
               // Bắn event ra Kafka để các service khác xử lý (Outbox pattern)
               kafkaTemplate.send("order-events", event.getFullDocument());
           });
   }
}