Mongo Go Driver Change Stream Watch

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

func watchOrders(ctx context.Context, coll *mongo.Collection, resumeToken bson.Raw) {
	match := bson.M{"operationType": bson.M{"$in": bson.A{"insert", "update"}}}

	stream, err := coll.Watch(ctx, mongo.Pipeline{
		{{Key: "$match", Value: match}},
	}, options.ChangeStream().
		SetFullDocument(options.UpdateLookup).
		SetResumeAfter(resumeToken))
	if err != nil {
		log.Printf("watch failed: %v", err)
		return
	}
	defer stream.Close(ctx)

	for stream.Next(ctx) {
		var event struct {
			OperationType string             `bson:"operationType"`
			FullDocument  bson.M             `bson:"fullDocument"`
			ResumeToken   bson.Raw           `bson:"_id"`
		}
		if err := stream.Decode(&event); err != nil {
			log.Printf("decode failed: %v", err)
			continue
		}
		fmt.Printf("%s -> %v\n", event.OperationType, event.FullDocument["status"])
		// 持久化 resumeToken 后即可在进程重启时断点续传
	}

	if err := stream.Err(); err != nil {
		log.Printf("stream error: %v", err)
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := mongo.Connect(ctx, options.Client().ApplyURI("mongodb://localhost:27017"))
	if err != nil {
		log.Fatal(err)
	}
	defer client.Disconnect(ctx)

	coll := client.Database("shop").Collection("orders")
	watchOrders(ctx, coll, nil)
}