Add mailer system

Signed-off-by: gearnode <bryan@frimin.fr>

pkg/coredata/email.go

@@ -0,0 +1,133 @@
+// Copyright (c) 2025 Probo Inc <hello@getprobo.com>.
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+package coredata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/getprobo/probo/pkg/gid"
+	"github.com/jackc/pgx/v5"
+	"go.gearno.de/kit/pg"
+)
+
+type (
+	Email struct {
+		ID             gid.GID    `db:"id"`
+		RecipientEmail string     `db:"recipient_email"`
+		RecipientName  string     `db:"recipient_name"`
+		Subject        string     `db:"subject"`
+		TextBody       string     `db:"text_body"`
+		CreatedAt      time.Time  `db:"created_at"`
+		UpdatedAt      time.Time  `db:"updated_at"`
+		SentAt         *time.Time `db:"sent_at"`
+	}
+)
+
+var (
+	ErrNoUnsentEmail = errors.New("no unsent email found")
+)
+
+func NewEmail(
+	recipientName string,
+	recipientEmail string,
+	subject string,
+	body string,
+) *Email {
+	return &Email{
+		ID:             gid.New(gid.NilTenant, EmailEntityType),
+		RecipientName:  recipientName,
+		RecipientEmail: recipientEmail,
+		Subject:        subject,
+		TextBody:       body,
+	}
+}
+
+func (e *Email) Insert(
+	ctx context.Context,
+	conn pg.Conn,
+) error {
+	q := `
+INSERT INTO emails (id, recipient_email, recipient_name, subject, text_body, created_at, updated_at)
+VALUES (@id, @recipient_email, @recipient_name, @subject, @text_body, @created_at, @updated_at)
+	`
+
+	args := pgx.StrictNamedArgs{
+		"id":              e.ID,
+		"recipient_email": e.RecipientEmail,
+		"recipient_name":  e.RecipientName,
+		"subject":         e.Subject,
+		"text_body":       e.TextBody,
+		"created_at":      e.CreatedAt,
+		"updated_at":      e.UpdatedAt,
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	return err
+}
+
+func (e *Email) LoadNextUnsentForUpdate(
+	ctx context.Context,
+	conn pg.Conn,
+) error {
+	q := `
+SELECT id, recipient_email, recipient_name, subject, text_body, created_at, updated_at, sent_at
+FROM emails
+WHERE sent_at IS NULL
+ORDER BY created_at ASC
+LIMIT 1
+FOR UPDATE
+	`
+
+	rows, err := conn.Query(ctx, q)
+	if err != nil {
+		return err
+	}
+
+	email, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[Email])
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ErrNoUnsentEmail
+		}
+
+		return fmt.Errorf("cannot collect email: %w", err)
+	}
+
+	*e = email
+
+	return nil
+}
+
+func (e *Email) Update(
+	ctx context.Context,
+	conn pg.Conn,
+) error {
+	q := `
+UPDATE emails
+SET sent_at = @sent_at, updated_at = @updated_at
+WHERE id = @id
+	`
+
+	args := pgx.StrictNamedArgs{
+		"id":         e.ID,
+		"sent_at":    e.SentAt,
+		"updated_at": e.UpdatedAt,
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	return err
+}

pkg/coredata/entity_type_reg.go

@@ -28,4 +28,5 @@ const (
 	PolicyEntityType
 	UserEntityType
 	SessionEntityType
+	EmailEntityType
 )

pkg/coredata/migrations/20250311T183100Z.sql

@@ -0,0 +1,11 @@
+CREATE TABLE emails (
+	id TEXT PRIMARY KEY,
+	email_address_to TEXT NOT NULL,
+	subject TEXT NOT NULL,
+	text_body TEXT NOT NULL,
+	created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+	updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+	sent_at TIMESTAMP WITH TIME ZONE
+);
+
+CREATE INDEX ON emails (sent_at) WHERE sent_at IS NULL;

pkg/coredata/migrations/20250312T084900Z.sql

@@ -0,0 +1,8 @@
+ALTER TABLE emails ADD COLUMN recipient_name TEXT NOT NULL;
+ALTER TABLE emails RENAME COLUMN email_address_to TO recipient_email;
+
+ALTER TABLE emails ALTER COLUMN created_at DROP DEFAULT;
+ALTER TABLE emails ALTER COLUMN updated_at DROP DEFAULT;
+
+ALTER TABLE emails ALTER COLUMN created_at SET NOT NULL;
+ALTER TABLE emails ALTER COLUMN updated_at SET NOT NULL;

pkg/mailer/mailer.go

@@ -0,0 +1,115 @@
+// Copyright (c) 2025 Probo Inc <hello@getprobo.com>.
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+package mailer
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net/smtp"
+	"time"
+
+	"github.com/getprobo/probo/pkg/coredata"
+	"github.com/jhillyerd/enmime"
+	"go.gearno.de/kit/log"
+	"go.gearno.de/kit/pg"
+)
+
+type (
+	Mailer struct {
+		pg  *pg.Client
+		l   *log.Logger
+		cfg Config
+	}
+
+	Config struct {
+		SenderName  string
+		SenderEmail string
+		Addr        string
+	}
+)
+
+func NewMailer(pg *pg.Client, l *log.Logger, cfg Config) *Mailer {
+	return &Mailer{pg: pg, l: l, cfg: cfg}
+}
+
+func (m *Mailer) Run(ctx context.Context) error {
+LOOP:
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-time.After(1 * time.Second):
+		ctx := context.Background()
+		if err := m.batchSendEmails(ctx); err != nil {
+			m.l.ErrorCtx(ctx, "cannot send email", log.Error(err))
+		}
+
+		goto LOOP
+	}
+}
+
+func (m *Mailer) batchSendEmails(ctx context.Context) error {
+	for {
+		err := m.pg.WithTx(
+			ctx,
+			func(tx pg.Conn) error {
+				email := &coredata.Email{}
+				err := email.LoadNextUnsentForUpdate(ctx, tx)
+				if err != nil {
+					return err
+				}
+
+				mail := enmime.Builder().
+					Subject(email.Subject).
+					From(m.cfg.SenderName, m.cfg.SenderEmail).
+					To(email.RecipientName, email.RecipientEmail).
+					Text([]byte(email.TextBody))
+
+				envelope, err := mail.Build()
+				if err != nil {
+					return fmt.Errorf("cannot build email: %w", err)
+				}
+
+				var buf bytes.Buffer
+				if err := envelope.Encode(&buf); err != nil {
+					return fmt.Errorf("cannot encode email: %w", err)
+				}
+
+				if err := smtp.SendMail(m.cfg.Addr, nil, m.cfg.SenderEmail, []string{email.RecipientEmail}, buf.Bytes()); err != nil {
+					return fmt.Errorf("cannot send email: %w", err)
+				}
+
+				now := time.Now()
+				email.SentAt = &now
+				email.UpdatedAt = now
+
+				if err := email.Update(ctx, tx); err != nil {
+					return fmt.Errorf("cannot update email: %w", err)
+				}
+
+				return nil
+			},
+		)
+
+		if errors.Is(err, coredata.ErrNoUnsentEmail) {
+			return nil
+		}
+
+		if err != nil {
+			return err
+		}
+	}
+}

