refactor: cqrs improvements;

Author: jcbbb

.gitignore

@@ -4,5 +4,5 @@
 tmp/
 
 .env*
-nats/
+natsdata/
 yeahapi*conf

aws/email.go

@@ -0,0 +1,63 @@
+package aws
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/ses"
+	"github.com/aws/aws-sdk-go-v2/service/ses/types"
+	"github.com/nats-io/nats.go/jetstream"
+	yeahapi "github.com/yeahuz/yeah-api"
+)
+
+type EmailService struct {
+	ses     *ses.Client
+	cqrssrv yeahapi.CQRSService
+}
+
+func NewEmailService(cfg aws.Config, cqrssrv yeahapi.CQRSService) *EmailService {
+	ses := ses.NewFromConfig(cfg)
+	return &EmailService{
+		ses:     ses,
+		cqrssrv: cqrssrv,
+	}
+}
+
+func (e *EmailService) HandleSendEmailCode(m jetstream.Msg) error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	var cmd yeahapi.SendEmailCodeCmd
+	if err := json.Unmarshal(m.Data(), &cmd); err != nil {
+		return err
+	}
+
+	_, err := e.ses.SendEmail(ctx, &ses.SendEmailInput{
+		Destination: &types.Destination{
+			ToAddresses: []string{
+				cmd.Email,
+			},
+		},
+		Source: aws.String("Needs <[email protected]>"),
+
+		Message: &types.Message{
+			Subject: &types.Content{
+				Data: aws.String(fmt.Sprintf("Your code: %s", cmd.Code)),
+			},
+			Body: &types.Body{
+				Text: &types.Content{Data: aws.String(cmd.Code)},
+			},
+		},
+	})
+
+	if err != nil {
+		return err
+	}
+
+	e.cqrssrv.Publish(ctx, yeahapi.NewEmailCodeSentEvent(cmd.Email))
+
+	return nil
+}

cmd/yeahapi/main.go

@@ -10,11 +10,16 @@ import (
 	"path/filepath"
 	"strings"
 
+	awsconf "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/jackc/pgx/v5/pgxpool"
+
 	"github.com/pelletier/go-toml/v2"
 	yeahapi "github.com/yeahuz/yeah-api"
+	"github.com/yeahuz/yeah-api/aws"
 	"github.com/yeahuz/yeah-api/http"
 	"github.com/yeahuz/yeah-api/inmem"
+	"github.com/yeahuz/yeah-api/nats"
 	"github.com/yeahuz/yeah-api/postgres"
 )
 
