SaveAny-Bot/storage/minio/stream.go

package minio

import (
	"context"
	"fmt"
	"io"

	"github.com/krau/SaveAny-Bot/logger"
	"github.com/minio/minio-go/v7"
)

type MinioWriter struct {
	pipeWriter *io.PipeWriter
	done       chan error
	path       string
	ctx        context.Context
	closed     bool
}

func (w *MinioWriter) Write(p []byte) (n int, err error) {
	select {
	case <-w.ctx.Done():
		return 0, w.ctx.Err()
	default:
		return w.pipeWriter.Write(p)
	}
}

func (w *MinioWriter) Close() error {
	if w.closed {
		return nil
	}
	w.closed = true

	if err := w.pipeWriter.Close(); err != nil {
		return fmt.Errorf("failed to close pipe writer: %w", err)
	}

	select {
	case err := <-w.done:
		if err != nil {
			return fmt.Errorf("upload failed: %w", err)
		}
		return nil
	case <-w.ctx.Done():
		return fmt.Errorf("upload cancelled: %w", w.ctx.Err())
	}
}

func (m *Minio) NewUploadStream(ctx context.Context, storagePath string) (io.WriteCloser, error) {
	logger.L.Infof("Creating upload stream for %s", storagePath)

	uploadCtx, cancel := context.WithCancel(ctx)
	pipeReader, pipeWriter := io.Pipe()
	done := make(chan error, 1)

	go func() {
		defer func() {
			if r := recover(); r != nil {
				done <- fmt.Errorf("panic during upload: %v", r)
			}
			pipeReader.Close()
			cancel()
		}()

		info, err := m.client.PutObject(
			uploadCtx,
			m.config.BucketName,
			storagePath,
			pipeReader,
			-1,
			minio.PutObjectOptions{},
		)

		if err != nil {
			logger.L.Errorf("Failed to upload to %s: %v", storagePath, err)
			done <- err
			return
		}

		logger.L.Infof("uploaded %d bytes to %s", info.Size, storagePath)
		done <- nil
	}()

	return &MinioWriter{
		pipeWriter: pipeWriter,
		done:       done,
		path:       storagePath,
		ctx:        uploadCtx,
		closed:     false,
	}, nil
}