courier: Implement message templates and SMTP delivery

.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)
+ }
+}