Свой канал через Kafka

Для команды организации, которая сама ведёт чат с клиентом: свой сайт, приложение, мессенджер. Операторы при этом работают в Shtab Connect. Система организации пишет в Kafka сообщения клиента и «печатает…». Из Kafka она получает ответы оператора и бота, квитанции, состояние обращения и «оператор печатает…».

Управлять обращениями (закрыть, передать, метки) и читать справочники и отчёты — через интеграционный API.

Как устроено

система организации ──► <inbound-топик> ──► Shtab Connect ──► рабочее место оператора
система организации ◄── <outbound-топик> ◄── Shtab Connect ◄── ответ оператора или бота

• Один аккаунт канала — пара топиков, входной и выходной. Значение записи — JSON одного сообщения контракта, вид — поле type. Ключ записи — customer.external_id: всё о клиенте идёт по порядку в одной партиции.
• Клиент — customer.external_id в системе организации. По нему находится карточка клиента (пространство имён custom:<аккаунт>), новому клиенту присваивается имя customer.name.
• Обращение выбирается так же, как в мессенджерах: одно активное на клиента в аккаунте. thread_id — метка нити у организации: ответы приходят с нитью, из которой клиент написал последним. Нужны параллельные обращения одного клиента — заведите отдельный аккаунт канала.
• Доставка — at least once в обе стороны. Повтор сообщения клиента с тем же message_id второго сообщения не создаёт. Повтор записи Shtab Connect несёт тот же event_id: по нему организация отбрасывает дубль.

Контракт

Поля каждого сообщения — на страницах контракта своего канала, собранных из JSON Schema. Те же схемы — в каталоге sdk/contracts/custom_channel/ комплекта поставки, версия схемы — поле schema_version (сейчас 1). SDK содержат те же типы.

Топик	type	Что это
входной	custom.message	Сообщение клиента: message_id, customer, thread_id, text, attachments[] (до 10, файлы скачиваются по url и проверяются антивирусом), sent_at
входной	custom.typing	Клиент набирает текст; draft — что набрано (необязательно)
выходной	custom.receipt	Итог приёма message_id: accepted с conversation_id и helpdesk_message_id или rejected с reason (invalid, empty)
выходной	custom.message	Фрагмент ответа оператора, бота или системы: текст, файл (attachment_id) или кнопки quick_replies. Сообщение из нескольких частей приходит несколькими фрагментами с одним message_id
выходной	custom.conversation_state	Состояние обращения: waiting, at_bot, assigned (с operator_name), on_hold, resolved, closed; номер HD-<n>
выходной	custom.typing	Оператор набирает ответ

• Черновик. draft видит только назначенный оператор и только если в организации включён показ набора («Показывать оператору набираемый текст»). Черновик не хранится и не пишется в журналы.
• Частота. «Печатает…» принимается не чаще раза в 3 секунды на клиента и нить, лишние сигналы отбрасываются. SDK троттлят сами.
• Файлы ответа. Скачиваются GET /api/integration/v1/attachments/{attachment_id} сервисным аккаунтом (интеграционный API). Прямых ссылок на хранилище нет.
• Нажатие быстрого ответа — обычное сообщение клиента с текстом кнопки.
• Прочтение. О доставке и прочтении организация не сообщает: ответ клиента считается прочтением.
• Канал выключен — входной топик не читается, сообщения ждут в нём.
• custom.conversation_state приходит только при смене состояния, видимого организации: у нового обращения — одно waiting.

Брокер: установки или организации

	Брокер установки (own)	Брокер организации (remote)
Где топики	<prefix>.custom.<ref>.inbound.v1 и ….outbound.v1 на Kafka установки	топики организации, их имена указываются при подключении
Учётка	helpdesk-custom-<ref> (SCRAM-SHA-512), пароль задаёт администратор при подключении	учётка Shtab Connect на брокере организации
Права	только запись во входной, чтение выходного, группы консьюмеров с префиксом custom.<ref>.	Shtab Connect нужны: READ входного топика и своей группы, WRITE выходного, DESCRIBE обоих
Сеть	система организации ходит на внешний адрес брокера установки	установка ходит на брокер организации: адрес — в «Исходящих подключениях», только TLS

