Checkpoint.

lib/cdc/mongo/debezium.go

@@ -3,9 +3,12 @@ package mongo
 import (
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"reflect"
 	"time"
 
+	"go.mongodb.org/mongo-driver/bson/primitive"
+
 	"github.com/artie-labs/transfer/lib/cdc"
 	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/debezium"
@@ -172,6 +175,12 @@ func (s *SchemaEventPayload) GetData(pkMap map[string]any, tc kafkalib.TopicConf
 		for k, v := range pkMap {
 			retMap[k] = v
 		}
+
+		if _, isOk := retMap["ts"]; !isOk {
+			slog.Info("ts not found in data, adding it", slog.String("table", s.GetTableName()))
+			objectID := retMap["_id"].(primitive.ObjectID)
+			retMap["ts"] = objectID.Timestamp()
+		}
 	case "r", "u", "c":
 		retMap = s.Payload.afterMap
 		// TODO: Remove this code.