add grpc client

dakma-dev · Sep 2, 2023 · d6b0da5 · d6b0da5
1 parent 0596ca9
commit d6b0da5
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 12 deletions.
diff --git a/internal/seawa/seawatcher.go b/internal/seawa/seawatcher.go
@@ -15,6 +15,7 @@ import (
 	"github.com/benleb/gloomberg/internal/gbl"
 	"github.com/benleb/gloomberg/internal/nemo/gloomberg"
 	gbgrpc "github.com/benleb/gloomberg/internal/nemo/gloomberg/gbgrpc/gen"
+	"github.com/benleb/gloomberg/internal/nemo/gloomberg/remote"
 	"github.com/benleb/gloomberg/internal/nemo/osmodels"
 	"github.com/benleb/gloomberg/internal/seawa/models"
 	"github.com/benleb/gloomberg/internal/style"
@@ -27,6 +28,7 @@ import (
 	"github.com/redis/rueidis"
 	"github.com/spf13/viper"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
 	emptypb "google.golang.org/protobuf/types/known/emptypb"
 )
 
@@ -391,8 +393,29 @@ func (sw *SeaWatcher) SubscribeForSlug(slug string, eventTypes []gbgrpc.EventTyp
 	return sw.SubscribeForSlugs([]string{slug}, eventTypes)
 }
 
-func (sw *SeaWatcher) SubscribeForSlugs(slug []string, eventTypes []gbgrpc.EventType) uint64 {
-	if !viper.GetBool("seawatcher.local") {
+func (sw *SeaWatcher) SubscribeForSlugs(slugs []string, eventTypes []gbgrpc.EventType) uint64 {
+	if viper.GetBool("grpc.client.enabled") {
+		// connect
+		grpcClient := remote.NewClient()
+
+		if grpcClient == nil {
+			log.Errorf("fail to connect to gRPC to subscribe for %+v", style.BoldAlmostWhite(strings.Join(slugs, ", ")))
+
+			return 0
+		}
+
+		sw.Prf("subscribing to %s...", strings.Join(slugs, ", "))
+
+		subsriptionRequest := &gbgrpc.SubscriptionRequest{EventTypes: []gbgrpc.EventType{gbgrpc.EventType_ITEM_LISTED}, Collections: slugs} //nolint:nosnakecase
+		_, err := grpcClient.Subscribe(context.Background(), subsriptionRequest, grpc.WaitForReady(true))
+		if err != nil {
+			log.Errorf("getting stream failed: %v, retrying", err)
+		}
+
+		// sw.Prf("subscribing to %s...", strings.Join(slugs, ", "))
+
+		return uint64(len(slugs))
+	} else if !viper.GetBool("seawatcher.local") {
 		log.Warn("⚓️ subscribe discarded - no local OpenSea clients")
 		log.Warn("⚓️ TODO implement subscribe via grpc (and maybe pubsub)")
 
@@ -401,7 +424,7 @@ func (sw *SeaWatcher) SubscribeForSlugs(slug []string, eventTypes []gbgrpc.Event
 
 	newEventSubscriptions := uint64(0)
 
-	for _, slug := range slug {
+	for _, slug := range slugs {
 		if sw.IsSubscribed(slug) {
 			log.Debugf("⚓️ ☕️ already subscribed to OpenSea events for %s", slug)
 

diff --git a/internal/trapri/trapri.go b/internal/trapri/trapri.go
@@ -15,6 +15,7 @@ import (
 	"github.com/benleb/gloomberg/internal/gbl"
 	"github.com/benleb/gloomberg/internal/jobs"
 	"github.com/benleb/gloomberg/internal/nemo/gloomberg"
+	"github.com/benleb/gloomberg/internal/nemo/gloomberg/gbgrpc"
 	"github.com/benleb/gloomberg/internal/nemo/gloomberg/gbgrpc/gen"
 	"github.com/benleb/gloomberg/internal/nemo/price"
 	"github.com/benleb/gloomberg/internal/nemo/standard"
@@ -198,15 +199,17 @@ func formatTokenTransaction(gb *gloomberg.Gloomberg, seawa *seawatcher.SeaWatche
 			continue
 		}
 
-		// fetch & store the first 1k txs for this contract
-		collectionFileNamePrefix := collection.Name
-		if collection.OpenseaSlug != "" {
-			collectionFileNamePrefix = collection.OpenseaSlug
-		}
+		//
+		// fetch & store the first txs for this contract
 
-		if viper.GetBool("experiments.firsttxs") && collectionFileNamePrefix != "" && ttx.GetPrice().Ether() >= viper.GetFloat64("gloomberg.firstTxs.min_value") {
-			jobs.AddJob("firsttxs", "etherscan", gloomberg.JobFirstTxsForContract, collectionFileNamePrefix, contractAddress)
-		}
+		// collectionFileNamePrefix := collection.Name
+		// if collection.OpenseaSlug != "" {
+		// 	collectionFileNamePrefix = collection.OpenseaSlug
+		// }
+
+		// if viper.GetBool("experiments.firsttxs") && collectionFileNamePrefix != "" && ttx.GetPrice().Ether() >= viper.GetFloat64("gloomberg.firstTxs.min_value") {
+		// 	jobs.AddJob("firsttxs", "etherscan", gloomberg.JobFirstTxsForContract, collectionFileNamePrefix, contractAddress)
+		// }
 
 		ttxCollections[contractAddress] = collection
 
@@ -795,7 +798,7 @@ func formatTokenTransaction(gb *gloomberg.Gloomberg, seawa *seawatcher.SeaWatche
 			return
 		}
 
-		if !currentCollection.Show.Mints && (ttx.Action == degendb.Mint || ttx.Action == degendb.Airdrop) && !viper.GetBool("show.mints") {
+		if !isOwnCollection && !currentCollection.Show.Mints && (ttx.Action == degendb.Mint || ttx.Action == degendb.Airdrop) && !viper.GetBool("show.mints") {
 			log.Debugf("skipping mint %s | viper.GetBool(show.mints): %v | %+v", style.Bold(txHash.String()), viper.GetBool("show.mints"), ttx)
 
 			return
@@ -850,6 +853,19 @@ func formatTokenTransaction(gb *gloomberg.Gloomberg, seawa *seawatcher.SeaWatche
 
 				if !seawa.IsSubscribed(currentCollection.OpenseaSlug) {
 					// if seawa.SubscribeForSlug(currentCollection.OpenseaSlug) {
+					// gprcClient := gbgrpc.NewClient("")
+
+					if gbgrpc.GRPCClient != nil {
+						subsriptionRequest := &gen.SubscriptionRequest{EventTypes: []gen.EventType{gen.EventType_ITEM_LISTED}, Collections: []string{currentCollection.OpenseaSlug}} //nolint:nosnakecase
+
+						_, err := gbgrpc.GRPCClient.Subscribe(context.Background(), subsriptionRequest, nil)
+						if err != nil {
+							log.Errorf("failed to subscribe to events for %s: %s", currentCollection.OpenseaSlug, err)
+						}
+
+						log.Printf("subsubsubbbyyyyyyyyyyyyyyy: %+v", subsriptionRequest)
+					}
+
 					if seawa.SubscribeForSlug(currentCollection.OpenseaSlug, []gen.EventType{gen.EventType_ITEM_LISTED, gen.EventType_ITEM_RECEIVED_BID, gen.EventType_COLLECTION_OFFER}) > 0 { //nolint:nosnakecase
 						seawa.Pr(fmt.Sprintf("auto-subscribed to events for %s (after %d sales) | stats resetted", style.AlmostWhiteStyle.Render(currentCollection.OpenseaSlug), autoSubscribeAfterSales))
 					}