Refactor kafka producer and add email templates

mailer/kafka.go

@@ -0,0 +1,59 @@
+package mailer
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"github.com/confluentinc/confluent-kafka-go/kafka"
+)
+
+type Config struct {
+	KafkaBrokers      []string `envconfig:"TIDEPOOL_KAFKA_BROKERS" validate:"required"`
+	KafkaFlushTimeout int      `envconfig:"TIDEPOOL_KAFKA_FLUSH_TIMEOUT" default:"30s" validate:"required"`
+	KafkaTopic        string   `envconfig:"TIDEPOOL_KAFKA_EMAILS_TOPIC" validate:"required"`
+}
+
+type KafkaMailer struct {
+	cfg          *Config
+	deliveryChan chan kafka.Event
+	producer     *kafka.Producer
+}
+
+var _ Mailer = &KafkaMailer{}
+
+func NewKafkaMailer(cfg *Config, deliveryChan chan kafka.Event) (*KafkaMailer, error) {
+	producer, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": cfg.KafkaBrokers})
+	if err != nil {
+		return nil, err
+	}
+
+	return &KafkaMailer{
+		cfg:          cfg,
+		deliveryChan: deliveryChan,
+		producer:     producer,
+	}, nil
+}
+
+func (k *KafkaMailer) Send(ctx context.Context, email *Email) error {
+	b, err := json.Marshal(email)
+	if err != nil {
+		return err
+	}
+
+	err = k.producer.Produce(&kafka.Message{
+		TopicPartition: kafka.TopicPartition{Topic: &k.cfg.KafkaTopic, Partition: kafka.PartitionAny},
+		Value:          b,
+	}, k.deliveryChan)
+	return err
+}
+
+func (k *KafkaMailer) Close(timeoutMs int) (err error) {
+	outstandingEvents := k.producer.Flush(timeoutMs)
+	if outstandingEvents != 0 {
+		err = errors.New(fmt.Sprintf("%v events were not delivered", outstandingEvents))
+	}
+	k.producer.Close()
+	close(k.deliveryChan)
+	return
+}

mailer/template.go

@@ -0,0 +1,55 @@
+package mailer
+
+import (
+	"bytes"
+	"fmt"
+	"github.com/pkg/errors"
+	"html/template"
+)
+
+type EmailTemplate struct {
+	body    *template.Template
+	name    string
+	subject *template.Template
+}
+
+func NewEmailTemplate(name string, subject string, body string) (*EmailTemplate, error) {
+	if name == "" {
+		return nil, errors.New("email template name cannot be empty")
+	}
+	if subject == "" {
+		return nil, errors.New("email template subject cannot be empty")
+	}
+	if body == "" {
+		return nil, errors.New("email template body cannot be empty")
+	}
+	precompiledSubject, err := template.New(fmt.Sprintf("%s_subject", name)).Parse(subject)
+	if err != nil {
+		return nil, err
+	}
+	precompiledBody, err := template.New(fmt.Sprintf("%s_body", name)).Parse(body)
+	if err != nil {
+		return nil, err
+	}
+	return &EmailTemplate{
+		body:    precompiledBody,
+		name:    name,
+		subject: precompiledSubject,
+	}, nil
+}
+
+func (e *EmailTemplate) RenderToEmail(params interface{}, email *Email) error {
+	var subject bytes.Buffer
+	var body bytes.Buffer
+
+	if err := e.subject.Execute(&subject, params); err != nil {
+		return err
+	}
+	if err := e.body.Execute(&body, params); err != nil {
+		return err
+	}
+
+	email.Subject = subject.String()
+	email.Body = body.String()
+	return nil
+}