@@ -11,6 +11,8 @@ const { Validators } = require('./validators');
1111const { HealthChecker } = require ( './health-checker' ) ;
1212const { displayValidationResults, displayTopicSummary } = require ( './utils' ) ;
1313const { SupabaseAnalytics } = require ( './analytics' ) ;
14+ const { NATSObjectStore } = require ( './nats-object-store' ) ;
15+ const { log } = require ( 'console' ) ;
1416
1517class CLI {
1618 constructor ( options = { } ) {
@@ -21,12 +23,31 @@ class CLI {
2123 } ;
2224 this . kafkaClient = null ;
2325 this . fileService = null ;
26+ this . natsObjectStore = null ;
2427 this . analytics = new SupabaseAnalytics ( ) ;
2528
2629 // Handle bootstrap-servers option mapping to brokers
2730 if ( options . bootstrapServers && ! options . brokers ) {
2831 this . options . brokers = options . bootstrapServers ;
2932 }
33+
34+ // Setup graceful shutdown
35+ this . setupGracefulShutdown ( ) ;
36+ }
37+
38+ setupGracefulShutdown ( ) {
39+ process . on ( 'SIGINT' , async ( ) => {
40+ console . log ( chalk . yellow ( '\n\n🛑 Received SIGINT, shutting down gracefully...' ) ) ;
41+
42+ // Stop NATS monitoring if active
43+ if ( this . natsObjectStore ) {
44+ this . natsObjectStore . stopMonitoring ( ) ;
45+ await this . natsObjectStore . disconnect ( ) ;
46+ }
47+
48+ console . log ( chalk . green ( '✅ Graceful shutdown completed' ) ) ;
49+ process . exit ( 0 ) ;
50+ } ) ;
3051 }
3152
3253 async loadConfigFromFile ( configPath ) {
@@ -115,6 +136,39 @@ class CLI {
115136 }
116137 }
117138
139+ async loadNATSConfig ( natsConfigPath ) {
140+ try {
141+ const fullPath = path . resolve ( natsConfigPath ) ;
142+ if ( ! fs . existsSync ( fullPath ) ) {
143+ throw new Error ( `NATS config file not found: ${ fullPath } ) ;
144+ }
145+
146+ const configContent = fs . readFileSync ( fullPath , 'utf8' ) ;
147+ const natsConfig = JSON . parse ( configContent ) ;
148+
149+ // Create NATS object store
150+ this . natsObjectStore = new NATSObjectStore ( natsConfig ) ;
151+
152+ // Connect to NATS
153+ const connected = await this . natsObjectStore . connect ( ) ;
154+ if ( ! connected ) {
155+ throw new Error ( 'Failed to connect to NATS' ) ;
156+ }
157+
158+ // Setup object store bucket
159+ const bucketReady = await this . natsObjectStore . getOrCreateBucket ( ) ;
160+ if ( ! bucketReady ) {
161+ throw new Error ( 'Failed to setup Object Store bucket' ) ;
162+ }
163+
164+ console . log ( chalk . green ( `✅ NATS configuration loaded from: ${ fullPath } ) ) ;
165+ return true ;
166+ } catch ( error ) {
167+ console . error ( chalk . red ( `❌ Failed to load NATS config file: ${ error . message } ) ) ;
168+ return false ;
169+ }
170+ }
171+
118172 async promptForConfig ( ) {
119173 console . log ( chalk . blue ( '\n🚀 Superstream Kafka Analyzer\n' ) ) ;
120174 console . log ( chalk . gray ( 'Configure your analysis settings:\n' ) ) ;
@@ -177,6 +231,7 @@ class CLI {
177231
178232 // Authentication configuration based on vendor
179233 let saslConfig = null ;
234+ let sslConfig = null ;
180235 if ( vendorAnswer . vendor === 'aws-msk' ) {
181236 console . log ( chalk . yellow ( '\n🔐 AWS MSK Authentication' ) ) ;
182237 const authType = await inquirer . prompt ( [
@@ -268,27 +323,50 @@ class CLI {
268323 {
269324 type : 'input' ,
270325 name : 'username' ,
271- message : 'Username:' ,
326+ message : 'Username (leave empty is using SSL only) :' ,
272327 validate : ( input ) => {
273- if ( ! input . trim ( ) ) return 'Username is required for Aiven' ;
274328 return true ;
275329 }
276330 } ,
277331 {
278332 type : 'password' ,
279333 name : 'password' ,
280- message : 'Password:' ,
334+ message : 'Password (leave empty if using SSL only) :' ,
281335 validate : ( input ) => {
282- if ( ! input . trim ( ) ) return 'Password is required for Aiven' ;
283336 return true ;
284337 }
338+ } ,
339+ {
340+ type : 'input' ,
341+ name : 'caPath' ,
342+ message : 'CA Certificate path (optional, e.g., "./certs/ca.pem"):' ,
343+ default : './certs/ca.pem'
344+ } ,
345+ {
346+ type : 'input' ,
347+ name : 'certPath' ,
348+ message : 'Client Certificate path (optional, eg. "./certs/service.cert"):' ,
349+ default : './certs/service.cert' ,
350+ } ,
351+ {
352+ type : 'input' ,
353+ name : 'keyPath' ,
354+ message : 'Client Private Key path (optional, eg. "./certs/service.key"):' ,
355+ default : './certs/service.key' ,
285356 }
286357 ] ) ;
358+
287359 saslConfig = {
288360 mechanism : 'scram-sha-256' ,
289361 username : aivenAnswers . username ,
290362 password : aivenAnswers . password
291363 } ;
364+
365+ sslConfig = {
366+ ca : aivenAnswers . caPath ,
367+ cert : aivenAnswers . certPath ,
368+ key : aivenAnswers . keyPath
369+ } ;
292370 } else {
293371 // For other vendors, ask if SASL is needed
294372 const useSasl = await inquirer . prompt ( [
@@ -346,7 +424,8 @@ class CLI {
346424 brokers : kafkaAnswers . brokers . split ( ',' ) . map ( broker => broker . trim ( ) ) , // Convert string to array
347425 vendor : vendorAnswer . vendor ,
348426 useSasl : ! ! saslConfig ,
349- sasl : saslConfig
427+ sasl : saslConfig ,
428+ ssl : sslConfig
350429 } ;
351430
352431 // File Output Configuration
@@ -417,6 +496,7 @@ class CLI {
417496
418497 // Use Commander.js options
419498 const configPath = this . options . config ;
499+ const natsConfigPath = this . options . natsConfig ;
420500
421501 if ( configPath ) {
422502 console . log ( chalk . gray ( `Debug: Loading config from: ${ configPath } ) ) ;
@@ -430,6 +510,16 @@ class CLI {
430510 await this . promptForConfig ( ) ;
431511 }
432512
513+ // Load NATS config if provided
514+ if ( natsConfigPath ) {
515+ console . log ( chalk . gray ( `Debug: Loading NATS config from: ${ natsConfigPath } ) ) ;
516+ const natsConfigLoaded = await this . loadNATSConfig ( natsConfigPath ) ;
517+ if ( ! natsConfigLoaded ) {
518+ process . exit ( 1 ) ;
519+ }
520+ console . log ( chalk . gray ( 'Debug: NATS config loaded successfully' ) ) ;
521+ }
522+
433523 // Initialize services without validation
434524 console . log ( chalk . yellow ( '\n⚠️ Validation skipped - proceeding directly to analysis' ) ) ;
435525 this . kafkaClient = new KafkaClient ( this . config . kafka ) ;
@@ -582,6 +672,16 @@ class CLI {
582672 if ( healthResults ) {
583673 analysisResults . healthChecks = healthResults ;
584674 }
675+
676+ // Store results in NATS
677+ if ( this . options . natsConfig ) {
678+ spinner . text = 'Saving results to NATS...' ;
679+ spinner . render ( ) ;
680+ await this . natsObjectStore . storeObject ( analysisResults ) ;
681+ spinner . stop ( ) ;
682+ console . log ( chalk . green ( '\n✅ Analysis completed and stored to NATS!' ) ) ;
683+ process . exit ( 0 ) ;
684+ }
585685
586686 // Check if email is provided for file generation
587687 if ( this . config . email && this . config . email . trim ( ) ) {
0 commit comments