优化结构

internal/integration/storage/compose.go

@@ -0,0 +1,56 @@
+package storage
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"io"
+)
+
+// composeStreams concatenates objects into destination while calculating the
+// MD5 of the exact bytes written. Provider-specific storage clients stay out
+// of this package; callbacks keep this helper independent from biz and SDKs.
+func composeStreams(
+	ctx context.Context,
+	names []string,
+	open func(context.Context, string) (io.ReadCloser, error),
+	put func(context.Context, string, io.Reader) error,
+	remove func(context.Context, string) error,
+	destination string,
+) (string, error) {
+	reader, writer := io.Pipe()
+	hash := md5.New()
+	errCh := make(chan error, 1)
+	go func() {
+		defer writer.Close()
+		for _, name := range names {
+			if err := ctx.Err(); err != nil {
+				errCh <- err
+				return
+			}
+			file, err := open(ctx, name)
+			if err != nil {
+				errCh <- err
+				return
+			}
+			_, copyErr := io.Copy(io.MultiWriter(writer, hash), file)
+			_ = file.Close()
+			if copyErr != nil {
+				errCh <- copyErr
+				return
+			}
+		}
+		errCh <- nil
+	}()
+	putErr := put(ctx, destination, reader)
+	composeErr := <-errCh
+	if putErr != nil {
+		_ = remove(ctx, destination)
+		return "", putErr
+	}
+	if composeErr != nil {
+		_ = remove(ctx, destination)
+		return "", composeErr
+	}
+	return hex.EncodeToString(hash.Sum(nil)), nil
+}

internal/integration/storage/compose_test.go

@@ -0,0 +1,50 @@
+package storage
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"errors"
+	"io"
+	"strings"
+	"testing"
+)
+
+func TestComposeStreams(t *testing.T) {
+	objects := map[string]string{"a": "hello ", "b": "world"}
+	var stored string
+	hash, err := composeStreams(context.Background(), []string{"a", "b"}, func(_ context.Context, name string) (io.ReadCloser, error) {
+		value, ok := objects[name]
+		if !ok {
+			return nil, errors.New("missing object")
+		}
+		return io.NopCloser(strings.NewReader(value)), nil
+	}, func(_ context.Context, _ string, reader io.Reader) error {
+		value, err := io.ReadAll(reader)
+		stored = string(value)
+		return err
+	}, func(context.Context, string) error { return nil }, "out")
+	if err != nil || stored != "hello world" {
+		t.Fatalf("compose = %q, %v", stored, err)
+	}
+	want := md5.Sum([]byte(stored))
+	if hash != hex.EncodeToString(want[:]) {
+		t.Fatalf("hash = %q", hash)
+	}
+}
+
+func TestComposeStreamsRemovesPartialDestination(t *testing.T) {
+	removed := false
+	_, err := composeStreams(context.Background(), []string{"missing"}, func(context.Context, string) (io.ReadCloser, error) {
+		return nil, errors.New("open failed")
+	}, func(_ context.Context, _ string, reader io.Reader) error {
+		_, _ = io.ReadAll(reader)
+		return nil
+	}, func(context.Context, string) error {
+		removed = true
+		return nil
+	}, "out")
+	if err == nil || !removed {
+		t.Fatalf("compose error = %v, removed = %v", err, removed)
+	}
+}

internal/integration/storage/local.go

@@ -12,7 +12,6 @@ import (
 
 	"kra/internal/biz"
 	"kra/internal/conf"
-	"kra/pkg/osskit"
 )
 
 type fileStorage struct {
@@ -66,7 +65,7 @@ func New(config *conf.AdminBackend) (biz.FileStorage, error) {
 
 func composeFiles(ctx context.Context, storage biz.FileStorage, names []string, destination string) (*biz.StoredFile, string, error) {
 	var stored *biz.StoredFile
-	hash, err := osskit.ComposeStreams(ctx, names, storage.Open, func(ctx context.Context, destination string, reader io.Reader) error {
+	hash, err := composeStreams(ctx, names, storage.Open, func(ctx context.Context, destination string, reader io.Reader) error {
 		var putErr error
 		stored, putErr = storage.Put(ctx, destination, reader)
 		return putErr