Feature/1 (#21)

* 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

* event processor implemented

* removed domain event handling from user repository

* moved event processing logic out of AuthService

* minor changes

* replaced plan controller with event handler

* new values in env

* new docker compose service

* implemented fetching plan data via ws

* refactored rest api

* plan handler refactoring

* error message update

* minor fix

* minor improvements in authenticator.go

* using dtos in plan_handler.go

* moved some files to other location

* minor change

* reduced user restriction service cache ttl to 15 seconds

* minor change in handler.go

* introduced a dialer service, changed project structure a bit

* introduced a dialer service, changed project structure a bit

* traffic reporter optimizations

* changed buf size in Proxy

* removed validateArgon2idHash

* buf optimizations

* fixed copyTrafficAndReport

* minor change in dialer_pool.go

* minor performance oriented improvements

* AuthService.AuthorizeBasic optimized with singleflight

* test fix

* gitignore update

* dependencies update

Author: NLipatov

.env

@@ -30,4 +30,14 @@ PROXY_KAFKA_GROUP_ID=PROXY
 PROXY_KAFKA_AUTO_OFFSET_RESET=earliest
 
 #PlansKafka
-PLAN_KAFKA_TOPIC=PLAN
+PLAN_KAFKA_TOPIC=PLAN
+
+#Google OAuth Client
+GOOGLE_AUTH_CLIENT_ID=secret
+GOOGLE_AUTH_CLIENT_SECRET=secret
+GOOGLE_AUTH_HOST=http://localhost
+GOOGLE_AUTH_PORT=3030
+
+#REST API KAFKA
+USERS_REST_API_KAFKA_GROUP_ID=rest-api
+USERS_REST_API_KAFKA_TOPIC=PROXY

.gitignore

@@ -28,3 +28,4 @@ go.work.sum
 /.idea/goproxy.iml
 /.idea/dataSources.xml
 /.idea/sqldialects.xml
+/.idea/forwardedPorts.xml

docker-compose.yml

@@ -99,6 +99,39 @@ services:
     depends_on:
       lt-billing-postgres:
         condition: service_healthy
+  
+  google-auth-api:
+    build:
+      context: ./src
+    container_name: google-auth-api
+    ports:
+      - "3030:3030"
+    environment:
+      - MODE=google-auth
+      - SERVE_PORT=3030
+      - ALLOWED_ORIGINS=http://localhost:5173
+      - DB_DATABASE=${DB_DATABASE}
+      - DB_USER=${DB_USER}
+      - DB_PASS=${DB_PASS}
+      - DB_HOST=${DB_HOST}
+      - DB_PORT=${DB_PORT}
+      - ENVIRONMENT=development
+      - GOOGLE_AUTH_CLIENT_ID=${GOOGLE_AUTH_CLIENT_ID}
+      - GOOGLE_AUTH_CLIENT_SECRET=${GOOGLE_AUTH_CLIENT_SECRET}
+      - GOOGLE_AUTH_HOST=${GOOGLE_AUTH_HOST}
+      - GOOGLE_AUTH_PORT=${GOOGLE_AUTH_PORT}
+      - PROXY_KAFKA_AUTO_OFFSET_RESET=${PROXY_KAFKA_AUTO_OFFSET_RESET}
+      - PROXY_KAFKA_BOOTSTRAP_SERVERS=kafka:9092
+      - PROXY_KAFKA_GROUP_ID=GOOGLE-AUTH
+      - PROXY_KAFKA_TOPIC=PROXY
+      - TC_CACHE_HOST=${TC_CACHE_HOST}
+      - TC_CACHE_PORT=${TC_CACHE_PORT}
+      - TC_CACHE_USER=${TC_CACHE_USER}
+    depends_on:
+      postgres:
+        condition: service_healthy
+      plan-controller-cache:
+        condition: service_healthy
 
   proxy:
     build:
@@ -143,6 +176,11 @@ services:
       - DB_HOST=${DB_HOST}
       - DB_PORT=${DB_PORT}
       - HTTP_REST_API_PORT=80
+      - PROXY_KAFKA_BOOTSTRAP_SERVERS=kafka:9092
+      - PROXY_KAFKA_GROUP_ID=${USERS_REST_API_KAFKA_GROUP_ID}
+      - PROXY_KAFKA_AUTO_OFFSET_RESET=${PROXY_KAFKA_AUTO_OFFSET_RESET}
+      - PROXY_KAFKA_TOPIC=${USERS_REST_API_KAFKA_TOPIC} 
+      - ALLOWED_ORIGINS=*
     depends_on:
       - auth-ctx-migrator
 
@@ -229,8 +267,11 @@ services:
     build:
       context: ./src
     container_name: plan-controller
+    ports:
+      - "3031:3031"
     environment:
       - MODE=plan-controller
+      - USERS_API_HOST=http://rest-api:80
       - DB_USER=${PLANS_DB_USER}
       - DB_PASS=${PLANS_DB_PASS}
       - DB_HOST=${PLANS_DB_HOST}
@@ -249,6 +290,7 @@ services:
       - TC_CACHE_PORT=${TC_CACHE_PORT}
       - TC_CACHE_USER=${TC_CACHE_USER}
       - TC_CACHE_PASSWORD=${TC_CACHE_PASSWORD}
+      - ALLOWED_ORIGINS=http://localhost:5173
     depends_on:
       init-kafka-topics:
         condition: service_completed_successfully

src/application/aplication_errors/ip_pool_empty.go

@@ -0,0 +1,8 @@
+package aplication_errors
+
+type ErrIpPoolEmpty struct {
+}
+
+func (ip ErrIpPoolEmpty) Error() string {
+	return "IP pool is empty"
+}

src/application/dialer_pool.go

@@ -0,0 +1,17 @@
+package application
+
+import (
+	"net"
+	"time"
+)
+
+type DialerPool interface {
+	// GetDialer retrieves a dialer for the given network and userId. If userId is bound, the corresponding IP is used.
+	GetDialer(network string, userId int) (*net.Dialer, error)
+
+	// BindDialerToUser binds an IP to a user for the specified TTL without creating a dialer.
+	BindDialerToUser(userId int, ttl time.Duration) error
+
+	// SetPool sets ip addresses available to be used with dialer
+	SetPool(ips []net.IP)
+}

src/application/event_processor.go

@@ -0,0 +1,87 @@
+package application
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+)
+
+type EventHandler interface {
+	Handle(payload string) error
+}
+
+type EventProcessor struct {
+	messageBus MessageBusService
+	handlerMap map[string]EventHandler
+	topics     []string
+}
+
+func NewEventProcessor(messageBus MessageBusService) *EventProcessor {
+	return &EventProcessor{
+		messageBus: messageBus,
+		handlerMap: make(map[string]EventHandler),
+	}
+}
+
+func (e *EventProcessor) RegisterHandler(eventType string, handler EventHandler) *EventProcessor {
+	e.handlerMap[eventType] = handler
+	return e
+}
+
+func (e *EventProcessor) RegisterTopic(topic string) *EventProcessor {
+	e.topics = append(e.topics, topic)
+	return e
+}
+
+func (e *EventProcessor) Build() error {
+	if e.topics == nil || len(e.topics) == 0 {
+		return errors.New("topics is empty")
+	}
+
+	if len(e.handlerMap) == 0 {
+		return errors.New("no handlers registered")
+	}
+
+	return nil
+}
+
+func (e *EventProcessor) Start(ctx context.Context) error {
+	if err := e.messageBus.Subscribe(e.topics); err != nil {
+		log.Fatalf("Failed to subscribe to topics: %v", err)
+	}
+
+	go func() {
+		defer func(messageBus MessageBusService) {
+			_ = messageBus.Close()
+		}(e.messageBus)
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				eventProcessingErr := e.ProcessNextEvent()
+				if eventProcessingErr != nil {
+					log.Printf("failed to process event: %v", eventProcessingErr)
+				}
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (e *EventProcessor) ProcessNextEvent() error {
+	event, err := e.messageBus.Consume()
+	if err != nil {
+		return fmt.Errorf("failed to consume event: %v", err)
+	}
+
+	eventHandler, ok := e.handlerMap[event.EventType.Value()]
+	if !ok {
+		return fmt.Errorf("no handler found for event type: %s", event.EventType.Value())
+	}
+
+	return eventHandler.Handle(event.Payload)
+}

src/application/event_processor_test.go

@@ -0,0 +1,138 @@
+package application
+
+import (
+	"context"
+	"errors"
+	"goproxy/application/mocks"
+	"goproxy/domain/events"
+	"goproxy/domain/valueobjects"
+	"testing"
+	"time"
+)
+
+type MockEventHandler struct {
+	handledEvents []string
+	returnError   bool
+}
+
+func (m *MockEventHandler) Handle(payload string) error {
+	m.handledEvents = append(m.handledEvents, payload)
+	if m.returnError {
+		return errors.New("mock handler error")
+	}
+	return nil
+}
+
+func TestEventProcessor_Start(t *testing.T) {
+	// Arrange
+	mockBus := mocks.NewMockMessageBusService()
+	mockHandler := &MockEventHandler{}
+	processor := NewEventProcessor(mockBus).
+		RegisterHandler("TestEvent", mockHandler).
+		RegisterTopic("TestTopic")
+
+	if err := processor.Build(); err != nil {
+		t.Fatalf("Failed to build EventProcessor: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	eventType, eventTypeErr := valueobjects.ParseEventTypeFromString("TestEvent")
+	if eventTypeErr != nil {
+		t.Fatalf("Failed to parse event type from payload: %v", eventTypeErr)
+	}
+
+	_ = mockBus.Produce("TestTopic", events.OutboxEvent{
+		EventType: eventType,
+		Payload:   "test_payload",
+	})
+
+	// Act
+	go func() {
+		if err := processor.Start(ctx); err != nil {
+			t.Errorf("Failed to start EventProcessor: %v", err)
+		}
+	}()
+
+	// Wait for the event to be processed
+	time.Sleep(100 * time.Millisecond)
+
+	// Assert
+	if len(mockHandler.handledEvents) != 1 {
+		t.Fatalf("Expected 1 handled event, got %d", len(mockHandler.handledEvents))
+	}
+	if mockHandler.handledEvents[0] != "test_payload" {
+		t.Errorf("Unexpected payload handled: %s", mockHandler.handledEvents[0])
+	}
+}
+
+func TestEventProcessor_MissingHandler(t *testing.T) {
+	// Arrange
+	mockBus := mocks.NewMockMessageBusService()
+	processor := NewEventProcessor(mockBus).
+		RegisterHandler("SomeEvent", &MockEventHandler{}).
+		RegisterTopic("TestTopic")
+
+	if err := processor.Build(); err != nil {
+		t.Fatalf("Failed to build EventProcessor: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	eventType, eventTypeErr := valueobjects.ParseEventTypeFromString("UnregisteredEvent")
+	if eventTypeErr != nil {
+		t.Fatalf("Failed to parse event type from payload: %v", eventTypeErr)
+	}
+
+	_ = mockBus.Produce("TestTopic", events.OutboxEvent{
+		EventType: eventType,
+		Payload:   "test_payload",
+	})
+
+	// Act
+	go func() {
+		if err := processor.Start(ctx); err != nil {
+			t.Errorf("Failed to start EventProcessor: %v", err)
+		}
+	}()
+
+	// Wait for the event to be processed
+	time.Sleep(100 * time.Millisecond)
+
+	// Assert
+	// No handlers registered for "UnregisteredEvent", so nothing should happen
+}
+
+func TestEventProcessor_Shutdown(t *testing.T) {
+	// Arrange
+	mockBus := mocks.NewMockMessageBusService()
+	mockHandler := &MockEventHandler{}
+	processor := NewEventProcessor(mockBus).
+		RegisterHandler("TestEvent", mockHandler).
+		RegisterTopic("TestTopic")
+
+	if err := processor.Build(); err != nil {
+		t.Fatalf("Failed to build EventProcessor: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	// Act
+	go func() {
+		if err := processor.Start(ctx); err != nil {
+			t.Errorf("Failed to start EventProcessor: %v", err)
+		}
+	}()
+
+	// Simulate shutdown
+	time.Sleep(100 * time.Millisecond)
+	cancel()
+
+	// Assert
+	time.Sleep(100 * time.Millisecond) // Give time for graceful shutdown
+	if !mockBus.WasCloseCalled() {
+		t.Errorf("MessageBus was not closed during shutdown")
+	}
+}

src/application/httpProxyListener.go

@@ -0,0 +1,9 @@
+package application
+
+import (
+	"net"
+)
+
+type HttpProxyListenerService interface {
+	Listen(port int) (net.Listener, error)
+}

src/application/ip_resolver.go

@@ -0,0 +1,9 @@
+package application
+
+import (
+	"net"
+)
+
+type IPResolver interface {
+	GetHostPublicIPs() ([]net.IP, error)
+}