badlyenginee.red

GraphQL & NATS

2023-10-06


I’ve been using gqlgen at work for a few services and while I don’t normally like code generators, it does a decent job of staying out of the way.

One thing I had hoped for was the ability to use the resolvers it generates with NATS instead of needing HTTP. I found an issue referencing this, and the gqlgen team mentioned they didn’t want to specifically support NATS because the resolvers were agnostic. So I took some time to figure out how to implement them over NATS.

This post won’t cover how to initally get started with gqlgen. I’m going to assume you have read through their documentation or have experience with it.

Resolvers

gqlgen will generate resolvers based on the queries and mutations defined in your schema. For example if your schema looked like this:

type Todo {
  id: ID!
  text: String!
  done: Boolean!
  user: User!
}
type User {
  id: ID!
  name: String!
}
type Query {
  todos: [Todo!]!
}
input NewTodo {
  text: String!
  userId: String!
}
type Mutation {
  createTodo(input: NewTodo!): Todo!
}
type Todo {
  id: ID!
  text: String!
  done: Boolean!
  user: User!
}
type User {
  id: ID!
  name: String!
}
type Query {
  todos: [Todo!]!
}
input NewTodo {
  text: String!
  userId: String!
}
type Mutation {
  createTodo(input: NewTodo!): Todo!
}

gqlgen would generate this resolver:

type Resolver struct{}
// CreateTodo is the resolver for the createTodo field.
func (r *mutationResolver) CreateTodo(ctx context.Context, input model.NewTodo) (*model.Todo, error) {
        panic(fmt.Errorf("not implemented: CreateTodo - createTodo"))
}
// Todos is the resolver for the todos field.
func (r *queryResolver) Todos(ctx context.Context) ([]*model.Todo, error) {
        panic(fmt.Errorf("not implemented: Todos - todos"))
}
// Mutation returns MutationResolver implementation.
func (r *Resolver) Mutation() MutationResolver { return &mutationResolver{r} }
// Query returns QueryResolver implementation.
func (r *Resolver) Query() QueryResolver { return &queryResolver{r} }
type mutationResolver struct{ *Resolver }
type queryResolver struct{ *Resolver }
type Resolver struct{}
// CreateTodo is the resolver for the createTodo field.
func (r *mutationResolver) CreateTodo(ctx context.Context, input model.NewTodo) (*model.Todo, error) {
        panic(fmt.Errorf("not implemented: CreateTodo - createTodo"))
}
// Todos is the resolver for the todos field.
func (r *queryResolver) Todos(ctx context.Context) ([]*model.Todo, error) {
        panic(fmt.Errorf("not implemented: Todos - todos"))
}
// Mutation returns MutationResolver implementation.
func (r *Resolver) Mutation() MutationResolver { return &mutationResolver{r} }
// Query returns QueryResolver implementation.
func (r *Resolver) Query() QueryResolver { return &queryResolver{r} }
type mutationResolver struct{ *Resolver }
type queryResolver struct{ *Resolver }

gqlgen also generates all of the boilerplate for validating the input data. It’s because of that validation that I wanted to use gqlgen’s generated resolvers. Sure, you could just add a NATS client to the resolver and that would be easy, but then you’d be on the hook for all of the validation.

Implementing the Transport Layer

gqlgen has a handler type that satisfies the ServeHTTP method. It takes the ExecutableSchemaExecutableSchema generated by gqlgen and adds transport options. So all we really have to do is the same setup using NATS instead of HTTP.

NATS Client

Let’s create a struct that contains our client and a struct for our execuable schema:

type NATSClient struct {
  Subject string
  Servers string
  Options []nats.Option
  Conn    *nats.Conn
  JS      nats.JetStreamContext
  NATSGraph
}
type NATSGraph struct {
  ExecutableSchema graphql.ExecutableSchema
  Exec             *executor.Executor
}
type NATSClient struct {
  Subject string
  Servers string
  Options []nats.Option
  Conn    *nats.Conn
  JS      nats.JetStreamContext
  NATSGraph
}
type NATSGraph struct {
  ExecutableSchema graphql.ExecutableSchema
  Exec             *executor.Executor
}

Now let’s build our constructor, connection, and resolve methods:

