hooks.go

package server

import (
	"bytes"
	"errors"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/tidwall/buntdb"
	"github.com/tidwall/gjson"
	"github.com/tidwall/resp"
	"github.com/tidwall/tile38/internal/endpoint"
	"github.com/tidwall/tile38/internal/glob"
	"github.com/tidwall/tile38/internal/log"
)

var hookLogSetDefaults = &buntdb.SetOptions{
	Expires: true, // automatically delete after 30 seconds
	TTL:     time.Second * 30,
}

func byHookName(a, b interface{}) bool {
	return a.(*Hook).Name < b.(*Hook).Name
}

func (s *Server) cmdSetHook(msg *Message) (
	res resp.Value, d commandDetails, err error,
) {
	channel := msg.Command() == "setchan"
	start := time.Now()
	vs := msg.Args[1:]
	var name, urls, cmd string
	var ok bool
	if vs, name, ok = tokenval(vs); !ok || name == "" {
		return NOMessage, d, errInvalidNumberOfArguments
	}
	var endpoints []string
	if channel {
		endpoints = []string{"local://" + name}
	} else {
		if vs, urls, ok = tokenval(vs); !ok || urls == "" {
			return NOMessage, d, errInvalidNumberOfArguments
		}
		for _, url := range strings.Split(urls, ",") {
			url = strings.TrimSpace(url)
			err := s.epc.Validate(url)
			if err != nil {
				log.Errorf("sethook: %v", err)
				return resp.SimpleStringValue(""), d, errInvalidArgument(url)
			}
			endpoints = append(endpoints, url)
		}
	}
	var commandvs []string
	var cmdlc string
	var types map[string]bool
	var expires float64
	var expiresSet bool
	metaMap := make(map[string]string)
	for {
		commandvs = vs
		if vs, cmd, ok = tokenval(vs); !ok || cmd == "" {
			return NOMessage, d, errInvalidNumberOfArguments
		}
		cmdlc = strings.ToLower(cmd)
		switch cmdlc {
		default:
			return NOMessage, d, errInvalidArgument(cmd)
		case "meta":
			var metakey string
			var metaval string
			if vs, metakey, ok = tokenval(vs); !ok || metakey == "" {
				return NOMessage, d, errInvalidNumberOfArguments
			}
			if vs, metaval, ok = tokenval(vs); !ok || metaval == "" {
				return NOMessage, d, errInvalidNumberOfArguments
			}
			metaMap[metakey] = metaval
			continue
		case "ex":
			var s string
			if vs, s, ok = tokenval(vs); !ok || s == "" {
				return NOMessage, d, errInvalidNumberOfArguments
			}
			v, err := strconv.ParseFloat(s, 64)
			if err != nil {
				return NOMessage, d, errInvalidArgument(s)
			}
			expires = v
			expiresSet = true
			continue
		case "nearby":
			types = nearbyTypes
		case "within", "intersects":
			types = withinOrIntersectsTypes
		}
		break
	}
	args, err := s.cmdSearchArgs(true, cmdlc, vs, types)
	if args.usingLua() {
		defer args.Close()
	}
	if err != nil {
		return NOMessage, d, err
	}
	if !args.fence {
		return NOMessage, d, errors.New("missing FENCE argument")
	}
	args.cmd = cmdlc
	cmsg := &Message{}
	*cmsg = *msg
	cmsg.Args = make([]string, len(commandvs))
	copy(cmsg.Args, commandvs)
	metas := make([]FenceMeta, 0, len(metaMap))
	for key, val := range metaMap {
		metas = append(metas, FenceMeta{key, val})
	}
	sort.Sort(hookMetaByName(metas))

	hook := &Hook{
		Key:       args.key,
		Name:      name,
		Endpoints: endpoints,
		Fence:     &args,
		Message:   cmsg,
		epm:       s.epc,
		Metas:     metas,
		channel:   channel,
		cond:      sync.NewCond(&sync.Mutex{}),
		counter:   &s.statsTotalMsgsSent,
	}
	if expiresSet {
		hook.expires =
			time.Now().Add(time.Duration(expires * float64(time.Second)))
	}
	if !channel {
		hook.db = s.qdb
	}
	var wr bytes.Buffer
	hook.ScanWriter, err = s.newScanWriter(
		&wr, cmsg, args.key, args.output, args.precision, args.globs, false,
		args.cursor, args.limit, args.wheres, args.whereins, args.whereevals,
		args.nofields)
	if err != nil {

		return NOMessage, d, err
	}
	prevHook, _ := s.hooks.Get(&Hook{Name: name}).(*Hook)
	if prevHook != nil {
		if prevHook.channel != channel {
			return NOMessage, d,
				errors.New("hooks and channels cannot share the same name")
		}
		if prevHook.Equals(hook) {
			// it was a match so we do nothing. But let's signal just
			// for good measure.
			prevHook.Signal()
			if !hook.expires.IsZero() {
				s.hookExpires.Set(hook)
			}
			switch msg.OutputType {
			case JSON:
				return OKMessage(msg, start), d, nil
			case RESP:
				return resp.IntegerValue(0), d, nil
			}
		}
		prevHook.Close()
		s.hooks.Delete(prevHook)
		s.hooksOut.Delete(prevHook)
		if !prevHook.expires.IsZero() {
			s.hookExpires.Delete(prevHook)
		}
		s.groupDisconnectHook(name)
	}

	d.updated = true
	d.timestamp = time.Now()

	s.hooks.Set(hook)
	if hook.Fence.detect == nil || hook.Fence.detect["outside"] {
		s.hooksOut.Set(hook)
	}

	// remove previous hook from spatial index
	if prevHook != nil && prevHook.Fence != nil && prevHook.Fence.obj != nil {
		rect := prevHook.Fence.obj.Rect()
		s.hookTree.Delete(
			[2]float64{rect.Min.X, rect.Min.Y},
			[2]float64{rect.Max.X, rect.Max.Y},
			prevHook)
		if prevHook.Fence.detect["cross"] {
			s.hookCross.Delete(
				[2]float64{rect.Min.X, rect.Min.Y},
				[2]float64{rect.Max.X, rect.Max.Y},
				prevHook)
		}
	}
	// add hook to spatial index
	if hook != nil && hook.Fence != nil && hook.Fence.obj != nil {
		rect := hook.Fence.obj.Rect()
		s.hookTree.Insert(
			[2]float64{rect.Min.X, rect.Min.Y},
			[2]float64{rect.Max.X, rect.Max.Y},
			hook)
		if hook.Fence.detect["cross"] {
			s.hookCross.Insert(
				[2]float64{rect.Min.X, rect.Min.Y},
				[2]float64{rect.Max.X, rect.Max.Y},
				hook)
		}
	}

	hook.Open() // Opens a goroutine to notify the hook
	if !hook.expires.IsZero() {
		s.hookExpires.Set(hook)
	}
	switch msg.OutputType {
	case JSON:
		return OKMessage(msg, start), d, nil
	case RESP:
		return resp.IntegerValue(1), d, nil
	}
	return NOMessage, d, nil
}

