fix: rewrite streaming logic and centralize in a helper function

controller/files.go

@@ -12,6 +12,8 @@ import (
 	"io"
 )
 
+var errStreamDone = errors.New("done")
+
 type FilesController struct {
 	Controller
 }
@@ -74,12 +76,8 @@ func (f *FilesController) GetDownloadBy(cidString string) {
 		return
 	}
 
-	err = ctx.StreamWriter(func(w io.Writer) error {
-		_, err = io.Copy(w, download)
-		_ = download.(io.Closer).Close()
-		return err
-	})
-	if internalError(ctx, err) {
+	err = passThroughStream(download, ctx)
+	if err != errStreamDone && internalError(ctx, err) {
 		logger.Get().Debug("failed streaming file", zap.Error(err))
 	}
 }
@@ -93,17 +91,13 @@ func (f *FilesController) GetProofBy(cidString string) {
 		return
 	}
 
-	download, err := files.DownloadProof(hashHex)
+	proof, err := files.DownloadProof(hashHex)
 	if internalError(ctx, err) {
 		logger.Get().Debug("failed fetching file proof", zap.Error(err))
 		return
 	}
 
-	err = ctx.StreamWriter(func(w io.Writer) error {
-		_, err = io.Copy(w, download)
-		_ = download.(io.Closer).Close()
-		return err
-	})
+	err = passThroughStream(proof, ctx)
 	if internalError(ctx, err) {
 		logger.Get().Debug("failed streaming file proof", zap.Error(err))
 	}
@@ -183,3 +177,36 @@ func validateCid(cidString string, validateStatus bool, ctx iris.Context) (strin
 
 	return hashHex, true
 }
+
+func passThroughStream(stream io.Reader, ctx iris.Context) error {
+	closed := false
+
+	err := ctx.StreamWriter(func(w io.Writer) error {
+		if closed {
+			return errStreamDone
+		}
+
+		count, err := io.CopyN(w, stream, 1024)
+		if count == 0 || err == io.EOF {
+			err = stream.(io.Closer).Close()
+			if err != nil {
+				logger.Get().Error("failed closing stream", zap.Error(err))
+				return err
+			}
+			closed = true
+			return nil
+		}
+
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+
+	if err == errStreamDone {
+		err = nil
+	}
+
+	return err
+}

service/files/files.go

@@ -164,7 +164,7 @@ func Download(hash string) (io.Reader, error) {
 	tusItem := db.Get().Table("tus").Where(&model.Tus{Hash: hash}).Row()
 
 	if uploadItem.Err() == nil {
-		fetch, err := client.R().SetDoNotParseResponse(true).Get(getWorkerProofUrl(hash))
+		fetch, err := client.R().SetDoNotParseResponse(true).Get(getWorkerObjectUrl(hash))
 		if err != nil {
 			logger.Get().Error(ErrFailedFetchObject.Error(), zap.Error(err))
 			return nil, ErrFailedFetchObject