feat(webhook): time-series histogram stats endpoint + frontend chart

2026-05-12 14:30:22 +00:00
parent 91f4310a77
commit 5479142093
11 changed files with 1049 additions and 2 deletions
@@ -0,0 +1,44 @@
+package cloud.nstart.terravault.ordinis.restapi.dto.webhook;
+
+import java.time.OffsetDateTime;
+import java.util.List;
+
+/**
+ * Response для time-bucketed histogram статистики delivery'и одной подписки.
+ *
+ * <p>Endpoint: {@code GET /api/v1/webhooks/subscriptions/{id}/stats
+ * ?bucketBy=hour|day&from=...&to=...}.
+ *
+ * <p>Структура:
+ * <ul>
+ *   <li>{@code bucketBy} — granularity, эхо из request'а ('hour' / 'day').</li>
+ *   <li>{@code from}/{@code to} — actual диапазон, который пошёл в SQL
+ *       (resolved дефолты если параметры были опущены).</li>
+ *   <li>{@code buckets} — массив continuous timeline'ов: каждый bucket'
+ *       (даже без events) представлен row'ом с zero count'ами. Frontend
+ *       chart рисует ровную time axis без gap'ов.</li>
+ * </ul>
+ *
+ * <p>Bucket row: {@code ts} — start of interval, остальные — count'ы по
+ * каждому status'у. {@code total} = сумма для convenience.
+ */
+public record WebhookDeliveryStatsResponse(
+    String bucketBy,
+    OffsetDateTime from,
+    OffsetDateTime to,
+    List<Bucket> buckets) {
+
+  /**
+   * Один time bucket с count'ами per status.
+   *
+   * <p>{@code ts} — начало интервала (date_trunc(bucketUnit, created_at)).
+   * Frontend chart использует это как X-axis label.
+   */
+  public record Bucket(
+      OffsetDateTime ts,
+      long pending,
+      long retrying,
+      long delivered,
+      long dlq,
+      long total) {}
+}
@@ -8,6 +8,7 @@ import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscription;
 import cloud.nstart.terravault.ordinis.domain.webhook.WebhookSubscriptionRepository;
 import cloud.nstart.terravault.ordinis.restapi.dto.webhook.CreateWebhookSubscriptionRequest;
 import cloud.nstart.terravault.ordinis.restapi.dto.webhook.WebhookDeliveryResponse;
+import cloud.nstart.terravault.ordinis.restapi.dto.webhook.WebhookDeliveryStatsResponse;
 import cloud.nstart.terravault.ordinis.restapi.dto.webhook.WebhookSubscriptionResponse;
 import cloud.nstart.terravault.ordinis.restapi.error.OrdinisException;
 import cloud.nstart.terravault.ordinis.restapi.service.webhook.WebhookSubscriptionService;
@@ -30,7 +31,14 @@ import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.ResponseStatus;
 import org.springframework.web.bind.annotation.RestController;