pkg/probod/mailer_config.go

@@ -0,0 +1,29 @@
+// Copyright (c) 2025 Probo Inc <hello@getprobo.com>.
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+package probod
+
+type (
+	mailerConfig struct {
+		SenderName  string     `json:"sender-name"`
+		SenderEmail string     `json:"sender-email"`
+		SMTP        smtpConfig `json:"smtp"`
+	}
+
+	smtpConfig struct {
+		Addr     string `json:"addr"`
+		User     string `json:"user"`
+		Password string `json:"password"`
+	}
+)

pkg/probod/probod.go

@@ -27,6 +27,7 @@ import (
 	"github.com/getprobo/probo/pkg/awsconfig"
 	"github.com/getprobo/probo/pkg/coredata"
 	"github.com/getprobo/probo/pkg/crypto/passwdhash"
+	"github.com/getprobo/probo/pkg/mailer"
 	"github.com/getprobo/probo/pkg/probo"
 	"github.com/getprobo/probo/pkg/server"
 	console_v1 "github.com/getprobo/probo/pkg/server/api/console/v1"
@@ -47,10 +48,11 @@ type (
 	}
 
 	config struct {
-		Pg   pgConfig   `json:"pg"`
-		Api  apiConfig  `json:"api"`
-		Auth authConfig `json:"auth"`
-		AWS  awsConfig  `json:"aws"`
+		Pg     pgConfig     `json:"pg"`
+		Api    apiConfig    `json:"api"`
+		Auth   authConfig   `json:"auth"`
+		AWS    awsConfig    `json:"aws"`
+		Mailer mailerConfig `json:"mailer"`
 	}
 )
 
@@ -94,6 +96,13 @@ func New() *Implm {
 				SecretAccessKey: "thisisnotasecret",
 				Endpoint:        "http://127.0.0.1:9000",
 			},
+			Mailer: mailerConfig{
+				SenderEmail: "no-reply@notification.getprobo.com",
+				SenderName:  "Probo",
+				SMTP: smtpConfig{
+					Addr: "localhost:1025",
+				},
+			},
 		},
 	}
 }
@@ -198,8 +207,23 @@ func (impl *Implm) Run(
 		}
 	}()
 
+	mailerCtx, stopMailer := context.WithCancel(context.Background())
+	mailer := mailer.NewMailer(pgClient, l, mailer.Config{
+		SenderEmail: impl.cfg.Mailer.SenderEmail,
+		SenderName:  impl.cfg.Mailer.SenderName,
+		Addr:        impl.cfg.Mailer.SMTP.Addr,
+	})
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := mailer.Run(mailerCtx); err != nil {
+			cancel(fmt.Errorf("mailer crashed: %w", err))
+		}
+	}()
+
 	<-ctx.Done()
 
+	stopMailer()
 	stopApiServer()
 
 	wg.Wait()