From 0c651498283fd4c69c28aab5d8249c0d29c80d71 Mon Sep 17 00:00:00 2001 From: Drew DeVault Date: Tue, 6 Oct 2020 15:56:13 -0400 Subject: [PATCH] Add email work queue --- config/middleware.go | 6 +++- email/send.go | 60 ++++++++++++++++++++++++++++++++++++ email/worker.go | 72 ++++++++++++++++++++++++++++++++++++++++++++ go.mod | 1 + go.sum | 2 ++ server/email.go | 42 ++------------------------ server/server.go | 12 +++++--- 7 files changed, 150 insertions(+), 45 deletions(-) create mode 100644 email/send.go create mode 100644 email/worker.go diff --git a/config/middleware.go b/config/middleware.go index fd238db3cf4c3503adfbcc3ec623e04efd1e5685..a936c57caec07201b454c23b31cb73046d79472a 100644 --- a/config/middleware.go +++ b/config/middleware.go @@ -19,7 +19,7 @@ func Middleware(conf ini.File, service string) func(next http.Handler) http.Hand svc := service return func(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - ctx := context.WithValue(r.Context(), configCtxKey, conf) + ctx := Context(r.Context(), conf) ctx = context.WithValue(ctx, serviceCtxKey, &svc) r = r.WithContext(ctx) next.ServeHTTP(w, r) @@ -27,6 +27,10 @@ func Middleware(conf ini.File, service string) func(next http.Handler) http.Hand } } +func Context(ctx context.Context, conf ini.File) context.Context { + return context.WithValue(ctx, configCtxKey, conf) +} + func ForContext(ctx context.Context) ini.File { raw, ok := ctx.Value(configCtxKey).(ini.File) if !ok { diff --git a/email/send.go b/email/send.go new file mode 100644 index 0000000000000000000000000000000000000000..bf7980ed2a55dffd7ae66fa8a99bcf4347096c97 --- /dev/null +++ b/email/send.go @@ -0,0 +1,60 @@ +package email + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/binary" + "errors" + "fmt" + "os" + "strconv" + "time" + + "github.com/martinlindhe/base36" + gomail "gopkg.in/mail.v2" + + "git.sr.ht/~sircmpwn/core-go/config" +) + +var attempts int = 0 + +// Sends an email. Blocks until it's sent or an error occurs. +func Send(ctx context.Context, m *gomail.Message) error { + conf := config.ForContext(ctx) + + portStr, ok := conf.Get("mail", "smtp-port") + if !ok { + return errors.New("internal system error") + } + port, _ := strconv.Atoi(portStr) + host, _ := conf.Get("mail", "smtp-host") + user, _ := conf.Get("mail", "smtp-user") + pass, _ := conf.Get("mail", "smtp-password") + + m.SetHeader("Message-ID", generateMessageID()) + m.SetDateHeader("Date", time.Now().UTC()) + + d := gomail.NewDialer(host, port, user, pass) + return d.DialAndSend(m) +} + +// Generates an RFC 2822-compliant Message-Id based on the informational draft +// "Recommendations for generating Message IDs", for lack of a better +// authoritative source. +func generateMessageID() string { + var ( + now bytes.Buffer + nonce []byte = make([]byte, 8) + ) + binary.Write(&now, binary.BigEndian, time.Now().UnixNano()) + rand.Read(nonce) + hostname, err := os.Hostname() + if err != nil { + hostname = "localhost" + } + return fmt.Sprintf("<%s.%s@%s>", + base36.EncodeBytes(now.Bytes()), + base36.EncodeBytes(nonce), + hostname) +} diff --git a/email/worker.go b/email/worker.go new file mode 100644 index 0000000000000000000000000000000000000000..37d3cebf382611a46aa9e8649e02b2408654e2bf --- /dev/null +++ b/email/worker.go @@ -0,0 +1,72 @@ +package email + +import ( + "context" + "errors" + "log" + "net/http" + "strings" + + "git.sr.ht/~sircmpwn/dowork" + gomail "gopkg.in/mail.v2" + + "git.sr.ht/~sircmpwn/core-go/config" +) + +var emailCtxKey = &contextKey{"email"} + +type contextKey struct { + name string +} + +// Returns a task which will send this email for the work queue. If the caller +// does not need to customize the task parameters, the Enqueue function may be +// more desirable. +func NewTask(ctx context.Context, m *gomail.Message) *work.Task { + conf := config.ForContext(ctx) + return work.NewTask(func(ctx context.Context) error { + return Send(config.Context(ctx, conf), m) + }).Retries(10).After(func(ctx context.Context, task *work.Task) { + if task.Result() == nil { + log.Printf("MAIL TO %s: '%s' sent after %d attempts", + strings.Join(m.GetHeader("To"), ";"), + strings.Join(m.GetHeader("Subject"), ";"), + task.Attempts()) + } else { + log.Printf("MAIL TO %s: '%s' failed after %d attempts: %v", + strings.Join(m.GetHeader("To"), ";"), + strings.Join(m.GetHeader("Subject"), ";"), + task.Attempts(), task.Result()) + } + }) +} + +// Enqueues an email for sending with the default parameters. +func Enqueue(ctx context.Context, m *gomail.Message) { + ForContext(ctx).Enqueue(NewTask(ctx, m)) +} + +// Creates a new email processing queue. +func NewQueue() *work.Queue { + return work.NewQueue("email") +} + +// Returns the email worker for this context. +func ForContext(ctx context.Context) *work.Queue { + q, ok := ctx.Value(emailCtxKey).(*work.Queue) + if !ok { + panic(errors.New("No email worker for this context")) + } + return q +} + +// Adds HTTP middleware to provide an email work queue to this context. +func Middleware(worker *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) + next.ServeHTTP(w, r) + }) + } +} diff --git a/go.mod b/go.mod index 3eb3910d9b8f08976042419aafe8eb307f68438d..e5d08cdba5473d1ba4dad28e1f5234a4a6eb4d45 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module git.sr.ht/~sircmpwn/core-go go 1.13 require ( + git.sr.ht/~sircmpwn/dowork v0.0.0-20201002192337-cc78e95c493c git.sr.ht/~sircmpwn/getopt v0.0.0-20191230200459-23622cc906b3 git.sr.ht/~sircmpwn/go-bare v0.0.0-20200812160916-d2c72e1a5018 github.com/99designs/gqlgen v0.13.0 diff --git a/go.sum b/go.sum index 8a90c4572ba56bc3ad729d42fbd3d1b6d9731297..fc2f8e26e2462cffcda5eae78eaa851a3c3a2994 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +git.sr.ht/~sircmpwn/dowork v0.0.0-20201002192337-cc78e95c493c h1:DHYVIt2TT6Nx+CK78om/isyzUp8ZCcGAs8lofsR8EAA= +git.sr.ht/~sircmpwn/dowork v0.0.0-20201002192337-cc78e95c493c/go.mod h1:8neHEO3503w/rNtttnR0JFpQgM/GFhaafVwvkPsFIDw= git.sr.ht/~sircmpwn/getopt v0.0.0-20191230200459-23622cc906b3 h1:4wDp4BKF7NQqoh73VXpZsB/t1OEhDpz/zEpmdQfbjDk= git.sr.ht/~sircmpwn/getopt v0.0.0-20191230200459-23622cc906b3/go.mod h1:wMEGFFFNuPos7vHmWXfszqImLppbc0wEhh6JBfJIUgw= git.sr.ht/~sircmpwn/go-bare v0.0.0-20200812160916-d2c72e1a5018 h1:89QMorzx6ML69PKPoayL3HuSfb7WqAlxD1dZ7DyzD0k= diff --git a/server/email.go b/server/email.go index 9c233240eea546b104804e949596418a970dda15..55d65b4a08e8d8be9454db7b89168d9df5951934 100644 --- a/server/email.go +++ b/server/email.go @@ -1,25 +1,19 @@ package server import ( - "bytes" "context" - "crypto/rand" - "encoding/binary" "errors" "fmt" "log" "net/mail" - "os" "runtime" - "strconv" - "time" "github.com/99designs/gqlgen/graphql" - "github.com/martinlindhe/base36" "github.com/vaughan0/go-ini" gomail "gopkg.in/mail.v2" "git.sr.ht/~sircmpwn/core-go/auth" + "git.sr.ht/~sircmpwn/core-go/email" ) // Provides a graphql.RecoverFunc which will print the stack trace, and if @@ -60,14 +54,6 @@ func EmailRecover(config ini.File, debug bool, srv string) graphql.RecoverFunc { return fmt.Errorf("internal system error") } from, _ := config.Get("mail", "error-from") - portStr, ok := config.Get("mail", "smtp-port") - if !ok { - return fmt.Errorf("internal system error") - } - port, _ := strconv.Atoi(portStr) - host, _ := config.Get("mail", "smtp-host") - user, _ := config.Get("mail", "smtp-user") - pass, _ := config.Get("mail", "smtp-password") m := gomail.NewMessage() sender, err := mail.ParseAddress(from) @@ -80,7 +66,6 @@ func EmailRecover(config ini.File, debug bool, srv string) graphql.RecoverFunc { log.Fatalf("Failed to parse recipient address") } m.SetAddressHeader("To", recipient.Address, recipient.Name) - m.SetHeader("Message-ID", generateMessageID()) m.SetHeader("Subject", fmt.Sprintf( "[%s] GraphQL query error: %v", srv, origErr)) @@ -99,30 +84,7 @@ The following stack trace was produced: %s`, origErr, quser.Username, quser.Email, octx.RawQuery, string(stack[:i]))) - d := gomail.NewDialer(host, port, user, pass) - if err := d.DialAndSend(m); err != nil { - log.Printf("Error sending email: %v\n", err) - } + email.Enqueue(ctx, m) return fmt.Errorf("internal system error") } } - -// Generates an RFC 2822-compliant Message-Id based on the informational draft -// "Recommendations for generating Message IDs", for lack of a better -// authoritative source. -func generateMessageID() string { - var ( - now bytes.Buffer - nonce []byte = make([]byte, 8) - ) - binary.Write(&now, binary.BigEndian, time.Now().UnixNano()) - rand.Read(nonce) - hostname, err := os.Hostname() - if err != nil { - hostname = "localhost" - } - return fmt.Sprintf("<%s.%s@%s>", - base36.EncodeBytes(now.Bytes()), - base36.EncodeBytes(nonce), - hostname) -} diff --git a/server/server.go b/server/server.go index d41b74ab2199c99b0dbe8ae0faaf3749383792f2..b4c0017f1dd92b2f4f73459c70b2b99e45f7077e 100644 --- a/server/server.go +++ b/server/server.go @@ -4,6 +4,7 @@ import ( "database/sql" "fmt" "log" + "net" "net/http" "os" "strconv" @@ -166,8 +167,11 @@ func MakeRouter(service string, conf ini.File, schema graphql.ExecutableSchema, return router } -// Runs the API server. -func ListenAndServe(router chi.Router) { - log.Printf("running on %s", addr) - log.Fatal(http.ListenAndServe(addr, router)) +func MakeServer(router chi.Router) (*http.Server, net.Listener) { + listen, err := net.Listen("tcp", addr) + if err != nil { + panic(err) + } + log.Printf("Running on %s", addr) + return &http.Server{Handler: router}, listen }