Режим после подключения не меняется: для другого брокера заведите новый канал. Брокер — Kafka 3.0 или новее.

Подключение

Администратор организации (право channels.manage): Настройки → Каналы → Подключить → Свой канал. Канал входит в модуль лицензии channel.custom: без него подключение отклоняется с FEATURE_NOT_LICENSED.

1. Название канала, очередь, режим брокера.
2. own: пароль учётки, не короче 16 символов. Кнопка генерирует его в браузере. Пароль виден только при вводе, дальше — только последние символы. Передайте его команде организации защищённым каналом.
3. remote: адреса брокера (host:port через запятую), протокол (SSL или SASL_SSL), механизм SASL, учётка и пароль, имена входного и выходного топиков, группа консьюмеров. При необходимости — PEM УЦ брокера и клиентский сертификат с ключом (mTLS). Хост брокера должен быть в Установка → Исходящие подключения: его добавляет администратор установки, иначе подключение отклоняется. Порт по умолчанию — 9092: при другом порте запись — host:port (kafka.acme.ru:9093).
4. После сохранения карточка канала показывает адрес брокера, топики, учётку и состояние: «Готовится» (учётка и топики заводятся, до 10 секунд), «Готово» или «Ошибка» с кодом. «Проверить подключение» проверяет брокер и топики.

Для инженера внедрения

• Процессы. custom-channel-ingest читает входные топики, по одной реплике (compose и Helm включают её по умолчанию). kafka-custom-channels заводит учётки, топики и ACL на брокере установки. Он работает под helpdesk-admin и нужен только для режима own. С внешней Kafka без прав администратора выключите его (workers.kafka-custom-channels.enabled: false) и заводите учётки вручную по таблице выше.
• Внешний адрес брокера (own). Значение config.kafkaExternalBootstrapServers в Helm или KAFKA_EXTERNAL_BOOTSTRAP_SERVERS в compose — адрес, который увидит система организации, протокол — config.kafkaExternalSecurityProtocol (SASL_SSL по умолчанию). Пусто — экран не предлагает брокер установки. Вне доверенной сети открывайте только листенер SASL_SSL с сертификатом.
• Права процессов на префикс <prefix>.custom. (READ, WRITE, DESCRIBE) и группы helpdesk.custom-ingest.* выдаёт bootstrap (kafka_ensure_acls). Для внешней Kafka возьмите их из manage.py kafka_ensure_acls --print.
• Изоляция. Учётка одного аккаунта не читает и не пишет топики другого: ACL только на свою пару. Проверка: kafka-console-consumer под учёткой аккаунта A на выходной топик аккаунта B даёт TopicAuthorizationException.
• Журналы содержат только id, коды, оффсеты и имена топиков. Тексты сообщений, черновики и пароли в них не попадают.

SDK

Каталог sdk/ комплекта поставки: Kafka-SDK на Python (shtab_helpdesk_kafka), Go и JS/TS. Отправка сообщения и «печатает…», подписка на выходной топик с разбором по контракту, троттлинг набора, идемпотентный продюсер (acks=all). Установка и примеры — в README пакета.

Проверка

1. Подключите свой канал в режиме own и дождитесь «Готово».
2. Пример из README Kafka-SDK с адресом брокера, учёткой helpdesk-custom-<ref> и паролем отправляет сообщение и «печатает…» с черновиком.
3. В рабочем месте оператора появляются обращение, «печатает…» и черновик (при включённом показе набора).
4. Оператор берёт обращение и отвечает. Пример получает custom.receipt (accepted), custom.conversation_state (assigned), custom.typing и custom.message. После завершения приходит custom.conversation_state (resolved, «Решено»), а клиенту — системное сообщение о решении; closed приходит, когда обращение закрыто окончательно.