Merge pull request #1 from ldebruijn/po-reloading

Add ability for Persisted Operations reloading

.gitignore

@@ -1,3 +1,4 @@
 armor.yml
 operations.json
-TODO.md
+TODO.md
+main

cmd/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"github.com/ardanlabs/conf/v3"
 	"github.com/ldebruijn/go-graphql-armor/internal/app/config"
@@ -22,8 +23,6 @@ import (
 var build = "develop"
 
 func main() {
-	ctx := context.Background()
-
 	log := slog.Default()
 
 	// cfg
@@ -40,13 +39,13 @@ func main() {
 	shutdown := make(chan os.Signal, 1)
 	signal.Notify(shutdown, syscall.SIGINT, syscall.SIGTERM)
 
-	if err := run(ctx, log, cfg, shutdown); err != nil {
+	if err := run(log, cfg, shutdown); err != nil {
 		log.Error("startup", "msg", err)
 		os.Exit(1)
 	}
 }
 
-func run(ctx context.Context, log *slog.Logger, cfg *config.Config, shutdown chan os.Signal) error {
+func run(log *slog.Logger, cfg *config.Config, shutdown chan os.Signal) error {
 	log.Info("startup", "GOMAXPROCS", runtime.GOMAXPROCS(0))
 
 	log.Info("Starting proxy", "target", cfg.Target.Host)
@@ -59,9 +58,20 @@ func run(ctx context.Context, log *slog.Logger, cfg *config.Config, shutdown cha
 		return nil
 	}
 
+	remoteLoader, err := persisted_operations.RemoteLoaderFromConfig(cfg.PersistedOperations)
+	if err != nil && !errors.Is(err, persisted_operations.ErrNoRemoteLoaderSpecified) {
+		log.Warn("Error initializing remote loader", "err", err)
+	}
+
+	po, err := persisted_operations.NewPersistedOperations(log, cfg.PersistedOperations, persisted_operations.NewLocalDirLoader(cfg.PersistedOperations), remoteLoader)
+	if err != nil {
+		log.Error("Error initializing Persisted Operations", "err", err)
+		return nil
+	}
+
 	mux := http.NewServeMux()
 
-	mid := middleware(log, cfg)
+	mid := middleware(log, po)
 	mux.Handle(cfg.Web.Path, mid(Handler(pxy)))
 
 	api := http.Server{
@@ -91,6 +101,8 @@ func run(ctx context.Context, log *slog.Logger, cfg *config.Config, shutdown cha
 		ctx, cancel := context.WithTimeout(context.Background(), cfg.Web.ShutdownTimeout)
 		defer cancel()
 
+		po.Shutdown()
+
 		if err := api.Shutdown(ctx); err != nil {
 			_ = api.Close()
 			return fmt.Errorf("could not stop server gracefully: %w", err)
@@ -100,14 +112,8 @@ func run(ctx context.Context, log *slog.Logger, cfg *config.Config, shutdown cha
 	return nil
 }
 
-func middleware(log *slog.Logger, cfg *config.Config) func(next http.Handler) http.Handler {
-	poLoader, err := persisted_operations.DetermineLoaderFromConfig(cfg.PersistedOperations)
-	if err != nil {
-		log.Error("Unable to determine loading strategy for persisted operations", "err", err)
-	}
-
+func middleware(log *slog.Logger, po *persisted_operations.PersistedOperationsHandler) func(next http.Handler) http.Handler {
 	rec := middleware2.Recover(log)
-	po, _ := persisted_operations.NewPersistedOperations(log, cfg.PersistedOperations, poLoader)
 
 	fn := func(next http.Handler) http.Handler {
 		return rec(po.Execute(next))

cmd/main_test.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"github.com/ldebruijn/go-graphql-armor/internal/app/config"
 	"github.com/stretchr/testify/assert"
@@ -41,6 +40,7 @@ func TestHttpServerIntegration(t *testing.T) {
 				}(),
 				cfgOverrides: func(cfg *config.Config) *config.Config {
 					cfg.PersistedOperations.Enabled = true
+					cfg.PersistedOperations.Store = "./"
 					return cfg
 				},
 				mockResponse: map[string]interface{}{
@@ -62,7 +62,8 @@ func TestHttpServerIntegration(t *testing.T) {
 					},
 				}
 				ex, _ := json.Marshal(expected)
-				actual, _ := io.ReadAll(response.Body)
+				actual, err := io.ReadAll(response.Body)
+				assert.NoError(t, err)
 				// perform string comparisons as map[string]interface seems incomparable
 				assert.Equal(t, string(ex), string(actual))
 			},
@@ -85,6 +86,7 @@ func TestHttpServerIntegration(t *testing.T) {
 				}(),
 				cfgOverrides: func(cfg *config.Config) *config.Config {
 					cfg.PersistedOperations.Enabled = true
+					cfg.PersistedOperations.Store = "./"
 					return cfg
 				},
 				mockResponse: map[string]interface{}{
@@ -105,7 +107,9 @@ func TestHttpServerIntegration(t *testing.T) {
 					},
 				}
 				ex, _ := json.Marshal(expected)
-				actual, _ := io.ReadAll(response.Body)
+				actual, err := io.ReadAll(response.Body)
+				assert.NoError(t, err)
+
 				ac := string(actual)
 				ac = strings.TrimSuffix(ac, "\n")
 
@@ -127,6 +131,8 @@ func TestHttpServerIntegration(t *testing.T) {
 				}(),
 				cfgOverrides: func(cfg *config.Config) *config.Config {
 					cfg.PersistedOperations.Enabled = true
+					cfg.PersistedOperations.Store = "./"
+					cfg.PersistedOperations.FailUnknownOperations = false
 					return cfg
 				},
 				mockResponse: map[string]interface{}{
@@ -160,7 +166,8 @@ func TestHttpServerIntegration(t *testing.T) {
 					},
 				}
 				ex, _ := json.Marshal(expected)
-				actual, _ := io.ReadAll(response.Body)
+				actual, err := io.ReadAll(response.Body)
+				assert.NoError(t, err)
 				// perform string comparisons as map[string]interface seems incomparable
 				assert.Equal(t, string(ex), string(actual))
 			},
@@ -184,13 +191,13 @@ func TestHttpServerIntegration(t *testing.T) {
 			cfg.Target.Host = mockServer.URL
 
 			go func() {
-				_ = run(context.Background(), slog.Default(), cfg, shutdown)
+				_ = run(slog.Default(), cfg, shutdown)
 			}()
 
 			url := "http://localhost:8080" + tt.args.request.URL.String()
 			res, err := http.Post(url, tt.args.request.Header.Get("Content-Type"), tt.args.request.Body)
 			if err != nil {
-				assert.NoError(t, err)
+				assert.NoError(t, err, tt.name)
 			}
 
 			tt.want(t, res)

docs/persisted_operations.md

@@ -20,19 +20,30 @@ persisted_operations:
   enabled: true
   # Fail unknown operations, disable this feature to allow unknown operations to reach your GraphQL API
   fail_unknown_operations: true
-  # Determines the strategy for loading the supported operations.
-  # Only one store will be used
-  store:
-    # Load persisted operations from a directory on the local filesystem. 
-    # Will look at all files in the directory and attempt to load any file with a `.json` extension
-    dir: "./my-dir"
+  # Store is the location on local disk where go-graphql-armor can find the persisted operations, it loads any `*.json` files on disk
+  store: "./store"
+  reload:
+    enabled: true
+    # The interval in which the local store dir is read and refreshes the internal state
+    interval: 5m
+    # The timeout for the remote operation
+    timeout: 10s
+  remote:
     # Load persisted operations from a GCP Cloud Storage bucket.
     # Will look at all the objects in the bucket and try to load any object with a `.json` extension
     gcp_bucket: "gs://somebucket"
 
 # ...
 ```
 
+## How it works
+
+`go-graphql-armor` looks at the `store` location on local disk to find any `*.json` files it can parse for persisted operations. 
+
+It can be configured to look at this directory and reload based on the files on local disk.
+
+Additionally, it can be configured to fetch operations from a remote location onto the local disk.
+
 ## Parsing Structure
 
 To be able to parse Persisted Operations go-graphql-armor expects a `key-value` structure for `hash-operation` in the files.

internal/business/block_field_suggestions/block_field_suggestions.go

@@ -51,13 +51,13 @@ func (b *BlockFieldSuggestionsHandler) processErrors(payload interface{}) interf
 func (b *BlockFieldSuggestionsHandler) processError(err map[string]interface{}) map[string]interface{} {
 	if msg, ok4 := err["message"]; ok4 {
 		if message, ok := msg.(string); ok {
-			err["message"] = b.ReplaceSuggestions(message)
+			err["message"] = b.replaceSuggestions(message)
 		}
 	}
 	return err
 }
 
-func (b *BlockFieldSuggestionsHandler) ReplaceSuggestions(message string) string {
+func (b *BlockFieldSuggestionsHandler) replaceSuggestions(message string) string {
 	if strings.HasPrefix(message, "Did you mean") {
 		return b.cfg.Mask
 	}

internal/business/persisted_operations/dir_loader.go

@@ -16,9 +16,9 @@ type DirLoader struct {
 	path string
 }
 
-func newDirLoader(cfg Config) *DirLoader {
+func NewLocalDirLoader(cfg Config) *DirLoader {
 	return &DirLoader{
-		path: cfg.Store.Dir,
+		path: cfg.Store,
 	}
 }
 

internal/business/persisted_operations/gcp_storage_loader.go

@@ -3,11 +3,13 @@ package persisted_operations
 import (
 	"cloud.google.com/go/storage"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"google.golang.org/api/iterator"
 	"io"
+	"os"
+	"path/filepath"
+	"time"
 )
 
 // GcpStorageLoader loads persisted operations from a GCP Storage bucket.
@@ -16,9 +18,10 @@ import (
 type GcpStorageLoader struct {
 	client *storage.Client
 	bucket string
+	store  string
 }
 
-func NewGcpStorageLoader(ctx context.Context, bucket string) (*GcpStorageLoader, error) {
+func NewGcpStorageLoader(ctx context.Context, bucket string, store string) (*GcpStorageLoader, error) {
 	client, err := storage.NewClient(ctx)
 	if err != nil {
 		return nil, err
@@ -27,40 +30,54 @@ func NewGcpStorageLoader(ctx context.Context, bucket string) (*GcpStorageLoader,
 	return &GcpStorageLoader{
 		client: client,
 		bucket: bucket,
+		store:  store,
 	}, nil
 }
-
-func (g *GcpStorageLoader) Load(ctx context.Context) (map[string]string, error) {
-	var store map[string]string
-
+func (g *GcpStorageLoader) Load(ctx context.Context) error {
 	it := g.client.Bucket(g.bucket).Objects(ctx, &storage.Query{
 		MatchGlob: "*.json",
 	})
+
+	var errs []error
 	for {
 		attrs, err := it.Next()
 		if errors.Is(err, iterator.Done) {
 			break
 		}
 		if err != nil {
-			continue
+			break
 		}
 
-		rc, err := g.client.Bucket(g.bucket).Object(attrs.Name).NewReader(ctx)
+		ctx, cancel := context.WithTimeout(ctx, time.Second*50)
+
+		f, err := os.Create(filepath.Join(g.store, attrs.Name))
 		if err != nil {
+			cancel()
+			errs = append(errs, fmt.Errorf("os.Create: %w", err))
 			continue
 		}
 
-		data, err := io.ReadAll(rc)
+		rc, err := g.client.Bucket(g.bucket).Object(attrs.Name).NewReader(ctx)
 		if err != nil {
-			return nil, fmt.Errorf("ioutil.ReadAll: %w", err)
+			cancel()
+			errs = append(errs, fmt.Errorf("Object(%q).NewReader: %w", attrs.Name, err))
+			continue
 		}
-		_ = rc.Close()
 
-		err = json.Unmarshal(data, &store)
-		if err != nil {
+		if _, err := io.Copy(f, rc); err != nil {
+			cancel()
+			errs = append(errs, fmt.Errorf("io.Copy: %w", err))
 			continue
 		}
+
+		if err = f.Close(); err != nil {
+			cancel()
+			errs = append(errs, fmt.Errorf("f.Close: %w", err))
+		}
+
+		cancel()
+		_ = rc.Close()
 	}
 
-	return store, nil
+	return errors.Join(errs...)
 }

internal/business/persisted_operations/loader.go

@@ -5,11 +5,19 @@ import (
 	"errors"
 )
 
-var ErrNoLoaderSpecified = errors.New("no loaders specified")
+type LocalLoader interface {
+	Load(ctx context.Context) (map[string]string, error)
+}
+
+type RemoteLoader interface {
+	Load(ctx context.Context) error
+}
+
+var ErrNoRemoteLoaderSpecified = errors.New("no remote loader specified")
 
-// DetermineLoaderFromConfig looks at the configuration applied and figures out which loader to initialize and return
-// If no loader is configured an error is returned
-func DetermineLoaderFromConfig(cfg Config) (PersistedOperationsLoader, error) {
+// RemoteLoaderFromConfig looks at the configuration applied and figures out which remoteLoader to initialize and return
+// If no remoteLoader is configured an error is returned
+func RemoteLoaderFromConfig(cfg Config) (RemoteLoader, error) {
 	loader, err := determineLoader(cfg)
 	if err != nil {
 		return loader, err
@@ -18,21 +26,13 @@ func DetermineLoaderFromConfig(cfg Config) (PersistedOperationsLoader, error) {
 }
 
 // load loads persisted operations from various sources
-func determineLoader(cfg Config) (PersistedOperationsLoader, error) {
-	if cfg.Store.Dir != "" {
-		loader := newDirLoader(cfg)
-		if loader == nil {
-			return nil, errors.New("unable to instantiate DirLoader")
-		}
-
-		return loader, nil
-	}
-	if cfg.Store.GcpBucket != "" {
-		loader, err := NewGcpStorageLoader(context.Background(), cfg.Store.GcpBucket)
+func determineLoader(cfg Config) (RemoteLoader, error) {
+	if cfg.Remote.GcpBucket != "" {
+		loader, err := NewGcpStorageLoader(context.Background(), cfg.Remote.GcpBucket, cfg.Store)
 		if err != nil {
 			return nil, errors.New("unable to instantiate GcpBucketLoader")
 		}
 		return loader, nil
 	}
-	return newMemoryLoader(map[string]string{}), ErrNoLoaderSpecified
+	return nil, ErrNoRemoteLoaderSpecified
 }