go_rabbitmq/app/lib/data_libs/version.go

package data_libs

import (
	"encoding/json"
	"fmt"
	"node_rabbit_go/app/lib"
	"time"
)

type Version struct{}

func New() *Version {
	return &Version{}
}

func (v *Version) GetPacket(requestController *lib.RequestController, data map[string]interface{}, devID string, timestamp int64) {
	selectResultChan := make(chan []struct {
		Version string `db:"version"`
	}, 1)
	errChan := make(chan error, 1)

	go v.selectVersion(requestController, devID, selectResultChan, errChan)

	select {
	case selectResult := <-selectResultChan:
		var oldVer map[string]interface{}
		if selectResult[0].Version != "" {
			err := json.Unmarshal([]byte(selectResult[0].Version), &oldVer)
			if err != nil {
				//log.Logger.Error(fmt.Sprintf("JSON unmarshal error: %v", err))
				return
			}
		}

		detectChangesChan := make(chan bool)
		go v.CheckChanges(oldVer, data, devID, requestController, detectChangesChan)

		detectChanges := <-detectChangesChan
		if detectChanges {
			oldVerJSON, _ := json.Marshal(oldVer)
			newVerJSON, _ := json.Marshal(data)

			go v.insertVersionHistory(requestController, devID, timestamp, string(oldVerJSON), string(newVerJSON), errChan)
			go v.updateDeviceVersion(requestController, devID, string(newVerJSON), errChan)

			requestController.STOMP.SendMessage("device_update_main", devID, map[string]interface{}{
				"action":   "UPDATE",
				"subtopic": "device_update_main",
			})
		}
	case err := <-errChan:
		fmt.Printf("%s", err)
		//logger_and_exception_handler.Error(fmt.Sprintf("SQL select error: %v", err))
		return
	}
}

func (v *Version) selectVersion(requestController *lib.RequestController, devID string, resultChan chan []struct {
	Version string `db:"version"`
}, errChan chan error) {
	query := fmt.Sprintf("SELECT version FROM device WHERE device_id=%d", devID)
	rows, err := requestController.MySQL.Query(query)
	if err != nil {
		errChan <- err
		return
	}
	//defer rows.Close()

	var selectResult []struct {
		Version string `db:"version"`
	}

	for rows.Next() {
		var version string
		if err := rows.Scan(&version); err != nil {
			errChan <- err
			return
		}
		selectResult = append(selectResult, struct {
			Version string `db:"version"`
		}{Version: version})
	}

	if err := rows.Err(); err != nil {
		errChan <- err
		return
	}

	resultChan <- selectResult
}

func (v *Version) insertVersionHistory(requestController *lib.RequestController, devID string, timestamp int64, oldVer, newVer string, errChan chan error) {
	_, err := requestController.MySQL.Query(fmt.Sprintf("INSERT INTO device_version_history(device_id, ts, old_ver, new_ver) VALUES (%s, %s, %s, %s)",
		devID, timestamp, oldVer, newVer))
	if err != nil {
		errChan <- err
	}
}

func (v *Version) updateDeviceVersion(requestController *lib.RequestController, devID, newVer string, errChan chan error) {
	_, err := requestController.MySQL.Query(fmt.Sprintf("UPDATE device SET version = %s WHERE device_id = %s", newVer, devID))
	if err != nil {
		errChan <- err
	}
}

func (v *Version) CheckChanges(oldVer, newVer map[string]interface{}, deviceID string, requestController *lib.RequestController, resultChan chan bool) {
	keysOld := getKeys(oldVer)
	keysNew := getKeys(newVer)

	findLoraNew := indexOf(keysNew, "lora_id")
	findOldLora := indexOf(keysOld, "lora_id")

	if findLoraNew != -1 && findOldLora == -1 {
		go v.insertLora(requestController, newVer[keysNew[findLoraNew]].(string), deviceID)
	} else if findOldLora != -1 && findLoraNew == -1 {
		go v.updateLoraStatus(requestController, deviceID, oldVer[keysOld[findOldLora]].(string))
	}

	if len(keysNew) != len(keysOld) {
		resultChan <- true
		return
	}

	diffRes := diff(keysNew, keysOld)
	if len(diffRes) > 0 {
		resultChan <- true
		return
	}

	for _, key := range keysNew {
		if oldVer[key] != newVer[key] {
			resultChan <- true
			return
		}
	}

	resultChan <- false
}

func (v *Version) insertLora(requestController *lib.RequestController, loraID, deviceID string) {
	_, err := requestController.MySQL.Query(fmt.Sprintf("REPLACE INTO device_lora (lora_id, device_id, status, created_at, updated_at) VALUES (%s, %s, %s, %s, %s)",
		loraID, deviceID, 1, time.Now().Unix(), time.Now().Unix()))
	if err != nil {
		//logger_and_exception_handler.Error(fmt.Sprintf("SQL replace error: %v", err))
	}
}

func (v *Version) updateLoraStatus(requestController *lib.RequestController, deviceID, loraID string) {
	_, err := requestController.MySQL.Query(fmt.Sprintf("UPDATE device_lora SET status=%s, updated_at=%s WHERE device_id=%s AND lora_id=%s",
		0, time.Now().Unix(), deviceID, loraID))
	if err != nil {
		//logger_and_exception_handler.Error(fmt.Sprintf("SQL update error: %v", err))
	}
}

func getKeys(m map[string]interface{}) []string {
	keys := make([]string, 0, len(m))
	for key := range m {
		keys = append(keys, key)
	}
	return keys
}

func indexOf(slice []string, item string) int {
	for i, v := range slice {
		if v == item {
			return i
		}
	}
	return -1
}

func diff(slice1, slice2 []string) []string {
	var diff []string
	m := make(map[string]bool)

	for _, item := range slice2 {
		m[item] = true
	}

	for _, item := range slice1 {
		if _, found := m[item]; !found {
			diff = append(diff, item)
		}
	}
	return diff
}