package de.bwaldvogel.mongo.oplog;

import java.time.Clock;
import java.time.Instant;
import java.util.List;
import java.util.stream.Collectors;

import de.bwaldvogel.mongo.MongoCollection;
import de.bwaldvogel.mongo.bson.BsonTimestamp;
import de.bwaldvogel.mongo.bson.Document;

public class CollectionBackedOplog extends AbstractOplog {

    private final MongoCollection<Document> collection;

    public CollectionBackedOplog(Clock clock, MongoCollection<Document> collection) {
        super(clock);
        this.collection = collection;
    }

    @Override
    public void handleInsert(String databaseName, Document query) {
        Instant instant = clock.instant();
        List<Document> documents = (List<Document>) query.get("documents");
        List<Document> oplogDocuments = documents.stream().map(d ->
            new OplogDocument()
                .withTimestamp(new BsonTimestamp(instant.toEpochMilli()))
                .withWall(instant)
                .withOperationType(OperationType.INSERT)
                .withOperationDocument(d)
                .withNamespace(String.format("%s.%s", databaseName, query.get("insert")))
                .asDocument()).collect(Collectors.toList());
        collection.insertDocuments(oplogDocuments);
    }

    @Override
    public void handleUpdate(String databaseName, Document query) {
    }

    @Override
    public void handleDelete(String databaseName, Document query) {

    }
}

@@ -637,40 +637,83 @@
     
-Instant instant = clock.instant(
+String namespace = String.format(%22%25s.%25s%22, databaseName, query.get(%22insert%22)
 );%0A 
@@ -849,54 +849,343 @@
 am()
-.map(d -%3E%0A            new OplogDocument()%0A    
+%0A            .map(oplogDocument -%3E toOplogDocument(oplogDocument, namespace))%0A            .collect(Collectors.toList());%0A        collection.insertDocuments(oplogDocuments);%0A    %7D%0A%0A    private Document toOplogDocument(Document oplogDocument, String namespace) %7B%0A        Instant now = clock.instant();%0A        return new OplogDocument()%0A
     
@@ -1229,30 +1229,11 @@
 amp(
-instant.toEpochMilli()
+now
 ))%0A 
@@ -1239,28 +1239,24 @@
             
-  
-  
 .withWall(in
@@ -1257,21 +1257,13 @@
 all(
-instant)%0A    
+now)%0A
     
@@ -1315,36 +1315,32 @@
 RT)%0A            
-    
 .withOperationDo
@@ -1350,14 +1350,23 @@
 ent(
-d)%0A   
+oplogDocument)%0A
     
@@ -1369,25 +1369,24 @@
             
- 
 .withNamespa
@@ -1392,177 +1392,43 @@
 ace(
-String.format(%22%25s.%25s%22, databaseName, query.get(%22insert%22)))%0A                .asDocument()).collect(Collectors.toList());%0A        collection.insertDocuments(oplogDocuments
+namespace)%0A            .asDocument(
 );%0A 

Refactor CollectionBackedOplog