func NewNATSClient(subject string, servers []string, opts ...nats.Option) *NATSClient {
        n := NATSClient{
                Subject: subject,
                Servers: strings.Join(servers, ","),
        Options: opts,
        }
        return &n
}
func (n *NATSClient) SetGraphQLExecutableSchema(e graphql.ExecutableSchema) {
        ng := NATSGraph{
                ExecutableSchema: e,
                Exec:             executor.New(e),
        }
    n.NATSGraph = ng
}
func (n *NATSClient) Connect() error {
        nc, err := nats.Connect(n.Servers, n.Options...)
        if err != nil {
                return err
        }
        n.Conn = nc
        js, err := nc.JetStream()
        if err != nil {
                return err
        }
        n.JS = js
        return nil
} 
func (n *NATSClient) Resolve(errChan chan<- error) {
    // ensure the executable schema has been set
        if n.NATSGraph.ExecutableSchema == nil || n.NATSGraph.Exec == nil {
                errChan <- fmt.Errorf("executable schema must be set")
        }
        n.resolve()
}
func (n *NATSClient) resolve() {
    // ensure the subject ends in graphql
        subject := fmt.Sprintf("%s.graphql", strings.TrimSuffix(n.Subject, ".>"))
        logr.Infof("listening for requests on %s", subject)
        _, err := n.Conn.Subscribe(subject, n.HandleAndLogRequests)
        if err != nil {
                logr.Errorf("Error in subscribing: %s", err)
        }
}
func NewNATSClient(subject string, servers []string, opts ...nats.Option) *NATSClient {
        n := NATSClient{
                Subject: subject,
                Servers: strings.Join(servers, ","),
        Options: opts,
        }
        return &n
}
func (n *NATSClient) SetGraphQLExecutableSchema(e graphql.ExecutableSchema) {
        ng := NATSGraph{
                ExecutableSchema: e,
                Exec:             executor.New(e),
        }
    n.NATSGraph = ng
}
func (n *NATSClient) Connect() error {
        nc, err := nats.Connect(n.Servers, n.Options...)
        if err != nil {
                return err
        }
        n.Conn = nc
        js, err := nc.JetStream()
        if err != nil {
                return err
        }
        n.JS = js
        return nil
} 
func (n *NATSClient) Resolve(errChan chan<- error) {
    // ensure the executable schema has been set
        if n.NATSGraph.ExecutableSchema == nil || n.NATSGraph.Exec == nil {
                errChan <- fmt.Errorf("executable schema must be set")
        }
        n.resolve()
}
func (n *NATSClient) resolve() {
    // ensure the subject ends in graphql
        subject := fmt.Sprintf("%s.graphql", strings.TrimSuffix(n.Subject, ".>"))
        logr.Infof("listening for requests on %s", subject)
        _, err := n.Conn.Subscribe(subject, n.HandleAndLogRequests)
        if err != nil {
                logr.Errorf("Error in subscribing: %s", err)
        }
}

Handling The Requests

Up until this point, everything is a pretty normal NATS setup. We have a struct containing the client and some options, but other than that fairly normal. This is where the magic happens. The HandleAndLogRequestsHandleAndLogRequests method conains our logic that mimics the HTTP Do logic gqlgen has set up. This is for the most part copy pasted from their Do method with very small modifications. Then we add the natsResponsenatsResponse method to ensure we can marshal the data.

