feat: initial commit

2022-10-09 15:23:52 +02:00
commit a1b4d4fc27
39 changed files with 5810 additions and 0 deletions
@@ -0,0 +1,95 @@
+package main
+
+import (
+	"io"
+	"net/url"
+	"os"
+	"sort"
+	"time"
+
+	"github.com/alecthomas/kong"
+	"github.com/apex/log"
+
+	"gitlab.com/unboundsoftware/schemas/ctl"
+)
+
+type Context struct {
+	ApiKey     string
+	SchemaRef  string
+	SchemasURL url.URL
+	logger     log.Interface
+}
+
+type PublishCmd struct {
+	Service string   `name:"service" env:"SERVICE" help:"The service to publish SDL for" required:""`
+	Url     *url.URL `name:"url" env:"URL" help:"The URL of the service" optional:""`
+	WSUrl   *url.URL `name:"ws-url" env:"WS_URL" help:"The Websocket URL of the service" optional:""`
+	SDL     *os.File `name:"sdl" env:"SDL" help:"The file containing the GraphQL SDL" required:""`
+}
+
+func (c *PublishCmd) Run(ctx Context) error {
+	buff, err := io.ReadAll(c.SDL)
+	if err != nil {
+		return err
+	}
+	subGraph, err := ctl.Publish(ctx.ApiKey, ctx.SchemaRef, c.Service, string(buff), c.Url, c.WSUrl, ctx.SchemasURL)
+	if err != nil {
+		return err
+	}
+	ctx.logger.Infof("published %s@%s %s", subGraph.Service, subGraph.ChangedAt.Format(time.RFC3339), subGraph.ChangedBy)
+	return nil
+}
+
+type ListCmd struct {
+	Service string `name:"service" env:"SERVICE" help:"The service to publish SDL for" optional:""`
+}
+
+func (c *ListCmd) Run(ctx Context) error {
+	subGraphs, err := ctl.List(ctx.ApiKey, ctx.SchemaRef, c.Service, ctx.SchemasURL)
+	if err != nil {
+		return err
+	}
+	sort.SliceStable(subGraphs, func(i, j int) bool {
+		return subGraphs[i].Service < subGraphs[j].Service
+	})
+	for _, subGraph := range subGraphs {
+		ctx.logger.Infof("%s URL: %s WS-URL: %s Changed: %s %s", subGraph.Service, emptyIfNil(subGraph.URL), emptyIfNil(subGraph.WSUrl), subGraph.ChangedAt.Format(time.RFC3339), subGraph.ChangedBy)
+	}
+	return nil
+}
+
+type CLI struct {
+	ApiKey             string     `name:"api-key" env:"API_KEY" help:"The API-key to use when communicating with the server" required:""`
+	SchemaRef          string     `name:"schema-ref" env:"SCHEMA_REF" help:"The schema reference to work with" required:""`
+	OverrideSchemasUrl *url.URL   `name:"override-schemas-url" env:"OVERRIDE_SCHEMAS_URL" help:"Use a specific URL when communicating with the Schemas-API" optional:""`
+	Publish            PublishCmd `cmd:""`
+	List               ListCmd    `cmd:""`
+	LogLevel           string     `name:"log-level" env:"LOG_LEVEL" help:"The level of logging to use (debug, info, warn, error, fatal)" default:"info"`
+}
+
+func main() {
+	cli := CLI{}
+	ctx := kong.Parse(&cli)
+	log.SetLevelFromString(cli.LogLevel)
+	logger := log.WithField("service", "schemactl")
+	err := ctx.Run(Context{ApiKey: cli.ApiKey, SchemaRef: cli.SchemaRef, SchemasURL: schemasUrl(cli.OverrideSchemasUrl), logger: logger})
+	ctx.FatalIfErrorf(err)
+}
+
+func emptyIfNil(s *string) string {
+	if s == nil {
+		return ""
+	}
+	return *s
+}
+
+func schemasUrl(u *url.URL) url.URL {
+	if u == nil {
+		u, err := url.Parse("https://schemas.unbound.se/query")
+		if err != nil {
+			return url.URL{}
+		}
+		return *u
+	}
+	return *u
+}
@@ -0,0 +1,216 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"reflect"
+	"sync"
+	"syscall"
+
+	"github.com/99designs/gqlgen/graphql/handler"
+	"github.com/99designs/gqlgen/graphql/playground"
+	"github.com/alecthomas/kong"
+	"github.com/apex/log"
+	"github.com/apex/log/handlers/json"
+	sentryhttp "github.com/getsentry/sentry-go/http"
+	"github.com/rs/cors"
+	"github.com/sparetimecoders/goamqp"
+	"gitlab.com/unboundsoftware/eventsourced/amqp"
+	"gitlab.com/unboundsoftware/eventsourced/eventsourced"
+	"gitlab.com/unboundsoftware/eventsourced/pg"
+
+	"gitlab.com/unboundsoftware/schemas/cache"
+	"gitlab.com/unboundsoftware/schemas/domain"
+	"gitlab.com/unboundsoftware/schemas/graph"
+	"gitlab.com/unboundsoftware/schemas/graph/generated"
+	"gitlab.com/unboundsoftware/schemas/middleware"
+	"gitlab.com/unboundsoftware/schemas/store"
+)
+
+var CLI struct {
+	AmqpURL            string `name:"amqp-url" env:"AMQP_URL" help:"URL to use to connect to RabbitMQ" default:"amqp://user:password@localhost:5672/"`
+	Port               int    `name:"port" env:"PORT" help:"Listen-port for GraphQL API" default:"8080"`
+	APIKey             string `name:"api-key" env:"API_KEY" help:"The API-key that is required"`
+	LogLevel           string `name:"log-level" env:"LOG_LEVEL" help:"The level of logging to use (debug, info, warn, error, fatal)" default:"info"`
+	DatabaseURL        string `name:"postgres-url" env:"POSTGRES_URL" help:"URL to use to connect to Postgres" default:"postgres://postgres:postgres@:5432/schemas?sslmode=disable"`
+	DatabaseDriverName string `name:"db-driver" env:"DB_DRIVER" help:"Driver to use to connect to db" default:"postgres"`
+}
+
+const serviceName = "schemas"
+
+func main() {
+	_ = kong.Parse(&CLI)
+	log.SetHandler(json.New(os.Stdout))
+	log.SetLevelFromString(CLI.LogLevel)
+	logger := log.WithField("service", serviceName)
+	closeEvents := make(chan error)
+
+	if err := start(
+		closeEvents,
+		logger,
+		ConnectAMQP,
+	); err != nil {
+		logger.WithError(err).Error("process error")
+	}
+}
+
+func start(closeEvents chan error, logger *log.Entry, connectToAmqpFunc func(url string) (Connection, error)) error {
+	rootCtx, rootCancel := context.WithCancel(context.Background())
+	defer rootCancel()
+
+	db, err := store.SetupDB(CLI.DatabaseDriverName, CLI.DatabaseURL)
+	if err != nil {
+		return fmt.Errorf("failed to setup DB: %v", err)
+	}
+
+	eventStore, err := pg.New(
+		db.DB,
+		pg.WithEventTypes(
+			&domain.SubGraphUpdated{},
+		),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create eventstore: %v", err)
+	}
+	eventPublisher, err := goamqp.NewPublisher(
+		goamqp.Route{
+			Type: domain.SubGraphUpdated{},
+			Key:  "SubGraph.Updated",
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create event publisher: %v", err)
+	}
+	amqp.New(eventPublisher)
+	conn, err := connectToAmqpFunc(CLI.AmqpURL)
+	if err != nil {
+		return fmt.Errorf("failed to connect to AMQP: %v", err)
+	}
+
+	serviceCache := cache.New(logger)
+	roots, err := eventStore.GetAggregateRoots(reflect.TypeOf(domain.SubGraph{}))
+	if err != nil {
+		return err
+	}
+	for _, root := range roots {
+		subGraph := &domain.SubGraph{BaseAggregate: eventsourced.BaseAggregateFromString(root.String())}
+		if _, err := eventsourced.NewHandler(subGraph, eventStore); err != nil {
+			return err
+		}
+		_, err := serviceCache.Update(subGraph, nil)
+		if err != nil {
+			return err
+		}
+	}
+	setups := []goamqp.Setup{
+		goamqp.UseLogger(logger.Errorf),
+		goamqp.CloseListener(closeEvents),
+		goamqp.WithPrefetchLimit(20),
+		goamqp.EventStreamPublisher(eventPublisher),
+		goamqp.TransientEventStreamConsumer("SubGraph.Updated", serviceCache.Update, domain.SubGraphUpdated{}),
+	}
+	if err := conn.Start(setups...); err != nil {
+		return fmt.Errorf("failed to setup AMQP: %v", err)
+	}
+
+	defer func() { _ = conn.Close() }()
+
+	logger.Info("Started")
+
+	mux := http.NewServeMux()
+	httpSrv := &http.Server{Addr: fmt.Sprintf(":%d", CLI.Port), Handler: mux}
+
+	wg := sync.WaitGroup{}
+
+	sigint := make(chan os.Signal, 1)
+	signal.Notify(sigint, os.Interrupt, syscall.SIGTERM)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		sig := <-sigint
+		if sig != nil {
+			// In case our shutdown logic is broken/incomplete we reset signal
+			// handlers so next signal goes to go itself. Go is more aggressive when
+			// shutting down goroutines
+			signal.Reset(os.Interrupt, syscall.SIGTERM)
+			logger.Info("Got shutdown signal..")
+			rootCancel()
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := <-closeEvents
+		if err != nil {
+			logger.WithError(err).Error("received close from AMQP")
+			rootCancel()
+		}
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		<-rootCtx.Done()
+
+		if err := httpSrv.Close(); err != nil {
+			logger.WithError(err).Error("close http server")
+		}
+		close(sigint)
+		close(closeEvents)
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		defer rootCancel()
+
+		resolver := &graph.Resolver{
+			EventStore: eventStore,
+			Publisher:  amqp.New(eventPublisher),
+			Logger:     logger,
+			Cache:      serviceCache,
+		}
+
+		config := generated.Config{
+			Resolvers:  resolver,
+			Complexity: generated.ComplexityRoot{},
+		}
+		apiKeyMiddleware := middleware.NewApiKey(CLI.APIKey, logger)
+		config.Directives.HasApiKey = apiKeyMiddleware.Directive
+		srv := handler.NewDefaultServer(generated.NewExecutableSchema(
+			config,
+		))
+
+		sentryHandler := sentryhttp.New(sentryhttp.Options{Repanic: true})
+		mux.Handle("/", sentryHandler.HandleFunc(playground.Handler("GraphQL playground", "/query")))
+		mux.Handle("/health", sentryHandler.HandleFunc(healthFunc))
+		mux.Handle("/query", cors.AllowAll().Handler(sentryHandler.Handle(apiKeyMiddleware.Handler(srv))))
+
+		logger.Infof("connect to http://localhost:%d/ for GraphQL playground", CLI.Port)
+
+		if err := httpSrv.ListenAndServe(); err != nil {
+			logger.WithError(err).Error("listen http")
+		}
+	}()
+
+	wg.Wait()
+
+	return nil
+}
+
+func healthFunc(w http.ResponseWriter, _ *http.Request) {
+	_, _ = w.Write([]byte("OK"))
+}
+
+func ConnectAMQP(url string) (Connection, error) {
+	return goamqp.NewFromURL(serviceName, url)
+}
+
+type Connection interface {
+	Start(opts ...goamqp.Setup) error
+	Close() error
+}