From 5479142093e428d081fbb548685f7559ddb1fb08 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=BB=D0=B5=D0=BA=D1=81=D0=B0=D0=BD=D0=B4=D1=80=20?= =?UTF-8?q?=D0=97=D0=B8=D0=BC=D0=B8=D0=BD?= Date: Tue, 12 May 2026 14:30:22 +0000 Subject: [PATCH] feat(webhook): time-series histogram stats endpoint + frontend chart --- docs/design/redis-fk-projection.md | 226 ++++++++++ ordinis-admin-ui/src/api/client.ts | 18 + ordinis-admin-ui/src/api/queries.ts | 42 ++ .../webhooks/WebhookStatsHistogram.tsx | 414 ++++++++++++++++++ ordinis-admin-ui/src/i18n.ts | 20 + ordinis-admin-ui/src/routes/webhooks.$id.tsx | 8 +- ordinis-app/pom.xml | 5 + .../ordinis/app/e2e/WebhookE2ETest.java | 98 +++++ .../webhook/WebhookDeliveryRepository.java | 34 ++ .../webhook/WebhookDeliveryStatsResponse.java | 44 ++ .../WebhookSubscriptionController.java | 142 ++++++ 11 files changed, 1049 insertions(+), 2 deletions(-) create mode 100644 docs/design/redis-fk-projection.md create mode 100644 ordinis-admin-ui/src/components/webhooks/WebhookStatsHistogram.tsx create mode 100644 ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/dto/webhook/WebhookDeliveryStatsResponse.java diff --git a/docs/design/redis-fk-projection.md b/docs/design/redis-fk-projection.md new file mode 100644 index 0000000..d6c5d3e --- /dev/null +++ b/docs/design/redis-fk-projection.md @@ -0,0 +1,226 @@ +# Design: Redis Projection FK Resolution + +**Author:** zimin.an +**Date:** 2026-05-12 +**Status:** PROPOSED — awaits CEO/eng review (`/plan-eng-review`) +**Sprint estimate:** 1-2 спринта (5-10 рабочих дней с тестами + observability) +**Blockers:** none, но рекомендую делать только после v2.12.0 prod stable + +--- + +## TL;DR + +Сейчас Redis projection (writer module) пишет per-locale flattened JSON записи в Redis для dict'ей с `redis_projection_enabled=true`. **FK ссылки** (поля с `x-references: "dict.field"`) хранятся как raw FK value (`satellite_type: "OPERATIONAL"`). Read-api / consumer делает N+1 lookup чтобы получить human label (`Действующий`). + +**Предложение:** при write resolve'ить FK и сохранять label рядом с FK value в projection. Per-dict / per-FK opt-in. Plus reverse-index для cascade invalidation когда referenced record меняется. + +**Win:** один Redis GET вместо N+1, latency 1ms вместо N×1ms (для записей с 5-10 FK). + +**Risk:** консистентность projection requires careful invalidation strategy, иначе projection stale до next write referencing'а dict'и. + +--- + +## Текущее состояние + +### Что есть + +```java +// ordinis-projection-writer/.../ProjectionWriter.java +public void upsert(String bundle, String dictionary, String businessKey, JsonNode data) { + DictionaryDefinition def = ... + if (!def.isRedisProjectionEnabled()) { + recordsSkippedCounter.increment(); + return; + } + // Per-locale flattening: + for (String locale : locales) { + JsonNode flat = flatten(schema, data, locale, def.getDefaultLocale()); + redis.opsForValue().set(RedisKeys.record(bundle, dictionary, businessKey, locale), ...); + } + // + index set: + redis.opsForSet().add(RedisKeys.dictionaryIndex(bundle, dictionary), businessKey); +} +``` + +`flatten()` сейчас обрабатывает **только** `x-localized` поля — для locale=ru заменяет `{name: {ru: "Земля", en: "Earth"}}` на `{name: "Земля"}`. FK поля проходят как есть. + +### Что хочется + +В projection rows вместо: +```json +{ + "businessKey": "ISS", + "satellite_type": "OPERATIONAL", + "country": "RU" +} +``` + +Получать: +```json +{ + "businessKey": "ISS", + "satellite_type": "OPERATIONAL", + "_resolved": { + "satellite_type": "Действующий", + "country": "Россия" + } +} +``` + +Read-api сразу отдаёт нужный label без второго round-trip. + +--- + +## Сложности (почему 1-2 sprint, не 4 часа) + +### 1. Cascade invalidation + +Если `satellite_types` dict обновляется (label changes), все projection'ы записей, ссылающихся на этот type, становятся stale. Нужен **reverse index**: + +``` +fk:: → SET of (referencing_dict, referencing_bk) +``` + +Пример: +``` +fk:satellite_types:OPERATIONAL → {spacecraft:ISS, spacecraft:MIR, ...} +``` + +При write записи в `satellite_types`: +1. Resolve reverse index → list of dependent records +2. Re-write projection каждого dependent record (заново resolve'ить FK) + +Это N+1 на каждый satellite_type update. С 1000 spacecrafts ссылающихся → 1000 Redis writes per type-label change. + +**Mitigation:** background invalidation queue. Type update → enqueue invalidation event → worker processes batch'ами. Eventual consistency ~few seconds. + +### 2. Cycle detection + +Schema A → B → A через FK chain — теоретически возможен. При resolve'е нужен depth limit (или цикл detection) чтобы не зацикливаться. **Solution:** только direct FK (1 level), nested FK через chain не resolve'им. Pragmatic limit. + +### 3. Multi-locale FK labels + +Referenced record имеет `name: {ru: "...", en: "..."}`. Projection пишется per-locale. Значит FK resolve тоже per-locale → 2× writes для каждой записи. + +В принципе не страшно — сейчас тоже per-locale writes, просто payload теперь больше. + +### 4. Schema-level config + +Где включать FK resolution? Варианты: +- **A) Per-dict flag** `fk_resolution_enabled` — глобальный для всех FK полей dict'a +- **B) Per-FK через schema annotation** `"x-references": "dict.field", "x-resolve-label": true` +- **C) Always-on когда `redis_projection_enabled=true`** — без отдельного toggle + +Recommendation: **B**. Granular control, schema-author знает какие FK «hot» (labels часто читаются) vs cold (только FK value матчат при join'е). Default false для backward compat. + +### 5. Backfill при включении флага + +Когда юзер впервые `x-resolve-label: true` на FK поле, существующие projection'ы записей без resolved label остаются stale. Нужен **one-shot backfill job** который проходит всех записей dict'a, для каждого resolve'ит FK поля и rewrite'ит projection. + +Backfill job — отдельный CLI / endpoint, не auto-trigger от schema change (иначе schema edit становится heavy operation). + +### 6. Observability + +Новые метрики: +- `ordinis_projection_fk_resolved_total{dict, fk_field}` — count successful FK resolutions +- `ordinis_projection_fk_resolve_miss_total{dict, fk_field}` — FK target dict not accessible или businessKey не найден +- `ordinis_projection_cascade_invalidation_total{trigger_dict}` — cascade fires count +- `ordinis_projection_fk_resolve_duration_seconds` — histogram + +--- + +## Архитектура + +### Phase A: read-side FK resolver (1 sprint) + +Add `FkResolver` service: + +```java +@Component +public class FkResolver { + private final FlattenedRecordRepository recordRepo; + private final DictionaryDefinitionRepository defRepo; + + /** Resolve FK value → human label per locale. Cached на 5 минут. */ + @Cacheable("fk-labels") + public Optional resolveLabel( + String refDict, // "satellite_types" + String refField, // "code" — default businessKey + String fkValue, // "OPERATIONAL" + String locale // "ru" + ) { ... } +} +``` + +Update `ProjectionWriter.flatten()`: +- Pass schema property + record data + locale through walker +- При `x-references` И `x-resolve-label: true`: + - Lookup label via FkResolver + - Inject в `_resolved.` JSON object + +### Phase B: cascade invalidation (1 sprint) + +Add reverse index update в `ProjectionWriter`: +- При upsert(record) проходим schema.properties, для каждого FK field add + `fk::` → SADD (recordDict, businessKey) +- При delete — SREM + +Add `ProjectionInvalidator`: +- Subscribe to OutboxEvents `RecordUpdated{dict=X}` +- Lookup `fk:X:` → set of (depDict, depBk) +- Enqueue `ProjectionRewriteRequest` for each — async worker re-pulls record + rewrite projection +- Metrics + retry если PG read fails + +### Phase C: backfill + ops (~3 дня) + +- CLI endpoint `POST /api/v1/admin/projections/{dict}/backfill` — iterate all records, rewrite projection +- Idempotent — safe to re-run +- Progress reporting через outbox event stream + +--- + +## Open questions + +1. **Кэш FkResolver TTL.** 5 минут — компромисс между «свежие label'ы» и «не долбить PG на каждый record write». Альтернатива — invalidate cache при `RecordUpdated{dict=referenced}` event'е. Pro: instant freshness. Con: cache coupling с event stream. + +2. **Что если FK target dict не accessible (scope-hide)?** Sub-FK fields внутри record вернут partial resolve. Predictable behavior: omit `_resolved` ключ для missing FK. Read-api получит null label → fallback на raw FK value. + +3. **Batch optimization.** При invalidation cascade на 1000 dependent records — лучше batch'ом писать в Redis pipeline? Да, через `redis.executePipelined()`. Прирост ~10× throughput для high-fanout cascade. + +4. **Storage cost.** Resolved labels удваивают payload size projection'а (label per locale × FK fields). Для dict'ей с 10 FK polями + 3 locales = 30 additional fields per record. Если 10k записей × 1KB → +300MB Redis для одного dict'a. Acceptable для AltUM-tier, но нужно alert thresholds. + +5. **Eventual consistency window.** Worst case: type-label update → 1000 dependent records → 5sec batch invalidate. Read-api в это окно может вернуть stale label. SLA нужно зафиксировать (recommendation: «projection eventually consistent within 30s of source change»). + +--- + +## Альтернатива: skip FK resolution, use read-api JOIN + +Вместо пре-resolve'инга при write — read-api делает JOIN на читающей стороне (через PG или client-side). Pro: simpler, no cascade. Con: N+1 на каждое чтение, ровно то что мы хотим избежать. + +Если RPS не упёрся в predisposed, можно отложить весь FK projection. Прагматичное правило: **включать FK resolution только когда конкретный dict упёрся в latency SLA**. Default off навсегда. + +--- + +## Implementation plan (если зелёный свет) + +| Step | Effort | Owner | Notes | +|---|---|---|---| +| 1. Schema annotation `x-resolve-label` + JSON Schema validator update | 0.5d | be | Update SchemaValidator | +| 2. FkResolver service + tests | 1d | be | @Cacheable, fallback on miss | +| 3. Update ProjectionWriter.flatten() | 1d | be | `_resolved` injection | +| 4. Reverse index update on upsert/delete | 1d | be | `fk::` keys | +| 5. ProjectionInvalidator (cascade worker) | 2d | be | Outbox listener + batch rewrite | +| 6. Backfill CLI endpoint | 1d | be | Idempotent paginated iterate | +| 7. Metrics + Grafana dashboard | 0.5d | infra/be | New panels | +| 8. Integration tests (Postgres + Redis testcontainers) | 1d | be | Happy path + cascade + cycle | +| 9. Docs (ops runbook, schema annotation guide) | 0.5d | be | docs/ | +| 10. Frontend: schema editor checkbox per FK field | 0.5d | fe | DictionaryEditorDialog | +| **Total** | **8.5d** | | within 1-2 sprint budget | + +--- + +## Recommendation + +**Defer until after v2.12.0 prod stable + 2 weeks dogfooding.** Текущий read path (PG read replica) handles RPS, no urgency. Когда первый dict упрётся в latency, активируй per-FK flag для cold start этого dict'a. Build full pipeline только если 3+ dict'ам нужно. + +**Next step (если decide go):** `/plan-eng-review` на этот doc, утверждение CEO, заведение epic + sprint allocation. diff --git a/ordinis-admin-ui/src/api/client.ts b/ordinis-admin-ui/src/api/client.ts index 607defd..dd9378a 100644 --- a/ordinis-admin-ui/src/api/client.ts +++ b/ordinis-admin-ui/src/api/client.ts @@ -330,6 +330,24 @@ export type WebhookDeliveryPage = { size: number } +/** Один bucket histogram time-series stats. ISO ts + counts по status. */ +export type WebhookStatsBucket = { + ts: string + pending: number + retrying: number + delivered: number + dlq: number + total: number +} + +/** Response от GET /admin/webhooks/subscriptions/{id}/stats. */ +export type WebhookDeliveryStats = { + bucketBy: 'hour' | 'day' + from: string + to: string + buckets: WebhookStatsBucket[] +} + // === Approval Workflow v2 === export type DraftStatus = 'PENDING' | 'APPROVED' | 'REJECTED' | 'WITHDRAWN' diff --git a/ordinis-admin-ui/src/api/queries.ts b/ordinis-admin-ui/src/api/queries.ts index 2a50bc8..4affe22 100644 --- a/ordinis-admin-ui/src/api/queries.ts +++ b/ordinis-admin-ui/src/api/queries.ts @@ -21,6 +21,7 @@ import { type SchemaDependent, type SearchResponse, type WebhookDeliveryPage, + type WebhookDeliveryStats, type WebhookSubscription, } from './client' @@ -440,6 +441,47 @@ export const webhookDlqQuery = (page: number, size: number) => export const useWebhookDlq = (page: number, size: number) => useQuery(webhookDlqQuery(page, size)) +export const webhookStatsQuery = ( + id: string, + bucketBy: 'hour' | 'day', + fromIso?: string, + toIso?: string, +) => + queryOptions({ + queryKey: [ + 'webhooks', + 'stats', + id, + bucketBy, + fromIso ?? null, + toIso ?? null, + ] as const, + queryFn: async (): Promise => { + const params: Record = { bucketBy } + if (fromIso) params.from = fromIso + if (toIso) params.to = toIso + const { data } = await apiClient.get( + `/admin/webhooks/subscriptions/${encodeURIComponent(id)}/stats`, + { params }, + ) + return data + }, + // Stats для admin dashboard — fresh data важна, но автополлинг мы + // отключаем (юзер сам refresh'ит / меняет range). + staleTime: 30 * 1000, + }) + +export const useWebhookStats = ( + id: string, + bucketBy: 'hour' | 'day', + fromIso?: string, + toIso?: string, +) => + useQuery({ + ...webhookStatsQuery(id, bucketBy, fromIso, toIso), + enabled: Boolean(id), + }) + // === Dependents (Phase 1 dict-relationships-v2) === /** diff --git a/ordinis-admin-ui/src/components/webhooks/WebhookStatsHistogram.tsx b/ordinis-admin-ui/src/components/webhooks/WebhookStatsHistogram.tsx new file mode 100644 index 0000000..363f10f --- /dev/null +++ b/ordinis-admin-ui/src/components/webhooks/WebhookStatsHistogram.tsx @@ -0,0 +1,414 @@ +import { useMemo, useState } from 'react' +import { useTranslation } from 'react-i18next' +import { ArrowsClockwiseIcon } from '@phosphor-icons/react' +import { + Alert, + Button, + Card, + CardContent, + CardHeader, + CardTitle, + LoadingBlock, + QueryErrorState, +} from '@/ui' +import { useWebhookStats } from '@/api/queries' +import type { WebhookStatsBucket } from '@/api/client' +import { cn } from '@/lib/utils' + +/** + * Time-series histogram delivery'и подписки. Stacked bar chart на чистом SVG + * (без recharts/chart-js dependency'и — bundle stays slim). + * + *