func (n *NATSClient) HandleAndLogRequests(m *nats.Msg) {
        ctx := context.Background()
        defer func() {
                if err := recover(); err != nil {
                        err := n.Exec.PresentRecoveredError(ctx, err)
                        gqlErr, _ := err.(*gqlerror.Error)
                        resp := &graphql.Response{Errors: []*gqlerror.Error{gqlErr}}
                        natsResponse(resp)
                }
        }()
        logr.Debugf("on subjeect %s, received request %+v", m.Subject, string(m.Data))
        ctx = graphql.StartOperationTrace(ctx)
        start := time.Now()
        params := &graphql.RawParams{
                ReadTime: graphql.TraceTiming{
                        Start: start,
                        End:   graphql.Now(),
                },
        }
        bodyReader := io.NopCloser(strings.NewReader(string(m.Data)))
        if err := jsonDecode(bodyReader, &params); err != nil {
                gqlErr := gqlerror.Errorf(
                        "json request body could not be decoded: %+v body:%s",
                        err,
                        string(m.Data),
                )
                resp := n.Exec.DispatchError(ctx, gqlerror.List{gqlErr})
                if err := m.RespondMsg(natsResponse(resp)); err != nil {
                        logr.Errorf("error sending message: %s", err)
                }
                return
        }
        rc, Operr := n.Exec.CreateOperationContext(ctx, params)
        if Operr != nil {
                resp := n.Exec.DispatchError(graphql.WithOperationContext(ctx, rc), Operr)
                m.RespondMsg(natsResponse(resp))
                return
        }
        var responses graphql.ResponseHandler
        responses, ctx = n.Exec.DispatchOperation(ctx, rc)
        m.RespondMsg(natsResponse(responses(ctx)))
}
func jsonDecode(r io.Reader, val interface{}) error {
        dec := json.NewDecoder(r)
        dec.UseNumber()
        return dec.Decode(val)
}
// natsResponse ensures the response can be marshaled and returned.
func natsResponse(resp *graphql.Response) *nats.Msg {
        var data []byte
        var err error
        data, err = json.Marshal(resp)
        if err != nil {
                data = []byte(`{"error": "internal server error"}`)
        }
        return &nats.Msg{
                Data: data,
        }
}
func (n *NATSClient) HandleAndLogRequests(m *nats.Msg) {
        ctx := context.Background()
        defer func() {
                if err := recover(); err != nil {
                        err := n.Exec.PresentRecoveredError(ctx, err)
                        gqlErr, _ := err.(*gqlerror.Error)
                        resp := &graphql.Response{Errors: []*gqlerror.Error{gqlErr}}
                        natsResponse(resp)
                }
        }()
        logr.Debugf("on subjeect %s, received request %+v", m.Subject, string(m.Data))
        ctx = graphql.StartOperationTrace(ctx)
        start := time.Now()
        params := &graphql.RawParams{
                ReadTime: graphql.TraceTiming{
                        Start: start,
                        End:   graphql.Now(),
                },
        }
        bodyReader := io.NopCloser(strings.NewReader(string(m.Data)))
        if err := jsonDecode(bodyReader, &params); err != nil {
                gqlErr := gqlerror.Errorf(
                        "json request body could not be decoded: %+v body:%s",
                        err,
                        string(m.Data),
                )
                resp := n.Exec.DispatchError(ctx, gqlerror.List{gqlErr})
                if err := m.RespondMsg(natsResponse(resp)); err != nil {
                        logr.Errorf("error sending message: %s", err)
                }
                return
        }
        rc, Operr := n.Exec.CreateOperationContext(ctx, params)
        if Operr != nil {
                resp := n.Exec.DispatchError(graphql.WithOperationContext(ctx, rc), Operr)
                m.RespondMsg(natsResponse(resp))
                return
        }
        var responses graphql.ResponseHandler
        responses, ctx = n.Exec.DispatchOperation(ctx, rc)
        m.RespondMsg(natsResponse(responses(ctx)))
}
func jsonDecode(r io.Reader, val interface{}) error {
        dec := json.NewDecoder(r)
        dec.UseNumber()
        return dec.Decode(val)
}
// natsResponse ensures the response can be marshaled and returned.
func natsResponse(resp *graphql.Response) *nats.Msg {
        var data []byte
        var err error
        data, err = json.Marshal(resp)
        if err != nil {
                data = []byte(`{"error": "internal server error"}`)
        }
        return &nats.Msg{
                Data: data,
        }
}

Starting Server

Now to start the NATS resolver you just need:

// backend is some backend where the data lives
backend := graph.NewBackend()
errChan := make(chan error, 1)
n := graph.NewNATSClient("my.subject", strings.Split("localhost:4222", ","))
n.SetGraphQLExecutableSchema(graph.NewExecutableSchema(graph.Config{Resolvers: &graph.Resolver{Backend: backend}}))
if err := n.Connect(); err != nil {
        logr.Fatal(err)
}
go n.Resolve(errChan)
// more here to set up HTTP or just keep go routine alive
// backend is some backend where the data lives
backend := graph.NewBackend()
errChan := make(chan error, 1)
n := graph.NewNATSClient("my.subject", strings.Split("localhost:4222", ","))
n.SetGraphQLExecutableSchema(graph.NewExecutableSchema(graph.Config{Resolvers: &graph.Resolver{Backend: backend}}))
if err := n.Connect(); err != nil {
        logr.Fatal(err)
}
go n.Resolve(errChan)
// more here to set up HTTP or just keep go routine alive

Results

Sending a mutation request returns this:

Querying returns this:

And since we are using the executable schema, we get all of the validation niceness: