MultiOpQueryer - File Upload Support

[mfinley3]

Hi - I was upgrading my gateway deployment to support file uploads. I've been using MultiOpQueryer as my go to queryer for a while now and ran into the error WARNING Network Error: map[string]interface {} is not an Upload why trying to use it for file uploads. I haven't dug too far into this but this doesn't happen when using the SingleRequestQueryer.

My Query looks something like

curl --request POST \
  --url http://localhost:8000/graphql \
  --header 'content-type: multipart/form-data; boundary=---011000010111000001101001' \
  --form 'operations={ "query": "mutation($file: Upload!) { import(file: $file) { id } }", "variables": { "file": null } }' \
  --form 'map={ "0": ["variables.file"] }' \
  --form 0=@file.yaml

        factory := gateway.QueryerFactory(func(ctx *gateway.PlanningContext, url string) graphql.Queryer {
		return graphql.NewMultiOpQueryer(url, 10*time.Millisecond, 1000)
	})

	// create the gateway instance
	gw, err := gateway.New(schemas, gateway.WithQueryerFactory(&factory))
	if err != nil {
		fmt.Println("Encountered error starting gateway:", err.Error())
		os.Exit(1)
	}

Unsure if this is expected behavior - wanted to ask because it took me a little while to figure out what was going on.
I looked through the docs but hadn't seen anything about it so it seemed like it should be supported; If not though a note here would be good :) )

[AlecAivazis]

Ah darn! It definitely should be supported, sorry you ran into that.

I haven't used this exact combo myself so I missed this issue, thanks for bringing it up! I suspect the problem is the way that we take a batch of queries and send them over the wire.

[AlecAivazis]

Poking around a little closer, it seems like what's missing is the equivalent of this block in the definition for the MultiOpQueryer defined here.

Do you have some bandwidth to check this out? If not, do you mind throwing together a repo that reproduces the issue so i can test against it?

[mfinley3]

Hey Alec - Thanks for getting back to me so quickly; wasn't expecting a near immediate reply!

I'd be happy to help where I can, unfortunately my bandwidth is a bit low right now so I've uploaded a small sample of what I was doing for testing/debugging incase you wanted to get to it quicker than I could. https://github.com/mfinley3/multiopqueryerfileupload

This isn't anything blocking so no rush looking at it! I've been able to put a small work around in the gateway.QueryerFactory to detect the mutations with uploads and use the SingleRequestQueryer for now.

I'd be happy to take a further look as time frees up as well, just let me know if you start on this :). I did end up doing a small dive into the code base to at least get more familiar with it and was wondering if it was going to be necessary to iterate over each of these keys now and determine which needs to be multipart and generate that complete payload differently. https://github.com/nautilus/graphql/blob/035738b350968205b9cbd8f2a15fe08f54d0d840/queryerMultiOp.go#L94-L99
Maybe you have a better way I missed :)

[prokaktus]

@mfinley3 @AlecAivazis is there any workaround exists? I see that support file upload PR #110 already merged, but when I try to submit Upload scalar I get the same error "map[string]interface {} is not an Upload"

Solutions without MultiOpQueryer works well for me!

@mfinley3 I could try to work on solution too! Because we really need this feature 🙂

[prokaktus]

@mfinley3

I've been able to put a small work around in the gateway.QueryerFactory to detect the mutations with uploads and use the SingleRequestQueryer for now.

Could you share it, by the way? I think it is ok to use single request, while sending mutation.

[mfinley3]

Hey @prokaktus!

I still haven't had time to look back into this. If you're able to pick it up that would be great :)

Here's the workaround (still feels pretty hacky to me so no guarantees, but it does work).

This is using the example here: https://github.com/mfinley3/multiopqueryerfileupload

func main() {

	//Start the service that accepts uploads
	go StartServiceUpload()

	schemas, err := graphql.IntrospectRemoteSchemas([]string{"http://localhost:5000/query"}...)
	if err != nil {
		log.Fatal(err)
	}

	schemasWithUpload := findSchemasWithUploadMutations(schemas)

	// enable transport-level batching of requests by default only use SingleRequestQueryer if necessary
	factory := gateway.QueryerFactory(func(ctx *gateway.PlanningContext, url string) graphql.Queryer {
		// If the service (url) has file upload mutations, check to see if the request is calling one of those mutations
		// If it is, use NewSingleRequestQueryer instead of the batch Queryer
		// This is necessary because MultiOpQueryer does not support Uploads
		if mutations, ok := schemasWithUpload[url]; ok {
			for _, mutation := range mutations {
				if strings.Contains(ctx.Query, mutation) {
					return graphql.NewSingleRequestQueryer(url)
				}
			}
		}
		return graphql.NewMultiOpQueryer(url, 10*time.Millisecond, 1000)
	})

	gw, err := gateway.New(schemas, gateway.WithQueryerFactory(&factory))
	if err != nil {
		log.Fatal(err)
	}

	http.HandleFunc("/graphql", gw.GraphQLHandler)

	// start the server
	log.Println("Starting server on port 4040")
	if err = http.ListenAndServe(":4040", nil); err != nil {
		log.Fatal(err.Error())
	}
}

// findSchemasWithUploadMutations accepts the list of all schemas
// returns a map of the service urls and mutations that use the Upload Scalar
// services that do not use the upload scalar will be completely omitted from the mapping
func findSchemasWithUploadMutations(schemas []*graphql.RemoteSchema) map[string][]string {
	uploadMutations := map[string][]string{}
	for _, schema := range schemas {
		if schema.Schema != nil {
			if schema.Schema.Types != nil {
				if uploadType, ok := schema.Schema.Types[upload]; ok && uploadType.Kind == ast.Scalar {
					//Check if the Upload Scalar is used in any of the input types. If it is, keep track of those type
					typesWithUpload := map[string]struct{}{}
					for k, v := range schema.Schema.Types {
						for _, field := range v.Fields {
							if field.Type != nil && field.Type.NamedType == upload {
								typesWithUpload[k] = struct{}{}
							}
						}
					}
					if schema.Schema.Mutation != nil {
						for _, mutation := range schema.Schema.Mutation.Fields {
							for _, argument := range mutation.Arguments {
								if argument.Type != nil {
									_, ok := typesWithUpload[argument.Type.NamedType] //If the type/import has fields that are of type the Upload Scalar ||
									if argument.Type.NamedType == upload || ok {      //If the mutation uses the Upload Scalar as a variable directly
										uploadMutations[schema.URL] = append(uploadMutations[schema.URL], mutation.Name)
									}
								}
							}
						}
					}
				}
			}
		}
		if _, ok := uploadMutations[schema.URL]; ok {
			logrus.WithField("host", schema.URL).WithField("mutations", strings.Join(uploadMutations[schema.URL], " ")).Infof("Registered Schema with Uploads")
		}
	}
	return uploadMutations
}

There might be a way to simplify findSchemasWithUploadMutations but it was working so... 🙂 Hopefully this helps!
(can also share the unit test for findSchemasWithUploadMutations if anyone needs it)

Edit: Updated to include the changes made to support uploads inside of inputs: #131