Signing and SIDs via Registrant, refactor tests (#112)

1. Creates a 'registrant' struct that holds the signing key and node ID:
- On creation, verify that the specified private key matches both the
registry and what is in the database
   - Issue SID's and sign payloads using the signing key and node ID
2. Refactors the tests:
   - Plumb registrant into existing tests
- Allow multiple servers to run simultaneously without overwriting each
other's state
   - Remove circular dependencies with the new mocks

.mockery.yaml

@@ -7,6 +7,7 @@ packages:
   github.com/xmtp/xmtpd/pkg/registry:
     interfaces:
       NodesContract:
+      NodeRegistry:
   github.com/xmtp/xmtpd/pkg/indexer/blockchain:
     interfaces:
       ChainClient:

cmd/replication/main.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/jessevdk/go-flags"
 	"github.com/xmtp/xmtpd/pkg/config"
+	"github.com/xmtp/xmtpd/pkg/db"
 	"github.com/xmtp/xmtpd/pkg/registry"
 	"github.com/xmtp/xmtpd/pkg/server"
 	"github.com/xmtp/xmtpd/pkg/tracing"
@@ -40,7 +41,23 @@ func main() {
 	var wg sync.WaitGroup
 	doneC := make(chan bool, 1)
 	tracing.GoPanicWrap(ctx, &wg, "main", func(ctx context.Context) {
-		s, err := server.NewReplicationServer(ctx, log, options, registry.NewFixedNodeRegistry([]registry.Node{}))
+		db, err := db.NewDB(
+			ctx,
+			options.DB.WriterConnectionString,
+			options.DB.WaitForDB,
+			options.DB.ReadTimeout,
+		)
+		if err != nil {
+			log.Fatal("initializing database", zap.Error(err))
+		}
+
+		s, err := server.NewReplicationServer(
+			ctx,
+			log,
+			options,
+			registry.NewFixedNodeRegistry([]registry.Node{}),
+			db,
+		)
 		if err != nil {
 			log.Fatal("initializing server", zap.Error(err))
 		}

pkg/api/server.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/pires/go-proxyproto"
 	"github.com/xmtp/xmtpd/pkg/proto/xmtpv4/message_api"
+	"github.com/xmtp/xmtpd/pkg/registrant"
 	"github.com/xmtp/xmtpd/pkg/tracing"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -25,22 +26,33 @@ type ApiServer struct {
 	db           *sql.DB
 	grpcListener net.Listener
 	log          *zap.Logger
+	registrant   *registrant.Registrant
 	service      message_api.ReplicationApiServer
 	wg           sync.WaitGroup
 }
 
-func NewAPIServer(ctx context.Context, writerDB *sql.DB, log *zap.Logger, port int) (*ApiServer, error) {
+func NewAPIServer(
+	ctx context.Context,
+	writerDB *sql.DB,
+	log *zap.Logger,
+	port int,
+	registrant *registrant.Registrant,
+) (*ApiServer, error) {
 	grpcListener, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%d", port))
 
 	if err != nil {
 		return nil, err
 	}
 	s := &ApiServer{
-		ctx:          ctx,
-		db:           writerDB,
-		grpcListener: &proxyproto.Listener{Listener: grpcListener, ReadHeaderTimeout: 10 * time.Second},
-		log:          log.Named("api"),
-		wg:           sync.WaitGroup{},
+		ctx: ctx,
+		db:  writerDB,
+		grpcListener: &proxyproto.Listener{
+			Listener:          grpcListener,
+			ReadHeaderTimeout: 10 * time.Second,
+		},
+		log:        log.Named("api"),
+		registrant: registrant,
+		wg:         sync.WaitGroup{},
 	}
 
 	// TODO: Add interceptors
@@ -61,7 +73,7 @@ func NewAPIServer(ctx context.Context, writerDB *sql.DB, log *zap.Logger, port i
 	healthcheck := health.NewServer()
 	healthgrpc.RegisterHealthServer(grpcServer, healthcheck)
 
-	replicationService, err := NewReplicationApiService(ctx, log, writerDB)
+	replicationService, err := NewReplicationApiService(ctx, log, registrant, writerDB)
 	if err != nil {
 		return nil, err
 	}

pkg/api/service.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/xmtp/xmtpd/pkg/db/queries"
 	"github.com/xmtp/xmtpd/pkg/proto/xmtpv4/message_api"
-	"github.com/xmtp/xmtpd/pkg/utils"
+	"github.com/xmtp/xmtpd/pkg/registrant"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
@@ -17,17 +17,19 @@ import (
 type Service struct {
 	message_api.UnimplementedReplicationApiServer
 
-	ctx     context.Context
-	log     *zap.Logger
-	queries *queries.Queries
+	ctx        context.Context
+	log        *zap.Logger
+	registrant *registrant.Registrant
+	queries    *queries.Queries
 }
 
 func NewReplicationApiService(
 	ctx context.Context,
 	log *zap.Logger,
+	registrant *registrant.Registrant,
 	writerDB *sql.DB,
 ) (*Service, error) {
-	return &Service{ctx: ctx, log: log, queries: queries.New(writerDB)}, nil
+	return &Service{ctx: ctx, log: log, registrant: registrant, queries: queries.New(writerDB)}, nil
 }
 
 func (s *Service) Close() {
@@ -74,7 +76,7 @@ func (s *Service) PublishEnvelope(
 		return nil, status.Errorf(codes.Internal, "could not insert staged envelope: %v", err)
 	}
 
-	originatorEnv, err := utils.SignStagedEnvelope(stagedEnv)
+	originatorEnv, err := s.registrant.SignStagedEnvelope(stagedEnv)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "could not sign envelope: %v", err)
 	}

pkg/api/service_test.go

@@ -5,9 +5,14 @@ import (
 	"database/sql"
 	"testing"
 
+	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/stretchr/testify/require"
+	"github.com/xmtp/xmtpd/pkg/db/queries"
+	"github.com/xmtp/xmtpd/pkg/mocks"
 	"github.com/xmtp/xmtpd/pkg/proto/identity/associations"
 	"github.com/xmtp/xmtpd/pkg/proto/xmtpv4/message_api"
+	"github.com/xmtp/xmtpd/pkg/registrant"
+	"github.com/xmtp/xmtpd/pkg/registry"
 	test "github.com/xmtp/xmtpd/pkg/testing"
 	"google.golang.org/protobuf/proto"
 )
@@ -16,8 +21,17 @@ func newTestService(t *testing.T) (*Service, *sql.DB, func()) {
 	ctx := context.Background()
 	log := test.NewLog(t)
 	db, _, dbCleanup := test.NewDB(t, ctx)
+	privKey, err := crypto.GenerateKey()
+	require.NoError(t, err)
+	privKeyStr := "0x" + test.HexEncode(crypto.FromECDSA(privKey))
+	mockRegistry := mocks.NewMockNodeRegistry(t)
+	mockRegistry.EXPECT().GetNodes().Return([]registry.Node{
+		{NodeID: 1, SigningKey: &privKey.PublicKey},
+	}, nil)
+	registrant, err := registrant.NewRegistrant(ctx, queries.New(db), mockRegistry, privKeyStr)
+	require.NoError(t, err)
 
-	svc, err := NewReplicationApiService(ctx, log, db)
+	svc, err := NewReplicationApiService(ctx, log, registrant, db)
 	require.NoError(t, err)
 
 	return svc, db, func() {

pkg/db/queries.sql

@@ -3,3 +3,11 @@ INSERT INTO staged_originator_envelopes(payer_envelope)
 	VALUES (@payer_envelope)
 RETURNING
 	*;
+
+-- name: InsertNodeInfo :one
+INSERT INTO node_info(node_id, public_key)
+	VALUES (@node_id, @public_key)
+	RETURNING *;
+
+-- name: SelectNodeInfo :one
+SELECT * FROM node_info WHERE singleton_id = 1;