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)
	// A failed destination may stop reading before the producer reaches EOF.
	// Close the pipe reader in that case so the producer's next write returns
	// instead of leaving the goroutine blocked forever.
	if putErr != nil {
		_ = reader.CloseWithError(putErr)
	} else {
		_ = reader.Close()
	}
	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
}