Make connecting via rooms possible (#77)

di/inject_adapters.go

@@ -17,6 +17,7 @@ import (
 	"github.com/planetary-social/scuttlego/service/domain/identity"
 	"github.com/planetary-social/scuttlego/service/domain/replication"
 	"github.com/planetary-social/scuttlego/service/domain/replication/ebt"
+	"github.com/planetary-social/scuttlego/service/domain/transport/boxstream"
 	"go.etcd.io/bbolt"
 )
 
@@ -98,6 +99,7 @@ func newFilesystemStorage(logger logging.Logger, config Config) (*blobs.Filesyst
 var adaptersSet = wire.NewSet(
 	adapters.NewCurrentTimeProvider,
 	wire.Bind(new(commands.CurrentTimeProvider), new(*adapters.CurrentTimeProvider)),
+	wire.Bind(new(boxstream.CurrentTimeProvider), new(*adapters.CurrentTimeProvider)),
 
 	adapters.NewBanListHasher,
 	wire.Bind(new(bolt.BanListHasher), new(*adapters.BanListHasher)),

di/inject_application.go

@@ -7,6 +7,7 @@ import (
 	"github.com/planetary-social/scuttlego/service/app/commands"
 	"github.com/planetary-social/scuttlego/service/app/queries"
 	"github.com/planetary-social/scuttlego/service/domain/replication"
+	"github.com/planetary-social/scuttlego/service/ports/pubsub"
 	portsrpc "github.com/planetary-social/scuttlego/service/ports/rpc"
 )
 
@@ -45,6 +46,9 @@ var commandsSet = wire.NewSet(
 
 	commands.NewRoomsAliasRegisterHandler,
 	commands.NewRoomsAliasRevokeHandler,
+
+	commands.NewProcessRoomAttendantEventHandler,
+	wire.Bind(new(pubsub.ProcessRoomAttendantEventHandler), new(*commands.ProcessRoomAttendantEventHandler)),
 )
 
 var queriesSet = wire.NewSet(

di/inject_networking.go

@@ -0,0 +1,31 @@
+package di
+
+import (
+	"github.com/google/wire"
+	"github.com/planetary-social/scuttlego/service/app/commands"
+	"github.com/planetary-social/scuttlego/service/app/queries"
+	"github.com/planetary-social/scuttlego/service/domain"
+	"github.com/planetary-social/scuttlego/service/domain/network"
+	"github.com/planetary-social/scuttlego/service/domain/rooms/tunnel"
+	domaintransport "github.com/planetary-social/scuttlego/service/domain/transport"
+	"github.com/planetary-social/scuttlego/service/domain/transport/boxstream"
+	"github.com/planetary-social/scuttlego/service/domain/transport/rpc"
+	portsnetwork "github.com/planetary-social/scuttlego/service/ports/network"
+)
+
+//nolint:unused
+var networkingSet = wire.NewSet(
+	domaintransport.NewPeerInitializer,
+	wire.Bind(new(portsnetwork.ServerPeerInitializer), new(*domaintransport.PeerInitializer)),
+	wire.Bind(new(network.ClientPeerInitializer), new(*domaintransport.PeerInitializer)),
+	wire.Bind(new(tunnel.ClientPeerInitializer), new(*domaintransport.PeerInitializer)),
+
+	rpc.NewConnectionIdGenerator,
+
+	boxstream.NewHandshaker,
+
+	network.NewDialer,
+	wire.Bind(new(commands.Dialer), new(*network.Dialer)),
+	wire.Bind(new(queries.Dialer), new(*network.Dialer)),
+	wire.Bind(new(domain.Dialer), new(*network.Dialer)),
+)

di/inject_ports.go

@@ -27,6 +27,8 @@ var portsSet = wire.NewSet(
 
 	portspubsub.NewRequestSubscriber,
 
+	portspubsub.NewRoomAttendantEventSubscriber,
+
 	local.NewDiscoverer,
 	portsnetwork.NewDiscoverer,
 	portsnetwork.NewConnectionEstablisher,

di/inject_pubsub.go

@@ -5,6 +5,7 @@ import (
 	"github.com/planetary-social/scuttlego/service/adapters/pubsub"
 	"github.com/planetary-social/scuttlego/service/app/queries"
 	blobReplication "github.com/planetary-social/scuttlego/service/domain/blobs/replication"
+	"github.com/planetary-social/scuttlego/service/domain/rooms"
 	"github.com/planetary-social/scuttlego/service/domain/transport/rpc"
 )
 
@@ -13,6 +14,7 @@ var pubSubSet = wire.NewSet(
 	requestPubSubSet,
 	messagePubSubSet,
 	blobDownloadedPubSubSet,
+	roomAttendantEventPubSubSet,
 )
 
 var requestPubSubSet = wire.NewSet(
@@ -30,3 +32,8 @@ var blobDownloadedPubSubSet = wire.NewSet(
 	wire.Bind(new(queries.BlobDownloadedSubscriber), new(*pubsub.BlobDownloadedPubSub)),
 	wire.Bind(new(blobReplication.BlobDownloadedPublisher), new(*pubsub.BlobDownloadedPubSub)),
 )
+
+var roomAttendantEventPubSubSet = wire.NewSet(
+	pubsub.NewRoomAttendantEventPubSub,
+	wire.Bind(new(rooms.AttendantEventPublisher), new(*pubsub.RoomAttendantEventPubSub)),
+)

di/service.go

@@ -16,13 +16,14 @@ import (
 type Service struct {
 	App app.Application
 
-	listener                   *networkport.Listener
-	discoverer                 *networkport.Discoverer
-	connectionEstablisher      *networkport.ConnectionEstablisher
-	requestSubscriber          *pubsubport.RequestSubscriber
-	advertiser                 *local.Advertiser
-	messageBuffer              *commands.MessageBuffer
-	createHistoryStreamHandler *queries.CreateHistoryStreamHandler
+	listener                     *networkport.Listener
+	discoverer                   *networkport.Discoverer
+	connectionEstablisher        *networkport.ConnectionEstablisher
+	requestSubscriber            *pubsubport.RequestSubscriber
+	roomAttendantEventSubscriber *pubsubport.RoomAttendantEventSubscriber
+	advertiser                   *local.Advertiser
+	messageBuffer                *commands.MessageBuffer
+	createHistoryStreamHandler   *queries.CreateHistoryStreamHandler
 }
 
 func NewService(
@@ -31,20 +32,22 @@ func NewService(
 	discoverer *networkport.Discoverer,
 	connectionEstablisher *networkport.ConnectionEstablisher,
 	requestSubscriber *pubsubport.RequestSubscriber,
+	roomAttendantEventSubscriber *pubsubport.RoomAttendantEventSubscriber,
 	advertiser *local.Advertiser,
 	messageBuffer *commands.MessageBuffer,
 	createHistoryStreamHandler *queries.CreateHistoryStreamHandler,
 ) Service {
 	return Service{
 		App: app,
 
-		listener:                   listener,
-		discoverer:                 discoverer,
-		connectionEstablisher:      connectionEstablisher,
-		requestSubscriber:          requestSubscriber,
-		advertiser:                 advertiser,
-		messageBuffer:              messageBuffer,
-		createHistoryStreamHandler: createHistoryStreamHandler,
+		listener:                     listener,
+		discoverer:                   discoverer,
+		connectionEstablisher:        connectionEstablisher,
+		requestSubscriber:            requestSubscriber,
+		roomAttendantEventSubscriber: roomAttendantEventSubscriber,
+		advertiser:                   advertiser,
+		messageBuffer:                messageBuffer,
+		createHistoryStreamHandler:   createHistoryStreamHandler,
 	}
 }
 
@@ -65,6 +68,11 @@ func (s Service) Run(ctx context.Context) error {
 		errCh <- s.requestSubscriber.Run(ctx)
 	}()
 
+	runners++
+	go func() {
+		errCh <- s.roomAttendantEventSubscriber.Run(ctx)
+	}()
+
 	runners++
 	go func() {
 		errCh <- s.advertiser.Run(ctx)

di/wire.go

@@ -24,15 +24,13 @@ import (
 	"github.com/planetary-social/scuttlego/service/domain/feeds/formats"
 	"github.com/planetary-social/scuttlego/service/domain/graph"
 	"github.com/planetary-social/scuttlego/service/domain/identity"
-	"github.com/planetary-social/scuttlego/service/domain/network"
+	domainmocks "github.com/planetary-social/scuttlego/service/domain/mocks"
 	"github.com/planetary-social/scuttlego/service/domain/network/local"
 	"github.com/planetary-social/scuttlego/service/domain/replication"
 	"github.com/planetary-social/scuttlego/service/domain/replication/ebt"
 	"github.com/planetary-social/scuttlego/service/domain/replication/gossip"
-	domaintransport "github.com/planetary-social/scuttlego/service/domain/transport"
-	"github.com/planetary-social/scuttlego/service/domain/transport/boxstream"
-	"github.com/planetary-social/scuttlego/service/domain/transport/rpc"
-	portsnetwork "github.com/planetary-social/scuttlego/service/ports/network"
+	"github.com/planetary-social/scuttlego/service/domain/rooms"
+	"github.com/planetary-social/scuttlego/service/domain/rooms/tunnel"
 	"go.etcd.io/bbolt"
 )
 
@@ -95,10 +93,12 @@ func BuildTestAdapters(*bbolt.DB) (TestAdapters, error) {
 }
 
 type TestCommands struct {
-	RoomsAliasRegister *commands.RoomsAliasRegisterHandler
-	RoomsAliasRevoke   *commands.RoomsAliasRevokeHandler
+	RoomsAliasRegister        *commands.RoomsAliasRegisterHandler
+	RoomsAliasRevoke          *commands.RoomsAliasRevokeHandler
+	ProcessRoomAttendantEvent *commands.ProcessRoomAttendantEventHandler
 
-	Dialer *mocks.DialerMock
+	PeerManager *domainmocks.PeerManagerMock
+	Dialer      *mocks.DialerMock
 }
 
 func BuildTestCommands(*testing.T) (TestCommands, error) {
@@ -108,6 +108,9 @@ func BuildTestCommands(*testing.T) (TestCommands, error) {
 		mocks.NewDialerMock,
 		wire.Bind(new(commands.Dialer), new(*mocks.DialerMock)),
 
+		domainmocks.NewPeerManagerMock,
+		wire.Bind(new(commands.PeerManager), new(*domainmocks.PeerManagerMock)),
+
 		identity.NewPrivate,
 
 		wire.Struct(new(TestCommands), "*"),
@@ -122,7 +125,7 @@ type TestQueries struct {
 	FeedRepository       *mocks.FeedRepositoryMock
 	MessagePubSub        *mocks.MessagePubSubMock
 	MessageRepository    *mocks.MessageRepositoryMock
-	PeerManager          *mocks.PeerManagerMock
+	PeerManager          *domainmocks.PeerManagerMock
 	BlobStorage          *mocks.BlobStorageMock
 	ReceiveLogRepository *mocks.ReceiveLogRepositoryMock
 	Dialer               *mocks.DialerMock
@@ -139,8 +142,8 @@ func BuildTestQueries(*testing.T) (TestQueries, error) {
 		mocks.NewMessagePubSubMock,
 		wire.Bind(new(queries.MessageSubscriber), new(*mocks.MessagePubSubMock)),
 
-		mocks.NewPeerManagerMock,
-		wire.Bind(new(queries.PeerManager), new(*mocks.PeerManagerMock)),
+		domainmocks.NewPeerManagerMock,
+		wire.Bind(new(queries.PeerManager), new(*domainmocks.PeerManagerMock)),
 
 		identity.NewPrivate,
 		privateIdentityToPublicIdentity,
@@ -197,21 +200,6 @@ func BuildService(context.Context, identity.Private, Config) (Service, error) {
 	wire.Build(
 		NewService,
 
-		extractFromConfigSet,
-
-		boxstream.NewHandshaker,
-
-		domaintransport.NewPeerInitializer,
-		wire.Bind(new(portsnetwork.ServerPeerInitializer), new(*domaintransport.PeerInitializer)),
-		wire.Bind(new(network.ClientPeerInitializer), new(*domaintransport.PeerInitializer)),
-
-		rpc.NewConnectionIdGenerator,
-
-		network.NewDialer,
-		wire.Bind(new(commands.Dialer), new(*network.Dialer)),
-		wire.Bind(new(queries.Dialer), new(*network.Dialer)),
-		wire.Bind(new(domain.Dialer), new(*network.Dialer)),
-
 		domain.NewPeerManager,
 		wire.Bind(new(commands.NewPeerHandler), new(*domain.PeerManager)),
 		wire.Bind(new(commands.PeerManager), new(*domain.PeerManager)),
@@ -221,11 +209,23 @@ func BuildService(context.Context, identity.Private, Config) (Service, error) {
 		wire.Bind(new(commands.TransactionProvider), new(*bolt.TransactionProvider)),
 		newAdaptersFactory,
 
+		newBolt,
+
 		newAdvertiser,
 		privateIdentityToPublicIdentity,
 
 		commands.NewMessageBuffer,
 
+		rooms.NewScanner,
+		wire.Bind(new(domain.RoomScanner), new(*rooms.Scanner)),
+
+		rooms.NewPeerRPCAdapter,
+		wire.Bind(new(rooms.MetadataGetter), new(*rooms.PeerRPCAdapter)),
+		wire.Bind(new(rooms.AttendantsGetter), new(*rooms.PeerRPCAdapter)),
+
+		tunnel.NewDialer,
+		wire.Bind(new(domain.RoomDialer), new(*tunnel.Dialer)),
+
 		portsSet,
 		applicationSet,
 		replicatorSet,
@@ -235,8 +235,8 @@ func BuildService(context.Context, identity.Private, Config) (Service, error) {
 		boltAdaptersSet,
 		blobsAdaptersSet,
 		adaptersSet,
-
-		newBolt,
+		extractFromConfigSet,
+		networkingSet,
 	)
 	return Service{}, nil
 }