Merge branch 'staging' into feat/kafka-driver-change

Author: saksham-datazip

drivers/mongodb/internal/cdc.go

@@ -29,11 +29,12 @@ const (
 var ErrIdleTermination = errors.New("change stream terminated due to idle timeout")
 
 type CDCDocument struct {
-	OperationType string              `json:"operationType"`
-	FullDocument  map[string]any      `json:"fullDocument"`
-	ClusterTime   primitive.Timestamp `json:"clusterTime"`
-	WallTime      primitive.DateTime  `json:"wallTime"`
-	DocumentKey   map[string]any      `json:"documentKey"`
+	OperationType            string              `json:"operationType"`
+	FullDocument             map[string]any      `json:"fullDocument"`
+	FullDocumentBeforeChange map[string]any      `json:"fullDocumentBeforeChange"`
+	ClusterTime              primitive.Timestamp `json:"clusterTime"`
+	WallTime                 primitive.DateTime  `json:"wallTime"`
+	DocumentKey              map[string]any      `json:"documentKey"`
 }
 
 func (m *Mongo) ChangeStreamConfig() (bool, bool, bool) {
@@ -45,7 +46,7 @@ func (m *Mongo) PreCDC(cdcCtx context.Context, streams []types.StreamInterface)
 		collection := m.client.Database(stream.Namespace(), options.Database().SetReadConcern(readconcern.Majority())).Collection(stream.Name())
 		pipeline := mongo.Pipeline{
 			{{Key: "$match", Value: bson.D{
-				{Key: "operationType", Value: bson.D{{Key: "$in", Value: bson.A{"insert", "update", "delete"}}}},
+				{Key: "operationType", Value: bson.D{{Key: "$in", Value: bson.A{"insert", "update", "replace", "delete"}}}},
 			}}},
 		}
 
@@ -95,10 +96,10 @@ func (m *Mongo) StreamChanges(ctx context.Context, streamIndex int, metadataStat
 	}
 	pipeline := mongo.Pipeline{
 		{{Key: "$match", Value: bson.D{
-			{Key: "operationType", Value: bson.D{{Key: "$in", Value: bson.A{"insert", "update", "delete"}}}},
+			{Key: "operationType", Value: bson.D{{Key: "$in", Value: bson.A{"insert", "update", "replace", "delete"}}}},
 		}}},
 	}
-	changeStreamOpts := options.ChangeStream().SetFullDocument(options.UpdateLookup).SetMaxAwaitTime(maxAwait)
+	changeStreamOpts := options.ChangeStream().SetFullDocument(options.UpdateLookup).SetFullDocumentBeforeChange(options.WhenAvailable).SetMaxAwaitTime(maxAwait)
 	collection := m.client.Database(stream.Namespace(), options.Database().SetReadConcern(readconcern.Majority())).Collection(stream.Name())
 
 	changeStreamOpts = changeStreamOpts.SetResumeAfter(map[string]any{cdcCursorField: prevResumeToken})
@@ -168,9 +169,19 @@ func (m *Mongo) handleChangeDoc(ctx context.Context, cursor *mongo.ChangeStream,
 		return fmt.Errorf("error while decoding: %s", err)
 	}
 
-	if record.OperationType == "delete" {
-		// replace full document(null) with documentKey
-		record.FullDocument = record.DocumentKey
+	record.OperationType = normalizeOperationType(record.OperationType)
+
+	switch record.OperationType {
+	case "delete":
+		if record.FullDocumentBeforeChange != nil {
+			record.FullDocument = record.FullDocumentBeforeChange
+		} else {
+			record.FullDocument = record.DocumentKey
+		}
+	case "update":
+		if record.FullDocument == nil && record.FullDocumentBeforeChange != nil {
+			record.FullDocument = record.FullDocumentBeforeChange
+		}
 	}
 
 	filterMongoObject(record.FullDocument)
@@ -298,3 +309,15 @@ func GetResumeToken(cursor *mongo.ChangeStream, streamID string) (string, error)
 
 	return token, nil
 }
+
+// normalizeOperationType maps MongoDB-specific operation types to the standard
+// set understood by the abstract CDC layer (insert, update, delete).
+// "replace" swaps the full document but keeps _id unchanged, so it is treated as an update.
+func normalizeOperationType(opType string) string {
+	switch opType {
+	case "replace":
+		return "update"
+	default:
+		return opType
+	}
+}