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

package sfive

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

	_ "github.com/mattn/go-sqlite3"

	"code.google.com/p/go-uuid/uuid"
	"github.com/coopernurse/gorp"
)

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

func tagsFromStatisticsData(value StatisticsData) []tagDb {
	tags := make([]tagDb, len(value.SourceId.Tags))
	for i := range value.SourceId.Tags {
		tags[i] = tagDb{Id: -1, Name: value.SourceId.Tags[i]}
	}
	return tags
}

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

func clientsFromStatisticsData(value StatisticsData) []clientDataDb {
	res := make([]clientDataDb, len(value.Data.Clients))
	for i := range value.Data.Clients {
		res[i] = clientDataDb{-1, -1, value.Data.Clients[i]}
	}
	return res
}

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

func updateFromStatisticsData(value StatisticsData) (dataUpdateDb, []clientDataDb, sourceDb, []tagDb) {
	du := dataUpdateFromStatisticsData(value)
	cd := clientsFromStatisticsData(value)
	src := sourceFromStatisticsData(value)
	tags := tagsFromStatisticsData(value)

	return du, cd, src, tags
}

func initDb(path string) (res *gorp.DbMap, hubId string, err error) {
	// connect to db using standard Go database/sql API
	db, err := sql.Open("sqlite3", path)
	if err != nil {
		return
	}

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

	dbmap.AddTableWithName(tagDb{}, tagsTn).SetKeys(true, "Id").ColMap("Name").SetUnique(true)
	dbmap.AddTableWithName(sourceTagsDb{}, sourceTagsTn).SetKeys(false, "TagId", "SourceId")
	dbmap.AddTableWithName(sourceDb{}, sourcesTn).SetKeys(true, "Id").SetUniqueTogether("ContentId", "Format", "Quality", "Hostname")
	dbmap.AddTableWithName(clientDataDb{}, clientdataUpdatesTn).SetKeys(true, "Id")
	dbmap.AddTableWithName(dataUpdateDb{}, dataUpdatesTn).SetKeys(true, "Id")
	dbmap.AddTableWithName(hubInfoDb{}, hubInfoTn).SetKeys(false, "Name")

	// TODO use some real migration, yadda yadda
	err = dbmap.CreateTablesIfNotExists()
	if 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
		}
	}

	res = dbmap
	return
}

func isEmptyFilter(filter *StatsFilter) bool {
	if filter == nil {
		return true
	}
	if filter.start == nil &&
		filter.end == nil &&
		filter.hostname == nil &&
		filter.contentId == nil &&
		filter.format == nil &&
		filter.quality == nil &&
		(filter.tagsAny == nil || len(filter.tagsAny) == 0) {
		return true
	}
	return false
}

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

type dataUpdateQueryResult struct {
	dataUpdateDb
	StreamId
	SourceId
}

func getFilteredDataUpdateSelect(filter *StatsFilter) (string, map[string]interface{}) {
	const baseQuery = "(select * from " + dataUpdatesTn + "," + sourcesTn + " on " + dataUpdatesTn + ".SourceId = " + sourcesTn + ".Id"
	if isEmptyFilter(filter) {
		return baseQuery + ")", nil
	}

	query := baseQuery + " where"
	parameters := make(map[string]interface{})
	needsAnd := false

	if filter.start != nil {
		query += insertAnd(&needsAnd)
		query += " StartTime >= :filterstart"
		parameters["filterstart"] = filter.start.Unix()
		needsAnd = true
	}
	if filter.end != nil {
		query += insertAnd(&needsAnd)
		query += " StartTime < :filterend"
		parameters["filterend"] = filter.end.Unix()
		needsAnd = true
	}

	// TODO other fields
	query += ")"
	return query, parameters
}

func (s sqliteStore) findTag(name string) (tag *tagDb, err error) {
	t := tagDb{}
	err = s.db.SelectOne(&t, "select * from "+tagsTn+" where Name = ?", name)
	if err == nil {
		tag = &t
	}
	return
}

func (s sqliteStore) insertNewTags(tags []tagDb) (err error) {
	for i := range tags {
		t, err := s.findTag(tags[i].Name)
		if err != nil {
			_, err = s.db.Exec("insert into "+tagsTn+" VALUES (NULL, ?)", tags[i].Name)
		}
		t, err = s.findTag(tags[i].Name)

		if err == nil {
			tags[i] = *t
		} else {
			break
		}
	}

	return
}

func (s sqliteStore) findSource(src sourceDb) (res *sourceDb, err error) {
	t := sourceDb{}
	err = s.db.SelectOne(
		&t,
		"select Id from "+sourcesTn+" where ContentId = ? and Format = ? and Quality = ? and Hostname = ?",
		src.ContentId,
		src.Format,
		src.Quality,
		src.Hostname)

	if err == nil {
		res = &t
	}

	return
}

func (s sqliteStore) insertNewSource(src *sourceDb) (err error) {
	t, err := s.findSource(*src)
	if err == nil {
		*src = *t
	} else {
		err = s.db.Insert(src)
	}

	return
}

