Fix Connect requiring signing key in dev mode (#114)

Author: amh4r

client.go

@@ -109,6 +109,10 @@ func NewClient(opts ClientOpts) (Client, error) {
 		return nil, err
 	}
 
+	if opts.Logger == nil {
+		opts.Logger = slog.Default()
+	}
+
 	c := &apiClient{
 		ClientOpts: opts,
 	}

connect.go

@@ -79,18 +79,21 @@ func Connect(ctx context.Context, opts ConnectOpts) (connect.WorkerConnection, e
 		return nil, fmt.Errorf("invalid handler passed")
 	}
 
+	var hashedKey []byte
+	var hashedFallbackKey []byte
 	signingKey := defaultClient.h.GetSigningKey()
 	if signingKey == "" {
-		return nil, fmt.Errorf("signing key is required")
-	}
-
-	hashedKey, err := hashedSigningKey([]byte(signingKey))
-	if err != nil {
-		return nil, fmt.Errorf("failed to hash signing key: %w", err)
-	}
+		if !defaultClient.h.isDev() {
+			// Signing key is only required in cloud mode.
+			return nil, fmt.Errorf("signing key is required")
+		}
+	} else {
+		var err error
+		hashedKey, err = hashedSigningKey([]byte(signingKey))
+		if err != nil {
+			return nil, fmt.Errorf("failed to hash signing key: %w", err)
+		}
 
-	var hashedFallbackKey []byte
-	{
 		if fallbackKey := defaultClient.h.GetSigningKeyFallback(); fallbackKey != "" {
 			hashedFallbackKey, err = hashedSigningKey([]byte(fallbackKey))
 			if err != nil {

connect/handler.go

@@ -5,6 +5,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"log/slog"
+	"net/url"
+	"os"
+	"runtime"
+	"sync/atomic"
+	"time"
+
 	"github.com/coder/websocket"
 	"github.com/inngest/inngest/pkg/execution/state"
 	"github.com/inngest/inngest/pkg/sdk"
@@ -13,13 +21,6 @@ import (
 	"github.com/inngest/inngestgo/internal/sdkrequest"
 	"github.com/pbnjay/memory"
 	"golang.org/x/sync/errgroup"
-	"io"
-	"log/slog"
-	"net/url"
-	"os"
-	"runtime"
-	"sync/atomic"
-	"time"
 )
 
 const (
@@ -159,7 +160,7 @@ type authContext struct {
 
 func (h *connectHandler) Connect(ctx context.Context) (WorkerConnection, error) {
 	signingKey := h.opts.HashedSigningKey
-	if len(signingKey) == 0 {
+	if len(signingKey) == 0 && !h.opts.IsDev {
 		return nil, fmt.Errorf("hashed signing key is required")
 	}
 

connect/workerapi.go

@@ -4,11 +4,12 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"github.com/inngest/inngest/proto/gen/connect/v1"
-	"google.golang.org/protobuf/proto"
 	"io"
 	"log/slog"
 	"net/http"
+
+	"github.com/inngest/inngest/proto/gen/connect/v1"
+	"google.golang.org/protobuf/proto"
 )
 
 type workerApiClient struct {
@@ -36,7 +37,9 @@ func (a *workerApiClient) start(ctx context.Context, hashedSigningKey []byte, re
 	}
 
 	httpReq.Header.Set("Content-Type", "application/protobuf")
-	httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", string(hashedSigningKey)))
+	if hashedSigningKey != nil {
+		httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", string(hashedSigningKey)))
+	}
 
 	if a.env != nil {
 		httpReq.Header.Add("X-Inngest-Env", *a.env)
@@ -96,7 +99,9 @@ func (a *workerApiClient) sendBufferedMessage(ctx context.Context, hashedSigning
 	}
 
 	httpReq.Header.Set("Content-Type", "application/protobuf")
-	httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", string(hashedSigningKey)))
+	if hashedSigningKey != nil {
+		httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", string(hashedSigningKey)))
+	}
 
 	if a.env != nil {
 		httpReq.Header.Add("X-Inngest-Env", *a.env)

handler.go

@@ -136,16 +136,16 @@ func (h handlerOpts) GetSigningKeyFallback() string {
 
 // GetAPIOrigin returns the host to use for sending API requests
 func (h handlerOpts) GetAPIBaseURL() string {
-	if h.isDev() {
-		return DevServerURL()
-	}
-
 	if h.APIBaseURL == nil {
 		base := os.Getenv("INNGEST_API_BASE_URL")
 		if base != "" {
 			return base
 		}
 
+		if h.isDev() {
+			return DevServerURL()
+		}
+
 		return defaultAPIOrigin
 	}
 

handler_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/inngest/inngest/pkg/syscode"
 	"github.com/inngest/inngestgo/internal/sdkrequest"
 	"github.com/inngest/inngestgo/step"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -1012,6 +1013,120 @@ func TestInBandSync(t *testing.T) {
 	})
 }
 
+func TestConnectSync(t *testing.T) {
+	t.Run("cloud", func(t *testing.T) {
+		// SDK sends an Authorization header when in cloud mode.
+
+		r := require.New(t)
+
+		// We need a cancellable context to stop the Connect worker.
+		connectCtx, cancelConnectCtx := context.WithCancel(context.Background())
+		defer cancelConnectCtx()
+
+		headers := http.Header{}
+		server := httptest.NewServer(http.HandlerFunc(
+			func(w http.ResponseWriter, r *http.Request) {
+				if r.URL.Path == "/v0/connect/start" {
+					for k, v := range r.Header {
+						headers.Add(k, v[0])
+					}
+					cancelConnectCtx()
+					w.WriteHeader(http.StatusOK)
+					return
+				}
+
+				w.WriteHeader(http.StatusNotFound)
+				_, _ = w.Write([]byte(`{}`))
+			},
+		))
+		defer server.Close()
+
+		c, err := NewClient(ClientOpts{
+			APIBaseURL: toPtr(server.URL),
+			AppID:      "app",
+			SigningKey: toPtr(testKey),
+		})
+		r.NoError(err)
+
+		_, err = CreateFunction(
+			c,
+			FunctionOpts{ID: "fn"},
+			EventTrigger("event", nil),
+			func(ctx context.Context, input Input[any]) (any, error) {
+				return nil, nil
+			},
+		)
+		r.NoError(err)
+
+		_, _ = Connect(connectCtx, ConnectOpts{
+			Apps:       []Client{c},
+			InstanceID: toPtr("instance"),
+		})
+
+		r.EventuallyWithT(func(t *assert.CollectT) {
+			a := assert.New(t)
+			a.NotEmpty(headers.Get("Authorization"))
+		}, 5*time.Second, 10*time.Millisecond)
+	})
+
+	t.Run("dev", func(t *testing.T) {
+		// SDK doesn't send an Authorization header when in dev mode.
+
+		r := require.New(t)
+
+		// We need a cancellable context to stop the Connect worker.
+		connectCtx, cancelConnectCtx := context.WithCancel(context.Background())
+		defer cancelConnectCtx()
+
+		headers := http.Header{}
+		called := false
+		server := httptest.NewServer(http.HandlerFunc(
+			func(w http.ResponseWriter, r *http.Request) {
+				if r.URL.Path == "/v0/connect/start" {
+					for k, v := range r.Header {
+						headers.Add(k, v[0])
+					}
+					called = true
+					cancelConnectCtx()
+					w.WriteHeader(http.StatusOK)
+					return
+				}
+
+				w.WriteHeader(http.StatusNotFound)
+			},
+		))
+		defer server.Close()
+
+		c, err := NewClient(ClientOpts{
+			APIBaseURL: toPtr(server.URL),
+			AppID:      "app",
+			Dev:        toPtr(true),
+		})
+		r.NoError(err)
+
+		_, err = CreateFunction(
+			c,
+			FunctionOpts{ID: "fn"},
+			EventTrigger("event", nil),
+			func(ctx context.Context, input Input[any]) (any, error) {
+				return nil, nil
+			},
+		)
+		r.NoError(err)
+
+		_, _ = Connect(connectCtx, ConnectOpts{
+			Apps:       []Client{c},
+			InstanceID: toPtr("instance"),
+		})
+
+		r.EventuallyWithT(func(t *assert.CollectT) {
+			a := assert.New(t)
+			a.Empty(headers.Get("Authorization"))
+			a.True(called)
+		}, 5*time.Second, 10*time.Millisecond)
+	})
+}
+
 func createRequest(t *testing.T, evt any) *sdkrequest.Request {
 	t.Helper()
 

tests/invoke_test.go

@@ -15,9 +15,7 @@ import (
 )
 
 func TestInvoke(t *testing.T) {
-	if testing.Short() {
-		t.Skip()
-	}
+	devEnv(t)
 
 	t.Run("success", func(t *testing.T) {
 		ctx := context.Background()

tests/main_test.go

@@ -29,13 +29,6 @@ func TestMain(m *testing.M) {
 }
 
 func setup() (func() error, error) {
-	os.Setenv("INNGEST_DEV", "1")
-
-	if os.Getenv("DEV_SERVER_ENABLED") == "0" {
-		// Don't start the Dev Server.
-		return func() error { return nil }, nil
-	}
-
 	stopDevServer, err := startDevServer()
 	if err != nil {
 		return nil, err
@@ -44,6 +37,11 @@ func setup() (func() error, error) {
 	return stopDevServer, nil
 }
 
+func devEnv(t *testing.T) {
+	t.Helper()
+	t.Setenv("INNGEST_DEV", "1")
+}
+
 func startDevServer() (func() error, error) {
 	fmt.Println("Starting Dev Server")
 	cmd := exec.Command(

tests/parallel_test.go

@@ -13,10 +13,7 @@ import (
 )
 
 func TestParallel(t *testing.T) {
-	if testing.Short() {
-		t.Skip()
-	}
-
+	devEnv(t)
 	t.Run("successful with a mix of step kinds", func(t *testing.T) {
 		ctx := context.Background()
 		r := require.New(t)

tests/probe_test.go

@@ -16,6 +16,8 @@ import (
 const sKey = "signkey-prod-000000"
 
 func TestTrustProbe(t *testing.T) {
+	devEnv(t)
+
 	t.Run("dev mode", func(t *testing.T) {
 		isDev := true