Add payment, new claim notifications

config/config.go

@@ -7,6 +7,8 @@ import (
 	"runtime"
 	"time"
 
+	"github.com/lbryio/chainquery/notifications"
+
 	"github.com/lbryio/chainquery/apiactions"
 	"github.com/lbryio/chainquery/auth"
 	"github.com/lbryio/chainquery/daemon"
@@ -126,6 +128,9 @@ func readConfig() {
 		logrus.Warning("Error reading config file...defaults will be used: ", err)
 	}
 	twilio.RecipientList = viper.GetStringSlice(smsrecipients)
+	notifications.ClearSubscribers()
+	subscriptions := viper.GetStringMap("subscription")
+	applySubscribers(subscriptions)
 	twilio.FromNumber = viper.GetString(smsfromphonenumber)
 	twilio.TwilioAuthToken = viper.GetString(twilioauthtoken)
 	twilio.TwilioSID = viper.GetString(twiliosid)
@@ -245,3 +250,28 @@ func getLbrycrdURLFromConfFile() (string, error) {
 
 	return "rpc://" + userpass + host + port, nil
 }
+
+func applySubscribers(subs map[string]interface{}) error {
+	for subType, p := range subs {
+		typeSubsInt, ok := p.([]interface{})
+		if ok {
+			for _, typeSub := range typeSubsInt {
+				params, ok := typeSub.(map[string]interface{})
+				if ok {
+					url, ok := params["url"].(string)
+					if ok {
+						delete(params, "url")
+						notifications.AddSubscriber(url, subType, params)
+					} else {
+						return errors.Err("url is required")
+					}
+				} else {
+					return errors.Err("could not find params map for the subscription type instance")
+				}
+			}
+		} else {
+			return errors.Err("could not find sub type array under subscription")
+		}
+	}
+	return nil
+}

config/default/chainqueryconfig.toml

@@ -123,4 +123,16 @@
 
 #Max Parallel Vout Processing - Specifies the maximum number of worker go routines created for processing outputs in a transaction.
 #DEFAULT: NumCPU
-#maxparallelvoutprocessing=
+#maxparallelvoutprocessing=
+
+#Subscribers - Lists the subscriptions for notifications. Possible types "payment" "new_claim"
+#DEFAULT: <none>
+#[[subscriber.payment]]
+#  url= "http://localhost:8080/event/payment"
+#  auth_token="mytoken"
+#[[subscriber.payment]]
+#  url= "http://localhost:8080/event/payment"
+#  auth_token="mytoken"
+#[[subscriber.newclaim]]
+#  url= "http://localhost:8080/event/claim"
+#  auth_token="mytoken"

daemon/processing/claim.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/lbryio/chainquery/notifications"
+
 	"github.com/lbryio/lbryschema.go/address/base58"
 	pb "github.com/lbryio/types/v2/go"
 
@@ -191,7 +193,7 @@ func processClaim(helper *c.ClaimHelper, claim *model.Claim, value []byte, outpu
 	if helper.LegacyClaim != nil {
 		claim.Version = helper.LegacyClaim.GetVersion().String()
 	}
-
+	notifications.ClaimEvent(claimID, claim.Name, claim.Title, tx.Hash, claim.PublisherID.String)
 	return claim, nil
 }
 

daemon/processing/outpoint.go

@@ -7,6 +7,8 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/lbryio/chainquery/notifications"
+
 	ds "github.com/lbryio/chainquery/datastore"
 	"github.com/lbryio/chainquery/lbrycrd"
 	m "github.com/lbryio/chainquery/model"
@@ -215,10 +217,13 @@ func processVout(jsonVout *lbrycrd.Vout, tx *m.Transaction, txDC *txDebitCredits
 	}
 
 	//Make sure there is a transaction address
-	if ds.GetTxAddress(tx.ID, address.ID) == nil {
+	txAddress := ds.GetTxAddress(tx.ID, address.ID)
+	if txAddress == nil {
 		return errors.Base("Missing txAddress for Tx:" + strconv.Itoa(int(tx.ID)) + "- Addr:" + strconv.Itoa(int(address.ID)))
 	}
 
+	notifications.PaymentEvent(vout.Value.Float64, address.Address, tx.Hash, vout.Vout)
+
 	// Process script for potential claims
 	claimid, err := processScriptForClaim(*vout, *tx, blockHeight)
 	if err != nil {

go.mod

@@ -40,6 +40,7 @@ require (
 	github.com/shopspring/decimal v0.0.0-20191009025716-f1972eb1d1f5
 	github.com/sirupsen/logrus v1.4.2
 	github.com/spf13/afero v1.1.1 // indirect
+	github.com/spf13/cast v1.3.0
 	github.com/spf13/cobra v0.0.3
 	github.com/spf13/jwalterweatherman v0.0.0-20180109140146-7c0cea34c8ec // indirect
 	github.com/spf13/pflag v1.0.1

go.sum

@@ -105,6 +105,7 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/lbryio/errors.go v0.0.0-20180223142025-ad03d3cc6a5c h1:BhdcWGsuKif/XoSZnqVGNqJ1iEmH0czWR5upj+AuR8M=
 github.com/lbryio/errors.go v0.0.0-20180223142025-ad03d3cc6a5c/go.mod h1:muH7wpUqE8hRA3OrYYosw9+Sl681BF9cwcjzE+OCNK8=
 github.com/lbryio/lbry.go v1.1.2 h1:Dyxc+glT/rVWJwHfIf7vjlPYYbjzrQz5ARmJd5Hp69c=
 github.com/lbryio/lbry.go v1.1.2/go.mod h1:JtyI30bU51rm0LZ/po3mQuzf++14OWb6kR/6mMRAmKU=

notifications/events.go

@@ -0,0 +1,29 @@
+package notifications
+
+import (
+	"net/url"
+
+	"github.com/spf13/cast"
+)
+
+const payment = "payment"
+const newClaim = "new_claim"
+
+func PaymentEvent(lbc float64, address, txid string, vout uint) {
+	values := url.Values{}
+	values.Add("lbc", cast.ToString(lbc))
+	values.Add("tx_id", txid)
+	values.Add("vout", cast.ToString(vout))
+	values.Add("address", address)
+	go Notify(payment, values)
+}
+
+func ClaimEvent(claimID, name, title, txID, channeClaimID string) {
+	values := url.Values{}
+	values.Add("claim_id", claimID)
+	values.Add("name", name)
+	values.Add("title", title)
+	values.Add("tx_id", txID)
+	values.Add("channel_claim_id", channeClaimID)
+	go Notify(newClaim, values)
+}

notifications/notification_test.go

@@ -0,0 +1,26 @@
+package notifications
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/spf13/viper"
+)
+
+func TestAddSubscriber(t *testing.T) {
+	viper.SetConfigType("toml")
+	err := viper.ReadConfig(strings.NewReader(`
+[[subscription.payment]]
+url= "http://localhost:8080/event/payment"
+auth_token="mytoken"
+[[subscription.payment]]
+url= "http://localhost:8080/event/payment"
+auth_token="mytoken"
+`))
+	if err != nil {
+		t.Error(err)
+	}
+	subs := viper.GetStringMap("subscription")
+	applySubscribers(subs)
+
+}

notifications/notifications.go

@@ -0,0 +1,55 @@
+package notifications
+
+import (
+	"net/http"
+	"net/url"
+
+	"github.com/lbryio/lbry.go/v2/extras/errors"
+
+	"github.com/sirupsen/logrus"
+)
+
+type subscriber struct {
+	URL    string
+	Type   string
+	Params url.Values
+}
+
+var subscriptions map[string][]subscriber
+
+func AddSubscriber(address, subType string, params map[string]interface{}) {
+	if subscriptions == nil {
+		subscriptions = make(map[string][]subscriber)
+	}
+	urlParams := url.Values{}
+	for param, v := range params {
+		value, ok := v.(string)
+		if ok {
+			urlParams.Set(param, value)
+		}
+	}
+	subscriptions[subType] = append(subscriptions[subType], subscriber{URL: address, Type: subType, Params: urlParams})
+}
+
+func ClearSubscribers() {
+	subscriptions = make(map[string][]subscriber)
+}
+
+func Notify(t string, values url.Values) {
+	subs, ok := subscriptions[t]
+	if ok {
+		for _, s := range subs {
+			for param, value := range s.Params {
+				values.Set(param, value[0])
+			}
+			s.notify(values)
+		}
+	}
+}
+
+func (s subscriber) notify(values url.Values) {
+	_, err := http.PostForm(s.URL, values)
+	if err != nil {
+		logrus.Error(errors.Prefix("Notify:", errors.Err(err)))
+	}
+}