path: root/src/hub/src/spreadspace.org/sfive/s5store.go

package sfive

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"time"

	// needed for gorp tracing
	// "log"
	// "os"

	"github.com/boltdb/bolt"
	_ "github.com/mattn/go-sqlite3"
	"github.com/pborman/uuid"
	"gopkg.in/gorp.v2"
)

type sqliteStore struct {
	db     *gorp.DbMap
	dbBolt *bolt.DB
	hubId  string
}

func sourceFromStatisticsData(value StatisticsData) sourceDb {
	return sourceDb{
		Hostname: value.SourceId.Hostname,
		StreamId: streamIdDb{
			ContentId: value.SourceId.StreamId.ContentId,
			Format:    value.SourceId.StreamId.Format,
			Quality:   value.SourceId.StreamId.Quality,
		},
		Tags: value.SourceId.Tags,
	}
}

func dataUpdateFromStatisticsData(value StatisticsData) dataUpdateDb {
	return dataUpdateDb{
		-1,
		-1,
		value.SourceHubUuid,
		value.SourceHubDataUpdateId,
		value.StartTime.Unix(),
		value.Duration,
		value.Data.ClientCount,
		value.Data.BytesReceived,
		value.Data.BytesSent}
}

func updateFromStatisticsData(value StatisticsData) (dataUpdateDb, []ClientData, sourceDb) {
	du := dataUpdateFromStatisticsData(value)
	cd := value.Data.Clients
	src := sourceFromStatisticsData(value)

	return du, cd, src
}

func initDbBolt(boltPath string) (boltDb *bolt.DB, err error) {
	boltDb, err = bolt.Open(boltPath, 0600, &bolt.Options{Timeout: 1 * time.Second})
	if err != nil {
		return
	}

	err = boltDb.Update(func(tx *bolt.Tx) error {
		if _, err := tx.CreateBucketIfNotExists([]byte(sourcesFwdBn)); err != nil {
			return err
		}
		if _, err := tx.CreateBucketIfNotExists([]byte(sourcesRevBn)); err != nil {
			return err
		}
		if _, err := tx.CreateBucketIfNotExists([]byte(clientDataBn)); err != nil {
			return err
		}
		if _, err := tx.CreateBucketIfNotExists([]byte(userAgentsFwdBn)); err != nil {
			return err
		}
		_, err := tx.CreateBucketIfNotExists([]byte(userAgentsRevBn))
		return err
	})

	return
}

func initDbSqlite(sqlitePath string) (dbmap *gorp.DbMap, hubId string, err error) {

	var db *sql.DB
	db, err = sql.Open("sqlite3", sqlitePath)
	if err != nil {
		return
	}

	dbmap = &gorp.DbMap{Db: db, Dialect: gorp.SqliteDialect{}}
	// dbmap.TraceOn("[gorp]", log.New(os.Stdout, "myapp:", log.Lmicroseconds))

	dbmap.AddTableWithName(dataUpdateDb{}, dataUpdatesTn).SetKeys(true, "Id")
	dbmap.AddTableWithName(hubInfoDb{}, hubInfoTn).SetKeys(false, "Name")

	// TODO use some real migration, yadda yadda
	if err = dbmap.CreateTablesIfNotExists(); err != nil {
		return
	}

	hubId, err = dbmap.SelectStr("select Value from " + hubInfoTn + " where Name = 'HubUuid'")
	// TODO handle only not-found this way
	if err != nil || hubId == "" {
		hubId = uuid.New()
		_, err = db.Exec("insert into "+hubInfoTn+" values ('HubUuid', ?)", hubId)
		if err != nil {
			hubId = ""
			return
		}
	}

	return
}

func insertAnd(needsAnd *bool) (res string) {
	if *needsAnd {
		res = " and"
		*needsAnd = false
	}
	return
}

func (s sqliteStore) insertDataUpdateEntry(srcId int, du *dataUpdateDb) (err error) {
	du.SourceId = srcId
	err = s.db.Insert(du)
	if err != nil {
		return
	}
	return
}

