pkg-proxy/internal/storage/blob.go

package storage

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"time"

	"gocloud.dev/blob"
	_ "gocloud.dev/blob/azureblob"
	_ "gocloud.dev/blob/fileblob"
	_ "gocloud.dev/blob/s3blob"
	"gocloud.dev/gcerrors"
)

const osWindows = "windows"

// attrsExt is fileblob's sidecar suffix, kept only to clear sidecars an
// earlier version wrote.
const attrsExt = ".attrs"

// Blob implements Storage using gocloud.dev/blob.
// Supports local filesystem (file://) and S3 (s3://) URLs.
type Blob struct {
	bucket *blob.Bucket
	url    string

	// fileRoot is the directory backing a file:// bucket, empty for cloud
	// backends. Used only to clear sidecars an earlier version wrote.
	fileRoot string
}

// OpenBucket opens a blob bucket from a URL.
//
// Supported URL schemes:
//   - file:///path/to/dir - Local filesystem storage
//   - s3://bucket-name - Amazon S3 (uses AWS_* environment variables)
//   - s3://bucket-name?region=us-east-1&endpoint=http://localhost:9000 - S3-compatible (MinIO, etc.)
//   - gs://bucket-name - Google Cloud Storage (uses Application Default Credentials;
//     supports Workload Identity on GKE/GCE without any extra configuration)
//   - azblob://container-name - Azure Blob Storage
//
// For local filesystem, the directory is created if it doesn't exist.
//
//nolint:ireturn // The URL scheme selects the storage implementation.
func OpenBucket(ctx context.Context, urlStr string) (Storage, error) {
	if strings.HasPrefix(urlStr, "gs://") {
		return OpenGCS(ctx, urlStr)
	}

	var fileRoot string

	// Handle file:// URLs specially to create the directory
	if strings.HasPrefix(urlStr, "file://") {
		path := strings.TrimPrefix(urlStr, "file://")

		// Handle file:/// (three slashes) for absolute paths
		if strings.HasPrefix(path, "/") && runtime.GOOS != osWindows {
			// Unix: file:///path -> /path
			// path is already correct
		} else if strings.HasPrefix(path, "/") && runtime.GOOS == osWindows {
			// Windows: file:///C:/path -> C:/path
			path = strings.TrimPrefix(path, "/")
		}

		// Convert forward slashes to native path separators for filesystem operations
		nativePath := filepath.FromSlash(path)

		// Ensure directory exists
		if err := os.MkdirAll(nativePath, dirPermissions); err != nil {
			return nil, fmt.Errorf("creating directory: %w", err)
		}

		// fileblob requires an absolute path with forward slashes
		absPath, err := filepath.Abs(nativePath)
		if err != nil {
			return nil, fmt.Errorf("resolving path: %w", err)
		}

		fileRoot = absPath

		// Convert back to URL format with forward slashes
		urlPath := filepath.ToSlash(absPath)
		if runtime.GOOS == osWindows {
			// Windows needs file:///C:/path format
			urlStr = "file:///" + urlPath
		} else {
			urlStr = "file://" + urlPath
		}

		// Create temp files next to the final path instead of in os.TempDir.
		// This avoids "invalid cross-device link" errors from os.Rename when
		// the bucket directory and os.TempDir are on different filesystems
		// (e.g. Docker volume mounts).
		//
		// Do not write fileblob's ".attrs" sidecar. It is rewritten with
		// os.Create, truncating in place outside the atomic rename that
		// protects the blob, so a read overlapping a write can decode a
		// partial file; a missing one defaults cleanly, a truncated one does
		// not. Nothing in the proxy needs it: Store sets no ContentType, and
		// Size reads os.Stat via Attributes.
		urlStr += "?no_tmp_dir=true&metadata=skip"
	}

	bucket, err := blob.OpenBucket(ctx, urlStr)
	if err != nil {
		return nil, fmt.Errorf("opening bucket: %w", err)
	}

	return &Blob{bucket: bucket, url: urlStr, fileRoot: fileRoot}, nil
}

// legacySidecarPath gives the ".attrs" path an earlier version wrote for key,
// or "" when that path would not be a file inside fileRoot.
func (b *Blob) legacySidecarPath(key string) string {
	if p := b.localPath(key); p != "" {
		return p + attrsExt
	}
	return ""
}

// localPath gives the file fileblob keeps key in, or "" when that would not
// be a file inside fileRoot.
//
// The key is escaped the way fileblob escapes it on the way to disk.
// filepath.Localize then validates the escaped form: it rejects an empty,
// absolute or ".." path, and "." would name fileRoot itself. What it declines
// are keys the proxy never produces.
func (b *Blob) localPath(key string) string {
	if b.fileRoot == "" {
		return ""
	}
	rel, err := filepath.Localize(escapeKey(key))
	if err != nil || rel == "." {
		return ""
	}
	return filepath.Join(b.fileRoot, rel)
}

// escapeKey mirrors fileblob's unexported escapeKey, which hex-escapes a rune
// as "__0x<hex>__". Slashes stay as "/" for filepath.Localize to convert.
func escapeKey(key string) string {
	runes := []rune(key)
	var out strings.Builder
	for i, r := range runes {
		if escapeRune(runes, i) {
			fmt.Fprintf(&out, "__%#x__", r)
		} else {
			out.WriteRune(r)
		}
	}
	return out.String()
}

// escapeRune is fileblob's rule for which runes of a key to escape: control
// characters, a raw path separator, a slash that would form "../", "//" or
// end the key, and on Windows the characters its filesystem reserves.
func escapeRune(r []rune, i int) bool {
	c := r[i]
	switch {
	case c < ' ':
		return true
	case os.PathSeparator != '/' && c == os.PathSeparator:
		return true
	case i > 1 && c == '/' && r[i-1] == '.' && r[i-2] == '.':
		return true
	case i > 0 && c == '/' && r[i-1] == '/':
		return true
	case c == '/' && i == len(r)-1:
		return true
	case os.PathSeparator == '\\' && strings.ContainsRune(`<>:"|?*`, c):
		return true
	}
	return false
}