func byHookExpires(a, b interface{}) bool {
	ha := a.(*Hook)
	hb := b.(*Hook)
	if ha.expires.Before(hb.expires) {
		return true
	}
	if ha.expires.After(hb.expires) {
		return false
	}
	return ha.Name < hb.Name
}

func (s *Server) cmdDELHOOKop(name string, channel bool) (updated bool) {
	hook, _ := s.hooks.Get(&Hook{Name: name}).(*Hook)
	if hook == nil || hook.channel != channel {
		return false
	}
	hook.Close()
	// remove hook from maps
	s.hooks.Delete(hook)
	s.hooksOut.Delete(hook)
	if !hook.expires.IsZero() {
		s.hookExpires.Delete(hook)
	}
	// remove any hook / object connections
	s.groupDisconnectHook(hook.Name)
	// remove hook from spatial index
	if hook.Fence != nil && hook.Fence.obj != nil {
		rect := hook.Fence.obj.Rect()
		s.hookTree.Delete(
			[2]float64{rect.Min.X, rect.Min.Y},
			[2]float64{rect.Max.X, rect.Max.Y},
			hook)
		if hook.Fence.detect["cross"] {
			s.hookCross.Delete(
				[2]float64{rect.Min.X, rect.Min.Y},
				[2]float64{rect.Max.X, rect.Max.Y},
				hook)
		}
	}
	return true
}

