Migrate the following Java code from spring-data-mongodb to pure java mongodb-driver-reactivestreams Use new DocumentfieldName if a JSON object is to be created You may use something like Fluxfromblo

@GetMapping("/frontendOverallStat") public Object frontendOverallStat( @RequestParam(name = "projectId", required = false) String projectId, @RequestParam(name = "submoduleId", required = false) Integer submoduleId ) { return Optional.ofNullable( Flux.from(mongoTemplate.aggregate( Aggregation.newAggregation( KapokLogPresenterApiController.optionalMatchSubmoduleId(submoduleId),

                Aggregation.match(Criteria.where("category").is("navigation").and("timestamp").exists(true).gt(
                    (double)(OffsetDateTime.now().minusDays(1).toInstant().toEpochMilli()) / 1000
                ).and("data.to").exists(true).ne("")),

                Aggregation.lookup(kapokMongoProvider.getMongoTransactionsCollectionName(projectId), "event_id", "event_id", "transactionObject"),
                Aggregation.lookup(kapokMongoProvider.getMongoEventsCollectionName(projectId), "event_id", "event_id", "eventObject"),

                Aggregation.project("timestamp", "data")
                    .and(ArrayOperators.ArrayElemAt.arrayOf("transactionObject").elementAt(0)).as("transactionObject")
                    .and(ArrayOperators.ArrayElemAt.arrayOf("eventObject").elementAt(0)).as("eventObject"),

                Aggregation.group("timestamp", "data.to")
                    .first("transactionObject.user.id").as("txUserId")
                    .first("eventObject.user.id").as("evUserId")
                    .first("eventObject.level").as("evLevel"),

                Aggregation.project()
                    .and("_id.timestamp").as("timestamp")
                    .and("_id.to").as("path")
                    .and(ConditionalOperators.ifNull("txUserId").thenValueOf("evUserId")).as("userId")
                    .and(ComparisonOperators.Eq.valueOf("evLevel").equalToValue("error")).as("hasError"),

                Aggregation.match(Criteria.where("userId").exists(true)),

                Aggregation.group("path", "userId")
                    .sum(ConditionalOperators.when("hasError").then(1).otherwise(0)).as("errorNum")
                    .count().as("pageAccessNum"),

                Aggregation.project("errorNum", "pageAccessNum")
                    .and("_id.path").as("path")
                    .and("_id.userId").as("userId"),
                    
                Aggregation.group("userId")
                    .sum("errorNum").as("errorNum")
                    .sum("pageAccessNum").as("pageAccessNum")
                    .count().as("pageNum"),

                Aggregation.project("errorNum", "pageNum", "pageAccessNum")
                    .and("_id").as("userId"),

                Aggregation.group()
                    .count().as("totalUserNum")
                    .sum("errorNum").as("totalErrorNum")
                    .sum("pageAccessNum").as("pageAccessNum"),

                Aggregation.project("totalUserNum")
                    .and(ArithmeticOperators.Divide.valueOf("totalErrorNum").divideBy("pageAccessNum")).as("avgErrorNumPerPageAccess")
                    .and(ArithmeticOperators.Divide.valueOf("pageAccessNum").divideBy("totalUserNum")).as("avgPageAccessNumPerUser")

            ).withOptions(AggregationOptions.builder().allowDiskUse(true).build()),
            kapokMongoProvider.getMongoBreadcrumbsCollectionName(projectId),
            Document.class
        ).blockLast())
    .orElse(frontendOverallStatDefaultDocument);
}