Layout: + * 

+ *   ┌────────────────────────────────────────────────────────┐
+ *   │ Header: title + bucketBy toggle (hour/day) + refresh   │
+ *   ├────────────────────────────────────────────────────────┤
+ *   │ Legend: pending · retrying · delivered · dlq           │
+ *   ├────────────────────────────────────────────────────────┤
+ *   │ Chart area:                                            │
+ *   │   stacked bars per bucket, y-axis = count              │
+ *   │   x-axis ticks: каждый 4-й bucket (hour) / day name    │
+ *   │   hover tooltip → counts breakdown                     │
+ *   └────────────────────────────────────────────────────────┘
+ *
+ * + *

Empty state: «Нет данных за выбранный период» — когда все buckets total=0. + * Visible-only: фильтруем chart area до bucket'ов, где total > 0, чтобы + * sparse data не растягивала ось. X-axis labels всё равно показывают полный + * диапазон чтобы юзер видел временное окно. + */ +export function WebhookStatsHistogram({ + subscriptionId, +}: { + subscriptionId: string +}) { + const { t } = useTranslation() + const [bucketBy, setBucketBy] = useState<'hour' | 'day'>('hour') + + const stats = useWebhookStats(subscriptionId, bucketBy) + + if (stats.isLoading) { + return ( + + + {t('webhooks.histogram.title')} + + + + + + ) + } + + if (stats.isError) { + return ( + + + {t('webhooks.histogram.title')} + + + stats.refetch()} /> + + + ) + } + + const data = stats.data + if (!data) return null + + return ( + + + {t('webhooks.histogram.title')} +

