Merge pull request #4 from christianselig/feature/fastjson

Implement fastjson

.github/dependabot.yml

@@ -0,0 +1,11 @@
+# To get started with Dependabot version updates, you'll need to specify which
+# package ecosystems to update and where the package manifests are located.
+# Please see the documentation for all configuration options:
+# https://help.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+
+version: 2
+updates:
+  - package-ecosystem: "gomod" # See documentation for possible values
+    directory: "/" # Location of package manifests
+    schedule:
+      interval: "daily"

.github/workflows/test.yml

@@ -0,0 +1,18 @@
+on: [push, pull_request]
+name: Unit Tests
+jobs:
+  test:
+    name: "go ${{ matrix.go-version }} (${{ matrix.platform }})"
+    strategy:
+      matrix:
+        go-version: [1.16.5]
+        platform: [ubuntu-latest]
+    runs-on: ${{ matrix.platform }}
+    steps:
+    - uses: actions/checkout@v2
+    - name: Setup Go
+      uses: actions/setup-go@v2
+      with:
+        go-version: ${{ matrix.go-version }}
+    - name: Test
+      run: go test ./... -v

go.mod

@@ -16,5 +16,6 @@ require (
 	github.com/sideshow/apns2 v0.20.0
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.2.1
+	github.com/stretchr/testify v1.7.0
 	github.com/valyala/fastjson v1.6.3
 )

internal/reddit/client.go

@@ -1,7 +1,6 @@
 package reddit
 
 import (
-	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -23,7 +22,7 @@ type Client struct {
 	secret string
 	client *http.Client
 	tracer *httptrace.ClientTrace
-	parser *fastjson.Parser
+	pool   *fastjson.ParserPool
 	statsd *statsd.Client
 }
 
@@ -74,14 +73,14 @@ func NewClient(id, secret string, statsd *statsd.Client, connLimit int) *Client
 
 	client := &http.Client{Transport: t}
 
-	parser := &fastjson.Parser{}
+	pool := &fastjson.ParserPool{}
 
 	return &Client{
 		id,
 		secret,
 		client,
 		tracer,
-		parser,
+		pool,
 		statsd,
 	}
 }
@@ -98,7 +97,7 @@ func (rc *Client) NewAuthenticatedClient(refreshToken, accessToken string) *Auth
 	return &AuthenticatedClient{rc, refreshToken, accessToken, nil}
 }
 
