ReleaseKit/internal/git/blob_reader.go

package git

import (
	"bufio"
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"os/exec"
	"strconv"
	"strings"
	"sync"

	"gitea.maximumdirect.net/eric/ReleaseKit/internal/process"
)

var errBlobObjectMissing = errors.New("Git object is missing")

// BlobReader reads requested blobs through one lazy git cat-file --batch
// process. Reads are serialized; callers must close the reader when finished.
type BlobReader struct {
	repositoryDir string
	parent        context.Context

	mu       sync.Mutex
	child    *process.PipedProcess
	stdout   *bufio.Reader
	stderr   bytes.Buffer
	closed   bool
	aborted  bool
	terminal error
	closeErr error
}

// NewBlobReader creates a lazy reader for one repository. It starts no child
// process until the first valid blob is requested.
func NewBlobReader(ctx context.Context, repositoryDir string) *BlobReader {
	return &BlobReader{repositoryDir: repositoryDir, parent: ctx}
}

// Read returns one blob body from the candidate repository. The returned bytes
// are newly allocated and owned by the caller.
func (reader *BlobReader) Read(ctx context.Context, objectID string) ([]byte, error) {
	if !validObjectID(objectID) {
		return nil, fmt.Errorf("invalid blob object ID %q", objectID)
	}
	if err := ctx.Err(); err != nil {
		return nil, fmt.Errorf("read blob %s: %w", objectID, err)
	}

	reader.mu.Lock()
	defer reader.mu.Unlock()
	if reader.terminal != nil {
		return nil, reader.terminal
	}
	if reader.closed {
		return nil, fmt.Errorf("read blob %s: Git blob reader is closed", objectID)
	}
	if err := reader.startLocked(); err != nil {
		return nil, fmt.Errorf("read blob %s: %w", objectID, err)
	}

	child := reader.child
	callbackDone := make(chan struct{})
	stopCancellation := context.AfterFunc(ctx, func() {
		defer close(callbackDone)
		child.Cancel()
		_ = child.Stdin().Close()
		_ = child.Stdout().Close()
	})
	body, err := reader.exchange(objectID)
	if !stopCancellation() {
		<-callbackDone
	}
	if contextErr := ctx.Err(); contextErr != nil {
		err = contextErr
	} else if contextErr := reader.parent.Err(); contextErr != nil {
		err = contextErr
	}
	if errors.Is(err, errBlobObjectMissing) {
		return nil, err
	}
	if err != nil {
		return nil, reader.abortLocked(objectID, err)
	}
	return body, nil
}

// Close closes a healthy session cleanly or reaps an aborted session. It is
// safe to call more than once.
func (reader *BlobReader) Close() error {
	reader.mu.Lock()
	defer reader.mu.Unlock()
	if reader.closed {
		return reader.closeErr
	}
	reader.closed = true
	if reader.aborted {
		reader.closeErr = reader.terminal
		return reader.closeErr
	}
	if reader.child == nil {
		return nil
	}
	if err := reader.child.Stdin().Close(); err != nil {
		reader.closeErr = reader.abortLocked("", fmt.Errorf("close Git cat-file input: %w", err))
		return reader.closeErr
	}
	if _, err := io.Copy(io.Discard, reader.child.Stdout()); err != nil {
		reader.closeErr = reader.abortLocked("", fmt.Errorf("finish Git cat-file output: %w", err))
		return reader.closeErr
	}
	if err := reader.child.Wait(); err != nil {
		reader.closeErr = reader.processError("close Git cat-file batch", err)
	}
	return reader.closeErr
}

func (reader *BlobReader) startLocked() error {
	if reader.child != nil {
		return nil
	}
	if err := reader.parent.Err(); err != nil {
		return err
	}
	child, err := process.StartPiped(reader.parent, process.Invocation{
		Args: []string{"git", "cat-file", "--batch"},
		Dir:  reader.repositoryDir,
		Env: map[string]string{
			"GIT_NO_LAZY_FETCH":  "1",
			"GIT_ALLOW_PROTOCOL": "",
		},
		Stderr: &reader.stderr,
	})
	if err != nil {
		return err
	}
	reader.child = child
	reader.stdout = bufio.NewReader(child.Stdout())
	return nil
}

func (reader *BlobReader) exchange(objectID string) ([]byte, error) {
	if _, err := io.WriteString(reader.child.Stdin(), objectID+"\n"); err != nil {
		return nil, fmt.Errorf("write object request: %w", err)
	}
	header, err := reader.stdout.ReadString('\n')
	if err != nil {
		return nil, fmt.Errorf("read object header: %w", err)
	}
	fields := strings.Fields(strings.TrimSuffix(header, "\n"))
	if len(fields) == 2 && strings.EqualFold(fields[0], objectID) && fields[1] == "missing" {
		return nil, fmt.Errorf("read blob %s: %w", objectID, errBlobObjectMissing)
	}
	if len(fields) != 3 {
		return nil, fmt.Errorf("invalid Git blob header %q", strings.TrimSuffix(header, "\n"))
	}
	if !strings.EqualFold(fields[0], objectID) {
		return nil, fmt.Errorf("Git blob response object ID %q does not match request %q", fields[0], objectID)
	}
	if fields[1] != "blob" {
		return nil, fmt.Errorf("Git object %s has type %q, want blob", objectID, fields[1])
	}
	size, err := strconv.ParseInt(fields[2], 10, strconv.IntSize)
	if err != nil || size < 0 {
		if err == nil {
			err = errors.New("negative object size")
		}
		return nil, fmt.Errorf("invalid Git blob size %q: %w", fields[2], err)
	}
	body := make([]byte, int(size))
	if _, err := io.ReadFull(reader.stdout, body); err != nil {
		return nil, fmt.Errorf("read %d-byte Git blob body: %w", size, err)
	}
	framing, err := reader.stdout.ReadByte()
	if err != nil {
		return nil, fmt.Errorf("read Git blob framing: %w", err)
	}
	if framing != '\n' {
		return nil, fmt.Errorf("invalid Git blob framing byte %q", framing)
	}
	return body, nil
}

func (reader *BlobReader) abortLocked(objectID string, cause error) error {
	if reader.aborted {
		return reader.terminal
	}
	reader.aborted = true
	var waitErr error
	if reader.child != nil {
		reader.child.Cancel()
		_ = reader.child.Stdin().Close()
		_ = reader.child.Stdout().Close()
		waitErr = reader.child.Wait()
	}
	if contextErr := reader.parent.Err(); contextErr != nil {
		cause = contextErr
	}
	reader.terminal = reader.processError("read blob "+objectID, errors.Join(cause, relevantProcessError(waitErr, cause)))
	return reader.terminal
}

func (reader *BlobReader) processError(operation string, cause error) error {
	message := strings.TrimSpace(reader.stderr.String())
	if message != "" {
		return fmt.Errorf("%s: %s: %w", operation, message, cause)
	}
	return fmt.Errorf("%s: %w", operation, cause)
}

func relevantProcessError(waitErr, cause error) error {
	if waitErr == nil || errors.Is(cause, context.Canceled) || errors.Is(cause, context.DeadlineExceeded) {
		return nil
	}
	var exitErr *exec.ExitError
	if errors.As(waitErr, &exitErr) && exitErr.ExitCode() == -1 {
		return nil
	}
	return waitErr
}