web/server.go - main - AngelCareMe/task_queue

web/server.go
428 строк11 KB
AngelCareMe
Dashboard done/project ready
29 июл 2025, 19:11
29 июл 2025, 19:111894643
package main

import (
	"context"
	"fmt"
	"net/http"
	"strings"
	"time"

	"task-queue/internal/adapter/queue/redis"
	redisConfig "task-queue/internal/adapter/queue/redis/config"
	redisConn "task-queue/internal/adapter/queue/redis/connection"
	"task-queue/internal/adapter/repository"
	"task-queue/internal/adapter/repository/postgres"
	postgresConfig "task-queue/internal/adapter/repository/postgres/config"
	postgresConn "task-queue/internal/adapter/repository/postgres/connection"
	"task-queue/internal/entity" // Убедись, что этот импорт есть
	"task-queue/pkg/config"

	"github.com/gin-gonic/gin"
	"github.com/google/uuid"
	"github.com/spf13/viper"
)

// Task represents a task for the dashboard
type Task struct {
	ID        string    `json:"id"`
	Name      string    `json:"name"`
	Status    string    `json:"status"`
	Retry     int       `json:"retry"`
	MaxRetry  int       `json:"max_retry"`
	CreatedAt time.Time `json:"created_at"`
	UpdatedAt time.Time `json:"updated_at"`
}

// Stats represents statistics for the dashboard
type Stats struct {
	Total      int64            `json:"total"`
	Pending    int64            `json:"pending"`
	Processing int64            `json:"processing"`
	Success    int64            `json:"success"`
	Failed     int64            `json:"failed"`
	ByDate     map[string]int64 `json:"by_date"`
}

// CreateTaskRequest represents the request body for creating a task
type CreateTaskRequest struct {
	Name     string `json:"name" binding:"required"`
	Payload  string `json:"payload"`
	MaxRetry int    `json:"max_retry"`
	Delay    int    `json:"delay"`
}

// Global variables for connections
var (
	globalDBPool      *postgresConn.Connection
	globalRedisClient *redisConn.Connection
)

func loadHostConfig() (*config.Config, error) {
	viper.Reset()

	viper.SetConfigName("config.host")
	viper.SetConfigType("yaml")
	viper.AddConfigPath(".")

	viper.AutomaticEnv()
	viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))

	viper.SetDefault("server.port", "8080")
	viper.SetDefault("queue.type", "redis")
	viper.SetDefault("log.level", "info")
	viper.SetDefault("database.max_connections", 25)

	if err := viper.ReadInConfig(); err != nil {
		if _, ok := err.(viper.ConfigFileNotFoundError); !ok {
			return nil, err
		}
	}

	var cfg config.Config
	err := viper.Unmarshal(&cfg)
	if err != nil {
		return nil, err
	}

	return &cfg, nil
}

func initConnections() error {
	cfg, err := loadHostConfig()
	if err != nil {
		return fmt.Errorf("failed to load host config: %w", err)
	}

	// Initialize database connection pool
	dbConfig := &postgresConfig.Config{
		Host:            cfg.Database.Host,
		Port:            cfg.Database.Port,
		User:            cfg.Database.User,
		Password:        cfg.Database.Password,
		Database:        cfg.Database.Name,
		SSLMode:         cfg.Database.SSLMode,
		MaxConnections:  cfg.Database.MaxConnections,
		MinConnections:  postgresConfig.DefaultConfig().MinConnections,
		MaxConnLifetime: postgresConfig.DefaultConfig().MaxConnLifetime,
		MaxConnIdleTime: postgresConfig.DefaultConfig().MaxConnIdleTime,
		ConnectTimeout:  postgresConfig.DefaultConfig().ConnectTimeout,
	}

	globalDBPool, err = postgresConn.NewConnection(
		cfg.Database.Host,
		cfg.Database.Port,
		cfg.Database.User,
		cfg.Database.Password,
		cfg.Database.Name,
		cfg.Database.SSLMode,
		cfg.Database.MaxConnections,
		dbConfig,
	)
	if err != nil {
		return fmt.Errorf("failed to connect to database: %w", err)
	}

	// Initialize Redis connection
	redisCfg := &redisConfig.Config{
		Host:     cfg.Redis.Host,
		Port:     cfg.Redis.Port,
		Password: cfg.Redis.Password,
		DB:       cfg.Redis.DB,
	}

	globalRedisClient, err = redisConn.NewConnection(redisCfg)
	if err != nil {
		globalDBPool.Close()
		return fmt.Errorf("failed to connect to Redis: %w", err)
	}

	return nil
}