+import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.UUID;

 /**
@@ -162,6 +170,140 @@ public class WebhookSubscriptionController {
    return deliveries.map(WebhookDeliveryResponse::from);
  }

+  /**
+   * Time-bucketed histogram delivery'и подписки для admin UI chart'а.
+   *
+   * <p>Параметры:
+   * <ul>
+   *   <li>{@code bucketBy} — granularity. Default {@code "hour"}. Allowed:
+   *       {@code "hour"} | {@code "day"}.</li>
+   *   <li>{@code from} — нижняя граница (inclusive). Default = now - 24h
+   *       для hour'a, now - 7d для day'a.</li>
+   *   <li>{@code to} — верхняя граница (exclusive). Default = now.</li>
+   * </ul>
+   *
+   * <p>Window cap: hour bucket'ом нельзя запросить больше 14 дней
+   * (336 bucket'ов), day'ем — больше 365 (год). Это защита от
+   * accident'ального overflow и keeps payload reasonable.
+   *
+   * <p>Response: {@link WebhookDeliveryStatsResponse} с continuous timeline
+   * (пустые bucket'ы заполнены нулями — chart рендерит ровный X-axis без
+   * gap'ов).
+   */
+  @GetMapping("/subscriptions/{id}/stats")
+  public WebhookDeliveryStatsResponse stats(
+      @PathVariable UUID id,
+      @RequestParam(value = "bucketBy", defaultValue = "hour") String bucketBy,
+      @RequestParam(value = "from", required = false) OffsetDateTime from,
+      @RequestParam(value = "to", required = false) OffsetDateTime to) {
+    requireInternal();
+
+    // Validate bucketBy и резолвим default window.
+    String unit = bucketBy.toLowerCase();
+    if (!ALLOWED_BUCKETS.contains(unit)) {
+      throw OrdinisException.badRequest(
+          "webhook_stats_bucket_invalid",
+          "Unknown bucketBy: " + bucketBy + ". Allowed: hour, day.");
+    }
+
+    OffsetDateTime now = OffsetDateTime.now();
+    OffsetDateTime resolvedTo = to != null ? to : now;
+    OffsetDateTime resolvedFrom =
+        from != null
+            ? from
+            : resolvedTo.minus(unit.equals("hour") ? Duration.ofHours(24) : Duration.ofDays(7));
+
+    if (!resolvedFrom.isBefore(resolvedTo)) {
+      throw OrdinisException.badRequest(
+          "webhook_stats_range_invalid",
+          "from must be before to (from=" + resolvedFrom + ", to=" + resolvedTo + ").");
+    }
+
+    // Window cap — prevent runaway аллокации bucket'ов.
+    long maxBuckets = unit.equals("hour") ? MAX_HOUR_BUCKETS : MAX_DAY_BUCKETS;
+    ChronoUnit chronoUnit = unit.equals("hour") ? ChronoUnit.HOURS : ChronoUnit.DAYS;
+    long requestedBuckets = chronoUnit.between(resolvedFrom, resolvedTo);
+    if (requestedBuckets > maxBuckets) {
+      throw OrdinisException.badRequest(
+          "webhook_stats_window_too_large",
+          "Запрошенное окно " + requestedBuckets + " " + unit
+              + " bucket'ов превышает лимит " + maxBuckets + ". "
+              + "Уменьшите диапазон или используйте более крупный bucketBy.");
+    }
+
+    // Native query → агрегируем в bucket map → fill empty bucket'ы.
+    List<Object[]> rows = deliveryRepository.aggregateByBucket(
+        id, unit, resolvedFrom, resolvedTo);
+
+    // Map<bucketTs, Map<status, count>> — промежуточный shape для merge'а
+    // status-row'ов в один Bucket dto. Native query возвращает по row на каждую
+    // (ts, status) пару.
+    //
+    // Hibernate native query mapping для Postgres timestamp-without-tz column
+    // возвращает {@link java.time.Instant} (default JDK time mapping). Bucket
+    // key хотим OffsetDateTime в UTC — convert через atOffset(UTC). Это
+    // совместимо с resolvedFrom.truncatedTo()/plus() ниже.
+    Map<OffsetDateTime, Map<String, Long>> byTs = new HashMap<>();
+    for (Object[] r : rows) {
+      OffsetDateTime ts = toOffsetDateTime(r[0]);
+      String status = (String) r[1];
+      Long cnt = ((Number) r[2]).longValue();
+      byTs.computeIfAbsent(ts, k -> new HashMap<>()).put(status, cnt);
+    }
+
+    // Continuous timeline — bucket'ы каждые `chronoUnit` от truncated(from) до to
+    // (exclusive). Truncated чтобы первый bucket совпал с DB date_trunc.
+    //
+    // Bucket cursor + byTs keys both в UTC чтобы equals() работал. Postgres
+    // date_trunc хранит результат в session TZ, но native query mapping
+    // возвращает Instant (UTC). Cursor конвертим в UTC same instant для
+    // консистентного lookup'a.
+    OffsetDateTime cursor = truncateTo(resolvedFrom.withOffsetSameInstant(
+        java.time.ZoneOffset.UTC), chronoUnit);
+    OffsetDateTime endUtc = resolvedTo.withOffsetSameInstant(java.time.ZoneOffset.UTC);
+    List<WebhookDeliveryStatsResponse.Bucket> buckets = new ArrayList<>();
+    while (cursor.isBefore(endUtc)) {
+      Map<String, Long> counts = byTs.getOrDefault(cursor, Map.of());
+      long pending = counts.getOrDefault("pending", 0L);
+      long retrying = counts.getOrDefault("retrying", 0L);
+      long delivered = counts.getOrDefault("delivered", 0L);
+      long dlq = counts.getOrDefault("dlq", 0L);
+      long total = pending + retrying + delivered + dlq;
+      buckets.add(new WebhookDeliveryStatsResponse.Bucket(
+          cursor, pending, retrying, delivered, dlq, total));
+      cursor = cursor.plus(1, chronoUnit);
+    }
+
+    return new WebhookDeliveryStatsResponse(unit, resolvedFrom, resolvedTo, buckets);
+  }
+
+  private static final Set<String> ALLOWED_BUCKETS = Set.of("hour", "day");
+  private static final long MAX_HOUR_BUCKETS = 14L * 24; // 14 дней × 24h = 336
+  private static final long MAX_DAY_BUCKETS = 365L;       // year
+
+  /** Truncate timestamp до начала bucket'а — match'ит SQL date_trunc. */
+  private static OffsetDateTime truncateTo(OffsetDateTime ts, ChronoUnit unit) {
+    return ts.truncatedTo(unit);
+  }
+
+  /**
+   * Convert native query timestamp value to {@code OffsetDateTime} в UTC.
+   * Hibernate возвращает {@link java.time.Instant} для PG timestamp без TZ,
+   * или OffsetDateTime для timestamp with TZ — handle оба case'а.
+   */
+  private static OffsetDateTime toOffsetDateTime(Object raw) {
+    if (raw == null) return null;
+    if (raw instanceof OffsetDateTime odt) return odt;
+    if (raw instanceof java.time.Instant inst) {
+      return inst.atOffset(java.time.ZoneOffset.UTC);
+    }
+    if (raw instanceof java.sql.Timestamp ts) {
+      return ts.toInstant().atOffset(java.time.ZoneOffset.UTC);
+    }
+    throw new IllegalStateException(
+        "Unexpected timestamp type from native query: " + raw.getClass().getName());
+  }
+
  /**
   * Replay delivery из DLQ или retrying state — сбрасывает attemptCount=0,
   * dlqAt=null, status=retrying, nextAttemptAt=now. Dispatcher подхватит на