@@ -88,18 +93,30 @@ func (m *Main) Run(ctx context.Context) (err error) {
 	userService := postgres.NewUserService(m.Pool)
 	localizerService := yeahapi.NewLocalizerService("en")
 
-	// cqrsService := nats.NewCQRSService(ctx, yeahapi.CQRSConfig{
-	// 	NatsURL:       m.Config.Nats.URL,
-	// 	NatsAuthToken: m.Config.Nats.AuthToken,
-	// 	Streams:       map[string][]string{},
-	// })
+	cqrsService, err := nats.NewCQRSService(ctx, yeahapi.CQRSConfig{
+		NatsURL:       m.Config.Nats.URL,
+		NatsAuthToken: m.Config.Nats.AuthToken,
+		Streams:       m.Config.Nats.Streams,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	awsconfig, err := awsconf.LoadDefaultConfig(ctx,
+		awsconf.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(m.Config.AWS.Key, m.Config.AWS.Secret, "")),
+		awsconf.WithRegion("en-north-1"),
+	)
+
+	emailService := aws.NewEmailService(awsconfig, cqrsService)
+	cqrsService.Handle("auth.sendEmailCode", emailService.HandleSendEmailCode)
 
 	m.Server.Addr = m.Config.HTTP.Addr
 
 	m.Server.UserService = userService
 	m.Server.AuthService = authService
 	m.Server.LocalizerService = localizerService
-	// m.Server.CQRSService = cqrsService
+	m.Server.CQRSService = cqrsService
 
 	if err := m.Server.Open(); err != nil {
 		return err
@@ -109,16 +126,16 @@ func (m *Main) Run(ctx context.Context) (err error) {
 }
 
 func (m *Main) Close() error {
+	if m.Pool != nil {
+		m.Pool.Close()
+	}
+
 	if m.Server != nil {
 		if err := m.Server.Close(); err != nil {
 			return err
 		}
 	}
 
-	if m.Pool != nil {
-		m.Pool.Close()
-	}
-
 	return nil
 }
 
@@ -167,9 +184,9 @@ type Config struct {
 	} `toml:"highwayhash"`
 
 	Nats struct {
-		AuthToken string            `toml:"auth-token"`
-		URL       string            `toml:"url"`
-		Streams   map[string]string `toml:"streams"`
+		AuthToken string              `toml:"auth-token"`
+		URL       string              `toml:"url"`
+		Streams   map[string][]string `toml:"streams"`
 	} `toml:"nats"`
 
 	Google struct {

cqrs.go

@@ -2,6 +2,15 @@ package yeahapi
 
 import (
 	"context"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+const (
+	sendPhoneCode = "auth.sendPhoneCode"
+	sendEmailCode = "auth.sendEmailCode"
+	emailCodeSent = "auth.emailCodeSent"
+	phoneCodeSent = "auth.phoneCodeSent"
 )
 
 type CQRSConfig struct {
@@ -14,6 +23,70 @@ type CQRSMessage interface {
 	Subject() string
 }
 
+type CQRSHandler func(message jetstream.Msg) error
+
 type CQRSService interface {
-	Send(ctx context.Context, message CQRSMessage) error
+	Publish(ctx context.Context, message CQRSMessage) error
+	Handle(subject string, handler CQRSHandler)
+	Close() error
+}
+
+type subject struct {
+	subject string
+}
+
+func (s subject) Subject() string {
+	return s.subject
+}
+
+type SendPhoneCodeCmd struct {
+	subject
+	PhoneNumber string `json:"phone_number"`
+	Code        string `json:"code"`
+}
+
+type EmailCodeSentEvent struct {
+	subject
+	Email string `json:"email"`
+}
+
+type PhoneCodeSentEvent struct {
+	subject
+	PhoneNumber string `json:"phone_number"`
+}
+
+type SendEmailCodeCmd struct {
+	subject
+	Email string `json:"email"`
+	Code  string `json:"code"`
+}
+
+func NewSendPhoneCodeCmd(phoneNumber string, code string) SendPhoneCodeCmd {
+	return SendPhoneCodeCmd{
+		subject:     subject{sendPhoneCode},
+		PhoneNumber: phoneNumber,
+		Code:        code,
+	}
+}
+
+func NewSendEmailCodeCmd(email string, code string) SendEmailCodeCmd {
+	return SendEmailCodeCmd{
+		subject: subject{sendEmailCode},
+		Email:   email,
+		Code:    code,
+	}
+}
+
+func NewEmailCodeSentEvent(email string) EmailCodeSentEvent {
+	return EmailCodeSentEvent{
+		subject: subject{emailCodeSent},
+		Email:   email,
+	}
+}
+
+func NewPhoneCodeSentEvent(phoneNumber string) PhoneCodeSentEvent {
+	return PhoneCodeSentEvent{
+		subject:     subject{phoneCodeSent},
+		PhoneNumber: phoneNumber,
+	}
 }

db/pool.go

@@ -4,36 +4,38 @@ go 1.21
 
 require (
 	github.com/aws/aws-sdk-go v1.48.9
+	github.com/aws/aws-sdk-go-v2 v1.24.0
+	github.com/aws/aws-sdk-go-v2/config v1.26.2
 	github.com/aws/aws-sdk-go-v2/service/ses v1.19.1
 	github.com/gofrs/uuid v4.4.0+incompatible
 	github.com/jackc/pgx/v5 v5.5.0
-	github.com/joho/godotenv v1.5.1
 	github.com/minio/highwayhash v1.0.2
 	github.com/nats-io/nats.go v1.31.0
 	github.com/pelletier/go-toml/v2 v2.1.1
 	golang.org/x/crypto v0.16.0
-	golang.org/x/oauth2 v0.15.0
 	golang.org/x/text v0.14.0
 )
 
 require (
-	cloud.google.com/go/compute v1.20.1 // indirect
-	cloud.google.com/go/compute/metadata v0.2.3 // indirect
-	github.com/aws/aws-sdk-go-v2 v1.23.4 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.7 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.7 // indirect
-	github.com/aws/smithy-go v1.18.1 // indirect
-	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.16.13 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.10 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.9 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.9 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.7.2 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.9 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.18.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.21.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.26.6 // indirect
+	github.com/aws/smithy-go v1.19.0 // indirect
+	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/klauspost/compress v1.17.0 // indirect
 	github.com/nats-io/nkeys v0.4.5 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	golang.org/x/net v0.19.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.15.0 // indirect
-	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/protobuf v1.31.0 // indirect
 )