Add Twitch events

Signed-off-by: Knut Ahlers <knut@ahlers.me>

action_counter.go

@@ -74,18 +74,18 @@ type ActorCounter struct {
 	Counter     *string `json:"counter" yaml:"counter"`
 }
 
-func (a ActorCounter) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a ActorCounter) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.Counter == nil {
 		return false, nil
 	}
 
-	counterName, err := formatMessage(*a.Counter, m, r, nil)
+	counterName, err := formatMessage(*a.Counter, m, r, eventData)
 	if err != nil {
 		return false, errors.Wrap(err, "preparing response")
 	}
 
 	if a.CounterSet != nil {
-		parseValue, err := formatMessage(*a.CounterSet, m, r, nil)
+		parseValue, err := formatMessage(*a.CounterSet, m, r, eventData)
 		if err != nil {
 			return false, errors.Wrap(err, "execute counter value template")
 		}

action_script.go

@@ -21,14 +21,14 @@ type ActorScript struct {
 	Command []string `json:"command" yaml:"command"`
 }
 
-func (a ActorScript) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a ActorScript) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if len(a.Command) == 0 {
 		return false, nil
 	}
 
 	var command []string
 	for _, arg := range a.Command {
-		tmp, err := formatMessage(arg, m, r, nil)
+		tmp, err := formatMessage(arg, m, r, eventData)
 		if err != nil {
 			return false, errors.Wrap(err, "execute command argument template")
 		}
@@ -80,7 +80,7 @@ func (a ActorScript) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (pr
 	}
 
 	for _, action := range actions {
-		apc, err := triggerActions(c, m, r, action)
+		apc, err := triggerActions(c, m, r, action, eventData)
 		if err != nil {
 			return preventCooldown, errors.Wrap(err, "execute returned action")
 		}

action_setvar.go

@@ -59,12 +59,12 @@ type ActorSetVariable struct {
 	Set      string `json:"set" yaml:"set"`
 }
 
-func (a ActorSetVariable) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a ActorSetVariable) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.Variable == "" {
 		return false, nil
 	}
 
-	varName, err := formatMessage(a.Variable, m, r, nil)
+	varName, err := formatMessage(a.Variable, m, r, eventData)
 	if err != nil {
 		return false, errors.Wrap(err, "preparing variable name")
 	}
@@ -76,7 +76,7 @@ func (a ActorSetVariable) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule
 		)
 	}
 
-	value, err := formatMessage(a.Set, m, r, nil)
+	value, err := formatMessage(a.Set, m, r, eventData)
 	if err != nil {
 		return false, errors.Wrap(err, "preparing value")
 	}

actions.go

@@ -24,7 +24,7 @@ func registerAction(af plugins.ActorCreationFunc) {
 	availableActions = append(availableActions, af)
 }
 
