Feature/1 (#20)

* users schema update: added email column, added indexes

* tests update

* user repo GetByEmail test added

* user email prop added

* new google oauth login will result in new user

* reset password handler

* username vo update

* test fix

* switched from bcrypt to argon2id

* fixed a bug

* added Delete method to cache interface

* UserPasswordChanged event added

* logging fixed

* removed auth context

Author: NLipatov

src/application/cache.go

@@ -5,6 +5,7 @@ import "time"
 type Cache[T any] interface {
 	Get(key string) (T, error)
 	Set(key string, value T) error
+	Delete(key string) error
 }
 
 type CacheWithTTL[T any] interface {

src/application/commands/post_user.go

@@ -4,5 +4,6 @@ import "goproxy/domain/valueobjects"
 
 type PostUser struct {
 	Username string
+	Email    string
 	Password valueobjects.Password
 }

src/application/crypto_service.go

@@ -1,7 +1,7 @@
 package application
 
 type CryptoService interface {
-	GenerateSalt() ([]byte, error)
-	HashValue(value string, salt []byte) ([]byte, error)
-	ValidateHash(hash []byte, salt []byte, password string) bool
+	GenerateRandomString(length int) (string, error)
+	HashValue(value string) (string, error)
+	ValidateHash(fullHash, password string) bool
 }

src/application/repository.go

@@ -15,6 +15,7 @@ type Repository[T any] interface {
 type UserRepository interface {
 	Repository[aggregates.User]
 	GetByUsername(username string) (aggregates.User, error)
+	GetByEmail(email string) (aggregates.User, error)
 }
 
 type EventRepository interface {

src/application/user_use_cases.go

@@ -31,18 +31,17 @@ func (u UserUseCases) GetById(id int) (aggregates.User, error) {
 	return u.repo.GetById(id)
 }
 
-func (u UserUseCases) Create(command commands.PostUser) (int, error) {
-	salt, err := u.cryptoService.GenerateSalt()
-	if err != nil {
-		return 0, err
-	}
+func (u UserUseCases) GetByEmail(email string) (aggregates.User, error) {
+	return u.repo.GetByEmail(email)
+}
 
-	hash, err := u.cryptoService.HashValue(command.Password.Value, salt)
+func (u UserUseCases) Create(command commands.PostUser) (int, error) {
+	hash, err := u.cryptoService.HashValue(command.Password.Value)
 	if err != nil {
 		return 0, err
 	}
 
-	user, err := aggregates.NewUser(-1, command.Username, hash, salt)
+	user, err := aggregates.NewUser(-1, command.Username, command.Email, hash)
 	if err != nil {
 		return 0, err
 	}
@@ -66,7 +65,7 @@ func (u UserUseCases) Delete(dto commands.DeleteUser) error {
 		return err
 	}
 
-	isPasswordValid := u.cryptoService.ValidateHash(user.PasswordHash(), user.PasswordSalt(), dto.Password.Value)
+	isPasswordValid := u.cryptoService.ValidateHash(user.PasswordHash(), dto.Password.Value)
 	if !isPasswordValid {
 		return fmt.Errorf("invalid password")
 	}

src/dal/db.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"fmt"
 	_ "github.com/lib/pq"
+	"log"
 	"os"
 )
 
@@ -22,13 +23,28 @@ func ConnectDB() (*sql.DB, error) {
 
 func buildConnectionString() (string, error) {
 	dbUser := os.Getenv("DB_USER")
+	if dbUser == "" {
+		log.Fatalf("DB_USER environment variable not set")
+	}
+
 	dbPass := os.Getenv("DB_PASS")
+	if dbPass == "" {
+		log.Fatalf("DB_PASS environment variable not set")
+	}
+
 	dbHost := os.Getenv("DB_HOST")
+	if dbHost == "" {
+		log.Fatalf("DB_HOST environment variable not set")
+	}
+
 	dbPort := os.Getenv("DB_PORT")
-	dbDatabase := os.Getenv("DB_DATABASE")
+	if dbPort == "" {
+		log.Fatalf("DB_PORT environment variable not set")
+	}
 
-	if dbUser == "" || dbPass == "" || dbHost == "" || dbPort == "" || dbDatabase == "" {
-		return "", fmt.Errorf("invalid db credentials")
+	dbDatabase := os.Getenv("DB_DATABASE")
+	if dbDatabase == "" {
+		log.Fatalf("DB_DATABASE environment variable not set")
 	}
 
 	dataSourceName := fmt.

src/dal/migrations/proxydb/v2_users_table.sql

@@ -1,7 +1,10 @@
 CREATE TABLE public.users (
     id SERIAL PRIMARY KEY,
     username VARCHAR(255) NOT NULL UNIQUE,
-    password_hash BYTEA NOT NULL,
-    password_salt BYTEA NOT NULL,
+    email VARCHAR(255) NOT NULL UNIQUE,
+    password_hash VARCHAR(256) NOT NULL,
     created_at TIMESTAMP DEFAULT now()
 );
+
+CREATE INDEX idx_email on users(email);
+CREATE INDEX idx_username on users(username);

src/dal/repositories/mocks/message_bus.go

@@ -0,0 +1,89 @@
+package mocks
+
+import (
+	"errors"
+	"goproxy/domain/events"
+	"sync"
+)
+
+type MockMessageBusService struct {
+	topics    map[string][]events.OutboxEvent
+	mutex     sync.RWMutex
+	isClosed  bool
+	consumeCh chan events.OutboxEvent
+}
+
+func NewMockMessageBusService() *MockMessageBusService {
+	return &MockMessageBusService{
+		topics:    make(map[string][]events.OutboxEvent),
+		consumeCh: make(chan events.OutboxEvent, 100),
+	}
+}
+
+func (m *MockMessageBusService) Subscribe(topics []string) error {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if m.isClosed {
+		return errors.New("message bus is closed")
+	}
+
+	for _, topic := range topics {
+		if _, exists := m.topics[topic]; !exists {
+			m.topics[topic] = []events.OutboxEvent{}
+		}
+	}
+
+	return nil
+}
+
+func (m *MockMessageBusService) Consume() (*events.OutboxEvent, error) {
+	m.mutex.RLock()
+	defer m.mutex.RUnlock()
+
+	if m.isClosed {
+		return nil, errors.New("message bus is closed")
+	}
+
+	event, ok := <-m.consumeCh
+	if !ok {
+		return nil, errors.New("no events to consume")
+	}
+
+	return &event, nil
+}
+
+func (m *MockMessageBusService) Produce(topic string, event events.OutboxEvent) error {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if m.isClosed {
+		return errors.New("message bus is closed")
+	}
+
+	if _, exists := m.topics[topic]; !exists {
+		return errors.New("topic not found")
+	}
+
+	m.topics[topic] = append(m.topics[topic], event)
+
+	select {
+	case m.consumeCh <- event:
+	default:
+	}
+
+	return nil
+}
+
+func (m *MockMessageBusService) Close() error {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if m.isClosed {
+		return errors.New("message bus is already closed")
+	}
+
+	m.isClosed = true
+	close(m.consumeCh)
+	return nil
+}