func (s *Server) cmdDelHook(msg *Message) (
	res resp.Value, d commandDetails, err error,
) {
	channel := msg.Command() == "delchan"
	start := time.Now()
	vs := msg.Args[1:]

	var name string
	var ok bool
	if vs, name, ok = tokenval(vs); !ok || name == "" {
		return NOMessage, d, errInvalidNumberOfArguments
	}
	if len(vs) != 0 {
		return NOMessage, d, errInvalidNumberOfArguments
	}

	d.updated = s.cmdDELHOOKop(name, channel)
	d.timestamp = time.Now()

	switch msg.OutputType {
	case JSON:
		return OKMessage(msg, start), d, nil
	case RESP:
		if d.updated {
			return resp.IntegerValue(1), d, nil
		}
		return resp.IntegerValue(0), d, nil
	}
	return
}

func (s *Server) cmdPDelHook(msg *Message) (
	res resp.Value, d commandDetails, err error,
) {
	channel := msg.Command() == "pdelchan"
	start := time.Now()
	vs := msg.Args[1:]

	var pattern string
	var ok bool
	if vs, pattern, ok = tokenval(vs); !ok || pattern == "" {
		return NOMessage, d, errInvalidNumberOfArguments
	}
	if len(vs) != 0 {
		return NOMessage, d, errInvalidNumberOfArguments
	}

	count := 0
	var hooks []*Hook
	s.forEachHookByPattern(pattern, channel, func(hook *Hook) bool {
		hooks = append(hooks, hook)
		return true
	})

	for _, hook := range hooks {
		if hook.channel != channel {
			continue
		}
		s.cmdDELHOOKop(hook.Name, channel)
		d.updated = true
		count++
	}
	d.timestamp = time.Now()

	switch msg.OutputType {
	case JSON:
		return OKMessage(msg, start), d, nil
	case RESP:
		return resp.IntegerValue(count), d, nil
	}
	return
}

func (s *Server) forEachHookByPattern(
	pattern string, channel bool, iter func(hook *Hook) bool,
) {
	g := glob.Parse(pattern, false)
	hasUpperLimit := g.Limits[1] != ""
	s.hooks.Ascend(&Hook{Name: g.Limits[0]}, func(v interface{}) bool {
		hook := v.(*Hook)
		if hasUpperLimit && hook.Name > g.Limits[1] {
			return false
		}
		if hook.channel == channel {
			match, _ := glob.Match(pattern, hook.Name)
			if match {
				return iter(hook)
			}
		}
		return true
	})
}

