dhth · dhth · Sep 12, 2025 · Sep 12, 2025 · Sep 12, 2025 · Sep 12, 2025
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- A command for scanning messages in a kafka topic, and optionally writing them
+  to the local filesystem
+
+### Changed
+
+- Messages get persisted to $HOME/.kplay by default; this can be overridden via
+  a flag
+
 ## [v2.0.0] - Apr 23, 2025
 
 ### Added

diff --git a/cmd/root.go b/cmd/root.go
@@ -46,9 +46,11 @@ func Execute() error {
 
 func NewRootCommand() (*cobra.Command, error) {
 	var (
-		configPath      string
-		configPathFull  string
-		homeDir         string
+		configPath     string
+		configPathFull string
+		homeDir        string
+		outputDir      string
+
 		persistMessages bool
 		skipMessages    bool
 		commitMessages  bool
@@ -151,7 +153,7 @@ Behaviours
 				return fmt.Errorf("%w: %s", errCouldntPingBrokers, err.Error())
 			}
 
-			return tui.Render(cl, config, behaviours, homeDir)
+			return tui.Render(cl, config, behaviours, outputDir)
 		},
 	}
 
@@ -261,20 +263,22 @@ Behaviours
 - consumer group          %s
 - authentication          %s
 - encoding                %s
-- brokers                 %v
+- brokers                 %s
 - number of messages      %d
 - save messages           %v
 - decode values           %v
