/* MIT License Copyright (c) 2024 Jonas Kaninda Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. */ package go_storage import ( "context" "encoding/json" "errors" "fmt" "io" "net/http" "os" "strconv" "strings" "cloud.google.com/go/storage" "google.golang.org/api/googleapi" "google.golang.org/api/iterator" "google.golang.org/api/option" ) // GCSStorage implements Storage interface for Google Cloud Storage. type GCSStorage struct { client *storage.Client bucket string } // NewGCSStorage creates a new Google Cloud Storage instance. // // Credentials: GCSCredentialsJSON, else GCSCredentialsFile, else Application // Default Credentials. The STORAGE_EMULATOR_HOST environment variable is // honoured by the Google client library itself. func NewGCSStorage(config Config) (*GCSStorage, error) { if err := validateConfig(config, map[string]string{ "bucket": config.GCSBucket, }); err != nil { return nil, fmt.Errorf("gcs: %w", err) } options := make([]option.ClientOption, 0, 2) if config.GCSEndpoint != "" { options = append(options, option.WithEndpoint(config.GCSEndpoint)) } switch { case config.GCSCredentialsJSON != "": data := []byte(config.GCSCredentialsJSON) options = append(options, option.WithAuthCredentialsJSON(gcsCredType(data), data)) case config.GCSCredentialsFile != "": data, err := os.ReadFile(config.GCSCredentialsFile) if err != nil { return nil, fmt.Errorf("gcs: failed to read credentials file: %w", err) } options = append(options, option.WithAuthCredentialsFile(gcsCredType(data), config.GCSCredentialsFile)) } return newGCSStorage(context.Background(), config.GCSBucket, options...) } func newGCSStorage(ctx context.Context, bucket string, options ...option.ClientOption) (*GCSStorage, error) { client, err := storage.NewClient(ctx, options...) if err != nil { return nil, fmt.Errorf("gcs: failed to create client: %w", err) } return &GCSStorage{client: client, bucket: bucket}, nil } // gcsCredType detects the credential type from the JSON "type" field. func gcsCredType(data []byte) option.CredentialsType { var v struct { Type string `json:"type"` } _ = json.Unmarshal(data, &v) switch v.Type { case "authorized_user": return option.AuthorizedUser case "impersonated_service_account": return option.ImpersonatedServiceAccount case "external_account": return option.ExternalAccount default: return option.ServiceAccount } } // Client returns the underlying GCS client. func (g *GCSStorage) Client() *storage.Client { return g.client } func gcsItem(attrs *storage.ObjectAttrs) Item { return Item{ Key: attrs.Name, ModifiedTime: attrs.Updated, Size: attrs.Size, ETag: strconv.FormatInt(attrs.Generation, 10), } } func (g *GCSStorage) put(ctx context.Context, key string, r io.Reader, opts *PutOptions, cond *storage.Conditions) (Item, error) { k, err := cleanKey(key) if err != nil { return Item{}, fmt.Errorf("gcs: %w", err) } obj := g.client.Bucket(g.bucket).Object(k) if cond != nil { obj = obj.If(*cond) } // A child context: cancelling it before Close aborts the upload so the // object is never finalised. wctx, cancel := context.WithCancel(ctx) defer cancel() w := obj.NewWriter(wctx) if opts != nil { w.ContentType = opts.ContentType if len(opts.Metadata) > 0 { w.Metadata = opts.Metadata } } if _, err := io.Copy(w, ctxReader{ctx: ctx, r: r}); err != nil { cancel() _ = w.Close() if cond != nil && isGCSPrecondition(err) { return Item{}, fmt.Errorf("gcs: conditional put of %s: %w", k, ErrPreconditionFailed) } return Item{}, fmt.Errorf("gcs: failed to upload %s: %w", k, err) } if err := ctx.Err(); err != nil { cancel() _ = w.Close() return Item{}, fmt.Errorf("gcs: failed to upload %s: %w", k, err) } if err := w.Close(); err != nil { if cond != nil && isGCSPrecondition(err) { return Item{}, fmt.Errorf("gcs: conditional put of %s: %w", k, ErrPreconditionFailed) } return Item{}, fmt.Errorf("gcs: failed to finalize upload of %s: %w", k, err) } return gcsItem(w.Attrs()), nil } // Put streams r to key. func (g *GCSStorage) Put(ctx context.Context, key string, r io.Reader, opts *PutOptions) (Item, error) { return g.put(ctx, key, r, opts, nil) } // PutIf writes r to key with a generation precondition (IfMatch is the // generation as returned in Item.ETag). func (g *GCSStorage) PutIf(ctx context.Context, key string, r io.Reader, cond Condition, opts *PutOptions) (Item, error) { if err := validateCondition(cond); err != nil { return Item{}, fmt.Errorf("gcs: %w", err) } c := storage.Conditions{} switch { case cond.IfNoneMatch: c.DoesNotExist = true default: gen, err := strconv.ParseInt(cond.IfMatch, 10, 64) if err != nil || gen <= 0 { return Item{}, fmt.Errorf("gcs: invalid generation %q: %w", cond.IfMatch, ErrPreconditionFailed) } c.GenerationMatch = gen } return g.put(ctx, key, r, opts, &c) } // Get streams the object body. func (g *GCSStorage) Get(ctx context.Context, key string) (io.ReadCloser, Item, error) { k, err := cleanKey(key) if err != nil { return nil, Item{}, fmt.Errorf("gcs: %w", err) } rd, err := g.client.Bucket(g.bucket).Object(k).NewReader(ctx) if err != nil { return nil, Item{}, g.mapErr("get", k, err) } return rd, Item{ Key: k, Size: rd.Attrs.Size, ModifiedTime: rd.Attrs.LastModified, ETag: strconv.FormatInt(rd.Attrs.Generation, 10), }, nil } // Stat returns object attributes. func (g *GCSStorage) Stat(ctx context.Context, key string) (Item, error) { k, err := cleanKey(key) if err != nil { return Item{}, fmt.Errorf("gcs: %w", err) } attrs, err := g.client.Bucket(g.bucket).Object(k).Attrs(ctx) if err != nil { return Item{}, g.mapErr("stat", k, err) } return gcsItem(attrs), nil } // Upload uploads a file to GCS. func (g *GCSStorage) Upload(ctx context.Context, localPath string, remotePath string) error { if err := uploadFile(ctx, g, localPath, remotePath); err != nil { return fmt.Errorf("gcs: %w", err) } return nil } // Download downloads a file from GCS. func (g *GCSStorage) Download(ctx context.Context, remotePath string, localPath string) error { if err := downloadFile(ctx, g, remotePath, localPath); err != nil { return fmt.Errorf("gcs: %w", err) } return nil } // List lists objects (recursively) with the given prefix. func (g *GCSStorage) List(ctx context.Context, prefix string) ([]Item, error) { p, err := cleanPrefix(prefix) if err != nil { return nil, fmt.Errorf("gcs: %w", err) } return g.list(ctx, &storage.Query{Prefix: p}) } // ListDir lists the objects directly under dir. func (g *GCSStorage) ListDir(ctx context.Context, dir string) ([]Item, error) { d, err := cleanDir(dir) if err != nil { return nil, fmt.Errorf("gcs: %w", err) } return g.list(ctx, &storage.Query{Prefix: dirPrefix(d), Delimiter: "/"}) } func (g *GCSStorage) list(ctx context.Context, q *storage.Query) ([]Item, error) { items := make([]Item, 0) it := g.client.Bucket(g.bucket).Objects(ctx, q) for { attrs, err := it.Next() if errors.Is(err, iterator.Done) { break } if err != nil { return nil, fmt.Errorf("gcs: failed to list objects in bucket %s: %w", g.bucket, err) } if attrs.Name == "" || (q.Delimiter != "" && strings.HasSuffix(attrs.Name, "/")) { continue // synthetic prefix or directory marker } items = append(items, gcsItem(attrs)) } return items, nil } // Remove deletes an object. func (g *GCSStorage) Remove(ctx context.Context, remotePath string) error { k, err := cleanKey(remotePath) if err != nil { return fmt.Errorf("gcs: %w", err) } if err := g.client.Bucket(g.bucket).Object(k).Delete(ctx); err != nil { return g.mapErr("remove", k, err) } return nil } // Close releases resources. func (g *GCSStorage) Close() error { return g.client.Close() } func (g *GCSStorage) mapErr(op, key string, err error) error { var ge *googleapi.Error if errors.Is(err, storage.ErrObjectNotExist) || (errors.As(err, &ge) && ge.Code == http.StatusNotFound) { return fmt.Errorf("gcs: %s %s in bucket %s: %w", op, key, g.bucket, ErrNotFound) } return fmt.Errorf("gcs: %s %s in bucket %s: %w", op, key, g.bucket, err) } func isGCSPrecondition(err error) bool { var ge *googleapi.Error return errors.As(err, &ge) && (ge.Code == http.StatusPreconditionFailed || ge.Code == http.StatusConflict) }