impl sending verification email

db/migrations/000004_add_verify_email.down.sql

@@ -0,0 +1,10 @@
+-- Drop the foreign key constraint on 'verify_emails'
+ALTER TABLE verify_emails
+DROP CONSTRAINT IF EXISTS fk_verify_emails_users;
+
+-- Drop the 'verify_emails' table
+DROP TABLE IF EXISTS verify_emails;
+
+-- Remove the 'is_email_verified' column from 'users'
+ALTER TABLE users
+DROP COLUMN IF EXISTS is_email_verified;

db/migrations/000004_add_verify_email.up.sql

@@ -0,0 +1,15 @@
+ALTER TABLE users ADD COLUMN is_email_verified BOOLEAN DEFAULT false;
+
+CREATE TABLE "verify_emails"
+(
+    "id" bigserial PRIMARY KEY,
+    "username" varchar NOT NULL,
+    "email" varchar NOT NULL,
+    "secret_code" varchar NOT NULL,
+    "is_used" boolean NOT NULL DEFAULT false,
+    "created_at" timestamptz NOT NULL DEFAULT (now()),
+    "expired_at" timestamptz NOT NULL DEFAULT (now() + INTERVAL '15 minutes')
+);
+
+ALTER TABLE "verify_emails" ADD FOREIGN KEY ("username") REFERENCES "users" ("username");
+

db/migrations/procs/user/create_user.sql

@@ -11,7 +11,8 @@ RETURNS TABLE (
     fullname VARCHAR,
     email VARCHAR,
     password_changed_at TIMESTAMP WITH TIME ZONE,
-    created_at TIMESTAMP WITH TIME ZONE
+    created_at TIMESTAMP WITH TIME ZONE,
+    is_email_verified BOOLEAN
 ) AS $$
 BEGIN
     RETURN QUERY

db/migrations/procs/user/create_verify_email.sql

@@ -0,0 +1,30 @@
+DROP FUNCTION IF EXISTS create_verify_email;
+CREATE OR REPLACE FUNCTION create_verify_email(
+    p_username VARCHAR,
+    p_email VARCHAR,
+    p_secret_code VARCHAR
+)
+RETURNS TABLE (
+    id bigint,
+    username VARCHAR,
+    email VARCHAR,
+    secret_code VARCHAR,
+    is_used BOOLEAN,
+    created_at TIMESTAMP WITH TIME ZONE,
+    expires_at_at TIMESTAMP WITH TIME ZONE
+) AS $$
+BEGIN
+    RETURN QUERY
+    INSERT INTO verify_emails (username, email, secret_code)
+    VALUES (p_username, p_email, p_secret_code)
+    RETURNING
+        verify_emails.id,
+        verify_emails.username,
+        verify_emails.email,
+        verify_emails.secret_code,
+        verify_emails.is_used,
+        verify_emails.created_at,
+        verify_emails.expired_at;
+
+END;
+$$ LANGUAGE plpgsql;

db/migrations/procs/user/get_user.sql

@@ -8,7 +8,8 @@ RETURNS TABLE (
     fullname VARCHAR,
     email VARCHAR,
     password_changed_at TIMESTAMP WITH TIME ZONE,
-    created_at TIMESTAMP WITH TIME ZONE
+    created_at TIMESTAMP WITH TIME ZONE,
+    is_email_verified BOOLEAN
 ) AS $$
 BEGIN
     RETURN QUERY
@@ -18,7 +19,8 @@ BEGIN
         users.fullname,
         users.email,
         users.password_changed_at,
-        users.created_at
+        users.created_at,
+        users.is_email_verified
     FROM users
     WHERE users.username = p_username
     LIMIT 1;

db/migrations/procs/user/update_user.sql

@@ -12,7 +12,8 @@ RETURNS TABLE (
     fullname VARCHAR,
     email VARCHAR,
     password_changed_at TIMESTAMP WITH TIME ZONE,
-    created_at TIMESTAMP WITH TIME ZONE
+    created_at TIMESTAMP WITH TIME ZONE,
+    is_email_verfied BOOLEAN
 ) AS $$
 BEGIN
     RETURN QUERY

db/store/migrate.go

