···

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

     

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

     

       
108
       
 
       
	RefCount  int

     

       
109
       
-
       
	Storage   string `gorm:"default:sqlite"`

     

       
110
       
 
       
}

     

       
111
       
 
       


     

       
112
       
 
       
type BlobPart struct {

     

···

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

     

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

     

       
108
       
 
       
	RefCount  int

     

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

     

       
110
       
 
       
}

     

       
111
       
 
       


     

       
112
       
 
       
type BlobPart struct {

     

···

       
64
       
 
       
		for _, p := range parts {

     

       
65
       
 
       
			buf.Write(p.Data)

     

       
66
       
 
       
		}

     

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

     

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

     

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

     

       
70
       
-
       
			return helpers.ServerError(e, nil)

     

       
71
       
-
       
		}

     

       
72
       
-
       


     

       
73
       
 
       
		config := &aws.Config{

     

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

     

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

     

···

       
64
       
 
       
		for _, p := range parts {

     

       
65
       
 
       
			buf.Write(p.Data)

     

       
66
       
 
       
		}

     

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

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
68
       
 
       
		config := &aws.Config{

     

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

     

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

     

···

       
1
       
 
       
package server

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
-
       
	"context"

     

       
5
       
-
       
	"time"

     

       
6
       
 
       


     

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

     

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

     
···

       
11
       
 
       
)

     

       
12
       
 
       


     

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

     

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

     

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

     

       
16
       
-
       


     

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

     

       
18
       
 
       
	if err != nil {

     

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

     

       
20
       
 
       
		return err

     

       
21
       
 
       
	}

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
22
       
 
       


     

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

     

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

     

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

     

       
26
       
 
       


     

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

     

       
28
       
 
       
		return true

     
···

       
36
       
 
       
	for evt := range evts {

     

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

     

       
38
       
 
       
		if err != nil {

     

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

     

       
40
       
-
       
			break

     

       
41
       
-
       
		}

     

       
42
       
-
       


     

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

     

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

     

       
45
       
-
       
			break

     

       
46
       
 
       
		}

     

       
47
       
 
       


     

       
48
       
 
       
		var obj util.CBOR

     

       
0
       
 
       

     

       
49
       
 
       
		switch {

     

       
50
       
 
       
		case evt.Error != nil:

     

       
51
       
 
       
			header.Op = events.EvtKindErrorFrame

     
···

       
63
       
 
       
			header.MsgType = "#info"

     

       
64
       
 
       
			obj = evt.RepoInfo

     

       
65
       
 
       
		default:

     

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

     

       
67
       
-
       
			return nil

     

       
68
       
 
       
		}

     

       
69
       
 
       


     

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

     

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

     

       
72
       
-
       
			break

     

       
73
       
 
       
		}

     

       
74
       
 
       


     

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

     

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

     

       
77
       
-
       
			break

     

       
78
       
 
       
		}

     

       
79
       
 
       


     

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

     

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

     

       
82
       
-
       
			break

     

       
83
       
 
       
		}

     

       
84
       
-
       
	}

     

       
85
       
-
       


     

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

     

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

     

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

     

       
89
       
-
       
	defer cancel()

     

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

     

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

     

       
92
       
 
       
	}

     

       
93
       
 
       


     

       
94
       
 
       
	return nil

     

···

       
1
       
 
       
package server

     

       
2
       
 
       


     

       
3
       
 
       
import (

     

       
4
       
+
       
	"fmt"

     

       
0
       
 
       

     

       
5
       
 
       


     

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

     

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

     
···

       
10
       
 
       
)

     

       
11
       
 
       


     

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

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

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

     

       
14
       
 
       
	if err != nil {

     

       
0
       
 
       

     

       
15
       
 
       
		return err

     

       
16
       
 
       
	}

     

       
17
       
+
       


     

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

     

       
19
       
+
       


     

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

     

       
21
       
 
       


     

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

     

       
0
       
 
       

     

       
0
       
 
       

     

       
23
       
 
       


     

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

     

       
25
       
 
       
		return true

     
