cmd/query/main.go - main - iezhelev/cppsh_micro

cmd/query/main.go
346 строк10 KB
iezhelev
ws server
23 мар 2025, 22:56
23 мар 2025, 22:56ac33f5f
package main

import (
	//"cppsh_micro/pkg/query"
	"context"
	"database/sql"
	"fmt"
	"log"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"regexp"
	"strings"
	"encoding/json"

	"github.com/go-redis/redis/v8"
	//"github.com/elastic/go-elasticsearch/v8"
	_ "github.com/lib/pq"
	"github.com/streadway/amqp"

	"cppsh_micro/internal/service"
)

var ctx = context.Background()

var (
	redisClient *redis.Client
	rabbitChan  *amqp.Channel
)

func main() {
	// Подключение к Elasticsearch
	// es, err := elasticsearch.NewDefaultClient()
	// if err != nil {
	// 	log.Fatalf("Error creating Elasticsearch client: %s", err)
	// }
	redisAddr := os.Getenv("REDIS_URL")
	fmt.Printf("Redis addr: %s\n", redisAddr)
    redisClient = redis.NewClient(&redis.Options{
        Addr:     os.Getenv("REDIS_URL"),
        Password: "", // no password
        DB:       0,  // default DB
    })


	// Подключение к RabbitMQ
	conn, err := amqp.Dial("amqp://guest:guest@rabbitmq:5672/")
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %s", err)
	}
	defer conn.Close()

	rabbitChan, err = conn.Channel()
	if err != nil {
		log.Fatalf("Failed to open a channel: %s", err)
	}
	defer rabbitChan.Close()

	// // Создаём очередь для событий
	// q, err := rabbitChan.QueueDeclare(
	// 	"user_events", // Имя очереди
	// 	false,         // durable
	// 	false,         // delete when unused
	// 	false,         // exclusive
	// 	false,         // no-wait
	// 	nil,           // arguments
	// )
	// if err != nil {
	// 	log.Fatalf("Failed to declare a queue: %s", err)
	// }

	// // Подписываемся на события
	// msgs, err := ch.Consume(
	// 	q.Name, // queue
	// 	"",     // consumer
	// 	true,   // auto-ack
	// 	false,  // exclusive
	// 	false,  // no-local
	// 	false,  // no-wait
	// 	nil,    // args
	// )
	// if err != nil {
	// 	log.Fatalf("Failed to register a consumer: %s", err)
	// }

	// // Обработчик событий
	// go func() {
	// 	for msg := range msgs {
	// 		log.Printf("Received a message: %s", msg.Body)
	// 		// Обрабатываем событие (например, обновляем данные в Elasticsearch)
	// 		//query.HandleEvent(es, msg.Body)
	// 	}
	// }()

    connStr := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
    os.Getenv("DB_HOST"), os.Getenv("DB_PORT"), os.Getenv("DB_USER"), os.Getenv("DB_PASSWORD"), os.Getenv("DB_NAME"))
	db, err := sql.Open("postgres", connStr)
	if err != nil {
		log.Fatalf("Error connecting to pg: %s", err)
		panic(err)
	}
	defer db.Close()
	// Initialize the DataRetrievalService
	dataRetrievalService := service.NewDataRetrievalService(db, redisAddr)
	// Define a handler for all paths
	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		switch {
		case r.URL.Path == "/health":
			// Handle the /health endpoint
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusOK)
			fmt.Fprint(w, `{"status":"ok"}`)
		case r.URL.Path == "/mobile/json/dayplan":
			//handleDayPlan(w, r, dataRetrievalService)
			handleScreenedQuery(w, r)
		case regexp.MustCompile(`^json/history`).MatchString(r.URL.Path):
			handleHistoryScreened(w, r)
		case regexp.MustCompile(`^/mobile/([^/]+)$`).MatchString(r.URL.Path):
			handleKafka(w, r, dataRetrievalService)
		case regexp.MustCompile(`^/mobile/attachment/([^/]+)$`).MatchString(r.URL.Path):
			handleMobileAttachment(w, r)
		case regexp.MustCompile(`^/mobile/json/m/v1/entity/user/([^/]+)$`).MatchString(r.URL.Path):
			handleMobileEntityUser(w, r)
		case regexp.MustCompile(`^/mobile/json/v2/xapi/entity/([^/]+)/sn/([^/]+)$`).MatchString(r.URL.Path):
			handleMobileEntityTableSn(w, r)
		}
	})

	// // Инициализация HTTP-обработчика
	// handler := query.NewQueryHandler()

	// // Запуск HTTP-сервера
	// http.HandleFunc("/user", handler.GetUser)
	log.Println("Query Service started on :8081")
	if err := http.ListenAndServe(":8081", nil); err != nil {
	 	log.Fatalf("Failed to start HTTP server: %s", err)
	}
}

