Receive loyalty pushes from stulbe

go.mod

@@ -11,5 +11,6 @@ require (
 	github.com/pkg/browser v0.0.0-20210115035449-ce105d075bb4
 	github.com/sirupsen/logrus v1.8.1
 	github.com/strimertul/kilovolt/v4 v4.0.1
+	github.com/strimertul/stulbe v0.4.2
 	github.com/strimertul/stulbe-client-go v0.2.0
 )

go.sum

@@ -112,8 +112,9 @@ github.com/strimertul/kilovolt/v4 v4.0.1 h1:81isohdSixVURO2+dZKKZBPw97HJmNN4/BXn
 github.com/strimertul/kilovolt/v4 v4.0.1/go.mod h1:AO2ZFQtSB+AcjCw0RTkXjbM6XBAjhsXsrRq10BX95kw=
 github.com/strimertul/strimertul v1.3.0/go.mod h1:1pSe9zVWF4BYt56ii1Hg+xTxvtfqfvT4FQ7bYffWsUA=
 github.com/strimertul/stulbe v0.2.5/go.mod h1:0AsY4OVf1dNCwOn9s7KySuAxJ85w88pXeostu1n9E7w=
-github.com/strimertul/stulbe v0.3.0 h1:RCELIetAMW+xopWArGA0dwk0ZHNPAPkkS08AyXpdVRg=
 github.com/strimertul/stulbe v0.3.0/go.mod h1:Pb0UQCKdyES7UKSKm2i2g9parkgXSJAFeMH/LSOSbgQ=
+github.com/strimertul/stulbe v0.4.2 h1:dxxjGtLiUEyMu/V7Rc7kurcY6DHS7jXS8/GEfQPSFR4=
+github.com/strimertul/stulbe v0.4.2/go.mod h1:Pb0UQCKdyES7UKSKm2i2g9parkgXSJAFeMH/LSOSbgQ=
 github.com/strimertul/stulbe-client-go v0.1.0/go.mod h1:KtfuDhxCHZ9DCFHnrBOHqb2Pu9zoj+EqA8ZRIUqLD/w=
 github.com/strimertul/stulbe-client-go v0.2.0 h1:zoUu+/ssOWz9VsoioJkQNMU88qJ/lQzAu6ak8GcTYxY=
 github.com/strimertul/stulbe-client-go v0.2.0/go.mod h1:5S+nDMyzkeoBHi9fq3s37A4mv3XUPJhwgW94UtGzp28=

main.go

@@ -123,12 +123,17 @@ func main() {
 	// Get Stulbe config, if enabled
 	var stulbeManager *stulbe.Manager = nil
 	if moduleConfig.EnableStulbe {
-		stulbeManager, err = stulbe.Initialize(db, wrapLogger("stulbe"))
+		stulbeLogger := wrapLogger("stulbe")
+		stulbeManager, err = stulbe.Initialize(db, stulbeLogger)
 		if err != nil {
 			log.WithError(err).Error("Stulbe initialization failed! Module was temporarely disabled")
 			moduleConfig.EnableStulbe = false
 		}
 		defer stulbeManager.Close()
+		go func() {
+			err := stulbeManager.ReceiveEvents()
+			stulbeLogger.WithError(err).Error("Stulbe subscription died unexpectedly!")
+		}()
 	}
 
 	var loyaltyManager *loyalty.Manager

modules/loyalty/data.go

@@ -58,6 +58,7 @@ type Redeem struct {
 	DisplayName string    `json:"display_name"`
 	Reward      Reward    `json:"reward"`
 	When        time.Time `json:"when"`
+	RequestText string    `json:"request_text"`
 }
 
 const CreateRedeemRPC = "loyalty/@create-redeem"

modules/loyalty/manager.go

@@ -5,10 +5,12 @@ import (
 	"errors"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/dgraph-io/badger/v3"
 	jsoniter "github.com/json-iterator/go"
 	"github.com/sirupsen/logrus"
+	"github.com/strimertul/stulbe/api"
 
 	"github.com/strimertul/strimertul/database"
 )
@@ -90,9 +92,8 @@ func NewManager(db *database.DB, log logrus.FieldLogger) (*Manager, error) {
 	}
 
 	// Subscribe for changes
-	go func() {
-		db.Subscribe(context.Background(), manager.update, "loyalty/")
-	}()
+	go db.Subscribe(context.Background(), manager.update, "loyalty/")
+	go db.Subscribe(context.Background(), manager.handleRemote, "stulbe/loyalty/")
 
 	return manager, nil
 }
