package db

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

	"code.justin.tv/live/plucky/api/rpc"

	"code.justin.tv/live/plucky/api/util"

	"code.justin.tv/creator-collab/log"
	"code.justin.tv/creator-collab/log/errors"
	_ "github.com/mattn/go-sqlite3"
)

type DB struct {
	db     *sql.DB
	logger log.Logger
	mutex  *sync.Mutex
}

func New(logger log.Logger) (*DB, error) {
	db, err := sql.Open("sqlite3", ":memory:")
	if err != nil {
		return nil, errors.Wrap(err, "opening db connection failed")
	}

	// Prevent the connection pool from closing all connections to the database.
	// Since we are using an in-memory database, if all connections are closed, sqlite will discard the data.
	db.SetConnMaxLifetime(0)
	db.SetMaxIdleConns(1)

	// Prevent the connection pool from creating additional connections.
	// Since we are using an in-memory database, every new connection creates creates a new database.
	db.SetMaxOpenConns(1)

	_, err = db.Exec(`
		CREATE TABLE occurrences (
			occurrence_id  TEXT PRIMARY KEY,
			service        TEXT NOT NULL,
			fingerprint    TEXT NOT NULL,
			timestamp      INTEGER NOT NULL,
			payload        TEXT NOT NULL,
			hour_bucket    INTEGER NOT NULL,
			minute_bucket  INTEGER NOT NULL
		);
		CREATE INDEX occurrences_service_idx ON occurrences(service);
		CREATE INDEX occurrences_timestamp_idx ON occurrences(timestamp);

		CREATE TABLE time_range (
			service  TEXT PRIMARY KEY,
			ranges_json TEXT NOT NULL 
		);		
	`)
	if err != nil {
		return nil, errors.Wrap(err, "creating the tables failed")
	}

	return &DB{
		db:     db,
		logger: logger,
		mutex:  &sync.Mutex{},
	}, nil
}

type CachedTimeRanges struct {
	Real []TimeRange `json:"real"`
	Safe []TimeRange `json:"safe"`
}

func (d *DB) GetCachedTimeRanges(ctx context.Context, service string) (*CachedTimeRanges, error) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	logStartTime := time.Now()
	defer func() {
		d.logger.Info("GetCachedTimeRanges", log.Fields{
			"duration": time.Since(logStartTime),
		})
	}()

	row := d.db.QueryRowContext(ctx, `
		SELECT ranges_json 
		FROM time_range
		WHERE service = ?
		LIMIT 1;
		`, service)

	var jsonStr string
	err := row.Scan(&jsonStr)
	if err == sql.ErrNoRows {
		return &CachedTimeRanges{}, nil
	} else if err != nil {
		return nil, errors.Wrap(err, "loading cached time ranges failed", errors.Fields{
			"service": service,
		})
	}

	var ranges CachedTimeRanges
	err = json.Unmarshal([]byte(jsonStr), &ranges)
	if err != nil {
		return nil, errors.Wrap(err, "unmarshalling cached time ranges failed", errors.Fields{
			"service":     service,
			"ranges_json": jsonStr,
		})
	}

	return &ranges, nil
}

func (d *DB) GetHourTrendPointMap(ctx context.Context, service string, queryRange TimeRange) (map[string][]*rpc.TrendPoint, error) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	logStartTime := time.Now()
	defer func() {
		d.logger.Info("GetHourTrendPointMap", log.Fields{
			"duration": time.Since(logStartTime),
		})
	}()

	startTime := queryRange.Start.Unix()
	endTime := queryRange.End.Unix()

	rows, err := d.db.QueryContext(ctx, `
		SELECT hour_bucket, fingerprint, COUNT(*) 
		FROM occurrences
		WHERE service = ? AND timestamp >= ? AND timestamp <= ?
		GROUP BY hour_bucket, fingerprint
		ORDER BY hour_bucket DESC;
		`, service, startTime, endTime)
	if err != nil {
		return nil, errors.Wrap(err, "loading trend failed", errors.Fields{
			"service": service,
		})
	}

	trendMap := make(map[string][]*rpc.TrendPoint, 0)
	for rows.Next() {
		var hourBucket int64
		var fingerprint string
		var count int32

		err = rows.Scan(&hourBucket, &fingerprint, &count)
		if err != nil {
			return nil, errors.Wrap(err, "GetHourTrendPointMap - scanning row failed", errors.Fields{
				"service": service,
			})
		}

		start := util.UnixToTimestamp(hourBucket)

		tp := &rpc.TrendPoint{
			Start: start,
			Count: count,
		}
		points := trendMap[fingerprint]
		trendMap[fingerprint] = append(points, tp)
	}

	return trendMap, nil
}

