Merge branch 'feat/webhook-stats-histogram' into 'main'

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

See merge request 2-6/2-6-4/terravault/ordinis!143

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:<referenced_dict>:<businessKey> → 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<String> 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.<field>` JSON object
+
+### Phase B: cascade invalidation (1 sprint)
+
+Add reverse index update в `ProjectionWriter`:
+- При upsert(record) проходим schema.properties, для каждого FK field add
+  `fk:<refDict>:<fkValue>` → SADD (recordDict, businessKey)
+- При delete — SREM
+
+Add `ProjectionInvalidator`:
+- Subscribe to OutboxEvents `RecordUpdated{dict=X}`
+- Lookup `fk:X:<businessKey>` → 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:<refDict>:<bk>` 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.

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'

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<WebhookDeliveryStats> => {
+      const params: Record<string, string> = { bucketBy }
+      if (fromIso) params.from = fromIso
+      if (toIso) params.to = toIso
+      const { data } = await apiClient.get<WebhookDeliveryStats>(
+        `/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) ===
 
 /**

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).
+ *
+ * <p>Layout:
+ * <pre>
+ *   ┌────────────────────────────────────────────────────────┐
+ *   │ 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                     │
+ *   └────────────────────────────────────────────────────────┘
+ * </pre>
+ *
+ * <p>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 (
+      <Card>
+        <CardHeader>
+          <CardTitle>{t('webhooks.histogram.title')}</CardTitle>
+        </CardHeader>
+        <CardContent>
+          <LoadingBlock />
+        </CardContent>
+      </Card>
+    )
+  }
+
+  if (stats.isError) {
+    return (
+      <Card>
+        <CardHeader>
+          <CardTitle>{t('webhooks.histogram.title')}</CardTitle>
+        </CardHeader>
+        <CardContent>
+          <QueryErrorState error={stats.error} onRetry={() => stats.refetch()} />
+        </CardContent>
+      </Card>
+    )
+  }
+
+  const data = stats.data
+  if (!data) return null
+
+  return (
+    <Card>
+      <CardHeader className="flex flex-row items-center justify-between gap-3">
+        <CardTitle>{t('webhooks.histogram.title')}</CardTitle>
+        <div className="flex items-center gap-2">
+          <div className="inline-flex rounded-md border border-line overflow-hidden text-cap">
+            <button
+              type="button"
+              onClick={() => setBucketBy('hour')}
+              className={cn(
+                'px-2.5 py-1 transition-colors',
+                bucketBy === 'hour'
+                  ? 'bg-accent text-on-accent'
+                  : 'bg-surface text-ink-2 hover:bg-surface-2',
+              )}
+            >
+              {t('webhooks.histogram.bucket.hour', { defaultValue: '24ч' })}
+            </button>
+            <button
+              type="button"
+              onClick={() => setBucketBy('day')}
+              className={cn(
+                'px-2.5 py-1 transition-colors border-l border-line',
+                bucketBy === 'day'
+                  ? 'bg-accent text-on-accent'
+                  : 'bg-surface text-ink-2 hover:bg-surface-2',
+              )}
+            >
+              {t('webhooks.histogram.bucket.day', { defaultValue: '7д' })}
+            </button>
+          </div>
+          <Button
+            variant="ghost"
+            size="sm"
+            onClick={() => stats.refetch()}
+            disabled={stats.isFetching}
+            aria-label={t('common.refresh')}
+            title={t('common.refresh')}
+          >
+            <ArrowsClockwiseIcon
+              size={14}
+              weight="regular"
+              className={stats.isFetching ? 'animate-spin' : undefined}
+            />
+          </Button>
+        </div>
+      </CardHeader>
+      <CardContent>
+        <HistogramChart buckets={data.buckets} bucketBy={data.bucketBy} />
+      </CardContent>
+    </Card>
+  )
+}
+
+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<number | null>(null)
+
+  const maxTotal = useMemo(
+    () => Math.max(1, ...buckets.map((b) => b.total)),
+    [buckets],
+  )
+
+  // Если всё пусто — показываем empty state, не пустой chart с одним пикселем.
+  if (buckets.every((b) => b.total === 0)) {
+    return (
+      <Alert variant="info" className="mt-2">
+        {t('webhooks.histogram.empty', {
+          defaultValue: 'Нет доставок за выбранный период.',
+        })}
+      </Alert>
+    )
+  }
+
+  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 (
+    <div className="space-y-2">
+      <Legend />
+      <div className="relative">
+        <svg
+          viewBox={`0 0 ${VB_W} ${VB_H}`}
+          preserveAspectRatio="none"
+          className="w-full h-44"
+          role="img"
+          aria-label={t('webhooks.histogram.title')}
+        >
+          {/* Y-axis ticks (0, mid, max) */}
+          <g className="text-mute" fontSize="10" fontFamily="ui-monospace, monospace">
+            {[0, 0.5, 1].map((frac) => {
+              const y = CHART_PADDING.top + innerHeight * (1 - frac)
+              const val = Math.round(maxTotal * frac)
+              return (
+                <g key={frac}>
+                  <line
+                    x1={CHART_PADDING.left}
+                    x2={VB_W - CHART_PADDING.right}
+                    y1={y}
+                    y2={y}
+                    stroke="currentColor"
+                    strokeOpacity={frac === 0 ? 0.4 : 0.15}
+                    strokeWidth={1}
+                  />
+                  <text
+                    x={CHART_PADDING.left - 4}
+                    y={y + 3}
+                    textAnchor="end"
+                    fill="currentColor"
+                  >
+                    {val}
+                  </text>
+                </g>
+              )
+            })}
+          </g>
+
+          {/* 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 (
+              <g
+                key={bucket.ts}
+                onMouseEnter={() => setHoveredIdx(idx)}
+                onMouseLeave={() => setHoveredIdx(null)}
+                className="cursor-pointer"
+              >
+                {/* Невидимая hover-zone — full column. Чтобы hover работал
+                  * даже на zero bucket'ах. */}
+                <rect
+                  x={x - barGap / 2}
+                  y={CHART_PADDING.top}
+                  width={barW}
+                  height={innerHeight}
+                  fill="transparent"
+                />
+                {STACK_ORDER.map((status) => {
+                  const value = bucket[status]
+                  if (!value) return null
+                  const segH = (value / maxTotal) * innerHeight
+                  yCursor -= segH
+                  return (
+                    <rect
+                      key={status}
+                      x={x}
+                      y={yCursor}
+                      width={actualBarW}
+                      height={segH}
+                      className={cn(
+                        STATUS_TOKENS[status].color,
+                        'transition-opacity',
+                        isHovered ? 'opacity-100' : 'opacity-85',
+                      )}
+                    />
+                  )
+                })}
+              </g>
+            )
+          })}
+
+          {/* X-axis ticks */}
+          <g className="text-mute" fontSize="10" fontFamily="ui-monospace, monospace">
+            {buckets.map((bucket, idx) => {
+              if (idx % tickStep !== 0 && idx !== buckets.length - 1) return null
+              const x = CHART_PADDING.left + idx * barW + barW / 2
+              return (
+                <text
+                  key={bucket.ts}
+                  x={x}
+                  y={VB_H - 10}
+                  textAnchor="middle"
+                  fill="currentColor"
+                >
+                  {formatTick(bucket.ts, bucketBy, i18n.language)}
+                </text>
+              )
+            })}
+          </g>
+        </svg>
+
+        {/* Tooltip overlay */}
+        {hoveredIdx !== null && (
+          <BucketTooltip
+            bucket={buckets[hoveredIdx]}
+            bucketBy={bucketBy}
+            locale={i18n.language}
+            // Position: справа от hovered bar если в первой половине, иначе слева
+            position={hoveredIdx < buckets.length / 2 ? 'right' : 'left'}
+          />
+        )}
+      </div>
+    </div>
+  )
+}
+
+function Legend() {
+  const { t } = useTranslation()
+  return (
+    <div className="flex flex-wrap items-center gap-3 text-cap text-mute">
+      {STACK_ORDER.map((status) => (
+        <div key={status} className="inline-flex items-center gap-1.5">
+          <svg width={10} height={10} className="shrink-0">
+            <rect
+              x={0}
+              y={0}
+              width={10}
+              height={10}
+              className={STATUS_TOKENS[status].color}
+              rx={2}
+            />
+          </svg>
+          <span>{t(STATUS_TOKENS[status].labelKey)}</span>
+        </div>
+      ))}
+    </div>
+  )
+}
+
+function BucketTooltip({
+  bucket,
+  bucketBy,
+  locale,
+  position,
+}: {
+  bucket: WebhookStatsBucket
+  bucketBy: 'hour' | 'day'
+  locale: string
+  position: 'left' | 'right'
+}) {
+  const { t } = useTranslation()
+  return (
+    <div
+      className={cn(
+        'pointer-events-none absolute top-1 z-10 rounded-md border border-line bg-surface-2 px-3 py-2 shadow-lg text-cap',
+        position === 'right' ? 'left-2' : 'right-2',
+      )}
+    >
+      <div className="text-mono text-ink-2 font-semibold mb-1">
+        {formatTooltip(bucket.ts, bucketBy, locale)}
+      </div>
+      <div className="space-y-0.5">
+        {STACK_ORDER.map((status) => {
+          const value = bucket[status]
+          if (!value) return null
+          return (
+            <div key={status} className="flex items-center gap-2">
+              <svg width={8} height={8} className="shrink-0">
+                <rect
+                  x={0}
+                  y={0}
+                  width={8}
+                  height={8}
+                  className={STATUS_TOKENS[status].color}
+                  rx={2}
+                />
+              </svg>
+              <span className="text-ink-2">{t(STATUS_TOKENS[status].labelKey)}</span>
+              <span className="text-mono text-ink ml-auto">{value}</span>
+            </div>
+          )
+        })}
+        <div className="border-t border-line mt-1 pt-1 flex items-center justify-between text-mute">
+          <span>{t('webhooks.histogram.total', { defaultValue: 'итого' })}</span>
+          <span className="text-mono text-ink">{bucket.total}</span>
+        </div>
+      </div>
+    </div>
+  )
+}
+
+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',
+  })
+}

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',

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() {
       </Panel>
 
       {/* Stats cards — счётчики per status + success rate. Aggregate
-          из 4 параллельных queries с size=1 (cheap). Visual time-series
+          из 4 параллельных queries с size=1 (cheap). */}
-          chart — deferred follow-up (требует /stats endpoint). */}
       <WebhookStatsCards subscriptionId={id} />
 
+      {/* Time-series histogram — backend /stats?bucketBy=hour|day. SVG
+          stacked bars без recharts depend'ы. 24h / 7d toggle. */}
+      <WebhookStatsHistogram subscriptionId={id} />
+
       <div>
         <div className="flex items-center justify-between mb-3">
           <h2 className="font-sans text-title-md text-accent">

ordinis-app/pom.xml

@@ -68,6 +68,11 @@
       <artifactId>spring-boot-starter-test</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.security</groupId>
+      <artifactId>spring-security-test</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.springframework.kafka</groupId>
       <artifactId>spring-kafka-test</artifactId>

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 ----------
 
   /**

ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDeliveryRepository.java

@@ -58,4 +58,38 @@ public interface WebhookDeliveryRepository extends JpaRepository<WebhookDelivery
   /** История доставок одной подписки фильтрованная по status. */
   Page<WebhookDelivery> findBySubscriptionIdAndStatusOrderByCreatedAtDesc(
       UUID subscriptionId, WebhookDelivery.Status status, Pageable pageable);
+
+  /**
+   * Histogram aggregation для admin UI time-series chart. Группирует deliveries
+   * подписки по time bucket (hour/day) × status, возвращает count'ы.
+   *
+   * <p>{@code bucketUnit} — Postgres date_trunc валидное значение:
+   * {@code 'hour'} или {@code 'day'} (другие unit'ы service rejects).
+   * Параметр инъектируется через native query (нельзя bind'ить interval label
+   * как ?, см. PG docs «String literal type cast»).
+   *
+   * <p>Returns rows как {@code Object[]{OffsetDateTime bucketTs, String status,
+   * Long count}}. Service aggregate'ит в structured DTO с пустыми bucket'ами
+   * заполненными нулями (continuous timeline).
+   *
+   * <p>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<Object[]> aggregateByBucket(
+      @Param("subscriptionId") UUID subscriptionId,
+      @Param("bucketUnit") String bucketUnit,
+      @Param("fromTs") OffsetDateTime fromTs,
+      @Param("toTs") OffsetDateTime toTs);
 }

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'и одной подписки.
+ *
+ * <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) {}
+}

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'а.
+   *
+   * <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 подхватит на