refactor: move progress tracking utilities to a new utils package

- Removed redundant progress tracking code from oci.go.
- Introduced NewCountingReader and StartTransferProgressTicker functions in transfer_progress.go for better modularity and reusability.
- Updated oci.go to utilize the new utilities for tracking download and upload progress.

Author: MarcStdt

internal/core/services/registry/oci.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -43,64 +42,6 @@ func NewOciClient(credentialStore *CredentialStore) *OciClient {
 	}
 }
 
-func formatBytes(b int64) string {
-	const (
-		kb = 1024
-		mb = kb * 1024
-		gb = mb * 1024
-	)
-	switch {
-	case b >= gb:
-		return fmt.Sprintf("%.2f GB", float64(b)/float64(gb))
-	case b >= mb:
-		return fmt.Sprintf("%.2f MB", float64(b)/float64(mb))
-	case b >= kb:
-		return fmt.Sprintf("%.2f KB", float64(b)/float64(kb))
-	default:
-		return fmt.Sprintf("%d B", b)
-	}
-}
-
-type progressReader struct {
-	inner       io.Reader
-	transferred *atomic.Int64
-}
-
-func (r *progressReader) Read(p []byte) (int, error) {
-	n, err := r.inner.Read(p)
-	if n > 0 {
-		r.transferred.Add(int64(n))
-	}
-	return n, err
-}
-
-func startProgressTicker(direction string, bytesTransferred *atomic.Int64, totalBytes *atomic.Int64, layersDone *atomic.Int64, totalLayers *atomic.Int64) func() {
-	ticker := time.NewTicker(60 * time.Second)
-	done := make(chan struct{})
-	go func() {
-		for {
-			select {
-			case <-ticker.C:
-				transferred := bytesTransferred.Load()
-				total := totalBytes.Load()
-				layers := layersDone.Load()
-				layerTotal := totalLayers.Load()
-				logger.Log().Info(fmt.Sprintf("%s progress", direction),
-					zap.String("transferred", formatBytes(transferred)),
-					zap.String("total", formatBytes(total)),
-					zap.String("layers", fmt.Sprintf("%d/%d", layers, layerTotal)),
-				)
-			case <-done:
-				return
-			}
-		}
-	}()
-	return func() {
-		ticker.Stop()
-		close(done)
-	}
-}
-
 func (c *OciClient) GetRepo(repoUrl string) (*remote.Repository, error) {
 	repo, err := remote.NewRepository(repoUrl)
 	if err != nil {
@@ -316,7 +257,7 @@ func (c *OciClient) PullSelective(dir string, artifact string, includeData bool,
 		},
 	}
 
-	stopProgress := startProgressTicker("Download", &bytesDownloaded, &totalPullBytes, &completed, &totalLayers)
+	stopProgress := utils.StartTransferProgressTicker("Download", &bytesDownloaded, &totalPullBytes, &completed, &totalLayers)
 
 	// Use a constant destination reference for the local file store so digest references
 	// (which contain ':' and other characters) don't become a tag key.
@@ -454,7 +395,7 @@ func pushBlobWithRetry(ctx context.Context, src *file.Store, dst *remote.Reposit
 			return false, fmt.Errorf("failed to read local blob: %w", err)
 		}
 
-		pr := &progressReader{inner: rc, transferred: bytesUploaded}
+		pr := utils.NewCountingReader(rc, bytesUploaded)
 		err = dst.Push(ctx, desc, pr)
 		rc.Close()
 		if err == nil {
@@ -503,7 +444,7 @@ func copyToRegistry(ctx context.Context, fs *file.Store, repo *remote.Repository
 	var totalLayers atomic.Int64
 	totalLayers.Store(int64(len(layers)))
 
-	stopProgress := startProgressTicker("Upload", &bytesUploaded, &totalBytes, &completed, &totalLayers)
+	stopProgress := utils.StartTransferProgressTicker("Upload", &bytesUploaded, &totalBytes, &completed, &totalLayers)
 
 	for _, desc := range layers {
 		title := desc.Annotations["org.opencontainers.image.title"]

internal/utils/transfer_progress.go

@@ -0,0 +1,74 @@
+package utils
+
+import (
+	"fmt"
+	"io"
+	"sync/atomic"
+	"time"
+
+	"github.com/highcard-dev/daemon/internal/utils/logger"
+	"go.uber.org/zap"
+)
+
+// countingReader wraps r and adds every successful Read size to transferred.
+type countingReader struct {
+	inner       io.Reader
+	transferred *atomic.Int64
+}
+
+// NewCountingReader returns an io.Reader that counts bytes read into transferred.
+func NewCountingReader(r io.Reader, transferred *atomic.Int64) io.Reader {
+	return &countingReader{inner: r, transferred: transferred}
+}
+
+func (r *countingReader) Read(p []byte) (int, error) {
+	n, err := r.inner.Read(p)
+	if n > 0 {
+		r.transferred.Add(int64(n))
+	}
+	return n, err
+}
+
+// StartTransferProgressTicker logs periodic progress and a final "done" line
+// (transferred, total, layers, average speed, elapsed on done).
+// Call the returned function when the transfer finishes.
+func StartTransferProgressTicker(direction string, bytesTransferred, totalBytes, layersDone, totalLayers *atomic.Int64) func() {
+	ticker := time.NewTicker(60 * time.Second)
+	done := make(chan struct{})
+	start := time.Now()
+
+	report := func(event string, withElapsed bool) {
+		n := bytesTransferred.Load()
+		sec := time.Since(start).Seconds()
+		speed := ""
+		if sec > 0 {
+			speed = HumanizeBytes(int64(float64(n)/sec)) + "/s"
+		}
+		fields := []zap.Field{
+			zap.String("transferred", HumanizeBytes(n)),
+			zap.String("total", HumanizeBytes(totalBytes.Load())),
+			zap.String("layers", fmt.Sprintf("%d/%d", layersDone.Load(), totalLayers.Load())),
+			zap.String("speed", speed),
+		}
+		if withElapsed {
+			fields = append(fields, zap.Duration("elapsed", time.Since(start)))
+		}
+		logger.Log().Info(direction+" "+event, fields...)
+	}
+
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				report("progress", false)
+			case <-done:
+				return
+			}
+		}
+	}()
+	return func() {
+		ticker.Stop()
+		close(done)
+		report("done", true)
+	}
+}