package storage

import (
	"context"
	"fmt"
	"net/http"
	"reflect"
	"time"

	"cloud.google.com/go/internal/optional"
	"cloud.google.com/go/internal/trace"
	"google.golang.org/api/googleapi"
	"google.golang.org/api/iterator"
	raw "google.golang.org/api/storage/v1"
)

type BucketHandle struct {
	c                *Client
	name             string
	acl              ACLHandle
	defaultObjectACL ACLHandle
	conds            *BucketConditions
	userProject      string // project for Requester Pays buckets
}

func (c *Client) Bucket(name string) *BucketHandle {
	return &BucketHandle{
		c:    c,
		name: name,
		acl: ACLHandle{
			c:      c,
			bucket: name,
		},
		defaultObjectACL: ACLHandle{
			c:         c,
			bucket:    name,
			isDefault: true,
		},
	}
}

func (b *BucketHandle) Create(ctx context.Context, projectID string, attrs *BucketAttrs) (err error) {
	ctx = trace.StartSpan(ctx, "cloud.google.com/go/storage.Bucket.Create")
	defer func() { trace.EndSpan(ctx, err) }()

	var bkt *raw.Bucket
	if attrs != nil {
		bkt = attrs.toRawBucket()
	} else {
		bkt = &raw.Bucket{}
	}

	if bkt.Location == "" && bkt.Lifecycle != nil {
		bkt.Location = "US"
	}
	req := b.c.raw.Buckets.Insert(projectID, bkt)
	setClientHeader(req.Header())
	if attrs != nil && attrs.PredefinedACL != "" {
		req.PredefinedAcl(attrs.PredefinedACL)
	}
	if attrs != nil && attrs.PredefinedDefaultObjectACL != "" {
		req.PredefinedDefaultObjectAcl(attrs.PredefinedDefaultObjectACL)
	}
	return runWithRetry(ctx, func() error { _, err := req.Context(ctx).Do(); return err })
}

func (b *BucketHandle) Delete(ctx context.Context) (err error) {
	ctx = trace.StartSpan(ctx, "cloud.google.com/go/storage.Bucket.Delete")
	defer func() { trace.EndSpan(ctx, err) }()

	req, err := b.newDeleteCall()
	if err != nil {
		return err
	}
	return runWithRetry(ctx, func() error { return req.Context(ctx).Do() })
}

func (b *BucketHandle) newDeleteCall() (*raw.BucketsDeleteCall, error) {
	req := b.c.raw.Buckets.Delete(b.name)
	setClientHeader(req.Header())
	if err := applyBucketConds("BucketHandle.Delete", b.conds, req); err != nil {
		return nil, err
	}
	if b.userProject != "" {
		req.UserProject(b.userProject)
	}
	return req, nil
}

func (b *BucketHandle) ACL() *ACLHandle {
	return &b.acl
}

func (b *BucketHandle) DefaultObjectACL() *ACLHandle {
	return &b.defaultObjectACL
}

func (b *BucketHandle) Object(name string) *ObjectHandle {
	return &ObjectHandle{
		c:      b.c,
		bucket: b.name,
		object: name,
		acl: ACLHandle{
			c:           b.c,
			bucket:      b.name,
			object:      name,
			userProject: b.userProject,
		},
		gen:         -1,
		userProject: b.userProject,
	}
}

func (b *BucketHandle) Attrs(ctx context.Context) (attrs *BucketAttrs, err error) {
	ctx = trace.StartSpan(ctx, "cloud.google.com/go/storage.Bucket.Attrs")
	defer func() { trace.EndSpan(ctx, err) }()

	req, err := b.newGetCall()
	if err != nil {
		return nil, err
	}
	var resp *raw.Bucket
	err = runWithRetry(ctx, func() error {
		resp, err = req.Context(ctx).Do()
		return err
	})
	if e, ok := err.(*googleapi.Error); ok && e.Code == http.StatusNotFound {
		return nil, ErrBucketNotExist
	}
	if err != nil {
		return nil, err
	}
	return newBucket(resp)
}

