func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, index int) (io.ReadCloser, error) {
	if err := c.Init(ctx); err != nil {
		return nil, err
	} else if generation == "" {
		return nil, fmt.Errorf("generation required")
	}
	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
	out, err := c.s3.GetObjectWithContext(ctx, &s3.GetObjectInput{
		Bucket: aws.String(c.Bucket),
		Key:    aws.String(key),
	})
	if isNotExists(err) {
		return nil, os.ErrNotExist
	} else if err != nil {
		return nil, err
	}
	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "GET").Inc()
	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(*out.ContentLength))
	return out.Body, nil
}
SnapshotReader returns a reader for snapshot data at the given generation/index.
package s3
import (
	"context"
	"crypto/tls"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"path"
	"regexp"
	"strings"
	"sync"
	"time"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/defaults"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/benbjohnson/litestream"
	"github.com/benbjohnson/litestream/internal"
	"golang.org/x/sync/errgroup"
)
const ReplicaClientType = "s3"
const MaxKeys = 1000
const DefaultRegion = "us-east-1"
var _ litestream.ReplicaClient = (*ReplicaClient)(nil)
type ReplicaClient struct {
	mu       sync.Mutex
	s3       *s3.S3  
	uploader *s3manager.Uploader
	AccessKeyID     string
	SecretAccessKey string
	Region         string
	Bucket         string
	Path           string
	Endpoint       string
	ForcePathStyle bool
	SkipVerify     bool
}
func NewReplicaClient() *ReplicaClient {
	return &ReplicaClient{}
}
func (c *ReplicaClient) Type() string {
	return ReplicaClientType
}
func (c *ReplicaClient) Init(ctx context.Context) (err error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.s3 != nil {
		return nil
	}
	region := c.Region
	if region == "" {
		if c.Endpoint == "" {
			if region, err = c.findBucketRegion(ctx, c.Bucket); err != nil {
				return fmt.Errorf("cannot lookup bucket region: %w", err)
			}
		} else {
			region = DefaultRegion  
		}
	}
	config := c.config()
	if region != "" {
		config.Region = aws.String(region)
	}
	sess, err := session.NewSession(config)
	if err != nil {
		return fmt.Errorf("cannot create aws session: %w", err)
	}
	c.s3 = s3.New(sess)
	c.uploader = s3manager.NewUploader(sess)
	return nil
}
func (c *ReplicaClient) config() *aws.Config {
	config := defaults.Get().Config
	if c.AccessKeyID != "" || c.SecretAccessKey != "" {
		config.Credentials = credentials.NewStaticCredentials(c.AccessKeyID, c.SecretAccessKey, "")
	}
	if c.Endpoint != "" {
		config.Endpoint = aws.String(c.Endpoint)
	}
	if c.ForcePathStyle {
		config.S3ForcePathStyle = aws.Bool(c.ForcePathStyle)
	}
	if c.SkipVerify {
		config.HTTPClient = &http.Client{Transport: &http.Transport{
			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
		}}
	}
	return config
}
func (c *ReplicaClient) findBucketRegion(ctx context.Context, bucket string) (string, error) {
	config := c.config()
	config.Region = aws.String(DefaultRegion)
	sess, err := session.NewSession(config)
	if err != nil {
		return "", err
	}
	if out, err := s3.New(sess).GetBucketLocation(&s3.GetBucketLocationInput{
		Bucket: aws.String(bucket),
	}); err != nil {
		return "", err
	} else if out.LocationConstraint != nil {
		return *out.LocationConstraint, nil
	}
	return DefaultRegion, nil
}
func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
	if err := c.Init(ctx); err != nil {
		return nil, err
	}
	var generations []string
	if err := c.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
		Bucket:    aws.String(c.Bucket),
		Prefix:    aws.String(path.Join(c.Path, "generations") + "/"),
		Delimiter: aws.String("/"),
	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
		for _, prefix := range page.CommonPrefixes {
			name := path.Base(*prefix.Prefix)
			if !litestream.IsGenerationName(name) {
				continue
			}
			generations = append(generations, name)
		}
		return true
	}); err != nil {
		return nil, err
	}
	return generations, nil
}
func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string) error {
	if err := c.Init(ctx); err != nil {
		return err
	} else if generation == "" {
		return fmt.Errorf("generation required")
	}
	var objIDs []*s3.ObjectIdentifier
	if err := c.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
		Bucket: aws.String(c.Bucket),
		Prefix: aws.String(path.Join(c.Path, "generations", generation)),
	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
		for _, obj := range page.Contents {
			objIDs = append(objIDs, &s3.ObjectIdentifier{Key: obj.Key})
		}
		return true
	}); err != nil {
		return err
	}
	for len(objIDs) > 0 {
		n := MaxKeys
		if len(objIDs) < n {
			n = len(objIDs)
		}
		if _, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
			Bucket: aws.String(c.Bucket),
			Delete: &s3.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
		}); err != nil {
			return err
		}
		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
		objIDs = objIDs[n:]
	}
	return nil
}
func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (litestream.SnapshotIterator, error) {
	if err := c.Init(ctx); err != nil {
		return nil, err
	}
	return newSnapshotIterator(ctx, c, generation), nil
}
func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, index int, rd io.Reader) (info litestream.SnapshotInfo, err error) {
	if err := c.Init(ctx); err != nil {
		return info, err
	} else if generation == "" {
		return info, fmt.Errorf("generation required")
	}
	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
	startTime := time.Now()
	rc := internal.NewReadCounter(rd)
	if _, err := c.uploader.UploadWithContext(ctx, &s3manager.UploadInput{
		Bucket: aws.String(c.Bucket),
		Key:    aws.String(key),
		Body:   rc,
	}); err != nil {
		return info, err
	}
	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "PUT").Inc()
	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "PUT").Add(float64(rc.N()))
	return litestream.SnapshotInfo{
		Generation: generation,
		Index:      index,
		Size:       rc.N(),
		CreatedAt:  startTime.UTC(),
	}, nil
}