inbox

Makefile

@@ -1,5 +1,5 @@
 .PHONY: build test deps build-dev
-SHELL=/bin/bash
+SHELL=/usr/bin/env bash
 export GOPRIVATE=github.com/anyproto
 export PATH:=$(CURDIR)/deps:$(PATH)
 export CGO_ENABLED:=1
@@ -28,4 +28,3 @@ deps:
 mocks:
 	echo 'Generating mocks...'
 	go generate ./...
-

cmd/coordinator/coordinator.go

@@ -35,8 +35,10 @@ import (
 	"github.com/anyproto/any-sync-coordinator/db"
 	"github.com/anyproto/any-sync-coordinator/deletionlog"
 	"github.com/anyproto/any-sync-coordinator/identityrepo"
+	"github.com/anyproto/any-sync-coordinator/inbox"
 	"github.com/anyproto/any-sync-coordinator/nodeconfsource"
 	"github.com/anyproto/any-sync-coordinator/spacestatus"
+	"github.com/anyproto/any-sync-coordinator/subscribe"
 
 	// import this to keep govvv in go.mod on mod tidy
 	_ "github.com/ahmetb/govvv/integration-test/app-different-package/mypkg"
@@ -124,6 +126,8 @@ func Bootstrap(a *app.App) {
 		Register(spacestatus.New()).
 		Register(consensusclient.New()).
 		Register(acl.New()).
+		Register(subscribe.New()).
+		Register(inbox.New()).
 		Register(accountlimit.New()).
 		Register(identityrepo.New()).
 		Register(coordinator.New()).

config/config.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/anyproto/any-sync-coordinator/accountlimit"
 	"github.com/anyproto/any-sync-coordinator/db"
+	"github.com/anyproto/any-sync-coordinator/inbox"
 	"github.com/anyproto/any-sync-coordinator/spacestatus"
 )
 
@@ -40,6 +41,7 @@ type Config struct {
 	NetworkUpdateIntervalSec int                      `yaml:"networkUpdateIntervalSec"`
 	Mongo                    db.Mongo                 `yaml:"mongo"`
 	SpaceStatus              spacestatus.Config       `yaml:"spaceStatus"`
+	Inbox                    inbox.Config             `yaml:"inbox"`
 	Yamux                    yamux.Config             `yaml:"yamux"`
 	Quic                     quic.Config              `yaml:"quic"`
 	AccountLimits            accountlimit.SpaceLimits `yaml:"defaultLimits"`
@@ -77,6 +79,10 @@ func (c Config) GetSpaceStatus() spacestatus.Config {
 	return c.SpaceStatus
 }
 
+func (c Config) GetInbox() inbox.Config {
+	return c.Inbox
+}
+
 func (c Config) GetNodeConfStorePath() string {
 	return c.NetworkStorePath
 }

coordinator/coordinator.go

@@ -30,7 +30,9 @@ import (
 	"github.com/anyproto/any-sync-coordinator/config"
 	"github.com/anyproto/any-sync-coordinator/coordinatorlog"
 	"github.com/anyproto/any-sync-coordinator/deletionlog"
+	"github.com/anyproto/any-sync-coordinator/inbox"
 	"github.com/anyproto/any-sync-coordinator/spacestatus"
+	"github.com/anyproto/any-sync-coordinator/subscribe"
 )
 
 var (
@@ -68,6 +70,8 @@ type coordinator struct {
 	deletionLog    deletionlog.DeletionLog
 	accountLimit   accountlimit.AccountLimit
 	acl            acl.AclService
+	inbox          inbox.InboxService
+	subscribe      subscribe.SubscribeService
 	drpcHandler    *rpcHandler
 	pool           pool.Service
 }
@@ -81,8 +85,10 @@ func (c *coordinator) Init(a *app.App) (err error) {
 	c.spaceStatus = a.MustComponent(spacestatus.CName).(spacestatus.SpaceStatus)
 	c.coordinatorLog = a.MustComponent(coordinatorlog.CName).(coordinatorlog.CoordinatorLog)
 	c.metric = a.MustComponent(metric.CName).(metric.Metric)
+	c.subscribe = a.MustComponent(subscribe.CName).(subscribe.SubscribeService)
 	c.deletionLog = app.MustComponent[deletionlog.DeletionLog](a)
 	c.acl = app.MustComponent[acl.AclService](a)
+	c.inbox = app.MustComponent[inbox.InboxService](a)
 	c.accountLimit = app.MustComponent[accountlimit.AccountLimit](a)
 	c.aclEventLog = app.MustComponent[acleventlog.AclEventLog](a)
 	c.pool = a.MustComponent(pool.CName).(pool.Service)
@@ -208,7 +214,7 @@ func (c *coordinator) SpaceSign(ctx context.Context, spaceId string, spaceHeader
 	if err != nil {
 		return
 	}
-	err = spacepayloads.ValidateSpaceHeader(&spacesyncproto.RawSpaceHeaderWithId{RawHeader: spaceHeader, Id: spaceId}, accountPubKey)
+	_, err = spacepayloads.ValidateSpaceHeader(&spacesyncproto.RawSpaceHeaderWithId{RawHeader: spaceHeader, Id: spaceId}, accountPubKey, nil, nil)
 	if err != nil {
 		return
 	}
@@ -465,3 +471,12 @@ func (c *coordinator) MakeSpaceUnshareable(ctx context.Context, spaceId, aclHead
 		EntryType: acleventlog.EntryTypeSpaceUnshared,
 	})
 }
+
+func (c *coordinator) InboxAddMessage(ctx context.Context, message *inbox.InboxMessage) (err error) {
+	err = c.inbox.InboxAddMessage(ctx, message)
+	return
+}
+
+func (c *coordinator) AddStream(eventType coordinatorproto.NotifyEventType, accountId, peerId string, stream coordinatorproto.DRPCCoordinator_NotifySubscribeStream) error {
+	return c.subscribe.AddStream(eventType, accountId, peerId, stream)
+}

coordinator/coordinator_test.go

@@ -33,8 +33,12 @@ import (
 	"github.com/anyproto/any-sync-coordinator/coordinatorlog/mock_coordinatorlog"
 	"github.com/anyproto/any-sync-coordinator/deletionlog"
 	"github.com/anyproto/any-sync-coordinator/deletionlog/mock_deletionlog"
+	"github.com/anyproto/any-sync-coordinator/inbox"
+	"github.com/anyproto/any-sync-coordinator/inbox/mock_inbox"
 	"github.com/anyproto/any-sync-coordinator/spacestatus"
 	"github.com/anyproto/any-sync-coordinator/spacestatus/mock_spacestatus"
+	"github.com/anyproto/any-sync-coordinator/subscribe"
+	"github.com/anyproto/any-sync-coordinator/subscribe/mock_subscribe"
 )
 
 var ctx = context.Background()
@@ -348,6 +352,8 @@ func newFixture(t *testing.T) *fixture {
 		coordLog:     mock_coordinatorlog.NewMockCoordinatorLog(ctrl),
 		aclEventLog:  mock_acleventlog.NewMockAclEventLog(ctrl),
 		deletionLog:  mock_deletionlog.NewMockDeletionLog(ctrl),
+		inbox:        mock_inbox.NewMockInboxService(ctrl),
+		subscribe:    mock_subscribe.NewMockSubscribeService(ctrl),
 		acl:          mock_acl.NewMockAclService(ctrl),
 		accountLimit: mock_accountlimit.NewMockAccountLimit(ctrl),
 		pool:         mock_pool.NewMockService(ctrl),
@@ -359,6 +365,8 @@ func newFixture(t *testing.T) *fixture {
 	anymock.ExpectComp(fx.spaceStatus.EXPECT(), spacestatus.CName)
 	anymock.ExpectComp(fx.coordLog.EXPECT(), coordinatorlog.CName)
 	anymock.ExpectComp(fx.deletionLog.EXPECT(), deletionlog.CName)
+	anymock.ExpectComp(fx.inbox.EXPECT(), inbox.CName)
+	anymock.ExpectComp(fx.subscribe.EXPECT(), subscribe.CName)
 	anymock.ExpectComp(fx.acl.EXPECT(), acl.CName)
 	anymock.ExpectComp(fx.accountLimit.EXPECT(), accountlimit.CName)
 	anymock.ExpectComp(fx.aclEventLog.EXPECT(), acleventlog.CName)
@@ -373,6 +381,8 @@ func newFixture(t *testing.T) *fixture {
 		Register(fx.aclEventLog).
 		Register(metric.New()).
 		Register(fx.deletionLog).
+		Register(fx.inbox).
+		Register(fx.subscribe).
 		Register(fx.acl).
 		Register(fx.accountLimit).
 		Register(fx.pool).
@@ -391,6 +401,8 @@ type fixture struct {
 	coordLog     *mock_coordinatorlog.MockCoordinatorLog
 	aclEventLog  *mock_acleventlog.MockAclEventLog
 	deletionLog  *mock_deletionlog.MockDeletionLog
+	subscribe    *mock_subscribe.MockSubscribeService
+	inbox        *mock_inbox.MockInboxService
 	acl          *mock_acl.MockAclService
 	accountLimit *mock_accountlimit.MockAccountLimit
 	pool         *mock_pool.MockService