func (d *DB) GetMinuteTrendPointMap(ctx context.Context, service string, queryRange TimeRange) (map[string][]*rpc.TrendPoint, error) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	logStartTime := time.Now()
	defer func() {
		d.logger.Info("GetMinuteTrendPointMap", log.Fields{
			"duration": time.Since(logStartTime),
		})
	}()

	startTime := queryRange.Start.Unix()
	endTime := queryRange.End.Unix()

	rows, err := d.db.QueryContext(ctx, `
		SELECT minute_bucket, fingerprint, COUNT(*) 
		FROM occurrences
		WHERE service = ? AND timestamp >= ? AND timestamp <= ?
		GROUP BY minute_bucket, fingerprint
		ORDER BY minute_bucket DESC;
		`, service, startTime, endTime)
	if err != nil {
		return nil, errors.Wrap(err, "loading trend failed", errors.Fields{
			"service": service,
		})
	}

	trendMap := make(map[string][]*rpc.TrendPoint, 0)
	for rows.Next() {
		var minuteBucket int64
		var fingerprint string
		var count int32

		err = rows.Scan(&minuteBucket, &fingerprint, &count)
		if err != nil {
			return nil, errors.Wrap(err, "GetMinuteTrendPointMap - scanning row failed", errors.Fields{
				"service": service,
			})
		}

		start := util.UnixToTimestamp(minuteBucket)

		tp := &rpc.TrendPoint{
			Start: start,
			Count: count,
		}
		points := trendMap[fingerprint]
		trendMap[fingerprint] = append(points, tp)
	}

	return trendMap, nil
}

func (d *DB) GetDayTrendPoints(ctx context.Context, service, fingerprint string, queryRange TimeRange) ([]*rpc.TrendPoint, error) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	logStartTime := time.Now()
	defer func() {
		d.logger.Info("GetDayTrendPoints", log.Fields{
			"duration": time.Since(logStartTime),
		})
	}()

	startTime := queryRange.Start.Unix()
	endTime := queryRange.End.Unix()

	rows, err := d.db.QueryContext(ctx, `
		SELECT hour_bucket, COUNT(*) 
		FROM occurrences
		WHERE service = ? AND fingerprint = ? AND timestamp >= ? AND timestamp <= ?
		GROUP BY hour_bucket
		ORDER BY hour_bucket DESC;
		`, service, fingerprint, startTime, endTime)
	if err != nil {
		return nil, errors.Wrap(err, "GetDayTrendPoints - loading rows failed", errors.Fields{
			"service":     service,
			"fingerprint": fingerprint,
		})
	}

	trendPoints := make([]*rpc.TrendPoint, 0)
	for rows.Next() {
		var hourBucket int64
		var count int32

		err = rows.Scan(&hourBucket, &count)
		if err != nil {
			return nil, errors.Wrap(err, "GetDayTrendPoints - scanning row failed", errors.Fields{
				"service":     service,
				"fingerprint": fingerprint,
			})
		}

		start := util.UnixToTimestamp(hourBucket)

		tp := &rpc.TrendPoint{
			Start: start,
			Count: count,
		}
		trendPoints = append(trendPoints, tp)
	}

	return trendPoints, nil
}

func (d *DB) GetOccurrences(ctx context.Context, serviceID string, queryRange TimeRange) ([]*rpc.Occurrence, error) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	logStartTime := time.Now()
	defer func() {
		d.logger.Info("GetOccurrences", log.Fields{
			"duration": time.Since(logStartTime),
		})
	}()

	startTime := queryRange.Start.Unix()
	endTime := queryRange.End.Unix()

	rows, err := d.db.QueryContext(ctx, `
		SELECT payload 
		FROM occurrences
		WHERE service = ? AND timestamp >= ? AND timestamp <= ?
		ORDER BY timestamp DESC;
		`, serviceID, startTime, endTime)
	if err != nil {
		return nil, errors.Wrap(err, "loading occurrences failed", errors.Fields{
			"service": serviceID,
		})
	}

	occurrences := make([]*rpc.Occurrence, 0)
	for rows.Next() {
		var payload string
		err = rows.Scan(&payload)
		if err != nil {
			return nil, errors.Wrap(err, "GetOccurrences - scanning row failed", errors.Fields{
				"service": serviceID,
			})
		}

		var o rpc.Occurrence
		err = json.Unmarshal([]byte(payload), &o)
		if err != nil {
			return nil, errors.Wrap(err, "GetOccurrences - unmarshalling payload failed", errors.Fields{
				"service": serviceID,
			})
		}

		occurrences = append(occurrences, &o)
	}

	if err = rows.Err(); err != nil {
		return nil, errors.Wrap(err, "GetOccurrences - scanning rows failed", errors.Fields{
			"service": serviceID,
		})
	}

	return occurrences, nil
}

