1package pipe
  2
  3import (
  4	"fmt"
  5	"time"
  6
  7	"github.com/picosh/pico/pkg/db"
  8	"github.com/picosh/pico/pkg/db/postgres"
  9)
 10
 11// DB is the storage pipe uses.
 12type DB interface {
 13	db.AuthDB
 14	FindUserByToken(token string) (*db.User, error)
 15	UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error
 16	UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error
 17	RemovePipeMonitor(userID, topic string) error
 18	FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error)
 19	FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error)
 20	InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error
 21	FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error)
 22}
 23
 24// PsqlDB is pipe's postgres storage: the queries apps share, and the ones
 25// only pipe makes.
 26type PsqlDB struct {
 27	*postgres.PsqlDB
 28}
 29
 30var _ DB = (*PsqlDB)(nil)
 31
 32func NewDB(base *postgres.PsqlDB) *PsqlDB {
 33	return &PsqlDB{PsqlDB: base}
 34}
 35
 36func (me *PsqlDB) UpsertPipeMonitor(userID, topic string, dur time.Duration, winEnd *time.Time) error {
 37	durStr := fmt.Sprintf("%d seconds", int64(dur.Seconds()))
 38	_, err := me.Db.Exec(
 39		`INSERT INTO pipe_monitors (user_id, topic, window_dur, window_end)
 40		VALUES ($1, $2, $3::interval, $4)
 41		ON CONFLICT (user_id, topic) DO UPDATE SET window_dur = $3::interval, window_end = $4, updated_at = NOW();`,
 42		userID,
 43		topic,
 44		durStr,
 45		winEnd,
 46	)
 47	return err
 48}
 49
 50func (me *PsqlDB) UpdatePipeMonitorLastPing(userID, topic string, lastPing *time.Time) error {
 51	_, err := me.Db.Exec(
 52		`UPDATE pipe_monitors SET last_ping = $3, updated_at = NOW() WHERE user_id = $1 AND topic = $2;`,
 53		userID,
 54		topic,
 55		lastPing,
 56	)
 57	return err
 58}
 59
 60func (me *PsqlDB) RemovePipeMonitor(userID, topic string) error {
 61	_, err := me.Db.Exec(
 62		`DELETE FROM pipe_monitors WHERE user_id = $1 AND topic = $2;`,
 63		userID,
 64		topic,
 65	)
 66	return err
 67}
 68
 69func (me *PsqlDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor, error) {
 70	monitor := &db.PipeMonitor{}
 71	err := me.Db.Get(monitor, `SELECT id, user_id, topic, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors WHERE user_id = $1 AND topic = $2;`, userID, topic)
 72	if err != nil {
 73		return nil, err
 74	}
 75	return monitor, nil
 76}
 77
 78func (me *PsqlDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
 79	var monitors []*db.PipeMonitor
 80	err := me.Db.Select(&monitors, `SELECT id, user_id, topic, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors WHERE user_id = $1 ORDER BY topic;`, userID)
 81	if err != nil {
 82		return nil, err
 83	}
 84	return monitors, nil
 85}
 86
 87func (me *PsqlDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error {
 88	durStr := fmt.Sprintf("%d seconds", int64(windowDur.Seconds()))
 89	_, err := me.Db.Exec(
 90		`INSERT INTO pipe_monitors_history (monitor_id, window_dur, window_end, last_ping) VALUES ($1, $2::interval, $3, $4)`,
 91		monitorID, durStr, windowEnd, lastPing,
 92	)
 93	return err
 94}
 95
 96func (me *PsqlDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error) {
 97	var history []*db.PipeMonitorHistory
 98	err := me.Db.Select(
 99		&history,
100		`SELECT id, monitor_id, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors_history WHERE monitor_id = $1 AND last_ping <= $2 AND window_end >= $3 ORDER BY last_ping ASC`,
101		monitorID, to, from,
102	)
103	if err != nil {
104		return nil, err
105	}
106	return history, nil
107}