fix(flow): grupo de consumidor próprio na fila de gatilhos de jornada do ClickHouse (CRM-777)

[nickoliveira23]

Summary

A tabela Kafka journey_trigger_kafka_queue do ClickHouse nomeava o consumer group temporal-workers, o mesmo dos workers que consomem os gatilhos de jornada.

• A fila de gatilhos passa a usar um grupo próprio: <KAFKA_GROUP_ID>-journey-triggers-clickhouse.
  • O ClickHouse só produz nesse tópico, porque a tabela é destino da events_to_journey_triggers_mv. Então hoje ele não entra no grupo, e isso foi conferido em system.kafka_consumers.
  • Mas, se um consumidor fosse iniciado nessa tabela, ele dividiria as partições com os workers e eles perderiam gatilhos. A mudança é defensiva.
• A guarda de boot (ensureKafkaEngineBroker) passa a comparar também o grupo. Quando ele diverge, a guarda derruba a MV e a tabela para recriá-las, como já fazia com o broker. Sem isso, a troca não chegaria às tabelas que já existem.
  • O parâmetro expectedGroup é opcional.
  • As duas filas o passam: contact_events passa o grupo que já tinha, sem efeito.
• O log de "tabela já correta" passa a citar o grupo.

Security

• Nenhuma mudança de autenticação, permissão ou dado de cliente.
• No primeiro boot depois do deploy, a fila de gatilhos é recriada uma vez. Ela não tem offset para perder, porque só produz. Eventos inseridos no contact_events nesse intervalo de milissegundos não viram gatilho.

Test plan

• npx jest src/modules/processing/clickhouse/ → 18/18. Os casos novos:
  • extrair o grupo do DDL, inclusive com o DDL em várias linhas;
  • grupo divergente → recria;
  • broker e grupo iguais → mantém;
  • wiring da fila de gatilhos com o grupo próprio, sem temporal-workers.
• npx jest --ci --maxWorkers=2 → 1134 passed, 0 falhas. npm run typecheck e npm run build ok.
• Mutações (todas vermelhas):
  • o grupo volta a temporal-workers;
  • a guarda ignora o grupo;
  • a regex do grupo captura o tópico;
  • contact_events sem passar o grupo;
  • um grupo divergente só gera aviso e não recria.
• Ao vivo no ambiente de dev:
  • 1º boot: o log uses consumer group 'temporal-workers' … Recreating it aparece, e a tabela é recriada com o grupo novo.
  • 2º boot: already points …, sem recriação.
  • Um evento de sonda inserido no contact_events chegou ao tópico journey-triggers; o offset passou de 390 para 391.

Changed Files

• src/modules/processing/clickhouse/clickhouse.service.ts
• src/modules/processing/clickhouse/clickhouse.service.spec.ts
• src/modules/processing/clickhouse/clickhouse.service.contact-events-broker.spec.ts

Related PRs

• evolution-foundation/evolution-ecosystem#143: o ajuste de configuração que faz as tabelas apontarem para o broker certo no ambiente de produção. Independente deste.

Linked Issue

• CRM-777

🤖 Generated with Claude Code

[sourcery-ai]

Reviewer's Guide

A implementação separa defensivamente o consumer group da tabela ClickHouse de gatilhos de jornada dos workers temporais e estende a guarda de boot para detectar divergências de grupo no DDL existente, recriando a MV e a tabela quando necessário; os testes cobrem parsing, recriação e wiring das filas.

Sequence diagram for ClickHouse Kafka queue boot guard

sequenceDiagram
    participant ClickHouseService
    participant ClickHouse
    participant Kafka

    ClickHouseService->>ClickHouse: ensureKafkaEngineBroker(expectedBrokers, expectedGroup)
    ClickHouse-->>ClickHouseService: Existing Kafka Engine DDL
    ClickHouseService->>ClickHouseService: extractKafkaBrokers()
    ClickHouseService->>ClickHouseService: extractKafkaGroup()
    alt broker or consumer group differs
        ClickHouseService->>ClickHouse: DROP VIEW IF EXISTS events_to_journey_triggers_mv
        ClickHouseService->>ClickHouse: DROP TABLE IF EXISTS journey_trigger_kafka_queue
        ClickHouseService->>ClickHouse: CREATE TABLE journey_trigger_kafka_queue
        ClickHouse->>Kafka: Produce to journey-triggers
    else broker and group match
        ClickHouseService-->>ClickHouse: Keep existing table
    end

Loading

File-Level Changes

Change	Details	Files
A fila Kafka de gatilhos de jornada passa a usar um consumer group exclusivo do ClickHouse.	Deriva o grupo como <KAFKA_GROUP_ID>-journey-triggers-clickhouse, evitando temporal-workers. Passa o grupo esperado ao provisionamento da tabela e mantém o mesmo valor no DDL Kafka. Preserva o grupo existente de contact_events ao integrar a nova validação.	src/modules/processing/clickhouse/clickhouse.service.ts
A guarda de inicialização detecta divergências de consumer group e força a recriação dos objetos Kafka dependentes.	Adiciona expectedGroup opcional à guarda, que compara o grupo extraído do DDL além do broker. Recria a MV dependente e a tabela quando broker ou grupo divergem; mantém objetos quando ambos coincidem. Extrai o terceiro argumento do engine Kafka, incluindo DDL formatado em múltiplas linhas, e atualiza os logs de estado correto e recriação.	src/modules/processing/clickhouse/clickhouse.service.ts
A cobertura de testes valida extração, detecção de configuração obsoleta e wiring das duas filas.	Testa extração de grupo em DDL normal e multilinha. Verifica recriação quando apenas o grupo diverge e preservação quando broker e grupo coincidem. Confirma que a fila de gatilhos usa o grupo exclusivo e nunca temporal-workers. Atualiza o contrato de teste da guarda para aceitar expectedGroup na fila de eventos de contato.	src/modules/processing/clickhouse/clickhouse.service.spec.ts src/modules/processing/clickhouse/clickhouse.service.contact-events-broker.spec.ts

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've reviewed your changes and they look great!

Sourcery assessment

Needs a human reviewer. The change alters the Kafka consumer group used by the ClickHouse engine table, which can change partition ownership and cause journey-trigger messages to be consumed by the wrong component or skipped. Reverting restores the configuration, but messages already consumed or offsets already advanced may not be recoverable by the revert alone.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook