Benny/opensea streaming #1343

benny-conn · 2024-01-03T21:43:28Z

Changes:

Added a new singleton service called opensea-streamer that listens with a websocket to all token transfers and sends them to the token-processing service
Added a handler on the token processing service for the webhook
Added some non-blocking and unrelated email digest adjustments

opensea-streamer/main.go

…llery into benny/opensea-streaming

radazen

Looking good! The email digest stuff looks A-OK to me. For OpenSea streaming, I left a few comments about how we store and update the bloom filter, but everything else looks great!

radazen · 2024-01-10T17:04:22Z

emails/digest.go

@@ -218,6 +202,13 @@ func getDigest(c context.Context, stg *storage.Client, f *publicapi.FeedAPI, q *
 		includeCommunities = *overrides.IncludeTopCommunities
 	}

+	topPostCount, _ := util.FindFirst([]int{overrides.FirstPostCount, defaultFirstPostCount, 5}, func(i int) bool {


Are these top posts or first posts? There seems to be a mix of terminology.

radazen · 2024-01-10T17:04:41Z

emails/digest.go

@@ -227,11 +218,44 @@ func getDigest(c context.Context, stg *storage.Client, f *publicapi.FeedAPI, q *
 			Selected: selectedCollections,
 			Include:  includeCommunities,
 		},
+		FirstPostCount: topPostCount,


Same as above, re: top posts vs first posts

radazen · 2024-01-10T17:06:37Z

emails/digest.go

 	}
 	return result, nil
 }