func handleHistoryScreened(w http.ResponseWriter, r *http.Request) {
	deviceid := r.URL.Query().Get("deviceid")
	userid := r.URL.Query().Get("userid")
	timestamp := r.URL.Query().Get("dt")
	entityId := r.URL.Query().Get("id")
	numitems := r.URL.Query().Get("num")


	cacheKey := fmt.Sprintf("%s:%s:%s:%s", deviceid, userid, entityId, timestamp)
	log.Printf("history %s", cacheKey)
	// Check Redis for cached response
	//ctx := context.Background()
	
	cachedResponse, err := redisClient.Get(ctx, cacheKey).Result()
	if err == nil {
		log.Println("return chached result");
		// Return cached response and delete it from Redis
		resdata := fmt.Sprintf("\"deviceid\":\"%s\",\"userid\":\"%s\",\"entityId\":\"%s\",\"dt\":\"%s\",", deviceid, userid, entityId, timestamp)

		w.WriteHeader(http.StatusOK)
		w.Write([]byte(fmt.Sprintf("{%s\"hasMore\":false,\"history\":[{}", resdata)+cachedResponse+"]}"))
		redisClient.Del(ctx, cacheKey)
		return
	}


	query := map[string]string{
		"deviceid": deviceid,
		"userid":  userid,
		"timestamp": timestamp,
		"id": entityId,
		"num": numitems,
	}
	// If no cached response, publish query to RabbitMQ
	err = publishToRabbitMQ2(query, "historyQueue")
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		w.Write([]byte("Failed to publish query to RabbitMQ"))
		return
	}
	log.Println("published to rabbit");

	// Return empty response
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("{}"))
}

// Helper function to forward requests to the main system
func forwardRequest(w http.ResponseWriter, r *http.Request) {
	// Parse the main system URL
	mainSystemURL, err := url.Parse(os.Getenv("MAIN_SYSTEM_URL"))
	if err != nil {
		http.Error(w, "Failed to parse main system URL", http.StatusInternalServerError)
		return
	}

	// Strip the "/mobile" prefix from the request path
	r.URL.Path = strings.TrimPrefix(r.URL.Path, "/mobile")

	// Create a reverse proxy
	proxy := httputil.NewSingleHostReverseProxy(mainSystemURL)

	// Update the request host to match the main system's host
	r.Host = mainSystemURL.Host

	// Forward the request to the main system
	proxy.ServeHTTP(w, r)
}

func handleDayPlan(w http.ResponseWriter, r *http.Request, drs *service.DataRetrievalService) {
	userID := r.URL.Query().Get("userid")
	//deviceID := r.URL.Query().Get("deviceid")
	dtStr := r.URL.Query().Get("dt")
	log.Printf("Day plan request for userId %s, time %s", userID, dtStr)
	// Call the DataRetrievalService
	data, err := drs.HandleDataRequest(userID, dtStr)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	// Return the response
	w.Header().Set("Content-Type", "application/json")
	w.Write([]byte(data))

}