func closeConnections() {
	if globalDBPool != nil {
		globalDBPool.Close()
	}
	if globalRedisClient != nil {
		globalRedisClient.Close()
	}
}

func main() {
	gin.SetMode(gin.ReleaseMode)

	if err := initConnections(); err != nil {
		fmt.Printf("Failed to initialize connections: %v\n", err)
		return
	}
	defer closeConnections()

	r := gin.Default()

	r.Static("/static", "./static")

	// API endpoints
	r.GET("/api/stats", getStats)
	r.GET("/api/tasks", getTasks)
	r.GET("/api/health", getHealth)
	r.POST("/api/tasks/:id/retry", retryTask)
	r.DELETE("/api/tasks/:id", deleteTask)
	r.POST("/api/tasks", createTask) // <-- Новый эндпоинт

	// Serve index.html for all other routes (SPA)
	r.GET("/", func(c *gin.Context) {
		c.File("./static/index.html")
	})

	r.GET("/dashboard", func(c *gin.Context) {
		c.File("./static/index.html")
	})

	fmt.Println("Dashboard server starting on :8085")
	if err := r.Run(":8085"); err != nil {
		fmt.Printf("Server failed to start: %v\n", err)
	}
}

func getDBPool() *postgresConn.Connection {
	return globalDBPool
}

func getRedisClient() *redisConn.Connection {
	return globalRedisClient
}

func getStats(c *gin.Context) {
	dbConn := getDBPool()
	if dbConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available"})
		return
	}

	taskRepo := postgres.NewPostgresRepository(dbConn.GetPool())

	stats, err := taskRepo.GetStats(context.Background())
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to get stats: " + err.Error()})
		return
	}

	dashboardStats := Stats{
		Total:      stats.Total,
		Pending:    stats.Pending,
		Processing: stats.Processing,
		Success:    stats.Success,
		Failed:     stats.Failed,
		ByDate:     stats.ByDate,
	}

	c.JSON(http.StatusOK, dashboardStats)
}

// getTasks получает список задач
func getTasks(c *gin.Context) {
	dbConn := getDBPool()
	if dbConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available"})
		return
	}

	taskRepo := postgres.NewPostgresRepository(dbConn.GetPool())

	// Используем правильный тип repository.TaskFilter
	filter := repository.TaskFilter{}

	limit := 50
	// В реальном приложении добавьте парсинг limit из query параметров
	// if limitParam := c.Query("limit"); limitParam != "" { ... }

	tasks, err := taskRepo.List(context.Background(), filter, limit, 0)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to get tasks: " + err.Error()})
		return
	}

	// Convert to dashboard tasks
	dashboardTasks := make([]Task, len(tasks))
	for i, task := range tasks {
		dashboardTasks[i] = Task{
			ID:        task.ID,
			Name:      task.Name,
			Status:    string(task.Status),
			Retry:     task.Retry,
			MaxRetry:  task.MaxRetry,
			CreatedAt: task.CreatedAt,
			UpdatedAt: task.UpdatedAt,
		}
	}

	c.JSON(http.StatusOK, dashboardTasks)
}

func getHealth(c *gin.Context) {
	dbConn := getDBPool()
	if dbConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"status": "unhealthy", "error": "Database connection not available"})
		return
	}

	redisConn := getRedisClient()
	if redisConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"status": "unhealthy", "error": "Redis connection not available"})
		return
	}

	if err := dbConn.HealthCheck(context.Background()); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"status": "unhealthy", "error": "Database health check failed: " + err.Error()})
		return
	}

	if err := redisConn.HealthCheck(context.Background()); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"status": "unhealthy", "error": "Redis health check failed: " + err.Error()})
		return
	}

	c.JSON(http.StatusOK, gin.H{"status": "healthy"})
}