+
+ + +
+ +
+ + + + + + ) +} + +const STATUS_TOKENS = { + delivered: { color: 'fill-jade', labelKey: 'webhooks.status.delivered' }, + retrying: { color: 'fill-amber', labelKey: 'webhooks.status.retrying' }, + pending: { color: 'fill-mute/60', labelKey: 'webhooks.status.pending' }, + dlq: { color: 'fill-mars', labelKey: 'webhooks.status.dlq' }, +} as const + +type StatusKey = keyof typeof STATUS_TOKENS + +const STACK_ORDER: StatusKey[] = ['delivered', 'retrying', 'pending', 'dlq'] + +const CHART_HEIGHT = 180 +const CHART_PADDING = { top: 8, right: 8, bottom: 28, left: 32 } + +function HistogramChart({ + buckets, + bucketBy, +}: { + buckets: WebhookStatsBucket[] + bucketBy: 'hour' | 'day' +}) { + const { t, i18n } = useTranslation() + const [hoveredIdx, setHoveredIdx] = useState(null) + + const maxTotal = useMemo( + () => Math.max(1, ...buckets.map((b) => b.total)), + [buckets], + ) + + // Если всё пусто — показываем empty state, не пустой chart с одним пикселем. + if (buckets.every((b) => b.total === 0)) { + return ( + + {t('webhooks.histogram.empty', { + defaultValue: 'Нет доставок за выбранный период.', + })} + + ) + } + + const innerHeight = CHART_HEIGHT - CHART_PADDING.top - CHART_PADDING.bottom + const bucketCount = buckets.length + + // ViewBox единиц: используем pixel-like coords чтобы text/тики читались. + // SVG растягивается через preserveAspectRatio (none) на full container width. + const VB_W = 800 + const VB_H = CHART_HEIGHT + const innerW = VB_W - CHART_PADDING.left - CHART_PADDING.right + const barW = innerW / bucketCount + const barGap = Math.max(1, Math.min(2, barW * 0.15)) + const actualBarW = Math.max(1, barW - barGap) + + const tickStep = bucketBy === 'hour' + ? Math.max(1, Math.floor(bucketCount / 6)) // 4 ticks per 24h + : 1 // каждый день для 7d view + + return ( +
+ +
+ + {/* Y-axis ticks (0, mid, max) */} + + {[0, 0.5, 1].map((frac) => { + const y = CHART_PADDING.top + innerHeight * (1 - frac) + const val = Math.round(maxTotal * frac) + return ( + + + + {val} + + + ) + })} + + + {/* Stacked bars */} + {buckets.map((bucket, idx) => { + const x = CHART_PADDING.left + idx * barW + barGap / 2 + let yCursor = CHART_PADDING.top + innerHeight + const isHovered = hoveredIdx === idx + return ( + setHoveredIdx(idx)} + onMouseLeave={() => setHoveredIdx(null)} + className="cursor-pointer" + > + {/* Невидимая hover-zone — full column. Чтобы hover работал + * даже на zero bucket'ах. */} + + {STACK_ORDER.map((status) => { + const value = bucket[status] + if (!value) return null + const segH = (value / maxTotal) * innerHeight + yCursor -= segH + return ( + + ) + })} + + ) + })} + + {/* X-axis ticks */} + + {buckets.map((bucket, idx) => { + if (idx % tickStep !== 0 && idx !== buckets.length - 1) return null + const x = CHART_PADDING.left + idx * barW + barW / 2 + return ( + + {formatTick(bucket.ts, bucketBy, i18n.language)} + + ) + })} + + + + {/* Tooltip overlay */} + {hoveredIdx !== null && ( + + )} +
+
+ ) +} + +function Legend() { + const { t } = useTranslation() + return ( +
+ {STACK_ORDER.map((status) => ( +
+ + + + {t(STATUS_TOKENS[status].labelKey)} +
+ ))} +
+ ) +} + +function BucketTooltip({ + bucket, + bucketBy, + locale, + position, +}: { + bucket: WebhookStatsBucket + bucketBy: 'hour' | 'day' + locale: string + position: 'left' | 'right' +}) { + const { t } = useTranslation() + return ( +
+
+ {formatTooltip(bucket.ts, bucketBy, locale)} +
+
+ {STACK_ORDER.map((status) => { + const value = bucket[status] + if (!value) return null + return ( +
+ + + + {t(STATUS_TOKENS[status].labelKey)} + {value} +
+ ) + })} +
+ {t('webhooks.histogram.total', { defaultValue: 'итого' })} + {bucket.total} +
+
+
+ ) +} + +const formatTick = (iso: string, bucketBy: 'hour' | 'day', locale: string): string => { + const d = new Date(iso) + if (bucketBy === 'hour') { + return d.toLocaleTimeString(locale === 'ru' ? 'ru-RU' : 'en-US', { + hour: '2-digit', + minute: '2-digit', + }) + } + return d.toLocaleDateString(locale === 'ru' ? 'ru-RU' : 'en-US', { + day: '2-digit', + month: '2-digit', + }) +} + +const formatTooltip = (iso: string, bucketBy: 'hour' | 'day', locale: string): string => { + const d = new Date(iso) + if (bucketBy === 'hour') { + return d.toLocaleString(locale === 'ru' ? 'ru-RU' : 'en-US', { + day: '2-digit', + month: '2-digit', + hour: '2-digit', + minute: '2-digit', + }) + } + return d.toLocaleDateString(locale === 'ru' ? 'ru-RU' : 'en-US', { + day: '2-digit', + month: 'long', + year: 'numeric', + }) +} diff --git a/ordinis-admin-ui/src/i18n.ts b/ordinis-admin-ui/src/i18n.ts index 6a33b6c..f1df44b 100644 --- a/ordinis-admin-ui/src/i18n.ts +++ b/ordinis-admin-ui/src/i18n.ts @@ -184,6 +184,17 @@ i18n 'webhooks.stats.successRateOf_one': 'из {{count}} завершённой', 'webhooks.stats.successRateOf_few': 'из {{count}} завершённых', 'webhooks.stats.successRateOf_many': 'из {{count}} завершённых', + // Status labels (used by histogram chart legend + tooltip) + 'webhooks.status.pending': 'в очереди', + 'webhooks.status.retrying': 'retry', + 'webhooks.status.delivered': 'доставлено', + 'webhooks.status.dlq': 'DLQ', + // Histogram chart + 'webhooks.histogram.title': 'История доставок', + 'webhooks.histogram.bucket.hour': '24ч', + 'webhooks.histogram.bucket.day': '7д', + 'webhooks.histogram.empty': 'Нет доставок за выбранный период.', + 'webhooks.histogram.total': 'итого', // === Friendly query error states === 'queryError.retry': 'Повторить', 'queryError.network.title': 'Нет соединения с сервером', @@ -870,6 +881,15 @@ i18n 'webhooks.stats.successRate': 'Success rate', 'webhooks.stats.successRateOf_one': 'of {{count}} terminal', 'webhooks.stats.successRateOf_other': 'of {{count}} terminal', + 'webhooks.status.pending': 'pending', + 'webhooks.status.retrying': 'retrying', + 'webhooks.status.delivered': 'delivered', + 'webhooks.status.dlq': 'DLQ', + 'webhooks.histogram.title': 'Delivery history', + 'webhooks.histogram.bucket.hour': '24h', + 'webhooks.histogram.bucket.day': '7d', + 'webhooks.histogram.empty': 'No deliveries in the selected period.', + 'webhooks.histogram.total': 'total', // === Friendly query error states === 'queryError.retry': 'Retry', 'queryError.network.title': 'No connection to server', diff --git a/ordinis-admin-ui/src/routes/webhooks.$id.tsx b/ordinis-admin-ui/src/routes/webhooks.$id.tsx index 0b9a6d5..f7c2835 100644 --- a/ordinis-admin-ui/src/routes/webhooks.$id.tsx +++ b/ordinis-admin-ui/src/routes/webhooks.$id.tsx @@ -37,6 +37,7 @@ import { import type { WebhookTestPingResult } from '@/api/client' import { SCOPE_DOT } from '@/lib/scope-style' import { WebhookStatsCards } from '@/components/webhooks/WebhookStatsCards' +import { WebhookStatsHistogram } from '@/components/webhooks/WebhookStatsHistogram' export const Route = createFileRoute('/webhooks/$id')({ component: WebhookDetailPage, @@ -207,10 +208,13 @@ function WebhookDetailPage() { {/* Stats cards — счётчики per status + success rate. Aggregate - из 4 параллельных queries с size=1 (cheap). Visual time-series - chart — deferred follow-up (требует /stats endpoint). */} + из 4 параллельных queries с size=1 (cheap). */} + {/* Time-series histogram — backend /stats?bucketBy=hour|day. SVG + stacked bars без recharts depend'ы. 24h / 7d toggle. */} + +