func (s sqliteStore) insertNewSource(tx *bolt.Tx, src sourceDb) (srcId int, err error) {
	bf := tx.Bucket([]byte(sourcesFwdBn))
	bf.FillPercent = 1.0 // we only do appends
	br := tx.Bucket([]byte(sourcesRevBn))
	br.FillPercent = 1.0 // we only do appends

	slug := src.String()
	bSrcId := bf.Get([]byte(slug))
	if bSrcId != nil {
		return btoi(bSrcId), nil
	}

	var jsonData []byte
	if jsonData, err = json.Marshal(src); err != nil {
		return
	}

	next, _ := bf.NextSequence()
	srcId = int(next)
	if err = bf.Put([]byte(slug), itob(srcId)); err != nil {
		return
	}
	if err = br.Put(itob(srcId), jsonData); err != nil {
		return
	}

	return srcId, err
}

func (s sqliteStore) insertNewUserAgent(tx *bolt.Tx, ua string) (uaId int, err error) {
	bf := tx.Bucket([]byte(userAgentsFwdBn))
	bf.FillPercent = 1.0 // we only do appends
	br := tx.Bucket([]byte(userAgentsRevBn))
	br.FillPercent = 1.0 // we only do appends

	bUaId := bf.Get([]byte(ua))
	if bUaId != nil {
		return btoi(bUaId), nil
	}

	next, _ := bf.NextSequence()
	uaId = int(next)
	if err = bf.Put([]byte(ua), itob(uaId)); err != nil {
		return
	}
	if err = br.Put(itob(uaId), []byte(ua)); err != nil {
		return
	}

	return uaId, err
}

func (s sqliteStore) insertDataUpdateClientEntries(tx *bolt.Tx, duId int, cd []ClientData) error {
	if len(cd) == 0 {
		return nil
	}

	b := tx.Bucket([]byte(clientDataBn))
	b.FillPercent = 1.0 // we only do appends

	data := []clientDataDb{}
	for _, c := range cd {
		uaId, err := s.insertNewUserAgent(tx, c.UserAgent)
		if err != nil {
			return err
		}
		data = append(data, clientDataDb{c.Ip, uaId, c.BytesSent})
	}

	jsonData, err := json.Marshal(data)
	if err != nil {
		return err
	}
	return b.Put(itob(duId), jsonData)
}

func (s sqliteStore) appendItem(tx *bolt.Tx, du dataUpdateDb, cd []ClientData, src sourceDb) (err error) {
	var srcId int
	if srcId, err = s.insertNewSource(tx, src); err != nil {
		return
	}

	if err = s.insertDataUpdateEntry(srcId, &du); err != nil {
		return
	}

	if err = s.insertDataUpdateClientEntries(tx, du.Id, cd); err != nil {
		return
	}

	return
}

func (s sqliteStore) Append(update StatisticsData) (err error) {
	return s.AppendMany([]StatisticsData{update})
}

func (s sqliteStore) AppendMany(updates []StatisticsData) (err error) {
	var tx *gorp.Transaction
	tx, err = s.db.Begin()
	if err != nil {
		return
	}

	err = s.dbBolt.Update(func(tx *bolt.Tx) error {
		for _, update := range updates {
			du, cd, src := updateFromStatisticsData(update)
			if err := s.appendItem(tx, du, cd, src); err != nil {
				return err
			}
		}
		return nil
	})

	if err != nil {
		tx.Rollback()
		return
	}

	return tx.Commit()
}

func (s sqliteStore) getSource(tx *bolt.Tx, id int) (res sourceDb, err error) {
	b := tx.Bucket([]byte(sourcesRevBn))

	jsonData := b.Get(itob(id))
	if jsonData == nil {
		err = ErrNotFound
		return
	}
	if err = json.Unmarshal(jsonData, &res); err != nil {
		return
	}
	return
}

func (s sqliteStore) GetSources() (res []SourceId, err error) {
	res = []SourceId{}
	err = s.dbBolt.View(func(tx *bolt.Tx) error {
		b := tx.Bucket([]byte(sourcesRevBn))
		c := b.Cursor()
		for k, v := c.First(); k != nil; k, v = c.Next() {
			var s sourceDb
			if err := json.Unmarshal(v, &s); err != nil {
				return err
			}
			var src SourceId
			src.CopyFromSourceDb(s)
			res = append(res, src)
		}
		return nil
	})
	return
}

func (s sqliteStore) GetSource(id int) (res SourceId, err error) {
	err = s.dbBolt.View(func(tx *bolt.Tx) error {
		src, err := s.getSource(tx, id)
		if err != nil {
			return err
		}
		res.CopyFromSourceDb(src)
		return nil
	})
	return
}

