register a grpc-gateway to mux http request (#509)

buf.gen.yaml

@@ -3,6 +3,8 @@ managed:
   enabled: true
   go_package_prefix:
     default: generated/protos
+    except:
+      - buf.build/googleapis/googleapis
 plugins:
   - plugin: buf.build/protocolbuffers/go:v1.31.0
     out: flow/generated/protos
@@ -27,3 +29,7 @@ plugins:
     opt:
       - esModuleInterop=true
       - outputServices=grpc-js
+  - plugin: buf.build/grpc-ecosystem/gateway:v2.18.0
+    out: flow/generated/protos
+    opt:
+      - paths=source_relative

docker-compose.yml

@@ -108,6 +108,7 @@ services:
       dockerfile: stacks/flow-api.Dockerfile
     ports:
       - 8112:8112
+      - 8113:8113
     environment:
       <<: [*catalog-config]
       TEMPORAL_HOST_PORT: temporal:7233

flow/cmd/api.go

@@ -4,11 +4,15 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"net/http"
+	"time"
 
 	utils "github.com/PeerDB-io/peer-flow/connectors/utils/catalog"
 	"github.com/PeerDB-io/peer-flow/generated/protos"
+	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/reflection"
 
 	"go.temporal.io/sdk/client"
@@ -19,9 +23,37 @@ import (
 type APIServerParams struct {
 	ctx              context.Context
 	Port             uint
+	GatewayPort      uint
 	TemporalHostPort string
 }
 
+// setupGRPCGatewayServer sets up the grpc-gateway mux
+func setupGRPCGatewayServer(args *APIServerParams) (*http.Server, error) {
+	conn, err := grpc.DialContext(
+		context.Background(),
+		fmt.Sprintf("0.0.0.0:%d", args.Port),
+		grpc.WithBlock(),
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+	)
+
+	if err != nil {
+		return nil, fmt.Errorf("unable to dial grpc server: %w", err)
+	}
+
+	gwmux := runtime.NewServeMux()
+	err = protos.RegisterFlowServiceHandler(context.Background(), gwmux, conn)
+	if err != nil {
+		return nil, fmt.Errorf("unable to register gateway: %w", err)
+	}
+
+	server := &http.Server{
+		Addr:              fmt.Sprintf(":%d", args.GatewayPort),
+		Handler:           gwmux,
+		ReadHeaderTimeout: 5 * time.Minute,
+	}
+	return server, nil
+}
+
 func APIMain(args *APIServerParams) error {
 	ctx := args.ctx
 
@@ -58,6 +90,18 @@ func APIMain(args *APIServerParams) error {
 		}
 	}()
 
+	gateway, err := setupGRPCGatewayServer(args)
+	if err != nil {
+		return fmt.Errorf("unable to setup gateway server: %w", err)
+	}
+
+	log.Infof("Starting API gateway on port %d", args.GatewayPort)
+	go func() {
+		if err := gateway.ListenAndServe(); err != nil {
+			log.Fatalf("failed to serve http: %v", err)
+		}
+	}()
+
 	<-ctx.Done()
 
 	grpcServer.GracefulStop()

flow/cmd/main.go

@@ -110,6 +110,11 @@ func main() {
 						Aliases: []string{"p"},
 						Value:   8110,
 					},
+					// gateway port is the port that the grpc-gateway listens on
+					&cli.UintFlag{
+						Name:  "gateway-port",
+						Value: 8111,
+					},
 					temporalHostPortFlag,
 				},
 				Action: func(ctx *cli.Context) error {
@@ -119,6 +124,7 @@ func main() {
 						ctx:              appCtx,
 						Port:             ctx.Uint("port"),
 						TemporalHostPort: temporalHostPort,
+						GatewayPort:      ctx.Uint("gateway-port"),
 					})
 				},
 			},