thola/internal/database/sqlDB.go

package database

import (
	"context"
	"encoding/json"
	"github.com/inexio/thola/internal/device"
	"github.com/inexio/thola/internal/network"
	"github.com/inexio/thola/internal/parser"
	"github.com/inexio/thola/internal/tholaerr"
	"github.com/jmoiron/sqlx"
	"github.com/pkg/errors"
	"github.com/rs/zerolog/log"
	"time"
)

type sqlDatabase struct {
	db *sqlx.DB
}

type sqlSelectResults []struct {
	Time     string
	Data     string
	Datatype string
}

var mysqlSchemaArr = []string{
	`DROP TABLE IF EXISTS cache;`,

	`CREATE TABLE cache (
		id INTEGER PRIMARY KEY,
		ip varchar(255) NOT NULL,
		datatype varchar(255) NOT NULL,
		data text NOT NULL,
		time datetime DEFAULT current_timestamp NOT NULL,
		CONSTRAINT unique_entries UNIQUE (ip, datatype)
		);`,
	`ALTER TABLE cache MODIFY id int(11) NOT NULL AUTO_INCREMENT;`,
}

func (d sqlDatabase) setupDatabase() error {
	for _, query := range mysqlSchemaArr {
		_, err := d.db.Exec(query)
		if err != nil {
			_, _ = d.db.Exec(`DROP TABLE IF EXISTS cache;`)
			return errors.Wrap(err, "Could not set up database schema - query: "+query)
		}
	}
	return nil
}

func (d *sqlDatabase) SetDeviceProperties(ctx context.Context, ip string, data device.Device) error {
	err := d.insertReplaceQuery(ctx, data, ip, "DeviceInfo")
	if err != nil {
		return errors.Wrap(err, "failed to store device data")
	}
	return nil
}

func (d *sqlDatabase) GetDeviceProperties(ctx context.Context, ip string) (device.Device, error) {
	var identifyResponse device.Device
	err := d.getEntry(ctx, &identifyResponse, ip, "DeviceInfo")
	if err != nil {
		return device.Device{}, err
	}
	return identifyResponse, nil
}

func (d *sqlDatabase) SetConnectionData(ctx context.Context, ip string, data network.ConnectionData) error {
	return d.insertReplaceQuery(ctx, data, ip, "ConnectionData")
}

func (d *sqlDatabase) GetConnectionData(ctx context.Context, ip string) (network.ConnectionData, error) {
	var connectionData network.ConnectionData
	err := d.getEntry(ctx, &connectionData, ip, "ConnectionData")
	if err != nil {
		return network.ConnectionData{}, err
	}
	return connectionData, nil
}

func (d *sqlDatabase) CheckConnection(ctx context.Context) error {
	return d.db.PingContext(ctx)
}

func (d *sqlDatabase) CloseConnection(ctx context.Context) error {
	log.Ctx(ctx).Debug().Msg("closing connection to mysql database")
	return d.db.Close()
}

func (d *sqlDatabase) insertReplaceQuery(ctx context.Context, data interface{}, ip, dataType string) error {
	JSONData, err := parser.ToJSON(data)
	if err != nil {
		return errors.Wrap(err, "failed to marshall data")
	}

	_, err = d.db.ExecContext(ctx, d.db.Rebind("REPLACE INTO cache (ip, datatype, data) VALUES (?, ?, ?);"), ip, dataType, string(JSONData))
	if err != nil {
		return errors.Wrap(err, "failed to exec sql query")
	}
	return nil
}

func (d *sqlDatabase) getEntry(ctx context.Context, dest interface{}, ip, dataType string) error {
	var results sqlSelectResults
	err := d.db.SelectContext(ctx, &results, d.db.Rebind("SELECT DATE_FORMAT(time, '%Y-%m-%d %H:%i:%S') as time, data, datatype FROM cache WHERE ip=? AND datatype=?;"), ip, dataType)
	if err != nil {
		return errors.Wrap(err, "db select failed")
	}
	if len(results) == 0 {
		return tholaerr.NewNotFoundError("cache entry not found")
	}

	res := results[0]
	t, err := time.Parse("2006-01-02 15:04:05", res.Time)
	if err != nil {
		return errors.Wrap(err, "failed to parse timestamp")
	}
	if time.Since(t) > cacheExpiration {
		_, err = d.db.ExecContext(ctx, d.db.Rebind("DELETE FROM cache WHERE ip=? AND datatype=?;"), ip, "IdentifyResponse")
		if err != nil {
			return errors.Wrap(err, "failed to delete expired cache element")
		}
		return tholaerr.NewNotFoundError("found only expired cache entry")
	}

	err = json.Unmarshal([]byte(res.Data), dest)
	if err != nil {
		return errors.Wrap(err, "failed to unmarshall entry data")
	}
	return nil
}