Александр Зимин
193 lines
8.2 KiB
Markdown
193 lines
8.2 KiB
Markdown
# События (Kafka outbox)
|
||
|
||
Ordinis публикует все изменения справочников в Kafka через
|
||
**outbox pattern** (at-least-once гарантия). Consumers подписываются на
|
||
топики и инвалидируют свой кэш / обновляют projection.
|
||
|
||
## Топики
|
||
|
||
| Топик | Scope | Назначение |
|
||
|---|---|---|
|
||
| `ordinis.cuod.events.public` | PUBLIC | Изменения публичных справочников. |
|
||
| `ordinis.cuod.events.internal` | INTERNAL | INTERNAL словари — нужна role `ordinis-internal`+. |
|
||
| `ordinis.cuod.events.restricted` | RESTRICTED | RESTRICTED — role `ordinis-restricted`. |
|
||
|
||
Параметры топиков (для consumer config):
|
||
|
||
- `cleanup.policy: delete`
|
||
- `retention.ms: 604800000` (7 дней)
|
||
- Multi-partition (точное число — спросить Ordinis team при registration,
|
||
partition count может меняться по operational reasons).
|
||
|
||
## Подключение
|
||
|
||
### Bootstrap address + creds
|
||
|
||
Запросить у Ordinis team:
|
||
- Bootstrap servers (внутренний DNS на per-environment basis)
|
||
- SASL credentials (SCRAM-SHA-512 username + password)
|
||
- Topic ACL (Read + Describe на нужные топики)
|
||
- Consumer group prefix (для group ACL)
|
||
|
||
Auth: **SASL/SCRAM-SHA-512**. TLS опциональный (отдельный listener).
|
||
|
||
Креденшелы выдаются per-consumer service. Не share между сервисами —
|
||
ACL и audit за каждым.
|
||
|
||
## Schema события
|
||
|
||
### `dictionary.changed`
|
||
|
||
Все события имеют общий envelope:
|
||
|
||
```json
|
||
{
|
||
"schemaVersion": "1.0.0",
|
||
"eventType": "dictionary.changed",
|
||
"occurredAt": "2026-05-07T15:23:45.123+03:00",
|
||
"txTime": "2026-05-07T15:23:45.100+03:00",
|
||
"changedBy": "admin@nstart.space",
|
||
"changeReason": "Обновление after launch",
|
||
"scope": "PUBLIC",
|
||
"dictionary": "spacecraft",
|
||
"businessKey": "RESURS-P-3",
|
||
"action": "updated",
|
||
"version": 3,
|
||
"validFrom": "2026-05-07T00:00:00+03:00",
|
||
"validTo": "9999-12-31T23:59:59+03:00",
|
||
"data": {
|
||
"name": { "ru-RU": "Ресурс-П № 3 (обновлён)" },
|
||
"status": "active",
|
||
"satelliteTypeCode": "EARTH-OBSERVATION-OPTICAL"
|
||
},
|
||
"previousData": {
|
||
"name": { "ru-RU": "Ресурс-П № 3" },
|
||
"status": "in-orbit"
|
||
}
|
||
}
|
||
```
|
||
|
||
### Поля envelope
|
||
|
||
| Поле | Описание |
|
||
|---|---|
|
||
| `schemaVersion` | Semver схемы события. Bump major = breaking change. |
|
||
| `eventType` | `dictionary.changed` для CRUD, `dictionary.bulk_imported` для bundle imports. |
|
||
| `occurredAt` | Wall-clock когда событие создано. |
|
||
| `txTime` | Transaction time изменения в БД (для ordering). |
|
||
| `action` | `created` / `updated` / `closed`. |
|
||
| `version` | Optimistic lock version. Monotonic per businessKey. |
|
||
| `data` | Payload новой версии. |
|
||
| `previousData` | Diff-like — поля, которые поменялись (только для `updated`). Полезен для consumer'ов которые хотят минимальный set re-fetches. |
|
||
| `validFrom` / `validTo` | Bitemporal границы. |
|
||
|
||
### `action` semantics
|
||
|
||
- **created** — новая запись с уникальным `businessKey`. `previousData = null`.
|
||
- **updated** — существующий businessKey, новая версия. `previousData` =
|
||
изменённые поля старой версии.
|
||
- **closed** — `validTo` обновлён в `now()` (раньше `9999-...`). Запись больше
|
||
не возвращается на `?at=now`. `previousData` = вся старая версия.
|
||
|
||
### Cascade events (v2 roadmap)
|
||
|
||
При cascade close (см. design doc `dict-relationships-v2` — внутренний
|
||
artifact в `~/.gstack/projects/claude/`) на каждую закрытую запись
|
||
отдельное событие emit'ится в outbox с дополнительным полем
|
||
`cascadeSource` = `{dictionary, businessKey}` источника. Consumer может
|
||
trace cascade chain.
|
||
|
||
### Definition events
|
||
|
||
Изменения **самой schema** dictionary (не records) emit'ятся отдельными
|
||
event types:
|
||
|
||
| `eventType` | Когда |
|
||
|---|---|
|
||
| `DefinitionCreated` | Новый dictionary зарегистрирован |
|
||
| `DefinitionUpdated` | Inline PATCH schema (см. `/dictionaries/{name}/schema`) |
|
||
|
||
Payload содержит `schemaJson`, `schemaVersion`, `previousSchema` (для
|
||
updated). Consumer'ы которые кэшируют schemas инвалидируют по этим
|
||
событиям.
|
||
|
||
### Schema workflow events (v2.2.0+)
|
||
|
||
Workflow draft lifecycle emit'ит дополнительные events. Все содержат
|
||
`draftId`, `dictionaryName`, `status`, `branchedFromVersion`, `makerId`
|
||
плюс event-specific extras:
|
||
|
||
| `eventType` | Trigger | Extras |
|
||
|---|---|---|
|
||
| `SchemaDraftCreated` | maker создаёт draft | `fromVersion` |
|
||
| `SchemaDraftReviewRequested` | submit на review | `reviewers[]`, `note` |
|
||
| `SchemaDraftDecided` | reviewer вынес verdict | `decision` (approve/request_changes/reject), `reviewerId`, `comment` |
|
||
| `SchemaDraftPublished` | publisher выпустил draft → live | `newVersion`, `publishedAt`, `publishNote` |
|
||
| `SchemaDraftWithdrawn` | maker отозвал draft | (без extras) |
|
||
|
||
Webhook router маппит эти типы в `dict.draft.*` category (см.
|
||
[webhooks.md](webhooks.md)). Полная state machine: [schema-workflow.md](schema-workflow.md).
|
||
|
||
## Partition key
|
||
|
||
Сообщения партиционируются по `dictionary + businessKey` (одна запись —
|
||
одна partition). Это даёт consumer'ам ordering guarantee per record.
|
||
|
||
```
|
||
key = "spacecraft:RESURS-P-3"
|
||
```
|
||
|
||
{% note info %}
|
||
|
||
Если consumer обрабатывает события параллельно (multiple consumer threads /
|
||
pods), Kafka гарантирует только partition-level ordering. Ordering между
|
||
records в разных partitions не определён. Для consumer'ов это норма —
|
||
each businessKey изолирован.
|
||
|
||
{% endnote %}
|
||
|
||
## At-least-once семантика
|
||
|
||
Ordinis использует **outbox pattern** — изменение БД и публикация в Kafka
|
||
атомарны через transactional outbox. Гарантия: **at-least-once**.
|
||
|
||
Что это значит для consumer'а:
|
||
|
||
- Можешь получить **дубликат** события (rare, при partial failures на
|
||
стороне Ordinis). Idempotency у consumer'а — must.
|
||
- При недоступности Kafka события накапливаются в Ordinis side и
|
||
доставляются после восстановления — **events не теряются**.
|
||
- Order — гарантирован per partition key (`dictionary:businessKey`),
|
||
не глобально.
|
||
|
||
## Idempotency
|
||
|
||
Используй `(dictionary, businessKey, txTime, version)` как dedup key:
|
||
|
||
```python
|
||
seen = redis.get(f"ordinis:event:{dict}:{key}:{tx_time}:{ver}")
|
||
if not seen:
|
||
process_event(event)
|
||
redis.setex(f"ordinis:event:...", 86400, "1")
|
||
```
|
||
|
||
## Persistent failures (Ordinis side)
|
||
|
||
Если событие не публикуется > 1000 retry попыток (multi-day Kafka outage),
|
||
оно складывается в DLQ внутри Ordinis. Не теряется, но требует ручного
|
||
разбора Ordinis team. Consumer не видит таких событий до их
|
||
re-publishing — но это очень редкий сценарий.
|
||
|
||
## Топик для discovery
|
||
|
||
В roadmap планируется `nsi.dictionary.changed` aggregator topic — single
|
||
firehose для consumers которые хотят видеть все scope'ы в одном потоке
|
||
(c фильтрацией consumer-side). Сейчас **нет**, подписывайся на 3 scope
|
||
топика отдельно.
|
||
|
||
## Дальше
|
||
|
||
- [Webhooks](webhooks.md) — alternative для consumer'ов без Kafka access
|
||
- [Bitemporal](bitemporal.md) — как интерпретировать `validFrom`/`validTo`
|
||
- [Caching](caching.md) — invalidation pattern с Kafka events
|