feat(group): expand topic extractor coverage and filter go test noise

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;
   }
 

gitnexus/test/unit/group/topic-extractor.test.ts

@@ -75,8 +75,7 @@ public void handleUserCreated(ConsumerRecord<String, String> 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([]);
+    });
   });
 });