func (s sqliteStore) GetUpdate(id int) (res dataUpdateDb, err error) {
	err = s.db.SelectOne(&res, "select * from "+dataUpdatesTn+" where Id = ?", id)
	return
}

func (s sqliteStore) getClientsByUpdateId(tx *bolt.Tx, id int) (res []ClientData, err error) {
	bc := tx.Bucket([]byte(clientDataBn))
	bu := tx.Bucket([]byte(userAgentsRevBn))

	jsonData := bc.Get(itob(id))
	if jsonData == nil {
		return
	}
	data := []clientDataDb{}
	if err = json.Unmarshal(jsonData, &data); err != nil {
		return
	}
	for _, c := range data {
		cd := ClientData{Ip: c.Ip, BytesSent: c.BytesSent}
		ua := bu.Get(itob(c.UserAgentId))
		if ua != nil {
			cd.UserAgent = string(ua)
		}
		res = append(res, cd)
	}
	return
}

func (s sqliteStore) CreateStatisticsDataFrom(tx *bolt.Tx, dat dataUpdateDb) (res StatisticsData, err error) {
	var clients []ClientData
	if clients, err = s.getClientsByUpdateId(tx, dat.Id); err != nil {
		return
	}
	var src sourceDb
	if src, err = s.getSource(tx, dat.SourceId); err != nil {
		return
	}

	res.CopyFromDataUpdateDb(dat, s.hubId)
	res.Hostname = src.Hostname
	res.StreamId.ContentId = src.StreamId.ContentId
	res.StreamId.Format = src.StreamId.Format
	res.StreamId.Quality = src.StreamId.Quality
	res.Tags = src.Tags
	res.Data.Clients = clients
	return
}

func (s sqliteStore) CreateStatisticsDatasFrom(dat []interface{}) (res []StatisticsData, err error) {
	err = s.dbBolt.View(func(tx *bolt.Tx) error {
		for i := range dat {
			sd, err := s.CreateStatisticsDataFrom(tx, *dat[i].(*dataUpdateDb))
			if err != nil {
				return err
			}
			res = append(res, sd)
		}
		return nil
	})
	return
}

func (s sqliteStore) GetUpdatesAfter(id int) (res []StatisticsData, err error) {
	parameters := make(map[string]interface{})
	sql := "SELECT * FROM " + dataUpdatesTn + " WHERE Id > :afterUpdateId limit :limit"
	parameters["afterUpdateId"] = id
	parameters["limit"] = 5000
	var updates []interface{}
	updates, err = s.db.Select(dataUpdateDb{}, sql, parameters)
	s5tl.Printf("sql: %s", sql)
	if err == nil {
		res, _ = s.CreateStatisticsDatasFrom(updates)
	}
	return
}

func (s sqliteStore) GetUpdates(filter *StatsFilter) (res []StatisticsData, err error) {
	err = fmt.Errorf("not implemented!")
	return
	// 	limit := 5000
	// 	if filter.limit == nil {
	// 		filter.limit = &limit
	// 	} else if *filter.limit > limit {
	// 		*filter.limit = limit
	// 	}
	// 	sourceSql, parameters := getFilteredDataUpdateSelect(filter)
	// 	sql := "SELECT " + updateColumnSelect + " FROM " + sourceSql
	// 	s5tl.Printf("store: sql: %s", sql)
	// 	var updates []interface{}
	// 	updates, err = s.db.Select(dataUpdateQueryResult{}, sql, parameters)
	// 	if err == nil {
	// 		res, _ = s.CreateStatisticsDatasFrom(updates)
	// 	}
	// 	return
}

type lastUpdateQueryResult struct {
	MaxDataUpdateId *int
}

func (s sqliteStore) GetLastUpdateForUuid(uuid string) (updateId *int, err error) {
	result := lastUpdateQueryResult{}
	err = s.db.SelectOne(
		&result,
		"select max(SourceHubDataUpdateId) as MaxDataUpdateId from "+dataUpdatesTn+" where SourceHubUuid = ?",
		uuid)
	if err == nil {
		updateId = result.MaxDataUpdateId
	} else {
		s5l.Printf("db: failed to find max SourceHubDataUpdateId for %s: %v", uuid, err)
	}
	return
}