func (b *BucketHandle) newGetCall() (*raw.BucketsGetCall, error) {
	req := b.c.raw.Buckets.Get(b.name).Projection("full")
	setClientHeader(req.Header())
	if err := applyBucketConds("BucketHandle.Attrs", b.conds, req); err != nil {
		return nil, err
	}
	if b.userProject != "" {
		req.UserProject(b.userProject)
	}
	return req, nil
}

func (b *BucketHandle) Update(ctx context.Context, uattrs BucketAttrsToUpdate) (attrs *BucketAttrs, err error) {
	ctx = trace.StartSpan(ctx, "cloud.google.com/go/storage.Bucket.Create")
	defer func() { trace.EndSpan(ctx, err) }()

	req, err := b.newPatchCall(&uattrs)
	if err != nil {
		return nil, err
	}
	if uattrs.PredefinedACL != "" {
		req.PredefinedAcl(uattrs.PredefinedACL)
	}
	if uattrs.PredefinedDefaultObjectACL != "" {
		req.PredefinedDefaultObjectAcl(uattrs.PredefinedDefaultObjectACL)

	rb, err := req.Context(ctx).Do()
	if err != nil {
		return nil, err
	}
	return newBucket(rb)
}

func (b *BucketHandle) newPatchCall(uattrs *BucketAttrsToUpdate) (*raw.BucketsPatchCall, error) {
	rb := uattrs.toRawBucket()
	req := b.c.raw.Buckets.Patch(b.name, rb).Projection("full")
	setClientHeader(req.Header())
	if err := applyBucketConds("BucketHandle.Update", b.conds, req); err != nil {
		return nil, err
	}
	if b.userProject != "" {
		req.UserProject(b.userProject)
	}
	return req, nil
}

	Name string

	ACL []ACLRule

	BucketPolicyOnly BucketPolicyOnly

	UniformBucketLevelAccess UniformBucketLevelAccess

	DefaultObjectACL []ACLRule

	DefaultEventBasedHold bool

	PredefinedACL string

	PredefinedDefaultObjectACL string

	Location string

	MetaGeneration int64

	StorageClass string

	Created time.Time

	VersioningEnabled bool

	Labels map[string]string

	RequesterPays bool

	Lifecycle Lifecycle

	RetentionPolicy *RetentionPolicy

	CORS []CORS

	Encryption *BucketEncryption

	Logging *BucketLogging

	Website *BucketWebsite

	Etag string

	LocationType string
}

	LockedTime time.Time
}

	LockedTime time.Time
}

type Lifecycle struct {
	Rules []LifecycleRule
}

	RetentionPeriod time.Duration

	EffectiveTime time.Time

	IsLocked bool
}

	rfc3339Date = "2006-01-02"

	DeleteAction = "Delete"

	SetStorageClassAction = "SetStorageClass"
)

	Action LifecycleAction

	Condition LifecycleCondition
}

	Type string

	StorageClass string
}

type Liveness int

	Archived
)

	AgeInDays int64

	CreatedBefore time.Time

	Liveness Liveness

	MatchesStorageClasses []string

	NumNewerVersions int64
}

	LogBucket string

	LogObjectPrefix string
}

	MainPageSuffix string

	NotFoundPage string
}

