From 1371ba4bbb75e323ad6db3c4c5273feefc518f3d Mon Sep 17 00:00:00 2001 From: Drew DeVault Date: Fri, 20 Aug 2021 15:16:50 +0200 Subject: [PATCH] Configure server to manage mail queue This also sets up the recovery function for webhooks, so we get emailed when we panic during a webhook query. --- config/middleware.go | 13 +++++++------ email/worker.go | 10 +++++++--- server/email.go | 2 +- server/server.go | 14 +++++++++++--- webhooks/context.go | 6 +++++- 5 files changed, 31 insertions(+), 14 deletions(-) diff --git a/config/middleware.go b/config/middleware.go index a936c57caec07201b454c23b31cb73046d79472a..dd1b1f4794707b02bf88cff4b174dbd4cbe8f49f 100644 --- a/config/middleware.go +++ b/config/middleware.go @@ -16,19 +16,20 @@ type contextKey struct { } func Middleware(conf ini.File, service string) func(next http.Handler) http.Handler { - svc := service return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - ctx := Context(r.Context(), conf) - ctx = context.WithValue(ctx, serviceCtxKey, &svc) + ctx := Context(r.Context(), conf, service) r = r.WithContext(ctx) next.ServeHTTP(w, r) }) } } -func Context(ctx context.Context, conf ini.File) context.Context { - return context.WithValue(ctx, configCtxKey, conf) +func Context(ctx context.Context, conf ini.File, service string) context.Context { + svc := service + ctx = context.WithValue(ctx, configCtxKey, conf) + ctx = context.WithValue(ctx, serviceCtxKey, &svc) + return ctx } func ForContext(ctx context.Context) ini.File { @@ -42,7 +43,7 @@ func ForContext(ctx context.Context) ini.File { func ServiceName(ctx context.Context) string { raw, ok := ctx.Value(serviceCtxKey).(*string) if !ok { - panic(errors.New("Invalid config context")) + panic(errors.New("Invalid service config context")) } return *raw } diff --git a/email/worker.go b/email/worker.go index 41cb120637396f8e6582fe78dc601e0229ad9318..57f35f6781e566d7da55ce612928b9e66cef2f02 100644 --- a/email/worker.go +++ b/email/worker.go @@ -192,12 +192,16 @@ func ForContext(ctx context.Context) *work.Queue { return q } +// Returns a context which includes the given mail worker. +func Context(ctx context.Context, queue *work.Queue) context.Context { + return context.WithValue(ctx, emailCtxKey, queue) +} + // Adds HTTP middleware to provide an email work queue to this context. -func Middleware(worker *work.Queue) func(next http.Handler) http.Handler { +func Middleware(queue *work.Queue) func(next http.Handler) http.Handler { return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - ctx := context.WithValue(r.Context(), emailCtxKey, worker) - r = r.WithContext(ctx) + r = r.WithContext(Context(r.Context(), queue)) next.ServeHTTP(w, r) }) } diff --git a/server/email.go b/server/email.go index 38f112f4f9f180fb33690814c670c4ee1402f5f2..584d5cb79d3fcfe278ab759b7b6e268ba9d2178f 100644 --- a/server/email.go +++ b/server/email.go @@ -19,7 +19,7 @@ import ( // Provides a graphql.RecoverFunc which will print the stack trace, and if // debug mode is not enabled, email it to the administrator. -func emailRecover(ctx context.Context, _origErr interface{}) error { +func EmailRecover(ctx context.Context, _origErr interface{}) error { var ( ok bool origErr error diff --git a/server/server.go b/server/server.go index 11e7b97f6d921f44ae2e2e23c4c52688dc6f81d1..7bb382226096ea11d1e903f26beeb7c89b17c394 100644 --- a/server/server.go +++ b/server/server.go @@ -30,6 +30,7 @@ import ( "git.sr.ht/~sircmpwn/core-go/auth" "git.sr.ht/~sircmpwn/core-go/config" "git.sr.ht/~sircmpwn/core-go/database" + "git.sr.ht/~sircmpwn/core-go/email" "git.sr.ht/~sircmpwn/core-go/redis" ) @@ -54,6 +55,7 @@ type Server struct { router chi.Router service string queues []*work.Queue + email *work.Queue } // Creates a new common server context for a SourceHut GraphQL daemon. @@ -95,14 +97,14 @@ func (server *Server) WithSchema( srv := handler.GraphQL(schema, handler.ComplexityLimit(complexity), - handler.RecoverFunc(emailRecover), + handler.RecoverFunc(EmailRecover), handler.UploadMaxSize(1073741824)) // 1 GiB (TODO: configurable?) - server.router.Handle("/query", srv) if config.Debug { server.router.Handle("/", playground.Handler("GraphQL playground", "/query")) } + server.router.Handle("/query", srv) server.router.Handle("/query/metrics", promhttp.Handler()) server.router.Get("/query/api-meta.json", func(w http.ResponseWriter, r *http.Request) { info := struct { @@ -133,6 +135,7 @@ type contextKey struct { // - PostgresSQL connection pool // - Redis connection // - Authentication middleware +// - An email queue // - Standard rigging: logging, x-real-ip, instrumentation, etc func (server *Server) WithDefaultMiddleware() *Server { pgcs, ok := server.conf.Get(server.service, "connection-string") @@ -169,6 +172,8 @@ func (server *Server) WithDefaultMiddleware() *Server { timeout = 3 * time.Second } + server.email = email.NewQueue() + server.router.Use(func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { start := time.Now() @@ -180,6 +185,7 @@ func (server *Server) WithDefaultMiddleware() *Server { }) }) server.router.Use(config.Middleware(server.conf, server.service)) + server.router.Use(email.Middleware(server.email)) server.router.Use(database.Middleware(db)) server.router.Use(redis.Middleware(rc)) server.router.Use(middleware.RealIP) @@ -194,6 +200,7 @@ func (server *Server) WithDefaultMiddleware() *Server { next.ServeHTTP(w, r) }) }) + server.WithQueues(server.email) return server } @@ -223,9 +230,10 @@ func (server *Server) WithMiddleware( // Add dowork task queues for this server to manage func (server *Server) WithQueues(queues ...*work.Queue) *Server { ctx := context.Background() - ctx = config.Context(ctx, server.conf) + ctx = config.Context(ctx, server.conf, server.service) ctx = database.Context(ctx, server.db) ctx = redis.Context(ctx, server.redis) + ctx = email.Context(ctx, server.email) server.queues = append(server.queues, queues...) for _, queue := range queues { diff --git a/webhooks/context.go b/webhooks/context.go index 320878c37110f79137decac630051548d8fc54c9..029b683ce983470533f8f5f50046b211852fce23 100644 --- a/webhooks/context.go +++ b/webhooks/context.go @@ -12,6 +12,7 @@ import ( "github.com/google/uuid" "git.sr.ht/~sircmpwn/core-go/auth" + "git.sr.ht/~sircmpwn/core-go/server" ) type contextKey struct { @@ -63,6 +64,7 @@ func (webhook *WebhookContext) Exec(ctx context.Context, return nil, err } + // TODO: Set complexity limit exec := executor.New(schema) params := graphql.RawParams{ Query: sub.Query, @@ -76,8 +78,10 @@ func (webhook *WebhookContext) Exec(ctx context.Context, if errors != nil { panic(errors) } - ctx = graphql.WithOperationContext(ctx, rc) + rc.RecoverFunc = server.EmailRecover + var resp graphql.ResponseHandler + ctx = graphql.WithOperationContext(ctx, rc) resp, ctx = exec.DispatchOperation(ctx, rc) payload, err := json.Marshal(resp(ctx)) if err != nil {