comparing 28876f4ef761606a6743531988c748f1d9837c20 and hailey/s3-blobstore on hailey.at/cocoon

+1 -1

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"`

     

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

     

       110
       110
        
       }

     

       111
       111
        
       

     

       112
       112
        
       type BlobPart struct {

+1 -6

server/handle_sync_get_blob.go

···

       64
       64
        
       		for _, p := range parts {

     

       65
       65
        
       			buf.Write(p.Data)

     

       66
       66
        
       		}

     

       67
       67
       -
       	} else if blob.Storage == "s3" {

     

       68
       68
       -
       		if  !(s.s3Config != nil && s.s3Config.BlobstoreEnabled) {

     

       69
       69
       -
       			s.logger.Error("s3 storage disabled")

     

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

     

       71
       71
       -
       		}

     

       72
       72
       -
       

     

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

     

       73
       68
        
       		config := &aws.Config{

     

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

     

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

+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

+5 -36

server/server.go

···

       78
       78
        
       	passport      *identity.Passport

     

       79
       79
        
       	fallbackProxy string

     

       80
       80
        
       

     

       81
       81
       -
       	lastRequestCrawl time.Time

     

       82
       82
       -
       	requestCrawlMu   sync.Mutex

     

       83
       83
       -
       

     

       84
       81
        
       	dbName   string

     

       85
       82
        
       	s3Config *S3Config

     

       86
       83
        
       }

     
···

       522
       519
        
       

     

       523
       520
        
       	go s.backupRoutine()

     

       524
       521
        
       

     

       525
       525
       -
       	go func() {

     

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

     

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

     

       528
       528
       -
       		}

     

       529
       529
       -
       	}()

     

       530
       530
       -
       

     

       531
       531
       -
       	<-ctx.Done()

     

       532
       532
       -
       

     

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

     

       534
       534
       -
       

     

       535
       535
       -
       	return nil

     

       536
       536
       -
       }

     

       537
       537
       -
       

     

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

     

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

     

       540
       540
       -
       	s.requestCrawlMu.Lock()

     

       541
       541
       -
       	defer s.requestCrawlMu.Unlock()

     

       542
       542
       -
       

     

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

     

       544
       544
       -
       

     

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

     

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

     

       547
       547
       -
       	}

     

       548
       548
       -
       

     

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

     

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

     

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

     

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

     

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

     

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

     

       554
       525
        
       			Hostname: s.config.Hostname,

     

       555
       555
       -
       		}); err != nil {

     

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

     

       557
       557
       -
       		} else {

     

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

     

       559
       559
       -
       		}

     

       526
       526
       +
       		})

     

       560
       527
        
       	}

     

       561
       528
        
       

     

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

     

       529
       529
       +
       	<-ctx.Done()

     

       530
       530
       +
       

     

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

     

       563
       532
        
       

     

       564
       533
        
       	return nil

     

       565
       534
        
       }

Compare changes