> Как отправляются события об изменении статусов? (Go)

Короткий ответ

События об изменении статусов отправляются через механизм событийного взаимодействия: сервис публикует событие в брокер сообщений (Kafka, NATS, RabbitMQ) или через внутреннюю шину, а подписчики обрабатывают его асинхронно. В Go это реализуется через паттерн outbox + publisher, чтобы гарантировать атомарность записи в БД и публикации. Альтернатива - синхронные webhook-вызовы или gRPC-stream, но они менее надёжны и масштабируемы.

Подробное объяснение

Основная проблема при отправке событий - согласованность между изменением статуса в базе данных и публикацией события. Если сначала обновить БД, а потом отправить событие, возможна потеря при сбое. Если наоборот - подписчики увидят событие до фактического изменения.

Стандартное решение - transactional outbox: в той же транзакции, где меняется статус, пишем запись в таблицу outbox. Отдельный процесс (relay) читает outbox и публикует события в брокер. Это даёт at-least-once доставку, поэтому подписчики должны быть идемпотентными.

Второй вариант - CDC (Change Data Capture) через Debezium: слушаем изменения в WAL PostgreSQL и транслируем их в Kafka. Это удобно, но добавляет инфраструктурную сложность.

Для внутренних событий в рамках одного процесса можно использовать простой pub/sub на каналах Go, но это не масштабируется на несколько инстансов.

Формат события: обычно JSON или protobuf, с полями event_id, entity_id, old_status, new_status, timestamp, version. Версионирование обязательно - подписчики могут отставать.

На практике

В Go-сервисах чаще всего используют:

• Kafka через segmentio/kafka-go или confluent-kafka-go
• NATS JetStream для лёгких сценариев
• outbox-реализации: watermill (с поддержкой outbox), go-outbox, или собственная реализация на pgx + pgxpool

Ретри-логика: при ошибке публикации relay должен делать backoff и не терять события. Подписчики обрабатывают события идемпотентно - по event_id или по ключу entity_id + new_status.

Важно: не отправлять события в обход outbox, даже если кажется, что "сейчас просто". Это приводит к рассинхронизации.

Пример кода

GO
// outbox record
type OutboxEvent struct {
    ID        int64
    EventID   string
    EntityID  string
    Status    string
    Payload   []byte
    CreatedAt time.Time
}

// publish in transaction
func ChangeStatus(ctx context.Context, db *pgxpool.Pool, entityID string, newStatus string) error {
    tx, err := db.Begin(ctx)
    if err != nil {
        return err
    }
    defer tx.Rollback(ctx)

    // update status
    _, err = tx.Exec(ctx, `UPDATE entities SET status = $1 WHERE id = $2`, newStatus, entityID)
    if err != nil {
        return err
    }

    // write outbox
    _, err = tx.Exec(ctx, `
        INSERT INTO outbox (event_id, entity_id, status, payload, created_at)
        VALUES ($1, $2, $3, $4, now())`,
        uuid.New().String(), entityID, newStatus, []byte(`{"source":"service-a"}`))
    if err != nil {
        return err
    }

    return tx.Commit(ctx)
}

// relay loop
func Relay(ctx context.Context, db *pgxpool.Pool, producer *kafka.Writer) {
    for {
        rows, err := db.Query(ctx, `
            SELECT id, event_id, entity_id, status, payload
            FROM outbox
            WHERE published_at IS NULL
            ORDER BY id
            LIMIT 100`)
        if err != nil {
            log.Printf("query outbox: %v", err)
            time.Sleep(time.Second)
            continue
        }

        var events []OutboxEvent
        for rows.Next() {
            var e OutboxEvent
            rows.Scan(&e.ID, &e.EventID, &e.EntityID, &e.Status, &e.Payload)
            events = append(events, e)
        }
        rows.Close()

        for _, e := range events {
            msg := kafka.Message{
                Key:   []byte(e.EntityID),
                Value: e.Payload,
            }
            if err := producer.WriteMessages(ctx, msg); err != nil {
                log.Printf("publish failed: %v", err)
                break
            }
            db.Exec(ctx, `UPDATE outbox SET published_at = now() WHERE id = $1`, e.ID)
        }

        time.Sleep(500 * time.Millisecond)
    }
}

Как отвечать на собеседовании

Начни с главного - почему нельзя просто отправить событие после изменения БД. Покажи понимание проблемы согласованности. Затем опиши outbox как базовое решение, упомяни альтернативы (CDC, синхронные вызовы) и их trade-off.

Подчеркни практические детали: идемпотентность подписчиков, ретраи, версионирование событий, порядок доставки (если важен). Если спросят про конкретный брокер - сравни Kafka и NATS по задержке и гарантиям.

Не углубляйся в детали реализации конкретной библиотеки, если не просят. Лучше показать архитектурное мышление.

Что проверяет интервьюер

• Понимание распределённых систем и проблемы согласованности
• Знание паттернов интеграции (outbox, CDC, event sourcing)
• Практический опыт с брокерами сообщений
• Умение проектировать надёжные асинхронные потоки
• Понимание идемпотентности и обработки ошибок

Типичные ошибки

• Предложение отправлять событие сразу после UPDATE без outbox - сразу показывает непонимание гарантий
• Игнорирование идемпотентности подписчиков
• Утверждение, что Kafka гарантирует exactly-once - это не так в общем случае
• Смешивание синхронного и асинхронного взаимодействия без чёткого обоснования
• Отсутствие упоминания версионирования событий
• Предложение использовать sync.Mutex или каналы для межсервисных событий - это работает только в одном процессе