// Handlers for each path
func handleKafka(w http.ResponseWriter, r *http.Request, drs *service.DataRetrievalService) {
	re := regexp.MustCompile(`^/mobile/([^/]+)$`)
	matches := re.FindStringSubmatch(r.URL.Path)
	shortname := matches[1]
	fmt.Printf("Data request for shortname: %s\n", shortname)

	dtStr := r.URL.Query().Get("dt")

	// Call the DataRetrievalService
	data, err := drs.HandleDataRequest(shortname, dtStr)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	// Return the response
	w.Header().Set("Content-Type", "application/json")
	w.Write([]byte(data))
}

func handleMobileAttachment(w http.ResponseWriter, r *http.Request) {
	re := regexp.MustCompile(`^/mobile/attachment/([^/]+)$`)
	matches := re.FindStringSubmatch(r.URL.Path)
	id := matches[1]
	fmt.Printf("Mobile attachment request for ID: %s\n", id)
	forwardRequest(w, r)
}

func handleMobileEntityUser(w http.ResponseWriter, r *http.Request) {
	re := regexp.MustCompile(`^/mobile/json/m/v1/entity/user/([^/]+)$`)
	matches := re.FindStringSubmatch(r.URL.Path)
	shortname := matches[1]
	fmt.Printf("Mobile entity user request for shortname: %s\n", shortname)
	forwardRequest(w, r)
}

func handleMobileEntityTableSn(w http.ResponseWriter, r *http.Request) {
	re := regexp.MustCompile(`^/mobile/json/v2/xapi/entity/([^/]+)/sn/([^/]+)$`)
	matches := re.FindStringSubmatch(r.URL.Path)
	tablename := matches[1]
	shortname := matches[2]
	fmt.Printf("Mobile entity table request for tablename: %s, shortname: %s\n", tablename, shortname)
	forwardRequest(w, r)
}

func handleScreenedQuery(w http.ResponseWriter, r *http.Request) {
	deviceid := r.URL.Query().Get("deviceid")
	userid := r.URL.Query().Get("userid")
	timestamp := r.URL.Query().Get("dt")
	log.Printf("%s:%s:%s", deviceid, userid, timestamp)
	// Check Redis for cached response
	//ctx := context.Background()
	// cacheKey := fmt.Sprintf("%s:%s", deviceid, userid)
	// cachedResponse, err := redisClient.Get(ctx, cacheKey).Result()
	// if err == nil {
	// 	log.Println("return chached result");
	// 	// Return cached response and delete it from Redis

	// 	w.WriteHeader(http.StatusOK)
	// 	w.Write([]byte("{\"hasMore\":false,\"messages\":[{}"+cachedResponse+"]}"))
	// 	redisClient.Del(ctx, cacheKey)
	// 	return
	// }

	// If no cached response, publish query to RabbitMQ
	err := publishToRabbitMQ(deviceid, userid, timestamp)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		w.Write([]byte("Failed to publish query to RabbitMQ"))
		return
	}
	log.Println("published to rabbit");

	// Return empty response
	w.WriteHeader(http.StatusOK)
	w.Write([]byte("{}"))
}

func publishToRabbitMQ(deviceid, userid, timestamp string) error {
	query := map[string]string{
		"deviceid": deviceid,
		"userid":  userid,
		"timestamp": timestamp,
	}
	body, err := json.Marshal(query)
	if err != nil {
		return err
	}

	return rabbitChan.Publish(
		"",          // exchange
		"queryQueue", // routing key
		false,       // mandatory
		false,       // immediate
		amqp.Publishing{
			ContentType: "application/json",
			Body:        body,
		},
	)
}

func publishToRabbitMQ2(query map[string]string, queue string) error {
	body, err := json.Marshal(query)
	if err != nil {
		return err
	}

	return rabbitChan.Publish(
		"",          // exchange
		queue, // routing key
		false,       // mandatory
		false,       // immediate
		amqp.Publishing{
			ContentType: "application/json",
			Body:        body,
		},
	)
}