From fea656ff61f6b95e1a8bb2dd86a1ad10a9ed18c0 Mon Sep 17 00:00:00 2001 From: ivkond Date: Fri, 10 Apr 2026 21:10:38 +0300 Subject: [PATCH] feat(group): expand topic extractor coverage and filter go test noise --- .../core/group/extractors/topic-extractor.ts | 86 ++++++++++ .../test/unit/group/topic-extractor.test.ts | 149 +++++++++++++++++- 2 files changed, 233 insertions(+), 2 deletions(-) diff --git a/gitnexus/src/core/group/extractors/topic-extractor.ts b/gitnexus/src/core/group/extractors/topic-extractor.ts index c27b419bb..d09129e2b 100644 --- a/gitnexus/src/core/group/extractors/topic-extractor.ts +++ b/gitnexus/src/core/group/extractors/topic-extractor.ts @@ -6,6 +6,9 @@ import type { ExtractedContract, RepoHandle } from '../types.js'; type Broker = 'kafka' | 'rabbitmq' | 'nats'; +const KAFKAJS_CONSUMER_RUN_RE = /consumer\.run\s*\(\s*\{\s*eachMessage:/; +const KAFKAJS_SUBSCRIBE_RE = /consumer\.subscribe\s*\(\s*\{\s*topic:\s*['"]([^'"]+)['"]/g; + function readSafe(repoPath: string, rel: string): string | null { const abs = path.resolve(repoPath, rel); const base = path.resolve(repoPath); @@ -116,6 +119,42 @@ const KAFKA_PATTERNS: PatternDef[] = [ topicGroup: 1, symbolName: 'producer.send', }, + // Go: sarama.NewSyncProducer(...); producer.SendMessage(&sarama.ProducerMessage{Topic: "xxx"}) + { + regex: /sarama\.NewSyncProducer[\s\S]{0,300}?Topic:\s*"([^"]+)"/g, + role: 'provider', + broker: 'kafka', + confidence: 0.75, + topicGroup: 1, + symbolName: 'sarama.ProducerMessage', + }, + // Go: sarama.NewAsyncProducer(...); producer.Input() <- &sarama.ProducerMessage{Topic: "xxx"} + { + regex: /sarama\.NewAsyncProducer[\s\S]{0,300}?Topic:\s*"([^"]+)"/g, + role: 'provider', + broker: 'kafka', + confidence: 0.75, + topicGroup: 1, + symbolName: 'sarama.ProducerMessage', + }, + // Go: kafka.Writer{Topic: "xxx"} or kafka.NewWriter(...Topic: "xxx") + { + regex: /kafka\.(?:NewWriter|Writer)\b[\s\S]{0,200}?Topic:\s*"([^"]+)"/g, + role: 'provider', + broker: 'kafka', + confidence: 0.75, + topicGroup: 1, + symbolName: 'kafka.Writer', + }, + // Go: kafka.NewReader(...Topic: "xxx") or kafka.Reader{Topic: "xxx"} + { + regex: /kafka\.(?:NewReader|Reader)\b[\s\S]{0,200}?Topic:\s*"([^"]+)"/g, + role: 'consumer', + broker: 'kafka', + confidence: 0.75, + topicGroup: 1, + symbolName: 'kafka.Reader', + }, ]; // --- RabbitMQ patterns --- @@ -205,6 +244,42 @@ const NATS_PATTERNS: PatternDef[] = [ topicGroup: 1, symbolName: 'nc.Publish', }, + // Go/Node JetStream: js.Subscribe("xxx" + { + regex: /js\.(?:S|s)ubscribe\s*\(\s*"([^"]+)"/g, + role: 'consumer', + broker: 'nats', + confidence: 0.8, + topicGroup: 1, + symbolName: 'js.Subscribe', + }, + // Go/Node JetStream: js.Publish("xxx" + { + regex: /js\.(?:P|p)ublish\s*\(\s*"([^"]+)"/g, + role: 'provider', + broker: 'nats', + confidence: 0.8, + topicGroup: 1, + symbolName: 'js.Publish', + }, + // Python: await nc.subscribe("xxx") + { + regex: /await\s+nc\.subscribe\s*\(\s*['"]([^'"]+)['"]/g, + role: 'consumer', + broker: 'nats', + confidence: 0.75, + topicGroup: 1, + symbolName: 'nc.subscribe', + }, + // Python: await nc.publish("xxx") + { + regex: /await\s+nc\.publish\s*\(\s*['"]([^'"]+)['"]/g, + role: 'provider', + broker: 'nats', + confidence: 0.75, + topicGroup: 1, + symbolName: 'nc.publish', + }, ]; const ALL_PATTERNS: PatternDef[] = [...KAFKA_PATTERNS, ...RABBITMQ_PATTERNS, ...NATS_PATTERNS]; @@ -229,6 +304,7 @@ export class TopicExtractor implements ContractExtractor { const out: ExtractedContract[] = []; for (const rel of files) { + if (rel.endsWith('_test.go')) continue; const content = readSafe(repoPath, rel); if (!content) continue; out.push(...this.scanFile(content, rel)); @@ -260,6 +336,16 @@ export class TopicExtractor implements ContractExtractor { } } + if (KAFKAJS_CONSUMER_RUN_RE.test(content)) { + const subscribeRe = new RegExp(KAFKAJS_SUBSCRIBE_RE.source, KAFKAJS_SUBSCRIBE_RE.flags); + let subscribeMatch: RegExpExecArray | null; + while ((subscribeMatch = subscribeRe.exec(content)) !== null) { + const topicName = subscribeMatch[1]; + if (!topicName) continue; + out.push(makeContract(topicName, 'consumer', filePath, 'consumer.run', 0.75, 'kafka')); + } + } + return out; } diff --git a/gitnexus/test/unit/group/topic-extractor.test.ts b/gitnexus/test/unit/group/topic-extractor.test.ts index c6a1161a0..38d8a11f9 100644 --- a/gitnexus/test/unit/group/topic-extractor.test.ts +++ b/gitnexus/test/unit/group/topic-extractor.test.ts @@ -75,8 +75,7 @@ public void handleUserCreated(ConsumerRecord record) { it('test_extract_kafkajs_subscribe_returns_consumer', async () => { writeFile( 'src/consumer.ts', - `await consumer.subscribe({ topic: 'order.placed', fromBeginning: true }); -await consumer.run({ eachMessage: async ({ message }) => {} });`, + `await consumer.subscribe({ topic: 'order.placed', fromBeginning: true });`, ); const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); @@ -101,6 +100,23 @@ await consumer.run({ eachMessage: async ({ message }) => {} });`, }); }); + describe('KafkaJS consumer run', () => { + it('test_extract_kafkajs_consumer_run_eachmessage_returns_consumer', async () => { + writeFile( + 'src/consumer.ts', + `await consumer.subscribe({ topic: 'user.logged-in' }); +await consumer.run({ eachMessage: async () => {} });`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const consumers = contracts.filter((c) => c.role === 'consumer'); + + expect(consumers).toHaveLength(1); + expect(consumers[0].contractId).toBe('topic::user.logged-in'); + expect(consumers[0].meta.broker).toBe('kafka'); + }); + }); + describe('RabbitMQ — Java', () => { it('test_extract_rabbit_listener_returns_consumer', async () => { writeFile( @@ -174,6 +190,62 @@ public void processOrder(OrderMessage msg) {}`, }); }); + describe('JetStream', () => { + it('test_extract_jetstream_publish_returns_provider', async () => { + writeFile('src/stream.go', `js.Publish("orders.created", payload)`); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const producers = contracts.filter((c) => c.role === 'provider'); + + expect(producers).toHaveLength(1); + expect(producers[0].contractId).toBe('topic::orders.created'); + expect(producers[0].meta.broker).toBe('nats'); + }); + + it('test_extract_jetstream_subscribe_returns_consumer', async () => { + writeFile('src/stream.go', `js.Subscribe("orders.created", handler)`); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const consumers = contracts.filter((c) => c.role === 'consumer'); + + expect(consumers).toHaveLength(1); + expect(consumers[0].contractId).toBe('topic::orders.created'); + expect(consumers[0].meta.broker).toBe('nats'); + }); + }); + + describe('Python NATS', () => { + it('test_extract_python_nats_subscribe_returns_consumer', async () => { + writeFile( + 'src/subscriber.py', + `nc = await nats.connect() +await nc.subscribe("orders.created", cb=handler)`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const consumers = contracts.filter((c) => c.role === 'consumer'); + + expect(consumers).toHaveLength(1); + expect(consumers[0].contractId).toBe('topic::orders.created'); + expect(consumers[0].meta.broker).toBe('nats'); + }); + + it('test_extract_python_nats_publish_returns_provider', async () => { + writeFile( + 'src/publisher.py', + `nc = await nats.connect() +await nc.publish("orders.created", payload)`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const producers = contracts.filter((c) => c.role === 'provider'); + + expect(producers).toHaveLength(1); + expect(producers[0].contractId).toBe('topic::orders.created'); + expect(producers[0].meta.broker).toBe('nats'); + }); + }); + describe('NATS', () => { it('test_extract_nats_subscribe_go_returns_consumer', async () => { writeFile( @@ -248,6 +320,68 @@ partConsumer, _ := consumer.ConsumePartition("inventory.update", 0, sarama.Offse expect(consumers[0].contractId).toBe('topic::inventory.update'); expect(consumers[0].meta.broker).toBe('kafka'); }); + + it('test_extract_sarama_sync_producer_returns_provider', async () => { + writeFile( + 'internal/publisher.go', + `package publisher +producer, _ := sarama.NewSyncProducer(brokers, cfg) +producer.SendMessage(&sarama.ProducerMessage{Topic: "inventory.update"})`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const producers = contracts.filter((c) => c.role === 'provider'); + + expect(producers).toHaveLength(1); + expect(producers[0].contractId).toBe('topic::inventory.update'); + expect(producers[0].meta.broker).toBe('kafka'); + }); + + it('test_extract_sarama_async_producer_returns_provider', async () => { + writeFile( + 'internal/publisher.go', + `package publisher +producer, _ := sarama.NewAsyncProducer(brokers, cfg) +producer.Input() <- &sarama.ProducerMessage{Topic: "inventory.update"}`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const producers = contracts.filter((c) => c.role === 'provider'); + + expect(producers).toHaveLength(1); + expect(producers[0].contractId).toBe('topic::inventory.update'); + expect(producers[0].meta.broker).toBe('kafka'); + }); + + it('test_extract_kafka_go_writer_returns_provider', async () => { + writeFile( + 'internal/writer.go', + `package publisher +writer := &kafka.Writer{Topic: "inventory.update"}`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const producers = contracts.filter((c) => c.role === 'provider'); + + expect(producers).toHaveLength(1); + expect(producers[0].contractId).toBe('topic::inventory.update'); + expect(producers[0].meta.broker).toBe('kafka'); + }); + + it('test_extract_kafka_go_reader_returns_consumer', async () => { + writeFile( + 'internal/reader.go', + `package consumer +reader := kafka.NewReader(kafka.ReaderConfig{Topic: "inventory.update"})`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + const consumers = contracts.filter((c) => c.role === 'consumer'); + + expect(consumers).toHaveLength(1); + expect(consumers[0].contractId).toBe('topic::inventory.update'); + expect(consumers[0].meta.broker).toBe('kafka'); + }); }); describe('Kafka — Python', () => { @@ -309,5 +443,16 @@ await consumer.subscribe({ topic: 'order.placed' });`, expect(producers).toHaveLength(2); expect(consumers).toHaveLength(1); }); + + it('test_extract_ignores_go_test_files', async () => { + writeFile( + 'src/orders_test.go', + `consumer.ConsumePartition("fake-topic", 0, sarama.OffsetNewest)`, + ); + + const contracts = await extractor.extract(null, tmpDir, makeRepo(tmpDir)); + + expect(contracts).toEqual([]); + }); }); });