comparing cfbcc31012529bcaec101039e1b70668538002b0 and hailey/s3-blobstore on hailey.at/cocoon

+11 -6

cmd/cocoon/main.go

···

       107
       107
        
       				Name:    "s3-backups-enabled",

     

       108
       108
        
       				EnvVars: []string{"COCOON_S3_BACKUPS_ENABLED"},

     

       109
       109
        
       			},

     

       110
       110
       +
       			&cli.BoolFlag{

     

       111
       111
       +
       				Name:    "s3-blobstore-enabled",

     

       112
       112
       +
       				EnvVars: []string{"COCOON_S3_BLOBSTORE_ENABLED"},

     

       113
       113
       +
       			},

     

       110
       114
        
       			&cli.StringFlag{

     

       111
       115
        
       				Name:    "s3-region",

     

       112
       116
        
       				EnvVars: []string{"COCOON_S3_REGION"},

     
···

       181
       185
        
       			SmtpEmail:       cmd.String("smtp-email"),

     

       182
       186
        
       			SmtpName:        cmd.String("smtp-name"),

     

       183
       187
        
       			S3Config: &server.S3Config{

     

       184
       184
       -
       				BackupsEnabled: cmd.Bool("s3-backups-enabled"),

     

       185
       185
       -
       				Region:         cmd.String("s3-region"),

     

       186
       186
       -
       				Bucket:         cmd.String("s3-bucket"),

     

       187
       187
       -
       				Endpoint:       cmd.String("s3-endpoint"),

     

       188
       188
       -
       				AccessKey:      cmd.String("s3-access-key"),

     

       189
       189
       -
       				SecretKey:      cmd.String("s3-secret-key"),

     

       188
       188
       +
       				BackupsEnabled:   cmd.Bool("s3-backups-enabled"),

     

       189
       189
       +
       				BlobstoreEnabled: cmd.Bool("s3-blobstore-enabled"),

     

       190
       190
       +
       				Region:           cmd.String("s3-region"),

     

       191
       191
       +
       				Bucket:           cmd.String("s3-bucket"),

     

       192
       192
       +
       				Endpoint:         cmd.String("s3-endpoint"),

     

       193
       193
       +
       				AccessKey:        cmd.String("s3-access-key"),

     

       194
       194
       +
       				SecretKey:        cmd.String("s3-secret-key"),

     

       190
       195
        
       			},

     

       191
       196
        
       			SessionSecret:     cmd.String("session-secret"),

     

       192
       197
        
       			BlockstoreVariant: server.MustReturnBlockstoreVariant(cmd.String("blockstore-variant")),

models/models.go

···

       106
       106
        
       	Did       string `gorm:"index;index:idx_blob_did_cid"`

     

       107
       107
        
       	Cid       []byte `gorm:"index;index:idx_blob_did_cid"`

     

       108
       108
        
       	RefCount  int

     

       109
       109
       +
       	Storage   string `gorm:"default:sqlite;check:storage in ('sqlite', 's3')"`

     

       109
       110
        
       }

     

       110
       111
        
       

     

       111
       112
        
       type BlobPart struct {

+50 -8

server/handle_repo_upload_blob.go

···

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
        
       	"bytes"

     

       5
       5
       +
       	"fmt"

     

       5
       6
        
       	"io"

     

       6
       7
        
       

     

       8
       8
       +
       	"github.com/aws/aws-sdk-go/aws"

     

       9
       9
       +
       	"github.com/aws/aws-sdk-go/aws/credentials"

     

       10
       10
       +
       	"github.com/aws/aws-sdk-go/aws/session"

     

       11
       11
       +
       	"github.com/aws/aws-sdk-go/service/s3"

     

       7
       12
        
       	"github.com/haileyok/cocoon/internal/helpers"

     

       8
       13
        
       	"github.com/haileyok/cocoon/models"

     

       9
       14
        
       	"github.com/ipfs/go-cid"

     
···

       34
       39
        
       		mime = "application/octet-stream"

     

       35
       40
        
       	}

     

       36
       41
        
       

     

       42
       42
       +
       	storage := "sqlite"

     

       43
       43
       +
       	s3Upload := s.s3Config != nil && s.s3Config.BlobstoreEnabled

     

       44
       44
       +
       	if s3Upload {

     

       45
       45
       +
       		storage = "s3"

     

       46
       46
       +
       	}

     

       37
       47
        
       	blob := models.Blob{

     

       38
       48
        
       		Did:       urepo.Repo.Did,

     

       39
       49
        
       		RefCount:  0,

     

       40
       50
        
       		CreatedAt: s.repoman.clock.Next().String(),

     

       51
       51
       +
       		Storage:   storage,

     

       41
       52
        
       	}

     

       42
       53
        
       

     

       43
       54
        
       	if err := s.db.Create(&blob, nil).Error; err != nil {

     
···

       66
       77
        
       		read += n

     

       67
       78
        
       		fulldata.Write(data)

     

       68
       79
        
       

     

       69
       69
       -
       		blobPart := models.BlobPart{

     

       70
       70
       -
       			BlobID: blob.ID,

     

       71
       71
       -
       			Idx:    part,

     

       72
       72
       -
       			Data:   data,

     

       73
       73
       -
       		}

     

       80
       80
       +
       		if !s3Upload {

     

       81
       81
       +
       			blobPart := models.BlobPart{

     

       82
       82
       +
       				BlobID: blob.ID,

     

       83
       83
       +
       				Idx:    part,

     

       84
       84
       +
       				Data:   data,

     

       85
       85
       +
       			}

     

       74
       86
        
       

     

       75
       75
       -
       		if err := s.db.Create(&blobPart, nil).Error; err != nil {

     

       76
       76
       -
       			s.logger.Error("error adding blob part to db", "error", err)

     

       77
       77
       -
       			return helpers.ServerError(e, nil)

     

       87
       87
       +
       			if err := s.db.Create(&blobPart, nil).Error; err != nil {

     

       88
       88
       +
       				s.logger.Error("error adding blob part to db", "error", err)

     

       89
       89
       +
       				return helpers.ServerError(e, nil)

     

       90
       90
       +
       			}

     

       78
       91
        
       		}

     

       79
       92
        
       		part++

     

       80
       93
        
       

     
···

       87
       100
        
       	if err != nil {

     

       88
       101
        
       		s.logger.Error("error creating cid prefix", "error", err)

     

       89
       102
        
       		return helpers.ServerError(e, nil)

     

       103
       103
       +
       	}

     

       104
       104
       +
       

     

       105
       105
       +
       	if s3Upload {

     

       106
       106
       +
       		config := &aws.Config{

     

       107
       107
       +
       			Region:      aws.String(s.s3Config.Region),

     

       108
       108
       +
       			Credentials: credentials.NewStaticCredentials(s.s3Config.AccessKey, s.s3Config.SecretKey, ""),

     

       109
       109
       +
       		}

     

       110
       110
       +
       

     

       111
       111
       +
       		if s.s3Config.Endpoint != "" {

     

       112
       112
       +
       			config.Endpoint = aws.String(s.s3Config.Endpoint)

     

       113
       113
       +
       			config.S3ForcePathStyle = aws.Bool(true)

     

       114
       114
       +
       		}

     

       115
       115
       +
       

     

       116
       116
       +
       		sess, err := session.NewSession(config)

     

       117
       117
       +
       		if err != nil {

     

       118
       118
       +
       			s.logger.Error("error creating aws session", "error", err)

     

       119
       119
       +
       			return helpers.ServerError(e, nil)

     

       120
       120
       +
       		}

     

       121
       121
       +
       

     

       122
       122
       +
       		svc := s3.New(sess)

     

       123
       123
       +
       

     

       124
       124
       +
       		if _, err := svc.PutObject(&s3.PutObjectInput{

     

       125
       125
       +
       			Bucket: aws.String(s.s3Config.Bucket),

     

       126
       126
       +
       			Key:    aws.String(fmt.Sprintf("blobs/%s/%s", urepo.Repo.Did, c.String())),

     

       127
       127
       +
       			Body:   bytes.NewReader(fulldata.Bytes()),

     

       128
       128
       +
       		}); err != nil {

     

       129
       129
       +
       			s.logger.Error("error uploading blob to s3", "error", err)

     

       130
       130
       +
       			return helpers.ServerError(e, nil)

     

       131
       131
       +
       		}

     

       90
       132
        
       	}

     

       91
       133
        
       

     

       92
       134
        
       	if err := s.db.Exec("UPDATE blobs SET cid = ? WHERE id = ?", nil, c.Bytes(), blob.ID).Error; err != nil {

+65 -8

server/handle_sync_get_blob.go

···

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
        
       	"bytes"

     

       5
       5
       +
       	"fmt"

     

       6
       6
       +
       	"io"

     

       5
       7
        
       

     

       6
       8
        
       	"github.com/Azure/go-autorest/autorest/to"

     

       9
       9
       +
       	"github.com/aws/aws-sdk-go/aws"

     

       10
       10
       +
       	"github.com/aws/aws-sdk-go/aws/credentials"

     

       11
       11
       +
       	"github.com/aws/aws-sdk-go/aws/session"

     

       12
       12
       +
       	"github.com/aws/aws-sdk-go/service/s3"

     

       7
       13
        
       	"github.com/haileyok/cocoon/internal/helpers"

     

       8
       14
        
       	"github.com/haileyok/cocoon/models"

     

       9
       15
        
       	"github.com/ipfs/go-cid"

     
···

       47
       53
        
       

     

       48
       54
        
       	buf := new(bytes.Buffer)

     

       49
       55
        
       

     

       50
       50
       -
       	var parts []models.BlobPart

     

       51
       51
       -
       	if err := s.db.Raw("SELECT * FROM blob_parts WHERE blob_id = ? ORDER BY idx", nil, blob.ID).Scan(&parts).Error; err != nil {

     

       52
       52
       -
       		s.logger.Error("error getting blob parts", "error", err)

     

       53
       53
       -
       		return helpers.ServerError(e, nil)

     

       54
       54
       -
       	}

     

       56
       56
       +
       	if blob.Storage == "sqlite" {

     

       57
       57
       +
       		var parts []models.BlobPart

     

       58
       58
       +
       		if err := s.db.Raw("SELECT * FROM blob_parts WHERE blob_id = ? ORDER BY idx", nil, blob.ID).Scan(&parts).Error; err != nil {

     

       59
       59
       +
       			s.logger.Error("error getting blob parts", "error", err)

     

       60
       60
       +
       			return helpers.ServerError(e, nil)

     

       61
       61
       +
       		}

     

       62
       62
       +
       

     

       63
       63
       +
       		// TODO: we can just stream this, don't need to make a buffer

     

       64
       64
       +
       		for _, p := range parts {

     

       65
       65
       +
       			buf.Write(p.Data)

     

       66
       66
       +
       		}

     

       67
       67
       +
       	} else if blob.Storage == "s3" && s.s3Config != nil && s.s3Config.BlobstoreEnabled {

     

       68
       68
       +
       		config := &aws.Config{

     

       69
       69
       +
       			Region:      aws.String(s.s3Config.Region),

     

       70
       70
       +
       			Credentials: credentials.NewStaticCredentials(s.s3Config.AccessKey, s.s3Config.SecretKey, ""),

     

       71
       71
       +
       		}

     

       72
       72
       +
       

     

       73
       73
       +
       		if s.s3Config.Endpoint != "" {

     

       74
       74
       +
       			config.Endpoint = aws.String(s.s3Config.Endpoint)

     

       75
       75
       +
       			config.S3ForcePathStyle = aws.Bool(true)

     

       76
       76
       +
       		}

     

       77
       77
       +
       

     

       78
       78
       +
       		sess, err := session.NewSession(config)

     

       79
       79
       +
       		if err != nil {

     

       80
       80
       +
       			s.logger.Error("error creating aws session", "error", err)

     

       81
       81
       +
       			return helpers.ServerError(e, nil)

     

       82
       82
       +
       		}

     

       83
       83
       +
       

     

       84
       84
       +
       		svc := s3.New(sess)

     

       85
       85
       +
       		if result, err := svc.GetObject(&s3.GetObjectInput{

     

       86
       86
       +
       			Bucket: aws.String(s.s3Config.Bucket),

     

       87
       87
       +
       			Key:    aws.String(fmt.Sprintf("blobs/%s/%s", urepo.Repo.Did, c.String())),

     

       88
       88
       +
       		}); err != nil {

     

       89
       89
       +
       			s.logger.Error("error getting blob from s3", "error", err)

     

       90
       90
       +
       			return helpers.ServerError(e, nil)

     

       91
       91
       +
       		} else {

     

       92
       92
       +
       			read := 0

     

       93
       93
       +
       			part := 0

     

       94
       94
       +
       			partBuf := make([]byte, 0x10000)

     

       55
       95
        
       

     

       56
       56
       -
       	// TODO: we can just stream this, don't need to make a buffer

     

       57
       57
       -
       	for _, p := range parts {

     

       58
       58
       -
       		buf.Write(p.Data)

     

       96
       96
       +
       			for {

     

       97
       97
       +
       				n, err := io.ReadFull(result.Body, partBuf)

     

       98
       98
       +
       				if err == io.ErrUnexpectedEOF || err == io.EOF {

     

       99
       99
       +
       					if n == 0 {

     

       100
       100
       +
       						break

     

       101
       101
       +
       					}

     

       102
       102
       +
       				} else if err != nil && err != io.ErrUnexpectedEOF {

     

       103
       103
       +
       					s.logger.Error("error reading blob", "error", err)

     

       104
       104
       +
       					return helpers.ServerError(e, nil)

     

       105
       105
       +
       				}

     

       106
       106
       +
       

     

       107
       107
       +
       				data := partBuf[:n]

     

       108
       108
       +
       				read += n

     

       109
       109
       +
       				buf.Write(data)

     

       110
       110
       +
       				part++

     

       111
       111
       +
       			}

     

       112
       112
       +
       		}

     

       113
       113
       +
       	} else {

     

       114
       114
       +
       		s.logger.Error("unknown storage", "storage", blob.Storage)

     

       115
       115
       +
       		return helpers.ServerError(e, nil)

     

       59
       116
        
       	}

     

       60
       117
        
       

     

       61
       118
        
       	e.Response().Header().Set(echo.HeaderContentDisposition, "attachment; filename="+c.String())

+11 -31

server/handle_sync_subscribe_repos.go

···

       1
       1
        
       package server

     

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
       -
       	"context"

     

       5
       5
       -
       	"time"

     

       4
       4
       +
       	"fmt"

     

       6
       5
        
       

     

       7
       6
        
       	"github.com/bluesky-social/indigo/events"

     

       8
       7
        
       	"github.com/bluesky-social/indigo/lex/util"

     
···

       11
       10
        
       )

     

       12
       11
        
       

     

       13
       12
        
       func (s *Server) handleSyncSubscribeRepos(e echo.Context) error {

     

       14
       14
       -
       	ctx := e.Request().Context()

     

       15
       15
       -
       	logger := s.logger.With("component", "subscribe-repos-websocket")

     

       16
       16
       -
       

     

       17
       13
        
       	conn, err := websocket.Upgrade(e.Response().Writer, e.Request(), e.Response().Header(), 1<<10, 1<<10)

     

       18
       14
        
       	if err != nil {

     

       19
       19
       -
       		logger.Error("unable to establish websocket with relay", "err", err)

     

       20
       15
        
       		return err

     

       21
       16
        
       	}

     

       17
       17
       +
       

     

       18
       18
       +
       	s.logger.Info("new connection", "ua", e.Request().UserAgent())

     

       19
       19
       +
       

     

       20
       20
       +
       	ctx := e.Request().Context()

     

       22
       21
        
       

     

       23
       22
        
       	ident := e.RealIP() + "-" + e.Request().UserAgent()

     

       24
       24
       -
       	logger = logger.With("ident", ident)

     

       25
       25
       -
       	logger.Info("new connection established")

     

       26
       23
        
       

     

       27
       24
        
       	evts, cancel, err := s.evtman.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool {

     

       28
       25
        
       		return true

     
···

       36
       33
        
       	for evt := range evts {

     

       37
       34
        
       		wc, err := conn.NextWriter(websocket.BinaryMessage)

     

       38
       35
        
       		if err != nil {

     

       39
       39
       -
       			logger.Error("error writing message to relay", "err", err)

     

       40
       40
       -
       			break

     

       41
       41
       -
       		}

     

       42
       42
       -
       

     

       43
       43
       -
       		if ctx.Err() != nil {

     

       44
       44
       -
       			logger.Error("context error", "err", err)

     

       45
       45
       -
       			break

     

       36
       36
       +
       			return err

     

       46
       37
        
       		}

     

       47
       38
        
       

     

       48
       39
        
       		var obj util.CBOR

     

       40
       40
       +
       

     

       49
       41
        
       		switch {

     

       50
       42
        
       		case evt.Error != nil:

     

       51
       43
        
       			header.Op = events.EvtKindErrorFrame

     
···

       63
       55
        
       			header.MsgType = "#info"

     

       64
       56
        
       			obj = evt.RepoInfo

     

       65
       57
        
       		default:

     

       66
       66
       -
       			logger.Warn("unrecognized event kind")

     

       67
       67
       -
       			return nil

     

       58
       58
       +
       			return fmt.Errorf("unrecognized event kind")

     

       68
       59
        
       		}

     

       69
       60
        
       

     

       70
       61
        
       		if err := header.MarshalCBOR(wc); err != nil {

     

       71
       71
       -
       			logger.Error("failed to write header to relay", "err", err)

     

       72
       72
       -
       			break

     

       62
       62
       +
       			return fmt.Errorf("failed to write header: %w", err)

     

       73
       63
        
       		}

     

       74
       64
        
       

     

       75
       65
        
       		if err := obj.MarshalCBOR(wc); err != nil {

     

       76
       76
       -
       			logger.Error("failed to write event to relay", "err", err)

     

       77
       77
       -
       			break

     

       66
       66
       +
       			return fmt.Errorf("failed to write event: %w", err)

     

       78
       67
        
       		}

     

       79
       68
        
       

     

       80
       69
        
       		if err := wc.Close(); err != nil {

     

       81
       81
       -
       			logger.Error("failed to flush-close our event write", "err", err)

     

       82
       82
       -
       			break

     

       70
       70
       +
       			return fmt.Errorf("failed to flush-close our event write: %w", err)

     

       83
       71
        
       		}

     

       84
       84
       -
       	}

     

       85
       85
       -
       

     

       86
       86
       -
       	// we should tell the relay to request a new crawl at this point if we got disconnected

     

       87
       87
       -
       	// use a new context since the old one might be cancelled at this point

     

       88
       88
       -
       	ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)

     

       89
       89
       -
       	defer cancel()

     

       90
       90
       -
       	if err := s.requestCrawl(ctx); err != nil {

     

       91
       91
       -
       		logger.Error("error requesting crawls", "err", err)

     

       92
       72
        
       	}

     

       93
       73
        
       

     

       94
       74
        
       	return nil

+12 -42

server/server.go

···

       52
       52
        
       )

     

       53
       53
        
       

     

       54
       54
        
       type S3Config struct {

     

       55
       55
       -
       	BackupsEnabled bool

     

       56
       56
       -
       	Endpoint       string

     

       57
       57
       -
       	Region         string

     

       58
       58
       -
       	Bucket         string

     

       59
       59
       -
       	AccessKey      string

     

       60
       60
       -
       	SecretKey      string

     

       55
       55
       +
       	BackupsEnabled   bool

     

       56
       56
       +
       	BlobstoreEnabled bool

     

       57
       57
       +
       	Endpoint         string

     

       58
       58
       +
       	Region           string

     

       59
       59
       +
       	Bucket           string

     

       60
       60
       +
       	AccessKey        string

     

       61
       61
       +
       	SecretKey        string

     

       61
       62
        
       }

     

       62
       63
        
       

     

       63
       64
        
       type Server struct {

     
···

       76
       77
        
       	evtman        *events.EventManager

     

       77
       78
        
       	passport      *identity.Passport

     

       78
       79
        
       	fallbackProxy string

     

       79
       79
       -
       

     

       80
       80
       -
       	lastRequestCrawl time.Time

     

       81
       81
       -
       	requestCrawlMu   sync.Mutex

     

       82
       80
        
       

     

       83
       81
        
       	dbName   string

     

       84
       82
        
       	s3Config *S3Config

     
···

       521
       519
        
       

     

       522
       520
        
       	go s.backupRoutine()

     

       523
       521
        
       

     

       524
       524
       -
       	go func() {

     

       525
       525
       -
       		if err := s.requestCrawl(ctx); err != nil {

     

       526
       526
       -
       			s.logger.Error("error requesting crawls", "err", err)

     

       527
       527
       -
       		}

     

       528
       528
       -
       	}()

     

       529
       529
       -
       

     

       530
       530
       -
       	<-ctx.Done()

     

       531
       531
       -
       

     

       532
       532
       -
       	fmt.Println("shut down")

     

       533
       533
       -
       

     

       534
       534
       -
       	return nil

     

       535
       535
       -
       }

     

       536
       536
       -
       

     

       537
       537
       -
       func (s *Server) requestCrawl(ctx context.Context) error {

     

       538
       538
       -
       	logger := s.logger.With("component", "request-crawl")

     

       539
       539
       -
       	s.requestCrawlMu.Lock()

     

       540
       540
       -
       	defer s.requestCrawlMu.Unlock()

     

       541
       541
       -
       

     

       542
       542
       -
       	logger.Info("requesting crawl with configured relays")

     

       543
       543
       -
       

     

       544
       544
       -
       	if time.Now().Sub(s.lastRequestCrawl) <= 1*time.Minute {

     

       545
       545
       -
       		return fmt.Errorf("a crawl request has already been made within the last minute")

     

       546
       546
       -
       	}

     

       547
       547
       -
       

     

       548
       522
        
       	for _, relay := range s.config.Relays {

     

       549
       549
       -
       		logger := logger.With("relay", relay)

     

       550
       550
       -
       		logger.Info("requesting crawl from relay")

     

       551
       523
        
       		cli := xrpc.Client{Host: relay}

     

       552
       552
       -
       		if err := atproto.SyncRequestCrawl(ctx, &cli, &atproto.SyncRequestCrawl_Input{

     

       524
       524
       +
       		atproto.SyncRequestCrawl(ctx, &cli, &atproto.SyncRequestCrawl_Input{

     

       553
       525
        
       			Hostname: s.config.Hostname,

     

       554
       554
       -
       		}); err != nil {

     

       555
       555
       -
       			logger.Error("error requesting crawl", "err", err)

     

       556
       556
       -
       		} else {

     

       557
       557
       -
       			logger.Info("crawl requested successfully")

     

       558
       558
       -
       		}

     

       526
       526
       +
       		})

     

       559
       527
        
       	}

     

       560
       528
        
       

     

       561
       561
       -
       	s.lastRequestCrawl = time.Now()

     

       529
       529
       +
       	<-ctx.Done()

     

       530
       530
       +
       

     

       531
       531
       +
       	fmt.Println("shut down")

     

       562
       532
        
       

     

       563
       533
        
       	return nil

     

       564
       534
        
       }

Compare changes