func retryTask(c *gin.Context) {
	taskID := c.Param("id")
	if taskID == "" {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Task ID is required"})
		return
	}

	dbConn := getDBPool()
	if dbConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available"})
		return
	}

	redisConn := getRedisClient()
	if redisConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Redis connection not available"})
		return
	}

	taskRepo := postgres.NewPostgresRepository(dbConn.GetPool())
	taskQueue := redis.NewRedisQueue(redisConn.GetClient(), taskRepo)

	task, err := taskRepo.GetByID(context.Background(), taskID)
	if err != nil {
		c.JSON(http.StatusNotFound, gin.H{"error": "Task not found: " + err.Error()})
		return
	}

	if task.Retry >= task.MaxRetry {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Max retry attempts exceeded"})
		return
	}

	task.Status = entity.TaskStatusPending
	task.UpdatedAt = time.Now()

	if err := taskRepo.Update(context.Background(), task); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to update task: " + err.Error()})
		return
	}

	if err := taskQueue.Retry(context.Background(), taskID, 0); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to retry task: " + err.Error()})
		return
	}

	c.JSON(http.StatusOK, gin.H{"message": "Task scheduled for retry"})
}

func deleteTask(c *gin.Context) {
	taskID := c.Param("id")
	if taskID == "" {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Task ID is required"})
		return
	}

	dbConn := getDBPool()
	if dbConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available"})
		return
	}

	taskRepo := postgres.NewPostgresRepository(dbConn.GetPool())

	if err := taskRepo.Delete(context.Background(), taskID); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to delete task: " + err.Error()})
		return
	}

	c.JSON(http.StatusOK, gin.H{"message": "Task deleted successfully"})
}

// createTask handles the creation of a new task via the Dashboard
func createTask(c *gin.Context) {
	var req CreateTaskRequest

	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid JSON: " + err.Error()})
		return
	}

	dbConn := getDBPool()
	if dbConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Database connection not available"})
		return
	}

	redisConn := getRedisClient()
	if redisConn == nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Redis connection not available"})
		return
	}

	taskRepo := postgres.NewPostgresRepository(dbConn.GetPool())
	taskQueue := redis.NewRedisQueue(redisConn.GetClient(), taskRepo)

	// Create the entity task
	entityTask := &entity.Task{
		ID:        uuid.New().String(),
		Name:      req.Name,
		Payload:   req.Payload,
		Status:    entity.TaskStatusPending,
		Retry:     0,
		MaxRetry:  req.MaxRetry,
		Delay:     req.Delay,
		CreatedAt: time.Now(),
		UpdatedAt: time.Now(),
	}

	// Save task to database
	if err := taskRepo.Create(context.Background(), entityTask); err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create task in database: " + err.Error()})
		return
	}

	// Enqueue the task
	var queueErr error
	if entityTask.Delay > 0 {
		queueErr = taskQueue.Retry(context.Background(), entityTask.ID, time.Duration(entityTask.Delay)*time.Second)
	} else {
		queueErr = taskQueue.Enqueue(context.Background(), entityTask)
	}

	if queueErr != nil {
		fmt.Printf("Warning: Task created but failed to enqueue: %v\n", queueErr)
		// We don't return an error here as the task is successfully saved in DB
	}

	// Return the created task in dashboard format
	responseTask := Task{
		ID:        entityTask.ID,
		Name:      entityTask.Name,
		Status:    string(entityTask.Status),
		Retry:     entityTask.Retry,
		MaxRetry:  entityTask.MaxRetry,
		CreatedAt: entityTask.CreatedAt,
		UpdatedAt: entityTask.UpdatedAt,
	}

	c.JSON(http.StatusCreated, responseTask)
}