@@ -19,7 +19,6 @@ func RunDBMigrations(db *sql.DB, migrationsURL string) {
 
 	driver, err := pgx.WithInstance(db, &pgx.Config{})
 	if err != nil {
-		// log.Fatal().Msg("cannot create postgres driver")
 		slog.Error("cannot create postgres driver %s", err)
 	}
 	migration, err := migrate.NewWithDatabaseInstance(
@@ -31,12 +30,10 @@ func RunDBMigrations(db *sql.DB, migrationsURL string) {
 	}
 	migration.Up()
 	if err = migration.Up(); err != nil && err != migrate.ErrNoChange {
-		// log.Fatal().Msg("failed to run migrate up")
 		slog.Error("failed to run migrate up %s", err)
 
 	}
 
-	// log.Info().Msg("DB migrated successfully")
 	slog.Info("DB migrated successfully")
 
 	// Run unversioned migrations

db/store/models.go

@@ -49,4 +49,15 @@ type User struct {
 	Email             string    `json:"email"`
 	PasswordChangedAt time.Time `json:"password_changed_at"`
 	CreatedAt         time.Time `json:"created_at"`
+	IsEmailVerified   bool      `json:"is_email_verified"`
+}
+
+type VerifyEmail struct {
+	ID         int64     `json:"id"`
+	Username   string    `json:"username"`
+	Email      string    `json:"email"`
+	SecretCode string    `json:"secret_code"`
+	IsUsed     bool      `json:"is_used"`
+	CreatedAt  time.Time `json:"created_at"`
+	ExpiredAt  time.Time `json:"expired_at"`
 }

db/store/querier.go

@@ -8,10 +8,11 @@ import (
 
 type Querier interface {
 	CreateUser(ctx context.Context, arg CreateUserParams) (User, error)
-	GetSession(ctx context.Context, id uuid.UUID) (Session, error)
-	CreateSession(ctx context.Context, arg CreateSessionParams) (Session, error)
 	GetUser(ctx context.Context, username string) (User, error)
 	UpdateUser(ctx context.Context, arg UpdateUserParams) (User, error)
+	GetSession(ctx context.Context, id uuid.UUID) (Session, error)
+	CreateSession(ctx context.Context, arg CreateSessionParams) (Session, error)
+	CreateVerifyEmail(ctx context.Context, arg CreateVerifyEmail) (VerifyEmail, error)
 }
 
 var _ Querier = (*Queries)(nil)

db/store/user.go

@@ -74,6 +74,7 @@ func scanUserFromRow(row *sql.Row, user *User) error {
 		&user.Email,
 		&user.PasswordChangedAt,
 		&user.CreatedAt,
+		&user.IsEmailVerified,
 	)
 
 	// Check for errors after scanning

db/store/verify_email.go

@@ -0,0 +1,93 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+	"log"
+)
+
+type CreateVerifyEmail struct {
+	Username   string `json:"username"`
+	Email      string `json:"email"`
+	SecretCode string `json:"secret_code"`
+}
+
+func (q *Queries) CreateVerifyEmail(ctx context.Context, arg CreateVerifyEmail) (VerifyEmail, error) {
+	var verifyEmail VerifyEmail
+	row := q.callStoredFunction(ctx, "create_verify_email",
+		arg.Username,
+		arg.Email,
+		arg.SecretCode,
+	)
+
+	err := scanVerifyEmailFromRow(row, &verifyEmail)
+	if err != nil {
+		return verifyEmail, err
+	}
+	return verifyEmail, nil
+}
+
+// func (q *Queries) GetUser(ctx context.Context, username string) (User, error) {
+// 	var user User
+// 	row := q.callStoredFunction(ctx, "get_user", username)
+// 	err := scanUserFromRow(row, &user)
+// 	if err != nil {
+// 		return user, err
+// 	}
+// 	return user, nil
+// }
+
+// type UpdateUserParams struct {
+// 	HashedPassword    sql.NullString `json:"hashed_password"`
+// 	PasswordChangedAt sql.NullTime   `json:"password_changed_at"`
+// 	FullName          sql.NullString `json:"fullname"`
+// 	Email             sql.NullString `json:"email"`
+// 	Username          string         `json:"username"`
+// }
+
+// func (q *Queries) UpdateUser(ctx context.Context, arg UpdateUserParams) (User, error) {
+// 	var user User
+// 	params := []interface{}{
+// 		arg.Username,
+// 		arg.HashedPassword,
+// 		arg.PasswordChangedAt,
+// 		arg.FullName,
+// 		arg.Email,
+// 	}
+// 	row := q.callStoredFunction(ctx, "update_user", params...)
+
+// 	// Execute the stored procedure and scan the results into the variables
+// 	err := scanUserFromRow(row, &user)
+// 	if err != nil {
+// 		return User{}, err
+// 	}
+
+// 	return user, nil
+// }
+
+func scanVerifyEmailFromRow(row *sql.Row, verifyEmail *VerifyEmail) error {
+	err := row.Scan(
+		&verifyEmail.ID,
+		&verifyEmail.Username,
+		&verifyEmail.Email,
+		&verifyEmail.SecretCode,
+		&verifyEmail.IsUsed,
+		&verifyEmail.CreatedAt,
+		&verifyEmail.ExpiredAt,
+	)
+
+	// Check for errors after scanning
+	if err != nil {
+		// Handle scan-related errors
+		if err == sql.ErrNoRows {
+			// fmt.Println("No rows were returned.")
+			return err
+		} else {
+			// Log and return other scan-related errors
+			log.Printf("Error scanning row: %s", err)
+			return err
+		}
+	}
+
+	return nil
+}

main.go

@@ -12,6 +12,7 @@ import (
 
 	db "github.com/aradwann/eenergy/db/store"
 	"github.com/aradwann/eenergy/gapi"
+	"github.com/aradwann/eenergy/mail"
 	"github.com/aradwann/eenergy/pb"
 	"github.com/aradwann/eenergy/util"
 	"github.com/aradwann/eenergy/worker"
@@ -43,12 +44,11 @@ func main() {
 	runDBMigrations(dbConn, config.MigrationsURL)
 
 	store := db.NewStore(dbConn)
-
 	redisOpts := asynq.RedisClientOpt{
 		Addr: config.RedisAddress,
 	}
 	taskDistributor := worker.NewRedisTaskDistributor(redisOpts)
-	go runTaskProcessor(redisOpts, store)
+	go runTaskProcessor(config, redisOpts, store)
 	go runGatewayServer(config, store, taskDistributor)
 	runGrpcServer(config, store, taskDistributor)
 }
@@ -142,8 +142,10 @@ func runGatewayServer(config util.Config, store db.Store, taskDistributor worker
 	}
 }
 
-func runTaskProcessor(redisOpts asynq.RedisClientOpt, store db.Store) {
-	taskProcessor := worker.NewRedisTaskProcessor(redisOpts, store)
+func runTaskProcessor(config util.Config, redisOpts asynq.RedisClientOpt, store db.Store) {
+	mailer := mail.NewGmailSender(config.EmailSenderName, config.EmailSenderAddress, config.EmailSenderPassword)
+
+	taskProcessor := worker.NewRedisTaskProcessor(redisOpts, store, mailer)
 	slog.Info("start task processor")
 	err := taskProcessor.Start()
 	if err != nil {

worker/processor.go

@@ -5,6 +5,7 @@ import (
 	"log/slog"
 
 	db "github.com/aradwann/eenergy/db/store"
+	"github.com/aradwann/eenergy/mail"
 	"github.com/hibiken/asynq"
 )
 
@@ -22,9 +23,10 @@ type TaskProcessor interface {
 type RedisTaskProcessor struct {
 	server *asynq.Server
 	store  db.Store
+	mailer mail.EmailSender
 }
 
-func NewRedisTaskProcessor(redisOpt asynq.RedisClientOpt, store db.Store) TaskProcessor {
+func NewRedisTaskProcessor(redisOpt asynq.RedisClientOpt, store db.Store, mailer mail.EmailSender) TaskProcessor {
 	server := asynq.NewServer(redisOpt, asynq.Config{
 		Queues: map[string]int{
 			QueueCritical: 6,
@@ -45,6 +47,7 @@ func NewRedisTaskProcessor(redisOpt asynq.RedisClientOpt, store db.Store) TaskPr
 	return &RedisTaskProcessor{
 		server: server,
 		store:  store,
+		mailer: mailer,
 	}
 }
 

worker/task_send_verify_email.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"log/slog"
 
+	db "github.com/aradwann/eenergy/db/store"
+	"github.com/aradwann/eenergy/util"
 	"github.com/hibiken/asynq"
 )
 
@@ -55,7 +57,25 @@ func (processor *RedisTaskProcessor) ProcessTaskSendVerifyEmail(ctx context.Cont
 		// }
 		return fmt.Errorf("failed to get user: %w", asynq.SkipRetry)
 	}
-	// TODO: send email to user
+	createVerifyEmailParams := db.CreateVerifyEmail{
+		Username:   user.Username,
+		Email:      user.Email,
+		SecretCode: util.RandomString(32),
+	}
+	verfiyEmail, err := processor.store.CreateVerifyEmail(ctx, createVerifyEmailParams)
+	if err != nil {
+		return fmt.Errorf("failed to create verify email instance: %w", err)
+	}
+	subject := "Welcome to Eennergy"
+	verifyURL := fmt.Sprintf("http://eenergy.io/verify_email?id=%d&secret_code=%s", verfiyEmail.ID, verfiyEmail.SecretCode)
+	content := fmt.Sprintf(`Hello %s, <br/>
+	Thank you for being a member in Eenergy community!</br>
+	Pleas click on <a href="%s">click here</a> to verify your email`, user.FullName, verifyURL)
+	to := []string{user.Email}
+	err = processor.mailer.SendEmail(subject, content, to, nil, nil, nil)
+	if err != nil {
+		return fmt.Errorf("failed to send verification email: %w", err)
+	}
 	slog.LogAttrs(context.Background(),
 		slog.LevelInfo,
 		"processed task",