chore: improve code quality in restic pkg (#736)

pkg/restic/outputs.go

@@ -125,56 +125,105 @@ func (b *BackupProgressEntry) Validate() error {
 	return nil
 }
 
-// readBackupProgressEntries returns the summary event or an error if the command failed.
-func readBackupProgressEntries(output io.Reader, logger io.Writer, callback func(event *BackupProgressEntry)) (*BackupProgressEntry, error) {
+func (b *BackupProgressEntry) IsError() bool {
+	return b.MessageType == "error"
+}
+
+func (b *BackupProgressEntry) IsSummary() bool {
+	return b.MessageType == "summary"
+}
+
+type RestoreProgressEntry struct {
+	MessageType    string  `json:"message_type"` // "summary" or "status"
+	SecondsElapsed float64 `json:"seconds_elapsed"`
+	TotalBytes     int64   `json:"total_bytes"`
+	BytesRestored  int64   `json:"bytes_restored"`
+	TotalFiles     int64   `json:"total_files"`
+	FilesRestored  int64   `json:"files_restored"`
+	PercentDone    float64 `json:"percent_done"`
+}
+
+func (e *RestoreProgressEntry) Validate() error {
+	if e.MessageType != "summary" && e.MessageType != "status" {
+		return fmt.Errorf("message_type must be 'summary' or 'status', got %v", e.MessageType)
+	}
+	return nil
+}
+
+func (r *RestoreProgressEntry) IsError() bool {
+	return r.MessageType == "error"
+}
+
+func (r *RestoreProgressEntry) IsSummary() bool {
+	return r.MessageType == "summary"
+}
+
+type ProgressEntryValidator interface {
+	Validate() error
+	IsError() bool
+	IsSummary() bool
+}
+
+// processProgressOutput handles common JSON output processing logic with proper type safety
+func processProgressOutput[T ProgressEntryValidator](
+	output io.Reader,
+	logger io.Writer,
+	callback func(T)) (T, error) {
+
 	scanner := bufio.NewScanner(output)
 	scanner.Split(bufio.ScanLines)
 
 	nonJSONOutput := bytes.NewBuffer(nil)
+	var captureNonJSON io.Writer = nonJSONOutput
+	if logger != nil {
+		captureNonJSON = io.MultiWriter(nonJSONOutput, logger)
+	}
 
-	var summary *BackupProgressEntry
+	var summary *T
+	var nullT T
 
-	// remaining events are parsed as JSON
 	for scanner.Scan() {
-		var event BackupProgressEntry
-		if err := json.Unmarshal(scanner.Bytes(), &event); err != nil {
-			nonJSONOutput.Write(scanner.Bytes())
-			if logger != nil {
-				logger.Write(scanner.Bytes())
-				logger.Write([]byte("\n"))
-			}
+		line := scanner.Bytes()
+		var event T
+
+		if err := json.Unmarshal(line, &event); err != nil {
+			captureNonJSON.Write(line)
+			captureNonJSON.Write([]byte("\n"))
 			continue
 		}
+
 		if err := event.Validate(); err != nil {
-			nonJSONOutput.Write(scanner.Bytes())
-			if logger != nil {
-				logger.Write(scanner.Bytes())
-				logger.Write([]byte("\n"))
-			}
+			captureNonJSON.Write(line)
+			captureNonJSON.Write([]byte("\n"))
 			continue
 		}
-		if event.MessageType == "error" && logger != nil {
-			logger.Write(scanner.Bytes())
-			logger.Write([]byte("\n"))
+
+		if event.IsError() && logger != nil {
+			captureNonJSON.Write(line)
+			captureNonJSON.Write([]byte("\n"))
 		}
+
 		if callback != nil {
-			callback(&event)
+			callback(event)
 		}
-		if event.MessageType == "summary" {
-			if logger != nil {
-				logger.Write(scanner.Bytes())
-				logger.Write([]byte("\n"))
-			}
-			summary = &event
+
+		if event.IsSummary() {
+			captureNonJSON.Write(line)
+			captureNonJSON.Write([]byte("\n"))
+			eventCopy := event // Make a copy to avoid issues with loop variable
+			summary = &eventCopy
 		}
 	}
+
 	if err := scanner.Err(); err != nil {
-		return summary, newErrorWithOutput(err, nonJSONOutput.String())
+		return nullT, newErrorWithOutput(err, nonJSONOutput.String())
 	}
+
 	if summary == nil {
-		return nil, newErrorWithOutput(errors.New("no summary event found"), nonJSONOutput.String())
+		return nullT, newErrorWithOutput(errors.New("no summary event found"), nonJSONOutput.String())
 	}
-	return summary, nil
+
+	return *summary, nil
 }
 
 type LsEntry struct {
@@ -248,80 +297,6 @@ func (r *ForgetResult) Validate() error {
 	return nil
 }
 
-type RestoreProgressEntry struct {
-	MessageType    string  `json:"message_type"` // "summary" or "status"
-	SecondsElapsed float64 `json:"seconds_elapsed"`
-	TotalBytes     int64   `json:"total_bytes"`
-	BytesRestored  int64   `json:"bytes_restored"`
-	TotalFiles     int64   `json:"total_files"`
-	FilesRestored  int64   `json:"files_restored"`
-	PercentDone    float64 `json:"percent_done"`
-}
-
-func (e *RestoreProgressEntry) Validate() error {
-	if e.MessageType != "summary" && e.MessageType != "status" {
-		return fmt.Errorf("message_type must be 'summary' or 'status', got %v", e.MessageType)
-	}
-	return nil
-}
-
-// readRestoreProgressEntries returns the summary event or an error if the command failed.
-func readRestoreProgressEntries(output io.Reader, logger io.Writer, callback func(event *RestoreProgressEntry)) (*RestoreProgressEntry, error) {
-	scanner := bufio.NewScanner(output)
-	scanner.Split(bufio.ScanLines)
-
-	nonJSONOutput := bytes.NewBuffer(nil)
-
-	var summary *RestoreProgressEntry
-
-	// remaining events are parsed as JSON
-	for scanner.Scan() {
-		var event RestoreProgressEntry
-		if err := json.Unmarshal(scanner.Bytes(), &event); err != nil {
-			nonJSONOutput.Write(scanner.Bytes())
-			if logger != nil {
-				logger.Write(scanner.Bytes())
-				logger.Write([]byte("\n"))
-			}
-			continue
-		}
-		if err := event.Validate(); err != nil {
-			// skip it. Best effort parsing, restic will return with a non-zero exit code if it fails.
-			nonJSONOutput.Write(scanner.Bytes())
-			if logger != nil {
-				logger.Write(scanner.Bytes())
-				logger.Write([]byte("\n"))
-			}
-			continue
-		}
-
-		if event.MessageType == "error" && logger != nil {
-			logger.Write(scanner.Bytes())
-			logger.Write([]byte("\n"))
-		}
-		if callback != nil {
-			callback(&event)
-		}
-		if event.MessageType == "summary" {
-			if logger != nil {
-				logger.Write(scanner.Bytes())
-				logger.Write([]byte("\n"))
-			}
-			summary = &event
-		}
-	}
-
-	if err := scanner.Err(); err != nil {
-		return summary, newErrorWithOutput(err, nonJSONOutput.String())
-	}
-
-	if summary == nil {
-		return nil, newErrorWithOutput(errors.New("no summary event found"), nonJSONOutput.String())
-	}
-
-	return summary, nil
-}
-
 func ValidateSnapshotId(id string) error {
 	if len(id) != 64 {
 		return fmt.Errorf("restic may be out of date (check with `restic self-upgrade`): snapshot ID must be 64 chars, got %v chars", len(id))

pkg/restic/outputs_test.go

@@ -12,7 +12,7 @@ func TestReadBackupProgressEntries(t *testing.T) {
 
 	b := bytes.NewBuffer([]byte(testInput))
 
-	summary, err := readBackupProgressEntries(b, nil, func(event *BackupProgressEntry) {
+	summary, err := processProgressOutput[*BackupProgressEntry](b, nil, func(event *BackupProgressEntry) {
 		t.Logf("event: %v", event)
 	})
 	if err != nil {

pkg/restic/restic.go

@@ -92,6 +92,35 @@ func (r *Repo) pipeCmdOutputToWriter(cmd *exec.Cmd, handlers ...io.Writer) {
 	cmd.Stderr = mw
 }
 
+// executeWithOutput runs a command and captures the output in a buffer
+func (r *Repo) executeWithOutput(ctx context.Context, args []string, opts ...GenericOption) ([]byte, error) {
+	cmd := r.commandWithContext(ctx, args, opts...)
+	output := bytes.NewBuffer(nil)
+	r.pipeCmdOutputToWriter(cmd, output)
+
+	err := cmd.Run()
+	if err != nil {
+		return output.Bytes(), newCmdError(ctx, cmd, newErrorWithOutput(err, output.String()))
+	}
+
+	return output.Bytes(), nil
+}
+
+// executeWithJSONOutput runs a command and parses its JSON output
+func (r *Repo) executeWithJSONOutput(ctx context.Context, args []string, result interface{}, opts ...GenericOption) error {
+	output, err := r.executeWithOutput(ctx, args, opts...)
+	if err != nil {
+		return err
+	}
+
+	if err := json.Unmarshal(output, result); err != nil {
+		return newCmdError(ctx, r.commandWithContext(ctx, args),
+			newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), string(output)))
+	}
+
+	return nil
+}
+
 // Exists checks if the repository exists.
 // Returns true if exists, false if it does not exist OR an access error occurred.
 func (r *Repo) Exists(ctx context.Context, opts ...GenericOption) error {
@@ -143,10 +172,7 @@ func (r *Repo) init(ctx context.Context, opts ...GenericOption) error {
 }
 
 func (r *Repo) Init(ctx context.Context, opts ...GenericOption) error {
-	if err := r.init(ctx, opts...); err != nil && !errors.Is(err, errAlreadyInitialized) {
-		return fmt.Errorf("init failed: %w", err)
-	}
-	return nil
+	return r.init(ctx, opts...)
 }
 
 func (r *Repo) Config(ctx context.Context, opts ...GenericOption) (RepoConfig, error) {
@@ -156,6 +182,56 @@ func (r *Repo) Config(ctx context.Context, opts ...GenericOption) (RepoConfig, e
 	return r.repoConfig, nil
 }
 
+type cmdRunnerWithProgress[T ProgressEntryValidator] struct {
+	repo       *Repo
+	callback   func(T)
+	failureErr error
+}
+
+func (cr *cmdRunnerWithProgress[T]) Run(ctx context.Context, args []string, opts ...GenericOption) (T, error) {
+	logger := LoggerFromContext(ctx)
+	cmdCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used
+	cmd := cr.repo.commandWithContext(cmdCtx, args, opts...)
+
+	// Ensure the command is logged since we're overriding the logger
+	if logger != nil {
+		fmt.Fprintf(logger, "command: %q\n", cmd)
+	}
+
+	buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing
+	reader, writer := nio.Pipe(buf)
+	cr.repo.pipeCmdOutputToWriter(cmd, writer)
+
+	var readErr error
+	var summary T
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		result, err := processProgressOutput[T](reader, logger, cr.callback)
+		summary = result
+		if err != nil {
+			readErr = fmt.Errorf("processing command output: %w", err)
+		}
+	}()
+
+	cmdErr := cmd.Run()
+	writer.Close()
+	wg.Wait()
+
+	if cmdErr != nil || readErr != nil {
+		if cmdErr != nil {
+			cmdErr = cr.repo.handleExitError(cmdErr, cr.failureErr)
+		}
+		return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr))
+	}
+
+	return summary, nil
+}
+
 func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func(*BackupProgressEntry), opts ...GenericOption) (*BackupProgressEntry, error) {
 	for _, p := range paths {
 		if _, err := os.Stat(p); err != nil {
@@ -167,114 +243,43 @@ func (r *Repo) Backup(ctx context.Context, paths []string, progressCallback func
 	args = append(args, paths...)
 	opts = append(slices.Clone(opts), WithEnv("RESTIC_PROGRESS_FPS=2"))
 
-	logger := LoggerFromContext(ctx)
-	cmdCtx, cancel := context.WithCancel(ctx)
-	cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used
-	cmd := r.commandWithContext(cmdCtx, args, opts...)
-
-	// Ensure the command is logged since we're overriding the logger
-	if logger != nil {
-		fmt.Fprintf(logger, "command: %q\n", cmd)
+	cr := cmdRunnerWithProgress[*BackupProgressEntry]{
+		repo:       r,
+		callback:   progressCallback,
+		failureErr: ErrBackupFailed,
 	}
-
-	buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing
-	reader, writer := nio.Pipe(buf)
-	r.pipeCmdOutputToWriter(cmd, writer)
-
-	var readErr error
-	var summary *BackupProgressEntry
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		defer cancel()
-		var err error
-		summary, err = readBackupProgressEntries(reader, logger, progressCallback)
-		if err != nil {
-			readErr = fmt.Errorf("processing command output: %w", err)
-		}
-	}()
-
-	cmdErr := cmd.Run()
-	writer.Close()
-	wg.Wait()
-
-	if cmdErr != nil || readErr != nil {
-		if cmdErr != nil {
-			var exitErr *exec.ExitError
-			if errors.As(cmdErr, &exitErr) {
-				if exitErr.ExitCode() == 3 {
-					cmdErr = ErrPartialBackup
-				} else {
-					cmdErr = fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), ErrBackupFailed)
-				}
-			}
-		}
-		return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr))
-	}
-	return summary, nil
+	return cr.Run(ctx, args, opts...)
 }
 
 func (r *Repo) Restore(ctx context.Context, snapshot string, callback func(*RestoreProgressEntry), opts ...GenericOption) (*RestoreProgressEntry, error) {
 	opts = append(slices.Clone(opts), WithEnv("RESTIC_PROGRESS_FPS=2"))
+	args := []string{"restore", "--json", snapshot}
 
-	logger := LoggerFromContext(ctx)
-	cmdCtx, cancel := context.WithCancel(ctx)
-	cmdCtx = ContextWithLogger(cmdCtx, nil) // ensure no logger is used
-	cmd := r.commandWithContext(cmdCtx, []string{"restore", "--json", snapshot}, opts...)
-	if logger != nil {
-		fmt.Fprintf(logger, "command: %v %v\n", cmd.Path, strings.Join(cmd.Args, " "))
+	cr := cmdRunnerWithProgress[*RestoreProgressEntry]{
+		repo:       r,
+		callback:   callback,
+		failureErr: ErrRestoreFailed,
 	}
-	buf := buffer.New(32 * 1024) // 32KB IO buffer for the realtime event parsing
-	reader, writer := nio.Pipe(buf)
-	r.pipeCmdOutputToWriter(cmd, writer)
+	return cr.Run(ctx, args, opts...)
+}
 
-	var readErr error
-	var summary *RestoreProgressEntry
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		defer cancel()
-		var err error
-		summary, err = readRestoreProgressEntries(reader, logger, callback)
-		if err != nil {
-			readErr = fmt.Errorf("processing command output: %w", err)
+// handleExitError processes a command exit error and converts it to an appropriate error type
+func (r *Repo) handleExitError(err error, failureErr error) error {
+	var exitErr *exec.ExitError
+	if errors.As(err, &exitErr) {
+		if exitErr.ExitCode() == 3 {
+			return ErrPartialBackup
+		} else {
+			return fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), failureErr)
 		}
-	}()
-
-	cmdErr := cmd.Run()
-	writer.Close()
-	wg.Wait()
-	if cmdErr != nil || readErr != nil {
-		if cmdErr != nil {
-			var exitErr *exec.ExitError
-			if errors.As(cmdErr, &exitErr) {
-				if exitErr.ExitCode() == 3 {
-					cmdErr = ErrPartialBackup
-				} else {
-					cmdErr = fmt.Errorf("exit code %d: %w", exitErr.ExitCode(), ErrRestoreFailed)
-				}
-			}
-		}
-
-		return summary, newCmdError(ctx, cmd, errors.Join(cmdErr, readErr))
 	}
-	return summary, nil
+	return err
 }
 
 func (r *Repo) Snapshots(ctx context.Context, opts ...GenericOption) ([]*Snapshot, error) {
-	cmd := r.commandWithContext(ctx, []string{"snapshots", "--json"}, opts...)
-	output := bytes.NewBuffer(nil)
-	r.pipeCmdOutputToWriter(cmd, output)
-
-	if err := cmd.Run(); err != nil {
-		return nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String()))
-	}
-
 	var snapshots []*Snapshot
-	if err := json.Unmarshal(output.Bytes(), &snapshots); err != nil {
-		return nil, newCmdError(ctx, cmd, newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), output.String()))
+	if err := r.executeWithJSONOutput(ctx, []string{"snapshots", "--json"}, &snapshots, opts...); err != nil {
+		return nil, err
 	}
 
 	for _, snapshot := range snapshots {
@@ -289,45 +294,36 @@ func (r *Repo) Forget(ctx context.Context, policy *RetentionPolicy, opts ...Gene
 	args := []string{"forget", "--json"}
 	args = append(args, policy.toForgetFlags()...)
 
-	cmd := r.commandWithContext(ctx, args, opts...)
-	output := bytes.NewBuffer(nil)
-	r.pipeCmdOutputToWriter(cmd, output)
-	if err := cmd.Run(); err != nil {
-		return nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String()))
+	var results []ForgetResult
+	if err := r.executeWithJSONOutput(ctx, args, &results, opts...); err != nil {
+		return nil, err
 	}
 
-	var result []ForgetResult
-	if err := json.Unmarshal(output.Bytes(), &result); err != nil {
-		return nil, newCmdError(ctx, cmd, newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), output.String()))
-	}
-	if len(result) != 1 {
-		return nil, fmt.Errorf("expected 1 output from forget, got %v", len(result))
-	}
-	if err := result[0].Validate(); err != nil {
-		return nil, newCmdError(ctx, cmd, fmt.Errorf("invalid forget result: %w", err))
+	if len(results) != 1 {
+		return nil, fmt.Errorf("expected 1 output from forget, got %v", len(results))
 	}
 
-	return &result[0], nil
+	if err := results[0].Validate(); err != nil {
+		return nil, fmt.Errorf("invalid forget result: %w", err)
+	}
+
+	return &results[0], nil
 }
 
 func (r *Repo) ForgetSnapshot(ctx context.Context, snapshotId string, opts ...GenericOption) error {
 	args := []string{"forget", "--json", snapshotId}
-
-	output := bytes.NewBuffer(nil)
-	cmd := r.commandWithContext(ctx, args, opts...)
-	r.pipeCmdOutputToWriter(cmd, output)
-	if err := cmd.Run(); err != nil {
-		return newCmdError(ctx, cmd, newErrorWithOutput(err, output.String()))
-	}
-
-	return nil
+	_, err := r.executeWithOutput(ctx, args, opts...)
+	return err
 }
 
 func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...GenericOption) error {
-	args := []string{"prune"}
-	cmd := r.commandWithContext(ctx, args, opts...)
-	if pruneOutput != nil {
-		r.pipeCmdOutputToWriter(cmd, pruneOutput)
+	return r.runSimpleCommand(ctx, []string{"prune"}, pruneOutput, opts...)
+}
+
+func (r *Repo) Check(ctx context.Context, checkOutput io.Writer, opts ...GenericOption) error {
+	cmd := r.commandWithContext(ctx, []string{"check"}, opts...)
+	if checkOutput != nil {
+		r.pipeCmdOutputToWriter(cmd, checkOutput)
 	}
 	if err := cmd.Run(); err != nil {
 		return newCmdError(ctx, cmd, err)
@@ -335,12 +331,11 @@ func (r *Repo) Prune(ctx context.Context, pruneOutput io.Writer, opts ...Generic
 	return nil
 }
 
-func (r *Repo) Check(ctx context.Context, checkOutput io.Writer, opts ...GenericOption) error {
-	args := []string{"check"}
+// runSimpleCommand executes a command with optional output capture
+func (r *Repo) runSimpleCommand(ctx context.Context, args []string, outputWriter io.Writer, opts ...GenericOption) error {
 	cmd := r.commandWithContext(ctx, args, opts...)
-	cmd.Stdin = bytes.NewBuffer(nil)
-	if checkOutput != nil {
-		r.pipeCmdOutputToWriter(cmd, checkOutput)
+	if outputWriter != nil {
+		r.pipeCmdOutputToWriter(cmd, outputWriter)
 	}
 	if err := cmd.Run(); err != nil {
 		return newCmdError(ctx, cmd, err)
@@ -357,43 +352,28 @@ func (r *Repo) ListDirectory(ctx context.Context, snapshot string, path string,
 	cmd := r.commandWithContext(ctx, []string{"ls", "--json", snapshot, path}, opts...)
 	output := bytes.NewBuffer(nil)
 	r.pipeCmdOutputToWriter(cmd, output)
-
 	if err := cmd.Run(); err != nil {
 		return nil, nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String()))
 	}
 
-	snapshots, entries, err := readLs(output)
+	snap, entries, err := readLs(output)
 	if err != nil {
-		return nil, nil, newCmdError(ctx, cmd, newErrorWithOutput(err, output.String()))
+		return nil, nil, newCmdError(ctx, cmd, fmt.Errorf("error parsing JSON: %w", err))
 	}
-
-	return snapshots, entries, nil
+	return snap, entries, nil
 }
 
 func (r *Repo) Unlock(ctx context.Context, opts ...GenericOption) error {
-	output := bytes.NewBuffer(nil)
-	cmd := r.commandWithContext(ctx, []string{"unlock"}, opts...)
-	r.pipeCmdOutputToWriter(cmd, output)
-	if err := cmd.Run(); err != nil {
-		return newCmdError(ctx, cmd, newErrorWithOutput(err, output.String()))
-	}
-	return nil
+	_, err := r.executeWithOutput(ctx, []string{"unlock"}, opts...)
+	return err
 }
 
 func (r *Repo) Stats(ctx context.Context, opts ...GenericOption) (*RepoStats, error) {
-	cmd := r.commandWithContext(ctx, []string{"stats", "--json", "--mode=raw-data"}, opts...)
-	output := bytes.NewBuffer(nil)
-	r.pipeCmdOutputToWriter(cmd, output)
-
-	if err := cmd.Run(); err != nil {
-		return nil, newCmdError(ctx, cmd, err)
-	}
-
 	var stats RepoStats
-	if err := json.Unmarshal(output.Bytes(), &stats); err != nil {
-		return nil, newCmdError(ctx, cmd, newErrorWithOutput(fmt.Errorf("command output is not valid JSON: %w", err), output.String()))
+	err := r.executeWithJSONOutput(ctx, []string{"stats", "--json", "--mode=raw-data"}, &stats, opts...)
+	if err != nil {
+		return nil, err
 	}
-
 	return &stats, nil
 }
 
@@ -403,11 +383,8 @@ func (r *Repo) AddTags(ctx context.Context, snapshotIDs []string, tags []string,
 	args = append(args, "--add", strings.Join(tags, ","))
 	args = append(args, snapshotIDs...)
 
-	cmd := r.commandWithContext(ctx, args, opts...)
-	if err := cmd.Run(); err != nil {
-		return newCmdError(ctx, cmd, err)
-	}
-	return nil
+	_, err := r.executeWithOutput(ctx, args, opts...)
+	return err
 }
 
 func (r *Repo) GenericCommand(ctx context.Context, args []string, opts ...GenericOption) error {