func newBucket(b *raw.Bucket) (*BucketAttrs, error) {
	if b == nil {
		return nil, nil
	}
	rp, err := toRetentionPolicy(b.RetentionPolicy)
	if err != nil {
		return nil, err
	}
	return &BucketAttrs{
		Name:                     b.Name,
		Location:                 b.Location,
		MetaGeneration:           b.Metageneration,
		DefaultEventBasedHold:    b.DefaultEventBasedHold,
		StorageClass:             b.StorageClass,
		Created:                  convertTime(b.TimeCreated),
		VersioningEnabled:        b.Versioning != nil && b.Versioning.Enabled,
		ACL:                      toBucketACLRules(b.Acl),
		DefaultObjectACL:         toObjectACLRules(b.DefaultObjectAcl),
		Labels:                   b.Labels,
		RequesterPays:            b.Billing != nil && b.Billing.RequesterPays,
		Lifecycle:                toLifecycle(b.Lifecycle),
		RetentionPolicy:          rp,
		CORS:                     toCORS(b.Cors),
		Encryption:               toBucketEncryption(b.Encryption),
		Logging:                  toBucketLogging(b.Logging),
		Website:                  toBucketWebsite(b.Website),
		BucketPolicyOnly:         toBucketPolicyOnly(b.IamConfiguration),
		UniformBucketLevelAccess: toUniformBucketLevelAccess(b.IamConfiguration),
		Etag:                     b.Etag,
		LocationType:             b.LocationType,
	}, nil
}

	var labels map[string]string
	if len(b.Labels) > 0 {
		labels = make(map[string]string, len(b.Labels))
		for k, v := range b.Labels {
			labels[k] = v
		}

	var v *raw.BucketVersioning
	if b.VersioningEnabled {
		v = &raw.BucketVersioning{Enabled: true}
	}
	var bb *raw.BucketBilling
	if b.RequesterPays {
		bb = &raw.BucketBilling{RequesterPays: true}
	}
	var bktIAM *raw.BucketIamConfiguration
	if b.UniformBucketLevelAccess.Enabled || b.BucketPolicyOnly.Enabled {
		bktIAM = &raw.BucketIamConfiguration{
			UniformBucketLevelAccess: &raw.BucketIamConfigurationUniformBucketLevelAccess{
				Enabled: true,
			},
		}
	}
	return &raw.Bucket{
		Name:             b.Name,
		Location:         b.Location,
		StorageClass:     b.StorageClass,
		Acl:              toRawBucketACL(b.ACL),
		DefaultObjectAcl: toRawObjectACL(b.DefaultObjectACL),
		Versioning:       v,
		Labels:           labels,
		Billing:          bb,
		Lifecycle:        toRawLifecycle(b.Lifecycle),
		RetentionPolicy:  b.RetentionPolicy.toRawRetentionPolicy(),
		Cors:             toRawCORS(b.CORS),
		Encryption:       b.Encryption.toRawBucketEncryption(),
		Logging:          b.Logging.toRawBucketLogging(),
		Website:          b.Website.toRawBucketWebsite(),
		IamConfiguration: bktIAM,
	}
}

	MaxAge time.Duration

	Methods []string

	Origins []string

	ResponseHeaders []string
}

	DefaultKMSKeyName string
}

	VersioningEnabled optional.Bool

	RequesterPays optional.Bool

	DefaultEventBasedHold optional.Bool

	BucketPolicyOnly *BucketPolicyOnly

	UniformBucketLevelAccess *UniformBucketLevelAccess

	RetentionPolicy *RetentionPolicy

	CORS []CORS

	Encryption *BucketEncryption

	Lifecycle *Lifecycle

	Logging *BucketLogging

	Website *BucketWebsite

	PredefinedACL string

	PredefinedDefaultObjectACL string

	setLabels    map[string]string
	deleteLabels map[string]bool
}

func (ua *BucketAttrsToUpdate) SetLabel(name, value string) {
	if ua.setLabels == nil {
		ua.setLabels = map[string]string{}
	}
	ua.setLabels[name] = value
}

func (ua *BucketAttrsToUpdate) DeleteLabel(name string) {
	if ua.deleteLabels == nil {
		ua.deleteLabels = map[string]bool{}
	}
	ua.deleteLabels[name] = true
}

