Migrate Java Code from Spring Data MongoDB to Pure Java MongoDB Driver Reactive Streams

@GetMapping("/mostClickedElements") public Object mostClickedElements( @RequestParam(name = "projectId", required = false) String projectId, @RequestParam(name = "submoduleId", required = false) Integer submoduleId, @RequestParam(value = "pageNum", defaultValue = "1") long pageNum, @RequestParam(value = "pageSize", defaultValue = "10") long pageSize ) { return Optional.ofNullable(Flux.from(dbClient.getCollection(kapokMongoProvider.getMongoBackendCollectionName(projectId)).aggregate(Arrays.asList( KapokLogPresenterApiController.optionalMatchSubmoduleId(submoduleId),

        Aggregates.match(
            Filters.and(
                Filters.eq("category", "ui.click"),
                Filters.exists("timestamp"),
                Filters.gt("timestamp", (double)(OffsetDateTime.now().minusDays(1).toInstant().toEpochMilli()) / 1000)
            )
        ),

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

        Aggregates.project(
            Projections.fields(
                Projections.include("timestamp", "message"),
                Projections.computed("transactionObject", Document.parse("{$arrayElemAt: ['$transactionObject', 0]} ")),
                Projections.computed("eventObject", Document.parse("{$arrayElemAt: ['$eventObject', 0]} "))
            )
        ),

        Aggregates.group(new Document("timestamp", "$timestamp").append("elemSelector", "$message"),
            Accumulators.first("txUserId", "$transactionObject.user.id"),
            Accumulators.first("evUserId", "$eventObject.user.id")
        ),

        Aggregates.project(
            Projections.fields(
                Projections.excludeId(),
                Projections.computed("timestamp", "$_id.timestamp"),
                Projections.computed("elemSelector", "$_id.elemSelector"),
                Projections.computed("userId", Document.parse(" { $ifNull: ['$txUserId', '$evUserId'] } "))
            )
        ),

        Aggregates.match(Filters.exists("userId")),

        Aggregates.group(new Document("elemSelector", "$elemSelector").append("userId", "$userId"),
            Accumulators.sum("opNum", 1)
        ),

        Aggregates.project(
            Projections.fields(
                Projections.excludeId(),
                Projections.computed("opNum", "$opNum"),
                Projections.computed("elemSelector", "$_id.elemSelector"),
                Projections.computed("userId", "$_id.userId")
            )
        ),

        Aggregates.group("$elemSelector",
            Accumulators.sum("totalOpNum", "$opNum"),
            Accumulators.sum("totalUserNum", 1)
        ),

        Aggregates.project(Projections.fields(
            Projections.excludeId(),
            Projections.computed("totalOpNum", "$totalOpNum"),
            Projections.computed("totalUserNum", "$totalUserNum"),
            Projections.computed("elemSelector", "$_id")
        )),

        Aggregates.sort(Sorts.descending("totalOpNum")),

        Aggregates.facet(
            Aggregates.skip((pageNum - 1) * pageSize),
            Aggregates.limit(pageSize)
        ).as("list"),
        Aggregates.count().as("count")
    )).allowDiskUse(true).first()).blockLast())
    .orElse(new Document(){{
        put("list", Collections.emptyList());
        put("count", Collections.emptyList());
    }});
}