@@ -157,6 +158,39 @@ func (m *Manager) update(kvs []database.ModifiedKV) error {
 	return nil
 }
 
+func (m *Manager) handleRemote(kvs []database.ModifiedKV) error {
+	for _, kv := range kvs {
+		m.logger.WithField("key", kv.Key).Trace("loyalty request from stulbe")
+		switch kv.Key {
+		case api.KVExLoyaltyRedeem:
+			// Parse request
+			var redeemRequest api.ExLoyaltyRedeem
+			err := jsoniter.ConfigFastest.Unmarshal(kv.Data, &redeemRequest)
+			if err != nil {
+				m.logger.WithError(err).Warn("error decoding redeem request")
+				break
+			}
+			// Find reward
+			reward := m.getReward(redeemRequest.RewardID)
+			if reward.ID == "" {
+				m.logger.WithField("reward-id", redeemRequest.RewardID).Warn("redeem request contains invalid reward id")
+				break
+			}
+			err = m.PerformRedeem(Redeem{
+				Username:    redeemRequest.Username,
+				DisplayName: redeemRequest.DisplayName,
+				Reward:      reward,
+				When:        time.Now(),
+				RequestText: redeemRequest.RequestText,
+			})
+			if err != nil {
+				m.logger.WithError(err).Warn("error performing redeem request")
+			}
+		}
+	}
+	return nil
+}
+
 func (m *Manager) GetPoints(user string) int64 {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -218,6 +252,17 @@ func (m *Manager) AddRedeem(redeem Redeem) error {
 	return m.saveQueue()
 }
 
+func (m *Manager) PerformRedeem(redeem Redeem) error {
+	// Add redeem
+	err := m.AddRedeem(redeem)
+	if err != nil {
+		return err
+	}
+
+	// Remove points from user
+	return m.TakePoints(map[string]int64{redeem.Username: redeem.Reward.Price})
+}
+
 func (m *Manager) RemoveRedeem(redeem Redeem) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -266,6 +311,17 @@ func (m *Manager) Rewards() []Reward {
 	return m.rewards[:]
 }
 
+func (m *Manager) getReward(id string) Reward {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, reward := range m.rewards {
+		if reward.ID == id {
+			return reward
+		}
+	}
+	return Reward{}
+}
+
 func (m *Manager) Config() Config {
 	m.mu.Lock()
 	defer m.mu.Unlock()

modules/stulbe/client.go

@@ -40,6 +40,20 @@ func Initialize(db *database.DB, logger logrus.FieldLogger) (*Manager, error) {
 	}, err
 }
 
+func (m *Manager) ReceiveEvents() error {
+	chn, err := m.Client.KV.SubscribePrefix("stulbe/")
+	if err != nil {
+		return err
+	}
+	for {
+		kv := <-chn
+		err := m.db.PutKey(kv.Key, []byte(kv.Value))
+		if err != nil {
+			return err
+		}
+	}
+}
+
 func (m *Manager) Close() {
 	m.Client.Close()
 }
@@ -51,11 +65,9 @@ func (m *Manager) ReplicateKey(prefix string) error {
 		return err
 	}
 
-	for k, v := range vals {
-		err = m.Client.KV.SetKey(k, v)
-		if err != nil {
-			return err
-		}
+	err = m.Client.KV.SetKeys(vals)
+	if err != nil {
+		return err
 	}
 
 	m.logger.WithFields(logrus.Fields{

modules/twitch/commands.go

@@ -90,19 +90,13 @@ func cmdRedeemReward(bot *Bot, message irc.PrivateMessage) {
 		}
 
 		// Perform redeem
-		if err := bot.Loyalty.AddRedeem(loyalty.Redeem{
+		if err := bot.Loyalty.PerformRedeem(loyalty.Redeem{
 			Username:    message.User.Name,
 			DisplayName: message.User.DisplayName,
 			When:        time.Now(),
 			Reward:      reward,
 		}); err != nil {
-			bot.logger.WithError(err).Error("error while adding redeem")
-			return
-		}
-
-		// Remove points from user
-		if err := bot.Loyalty.TakePoints(map[string]int64{message.User.Name: reward.Price}); err != nil {
-			bot.logger.WithError(err).Error("error while taking points for redeem")
+			bot.logger.WithError(err).Error("error while performing redeem")
 			return
 		}