func (ua *BucketAttrsToUpdate) toRawBucket() *raw.Bucket {
	rb := &raw.Bucket{}
	if ua.CORS != nil {
		rb.Cors = toRawCORS(ua.CORS)
		rb.ForceSendFields = append(rb.ForceSendFields, "Cors")
	}
	if ua.DefaultEventBasedHold != nil {
		rb.DefaultEventBasedHold = optional.ToBool(ua.DefaultEventBasedHold)
		rb.ForceSendFields = append(rb.ForceSendFields, "DefaultEventBasedHold")
	}
	if ua.RetentionPolicy != nil {
		if ua.RetentionPolicy.RetentionPeriod == 0 {
			rb.NullFields = append(rb.NullFields, "RetentionPolicy")
			rb.RetentionPolicy = nil
		} else {
			rb.RetentionPolicy = ua.RetentionPolicy.toRawRetentionPolicy()
		}
	}
	if ua.VersioningEnabled != nil {
		rb.Versioning = &raw.BucketVersioning{
			Enabled:         optional.ToBool(ua.VersioningEnabled),
			ForceSendFields: []string{"Enabled"},
		}
	}
	if ua.RequesterPays != nil {
		rb.Billing = &raw.BucketBilling{
			RequesterPays:   optional.ToBool(ua.RequesterPays),
			ForceSendFields: []string{"RequesterPays"},
		}
	}
	if ua.BucketPolicyOnly != nil {
		rb.IamConfiguration = &raw.BucketIamConfiguration{
			UniformBucketLevelAccess: &raw.BucketIamConfigurationUniformBucketLevelAccess{
				Enabled:         ua.BucketPolicyOnly.Enabled,
				ForceSendFields: []string{"Enabled"},
			},
		}
	}
	if ua.UniformBucketLevelAccess != nil {
		rb.IamConfiguration = &raw.BucketIamConfiguration{
			UniformBucketLevelAccess: &raw.BucketIamConfigurationUniformBucketLevelAccess{
				Enabled:         ua.UniformBucketLevelAccess.Enabled,
				ForceSendFields: []string{"Enabled"},
			},
		}
	}
	if ua.Encryption != nil {
		if ua.Encryption.DefaultKMSKeyName == "" {
			rb.NullFields = append(rb.NullFields, "Encryption")
			rb.Encryption = nil
		} else {
			rb.Encryption = ua.Encryption.toRawBucketEncryption()
		}
	}
	if ua.Lifecycle != nil {
		rb.Lifecycle = toRawLifecycle(*ua.Lifecycle)
		rb.ForceSendFields = append(rb.ForceSendFields, "Lifecycle")
	}
	if ua.Logging != nil {
		if *ua.Logging == (BucketLogging{}) {
			rb.NullFields = append(rb.NullFields, "Logging")
			rb.Logging = nil
		} else {
			rb.Logging = ua.Logging.toRawBucketLogging()
		}
	}
	if ua.Website != nil {
		if *ua.Website == (BucketWebsite{}) {
			rb.NullFields = append(rb.NullFields, "Website")
			rb.Website = nil
		} else {
			rb.Website = ua.Website.toRawBucketWebsite()
		}
	}

		rb.Acl = nil
		rb.ForceSendFields = append(rb.ForceSendFields, "Acl")
	}

		rb.DefaultObjectAcl = nil
		rb.ForceSendFields = append(rb.ForceSendFields, "DefaultObjectAcl")
	}
	if ua.setLabels != nil || ua.deleteLabels != nil {
		rb.Labels = map[string]string{}
		for k, v := range ua.setLabels {
			rb.Labels[k] = v
		}
		if len(rb.Labels) == 0 && len(ua.deleteLabels) > 0 {
			rb.ForceSendFields = append(rb.ForceSendFields, "Labels")
		}
		for l := range ua.deleteLabels {
			rb.NullFields = append(rb.NullFields, "Labels."+l)
		}
	}
	return rb
}

func (b *BucketHandle) If(conds BucketConditions) *BucketHandle {
	b2 := *b
	b2.conds = &conds
	return &b2
}

	MetagenerationMatch int64

	MetagenerationNotMatch int64
}

func (c *BucketConditions) validate(method string) error {
	if *c == (BucketConditions{}) {
		return fmt.Errorf("storage: %s: empty conditions", method)
	}
	if c.MetagenerationMatch != 0 && c.MetagenerationNotMatch != 0 {
		return fmt.Errorf("storage: %s: multiple conditions specified for metageneration", method)
	}
	return nil
}