func (s *Server) cmdHooks(msg *Message) (
	res resp.Value, err error,
) {
	channel := msg.Command() == "chans"
	start := time.Now()
	vs := msg.Args[1:]

	var pattern string
	var ok bool

	if vs, pattern, ok = tokenval(vs); !ok || pattern == "" {
		return NOMessage, errInvalidNumberOfArguments
	}
	if len(vs) != 0 {
		return NOMessage, errInvalidNumberOfArguments
	}

	switch msg.OutputType {
	case JSON:
		buf := &bytes.Buffer{}
		buf.WriteString(`{"ok":true,`)
		if channel {
			buf.WriteString(`"chans":[`)
		} else {
			buf.WriteString(`"hooks":[`)
		}
		var i int
		s.forEachHookByPattern(pattern, channel, func(hook *Hook) bool {
			var ttl = -1
			if !hook.expires.IsZero() {
				ttl = int(hook.expires.Sub(start).Seconds())
				if ttl < 0 {
					ttl = 0
				}
			}
			if i > 0 {
				buf.WriteByte(',')
			}
			buf.WriteString(`{`)
			buf.WriteString(`"name":` + jsonString(hook.Name))
			buf.WriteString(`,"key":` + jsonString(hook.Key))
			buf.WriteString(`,"ttl":` + strconv.Itoa(ttl))
			if !channel {
				buf.WriteString(`,"endpoints":[`)
				for i, endpoint := range hook.Endpoints {
					if i > 0 {
						buf.WriteByte(',')
					}
					buf.WriteString(jsonString(endpoint))
				}
				buf.WriteString(`]`)
			}
			buf.WriteString(`,"command":[`)
			for i, v := range hook.Message.Args {
				if i > 0 {
					buf.WriteString(`,`)
				}
				buf.WriteString(jsonString(v))
			}
			buf.WriteString(`],"meta":{`)
			for i, meta := range hook.Metas {
				if i > 0 {
					buf.WriteString(`,`)
				}
				buf.WriteString(jsonString(meta.Name))
				buf.WriteString(`:`)
				buf.WriteString(jsonString(meta.Value))
			}
			buf.WriteString(`}}`)
			i++
			return true
		})
		buf.WriteString(`],"elapsed":"` +
			time.Since(start).String() + "\"}")
		return resp.StringValue(buf.String()), nil
	case RESP:
		var vals []resp.Value
		s.forEachHookByPattern(pattern, channel, func(hook *Hook) bool {
			var hvals []resp.Value
			hvals = append(hvals, resp.StringValue(hook.Name))
			hvals = append(hvals, resp.StringValue(hook.Key))
			var evals []resp.Value
			for _, endpoint := range hook.Endpoints {
				evals = append(evals, resp.StringValue(endpoint))
			}
			hvals = append(hvals, resp.ArrayValue(evals))
			avals := make([]resp.Value, len(hook.Message.Args))
			for i := 0; i < len(hook.Message.Args); i++ {
				avals[i] = resp.StringValue(hook.Message.Args[i])
			}
			hvals = append(hvals, resp.ArrayValue(avals))
			var metas []resp.Value
			for _, meta := range hook.Metas {
				metas = append(metas, resp.StringValue(meta.Name))
				metas = append(metas, resp.StringValue(meta.Value))
			}
			hvals = append(hvals, resp.ArrayValue(metas))
			vals = append(vals, resp.ArrayValue(hvals))
			return true
		})
		return resp.ArrayValue(vals), nil
	}
	return resp.SimpleStringValue(""), nil
}

// Hook represents a hook.
type Hook struct {
	cond       *sync.Cond
	Key        string
	Name       string
	Endpoints  []string
	Message    *Message
	Fence      *liveFenceSwitches
	ScanWriter *scanWriter
	Metas      []FenceMeta
	db         *buntdb.DB
	channel    bool
	closed     bool
	opened     bool
	query      string
	epm        *endpoint.Manager
	expires    time.Time
	counter    *atomic.Int64 // counter that grows when a message was sent
	sig        int
}

// Expires returns when the hook expires. Required by the expire.Item interface.
func (h *Hook) Expires() time.Time {
	return h.expires
}

// Equals returns true if two hooks are equal
func (h *Hook) Equals(hook *Hook) bool {
	if h.Key != hook.Key ||
		h.Name != hook.Name ||
		len(h.Endpoints) != len(hook.Endpoints) ||
		len(h.Metas) != len(hook.Metas) {
		return false
	}
	if !h.expires.Equal(hook.expires) {
		return false
	}
	for i, endpoint := range h.Endpoints {
		if endpoint != hook.Endpoints[i] {
			return false
		}
	}
	for i, meta := range h.Metas {
		if meta.Name != hook.Metas[i].Name ||
			meta.Value != hook.Metas[i].Value {
			return false
		}
	}
	if len(h.Message.Args) != len(hook.Message.Args) {
		return false
	}
	for i := 0; i < len(h.Message.Args); i++ {
		if h.Message.Args[i] != hook.Message.Args[i] {
			return false
		}
	}
	return true
}

// FenceMeta is a meta key/value pair for fences
type FenceMeta struct {
	Name, Value string
}

type hookMetaByName []FenceMeta

func (arr hookMetaByName) Len() int {
	return len(arr)
}

func (arr hookMetaByName) Less(a, b int) bool {
	return arr[a].Name < arr[b].Name
}

func (arr hookMetaByName) Swap(a, b int) {
	arr[a], arr[b] = arr[b], arr[a]
}

