<commit_before>package DatabaseConnector

import (
	"database/sql"
	"errors"
	"fmt"
	"log"
	"sync"

	mdl "github.com/LoRaWanSoFa/LoRaWanSoFa/Components"
	_ "github.com/lib/pq"
)

type DatabaseConnector struct {
	Database          *sql.DB
	CheckDevEUISTMT   *sql.Stmt
	GetNodeHeaderSTMT *sql.Stmt
	InsertMessageSTMT *sql.Stmt
	InsertPayloadSTMT *sql.Stmt
}

type WorkRequest struct {
	Query         string
	ResultChannel chan (WorkResult)
	Arguments     []interface{}
	F             func(w *WorkRequest)
}

type WorkResult struct {
	Result interface{}
	err    error
}

func newDBC(db *sql.DB) *DatabaseConnector {
	dbc := DatabaseConnector{}
	dbc.Database = db
	return &dbc
}

var instantiated *DatabaseConnector
var once sync.Once
var WorkQueue = make(chan WorkRequest, 100)

// Get the instantiated instance of the DatabaseConnector or create it.
func GetInstance() *DatabaseConnector {
	once.Do(func() {
		settings := mdl.GetConfiguration().Db
		dbConnectionInfo := fmt.Sprintf("user=%s password=%s dbname=%s host=%s port=%v sslmode=disable", settings.User, settings.Password, settings.Name, settings.Network, settings.Port)
		println(dbConnectionInfo)
		actualDb, err := sql.Open("postgres", dbConnectionInfo)
		instantiated = newDBC(actualDb)
		if err != nil {
			log.Fatal(err)
		}
		log.Print("not connected yet")
		StartDispatcher(settings.NumberOfWorkers)
	})
	return instantiated
}

func Connect() error {
	db := GetInstance()
	err := db.Database.Ping()
	if err != nil {
		return err
	}
	db.CheckDevEUISTMT, err = db.Database.Prepare("select exists(select 1 from nodes where deveui=$1)")
	if err != nil {
		return err
	}
	db.GetNodeHeaderSTMT, err = db.Database.Prepare("select sensors.id, number_of_values, lenght_of_values, header_order, conversion_expression, data_type " +
		"from sensors " +
		"join public.sensortypes on sensors.sensortype_id = sensortypes.id " +
		"where deveui =$1 " +
		"order by header_order asc;")
	if err != nil {
		return err
	}
	db.InsertMessageSTMT, err = db.Database.Prepare("INSERT INTO public.messages(" +
		"deveui, created_at, down) " +
		"VALUES ($1, NOW(), false) " +
		"RETURNING id;")
	if err != nil {
		return err
	}
	db.InsertPayloadSTMT, err = db.Database.Prepare("INSERT INTO public.message_payloads(" +
		"message_id, sensor_id, payload) " +
		"VALUES ($1, $2, $3);")
	return err
}

func Close() error {
	return GetInstance().Database.Close()
}

//Not used, but can be used as example
// Checks if the devEUI exists in the database.
// Uses a database worker to execute the query.
func CheckDevEUI(devEUI string) bool {
	//log.Println(devEUI)
	result := make(chan WorkResult)
	args := make([]interface{}, 1)
	args[0] = devEUI
	WorkQueue <- WorkRequest{Query: "", Arguments: args, ResultChannel: result, F: func(w *WorkRequest) {
		rows, err := GetInstance().CheckDevEUISTMT.Query(w.Arguments...)
		defer rows.Close()
		checkErr(err)
		if err != nil {
			w.ResultChannel <- WorkResult{Result: false, err: err}
			return
		}
		rows.Next()
		var exisistsboolean = false
		err = rows.Scan(&exisistsboolean)
		w.ResultChannel <- WorkResult{Result: exisistsboolean, err: err}
	}}
	defer close(result)
	var workResult = <-result
	checkErr(workResult.err)
	exists := workResult.Result.(bool)
	log.Printf("Work done for Q:%s A:%t\n", devEUI, exists)
	return exists
}

//Store a message
func AddMessage(devEUI string) (mdl.MessageUplinkI, error) {
	//create response channel
	result := make(chan WorkResult)
	defer close(result)
	//create arguments
	args := make([]interface{}, 1)
	args[0] = devEUI
	//create and add new WorkRequest
	WorkQueue <- WorkRequest{Query: "", Arguments: args, ResultChannel: result, F: func(w *WorkRequest) {
		var messageId int64
		rows, err := GetInstance().InsertMessageSTMT.Query(w.Arguments...)

		checkErr(err)
		if err != nil {
			log.Println("check errors1")
			w.ResultChannel <- WorkResult{Result: 0, err: err}
			return
		}
		defer rows.Close()

		rows.Next()
		err = rows.Scan(&messageId)
		if err != nil {
			w.ResultChannel <- WorkResult{Result: 0, err: err}
		}
		w.ResultChannel <- WorkResult{Result: messageId, err: nil}
	}}
	response := <-result
	if response.err != nil {
		log.Println("check errors3")
		return nil, response.err
	}
	message := mdl.NewMessageUplink(response.Result.(int64), devEUI)

	return message, response.err
}