func (b *BucketHandle) UserProject(projectID string) *BucketHandle {
	b2 := *b
	b2.userProject = projectID
	b2.acl.userProject = projectID
	b2.defaultObjectACL.userProject = projectID
	return &b2
}

func (b *BucketHandle) LockRetentionPolicy(ctx context.Context) error {
	var metageneration int64
	if b.conds != nil {
		metageneration = b.conds.MetagenerationMatch
	}
	req := b.c.raw.Buckets.LockRetentionPolicy(b.name, metageneration)
	_, err := req.Context(ctx).Do()
	return err
}

func applyBucketConds(method string, conds *BucketConditions, call interface{}) error {
	if conds == nil {
		return nil
	}
	if err := conds.validate(method); err != nil {
		return err
	}
	cval := reflect.ValueOf(call)
	switch {
	case conds.MetagenerationMatch != 0:
		if !setConditionField(cval, "IfMetagenerationMatch", conds.MetagenerationMatch) {
			return fmt.Errorf("storage: %s: ifMetagenerationMatch not supported", method)
		}
	case conds.MetagenerationNotMatch != 0:
		if !setConditionField(cval, "IfMetagenerationNotMatch", conds.MetagenerationNotMatch) {
			return fmt.Errorf("storage: %s: ifMetagenerationNotMatch not supported", method)
		}
	}
	return nil
}

func (rp *RetentionPolicy) toRawRetentionPolicy() *raw.BucketRetentionPolicy {
	if rp == nil {
		return nil
	}
	return &raw.BucketRetentionPolicy{
		RetentionPeriod: int64(rp.RetentionPeriod / time.Second),
	}
}

func toRetentionPolicy(rp *raw.BucketRetentionPolicy) (*RetentionPolicy, error) {
	if rp == nil {
		return nil, nil
	}
	t, err := time.Parse(time.RFC3339, rp.EffectiveTime)
	if err != nil {
		return nil, err
	}
	return &RetentionPolicy{
		RetentionPeriod: time.Duration(rp.RetentionPeriod) * time.Second,
		EffectiveTime:   t,
		IsLocked:        rp.IsLocked,
	}, nil
}

func toRawCORS(c []CORS) []*raw.BucketCors {
	var out []*raw.BucketCors
	for _, v := range c {
		out = append(out, &raw.BucketCors{
			MaxAgeSeconds:  int64(v.MaxAge / time.Second),
			Method:         v.Methods,
			Origin:         v.Origins,
			ResponseHeader: v.ResponseHeaders,
		})
	}
	return out
}

func toCORS(rc []*raw.BucketCors) []CORS {
	var out []CORS
	for _, v := range rc {
		out = append(out, CORS{
			MaxAge:          time.Duration(v.MaxAgeSeconds) * time.Second,
			Methods:         v.Method,
			Origins:         v.Origin,
			ResponseHeaders: v.ResponseHeader,
		})
	}
	return out
}

func toRawLifecycle(l Lifecycle) *raw.BucketLifecycle {
	var rl raw.BucketLifecycle
	if len(l.Rules) == 0 {
		rl.ForceSendFields = []string{"Rule"}
	}
	for _, r := range l.Rules {
		rr := &raw.BucketLifecycleRule{
			Action: &raw.BucketLifecycleRuleAction{
				Type:         r.Action.Type,
				StorageClass: r.Action.StorageClass,
			},
			Condition: &raw.BucketLifecycleRuleCondition{
				Age:                 r.Condition.AgeInDays,
				MatchesStorageClass: r.Condition.MatchesStorageClasses,
				NumNewerVersions:    r.Condition.NumNewerVersions,
			},
		}

		switch r.Condition.Liveness {
		case LiveAndArchived:
			rr.Condition.IsLive = nil
		case Live:
			rr.Condition.IsLive = googleapi.Bool(true)
		case Archived:
			rr.Condition.IsLive = googleapi.Bool(false)
		}

		if !r.Condition.CreatedBefore.IsZero() {
			rr.Condition.CreatedBefore = r.Condition.CreatedBefore.Format(rfc3339Date)
		}
		rl.Rule = append(rl.Rule, rr)
	}
	return &rl
}