func (d *DB) GetOccurrencesByFingerprint(ctx context.Context, serviceID, fingerprint string, queryRange TimeRange) ([]*rpc.Occurrence, error) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	logStartTime := time.Now()
	defer func() {
		d.logger.Info("GetOccurrencesByFingerprint", log.Fields{
			"duration": time.Since(logStartTime),
		})
	}()

	startTime := queryRange.Start.Unix()
	endTime := queryRange.End.Unix()

	rows, err := d.db.QueryContext(ctx, `
		SELECT payload 
		FROM occurrences
		WHERE service = ? AND fingerprint = ? AND timestamp >= ? AND timestamp <= ?
		ORDER BY timestamp DESC;
		`, serviceID, fingerprint, startTime, endTime)
	if err != nil {
		return nil, errors.Wrap(err, "GetOccurrencesByFingerprint - loading occurrences failed", errors.Fields{
			"service":     serviceID,
			"fingerprint": fingerprint,
		})
	}

	occurrences := make([]*rpc.Occurrence, 0)
	for rows.Next() {
		var payload string
		err = rows.Scan(&payload)
		if err != nil {
			return nil, errors.Wrap(err, "GetOccurrencesByFingerprint - scanning row failed", errors.Fields{
				"service":     serviceID,
				"fingerprint": fingerprint,
			})
		}

		var o rpc.Occurrence
		err = json.Unmarshal([]byte(payload), &o)
		if err != nil {
			return nil, errors.Wrap(err, "GetOccurrencesByFingerprint - unmarshalling payload failed", errors.Fields{
				"service":     serviceID,
				"fingerprint": fingerprint,
			})
		}

		occurrences = append(occurrences, &o)
	}

	if err = rows.Err(); err != nil {
		return nil, errors.Wrap(err, "GetOccurrencesByFingerprint - scanning rows failed", errors.Fields{
			"service":     serviceID,
			"fingerprint": fingerprint,
		})
	}

	return occurrences, nil
}

func (d *DB) InsertOccurrences(
	ctx context.Context, serviceID string, occurrences []*rpc.Occurrence, cachedTimeRanges *CachedTimeRanges) error {

	d.mutex.Lock()
	defer d.mutex.Unlock()

	logStartTime := time.Now()
	defer func() {
		d.logger.Info("InsertOccurrences", log.Fields{
			"duration":         time.Since(logStartTime),
			"occurrence_count": len(occurrences),
			"service_id":       serviceID,
		})
	}()

	d.logger.Debug(fmt.Sprintf("InsertOccurrences - cachedTimeRanges.Real"))
	for _, qr := range cachedTimeRanges.Real {
		d.logger.Debug(qr.String())
	}
	d.logger.Debug(fmt.Sprintf("InsertOccurrences - cachedTimeRanges.Safe"))
	for _, qr := range cachedTimeRanges.Safe {
		d.logger.Debug(qr.String())
	}

	rangesJson, err := json.Marshal(*cachedTimeRanges)
	if err != nil {
		return errors.Wrap(err, "InsertOccurrences - converting updatedRanges to json failed")
	}

	d.logger.Info("InsertOccurrences - create cachedTimeRanges", log.Fields{
		"service_id":      serviceID,
		"time_range_json": string(rangesJson),
	})

	tx, err := d.db.BeginTx(ctx, nil)
	if err != nil {
		return errors.Wrap(err, "InsertOccurrences - starting transaction failed")
	}

	for _, o := range occurrences {
		payloadBytes, err := json.Marshal(&o)
		if err != nil {
			d.logger.Error(err)
			continue
		}
		payloadStr := string(payloadBytes)

		hourBucket := util.GetHourBucket(o.Timestamp)
		minBucket := util.GetMinuteBucket(o.Timestamp)

		_, err = tx.ExecContext(ctx, `
			INSERT INTO occurrences (occurrence_id, service, fingerprint, timestamp, payload, hour_bucket, minute_bucket)
			VALUES (?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT (occurrence_id) DO UPDATE SET
				service=excluded.service,
			    fingerprint=excluded.fingerprint,
			    timestamp=excluded.timestamp,
			    payload=excluded.payload,
				hour_bucket=excluded.hour_bucket,
				minute_bucket=excluded.minute_bucket;				
		`, o.OccurrenceId, o.Service, o.Fingerprint, o.Timestamp.Seconds, payloadStr, hourBucket, minBucket)
		if err != nil {
			rollbackErr := tx.Rollback()
			if rollbackErr != nil {
				d.logger.Error(rollbackErr)
			}

			return errors.Wrap(err, "inserting occurrence failed", errors.Fields{
				"occurrence": o,
			})
		}
	}

	_, err = tx.ExecContext(ctx, `
			INSERT INTO time_range (service, ranges_json)
			VALUES (?, ?)
			ON CONFLICT (service) DO UPDATE SET
				service=excluded.service,
			    ranges_json=excluded.ranges_json;				
		`, serviceID, rangesJson)
	if err != nil {
		rollbackErr := tx.Rollback()
		if rollbackErr != nil {
			d.logger.Error(rollbackErr)
		}

		return errors.Wrap(err, "inserting time_range failed", errors.Fields{
			"service":         serviceID,
			"time_range_json": rangesJson,
		})
	}

	err = tx.Commit()
	if err != nil {
		return errors.Wrap(err, "InsertOccurrences - committing transaction failed")
	}

	return nil
}
