code review

Author: Elazul

cmd/watcher/main.go

@@ -46,15 +46,18 @@ func handleExitSignal(cancel context.CancelFunc, container *service.Container) f
 
 		// Cancel the main context immediately so all goroutines start shutting down.
 		cancel()
-		container.GetKafkaClient().Close()
 
 		// Use independent background contexts so these calls succeed even after cancel().
+		// Disconnect MongoDB before Kafka close: Kafka.Close() busy-waits with no timeout
+		// and could block indefinitely, preventing Disconnect() from running.
 		disconnectCtx, disconnectCancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer disconnectCancel()
 		if err := container.GetMongoConnection().Client().Disconnect(disconnectCtx); err != nil {
 			log.Error("Failed to disconnect MongoDB client", logger.Error("error", err))
 		}
 
+		container.GetKafkaClient().Close()
+
 		httpShutdownCtx, httpShutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer httpShutdownCancel()
 		if err := container.GetHttpServer().Close(httpShutdownCtx); err != nil {

internal/mongo/replay_producer.go

@@ -57,7 +57,7 @@ func (r *ReplayProducer) Produce(ctx context.Context) (chan *ChangeEvent, error)
 	var events = make(chan *ChangeEvent)
 
 	go func() {
-		defer func() { closeCursor(cursor) }()
+		defer closeCursor(cursor)
 		defer close(events)
 
 		r.sendEvents(ctx, cursor, events)

internal/mongo/watch_producer.go

@@ -42,11 +42,7 @@ func (w *WatchProducer) GetProducer(o ...WatchOption) ChangeEventProducer {
 
 		go func() {
 			defer close(events)
-			defer func() {
-				if cursor != nil {
-					closeCursor(cursor)
-				}
-			}()
+			defer func() { closeCursor(cursor) }()
 			for {
 				select {
 				case <-ctx.Done():