// clearLegacySidecar removes the ".attrs" file an earlier version wrote for
// key. Nothing rewrites one now, so a sidecar left partial by an interrupted
// write would fail every read of that key for good. Removing is atomic where
// the rewrite was not, so a concurrent reader gets the whole old file or
// nothing.
//
// Failure is deliberately not fatal. Usually the key never had a sidecar and
// os.Remove reports not-exist. A real failure leaves exactly the state this
// change inherited, while failing the write would turn a cleanup miss into a
// failed request. Windows makes that concrete: Go opens files without
// FILE_SHARE_DELETE, so a reader holding the sidecar open blocks deletion, and
// that reader is the very workload this change protects. The next store of the
// key retries.
func (b *Blob) clearLegacySidecar(key string) {
	if sidecar := b.legacySidecarPath(key); sidecar != "" {
		_ = os.Remove(sidecar)
	}
}

func (b *Blob) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) {
	b.clearLegacySidecar(path)

	// Compute hash while writing
	h := sha256.New()
	tee := io.TeeReader(r, h)

	// gocloud commits whatever was written when the writer closes, unless
	// its context is cancelled first, so a failed copy cancels it to abort.
	writeCtx, abort := context.WithCancel(ctx)
	defer abort()

	opts := &blob.WriterOptions{}
	w, err := b.bucket.NewWriter(writeCtx, path, opts)
	if err != nil {
		return 0, "", fmt.Errorf("creating writer: %w", err)
	}

	size, err := io.Copy(w, tee)
	if err != nil {
		abort()
		_ = w.Close()
		return 0, "", fmt.Errorf("writing content: %w", err)
	}

	if err := w.Close(); err != nil {
		return 0, "", fmt.Errorf("closing writer: %w", err)
	}

	hash := hex.EncodeToString(h.Sum(nil))
	return size, hash, nil
}

func (b *Blob) Open(ctx context.Context, path string) (io.ReadCloser, error) {
	if localPath := b.localPath(path); localPath != "" {
		r, err := os.Open(localPath)
		if err != nil {
			if os.IsNotExist(err) {
				return nil, ErrNotFound
			}
			return nil, fmt.Errorf("opening local reader: %w", err)
		}
		return r, nil
	}

	r, err := b.bucket.NewReader(ctx, path, nil)
	if err != nil {
		if isNotExist(err) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("opening reader: %w", err)
	}
	return r, nil
}

func (b *Blob) Exists(ctx context.Context, path string) (bool, error) {
	exists, err := b.bucket.Exists(ctx, path)
	if err != nil {
		return false, fmt.Errorf("checking existence: %w", err)
	}
	return exists, nil
}

// Delete removes the object at path. On a file:// bucket it also removes the
// object's fetch directory once empty, since fileblob leaves directories
// behind. Any other directory, such as a version directory another fetch may
// be creating its own directory in, is left alone.
func (b *Blob) Delete(ctx context.Context, path string) error {
	err := b.bucket.Delete(ctx, path)
	if err != nil && !isNotExist(err) {
		return fmt.Errorf("deleting object: %w", err)
	}
	if p := b.localPath(path); p != "" {
		if dir := filepath.Dir(p); dir != b.fileRoot && isFetchDir(filepath.Base(dir)) {
			_ = os.Remove(dir) // fails, harmlessly, while the directory holds anything
		}
	}
	return nil
}

func (b *Blob) SignedURL(ctx context.Context, path string, expiry time.Duration) (string, error) {
	url, err := b.bucket.SignedURL(ctx, path, &blob.SignedURLOptions{
		Method: http.MethodGet,
		Expiry: expiry,
	})
	if err != nil {
		if gcerrors.Code(err) == gcerrors.Unimplemented {
			return "", ErrSignedURLUnsupported
		}
		return "", fmt.Errorf("signing URL: %w", err)
	}
	return url, nil
}

func (b *Blob) Size(ctx context.Context, path string) (int64, error) {
	attrs, err := b.bucket.Attributes(ctx, path)
	if err != nil {
		if isNotExist(err) {
			return 0, ErrNotFound
		}
		return 0, fmt.Errorf("getting attributes: %w", err)
	}
	return attrs.Size, nil
}

func (b *Blob) UsedSpace(ctx context.Context) (int64, error) {
	var total int64

	iter := b.bucket.List(nil)
	for {
		obj, err := iter.Next(ctx)
		if err == io.EOF {
			break
		}
		if err != nil {
			return 0, fmt.Errorf("listing objects: %w", err)
		}
		total += obj.Size
	}

	return total, nil
}

// ListPrefix returns object metadata for keys under a prefix.
func (b *Blob) ListPrefix(ctx context.Context, prefix string) ([]ObjectInfo, error) {
	iter := b.bucket.List(&blob.ListOptions{Prefix: prefix})
	objects := make([]ObjectInfo, 0)

	for {
		obj, err := iter.Next(ctx)
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("listing objects: %w", err)
		}
		if obj.IsDir {
			continue
		}

		info := ObjectInfo{
			Path:    obj.Key,
			Size:    obj.Size,
			ModTime: obj.ModTime,
		}

		objects = append(objects, info)
	}

	return objects, nil
}

func (b *Blob) Close() error {
	return b.bucket.Close()
}

func (b *Blob) URL() string {
	return b.url
}

func isNotExist(err error) bool {
	return gcerrors.Code(err) == gcerrors.NotFound
}