···

       
33
       
 
       
	for evt := range evts {

     

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

     

       
35
       
 
       
		if err != nil {

     

       
36
       
+
       
			return err

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
37
       
 
       
		}

     

       
38
       
 
       


     

       
39
       
 
       
		var obj util.CBOR

     

       
40
       
+
       


     

       
41
       
 
       
		switch {

     

       
42
       
 
       
		case evt.Error != nil:

     

       
43
       
 
       
			header.Op = events.EvtKindErrorFrame

     
···

       
55
       
 
       
			header.MsgType = "#info"

     

       
56
       
 
       
			obj = evt.RepoInfo

     

       
57
       
 
       
		default:

     

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

     

       
0
       
 
       

     

       
59
       
 
       
		}

     

       
60
       
 
       


     

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

     

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

     

       
0
       
 
       

     

       
63
       
 
       
		}

     

       
64
       
 
       


     

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

     

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

     

       
0
       
 
       

     

       
67
       
 
       
		}

     

       
68
       
 
       


     

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

     

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

     

       
0
       
 
       

     

       
71
       
 
       
		}

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
72
       
 
       
	}

     

       
73
       
 
       


     

       
74
       
 
       
	return nil

     

···

       
78
       
 
       
	passport      *identity.Passport

     

       
79
       
 
       
	fallbackProxy string

     

       
80
       
 
       


     

       
81
       
-
       
	lastRequestCrawl time.Time

     

       
82
       
-
       
	requestCrawlMu   sync.Mutex

     

       
83
       
-
       


     

       
84
       
 
       
	dbName   string

     

       
85
       
 
       
	s3Config *S3Config

     

       
86
       
 
       
}

     
···

       
522
       
 
       


     

       
523
       
 
       
	go s.backupRoutine()

     

       
524
       
 
       


     

       
525
       
-
       
	go func() {

     

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

     

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

     

       
528
       
-
       
		}

     

       
529
       
-
       
	}()

     

       
530
       
-
       


     

       
531
       
-
       
	<-ctx.Done()

     

       
532
       
-
       


     

       
533
       
-
       
	fmt.Println("shut down")

     

       
534
       
-
       


     

       
535
       
-
       
	return nil

     

       
536
       
-
       
}

     

       
537
       
-
       


     

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

     

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

     

       
540
       
-
       
	s.requestCrawlMu.Lock()

     

       
541
       
-
       
	defer s.requestCrawlMu.Unlock()

     

       
542
       
-
       


     

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

     

       
544
       
-
       


     

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

     

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

     

       
547
       
-
       
	}

     

       
548
       
-
       


     

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

     

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

     

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

     

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

     

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

     

       
554
       
 
       
			Hostname: s.config.Hostname,

     

       
555
       
-
       
		}); err != nil {

     

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

     

       
557
       
-
       
		} else {

     

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

     

       
559
       
-
       
		}

     

       
560
       
 
       
	}

     

       
561
       
 
       


     

       
562
       
-
       
	s.lastRequestCrawl = time.Now()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
563
       
 
       


     

       
564
       
 
       
	return nil

     

       
565
       
 
       
}

     

···

       
78
       
 
       
	passport      *identity.Passport

     

       
79
       
 
       
	fallbackProxy string

     

       
80
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
81
       
 
       
	dbName   string

     

       
82
       
 
       
	s3Config *S3Config

     

       
83
       
 
       
}

     
···

       
519
       
 
       


     

       
520
       
 
       
	go s.backupRoutine()

     

       
521
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

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

     

       
0
       
 
       

     

       
0
       
 
       

     

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

     

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

     

       
525
       
 
       
			Hostname: s.config.Hostname,

     

       
526
       
+
       
		})

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
527
       
 
       
	}

     

       
528
       
 
       


     

       
529
       
+
       
	<-ctx.Done()

     

       
530
       
+
       


     

       
531
       
+
       
	fmt.Println("shut down")

     

       
532
       
 
       


     

       
533
       
 
       
	return nil

     

       
534
       
 
       
}