api.go

package apifu

import (
	"context"
	"net/http"
	"reflect"
	"strconv"
	"sync"

	jsoniter "github.com/json-iterator/go"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"

	"github.com/ccbrown/api-fu/graphql"
)

// API is responsible for serving your API traffic. Construct an API by creating a Config, then
// calling NewAPI.
type API struct {
	schema  *graphql.Schema
	config  *Config
	logger  logrus.FieldLogger
	execute func(*graphql.Request, *RequestInfo) *graphql.Response

	graphqlWSConnectionsMutex sync.Mutex
	graphqlWSConnections      map[graphqlWSConnection]struct{}
}

func (api *API) Schema() *graphql.Schema {
	return api.schema
}

type RequestInfo struct {
	Cost int
}

func normalizeModelType(t reflect.Type) reflect.Type {
	for t.Kind() == reflect.Ptr {
		t = t.Elem()
	}
	return t
}

// NewAPI validates your schema and builds an API ready to serve requests.
func NewAPI(cfg *Config) (*API, error) {
	schema, err := cfg.graphqlSchema()
	if err != nil {
		return nil, errors.Wrap(err, "error building graphql schema")
	}
	logger := cfg.Logger
	if logger == nil {
		logger = logrus.StandardLogger()
	}
	execute := cfg.Execute
	if execute == nil {
		execute = func(r *graphql.Request, info *RequestInfo) *graphql.Response {
			return graphql.Execute(r)
		}
	}
	return &API{
		config:               cfg,
		schema:               schema,
		logger:               logger,
		execute:              execute,
		graphqlWSConnections: map[graphqlWSConnection]struct{}{},
	}, nil
}

type apiContextKeyType int

var apiContextKey apiContextKeyType

func ctxAPI(ctx context.Context) *API {
	return ctx.Value(apiContextKey).(*API)
}

type asyncResolution struct {
	Result graphql.ResolveResult
	Dest   graphql.ResolvePromise
}

type apiRequest struct {
	asyncResolutions        chan asyncResolution
	chainedAsyncResolutions map[graphql.ResolvePromise]struct{}
	batches                 map[*int]*batch
}

func (r *apiRequest) IdleHandler() {
	for {
		if len(r.batches) > 0 {
			// Go ahead and resolve all the batches.
			var wg sync.WaitGroup
			for _, b := range r.batches {
				wg.Add(1)
				b := b
				go func() {
					defer wg.Done()
					for i, result := range b.resolver(b.items) {
						b.dests[i] <- result
					}
				}()
			}
			wg.Wait()
			r.batches = map[*int]*batch{}
		} else {
			// Block until we've fully resolved something.
			resolution := <-r.asyncResolutions
			resolution.Dest <- resolution.Result
			if _, ok := r.chainedAsyncResolutions[resolution.Dest]; ok {
				delete(r.chainedAsyncResolutions, resolution.Dest)
				continue
			}
		}

		// Move along anything else that we happen to also be done resolving.
		for {
			select {
			case resolution := <-r.asyncResolutions:
				resolution.Dest <- resolution.Result
			default:
				return
			}
		}
	}
}

type apiRequestContextKeyType int

var apiRequestContextKey apiRequestContextKeyType

func ctxAPIRequest(ctx context.Context) *apiRequest {
	return ctx.Value(apiRequestContextKey).(*apiRequest)
}

func chain(ctx context.Context, p graphql.ResolvePromise, f func(interface{}) (interface{}, error)) graphql.ResolvePromise {
	apiRequest := ctxAPIRequest(ctx)
	if apiRequest.chainedAsyncResolutions == nil {
		apiRequest.chainedAsyncResolutions = map[graphql.ResolvePromise]struct{}{}
	}
	apiRequest.chainedAsyncResolutions[p] = struct{}{}
	return Go(ctx, func() (interface{}, error) {
		result := <-p
		if !isNil(result.Error) {
			return nil, result.Error
		}
		return f(result.Value)
	})
}

