[IMPL] email queue processing (#176) (#187)

application.yaml

@@ -91,6 +91,8 @@ auth/telegram:
 auth/email-link:
   extraLoadBalancersCount: 2
   wintr/connectors/storage/v2: *db
+  wintr/connectors/storage/v3:
+    url: redis://default:@localhost:6379/
   fromEmailAddress: no-reply@ice.io
   fromEmailName: ice
   emailValidation:

auth/email_link/.testdata/docker-compose.yaml

@@ -0,0 +1,12 @@
+# SPDX-License-Identifier: ice License 1.0
+
+version: '3.7'
+services:
+  eskimo-dfly:
+    container_name: eskimo-dfly
+    image: 'docker.dragonflydb.io/dragonflydb/dragonfly'
+    ulimits:
+      memlock: -1
+    ports:
+      - 6379:6379
+    command: "dragonfly --dbnum=1"

auth/email_link/DDL.sql

@@ -9,6 +9,7 @@ CREATE TABLE IF NOT EXISTS email_link_sign_ins (
            previously_issued_token_seq            BIGINT DEFAULT 0 NOT NULL,
            confirmation_code_wrong_attempts_count BIGINT DEFAULT 0 NOT NULL,
            email                                  TEXT NOT NULL,
+           language                               TEXT NOT NULL,
            confirmation_code                      TEXT,
            user_id                                TEXT,
            phone_number_to_email_migration_user_id TEXT,
@@ -33,4 +34,7 @@ CREATE TABLE IF NOT EXISTS sign_ins_per_ip (
        login_attempts        BIGINT DEFAULT 0 NOT NULL CONSTRAINT sign_ins_per_ip_login_attempts_count CHECK (login_attempts <= 10),
        ip                    TEXT NOT NULL,
        PRIMARY KEY (login_session_number, ip)
-);
+);
+
+ALTER TABLE email_link_sign_ins
+    ADD COLUMN IF NOT EXISTS language TEXT NOT NULL DEFAULT 'en';

auth/email_link/contract.go

@@ -7,6 +7,7 @@ import (
 	"embed"
 	"io"
 	"mime/multipart"
+	"sync"
 	"text/template"
 	stdlibtime "time"
 
@@ -17,6 +18,7 @@ import (
 	"github.com/ice-blockchain/eskimo/users"
 	wintrauth "github.com/ice-blockchain/wintr/auth"
 	"github.com/ice-blockchain/wintr/connectors/storage/v2"
+	storagev3 "github.com/ice-blockchain/wintr/connectors/storage/v3"
 	"github.com/ice-blockchain/wintr/email"
 	"github.com/ice-blockchain/wintr/time"
 )
@@ -29,9 +31,10 @@ type (
 	}
 	Client interface {
 		IceUserIDClient
-		SendSignInLinkToEmail(ctx context.Context, emailValue, deviceUniqueID, language, clientIP string) (loginSession string, err error)
+		SendSignInLinkToEmail(ctx context.Context, emailValue, deviceUniqueID, language, clientIP string) (queuePos int64, rateLimit, loginSession string, err error)
 		SignIn(ctx context.Context, loginFlowToken, confirmationCode string) (tokens *auth.Tokens, emailConfirmed bool, err error)
 		UpdateMetadata(ctx context.Context, userID string, metadata *users.JSON) (*users.JSON, error)
+		CheckHealth(ctx context.Context) error
 		RefreshToken(ctx context.Context, token *wintrauth.IceToken) (tokens *auth.Tokens, err error)
 	}
 	IceUserIDClient interface {
@@ -83,18 +86,24 @@ const (
 	sameIPCheckRate = 24 * stdlibtime.Hour
 
 	duplicatedSignInRequestsInLessThan = 2 * stdlibtime.Second
+	loginQueueKey                      = "login_queue"
+	loginRateLimitKey                  = "login_rate_limit"
+	initEmailRateLimit                 = "1000:1m"
 )
 
 type (
 	languageCode = string
 	client       struct {
+		queueDB            storagev3.DB
 		db                 *storage.DB
 		cfg                *config
 		shutdown           func() error
 		authClient         wintrauth.Client
 		userModifier       UserModifier
+		cancel             context.CancelFunc
 		emailClients       []email.Client
 		fromRecipients     []fromRecipient
+		queueWg            sync.WaitGroup
 		emailClientLBIndex uint64
 	}
 	config struct {
@@ -177,4 +186,5 @@ var (
 		modifyEmailType,
 		notifyEmailChangedType,
 	}
+	errAlreadyEnqueued = errors.New("already enqueued")
 )

auth/email_link/email_modify.go

@@ -44,7 +44,7 @@ func (c *client) handleEmailModification(ctx context.Context, els *emailLinkSign
 	if notifyEmail != "" {
 		now := time.Now()
 		resetConfirmationCode := generateConfirmationCode()
-		uErr := c.upsertEmailLinkSignIn(ctx, oldEmail, els.DeviceUniqueID, resetConfirmationCode, now)
+		uErr := c.upsertEmailLinkSignIn(ctx, oldEmail, els.DeviceUniqueID, resetConfirmationCode, els.Language, now)
 		if uErr != nil {
 			return multierror.Append( //nolint:wrapcheck // .
 				errors.Wrapf(c.resetEmailModification(ctx, usr.ID, oldEmail), "[reset] resetEmailModification failed for email:%v", oldEmail),

auth/email_link/emaillink.go

@@ -19,6 +19,7 @@ import (
 	"github.com/ice-blockchain/wintr/auth"
 	appcfg "github.com/ice-blockchain/wintr/config"
 	"github.com/ice-blockchain/wintr/connectors/storage/v2"
+	storagev3 "github.com/ice-blockchain/wintr/connectors/storage/v3"
 	"github.com/ice-blockchain/wintr/email"
 	"github.com/ice-blockchain/wintr/log"
 	"github.com/ice-blockchain/wintr/time"
@@ -29,15 +30,20 @@ func init() {
 	loadEmailMagicLinkTranslationTemplates()
 }
 
-func NewClient(ctx context.Context, userModifier UserModifier, authClient auth.Client) Client {
+//nolint:funlen // .
+func NewClient(ctx context.Context, cancel context.CancelFunc, userModifier UserModifier, authClient auth.Client) Client {
 	cfg := loadConfiguration()
 	cfg.validate()
 	db := storage.MustConnect(ctx, ddl, applicationYamlKey)
-
+	//nolint:contextcheck // Used in queue processing.
+	queueDB := storagev3.MustConnect(context.Background(), applicationYamlKey)
+	log.Panic(errors.Wrapf(queueDB.SetNX(ctx, loginRateLimitKey, initEmailRateLimit, 0).Err(), "failed to init email sending rate limit")) //nolint:revive // .
 	cl := &client{
 		cfg:            cfg,
 		shutdown:       db.Close,
 		db:             db,
+		cancel:         cancel,
+		queueDB:        queueDB,
 		authClient:     authClient,
 		userModifier:   userModifier,
 		emailClients:   make([]email.Client, 0, cfg.ExtraLoadBalancersCount+1),
@@ -52,6 +58,7 @@ func NewClient(ctx context.Context, userModifier UserModifier, authClient auth.C
 			cl.emailClients = append(cl.emailClients, email.New(fmt.Sprintf("%v/%v", applicationYamlKey, i+1)))
 			cl.fromRecipients = append(cl.fromRecipients, fromRecipient{nestedCfg.FromEmailName, nestedCfg.FromEmailAddress})
 		}
+		go cl.processEmailQueue(ctx)
 	}
 	go cl.startOldLoginAttemptsCleaner(ctx)
 
@@ -68,9 +75,33 @@ func NewROClient(ctx context.Context) IceUserIDClient {
 }
 
 func (c *client) Close() error {
+	if c.cancel != nil {
+		c.cancel()
+	}
+	c.queueWg.Wait()
+
 	return errors.Wrap(c.shutdown(), "closing auth/emaillink repository failed")
 }
 
+func (c *client) CheckHealth(ctx context.Context) error {
+	return errors.Wrapf(c.checkQueueDBHealth(ctx), "[health-check] failed to ping queueDB/dfly for email client")
+}
+
+func (c *client) checkQueueDBHealth(ctx context.Context) error {
+	if resp := c.queueDB.Ping(ctx); resp.Err() != nil || resp.Val() != "PONG" {
+		if resp.Err() == nil {
+			resp.SetErr(errors.Errorf("response `%v` is not `PONG`", resp.Val()))
+		}
+
+		return errors.Wrap(resp.Err(), "[health-check] failed to ping DB")
+	}
+	if !c.queueDB.IsRW(ctx) {
+		return errors.New("db is not writeable")
+	}
+
+	return nil
+}
+
 func loadConfiguration() *config {
 	var cfg config
 	appcfg.MustLoadFromKey(applicationYamlKey, &cfg)