+- output directory        %s
 - batch size              %d
 `,
 					config.Topic,
 					config.ConsumerGroup,
 					config.AuthenticationDisplay(),
 					config.EncodingDisplay(),
-					config.Brokers,
+					strings.Join(config.Brokers, "\n                          "),
 					scanBehaviours.NumMessages,
 					scanBehaviours.SaveMessages,
 					scanBehaviours.Decode,
+					outputDir,
 					scanBehaviours.BatchSize,
 				)
 
@@ -310,7 +314,7 @@ Behaviours
 
 			defer client.Close()
 
-			scanner := scan.New(client, config, scanBehaviours, homeDir)
+			scanner := scan.New(client, config, scanBehaviours, outputDir)
 
 			return scanner.Execute()
 		},
@@ -328,13 +332,15 @@ Behaviours
 	}
 
 	defaultConfigPath := filepath.Join(configDir, configFileName)
+	defaultOutputDir := filepath.Join(homeDir, ".kplay")
 
 	tuiCmd.Flags().StringVarP(&configPath, "config-path", "c", defaultConfigPath, "location of kplay's config file")
 	tuiCmd.Flags().BoolVarP(&persistMessages, "persist-messages", "p", false, "whether to start the TUI with the setting \"persist messages\" ON")
 	tuiCmd.Flags().BoolVarP(&skipMessages, "skip-messages", "s", false, "whether to start the TUI with the setting \"skip messages\" ON")
 	tuiCmd.Flags().BoolVarP(&commitMessages, "commit-messages", "C", true, "whether to start the TUI with the setting \"commit messages\" ON")
 	tuiCmd.Flags().StringVarP(&consumerGroup, consumerGroupFlag, "g", "", "consumer group to use (overrides the one in kplay's config file)")
 	tuiCmd.Flags().BoolVar(&debug, "debug", false, "whether to only display config picked up by kplay without running it")
+	tuiCmd.Flags().StringVarP(&outputDir, "output-dir", "O", defaultOutputDir, "directory to persist messages in")
 
 	serveCmd.Flags().StringVarP(&configPath, "config-path", "c", defaultConfigPath, "location of kplay's config file")
 	serveCmd.Flags().StringVarP(&consumerGroup, consumerGroupFlag, "g", "", "consumer group to use (overrides the one in kplay's config file)")
@@ -352,6 +358,7 @@ Behaviours
 	scanCmd.Flags().BoolVarP(&scanSaveMessages, "save-messages", "s", false, "whether to save kafka messages to the local filesystem")
 	scanCmd.Flags().BoolVarP(&scanDecode, "decode", "d", true, "whether to decode message values (false is equivalent to 'encodingFormat: raw' in kplay's config)")
 	scanCmd.Flags().UintVarP(&scanBatchSize, "batch-size", "b", 100, "number of messages to fetch per batch (must be greater than 0)")
+	scanCmd.Flags().StringVarP(&outputDir, "output-dir", "O", defaultOutputDir, "directory to save scan results in")
 
 	rootCmd.AddCommand(tuiCmd)
 	rootCmd.AddCommand(serveCmd)

diff --git a/internal/scan/scan.go b/internal/scan/scan.go
@@ -29,7 +29,7 @@ type Scanner struct {
 	client     *kgo.Client
 	config     t.Config
 	behaviours Behaviours
-	homeDir    string
+	outputDir  string
 	progress   scanProgress
 }
 
@@ -47,12 +47,12 @@ type scanProgress struct {
 	fsErrors           []fsError
 }
 
-func New(client *kgo.Client, config t.Config, behaviours Behaviours, homeDir string) Scanner {
+func New(client *kgo.Client, config t.Config, behaviours Behaviours, outputDir string) Scanner {
 	scanner := Scanner{
 		client:     client,
 		config:     config,
 		behaviours: behaviours,
-		homeDir:    homeDir,
+		outputDir:  outputDir,
 	}
 
 	return scanner
@@ -101,7 +101,7 @@ func (s *Scanner) scan(ctx context.Context) error {
 	var recordWriter *messageWriter
 
 	now := time.Now().Unix()
-	scanOutputDir := filepath.Join(s.homeDir, ".kplay", "messages", s.config.Topic)
+	scanOutputDir := filepath.Join(s.outputDir, "messages", s.config.Topic)
 
 	err := os.MkdirAll(scanOutputDir, 0o755)
 	if err != nil {

diff --git a/internal/tui/cmds.go b/internal/tui/cmds.go
@@ -31,11 +31,10 @@ func FetchMessages(cl *kgo.Client, config t.Config, commit bool, numRecords int)
 	}
 }
 
-func saveRecordDetailsToDisk(msg t.Message, homeDir, topic string, notifyUserOnSuccess bool) tea.Cmd {
+func saveRecordDetailsToDisk(msg t.Message, outputDir, topic string, notifyUserOnSuccess bool) tea.Cmd {
 	return func() tea.Msg {
 		filePath := filepath.Join(
-			homeDir,
-			".kplay",
+			outputDir,
 			"messages",
 			topic,
 			fmt.Sprintf("partition-%d", msg.Partition),

diff --git a/internal/tui/initial.go b/internal/tui/initial.go
@@ -7,7 +7,7 @@ import (
 	"github.com/twmb/franz-go/pkg/kgo"
 )
 
-func InitialModel(kCl *kgo.Client, config t.Config, behaviours t.TUIBehaviours, homeDir string) Model {
+func InitialModel(kCl *kgo.Client, config t.Config, behaviours t.TUIBehaviours, outputDir string) Model {
 	appDelegateKeys := newAppDelegateKeyMap()
 	appDelegate := newAppItemDelegate(appDelegateKeys)
 	jobItems := make([]list.Item, 0)
@@ -17,7 +17,7 @@ func InitialModel(kCl *kgo.Client, config t.Config, behaviours t.TUIBehaviours,
 		client:            kCl,
 		msgsList:          list.New(jobItems, appDelegate, listWidth, 0),
 		currentMsgIndex:   -1,
-		homeDir:           homeDir,
+		outputDir:         outputDir,
 		behaviours:        behaviours,
 		showHelpIndicator: true,
 	}

diff --git a/internal/tui/model.go b/internal/tui/model.go
@@ -30,7 +30,7 @@ type Model struct {
 	msgDetailsVP       viewport.Model
 	msgDetailsVPReady  bool
 	showHelpIndicator  bool
-	homeDir            string
+	outputDir          string
 	behaviours         t.TUIBehaviours
 	helpVPReady        bool
 	terminalWidth      int

diff --git a/internal/tui/ui.go b/internal/tui/ui.go
@@ -12,7 +12,7 @@ import (
 
 var errCouldntSetupDebugLogging = errors.New("couldn't set up debug logging")
 
-func Render(kCl *kgo.Client, config t.Config, behaviours t.TUIBehaviours, homeDir string) error {
+func Render(kCl *kgo.Client, config t.Config, behaviours t.TUIBehaviours, outputDir string) error {
 	if len(os.Getenv("DEBUG")) > 0 {
 		f, err := tea.LogToFile("debug.log", "debug")
 		if err != nil {
@@ -21,7 +21,7 @@ func Render(kCl *kgo.Client, config t.Config, behaviours t.TUIBehaviours, homeDi
 		defer f.Close()
 	}
 
-	p := tea.NewProgram(InitialModel(kCl, config, behaviours, homeDir), tea.WithAltScreen())
+	p := tea.NewProgram(InitialModel(kCl, config, behaviours, outputDir), tea.WithAltScreen())
 	_, err := p.Run()
 
 	return err

diff --git a/internal/tui/update.go b/internal/tui/update.go
@@ -193,7 +193,7 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 				break
 			}
 
-			cmds = append(cmds, saveRecordDetailsToDisk(message, m.homeDir, m.config.Topic, true))
+			cmds = append(cmds, saveRecordDetailsToDisk(message, m.outputDir, m.config.Topic, true))
 		}
 	case tea.WindowSizeMsg:
 		w1, h1 := messageListStyle.GetFrameSize()
@@ -247,7 +247,7 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 			for _, message := range msg.messages {
 				m.msgsList.InsertItem(len(m.msgsList.Items()), message)
 				if m.behaviours.PersistMessages {
-					cmds = append(cmds, saveRecordDetailsToDisk(message, m.homeDir, m.config.Topic, false))
+					cmds = append(cmds, saveRecordDetailsToDisk(message, m.outputDir, m.config.Topic, false))
 				}
 			}
 			m.msg = fmt.Sprintf("%d message(s) fetched", len(msg.messages))