-func contractToUserFacing(ctx context.Context, q *coredb.Queries, l *dataloader.Loaders, collection coredb.Contract) UserFacingContract {
+func getOverrides(c context.Context, stg *storage.Client) (DigestValueOverrides, error) {
+	_, err := stg.Bucket(env.GetString("CONFIGURATION_BUCKET")).Object(overrideFile).Attrs(c)


Can we make a variable for stg.Bucket(env.GetString("CONFIGURATION_BUCKET")).Object(overrideFile) since we're using it several times in this function?

radazen · 2024-01-10T17:11:46Z

emails/digest.go

+	for _, p := range second.TopPosts {
+		seenPostPositions[p.Position] = true
+	}
+	for _, p := range second.TopCommunities {


Do we need to do this loop for TopGalleries and TopFirstPosts, too?

radazen · 2024-01-10T17:18:47Z

emails/send.go

@@ -357,7 +359,7 @@ func sendDigestEmailsToAllUsers(c context.Context, v DigestValues, queries *core
 	})
 }

-func sendDigestEmailToUser(c context.Context, u coredb.PiiUserView, emailRecipient persist.Email, digestValues DigestValues, s *sendgrid.Client, sendRealEmail bool) (*rest.Response, error) {


What's the net effect of removing the sendRealEmail parameter? For example, if we get here on dev, would we send emails to all the dummy PII-scrubbed email addresses we have in the database?

I realized it was unnecessary because on anything but prod we only send emails to Admin verified email accounts.

radazen · 2024-01-10T17:38:23Z

opensea-streamer/main.go

+			_, message, err := conn.ReadMessage()
+			if err != nil {
+				logger.For(ctx).Error(err)
+				break


Assume we error here and this break is executed. What happens? Would it be the case that the service keeps running (because of the ticker loop below), but we wouldn't be listening for messages anymore? Seems like we'd want to log the error and restart the connection or something along those lines.

radazen · 2024-01-10T17:39:19Z

opensea-streamer/main.go

+
+			var oe openseaEvent
+			err = json.Unmarshal(message, &oe)
+			if err != nil {


If there's an unmarshaling error, I feel like we should log it so we know what happened

This one is an interesting one because with the JS SDK we are able to filter down to specific events but because we are doing this in go we get all incoming events from every type of event. Because of this, most of them error out here simply because they are the wrong event type and don't match the JSON schema we defined. It will get really spammy if we log this.

radazen · 2024-01-10T19:07:05Z

opensea-streamer/main.go

+		c.String(http.StatusOK, "OK")
+	})
+
+	router.GET("/updateBloomFilter", func(c *gin.Context) {


Is this used by anything, or is it just for debugging?

Just used for debugging for now but I imagine can be used when a user signs up if we want to add that functionality later?

radazen · 2024-01-10T19:24:06Z

opensea-streamer/main.go

+				continue
+			}
+
+			bf, err = resetBloomFilter(ctx, queries, bloomCache)


Updating bf here won't change the reference being used by streamOpenseaTransfers. It'll change what the outer bf variable in main() points to, but it's a pointer -- not a pointer to a pointer -- so it isn't going to change what streamOpenseaTransfers sees.

Assuming it's quick to update the bloom filter -- and I think that's true, right? -- we probably don't need to use Redis at all. I was thinking we'd want to use it because it has its own bloom filter implementation and we wouldn't need to store everything in memory on a server if we were using Redis. But given that we're actually storing everything in memory here, and given that rebuilding the filter is probably pretty quick, I don't see much benefit to using Redis. At best, it saves us a small amount of time at startup, but only if there hasn't been a change in the number of active wallets we're tracking.

Instead, let's get rid of Redis entirely. We can use an atomic.Pointer[bloomFilter] to track the active bloom filter, and when we update the bloom filter, use the atomic pointer's Store function to point to the newly created filter. In your processing loop, you'd just want to load the current reference every time you're going to use the bloom filter. There's some overhead, but using atomic is generally much more efficient than using locks, and I think for our purposes it'll be okay. Something like:

// check if the wallet is in the bloom filter chainAddress, err := persist.NewL1ChainAddress(persist.Address(win.Payload.ToAccount.Address.String()), win.Payload.Item.NFTID.Chain).MarshalJSON() if err != nil { logger.For(ctx).Error(err) continue } bf := myAtomicPointer.Load() if !bf.Test(chainAddress) { continue }

I think you'll also want to put the atomic.Pointer in the global scope to make it easier to ensure that all methods are using the same reference to the same atomic pointer.

radazen · 2024-01-10T19:29:07Z

db/queries/core/query.sql

+
+
+-- name: CountActiveWallets :one
+select count(w.*) from users u join wallets w on w.id = any(u.wallets) where not u.deleted and not w.deleted and not u.universal;


One edge case here: if one wallet gets added and a different wallet gets deleted, the count stays the same but our active wallets have changed. We could either:

skip this check for now and just update the filter every hour, or

use a query that checks whether any active wallets have been added since <timestamp>, and check that every hour

I think either is fine; if updating the bloom filter is fast, updating it hourly (even if it hasn't changed) sounds okay to me. As more and more people start using Gallery, we probably will end up seeing a change in our active wallets every hour anyway.

benny-conn added 3 commits January 3, 2024 12:43

stream zora events

dcf3ee3

all chains but matic

7064848

back to only zora for now

adbd103

benny-conn requested review from radazen and jarrel-b as code owners January 3, 2024 21:43

benny-conn added 11 commits January 3, 2024 14:22

emails secret

e41adf2

jobs for emails

b7989ad

merge

83d3587

trending feed no override

d529085

dev only admin emails

83e7963

dev lock

3b5b702

dev admin email

9406245

post count

fbffc46

filter

963f1a1

support new additions, remove unused param

87a7f0b

opensea streamer in golang

98e483b

github-actions bot added the query label Jan 4, 2024

correct path

d271aa7

github-advanced-security bot found potential problems Jan 4, 2024

View reviewed changes

opensea-streamer/main.go Fixed Show fixed Hide fixed

benny-conn added 11 commits January 4, 2024 15:20

read directly to json

b7c5fc0

send correct input

f553211

logs

0d58241

to account

cad38ea

no json log

75c5c98

read from correctly

93560c2

ctx

bc6e426

timeouts

a580b6b

actions

258c1f1

alternate check

32ad274

break on err

4cfd1ac

benny-conn added 6 commits January 5, 2024 14:19

no log on json err

7f9faaa

marshalling

5b88619

no throttling

2bb085e

Merge branch 'main' into benny/opensea-streaming

63ce9f1

context logging

24d5cfb

Merge branch 'benny/opensea-streaming' of github.com:gallery-so/go-ga…

c7e7d67

…llery into benny/opensea-streaming

radazen reviewed Jan 10, 2024

View reviewed changes

benny-conn added 2 commits January 11, 2024 13:55

comments

4e018a0

override with extras

ae4d6cc

benny-conn merged commit 7a70e2a into main Jan 11, 2024
6 checks passed

benny-conn deleted the benny/opensea-streaming branch January 11, 2024 22:12

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Benny/opensea streaming #1343

Benny/opensea streaming #1343

benny-conn commented Jan 3, 2024 •

edited

Loading

radazen left a comment

radazen Jan 10, 2024

radazen Jan 10, 2024

radazen Jan 10, 2024

radazen Jan 10, 2024

radazen Jan 10, 2024

benny-conn Jan 11, 2024

radazen Jan 10, 2024

radazen Jan 10, 2024

benny-conn Jan 11, 2024 •

edited

Loading

radazen Jan 10, 2024

benny-conn Jan 11, 2024

radazen Jan 10, 2024

radazen Jan 10, 2024 •

edited

Loading



		-- name: CountActiveWallets :one
		select count(w.*) from users u join wallets w on w.id = any(u.wallets) where not u.deleted and not w.deleted and not u.universal;

Benny/opensea streaming #1343

Benny/opensea streaming #1343

Conversation

benny-conn commented Jan 3, 2024 • edited Loading

radazen left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benny-conn Jan 11, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

radazen Jan 10, 2024 • edited Loading

Choose a reason for hiding this comment

benny-conn commented Jan 3, 2024 •

edited

Loading

benny-conn Jan 11, 2024 •

edited

Loading

radazen Jan 10, 2024 •

edited

Loading