func (s sqliteStore) insertSourceTagLinks(src sourceDb, tags []tagDb) (err error) {
	st := make([]sourceTagsDb, len(tags))
	for i := range tags {
		st[i].TagId = tags[i].Id
		st[i].SourceId = src.Id
	}
	for i := range st {
		_, err = s.db.Exec(
			"insert or ignore into "+sourceTagsTn+" values (?,?)",
			st[i].SourceId,
			st[i].TagId)
		// err = s.db.Insert(&st[i])
		if err != nil {
			// TODO
			//fmt.Printf("st\n")
			return
		}
	}
	return
}

func (s sqliteStore) insertDataUpdateEntry(src sourceDb, du *dataUpdateDb) (err error) {
	du.SourceId = src.Id
	err = s.db.Insert(du)
	if err != nil {
		//fmt.Printf("du\n")
		return
	}
	return
}

func (s sqliteStore) insertDataUpdateClientEntries(cd []clientDataDb, du dataUpdateDb) (err error) {
	for i := range cd {
		cd[i].DataUpdatesId = du.Id
		err = s.db.Insert(&cd[i])
		if err != nil {
			// TODO
			return
		}
	}
	return
}

// this function is the biggest pile of copy/pasted crap while sick that is still compilable.
func (s sqliteStore) Append(update StatisticsData) (err error) {
	du, cd, src, tags := updateFromStatisticsData(update)

	// s.db.TraceOn("", log.New(os.Stdout, "gorptest: ", log.Lmicroseconds))
	tx, err := s.db.Begin()
	if err != nil {
		return
	}

	err = s.insertNewTags(tags)
	if err != nil {
		return
	}

	err = s.insertNewSource(&src)
	if err != nil {
		//fmt.Printf("src\n")
		return
	}

	err = s.insertSourceTagLinks(src, tags)
	if err != nil {
		return
	}

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

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

	return tx.Commit()
}

func castArrayToString(value []interface{}) []string {
	res := make([]string, len(value))
	for i := range value {
		res[i] = value[i].(string)
	}
	return res
}

func (s sqliteStore) GetTags() ([]string, error) {
	res, dbErr := s.db.Select("", "select Name from "+tagsTn)
	if dbErr == nil {
		sRes := castArrayToString(res)
		return sRes, nil
	}
	return nil, dbErr
}

func (s sqliteStore) GetSources() (res []sourceDb, err error) {
	qres, err := s.db.Select(sourceDb{}, "select * from "+sourcesTn)
	if err == nil {
		res = make([]sourceDb, len(qres))
		for i := range qres {
			res[i] = *qres[i].(*sourceDb)
		}
	}
	return
}

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

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

var (
	updateColumnSelect = `
	Id,
	SourceHubUuid,
	SourceHubDataUpdateId,
	StartTime,
	Duration,
	ClientCount,
	BytesReceived,
	BytesSent,
	Hostname,
	ContentId,
	Format,
	Quality
`
)

func (s sqliteStore) GetUpdatesAfter(id int) (res []StatisticsData, err error) {
	updates, err := s.db.Select(
		dataUpdateQueryResult{},
		"SELECT "+updateColumnSelect+" FROM "+dataUpdatesTn+","+sourcesTn+" ON "+dataUpdatesTn+".SourceId = "+sourcesTn+".Id WHERE "+dataUpdatesTn+".Id > ?",
		id)
	if err == nil {
		res = make([]StatisticsData, len(updates))
		for i := range updates {
			t := *updates[i].(*dataUpdateQueryResult)
			res[i].CopyFromDataUpdateDb(t.dataUpdateDb, s.hubId)
			res[i].Hostname = t.Hostname
			res[i].StreamId.ContentId = t.ContentId
			res[i].StreamId.Format = t.Format
			res[i].StreamId.Quality = t.Quality
			// TODO clients
			// TODO tags
		}
	}
	return
}

func (s sqliteStore) GetUpdates(filter *StatsFilter) (res []StatisticsData, err error) {
	sourceSql, parameters := getFilteredDataUpdateSelect(filter)
	updates, err := s.db.Select(
		dataUpdateQueryResult{},
		"SELECT "+updateColumnSelect+" FROM "+sourceSql,
		parameters)
	if err == nil {
		res = make([]StatisticsData, len(updates))
		for i := range updates {
			t := *updates[i].(*dataUpdateQueryResult)
			res[i].CopyFromDataUpdateDb(t.dataUpdateDb, s.hubId)
			res[i].Hostname = t.Hostname
			res[i].StreamId.ContentId = t.ContentId
			res[i].StreamId.Format = t.Format
			res[i].StreamId.Quality = t.Quality
			// TODO clients
			// TODO tags
		}
	}
	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
}

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) {
	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(path string) (store sqliteStore, err error) {
	db, hubid, err := initDb(path)
	if err != nil {
		return
	}
	store = sqliteStore{db, hubid}
	return
}

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