func toLifecycle(rl *raw.BucketLifecycle) Lifecycle {
	var l Lifecycle
	if rl == nil {
		return l
	}
	for _, rr := range rl.Rule {
		r := LifecycleRule{
			Action: LifecycleAction{
				Type:         rr.Action.Type,
				StorageClass: rr.Action.StorageClass,
			},
			Condition: LifecycleCondition{
				AgeInDays:             rr.Condition.Age,
				MatchesStorageClasses: rr.Condition.MatchesStorageClass,
				NumNewerVersions:      rr.Condition.NumNewerVersions,
			},
		}

		if rr.Condition.IsLive == nil {
			r.Condition.Liveness = LiveAndArchived
		} else if *rr.Condition.IsLive {
			r.Condition.Liveness = Live
		} else {
			r.Condition.Liveness = Archived
		}

		if rr.Condition.CreatedBefore != "" {
			r.Condition.CreatedBefore, _ = time.Parse(rfc3339Date, rr.Condition.CreatedBefore)
		}
		l.Rules = append(l.Rules, r)
	}
	return l
}

func (e *BucketEncryption) toRawBucketEncryption() *raw.BucketEncryption {
	if e == nil {
		return nil
	}
	return &raw.BucketEncryption{
		DefaultKmsKeyName: e.DefaultKMSKeyName,
	}
}

func toBucketEncryption(e *raw.BucketEncryption) *BucketEncryption {
	if e == nil {
		return nil
	}
	return &BucketEncryption{DefaultKMSKeyName: e.DefaultKmsKeyName}
}

func (b *BucketLogging) toRawBucketLogging() *raw.BucketLogging {
	if b == nil {
		return nil
	}
	return &raw.BucketLogging{
		LogBucket:       b.LogBucket,
		LogObjectPrefix: b.LogObjectPrefix,
	}
}

func toBucketLogging(b *raw.BucketLogging) *BucketLogging {
	if b == nil {
		return nil
	}
	return &BucketLogging{
		LogBucket:       b.LogBucket,
		LogObjectPrefix: b.LogObjectPrefix,
	}
}

func (w *BucketWebsite) toRawBucketWebsite() *raw.BucketWebsite {
	if w == nil {
		return nil
	}
	return &raw.BucketWebsite{
		MainPageSuffix: w.MainPageSuffix,
		NotFoundPage:   w.NotFoundPage,
	}
}

func toBucketWebsite(w *raw.BucketWebsite) *BucketWebsite {
	if w == nil {
		return nil
	}
	return &BucketWebsite{
		MainPageSuffix: w.MainPageSuffix,
		NotFoundPage:   w.NotFoundPage,
	}
}

func toBucketPolicyOnly(b *raw.BucketIamConfiguration) BucketPolicyOnly {
	if b == nil || b.BucketPolicyOnly == nil || !b.BucketPolicyOnly.Enabled {
		return BucketPolicyOnly{}
	}
	lt, err := time.Parse(time.RFC3339, b.BucketPolicyOnly.LockedTime)
	if err != nil {
		return BucketPolicyOnly{
			Enabled: true,
		}
	}
	return BucketPolicyOnly{
		Enabled:    true,
		LockedTime: lt,
	}
}

func toUniformBucketLevelAccess(b *raw.BucketIamConfiguration) UniformBucketLevelAccess {
	if b == nil || b.UniformBucketLevelAccess == nil || !b.UniformBucketLevelAccess.Enabled {
		return UniformBucketLevelAccess{}
	}
	lt, err := time.Parse(time.RFC3339, b.UniformBucketLevelAccess.LockedTime)
	if err != nil {
		return UniformBucketLevelAccess{
			Enabled: true,
		}
	}
	return UniformBucketLevelAccess{
		Enabled:    true,
		LockedTime: lt,
	}
}

