diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDelivery.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDelivery.java new file mode 100644 index 0000000..5eae9fb --- /dev/null +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDelivery.java @@ -0,0 +1,153 @@ +package cloud.nstart.terravault.ordinis.domain.webhook; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.Table; + +import java.time.OffsetDateTime; +import java.util.UUID; + +/** + * Webhook delivery — одна попытка доставки event'а подписчику. Один (subscription_id, + * outbox_event_id) row на match (uniq index). Жизненный цикл: + * + * 
+ *   pending → retrying → delivered  (2xx response)
+ *                     ↓
+ *                     dlq           (после attempt_count >= 1000)
+ *
+ * + *

Backoff schedule: 1m, 5m, 15m, 1h, 6h, 24h (exponential), capped 24h. + * Computed в {@link #scheduleNextAttempt()}. + */ +@Entity +@Table(name = "webhook_deliveries") +public class WebhookDelivery { + + public enum Status { + pending, + retrying, + delivered, + dlq + } + + /** Backoff schedule in seconds. attemptCount-1 индексирует. После — last value. */ + private static final long[] BACKOFF_SECONDS = { + 60L, // 1 min + 5 * 60L, // 5 min + 15 * 60L, // 15 min + 60 * 60L, // 1 h + 6 * 60 * 60L, // 6 h + 24 * 60 * 60L // 24 h + }; + + /** После этого attempt_count → DLQ flag. Совпадает с outbox cap. */ + public static final int MAX_ATTEMPTS = 1000; + + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + @Column(name = "id") + private Long id; + + @Column(name = "subscription_id", nullable = false) + private UUID subscriptionId; + + @Column(name = "outbox_event_id", nullable = false) + private Long outboxEventId; + + @Enumerated(EnumType.STRING) + @Column(name = "status", nullable = false, length = 16) + private Status status = Status.pending; + + @Column(name = "attempt_count", nullable = false) + private int attemptCount = 0; + + @Column(name = "created_at", nullable = false) + private OffsetDateTime createdAt; + + @Column(name = "next_attempt_at", nullable = false) + private OffsetDateTime nextAttemptAt; + + @Column(name = "last_attempt_at") + private OffsetDateTime lastAttemptAt; + + @Column(name = "delivered_at") + private OffsetDateTime deliveredAt; + + @Column(name = "last_error", columnDefinition = "TEXT") + private String lastError; + + @Column(name = "last_status_code") + private Integer lastStatusCode; + + @Column(name = "dlq_at") + private OffsetDateTime dlqAt; + + protected WebhookDelivery() {} + + public WebhookDelivery(UUID subscriptionId, Long outboxEventId) { + this.subscriptionId = subscriptionId; + this.outboxEventId = outboxEventId; + OffsetDateTime now = OffsetDateTime.now(); + this.createdAt = now; + this.nextAttemptAt = now; + } + + public void markDelivered(int statusCode) { + this.status = Status.delivered; + this.deliveredAt = OffsetDateTime.now(); + this.lastAttemptAt = this.deliveredAt; + this.lastStatusCode = statusCode; + this.lastError = null; + } + + /** Отметить попытку как failed и запланировать retry, либо DLQ если cap. */ + public void markFailed(String error, Integer statusCode) { + this.attemptCount++; + this.lastAttemptAt = OffsetDateTime.now(); + this.lastError = error; + this.lastStatusCode = statusCode; + + if (this.attemptCount >= MAX_ATTEMPTS) { + this.status = Status.dlq; + this.dlqAt = this.lastAttemptAt; + return; + } + this.status = Status.retrying; + scheduleNextAttempt(); + } + + /** Reset DLQ flag — поллер снова возьмётся (e.g. оператор fixed receiver URL). */ + public void resetDlq() { + this.status = Status.retrying; + this.dlqAt = null; + this.attemptCount = 0; + this.nextAttemptAt = OffsetDateTime.now(); + } + + private void scheduleNextAttempt() { + int idx = Math.min(this.attemptCount - 1, BACKOFF_SECONDS.length - 1); + long seconds = BACKOFF_SECONDS[idx]; + this.nextAttemptAt = OffsetDateTime.now().plusSeconds(seconds); + } + + public Long getId() { return id; } + public UUID getSubscriptionId() { return subscriptionId; } + public Long getOutboxEventId() { return outboxEventId; } + public Status getStatus() { return status; } + public int getAttemptCount() { return attemptCount; } + public OffsetDateTime getCreatedAt() { return createdAt; } + public OffsetDateTime getNextAttemptAt() { return nextAttemptAt; } + public OffsetDateTime getLastAttemptAt() { return lastAttemptAt; } + public OffsetDateTime getDeliveredAt() { return deliveredAt; } + public String getLastError() { return lastError; } + public Integer getLastStatusCode() { return lastStatusCode; } + public OffsetDateTime getDlqAt() { return dlqAt; } + + public boolean isDlq() { return status == Status.dlq; } +} 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 new file mode 100644 index 0000000..4db3d02 --- /dev/null +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookDeliveryRepository.java @@ -0,0 +1,57 @@ +package cloud.nstart.terravault.ordinis.domain.webhook; + +import org.springframework.data.domain.Page; +import org.springframework.data.domain.Pageable; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; + +import java.time.OffsetDateTime; +import java.util.List; +import java.util.UUID; + +public interface WebhookDeliveryRepository extends JpaRepository { + + /** + * Due for retry: status IN (pending, retrying) AND next_attempt_at <= now(). + * Использует partial index {@code idx_webhook_deliveries_due}. Limit через + * Pageable чтобы dispatcher не выгребал >1000 за тик. + */ + @Query(""" + SELECT d FROM WebhookDelivery d + WHERE d.status IN ('pending','retrying') + AND d.nextAttemptAt <= :now + ORDER BY d.nextAttemptAt ASC + """) + List findDue(@Param("now") OffsetDateTime now, Pageable pageable); + + /** DLQ листинг для admin UI. Partial index {@code idx_webhook_deliveries_dlq}. */ + Page findByStatus(WebhookDelivery.Status status, Pageable pageable); + + /** Метрика {@code nsi_webhook_dlq_count}. */ + long countByStatus(WebhookDelivery.Status status); + + /** + * Idempotent insert (sub × event = uniq). На race между dispatcher tick'ами + * вторая вставка попадёт в conflict, ON CONFLICT DO NOTHING. Native query + * чтобы не тащить ON CONFLICT через JPA. + */ + @Modifying + @Query(value = """ + INSERT INTO webhook_deliveries (subscription_id, outbox_event_id, status, + attempt_count, created_at, next_attempt_at) + VALUES (:subscriptionId, :outboxEventId, 'pending', 0, now(), now()) + ON CONFLICT (subscription_id, outbox_event_id) DO NOTHING + """, nativeQuery = true) + int insertIfAbsent( + @Param("subscriptionId") UUID subscriptionId, + @Param("outboxEventId") Long outboxEventId); + + /** История доставок одного event'а — для drilldown в UI. */ + List findByOutboxEventIdOrderByCreatedAtDesc(Long outboxEventId); + + /** История доставок одной подписки. */ + Page findBySubscriptionIdOrderByCreatedAtDesc( + UUID subscriptionId, Pageable pageable); +} diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookSubscription.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookSubscription.java new file mode 100644 index 0000000..c9bef6d --- /dev/null +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookSubscription.java @@ -0,0 +1,116 @@ +package cloud.nstart.terravault.ordinis.domain.webhook; + +import cloud.nstart.terravault.ordinis.domain.DataScope; +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.Table; +import org.hibernate.annotations.JdbcTypeCode; +import org.hibernate.type.SqlTypes; + +import java.time.OffsetDateTime; +import java.util.List; +import java.util.UUID; + +/** + * Webhook subscription — HTTP callback delivery параллельно с Kafka publishing. + * Для consumer'ов которые не могут Kafka subscribe (геопортал PHP, legacy admin-ui'и + * заказчиков, BI-системы). + * + *

Filter arrays: NULL/empty = match all events. Match logic в + * {@code WebhookDispatcher} — event matches subscription iff + * (scope_filter is null OR scope IN scope_filter) AND + * (dictionary_filter is null OR dictionary_name IN dictionary_filter) AND + * (event_type_filter is null OR event_type IN event_type_filter). + * + *

HMAC secret hex-encoded 64 chars (32 bytes random). Generated server-side + * на create, returned ONCE в response, потом hidden — admin-ui показывает + * только маскированный preview {@code sk_****abc123}. + */ +@Entity +@Table(name = "webhook_subscriptions") +public class WebhookSubscription { + + @Id + @Column(name = "id") + private UUID id; + + @Column(name = "name", nullable = false, length = 128) + private String name; + + @Column(name = "url", nullable = false, length = 2048) + private String url; + + /** TEXT[] postgres → List via SqlTypes.ARRAY. */ + @JdbcTypeCode(SqlTypes.ARRAY) + @Column(name = "scope_filter", columnDefinition = "TEXT[]") + private List scopeFilter; + + @JdbcTypeCode(SqlTypes.ARRAY) + @Column(name = "dictionary_filter", columnDefinition = "TEXT[]") + private List dictionaryFilter; + + @JdbcTypeCode(SqlTypes.ARRAY) + @Column(name = "event_type_filter", columnDefinition = "TEXT[]") + private List eventTypeFilter; + + @Column(name = "hmac_secret", nullable = false, length = 128) + private String hmacSecret; + + @Column(name = "active", nullable = false) + private boolean active = true; + + @Column(name = "created_at", nullable = false) + private OffsetDateTime createdAt; + + @Column(name = "created_by", nullable = false, length = 128) + private String createdBy; + + @Column(name = "updated_at", nullable = false) + private OffsetDateTime updatedAt; + + @Column(name = "description", columnDefinition = "TEXT") + private String description; + + protected WebhookSubscription() {} + + public WebhookSubscription( + String name, + String url, + String hmacSecret, + String createdBy) { + this.id = UUID.randomUUID(); + this.name = name; + this.url = url; + this.hmacSecret = hmacSecret; + this.createdBy = createdBy; + OffsetDateTime now = OffsetDateTime.now(); + this.createdAt = now; + this.updatedAt = now; + } + + public void touch() { + this.updatedAt = OffsetDateTime.now(); + } + + public UUID getId() { return id; } + public String getName() { return name; } + public String getUrl() { return url; } + public List getScopeFilter() { return scopeFilter; } + public List getDictionaryFilter() { return dictionaryFilter; } + public List getEventTypeFilter() { return eventTypeFilter; } + public String getHmacSecret() { return hmacSecret; } + public boolean isActive() { return active; } + public OffsetDateTime getCreatedAt() { return createdAt; } + public String getCreatedBy() { return createdBy; } + public OffsetDateTime getUpdatedAt() { return updatedAt; } + public String getDescription() { return description; } + + public void setName(String v) { this.name = v; touch(); } + public void setUrl(String v) { this.url = v; touch(); } + public void setScopeFilter(List v) { this.scopeFilter = v; touch(); } + public void setDictionaryFilter(List v) { this.dictionaryFilter = v; touch(); } + public void setEventTypeFilter(List v) { this.eventTypeFilter = v; touch(); } + public void setActive(boolean v) { this.active = v; touch(); } + public void setDescription(String v) { this.description = v; touch(); } +} diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookSubscriptionRepository.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookSubscriptionRepository.java new file mode 100644 index 0000000..23e736f --- /dev/null +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/webhook/WebhookSubscriptionRepository.java @@ -0,0 +1,20 @@ +package cloud.nstart.terravault.ordinis.domain.webhook; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Query; + +import java.util.List; +import java.util.UUID; + +public interface WebhookSubscriptionRepository extends JpaRepository { + + /** + * Активные подписки. Использует partial index {@code idx_webhook_subscriptions_active} + * (where active = true). Hot path в WebhookDispatcher. + */ + @Query("SELECT w FROM WebhookSubscription w WHERE w.active = true") + List findActive(); + + /** Метрика {@code nsi_webhook_subscriptions_active}. */ + long countByActiveTrue(); +} diff --git a/ordinis-migrations/src/main/resources/db/changelog/changes/0013-webhook-subscriptions.xml b/ordinis-migrations/src/main/resources/db/changelog/changes/0013-webhook-subscriptions.xml new file mode 100644 index 0000000..a4ffaf1 --- /dev/null +++ b/ordinis-migrations/src/main/resources/db/changelog/changes/0013-webhook-subscriptions.xml @@ -0,0 +1,96 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + URL должен начинаться с https:// (production) или http:// (dev). + Scope values constrained тот же enum что DataScope в Java. + + + ALTER TABLE webhook_subscriptions + ADD CONSTRAINT chk_webhook_url_scheme + CHECK (url ~ '^https?://[^\s]+$'); + + ALTER TABLE webhook_subscriptions + ADD CONSTRAINT chk_webhook_scope_filter + CHECK (scope_filter IS NULL OR scope_filter <@ ARRAY['PUBLIC','INTERNAL','RESTRICTED']::TEXT[]); + + + + ALTER TABLE webhook_subscriptions DROP CONSTRAINT IF EXISTS chk_webhook_url_scheme; + ALTER TABLE webhook_subscriptions DROP CONSTRAINT IF EXISTS chk_webhook_scope_filter; + + + + + + + Partial index — dispatcher poller сканит только активные subscriptions + (горячий path). Inactive sit без overhead. + + + CREATE INDEX idx_webhook_subscriptions_active + ON webhook_subscriptions (created_at) + WHERE active = true; + + + DROP INDEX IF EXISTS idx_webhook_subscriptions_active; + + + + diff --git a/ordinis-migrations/src/main/resources/db/changelog/changes/0014-webhook-deliveries.xml b/ordinis-migrations/src/main/resources/db/changelog/changes/0014-webhook-deliveries.xml new file mode 100644 index 0000000..976c88e --- /dev/null +++ b/ordinis-migrations/src/main/resources/db/changelog/changes/0014-webhook-deliveries.xml @@ -0,0 +1,117 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + ALTER TABLE webhook_deliveries + ADD CONSTRAINT chk_webhook_delivery_status + CHECK (status IN ('pending','retrying','delivered','dlq')); + + + + ALTER TABLE webhook_deliveries DROP CONSTRAINT IF EXISTS chk_webhook_delivery_status; + + + + + + + Hot poller path: WHERE status IN ('pending','retrying') AND + next_attempt_at <= now(). Partial index покрывает обе колонки. + + + CREATE INDEX idx_webhook_deliveries_due + ON webhook_deliveries (next_attempt_at) + WHERE status IN ('pending', 'retrying'); + + + DROP INDEX IF EXISTS idx_webhook_deliveries_due; + + + + + Для UI listing DLQ + reset операций. + + CREATE INDEX idx_webhook_deliveries_dlq + ON webhook_deliveries (dlq_at DESC) + WHERE status = 'dlq'; + + + DROP INDEX IF EXISTS idx_webhook_deliveries_dlq; + + + + + + Idempotency: один subscription × event = один row. Если dispatcher + ловит retry где event/subscription уже создавал delivery — INSERT + ON CONFLICT DO NOTHING. + + + CREATE UNIQUE INDEX uniq_webhook_deliveries_subscription_event + ON webhook_deliveries (subscription_id, outbox_event_id); + + + DROP INDEX IF EXISTS uniq_webhook_deliveries_subscription_event; + + + + diff --git a/ordinis-migrations/src/main/resources/db/changelog/master.xml b/ordinis-migrations/src/main/resources/db/changelog/master.xml index 95289a3..9bc6291 100644 --- a/ordinis-migrations/src/main/resources/db/changelog/master.xml +++ b/ordinis-migrations/src/main/resources/db/changelog/master.xml @@ -22,5 +22,7 @@ + +