//Get a message
func StoreMessagePayloads(message mdl.MessageUplinkI) error {
	if message == nil {
		return errors.New("nil given as message parameter")
	}
	if message.GetId() == 0 {
		return errors.New("Message has not been initalized/stored")
	}
	payloads := message.GetPayloads()
	if len(payloads) == 0 {
		return errors.New("Nothing to store!")
	}
	var parameters []interface{}
	for _, payload := range payloads {
		parameters = append(parameters, message.GetId())        //message id
		parameters = append(parameters, payload.GetSensor().Id) //sensor id
		parameters = append(parameters, payload.GetPayload())   //payload
		log.Printf("parameters: %+v", parameters)
	}
	return nil
}

func getIndex(i int) string {
	i++
	return string(i)
}

//Get messages from one node
// TODO define return type
func GetNodeMessages(NodeId, maxMessages int) {
	// TODO
}

//Get sensors that belong to one node
func GetNodeSensors(devEUI string) []mdl.Sensor {
	result := make(chan WorkResult)
	defer close(result)
	args := make([]interface{}, 1)
	args[0] = devEUI
	WorkQueue <- WorkRequest{Query: "", Arguments: args, ResultChannel: result, F: func(w *WorkRequest) {
		//rows, err := GetInstance().Database.Query(w.Query, w.Arguments...)
		rows, err := GetInstance().GetNodeHeaderSTMT.Query(w.Arguments...)
		defer rows.Close()
		checkErr(err)
		if err != nil {
			w.ResultChannel <- WorkResult{Result: false, err: err}
			return
		}
		sensors := make([]mdl.Sensor, 0)<commit_msg>Insert payloads into the database<commit_after>package DatabaseConnector

import (
	"database/sql"
	"errors"
	"fmt"
	"log"
	"sync"
	"time"

	mdl "github.com/LoRaWanSoFa/LoRaWanSoFa/Components"
	_ "github.com/lib/pq"
)

type DatabaseConnector struct {
	Database          *sql.DB
	CheckDevEUISTMT   *sql.Stmt
	GetNodeHeaderSTMT *sql.Stmt
	InsertMessageSTMT *sql.Stmt
	InsertPayloadSTMT *sql.Stmt
}

type WorkRequest struct {
	Query         string
	ResultChannel chan (WorkResult)
	Arguments     []interface{}
	F             func(w *WorkRequest)
}

type WorkResult struct {
	Result interface{}
	err    error
}

func newDBC(db *sql.DB) *DatabaseConnector {
	dbc := DatabaseConnector{}
	dbc.Database = db
	return &dbc
}

var instantiated *DatabaseConnector
var once sync.Once
var WorkQueue = make(chan WorkRequest, 100)

// Get the instantiated instance of the DatabaseConnector or create it.
func GetInstance() *DatabaseConnector {
	once.Do(func() {
		settings := mdl.GetConfiguration().Db
		dbConnectionInfo := fmt.Sprintf("user=%s password=%s dbname=%s host=%s port=%v sslmode=disable", settings.User, settings.Password, settings.Name, settings.Network, settings.Port)
		println(dbConnectionInfo)
		actualDb, err := sql.Open("postgres", dbConnectionInfo)
		instantiated = newDBC(actualDb)
		if err != nil {
			log.Fatal(err)
		}
		log.Print("not connected yet")
		StartDispatcher(settings.NumberOfWorkers)
	})
	return instantiated
}

func Connect() error {
	//IDEA move sql.Open to here so we can reconnect.
	db := GetInstance()
	err := db.Database.Ping()
	if err != nil {
		return err
	}
	time.Sleep(1000 * time.Millisecond) //Wait for connections to finnish setting up. 1 sec is propably too long, can be finetuned later
	db.CheckDevEUISTMT, err = db.Database.Prepare("select exists(select 1 from nodes where deveui=$1)")
	if err != nil {
		return err
	}
	db.GetNodeHeaderSTMT, err = db.Database.Prepare("select sensors.id, number_of_values, lenght_of_values, header_order, conversion_expression, data_type " +
		"from sensors " +
		"join public.sensortypes on sensors.sensortype_id = sensortypes.id " +
		"where deveui =$1 " +
		"order by header_order asc;")
	if err != nil {
		return err
	}
	db.InsertMessageSTMT, err = db.Database.Prepare("INSERT INTO public.messages(" +
		"deveui, created_at, down) " +
		"VALUES ($1, NOW(), false) " +
		"RETURNING id;")
	if err != nil {
		return err
	}
	db.InsertPayloadSTMT, err = db.Database.Prepare("INSERT INTO public.message_payloads(" +
		"message_id, sensor_id, payload) " +
		"VALUES ($1, $2, $3);")
	return err
}

func Close() error {
	return GetInstance().Database.Close()
}

//Not used, but can be used as example
// Checks if the devEUI exists in the database.
// Uses a database worker to execute the query.
func CheckDevEUI(devEUI string) bool {
	//log.Println(devEUI)
	result := make(chan WorkResult)
	args := make([]interface{}, 1)
	args[0] = devEUI
	WorkQueue <- WorkRequest{Query: "", Arguments: args, ResultChannel: result, F: func(w *WorkRequest) {
		rows, err := GetInstance().CheckDevEUISTMT.Query(w.Arguments...)
		defer rows.Close()
		checkErr(err)
		if err != nil {
			w.ResultChannel <- WorkResult{Result: false, err: err}
			return
		}
		rows.Next()
		var exisistsboolean = false
		err = rows.Scan(&exisistsboolean)
		w.ResultChannel <- WorkResult{Result: exisistsboolean, err: err}
	}}
	defer close(result)
	var workResult = <-result
	checkErr(workResult.err)
	exists := workResult.Result.(bool)
	log.Printf("Work done for Q:%s A:%t\n", devEUI, exists)
	return exists
}

//Store a message
func AddMessage(devEUI string) (mdl.MessageUplinkI, error) {
	//create response channel
	result := make(chan WorkResult)
	defer close(result)
	//create arguments
	args := make([]interface{}, 1)
	args[0] = devEUI
	//create and add new WorkRequest
	WorkQueue <- WorkRequest{Query: "", Arguments: args, ResultChannel: result, F: func(w *WorkRequest) {
		var messageId int64
		rows, err := GetInstance().InsertMessageSTMT.Query(w.Arguments...)

		checkErr(err)
		if err != nil {
			log.Println("check errors1")
			w.ResultChannel <- WorkResult{Result: 0, err: err}
			return
		}
		defer rows.Close()

		rows.Next()
		err = rows.Scan(&messageId)
		if err != nil {
			w.ResultChannel <- WorkResult{Result: 0, err: err}
		}
		w.ResultChannel <- WorkResult{Result: messageId, err: nil}
	}}
	response := <-result
	if response.err != nil {
		log.Println("check errors3")
		return nil, response.err
	}
	message := mdl.NewMessageUplink(response.Result.(int64), devEUI)

	return message, response.err
}

//Get a message
func StoreMessagePayloads(message mdl.MessageUplinkI) error {
	if message == nil {
		return errors.New("nil given as message parameter")
	}
	if message.GetId() == 0 {
		return errors.New("Message has not been initalized/stored")
	}
	payloads := message.GetPayloads()
	if len(payloads) == 0 {
		return errors.New("Nothing to store!")
	}
	var parameters []interface{}
	var err error
	for _, payload := range payloads {
		parameters = make([]interface{}, 0)
		parameters = append(parameters, message.GetId())               //message id
		parameters = append(parameters, payload.GetSensor().Id)        //sensor id
		parameters = append(parameters, payload.GetPayload().(string)) //payload
		//log.Printf("parameters: %+v", parameters)
		err = insertPayload(parameters)
	}
	if err != nil {
		return nil
	}
	return nil
}

func insertPayload(parameters []interface{}) error {
	result := make(chan WorkResult)
	defer close(result)
	WorkQueue <- WorkRequest{Query: "", Arguments: parameters, ResultChannel: result, F: func(w *WorkRequest) {
		rows, err := GetInstance().InsertPayloadSTMT.Query(w.Arguments...)

		checkErr(err)
		if err != nil {
			log.Println("Query could not be executed!")
			w.ResultChannel <- WorkResult{Result: nil, err: err}
			return
		}
		defer rows.Close()

		rows.Next()
		if err != nil {
			w.ResultChannel <- WorkResult{Result: nil, err: err}
		}
		w.ResultChannel <- WorkResult{Result: true, err: nil}
	}}
	response := <-result
	if response.err != nil {
		log.Println("The worker could not execute the work properly")
		return response.err
	}
	return nil
}

func getIndex(i int) string {
	i++
	return string(i)
}

//Get messages from one node
// TODO define return type
func GetNodeMessages(NodeId, maxMessages int) {
	// TODO
}

//Get sensors that belong to one node
func GetNodeSensors(devEUI string) []mdl.Sensor {
	result := make(chan WorkResult)
	defer close(result)
	args := make([]interface{}, 1)
	args[0] = devEUI
	WorkQueue <- WorkRequest{Query: "", Arguments: args, ResultChannel: result, F: func(w *WorkRequest) {
		//rows, err := GetInstance().Database.Query(w.Query, w.Arguments...)
		rows, err := GetInstance().GetNodeHeaderSTMT.Query(w.Arguments...)
		defer rows.Close()
		checkErr(err)
		if err != nil {
			w.ResultChannel <- WorkResult{Result: false, err: err}
			return
		}
		sensors := make([]mdl.Sensor, 0)
{'fraction_non_alphanumeric': 0.13296071794064393, 'fraction_numerical': 0.004723293710147209, 'mean_word_length': 4.097913322632424, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '490412', 'n_tokens_mistral': 4490, 'n_tokens_neox': 4230, 'n_words': 1140}