func join(ctx context.Context, p []graphql.ResolvePromise, f func([]interface{}) (interface{}, error)) graphql.ResolvePromise {
	apiRequest := ctxAPIRequest(ctx)
	if apiRequest.chainedAsyncResolutions == nil {
		apiRequest.chainedAsyncResolutions = map[graphql.ResolvePromise]struct{}{}
	}
	for _, p := range p {
		apiRequest.chainedAsyncResolutions[p] = struct{}{}
	}
	return Go(ctx, func() (interface{}, error) {
		values := make([]interface{}, len(p))
		for i, p := range p {
			result := <-p
			if !isNil(result.Error) {
				return nil, result.Error
			}
			values[i] = result.Value
		}
		return f(values)
	})
}

// Go completes resolution asynchronously and concurrently with any other asynchronous resolutions.
func Go(ctx context.Context, f func() (interface{}, error)) graphql.ResolvePromise {
	apiRequest := ctxAPIRequest(ctx)
	if apiRequest.asyncResolutions == nil {
		apiRequest.asyncResolutions = make(chan asyncResolution)
	}
	ch := make(graphql.ResolvePromise, 1)
	go func() {
		v, err := f()
		apiRequest.asyncResolutions <- asyncResolution{
			Result: graphql.ResolveResult{
				Value: v,
				Error: err,
			},
			Dest: ch,
		}
	}()
	return ch
}

type batch struct {
	resolver func([]graphql.FieldContext) []graphql.ResolveResult
	items    []graphql.FieldContext
	dests    []chan graphql.ResolveResult
}

// Batch batches up the resolver invocations into a single call. As queries are executed, whenever
// resolution gets "stuck", all pending batch resolvers will be triggered concurrently. Batch
// resolvers must return one result for every field context it receives.
func Batch(f func([]graphql.FieldContext) []graphql.ResolveResult) func(graphql.FieldContext) (interface{}, error) {
	var x int
	key := &x
	return func(ctx graphql.FieldContext) (interface{}, error) {
		apiRequest := ctxAPIRequest(ctx.Context)
		b, ok := apiRequest.batches[key]
		if !ok {
			b = &batch{
				resolver: f,
			}
			if apiRequest.batches == nil {
				apiRequest.batches = map[*int]*batch{}
			}
			apiRequest.batches[key] = b
		}
		ch := make(graphql.ResolvePromise, 1)
		b.items = append(b.items, ctx)
		b.dests = append(b.dests, ch)
		return ch, nil
	}
}

// ServeGraphQL serves GraphQL HTTP requests. Requests may be GET requests using query string
// parameters or POST requests with either the application/json or application/graphql content type.
func (api *API) ServeGraphQL(w http.ResponseWriter, r *http.Request) {
	ctx := context.WithValue(r.Context(), apiContextKey, api)
	apiRequest := &apiRequest{}
	ctx = context.WithValue(ctx, apiRequestContextKey, apiRequest)
	r = r.WithContext(ctx)

	req, code, err := graphql.NewRequestFromHTTP(r)
	if err != nil {
		http.Error(w, err.Error(), code)
		return
	}
	req.Schema = api.schema
	req.IdleHandler = apiRequest.IdleHandler
	if api.config.Features != nil {
		req.Features = api.config.Features(ctx)
	}

	execute := func(req *graphql.Request) *graphql.Response {
		var info RequestInfo
		if doc, errs := graphql.ParseAndValidate(req.Query, req.Schema, req.Features, req.ValidateCost(-1, &info.Cost, api.config.DefaultFieldCost)); len(errs) > 0 {
			return &graphql.Response{
				Errors: errs,
			}
		} else {
			req.Document = doc
			return api.execute(req, &info)
		}
	}
	if storage := api.config.PersistedQueryStorage; storage != nil {
		execute = PersistedQueryExtension(storage, execute)
	}

	body, err := jsoniter.Marshal(execute(req))
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Content-Length", strconv.Itoa(len(body)))
	w.Write(body)
}

func isNil(v interface{}) bool {
	if v == nil {
		return true
	}
	rv := reflect.ValueOf(v)
	return (rv.Kind() == reflect.Ptr || rv.Kind() == reflect.Interface) && rv.IsNil()
}