func (s sqliteStore) GetLastUpdateId() (updateId *int, err error) {
	result := lastUpdateQueryResult{}
	err = s.db.SelectOne(&result, "select max(Id) as MaxDataUpdateId from "+dataUpdatesTn)
	if err == nil {
		updateId = result.MaxDataUpdateId
	} else {
		s5l.Printf("db: failed to find max DataUpdateId: %v", err)
	}
	return
}

type statsResult struct {
	UpdateCount   *int
	HubCount      *int
	SourcesCount  *int
	ClientCount   *float32
	BytesSent     *uint
	BytesReceived *uint
	StartTime     *int64
	LastStartTime *int64
}

type StatsResult struct {
	UpdateCount   int
	HubCount      int
	SourcesCount  int
	ClientCount   float32
	BytesSent     uint
	BytesReceived uint
	StartTime     time.Time
	LastStartTime time.Time
}

func toApiStatsResult(value statsResult) (res StatsResult) {
	if value.UpdateCount != nil {
		res.UpdateCount = *value.UpdateCount
	}
	if value.HubCount != nil {
		res.HubCount = *value.HubCount
	}
	if value.SourcesCount != nil {
		res.SourcesCount = *value.SourcesCount
	}
	if value.ClientCount != nil {
		res.ClientCount = *value.ClientCount
	}
	if value.BytesSent != nil {
		res.BytesSent = *value.BytesSent
	}
	if value.BytesReceived != nil {
		res.BytesReceived = *value.BytesReceived
	}
	if value.StartTime != nil {
		res.StartTime = time.Unix(*value.StartTime, 0)
	}
	if value.LastStartTime != nil {
		res.LastStartTime = time.Unix(*value.LastStartTime, 0)
	}
	return res
}

// var (
// 	statsGroupSelect = `
// SELECT
// 	count(*) as UpdateCount,
// 	SourceHubUuid as SourceHubUuid,
// 	count(distinct SourceId) as SourcesCount,
// 	avg(ClientCount) as ClientCount,
// 	sum(BytesSent) as BytesSent,
// 	sum(BytesReceived) as BytesReceived,
// 	min(StartTime) as StartTime,
// 	max(StartTime) as LastStartTime
// FROM
// `
// 	statsGroupClause = `
// GROUP BY
// 	SourceHubUuid
// `
// 	statsAggregateSelect = `
// SELECT
// 	sum(UpdateCount) as UpdateCount,
// 	count(distinct SourceHubUuid) as HubCount,
// 	sum(SourcesCount) as SourcesCount,
// 	sum(ClientCount) as ClientCount,
// 	sum(BytesSent) as BytesSent,
// 	sum(BytesReceived) as BytesReceived,
// 	min(StartTime) as StartTime,
// 	max(LastStartTime) as LastStartTime
// FROM
// `
// )

func (s sqliteStore) GetStats(filter *StatsFilter) (StatsResult, error) { // (map[string]interface{}, error) {
	return StatsResult{}, fmt.Errorf("not implemented!")
	// sourceSql, parameters := getFilteredDataUpdateSelect(filter)
	// _ = sourceSql
	// sql := fmt.Sprintf("%s (%s %s %s)", statsAggregateSelect, statsGroupSelect, sourceSql, statsGroupClause)
	// s5tl.Printf("store: stats sql: %s", sql)
	// res := statsResult{}
	// err := s.db.SelectOne(&res, sql, parameters)
	// if err == nil {
	// 	return toApiStatsResult(res), nil
	// }
	// return StatsResult{}, err
}

func (s sqliteStore) GetStoreId() (uuid string, err error) {
	uuid, err = s.db.SelectStr("select Value from HubInfo where Name = ?", "HubUuid")
	return
}

func NewStore(sqlitePath, boltPath string) (sqliteStore, error) {
	boltDb, err := initDbBolt(boltPath)
	if err != nil {
		return sqliteStore{}, err
	}

	db, hubid, err := initDbSqlite(sqlitePath)
	if err != nil {
		boltDb.Close()
		return sqliteStore{}, err
	}
	return sqliteStore{db, boltDb, hubid}, nil
}

func (s sqliteStore) Close() {
	s.db.Db.Close()
	s.dbBolt.Close()
}