func (b *BucketHandle) Objects(ctx context.Context, q *Query) *ObjectIterator {
	it := &ObjectIterator{
		ctx:    ctx,
		bucket: b,
	}
	it.pageInfo, it.nextFunc = iterator.NewPageInfo(
		it.fetch,
		func() int { return len(it.items) },
		func() interface{} { b := it.items; it.items = nil; return b })
	if q != nil {
		it.query = *q
	}
	return it
}

type ObjectIterator struct {
	ctx      context.Context
	bucket   *BucketHandle
	query    Query
	pageInfo *iterator.PageInfo
	nextFunc func() error
	items    []*ObjectAttrs
}

func (it *ObjectIterator) PageInfo() *iterator.PageInfo { return it.pageInfo }

func (it *ObjectIterator) Next() (*ObjectAttrs, error) {
	if err := it.nextFunc(); err != nil {
		return nil, err
	}
	item := it.items[0]
	it.items = it.items[1:]
	return item, nil
}

func (it *ObjectIterator) fetch(pageSize int, pageToken string) (string, error) {
	req := it.bucket.c.raw.Objects.List(it.bucket.name)
	setClientHeader(req.Header())
	req.Projection("full")
	req.Delimiter(it.query.Delimiter)
	req.Prefix(it.query.Prefix)
	req.Versions(it.query.Versions)
	if len(it.query.fieldSelection) > 0 {
		req.Fields("nextPageToken", googleapi.Field(it.query.fieldSelection))
	}
	req.PageToken(pageToken)
	if it.bucket.userProject != "" {
		req.UserProject(it.bucket.userProject)
	}
	if pageSize > 0 {
		req.MaxResults(int64(pageSize))
	}
	var resp *raw.Objects
	var err error
	err = runWithRetry(it.ctx, func() error {
		resp, err = req.Context(it.ctx).Do()
		return err
	})
	if err != nil {
		if e, ok := err.(*googleapi.Error); ok && e.Code == http.StatusNotFound {
			err = ErrBucketNotExist
		}
		return "", err
	}
	for _, item := range resp.Items {
		it.items = append(it.items, newObject(item))
	}
	for _, prefix := range resp.Prefixes {
		it.items = append(it.items, &ObjectAttrs{Prefix: prefix})
	}
	return resp.NextPageToken, nil
}

func (c *Client) Buckets(ctx context.Context, projectID string) *BucketIterator {
	it := &BucketIterator{
		ctx:       ctx,
		client:    c,
		projectID: projectID,
	}
	it.pageInfo, it.nextFunc = iterator.NewPageInfo(
		it.fetch,
		func() int { return len(it.buckets) },
		func() interface{} { b := it.buckets; it.buckets = nil; return b })

	return it
}

	Prefix string

	ctx       context.Context
	client    *Client
	projectID string
	buckets   []*BucketAttrs
	pageInfo  *iterator.PageInfo
	nextFunc  func() error
}

func (it *BucketIterator) Next() (*BucketAttrs, error) {
	if err := it.nextFunc(); err != nil {
		return nil, err
	}
	b := it.buckets[0]
	it.buckets = it.buckets[1:]
	return b, nil
}

func (it *BucketIterator) PageInfo() *iterator.PageInfo { return it.pageInfo }

func (it *BucketIterator) fetch(pageSize int, pageToken string) (token string, err error) {
	req := it.client.raw.Buckets.List(it.projectID)
	setClientHeader(req.Header())
	req.Projection("full")
	req.Prefix(it.Prefix)
	req.PageToken(pageToken)
	if pageSize > 0 {
		req.MaxResults(int64(pageSize))
	}
	var resp *raw.Buckets
	err = runWithRetry(it.ctx, func() error {
		resp, err = req.Context(it.ctx).Do()
		return err
	})
	if err != nil {
		return "", err
	}
	for _, item := range resp.Items {
		b, err := newBucket(item)
		if err != nil {
			return "", err
		}
		it.buckets = append(it.buckets, b)
	}
	return resp.NextPageToken, nil