chore(blobs): generalised blob storage

internal/blob/azureblob/azureblob.go

@@ -0,0 +1,74 @@
+package azureblob
+
+import (
+	"context"
+	"io"
+	"time"
+
+	stblob "github.com/syncthing/syncthing/internal/blob"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
+)
+
+var _ stblob.Store = (*BlobStore)(nil)
+
+type BlobStore struct {
+	client    *azblob.Client
+	container string
+}
+
+func NewBlobStore(accountName, accountKey, containerName string) (*BlobStore, error) {
+	credential, err := azblob.NewSharedKeyCredential(accountName, accountKey)
+	if err != nil {
+		return nil, err
+	}
+	url := "https://" + accountName + ".blob.core.windows.net/"
+	sc, err := azblob.NewClientWithSharedKeyCredential(url, credential, &azblob.ClientOptions{})
+	if err != nil {
+		return nil, err
+	}
+	// This errors when the container already exists, which we ignore.
+	_, _ = sc.CreateContainer(context.Background(), containerName, &container.CreateOptions{})
+	return &BlobStore{
+		client:    sc,
+		container: containerName,
+	}, nil
+}
+
+func (a *BlobStore) Upload(ctx context.Context, key string, data io.Reader) error {
+	_, err := a.client.UploadStream(ctx, a.container, key, data, &blockblob.UploadStreamOptions{})
+	return err
+}
+
+func (a *BlobStore) Download(ctx context.Context, key string, w stblob.Writer) error {
+	resp, err := a.client.DownloadStream(ctx, a.container, key, &blob.DownloadStreamOptions{})
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	_, err = io.Copy(w, resp.Body)
+	return err
+}
+
+func (a *BlobStore) LatestKey(ctx context.Context) (string, error) {
+	opts := &azblob.ListBlobsFlatOptions{}
+	pager := a.client.NewListBlobsFlatPager(a.container, opts)
+	var latest string
+	var lastModified time.Time
+	for pager.More() {
+		page, err := pager.NextPage(ctx)
+		if err != nil {
+			return "", err
+		}
+		for _, blob := range page.Segment.BlobItems {
+			if latest == "" || blob.Properties.LastModified.After(lastModified) {
+				latest = *blob.Name
+				lastModified = *blob.Properties.LastModified
+			}
+		}
+	}
+	return latest, nil
+}

internal/blob/interface.go

@@ -0,0 +1,17 @@
+package blob
+
+import (
+	"context"
+	"io"
+)
+
+type Store interface {
+	Upload(ctx context.Context, key string, r io.Reader) error
+	Download(ctx context.Context, key string, w Writer) error
+	LatestKey(ctx context.Context) (string, error)
+}
+
+type Writer interface {
+	io.Writer
+	io.WriterAt
+}

internal/blob/s3/s3.go

@@ -0,0 +1,106 @@
+// Copyright (C) 2024 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at https://mozilla.org/MPL/2.0/.
+
+package s3
+
+import (
+	"context"
+	"io"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"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/syncthing/syncthing/internal/blob"
+)
+
+var _ blob.Store = (*Session)(nil)
+
+type Session struct {
+	bucket string
+	s3sess *session.Session
+}
+
+type Object = s3.Object
+
+func NewSession(endpoint, region, bucket, accessKeyID, secretKey string) (*Session, error) {
+	sess, err := session.NewSession(&aws.Config{
+		Region:           aws.String(region),
+		Endpoint:         aws.String(endpoint),
+		Credentials:      credentials.NewStaticCredentials(accessKeyID, secretKey, ""),
+		S3ForcePathStyle: aws.Bool(true),
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &Session{
+		bucket: bucket,
+		s3sess: sess,
+	}, nil
+}
+
+func (s *Session) Upload(_ context.Context, key string, r io.Reader) error {
+	uploader := s3manager.NewUploader(s.s3sess)
+	_, err := uploader.Upload(&s3manager.UploadInput{
+		Bucket: aws.String(s.bucket),
+		Key:    aws.String(key),
+		Body:   r,
+	})
+	return err
+}
+
+func (s *Session) Download(_ context.Context, key string, w blob.Writer) error {
+	downloader := s3manager.NewDownloader(s.s3sess)
+	_, err := downloader.Download(w, &s3.GetObjectInput{
+		Bucket: aws.String(s.bucket),
+		Key:    aws.String(key),
+	})
+	return err
+}
+
+func (s *Session) LatestKey(_ context.Context) (string, error) {
+	var latestKey string
+	var lastModified time.Time
+	if err := s.list(func(obj *Object) bool {
+		if latestKey == "" || obj.LastModified.After(lastModified) {
+			latestKey = *obj.Key
+			lastModified = *obj.LastModified
+		}
+		return true
+	}); err != nil {
+		return "", err
+	}
+	return latestKey, nil
+}
+
+func (s *Session) list(fn func(*Object) bool) error {
+	svc := s3.New(s.s3sess)
+
+	opts := &s3.ListObjectsV2Input{
+		Bucket: aws.String(s.bucket),
+	}
+	for {
+		resp, err := svc.ListObjectsV2(opts)
+		if err != nil {
+			return err
+		}
+
+		for _, item := range resp.Contents {
+			if !fn(item) {
+				return nil
+			}
+		}
+
+		if resp.NextContinuationToken == nil || *resp.NextContinuationToken == "" {
+			break
+		}
+		opts.ContinuationToken = resp.NextContinuationToken
+	}
+
+	return nil
+}