@Override
    public Stream<CasEventAggregate> aggregate(final Class type, final Duration start) {
        val initialSql = """
            SELECT
                DATE_BIN(INTERVAL '1 hour', time) AS window,
                "type",
                "tenant",
                COUNT("value") AS count
              FROM "${measurement}"
              WHERE ${where}
              GROUP BY window, "type", "tenant"
              ORDER BY window, "type", "tenant";
            """
            .stripLeading()
            .stripTrailing();

        var whereClause = "time >= NOW() - INTERVAL '%s hours'".formatted(start.toHours());
        if (type != null) {
            whereClause += " AND \"type\" = '%s'".formatted(type.getName());
        }
        val sub = new StringSubstitutor(Map.of(
            "measurement", MEASUREMENT,
            "where", whereClause
        ));
        val sql = sub.replace(initialSql);
        LOGGER.debug("Executing SQL query [{}]", sql);
        try (val rows = influxDbConnectionFactory.query(MEASUREMENT, sql)) {
            return rows.map(row -> new CasEventAggregate(
                row.getField("window", LocalDateTime.class),
                row.getTag("type"),
                row.getIntegerField("count"),
                row.getTag("tenant")
            ));
        }
    }
This is {@link InfluxDbCasEventRepository}.

@author Misagh Moayyed
@since 5.2.0
support/cas-server-support-events-influxdb/src/main/java/org/apereo/cas/support/events/dao/InfluxDbCasEventRepository.java