Добавил хранение в arango

.vscode/launch.json

@@ -15,6 +15,10 @@
         "KAFKA_PORT": "19092",
         "KAFKA_HOST": "37.143.12.169",
         // "SENTRY_DSN": "https://3f4b31dbbd9a4a6b8a71f9881d962f25@o4504654569799680.ingest.sentry.io/4504654572683264"
+        "ARANGO_PORT": "8530",
+        "ARANGO_HOST": "127.0.0.1",
+        "ARANGO_USER": "root",
+        "ARANGO_PASSWORD": "dbpassword",
       },
       "args": []
     }

internal/authDB.go

@@ -2,15 +2,19 @@ package authDB
 
 import (
 	"context"
+	"crypto/md5"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
+	"log"
 
 	"net"
 	"strconv"
-	"sync"
 	"time"
 
 	"git.slaventius.ru/test3k/authDB/internal/config"
+	repo "git.slaventius.ru/test3k/authDB/internal/customer"
+	arango "git.slaventius.ru/test3k/authDB/internal/transport/arango"
 	kafka "git.slaventius.ru/test3k/authDB/internal/transport/kafka"
 	api "git.slaventius.ru/test3k/umate/pkg/api"
 	apiKafka "git.slaventius.ru/test3k/umate/pkg/kafka"
@@ -18,20 +22,10 @@ import (
 	// health "google.golang.org/grpc/health/grpc_health_v1"
 )
 
-type user struct {
-	ID        int32
-	Login     string
-	Password  string
-	Confirmed bool
-	Time      time.Time
-	apiKafka.MessageRegistration
-}
-
 type AuthDBServer struct {
-	mu          sync.Mutex
-	users       map[string]*user
 	kafkaWriter *kafka.KafkaWriter
 	logger      *logger.Logger
+	users       *repo.CustomerRepository
 	api.UnimplementedAuthDBServer
 	api.UnimplementedHealthServer
 	ctx context.Context
@@ -39,11 +33,20 @@ type AuthDBServer struct {
 }
 
 func NewServer(ctx context.Context, config *config.Config) *AuthDBServer {
+	conn := arango.NewConnection(ctx, config.Arango.Host, config.Arango.Port)
+	client := arango.NewClient(ctx, conn, config.Arango.User, config.Arango.Password)
+	arangoDB := arango.NewDataBase(ctx, client, "test3k")
 	logger := logger.NewLogger("test3k:authDBService", config.Sentry.DSN)
+	users := repo.NewCustomerRepository(ctx, arangoDB)
+
+	//
+	err := users.Fetch()
+	if err != nil {
+		log.Fatal(err)
+	}
 
 	return &AuthDBServer{
-		mu:          sync.Mutex{},
+		users:       users,
-		users:       make(map[string]*user),
 		kafkaWriter: kafka.NewWriter(ctx, logger, apiKafka.TopicRegistrations, net.JoinHostPort(config.Kafka.Host, strconv.Itoa(config.Kafka.Port))),
 		logger:      logger,
 		ctx:         ctx,
@@ -51,16 +54,18 @@ func NewServer(ctx context.Context, config *config.Config) *AuthDBServer {
 	}
 }
 
+func (r *AuthDBServer) getMD5Hash(text string) string {
+	hash := md5.Sum([]byte(text))
+
+	return hex.EncodeToString(hash[:])
+}
+
 func (s *AuthDBServer) GracefulStop() error {
 	return s.kafkaWriter.Close()
 }
 
 func (s *AuthDBServer) Login(ctx context.Context, req *api.LoginRequest) (*api.LoginResponse, error) {
-	s.mu.Lock()
+	user, ok := s.users.Customers[req.GetLogin()]
-	defer s.mu.Unlock()
-
-	//
-	user, ok := s.users[req.GetLogin()]
 	if !ok {
 		return nil, errors.New("login unknown")
 	}
@@ -71,59 +76,51 @@ func (s *AuthDBServer) Login(ctx context.Context, req *api.LoginRequest) (*api.L
 	}
 
 	//
-	if user.Password != req.Password {
+	if user.Password != s.getMD5Hash(req.Password) {
 		return nil, errors.New("password incorrect")
 	}
 
 	return &api.LoginResponse{
-		ID: user.ID,
+		ID: 0,
 	}, nil
 }
 
 func (s *AuthDBServer) Registration(ctx context.Context, req *api.RegistrationRequest) (*api.RegistrationResponse, error) {
-	s.mu.Lock()
+	user, ok := s.users.Customers[req.GetLogin()]
-	defer s.mu.Unlock()
+	if !ok {
+		hash := s.getMD5Hash(req.GetPassword())
+		tmpUser, eru := s.users.NewCustomer(req.GetLogin(), hash, req.GetEmail(), strconv.Itoa(time.Now().Nanosecond()))
+		if eru != nil {
+			s.logger.Error(eru)
 
-	//
+			return nil, eru
-	if val, ok := s.users[req.GetLogin()]; ok {
-		if time.Now().Before(val.Time) {
-			return nil, errors.New("login already registered")
 		}
-	} else {
-		s.id = s.id + 1
-	}
 
-	//
+		user = tmpUser
-	user := &user{
+	} else if user.Confirmed || time.Now().Before(user.Time) {
-		ID:        s.id,
+		return nil, errors.New("login already registered")
-		Login:     req.GetLogin(),
+	} else { // Обновим время регистрации
-		Password:  req.GetPassword(),
+		user.Refresh()
-		Confirmed: false,
-		Time:      time.Now().Add(time.Minute * 15),
-		MessageRegistration: apiKafka.MessageRegistration{
-			Code:  strconv.Itoa(time.Now().Nanosecond()),
-			Email: req.GetEmail(),
-		},
 	}
 
 	// TODO
-	value, eru := json.Marshal(user.MessageRegistration)
+	_, era := json.Marshal(user.MessageRegistration)
-	if eru != nil {
+	if era != nil {
-		return nil, eru
+		return nil, era
 	}
 
 	//
 	s.logger.Printf("publication code %s to %s ...", user.MessageRegistration.Code, user.MessageRegistration.Email)
 
-	//
+	// //
-	err := s.kafkaWriter.WriteMessage([]byte(user.Login), value)
+	// err := s.kafkaWriter.WriteMessage([]byte(user.Login), value)
-	if err != nil {
+	// if err != nil {
-		s.logger.Error(err)
+	// 	s.logger.Error(err)
 
-		return nil, err
+	// 	return nil, err
-	} else {
+	// } else {
-		s.users[req.Login] = user
+	s.users.Customers[req.Login] = user
-	}
+	// }
 
 	//
 	s.logger.Printf("publication code %s to %s completed", user.MessageRegistration.Code, user.MessageRegistration.Email)
@@ -135,21 +132,22 @@ func (s *AuthDBServer) Registration(ctx context.Context, req *api.RegistrationRe
 }
 
 func (s *AuthDBServer) Confirmation(ctx context.Context, req *api.ConfirmationRequest) (*api.ConfirmationResponse, error) {
-	s.mu.Lock()
+	for _, x := range s.users.Customers {
-	defer s.mu.Unlock()
-
-	//
-	for _, x := range s.users {
 		if x.Code == req.GetCode() {
 			if x.Confirmed {
 				return nil, errors.New("already confirmed")
 			}
 
 			//
-			x.Confirmed = true
+			err := x.Confirm()
+			if err != nil {
+				return nil, err
+			} else { // Что бы не перечитывать из БД, обновим локальную копию
+				s.users.Customers[x.Login] = x
+			}
 
 			return &api.ConfirmationResponse{
-				ID: x.ID,
+				ID: 0,
 			}, nil
 		}
 	}

internal/config/config.go

@@ -15,6 +15,13 @@ type kafkaConfig struct {
 	Host string `envconfig:"KAFKA_HOST"`
 }
 
+type arangoConfig struct {
+	Port     int    `envconfig:"ARANGO_PORT"`
+	Host     string `envconfig:"ARANGO_HOST"`
+	User     string `envconfig:"ARANGO_USER"`
+	Password string `envconfig:"ARANGO_PASSWORD"`
+}
+
 type sentryConfig struct {
 	DSN string `envconfig:"SENTRY_DSN"`
 }
@@ -24,6 +31,7 @@ type Config struct {
 	App    appConfig
 	Kafka  kafkaConfig
 	Sentry sentryConfig
+	Arango arangoConfig
 }
 
 func NewConfig() *Config {

internal/customer/customer.go

@@ -0,0 +1,39 @@
+package customer
+
+import (
+	"time"
+
+	apiKafka "git.slaventius.ru/test3k/umate/pkg/kafka"
+)
+
+// Покупатель
+type Customer struct {
+	SimpleRow
+	Login     string
+	Password  string
+	Confirmed bool
+	apiKafka.MessageRegistration
+	Time time.Time
+}
+
+func (c *Customer) Update() error {
+	_, err := c.collection.UpdateDocument(c.ctx, c.key, c)
+
+	return err
+}
+
+func (c *Customer) Delete() error {
+	return c.SimpleRow.Delete()
+}
+
+func (c *Customer) Refresh() error {
+	c.Time = time.Now().Add(time.Minute * 15)
+
+	return c.Update()
+}
+
+func (c *Customer) Confirm() error {
+	c.Confirmed = true
+
+	return c.Update()
+}

internal/customer/customer_model.go

@@ -0,0 +1,58 @@
+package customer
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	driver "github.com/arangodb/go-driver"
+)
+
+const (
+	DateTemplate     string = "2006-01-02"                      // Шаблон даты
+	TimeTemplate     string = "15:04:05"                        // Шаблон времени
+	StampTemplate    string = DateTemplate + " " + TimeTemplate // Шаблон метки времени
+	DateTemplatePast string = "1900-01-01" + " " + TimeTemplate // Далекое прошлое
+)
+
+type Row struct {
+	ctx        context.Context
+	collection driver.Collection
+	database   driver.Database
+	key        string
+}
+
+// Базовые поля
+type SimpleRow struct {
+	Row
+	ID        string `json:"id"`
+	CreatedAt string `json:"created_at"`
+	UpdatedAt string `json:"updated_at"`
+	DeletedAt string `json:"deleted_at"`
+}
+
+func NewSimpleRow(ctx context.Context, database driver.Database, collection driver.Collection) *SimpleRow {
+	now := time.Now()
+	nowStr := now.Format(StampTemplate)
+	key := strconv.Itoa(now.Nanosecond())
+
+	return &SimpleRow{
+		Row: Row{
+			ctx:        ctx,
+			collection: collection,
+			database:   database,
+			key:        key,
+		},
+		ID:        key,
+		CreatedAt: nowStr,
+		UpdatedAt: nowStr,
+		DeletedAt: DateTemplatePast,
+	}
+}
+
+// Удаление
+func (r *SimpleRow) Delete() error {
+	_, err := r.collection.RemoveDocument(r.ctx, r.key)
+
+	return err
+}

internal/customer/repository.go

@@ -0,0 +1,91 @@
+package customer
+
+import (
+	"context"
+	"log"
+	"time"
+
+	arango_db "git.slaventius.ru/test3k/authDB/internal/transport/arango"
+	"git.slaventius.ru/test3k/umate/pkg/kafka"
+
+	"github.com/arangodb/go-driver"
+)
+
+type CustomerRepository struct {
+	ctx        context.Context
+	database   driver.Database
+	collection driver.Collection
+	Customers  map[string]Customer
+}
+
+// Хранилище покупателей
+func NewCustomerRepository(ctx context.Context, database *arango_db.DataBase) *CustomerRepository {
+	collName := "customers"
+	collection, err := database.AddCollection(collName)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return &CustomerRepository{
+		ctx:        ctx,
+		database:   database.DB,
+		collection: collection,
+		Customers:  make(map[string]Customer),
+	}
+}
+
+// Новый покупатель
+func (r *CustomerRepository) NewCustomer(login string, password string, email string, code string) (Customer, error) {
+	customer := Customer{
+		SimpleRow: *NewSimpleRow(r.ctx, r.database, r.collection),
+		Login:     login,
+		Password:  password,
+		Time:      time.Now().Add(time.Minute * 15),
+		MessageRegistration: kafka.MessageRegistration{
+			Email: email,
+			Code:  code,
+		},
+	}
+
+	//
+	meta, err := r.collection.CreateDocument(r.ctx, customer)
+	if err != nil {
+		return customer, err
+	} else {
+		customer.key = meta.Key
+	}
+
+	return customer, nil
+}
+
+func (r *CustomerRepository) Fetch() error {
+	params := make(map[string]interface{})
+	params["@coll"] = r.collection.Name()
+
+	//
+	query := "for el in @@coll return el"
+	cursor, err := r.database.Query(r.ctx, query, params)
+	if err != nil {
+		return err
+	}
+	defer cursor.Close()
+
+	//
+	for {
+		rec := Customer{}
+		meta, era := cursor.ReadDocument(r.ctx, &rec)
+		if driver.IsNoMoreDocuments(era) {
+			break
+		}
+
+		//
+		rec.ctx = r.ctx
+		rec.collection = r.collection
+		rec.database = r.database
+		rec.key = meta.Key
+
+		r.Customers[rec.Login] = rec
+	}
+
+	return nil
+}