-func (rac *AuthenticatedClient) request(r *Request) ([]byte, error) {
+func (rac *AuthenticatedClient) request(r *Request) (*fastjson.Value, error) {
 	req, err := r.HTTPRequest()
 	if err != nil {
 		return nil, err
@@ -122,16 +121,21 @@ func (rac *AuthenticatedClient) request(r *Request) ([]byte, error) {
 		rac.statsd.Incr("reddit.api.errors", r.tags, 0.1)
 		return nil, err
 	}
+
+	parser := rac.pool.Get()
+	defer rac.pool.Put(parser)
+
 	if resp.StatusCode != 200 {
 		rac.statsd.Incr("reddit.api.errors", r.tags, 0.1)
+
 		// Try to parse a json error. Otherwise we generate a generic one
-		rerr := &Error{}
+		val, jerr := parser.ParseBytes(bb)
-		if jerr := json.Unmarshal(bb, rerr); jerr != nil {
+		if jerr != nil {
 			return nil, fmt.Errorf("error from reddit: %d", resp.StatusCode)
 		}
-		return nil, rerr
+		return nil, NewError(val)
 	}
-	return bb, nil
+	return parser.ParseBytes(bb)
 }
 
 func (rac *AuthenticatedClient) RefreshTokens() (*RefreshTokenResponse, error) {
@@ -144,55 +148,47 @@ func (rac *AuthenticatedClient) RefreshTokens() (*RefreshTokenResponse, error) {
 		WithBasicAuth(rac.id, rac.secret),
 	)
 
-	body, err := rac.request(req)
+	val, err := rac.request(req)
-
 	if err != nil {
 		return nil, err
 	}
 
-	rtr := &RefreshTokenResponse{}
+	return NewRefreshTokenResponse(val), nil
-	json.Unmarshal([]byte(body), rtr)
-	return rtr, nil
 }
 
-func (rac *AuthenticatedClient) MessageInbox(from string) (*MessageListingResponse, error) {
+func (rac *AuthenticatedClient) MessageInbox(opts ...RequestOption) (*ListingResponse, error) {
-	req := NewRequest(
+	opts = append([]RequestOption{
 		WithTags([]string{"url:/api/v1/message/inbox"}),
 		WithMethod("GET"),
 		WithToken(rac.accessToken),
 		WithURL("https://oauth.reddit.com/message/inbox.json"),
-		WithQuery("before", from),
+	}, opts...)
-	)
+	req := NewRequest(opts...)
-
-	body, err := rac.request(req)
 
+	val, err := rac.request(req)
 	if err != nil {
 		return nil, err
 	}
 
-	mlr := &MessageListingResponse{}
+	return NewListingResponse(val), nil
-	json.Unmarshal([]byte(body), mlr)
-	return mlr, nil
 }
 
-func (rac *AuthenticatedClient) MessageUnread(from string) (*MessageListingResponse, error) {
+func (rac *AuthenticatedClient) MessageUnread(opts ...RequestOption) (*ListingResponse, error) {
-	req := NewRequest(
+	opts = append([]RequestOption{
 		WithTags([]string{"url:/api/v1/message/unread"}),
 		WithMethod("GET"),
 		WithToken(rac.accessToken),
 		WithURL("https://oauth.reddit.com/message/unread.json"),
-		WithQuery("before", from),
+	}, opts...)
-	)
 
-	body, err := rac.request(req)
+	req := NewRequest(opts...)
 
+	val, err := rac.request(req)
 	if err != nil {
 		return nil, err
 	}
 
-	mlr := &MessageListingResponse{}
+	return NewListingResponse(val), nil
-	json.Unmarshal([]byte(body), mlr)
-	return mlr, nil
 }
 
 func (rac *AuthenticatedClient) Me() (*MeResponse, error) {
@@ -203,14 +199,10 @@ func (rac *AuthenticatedClient) Me() (*MeResponse, error) {
 		WithURL("https://oauth.reddit.com/api/v1/me"),
 	)
 
-	body, err := rac.request(req)
+	val, err := rac.request(req)
-
 	if err != nil {
 		return nil, err
 	}
 
-	mr := &MeResponse{}
+	return NewMeResponse(val), nil
-	err = json.Unmarshal(body, mr)
-
-	return mr, err
 }

internal/reddit/testdata/error.json

@@ -0,0 +1,4 @@
+{
+  "message": "Unauthorized",
+  "error": 401
+}

internal/reddit/testdata/me.json

@@ -0,0 +1,147 @@
+{
+  "is_employee": false,
+  "seen_layout_switch": true,
+  "has_visited_new_profile": false,
+  "pref_no_profanity": true,
+  "has_external_account": false,
+  "pref_geopopular": "GLOBAL",
+  "seen_redesign_modal": true,
+  "pref_show_trending": true,
+  "subreddit": {
+    "default_set": true,
+    "user_is_contributor": false,
+    "banner_img": "",
+    "restrict_posting": true,
+    "user_is_banned": false,
+    "free_form_reports": true,
+    "community_icon": null,
+    "show_media": true,
+    "icon_color": "#545452",
+    "user_is_muted": false,
+    "display_name": "u_hugocat",
+    "header_img": null,
+    "title": "",
+    "coins": 0,
+    "previous_names": [],
+    "over_18": false,
+    "icon_size": [
+      256,
+      256
+    ],
+    "primary_color": "",
+    "icon_img": "https://www.redditstatic.com/avatars/avatar_default_07_545452.png",
+    "description": "",
+    "submit_link_label": "",
+    "header_size": null,
+    "restrict_commenting": false,
+    "subscribers": 0,
+    "submit_text_label": "",
+    "is_default_icon": true,
+    "link_flair_position": "",
+    "display_name_prefixed": "u/hugocat",
+    "key_color": "",
+    "name": "t5_1e6nc8",
+    "is_default_banner": true,
+    "url": "/user/hugocat/",
+    "quarantine": false,
+    "banner_size": null,
+    "user_is_moderator": true,
+    "public_description": "",
+    "link_flair_enabled": false,
+    "disable_contributor_requests": false,
+    "subreddit_type": "user",
+    "user_is_subscriber": false
+  },
+  "pref_show_presence": true,
+  "snoovatar_img": "",
+  "snoovatar_size": null,
+  "gold_expiration": null,
+  "has_gold_subscription": false,
+  "is_sponsor": false,
+  "num_friends": 1,
+  "features": {
+    "mod_service_mute_writes": true,
+    "promoted_trend_blanks": true,
+    "show_amp_link": true,
+    "top_content_email_digest_v2": {
+      "owner": "growth",
+      "variant": "control_1",
+      "experiment_id": 363
+    },
+    "chat": true,
+    "is_email_permission_required": true,
+    "mod_awards": true,
+    "expensive_coins_package": true,
+    "mweb_xpromo_revamp_v2": {
+      "owner": "growth",
+      "variant": "treatment_1",
+      "experiment_id": 457
+    },
+    "awards_on_streams": true,
+    "mweb_xpromo_modal_listing_click_daily_dismissible_ios": true,
+    "chat_subreddit": true,
+    "cookie_consent_banner": true,
+    "modlog_copyright_removal": true,
+    "do_not_track": true,
+    "mod_service_mute_reads": true,
+    "chat_user_settings": true,
+    "use_pref_account_deployment": true,
+    "mweb_xpromo_interstitial_comments_ios": true,
+    "noreferrer_to_noopener": true,
+    "premium_subscriptions_table": true,
+    "mweb_xpromo_interstitial_comments_android": true,
+    "chat_group_rollout": true,
+    "resized_styles_images": true,
+    "spez_modal": true,
+    "mweb_xpromo_modal_listing_click_daily_dismissible_android": true
+  },
+  "can_edit_name": false,
+  "verified": true,
+  "new_modmail_exists": true,
+  "pref_autoplay": true,
+  "coins": 100,
+  "has_paypal_subscription": false,
+  "has_subscribed_to_premium": false,
+  "id": "xgeee",
+  "has_stripe_subscription": false,
+  "oauth_client_id": "5JHxEu-4wnFfBA",
+  "can_create_subreddit": true,
+  "over_18": true,
+  "is_gold": false,
+  "is_mod": true,
+  "awarder_karma": 0,
+  "suspension_expiration_utc": null,
+  "has_verified_email": true,
+  "is_suspended": false,
+  "pref_video_autoplay": true,
+  "in_chat": true,
+  "has_android_subscription": false,
+  "in_redesign_beta": true,
+  "icon_img": "https://www.redditstatic.com/avatars/avatar_default_07_545452.png",
+  "has_mod_mail": true,
+  "pref_nightmode": false,
+  "awardee_karma": 55,
+  "hide_from_robots": false,
+  "password_set": true,
+  "link_karma": 166,
+  "force_password_reset": false,
+  "total_karma": 324,
+  "seen_give_award_tooltip": false,
+  "inbox_count": 0,
+  "seen_premium_adblock_modal": false,
+  "pref_top_karma_subreddits": true,
+  "has_mail": false,
+  "pref_show_snoovatar": false,
+  "name": "hugocat",
+  "pref_clickgadget": 5,
+  "created": 1461652799.0,
+  "gold_creddits": 0,
+  "created_utc": 1461623999.0,
+  "has_ios_subscription": false,
+  "pref_show_twitter": false,
+  "in_beta": false,
+  "comment_karma": 103,
+  "has_subscribed": true,
+  "linked_identities": [],
+  "seen_subreddit_chat_ftux": true
+}

internal/reddit/testdata/refresh_token.json

@@ -0,0 +1,7 @@
+{
+  "access_token": "***REMOVED***",
+  "token_type": "bearer",
+  "expires_in": 3600,
+  "refresh_token": "***REMOVED***",
+  "scope": "account creddits edit flair history identity livemanage modconfig modcontributors modflair modlog modmail modothers modposts modself modtraffic modwiki mysubreddits privatemessages read report save structuredstyles submit subscribe vote wikiedit wikiread"
+}

internal/reddit/types.go

@@ -3,6 +3,8 @@ package reddit
 import (
 	"fmt"
 	"strings"
+
+	"github.com/valyala/fastjson"
 )
 
 type Error struct {
@@ -14,9 +16,50 @@ func (err *Error) Error() string {
 	return fmt.Sprintf("%s (%d)", err.Message, err.Code)
 }
 
-type Message struct {
+func NewError(val *fastjson.Value) *Error {
+	err := &Error{}
+
+	err.Message = string(val.GetStringBytes("message"))
+	err.Code = val.GetInt("error")
+
+	return err
+}
+
+type RefreshTokenResponse struct {
+	AccessToken  string `json:"access_token"`
+	RefreshToken string `json:"refresh_token"`
+}
+
+func NewRefreshTokenResponse(val *fastjson.Value) *RefreshTokenResponse {
+	rtr := &RefreshTokenResponse{}
+
+	rtr.AccessToken = string(val.GetStringBytes("access_token"))
+	rtr.RefreshToken = string(val.GetStringBytes("refresh_token"))
+
+	return rtr
+}
+
+type MeResponse struct {
 	ID   string `json:"id"`
+	Name string
+}
+
+func (mr *MeResponse) NormalizedUsername() string {
+	return strings.ToLower(mr.Name)
+}
+
+func NewMeResponse(val *fastjson.Value) *MeResponse {
+	mr := &MeResponse{}
+
+	mr.ID = string(val.GetStringBytes("id"))
+	mr.Name = string(val.GetStringBytes("name"))
+
+	return mr
+}
+
+type Thing struct {
 	Kind        string  `json:"kind"`
+	ID          string  `json:"id"`
 	Type        string  `json:"type"`
 	Author      string  `json:"author"`
 	Subject     string  `json:"subject"`
@@ -29,33 +72,57 @@ type Message struct {
 	Subreddit   string  `json:"subreddit"`
 }
 
-type MessageData struct {
+func (t *Thing) FullName() string {
-	Message `json:"data"`
+	return fmt.Sprintf("%s_%s", t.Kind, t.ID)
-	Kind    string `json:"kind"`
 }
 
-func (md MessageData) FullName() string {
+func NewThing(val *fastjson.Value) *Thing {
-	return fmt.Sprintf("%s_%s", md.Kind, md.ID)
+	t := &Thing{}
+
+	t.Kind = string(val.GetStringBytes("kind"))
+
+	data := val.Get("data")
+
+	t.ID = string(data.GetStringBytes("id"))
+	t.Type = string(data.GetStringBytes("type"))
+	t.Author = string(data.GetStringBytes("author"))
+	t.Subject = string(data.GetStringBytes("subject"))
+	t.Body = string(data.GetStringBytes("body"))
+	t.CreatedAt = data.GetFloat64("created_utc")
+	t.Context = string(data.GetStringBytes("context"))
+	t.ParentID = string(data.GetStringBytes("parent_id"))
+	t.LinkTitle = string(data.GetStringBytes("link_title"))
+	t.Destination = string(data.GetStringBytes("dest"))
+	t.Subreddit = string(data.GetStringBytes("subreddit"))
+
+	return t
 }
 
-type MessageListing struct {
+type ListingResponse struct {
-	Messages []MessageData `json:"children"`
+	Count    int
+	Children []*Thing
+	After    string
+	Before   string
 }
 
-type MessageListingResponse struct {
+func NewListingResponse(val *fastjson.Value) *ListingResponse {
-	MessageListing MessageListing `json:"data"`
+	lr := &ListingResponse{}
+
+	data := val.Get("data")
+	lr.After = string(data.GetStringBytes("after"))
+	lr.Before = string(data.GetStringBytes("before"))
+	lr.Count = data.GetInt("dist")
+
+	if lr.Count == 0 {
+		return lr
 	}
 
-type RefreshTokenResponse struct {
+	lr.Children = make([]*Thing, lr.Count)
-	AccessToken  string `json:"access_token"`
+	children := data.GetArray("children")
-	RefreshToken string `json:"refresh_token"`
+	for i := 0; i < lr.Count; i++ {
+		t := NewThing(children[i])
+		lr.Children[i] = t
 	}
 
-type MeResponse struct {
+	return lr
-	ID   string `json:"id"`
-	Name string
-}
-
-func (mr *MeResponse) NormalizedUsername() string {
-	return strings.ToLower(mr.Name)
 }

internal/reddit/types_test.go

@@ -0,0 +1,73 @@
+package reddit
+
+import (
+	"io/ioutil"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/valyala/fastjson"
+)
+
+var (
+	parser = &fastjson.Parser{}
+)
+
+func TestMeResponseParsing(t *testing.T) {
+	bb, err := ioutil.ReadFile("testdata/me.json")
+	assert.NoError(t, err)
+
+	val, err := parser.ParseBytes(bb)
+	assert.NoError(t, err)
+
+	me := NewMeResponse(val)
+	assert.NotNil(t, me)
+
+	assert.Equal(t, "xgeee", me.ID)
+	assert.Equal(t, "hugocat", me.Name)
+}
+
+func TestRefreshTokenResponseParsing(t *testing.T) {
+	bb, err := ioutil.ReadFile("testdata/refresh_token.json")
+	assert.NoError(t, err)
+
+	val, err := parser.ParseBytes(bb)
+	assert.NoError(t, err)
+
+	rtr := NewRefreshTokenResponse(val)
+	assert.NotNil(t, rtr)
+
+	assert.Equal(t, "***REMOVED***", rtr.AccessToken)
+	assert.Equal(t, "***REMOVED***", rtr.RefreshToken)
+}
+
+func TestListingResponseParsing(t *testing.T) {
+	bb, err := ioutil.ReadFile("testdata/message_inbox.json")
+	assert.NoError(t, err)
+
+	val, err := parser.ParseBytes(bb)
+	assert.NoError(t, err)
+
+	l := NewListingResponse(val)
+	assert.NotNil(t, l)
+
+	assert.Equal(t, 25, l.Count)
+	assert.Equal(t, 25, len(l.Children))
+	assert.Equal(t, "t1_h470gjv", l.After)
+	assert.Equal(t, "", l.Before)
+
+	thing := l.Children[0]
+	assert.Equal(t, "t4", thing.Kind)
+	assert.Equal(t, "138z6ke", thing.ID)
+	assert.Equal(t, "unknown", thing.Type)
+	assert.Equal(t, "iamthatis", thing.Author)
+	assert.Equal(t, "how goes it", thing.Subject)
+	assert.Equal(t, "how are you today", thing.Body)
+	assert.Equal(t, 1626285395.0, thing.CreatedAt)
+	assert.Equal(t, "hugocat", thing.Destination)
+
+	thing = l.Children[6]
+	assert.Equal(t, "/r/calicosummer/comments/ngcapc/hello_i_am_a_cat/h4q5j98/?context=3", thing.Context)
+	assert.Equal(t, "t1_h46tec3", thing.ParentID)
+	assert.Equal(t, "hello i am a cat", thing.LinkTitle)
+	assert.Equal(t, "calicosummer", thing.Subreddit)
+}

internal/worker/notifications.go

@@ -248,7 +248,7 @@ func (nc *notificationsConsumer) Consume(delivery rmq.Delivery) {
 	nc.logger.WithFields(logrus.Fields{
 		"accountID": id,
 	}).Debug("fetching message inbox")
-	msgs, err := rac.MessageInbox(account.LastMessageID)
+	msgs, err := rac.MessageInbox(reddit.WithQuery("limit", "10"))
 
 	if err != nil {
 		nc.logger.WithFields(logrus.Fields{
@@ -258,20 +258,33 @@ func (nc *notificationsConsumer) Consume(delivery rmq.Delivery) {
 		return
 	}
 
-	nc.logger.WithFields(logrus.Fields{
+	// Figure out where we stand
-		"accountID": id,
+	if msgs.Count == 0 || msgs.Children[0].FullName() == account.LastMessageID {
-		"count":     len(msgs.MessageListing.Messages),
-	}).Debug("fetched messages")
-
-	if len(msgs.MessageListing.Messages) == 0 {
 		nc.logger.WithFields(logrus.Fields{
 			"accountID": id,
 		}).Debug("no new messages, bailing early")
 		return
 	}
 
+	// Find which one is the oldest we haven't notified on
+	oldest := 0
+	for i, t := range msgs.Children {
+		if t.FullName() == account.LastMessageID {
+			break
+		}
+
+		oldest = i
+	}
+
+	tt := msgs.Children[:oldest]
+
+	nc.logger.WithFields(logrus.Fields{
+		"accountID": id,
+		"count":     len(tt),
+	}).Debug("fetched messages")
+
 	// Set latest message we alerted on
-	latestMsg := msgs.MessageListing.Messages[0]
+	latestMsg := tt[0]
 	if err = nc.db.BeginFunc(ctx, func(tx pgx.Tx) error {
 		stmt := `
 			UPDATE accounts
@@ -316,10 +329,12 @@ func (nc *notificationsConsumer) Consume(delivery rmq.Delivery) {
 		devices = append(devices, device)
 	}
 
-	for _, msg := range msgs.MessageListing.Messages {
+	// Iterate backwards so we notify from older to newer
+	for i := len(tt) - 1; i >= 0; i-- {
+		msg := tt[i]
 		notification := &apns2.Notification{}
 		notification.Topic = "com.christianselig.Apollo"
-		notification.Payload = payloadFromMessage(account, &msg, len(msgs.MessageListing.Messages))
+		notification.Payload = payloadFromMessage(account, msg, len(tt))
 
 		for _, device := range devices {
 			notification.DeviceToken = device.APNSToken
@@ -353,7 +368,7 @@ func (nc *notificationsConsumer) Consume(delivery rmq.Delivery) {
 	}).Debug("finishing job")
 }
 
-func payloadFromMessage(acct *data.Account, msg *reddit.MessageData, badgeCount int) *payload.Payload {
+func payloadFromMessage(acct *data.Account, msg *reddit.Thing, badgeCount int) *payload.Payload {
 	postBody := msg.Body
 	if len(postBody) > 2000 {
 		postBody = msg.Body[:2000]