courier: Implement message templates and SMTP delivery (#146)

This patch adds a message templates (with override capabilities)
and SMTP delivery.

Integration tests using MailHog test fault resilience and e2e email
delivery.

This system is designed to be extended for SMS and other use cases.

Closes #99

.circleci/config.yml

@@ -34,12 +34,17 @@ jobs:
         image: circleci/golang:1.13
         environment:
           - GO111MODULE=on
+          - TEST_MAILHOG_SMTP=smtp://test:[email protected]:1025
+          - TEST_MAILHOG_API=smtp://127.0.0.1:8025
           - TEST_SELFSERVICE_OIDC_HYDRA_ADMIN=http://127.0.0.1:4445
           - TEST_SELFSERVICE_OIDC_HYDRA_PUBLIC=http://127.0.0.1:4444
           - TEST_SELFSERVICE_OIDC_HYDRA_INTEGRATION_ADDR=127.0.0.1:4499
           - TEST_DATABASE_POSTGRESQL=postgres://test:test@localhost:5432/postgres?sslmode=disable
           - TEST_DATABASE_MYSQL=mysql://root:test@(localhost:3306)/mysql?parseTime=true
 #          - TEST_DATABASE_COCKROACHDB=cockroach://root@localhost:26257/defaultdb?sslmode=disable
+      -
+        image: mailhog/mailhog:v1.0.0
+        command: MailHog -invite-jim -jim-linkspeed-affect=0.25 -jim-reject-auth=0.25 -jim-reject-recipient=0.25 -jim-reject-sender=0.25 -jim-disconnect=0.25 -jim-linkspeed-min=1250 -jim-linkspeed-max=12500
       -
         image: postgres:9.6
         environment:

contrib/sql/migrations/20191120000000_identities.up.fizz → contrib/sql/migrations/20191100000001_identities.up.fizz

@@ -2,8 +2,6 @@ create_table("identities") {
 	t.Column("id", "uuid", {primary: true})
     t.Column("traits_schema_url", "string", {"size": 2048})
     t.Column("traits", "json")
-
-	t.Timestamps()
 }
 
 create_table("identity_credential_types") {
@@ -22,8 +20,6 @@ create_table("identity_credentials") {
     t.Column("identity_credential_type_id", "uuid", { "size": 32 })
     t.Column("identity_id", "uuid")
 
- 	t.Timestamps()
-
     t.ForeignKey("identity_id", {"identities": ["id"]}, {"on_delete": "cascade"})
     t.ForeignKey("identity_credential_type_id", {"identity_credential_types": ["id"]}, {"on_delete": "cascade"})
 }
@@ -33,8 +29,6 @@ create_table("identity_credential_identifiers") {
     t.Column("identifier", "string", {"size": 255})
     t.Column("identity_credential_id", "uuid")
 
- 	t.Timestamps()
-
     t.ForeignKey("identity_credential_id", {"identity_credentials": ["id"]}, {"on_delete": "cascade"})
 }
 

contrib/sql/migrations/20191121000000_requests.up.fizz → contrib/sql/migrations/20191100000002_requests.up.fizz

@@ -4,8 +4,6 @@ create_table("selfservice_login_requests") {
     t.Column("issued_at", "timestamp", { default_raw: "CURRENT_TIMESTAMP" })
     t.Column("expires_at", "timestamp")
     t.Column("active_method", "string", {"size": 32})
-
-	t.Timestamps()
 }
 
 create_table("selfservice_login_request_methods") {
@@ -14,8 +12,6 @@ create_table("selfservice_login_request_methods") {
     t.Column("selfservice_login_request_id", "uuid")
     t.Column("config", "json")
 
-	t.Timestamps()
-
     t.ForeignKey("selfservice_login_request_id", {"selfservice_login_requests": ["id"]}, {"on_delete": "cascade"})
 }
 
@@ -25,8 +21,6 @@ create_table("selfservice_registration_requests") {
     t.Column("issued_at", "timestamp", { default_raw: "CURRENT_TIMESTAMP" })
     t.Column("expires_at", "timestamp")
     t.Column("active_method", "string", {"size": 32})
-
-	t.Timestamps()
 }
 
 create_table("selfservice_registration_request_methods") {
@@ -35,8 +29,6 @@ create_table("selfservice_registration_request_methods") {
     t.Column("selfservice_registration_request_id", "uuid")
     t.Column("config", "json")
 
-	t.Timestamps()
-
     t.ForeignKey("selfservice_registration_request_id", {"selfservice_registration_requests": ["id"]}, {"on_delete": "cascade"})
 }
 
@@ -49,7 +41,5 @@ create_table("selfservice_profile_management_requests") {
     t.Column("update_successful", "bool")
     t.Column("identity_id", "uuid")
 
-	t.Timestamps()
-
     t.ForeignKey("identity_id", {"identities": ["id"]}, {"on_delete": "cascade"})
 }

contrib/sql/migrations/20191122000000_sessions.up.fizz → contrib/sql/migrations/20191100000003_sessions.up.fizz

@@ -5,7 +5,5 @@ create_table("sessions") {
     t.Column("authenticated_at", "timestamp")
     t.Column("identity_id", "uuid")
 
- 	t.Timestamps()
-
     t.ForeignKey("identity_id", {"identities": ["id"]}, {"on_delete": "cascade"})
 }

contrib/sql/migrations/20191123000000_errors.up.fizz → contrib/sql/migrations/20191100000004_errors.up.fizz

@@ -3,6 +3,4 @@ create_table("selfservice_errors") {
     t.Column("errors", "json")
     t.Column("seen_at", "timestamp")
     t.Column("was_seen", "bool")
-
- 	t.Timestamps()
 }

contrib/sql/migrations/20191100000005_identities.mysql.down.sql

@@ -0,0 +1 @@
+ALTER TABLE identity_credential_identifiers MODIFY COLUMN identifier VARCHAR(255);

contrib/sql/migrations/20191100000006_courier.down.fizz

@@ -0,0 +1 @@
+drop_table("courier_messages")

contrib/sql/migrations/20191100000006_courier.up.fizz

@@ -0,0 +1,10 @@
+create_table("courier_messages") {
+	t.Column("id", "uuid", {primary: true})
+
+    t.Column("type", "int")
+    t.Column("status", "int")
+
+    t.Column("body", "string")
+    t.Column("subject", "string")
+    t.Column("recipient", "string")
+}

courier/courier.go

@@ -0,0 +1,145 @@
+package courier
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/cenkalti/backoff"
+	"github.com/gofrs/uuid"
+	"github.com/pkg/errors"
+	"gopkg.in/gomail.v2"
+
+	"github.com/ory/kratos/driver/configuration"
+	"github.com/ory/kratos/x"
+)
+
+type (
+	smtpDependencies interface {
+		PersistenceProvider
+		x.LoggingProvider
+	}
+	Courier struct {
+		dialer *gomail.Dialer
+		d      smtpDependencies
+		c      configuration.Provider
+	}
+	Provider interface {
+		Courier() *Courier
+	}
+)
+
+func NewSMTP(d smtpDependencies, c configuration.Provider) *Courier {
+	uri := c.CourierSMTPURL()
+	sslSkipVerify, _ := strconv.ParseBool(uri.Query().Get("skip_ssl_verify"))
+	password, _ := uri.User.Password()
+	port, _ := strconv.ParseInt(uri.Port(), 10, 64)
+	return &Courier{
+		d: d,
+		c: c,
+		dialer: &gomail.Dialer{
+			Host:      uri.Hostname(),
+			Port:      int(port),
+			Username:  uri.User.Username(),
+			Password:  password,
+			SSL:       uri.Scheme == "smtps",
+			TLSConfig: &tls.Config{InsecureSkipVerify: sslSkipVerify},
+		},
+	}
+}
+
+func (m *Courier) SendEmail(ctx context.Context, t EmailTemplate) (uuid.UUID, error) {
+	body, err := t.EmailBody()
+	if err != nil {
+		return uuid.Nil, err
+	}
+
+	subject, err := t.EmailSubject()
+	if err != nil {
+		return uuid.Nil, err
+	}
+
+	recipient, err := t.EmailRecipient()
+	if err != nil {
+		return uuid.Nil, err
+	}
+
+	message := &Message{
+		Status:    MessageStatusQueued,
+		Type:      MessageTypeEmail,
+		Body:      body,
+		Subject:   subject,
+		Recipient: recipient,
+	}
+	if err := m.d.CourierPersister().AddMessage(ctx, message); err != nil {
+		return uuid.Nil, err
+	}
+	return message.ID, nil
+}
+
+func (m *Courier) Work(ctx context.Context) error {
+	errChan := make(chan error)
+	defer close(errChan)
+
+	go m.watchMessages(ctx, errChan)
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case err := <-errChan:
+		return err
+	}
+}
+
+func (m *Courier) watchMessages(ctx context.Context, errChan chan error) {
+	for {
+		if err := backoff.Retry(func() error {
+			messages, err := m.d.CourierPersister().NextMessages(ctx, 10)
+			if err != nil {
+				return err
+			}
+
+			for k := range messages {
+				var msg Message = messages[k]
+
+				switch msg.Type {
+				case MessageTypeEmail:
+					from := m.c.CourierSMTPFrom()
+					gm := gomail.NewMessage()
+					gm.SetHeader("From", from)
+					gm.SetHeader("To", msg.Recipient)
+					gm.SetHeader("Subject", msg.Subject)
+					gm.SetBody("text/plain", msg.Body)
+					gm.AddAlternative("text/html", msg.Body)
+
+					if err := m.dialer.DialAndSend(gm); err != nil {
+						m.d.Logger().
+							WithError(err).
+							WithField("smtp_server", fmt.Sprintf("smtp(s)://%s:%d", m.dialer.Host, m.dialer.Port)).
+							WithField("email_to", msg.Recipient).WithField("email_from", from).
+							Error("Unable to send email using SMTP connection.")
+						continue
+					}
+
+					if err := m.d.CourierPersister().SetMessageStatus(ctx, msg.ID, MessageStatusSent); err != nil {
+						m.d.Logger().
+							WithError(err).
+							WithField("message_id", msg.ID).
+							Error(`Unable to set the message status to "sent".`)
+						return err
+					}
+				default:
+					return errors.Errorf("received unexpected message type: %d", msg.Type)
+				}
+			}
+
+			return nil
+		}, backoff.NewExponentialBackOff()); err != nil {
+			errChan <- err
+			return
+		}
+		time.Sleep(time.Second)
+	}
+}