// Open is called when a hook is first created. It calls the manager
// function in a goroutine
func (h *Hook) Open() {
	if h.channel {
		// nothing to open for channels
		return
	}
	h.cond.L.Lock()
	defer h.cond.L.Unlock()
	if h.opened {
		return
	}
	h.opened = true
	h.query = `{"hook":` + jsonString(h.Name) + `}`
	go h.manager()
}

// Close closed the hook and stop the manager function
func (h *Hook) Close() {
	if h.channel {
		// nothing to close for channels
		return
	}
	h.cond.L.Lock()
	defer h.cond.L.Unlock()
	if h.closed {
		return
	}
	h.closed = true
	h.cond.Broadcast()
}

// Signal can be called at any point to wake up the hook and
// notify the manager that there may be something new in the queue.
func (h *Hook) Signal() {
	if h.channel {
		// nothing to signal for channels
		return
	}
	h.cond.L.Lock()
	h.sig++
	h.cond.Broadcast()
	h.cond.L.Unlock()
}

// the manager is a forever loop that calls proc whenever there's a signal.
// it ends when the "closed" flag is set.
func (h *Hook) manager() {
	// lock the hook to waiting on signals
	h.cond.L.Lock()
	defer h.cond.L.Unlock()
	var sig int
	for {
		if h.closed {
			// the hook has closed, end manager
			return
		}
		sig = h.sig
		// unlock/logk the hook and send outgoing messages
		if !func() bool {
			h.cond.L.Unlock()
			defer h.cond.L.Lock()
			return h.proc()
		}() {
			// a send failed, try again in a moment
			time.Sleep(time.Second / 2)
			continue
		}
		if sig != h.sig {
			// there was another incoming signal
			continue
		}
		// wait on signal
		h.cond.Wait()
	}
}

// proc processes queued hook logs.
// returning true will indicate that all log entries have been
// successfully handled.
func (h *Hook) proc() (ok bool) {
	var keys, vals []string
	var ttls []time.Duration
	start := time.Now()
	err := h.db.Update(func(tx *buntdb.Tx) error {
		// get keys and vals
		err := tx.AscendGreaterOrEqual("hooks",
			h.query, func(key, val string) bool {
				if strings.HasPrefix(key, hookLogPrefix) {
					// Verify this hooks name matches the one in the notif
					if h.Name == gjson.Get(val, "hook").String() {
						keys = append(keys, key)
						vals = append(vals, val)
					}
				}
				return true
			},
		)
		if err != nil {
			return err
		}

		// delete the keys
		for _, key := range keys {
			ttl, err := tx.TTL(key)
			if err != nil {
				if err != buntdb.ErrNotFound {
					return err
				}
			}
			ttls = append(ttls, ttl)
			_, err = tx.Delete(key)
			if err != nil {
				if err != buntdb.ErrNotFound {
					return err
				}
			}
		}
		return nil
	})
	if err != nil {
		log.Error(err)
		return false
	}

	// send each val. on failure reinsert that one and all of the following
	for i, key := range keys {
		val := vals[i]
		idx := stringToUint64(key[len(hookLogPrefix):])
		var sent bool
		for _, endpoint := range h.Endpoints {
			err := h.epm.Send(endpoint, val)
			if err != nil {
				log.Debugf("Endpoint connect/send error: %v: %v: %v",
					idx, endpoint, err)
				continue
			}
			log.Debugf("Endpoint send ok: %v: %v: %v", idx, endpoint, err)
			sent = true
			h.counter.Add(1)
			break
		}
		if !sent {
			// failed to send. try to reinsert the remaining.
			// if this fails we lose log entries.
			keys = keys[i:]
			vals = vals[i:]
			ttls = ttls[i:]
			h.db.Update(func(tx *buntdb.Tx) error {
				for i, key := range keys {
					val := vals[i]
					ttl := ttls[i] - time.Since(start)
					if ttl > 0 {
						opts := &buntdb.SetOptions{
							Expires: true,
							TTL:     ttl,
						}
						_, _, err := tx.Set(key, val, opts)
						if err != nil {
							return err
						}
					}
				}
				return nil
			})
			return false
		}
	}
	return true
}