diff --git a/ordinis-app/pom.xml b/ordinis-app/pom.xml index fc6f8a7..de86267 100644 --- a/ordinis-app/pom.xml +++ b/ordinis-app/pom.xml @@ -68,6 +68,11 @@ spring-boot-starter-test test + + org.springframework.security + spring-security-test + test + org.springframework.kafka spring-kafka-test diff --git a/ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java b/ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java index aaaf257..2aac88e 100644 --- a/ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java +++ b/ordinis-app/src/test/java/cloud/nstart/terravault/ordinis/app/e2e/WebhookE2ETest.java @@ -22,6 +22,9 @@ import org.springframework.context.annotation.Primary; import org.springframework.http.MediaType; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.Trigger; +import org.springframework.security.oauth2.jwt.Jwt; +import org.springframework.security.oauth2.server.resource.authentication.JwtAuthenticationToken; +import org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors; import org.springframework.test.context.ActiveProfiles; import org.springframework.test.context.DynamicPropertyRegistry; import org.springframework.test.context.DynamicPropertySource; @@ -42,7 +45,9 @@ import java.util.concurrent.TimeoutException; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; /** @@ -297,6 +302,99 @@ class WebhookE2ETest { } } + /** + * Mock JWT с INTERNAL role для admin endpoints. spring-security-test + * authentication() post-processor правильно проходит через filter chain + * (filter не overwrite'ит auth когда передан таким образом). + */ + private static org.springframework.test.web.servlet.request.RequestPostProcessor internalJwt() { + Jwt jwt = Jwt.withTokenValue("test-token") + .header("alg", "none") + .claim("sub", "test-user") + .claim("realm_access", java.util.Map.of("roles", java.util.List.of("ordinis-internal"))) + .build(); + return SecurityMockMvcRequestPostProcessors.authentication( + new JwtAuthenticationToken(jwt, java.util.Collections.emptyList())); + } + + @Test + void statsEndpoint_returnsContinuousTimeline_withCorrectCounts() throws Exception { + WebhookSubscription sub = subRepo.save(new WebhookSubscription( + "stats-test", "http://127.0.0.1:1/null", "stats-secret", "test-user")); + UUID subId = sub.getId(); + + Long eventId1 = insertMinimalOutboxEvent("EVT-stats-1"); + Long eventId2 = insertMinimalOutboxEvent("EVT-stats-2"); + Long eventId3 = insertMinimalOutboxEvent("EVT-stats-3"); + + WebhookDelivery del1 = deliveryRepo.save(new WebhookDelivery(subId, eventId1)); + del1.markDelivered(200); + deliveryRepo.save(del1); + + WebhookDelivery del2 = deliveryRepo.save(new WebhookDelivery(subId, eventId2)); + del2.markFailed("connection refused", null); + deliveryRepo.save(del2); + + deliveryRepo.save(new WebhookDelivery(subId, eventId3)); + + mvc.perform(get("/api/v1/admin/webhooks/subscriptions/{id}/stats", subId) + .with(internalJwt()) + .param("bucketBy", "hour")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.bucketBy").value("hour")) + .andExpect(jsonPath("$.buckets").isArray()) + .andExpect(jsonPath("$.buckets[?(@.total > 0)].pending").value(org.hamcrest.Matchers.hasItem(1))) + .andExpect(jsonPath("$.buckets[?(@.total > 0)].retrying").value(org.hamcrest.Matchers.hasItem(1))) + .andExpect(jsonPath("$.buckets[?(@.total > 0)].delivered").value(org.hamcrest.Matchers.hasItem(1))); + } + + @Test + void statsEndpoint_rejectsInvalidBucketBy() throws Exception { + WebhookSubscription sub = subRepo.save(new WebhookSubscription( + "stats-bad", "http://127.0.0.1:1/null", "secret-bad", "test-user")); + mvc.perform(get("/api/v1/admin/webhooks/subscriptions/{id}/stats", sub.getId()) + .with(internalJwt()) + .param("bucketBy", "year")) + .andExpect(status().isBadRequest()) + .andExpect(jsonPath("$.code").value("webhook_stats_bucket_invalid")); + } + + @Test + void statsEndpoint_rejectsWindowTooLarge() throws Exception { + WebhookSubscription sub = subRepo.save(new WebhookSubscription( + "stats-big", "http://127.0.0.1:1/null", "secret-big", "test-user")); + String fromIso = java.time.OffsetDateTime.now().minusDays(60).toString(); + String toIso = java.time.OffsetDateTime.now().toString(); + mvc.perform(get("/api/v1/admin/webhooks/subscriptions/{id}/stats", sub.getId()) + .with(internalJwt()) + .param("bucketBy", "hour") + .param("from", fromIso) + .param("to", toIso)) + .andExpect(status().isBadRequest()) + .andExpect(jsonPath("$.code").value("webhook_stats_window_too_large")); + } + + @Autowired org.springframework.jdbc.core.JdbcTemplate jdbcTemplate; + + /** + * Минимальный outbox_event insert для удовлетворения FK constraint + * webhook_deliveries.outbox_event_id. NOT NULL columns set к dummy values. + */ + private Long insertMinimalOutboxEvent(String key) { + return jdbcTemplate.queryForObject( + """ + INSERT INTO outbox_events + (event_type, aggregate_type, aggregate_id, data_scope, + payload, kafka_topic, kafka_key, created_at, attempt_count) + VALUES + ('TestEvent', 'Test', ?, 'PUBLIC', + '{}'::jsonb, 'test-topic', ?, now(), 0) + RETURNING id + """, + Long.class, + key, key); + } + // ---------- no-op scheduler implementation ---------- /** diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDeliveryRepository.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDeliveryRepository.java index 84f626a..cd5dbef 100644 --- a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDeliveryRepository.java +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDeliveryRepository.java @@ -58,4 +58,38 @@ public interface WebhookDeliveryRepository extends JpaRepository findBySubscriptionIdAndStatusOrderByCreatedAtDesc( UUID subscriptionId, WebhookDelivery.Status status, Pageable pageable); + + /** + * Histogram aggregation для admin UI time-series chart. Группирует deliveries + * подписки по time bucket (hour/day) × status, возвращает count'ы. + * + *

{@code bucketUnit} — Postgres date_trunc валидное значение: + * {@code 'hour'} или {@code 'day'} (другие unit'ы service rejects). + * Параметр инъектируется через native query (нельзя bind'ить interval label + * как ?, см. PG docs «String literal type cast»). + * + *

Returns rows как {@code Object[]{OffsetDateTime bucketTs, String status, + * Long count}}. Service aggregate'ит в structured DTO с пустыми bucket'ами + * заполненными нулями (continuous timeline). + * + *

Index plan: {@code idx_webhook_deliveries_sub_created} (subscription_id, + * created_at). Если будем фильтровать без сабскрипции (всё-про-всё) — + * нужен будет ещё один index, но текущий use case admin-UI всегда per-sub. + */ + @Query(value = """ + SELECT date_trunc(:bucketUnit, created_at) AS bucket_ts, + status::text AS status, + COUNT(*) AS cnt + FROM webhook_deliveries + WHERE subscription_id = :subscriptionId + AND created_at >= :fromTs + AND created_at < :toTs + GROUP BY bucket_ts, status + ORDER BY bucket_ts ASC + """, nativeQuery = true) + List aggregateByBucket( + @Param("subscriptionId") UUID subscriptionId, + @Param("bucketUnit") String bucketUnit, + @Param("fromTs") OffsetDateTime fromTs, + @Param("toTs") OffsetDateTime toTs); } diff --git a/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/dto/webhook/WebhookDeliveryStatsResponse.java b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/dto/webhook/WebhookDeliveryStatsResponse.java new file mode 100644 index 0000000..4dbe721 --- /dev/null +++ b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/dto/webhook/WebhookDeliveryStatsResponse.java @@ -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'и одной подписки. + * + *

Endpoint: {@code GET /api/v1/webhooks/subscriptions/{id}/stats + * ?bucketBy=hour|day&from=...&to=...}. + * + *

Структура: + *

    + *
  • {@code bucketBy} — granularity, эхо из request'а ('hour' / 'day').
    • + *
  • {@code from}/{@code to} — actual диапазон, который пошёл в SQL + * (resolved дефолты если параметры были опущены).
    • + *
  • {@code buckets} — массив continuous timeline'ов: каждый bucket' + * (даже без events) представлен row'ом с zero count'ами. Frontend + * chart рисует ровную time axis без gap'ов.
    • + *
+ * + *

Bucket row: {@code ts} — start of interval, остальные — count'ы по + * каждому status'у. {@code total} = сумма для convenience. + */ +public record WebhookDeliveryStatsResponse( + String bucketBy, + OffsetDateTime from, + OffsetDateTime to, + List buckets) { + + /** + * Один time bucket с count'ами per status. + * + *

{@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) {} +} diff --git a/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/webhook/WebhookSubscriptionController.java b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/webhook/WebhookSubscriptionController.java index 47bbfb7..26dff11 100644 --- a/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/webhook/WebhookSubscriptionController.java +++ b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/web/webhook/WebhookSubscriptionController.java @@ -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'а. + * + *

Параметры: + *

    + *
  • {@code bucketBy} — granularity. Default {@code "hour"}. Allowed: + * {@code "hour"} | {@code "day"}.
    • + *
  • {@code from} — нижняя граница (inclusive). Default = now - 24h + * для hour'a, now - 7d для day'a.
    • + *
  • {@code to} — верхняя граница (exclusive). Default = now.
    • + *
+ * + *

Window cap: hour bucket'ом нельзя запросить больше 14 дней + * (336 bucket'ов), day'ем — больше 365 (год). Это защита от + * accident'ального overflow и keeps payload reasonable. + * + *

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 rows = deliveryRepository.aggregateByBucket( + id, unit, resolvedFrom, resolvedTo); + + // Map> — промежуточный 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> 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 buckets = new ArrayList<>(); + while (cursor.isBefore(endUtc)) { + Map 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 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 подхватит на