# События (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`. |

Topology: 1 broker (single-node на cluster.142), 3+ partitions per topic
(historical setup имел 12 partitions, при cluster recreate 2026-05-07
spec задал 3; existing topics annotated `strimzi.io/managed=false`,
broker partitions сохранены), `cleanup.policy: delete`,
`retention.ms: 604800000` (7 дней).

## Подключение

### Адреса bootstrap

| Среда | Bootstrap servers | Auth |
|---|---|---|
| prod | `ordinis-kafka-kafka-bootstrap.ordinis-prod:9092` (TLS+SASL: `:9093`) | SCRAM-SHA-512 |
| staging | `ordinis-kafka-kafka-bootstrap.ordinis-staging:9092` | SCRAM-SHA-512 |

### Получение SASL креденшелов

Запросить у Ordinis team создание Kafka user (Strimzi `KafkaUser` CR):

```yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: altum-consumer
  namespace: ordinis-prod
  labels:
    strimzi.io/cluster: ordinis-kafka
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: ordinis.cuod.events.public
          patternType: literal
        operations: [Read, Describe]
      - resource:
          type: group
          name: altum-consumer-
          patternType: prefix
        operations: [Read]
```

Strimzi user-operator создаст Secret `altum-consumer` с полями `username`,
`password`. Используй их в Kafka client config.

## 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.

## 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 семантика

Outbox pattern гарантирует **at-least-once** delivery:

1. Writer пишет запись в БД + outbox table в одной транзакции.
2. OutboxPoller (scheduled @1s) читает unpublished rows, отправляет в Kafka.
3. После Kafka ACK — отмечает row как published (`published_at` set).

Что это значит для consumer'а:

- Можешь получить **дубликат** события если poller crashed после ACK но до
  `published_at` write. Idempotency у consumer'а — must.
- При network partition между Kafka и poller — сообщения накапливаются в
  outbox (alert `OrdinisOutboxLagHigh` срабатывает > 100 events).

## 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")
```

## DLQ (Dead-letter queue)

Если событие не публикуется > 1000 retry попыток (network issues,
Kafka unavailable) — переходит в DLQ table в Ordinis. Не теряется, но
требует ручного разбора.

Симптом: alert `OrdinisOutboxDLQNonEmpty` → check `/api/v1/admin/audit?action=outbox.dlq`.

## Топик для 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