-func triggerActions(c *irc.Client, m *irc.Message, rule *plugins.Rule, ra *plugins.RuleAction) (preventCooldown bool, err error) {
+func triggerActions(c *irc.Client, m *irc.Message, rule *plugins.Rule, ra *plugins.RuleAction, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	availableActionsLock.RLock()
 	defer availableActionsLock.RUnlock()
 
@@ -41,14 +41,14 @@ func triggerActions(c *irc.Client, m *irc.Message, rule *plugins.Rule, ra *plugi
 
 		if a.IsAsync() {
 			go func() {
-				if _, err := a.Execute(c, m, rule); err != nil {
+				if _, err := a.Execute(c, m, rule, eventData); err != nil {
 					logger.WithError(err).Error("Error in async actor")
 				}
 			}()
 			continue
 		}
 
-		apc, err := a.Execute(c, m, rule)
+		apc, err := a.Execute(c, m, rule, eventData)
 		preventCooldown = preventCooldown || apc
 		if err != nil {
 			return preventCooldown, errors.Wrap(err, "execute action")
@@ -58,12 +58,12 @@ func triggerActions(c *irc.Client, m *irc.Message, rule *plugins.Rule, ra *plugi
 	return preventCooldown, nil
 }
 
-func handleMessage(c *irc.Client, m *irc.Message, event *string) {
+func handleMessage(c *irc.Client, m *irc.Message, event *string, eventData map[string]interface{}) {
 	for _, r := range config.GetMatchingRules(m, event) {
 		var preventCooldown bool
 
 		for _, a := range r.Actions {
-			apc, err := triggerActions(c, m, r, a)
+			apc, err := triggerActions(c, m, r, a, eventData)
 			if err != nil {
 				log.WithError(err).Error("Unable to trigger action")
 			}

events.go

@@ -12,4 +12,9 @@ var (
 	eventTypeSub     = ptrStr("sub")
 	eventTypeSubgift = ptrStr("subgift")
 	eventTypeWhisper = ptrStr("whisper")
+
+	eventTypeTwitchCategoryUpdate = ptrStr("category_update")
+	eventTypeTwitchStreamOffline  = ptrStr("stream_offline")
+	eventTypeTwitchStreamOnline   = ptrStr("stream_online")
+	eventTypeTwitchTitleUpdate    = ptrStr("title_update")
 )

internal/actors/ban/actor.go

@@ -18,7 +18,7 @@ type actor struct {
 	Ban *string `json:"ban" yaml:"ban"`
 }
 
-func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.Ban == nil {
 		return false, nil
 	}

internal/actors/delay/actor.go

@@ -19,7 +19,7 @@ type actor struct {
 	DelayJitter time.Duration `json:"delay_jitter" yaml:"delay_jitter"`
 }
 
-func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.Delay == 0 && a.DelayJitter == 0 {
 		return false, nil
 	}

internal/actors/delete/actor.go

@@ -18,7 +18,7 @@ type actor struct {
 	DeleteMessage *bool `json:"delete_message" yaml:"delete_message"`
 }
 
-func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.DeleteMessage == nil || !*a.DeleteMessage {
 		return false, nil
 	}

internal/actors/raw/actor.go

@@ -20,12 +20,12 @@ type actor struct {
 	RawMessage *string `json:"raw_message" yaml:"raw_message"`
 }
 
-func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.RawMessage == nil {
 		return false, nil
 	}
 
-	rawMsg, err := formatMessage(*a.RawMessage, m, r, nil)
+	rawMsg, err := formatMessage(*a.RawMessage, m, r, eventData)
 	if err != nil {
 		return false, errors.Wrap(err, "preparing raw message")
 	}

internal/actors/respond/actor.go

@@ -22,17 +22,17 @@ type actor struct {
 	RespondFallback *string `json:"respond_fallback" yaml:"respond_fallback"`
 }
 
-func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.Respond == nil {
 		return false, nil
 	}
 
-	msg, err := formatMessage(*a.Respond, m, r, nil)
+	msg, err := formatMessage(*a.Respond, m, r, eventData)
 	if err != nil {
 		if a.RespondFallback == nil {
 			return false, errors.Wrap(err, "preparing response")
 		}
-		if msg, err = formatMessage(*a.RespondFallback, m, r, nil); err != nil {
+		if msg, err = formatMessage(*a.RespondFallback, m, r, eventData); err != nil {
 			return false, errors.Wrap(err, "preparing response fallback")
 		}
 	}

internal/actors/timeout/actor.go

@@ -19,7 +19,7 @@ type actor struct {
 	Timeout *time.Duration `json:"timeout" yaml:"timeout"`
 }
 
-func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.Timeout == nil {
 		return false, nil
 	}

internal/actors/whisper/actor.go

@@ -23,17 +23,17 @@ type actor struct {
 	WhisperTo      *string `json:"whisper_to" yaml:"whisper_to"`
 }
 
-func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule) (preventCooldown bool, err error) {
+func (a actor) Execute(c *irc.Client, m *irc.Message, r *plugins.Rule, eventData map[string]interface{}) (preventCooldown bool, err error) {
 	if a.WhisperTo == nil || a.WhisperMessage == nil {
 		return false, nil
 	}
 
-	to, err := formatMessage(*a.WhisperTo, m, r, nil)
+	to, err := formatMessage(*a.WhisperTo, m, r, eventData)
 	if err != nil {
 		return false, errors.Wrap(err, "preparing whisper receiver")
 	}
 
-	msg, err := formatMessage(*a.WhisperMessage, m, r, nil)
+	msg, err := formatMessage(*a.WhisperMessage, m, r, eventData)
 	if err != nil {
 		return false, errors.Wrap(err, "preparing whisper message")
 	}

irc.go

@@ -174,11 +174,11 @@ func (ircHandler) getChannel(m *irc.Message) string {
 }
 
 func (i ircHandler) handleJoin(m *irc.Message) {
-	go handleMessage(i.c, m, eventTypeJoin)
+	go handleMessage(i.c, m, eventTypeJoin, nil)
 }
 
 func (i ircHandler) handlePart(m *irc.Message) {
-	go handleMessage(i.c, m, eventTypePart)
+	go handleMessage(i.c, m, eventTypePart, nil)
 }
 
 func (i ircHandler) handlePermit(m *irc.Message) {
@@ -198,7 +198,7 @@ func (i ircHandler) handlePermit(m *irc.Message) {
 	log.WithField("user", username).Debug("Added permit")
 	timerStore.AddPermit(m.Params[0], username)
 
-	go handleMessage(i.c, m, eventTypePermit)
+	go handleMessage(i.c, m, eventTypePermit, nil)
 }
 
 func (i ircHandler) handleTwitchNotice(m *irc.Message) {
@@ -216,7 +216,7 @@ func (i ircHandler) handleTwitchNotice(m *irc.Message) {
 	case "host_success", "host_success_viewers":
 		log.WithField("trailing", m.Trailing()).Warn("Incoming host")
 
-		go handleMessage(i.c, m, eventTypeHost)
+		go handleMessage(i.c, m, eventTypeHost, nil)
 
 	}
 }
@@ -244,7 +244,7 @@ func (i ircHandler) handleTwitchPrivmsg(m *irc.Message) {
 		return
 	}
 
-	go handleMessage(i.c, m, nil)
+	go handleMessage(i.c, m, nil, nil)
 }
 
 func (i ircHandler) handleTwitchUsernotice(m *irc.Message) {
@@ -265,20 +265,20 @@ func (i ircHandler) handleTwitchUsernotice(m *irc.Message) {
 			"viewercount": m.Tags["msg-param-viewerCount"],
 		}).Info("Incoming raid")
 
-		go handleMessage(i.c, m, eventTypeRaid)
+		go handleMessage(i.c, m, eventTypeRaid, nil)
 
 	case "resub":
-		go handleMessage(i.c, m, eventTypeResub)
+		go handleMessage(i.c, m, eventTypeResub, nil)
 
 	case "sub":
-		go handleMessage(i.c, m, eventTypeSub)
+		go handleMessage(i.c, m, eventTypeSub, nil)
 
 	case "subgift", "anonsubgift":
-		go handleMessage(i.c, m, eventTypeSubgift)
+		go handleMessage(i.c, m, eventTypeSubgift, nil)
 
 	}
 }
 
 func (i ircHandler) handleTwitchWhisper(m *irc.Message) {
-	go handleMessage(i.c, m, eventTypeWhisper)
+	go handleMessage(i.c, m, eventTypeWhisper, nil)
 }

main.go

@@ -82,6 +82,9 @@ func main() {
 	cronService = cron.New()
 	twitchClient = twitch.New(cfg.TwitchClient, cfg.TwitchToken)
 
+	twitchWatch := newTwitchWatcher()
+	cronService.AddFunc("* * * * *", twitchWatch.Check)
+
 	router.HandleFunc("/", handleSwaggerHTML)
 	router.HandleFunc("/openapi.json", handleSwaggerRequest)
 
@@ -170,11 +173,20 @@ func main() {
 			}
 
 			irc.ExecuteJoins(config.Channels)
+			for _, c := range config.Channels {
+				if err := twitchWatch.AddChannel(c); err != nil {
+					log.WithError(err).WithField("channel", c).Error("Unable to add channel to watcher")
+				}
+			}
 
 			for _, c := range previousChannels {
 				if !str.StringInSlice(c, config.Channels) {
 					log.WithField("channel", c).Info("Leaving removed channel...")
 					irc.ExecutePart(c)
+
+					if err := twitchWatch.RemoveChannel(c); err != nil {
+						log.WithError(err).WithField("channel", c).Error("Unable to remove channel from watcher")
+					}
 				}
 			}
 

plugins/interface.go

@@ -9,7 +9,7 @@ import (
 type (
 	Actor interface {
 		// Execute will be called after the config was read into the Actor
-		Execute(*irc.Client, *irc.Message, *Rule) (preventCooldown bool, err error)
+		Execute(*irc.Client, *irc.Message, *Rule, map[string]interface{}) (preventCooldown bool, err error)
 		// IsAsync may return true if the Execute function is to be executed
 		// in a Go routine as of long runtime. Normally it should return false
 		// except in very specific cases

twitchWatcher.go

@@ -0,0 +1,116 @@
+package main
+
+import (
+	"sync"
+
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+type (
+	twitchChannelState struct {
+		Category string
+		IsLive   bool
+		Title    string
+	}
+
+	twitchWatcher struct {
+		ChannelStatus map[string]*twitchChannelState
+
+		lock sync.RWMutex
+	}
+)
+
+func (t twitchChannelState) Equals(c twitchChannelState) bool {
+	return t.Category == c.Category &&
+		t.IsLive == c.IsLive &&
+		t.Title == c.Title
+}
+
+func newTwitchWatcher() *twitchWatcher {
+	return &twitchWatcher{
+		ChannelStatus: make(map[string]*twitchChannelState),
+	}
+}
+
+func (r *twitchWatcher) AddChannel(channel string) error {
+	r.lock.RLock()
+	if _, ok := r.ChannelStatus[channel]; ok {
+		r.lock.RUnlock()
+		return nil
+	}
+	r.lock.RUnlock()
+
+	return r.updateChannelFromAPI(channel, false)
+}
+
+func (r *twitchWatcher) Check() {
+	var channels []string
+	r.lock.RLock()
+	for c := range r.ChannelStatus {
+		channels = append(channels, c)
+	}
+	r.lock.RUnlock()
+
+	for _, ch := range channels {
+		if err := r.updateChannelFromAPI(ch, true); err != nil {
+			log.WithError(err).WithField("channel", ch).Error("Unable to update channel status")
+		}
+	}
+}
+
+func (r *twitchWatcher) RemoveChannel(channel string) error {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	delete(r.ChannelStatus, channel)
+	return nil
+}
+
+func (r *twitchWatcher) updateChannelFromAPI(channel string, sendUpdate bool) error {
+	var (
+		err    error
+		status twitchChannelState
+	)
+
+	status.IsLive, err = twitchClient.HasLiveStream(channel)
+	if err != nil {
+		return errors.Wrap(err, "getting live status")
+	}
+
+	status.Category, status.Title, err = twitchClient.GetRecentStreamInfo(channel)
+	if err != nil {
+		return errors.Wrap(err, "getting stream info")
+	}
+
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	if r.ChannelStatus[channel] != nil && r.ChannelStatus[channel].Equals(status) {
+		return nil
+	}
+
+	if r.ChannelStatus[channel].Category != status.Category {
+		go handleMessage(nil, nil, eventTypeTwitchCategoryUpdate, map[string]interface{}{
+			"category": status.Category,
+		})
+	}
+
+	if r.ChannelStatus[channel].Title != status.Title {
+		go handleMessage(nil, nil, eventTypeTwitchTitleUpdate, map[string]interface{}{
+			"title": status.Title,
+		})
+	}
+
+	if r.ChannelStatus[channel].IsLive != status.IsLive {
+		evt := eventTypeTwitchStreamOnline
+		if !status.IsLive {
+			evt = eventTypeTwitchStreamOffline
+		}
+
+		go handleMessage(nil, nil, evt, nil)
+	}
+
+	r.ChannelStatus[channel] = &status
+	return nil
+}

wiki/Home.md

@@ -128,7 +128,8 @@ rules: # See below for examples
     match_users: ['mychannel'] # List of users, all names MUST be all lower-case
 
     # Execute actions when this event occurs
-    # Available events: join, host, part, permit, raid, resub, sub, subgift, whisper
+    # Available events: category_update, join, host, part, permit, raid, resub,
+    #                   stream_offline, stream_online, sub, subgift, title_update, whisper
     match_event: 'permit'
 
     # Execute action when the chat message matches this regular expression