Voice message waveforms, an XMA suppression option, and readable polls

pkg/connector/client.go

@@ -130,7 +130,11 @@ func (m *MetaConnector) getProxy(reason string) (string, error) {
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return "", fmt.Errorf("failed to send request: %w", err)
-	} else if resp.StatusCode >= 300 || resp.StatusCode < 200 {
+	}
+	defer func() {
+		_ = resp.Body.Close()
+	}()
+	if resp.StatusCode >= 300 || resp.StatusCode < 200 {
 		return "", fmt.Errorf("unexpected status code %d", resp.StatusCode)
 	}
 	var respData respGetProxy
@@ -507,7 +511,7 @@ func (m *MetaClient) saveConnectionState(ctx context.Context, state json.RawMess
 	}
 	m.lastStateSaveLock.Unlock()
 	if state == nil {
-		if !ratelimited {
+		if ratelimited {
 			return
 		}
 		state, _ = m.Client.DumpState()

pkg/connector/handlemeta.go

@@ -143,6 +143,7 @@ func (m *MetaClient) parseAndQueueTable(ctx context.Context, tbl *table.LSTable,
 	}
 	if ctx.Err() != nil {
 		zerolog.Ctx(ctx).Warn().Err(ctx.Err()).Msg("Not dispatching parsed table, context is canceled")
+		return
 	}
 	select {
 	case m.parsedTables <- wrapped:

pkg/connector/userinfo.go

@@ -46,8 +46,21 @@ const (
 	MetaAIMessengerID = 156025504001094
 )
 
+// makeUserIdentifiers returns the identifier list for a Facebook user, in the same
+// `network:handle` form the Instagram connector uses. Vanity usernames are optional
+// on Facebook, so fall back to the numeric ID, which works in profile URLs too.
+func makeUserIdentifiers(info types.UserInfo) []string {
+	if username := info.GetUsername(); username != "" {
+		return []string{fmt.Sprintf("facebook:%s", username)}
+	} else if fbid := info.GetFBID(); fbid != 0 {
+		return []string{fmt.Sprintf("facebook:%d", fbid)}
+	}
+	return nil
+}
+
 func (m *MetaClient) wrapUserInfo(info types.UserInfo) *bridgev2.UserInfo {
 	return &bridgev2.UserInfo{
+		Identifiers: makeUserIdentifiers(info),
 		Name: ptr.Ptr(m.Main.Config.FormatDisplayname(DisplaynameParams{
 			DisplayName: info.GetName(),
 			Username:    info.GetUsername(),

pkg/igconnector/chatsync.go

@@ -130,37 +130,46 @@ func (ic *IGClient) doChatBackfill(ctx context.Context, startCursor string) {
 	}
 
 	batchCount := 0
-	if startCursor == "" {
+	cursor := startCursor
+	hasNextPage := true
+	if cursor == "" {
 		log.Info().Msg("No start cursor, loading from scratch")
 		resp, err := ic.Client.GetMailbox(ctx)
 		if err != nil {
 			log.Err(err).Msg("Failed to fetch initial inbox")
 			return
+		} else if resp.Mailbox == nil {
+			log.Warn().Msg("Initial inbox response didn't contain a mailbox")
+			return
 		}
 		if !processThreads(resp.Mailbox.ThreadsByFolder) {
 			return
 		}
+		cursor = resp.Mailbox.ThreadsByFolder.PageInfo.EndCursor
+		hasNextPage = resp.Mailbox.ThreadsByFolder.PageInfo.HasNextPage
 	} else {
 		batchCount++
 	}
-	for batchCount < ic.Main.Config.ThreadBackfill.BatchCount {
+	for hasNextPage && cursor != "" && batchCount < ic.Main.Config.ThreadBackfill.BatchCount {
 		select {
 		case <-time.After(ic.Main.Config.ThreadBackfill.BatchDelay):
 		case <-ctx.Done():
 			return
 		}
-		resp, err := ic.Client.PaginateMailbox(ctx, slidetypes.MakePaginateMailboxRequest(viewerFBID, startCursor, "INBOX", nil))
+		resp, err := ic.Client.PaginateMailbox(ctx, slidetypes.MakePaginateMailboxRequest(viewerFBID, cursor, "INBOX", nil))
 		if err != nil {
 			log.Err(err).Msg("Failed to fetch more chats")
 			return
+		} else if resp.Mailbox == nil {
+			log.Warn().Msg("Pagination response didn't contain a mailbox")
+			return
 		}
 		if !processThreads(resp.Mailbox.ThreadsByFolder) {
 			return
 		}
 		batchCount++
-		if !resp.Mailbox.ThreadsByFolder.PageInfo.HasNextPage {
-			break
-		}
+		cursor = resp.Mailbox.ThreadsByFolder.PageInfo.EndCursor
+		hasNextPage = resp.Mailbox.ThreadsByFolder.PageInfo.HasNextPage
 	}
 	log.Info().Int("total_batch_count", batchCount).Msg("Completed chat backfill successfully")
 	ic.LoginMeta.BackfillCompleted = true

pkg/igconnector/client.go

@@ -106,7 +106,11 @@ func (ic *IGConnector) getProxy(reason string) (string, error) {
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return "", fmt.Errorf("failed to send request: %w", err)
-	} else if resp.StatusCode >= 300 || resp.StatusCode < 200 {
+	}
+	defer func() {
+		_ = resp.Body.Close()
+	}()
+	if resp.StatusCode >= 300 || resp.StatusCode < 200 {
 		return "", fmt.Errorf("unexpected status code %d", resp.StatusCode)
 	}
 	var respData respGetProxy

pkg/igconnector/config.go

@@ -37,6 +37,7 @@ type Config struct {
 
 	DisableXMABackfill bool `yaml:"disable_xma_backfill"`
 	DisableXMAAlways   bool `yaml:"disable_xma_always"`
+	SuppressXMA        bool `yaml:"suppress_xma"`
 
 	MinFullReconnectIntervalSeconds int  `yaml:"min_full_reconnect_interval_seconds"`
 	ForceRefreshIntervalSeconds     int  `yaml:"force_refresh_interval_seconds"`
@@ -86,6 +87,7 @@ func upgradeConfig(helper up.Helper) {
 	helper.Copy(up.Bool, "cache_connection_state")
 	helper.Copy(up.Bool, "disable_xma_backfill")
 	helper.Copy(up.Bool, "disable_xma_always")
+	helper.Copy(up.Bool, "suppress_xma")
 	helper.Copy(up.Bool, "disable_typing")
 	helper.Copy(up.Bool, "disable_view_once")
 	helper.Copy(up.Int, "thread_backfill", "batch_count")

pkg/igconnector/connector.go

@@ -44,6 +44,7 @@ func (ic *IGConnector) Init(bridge *bridgev2.Bridge) {
 	ic.DB = metadb.New(bridge.ID, bridge.DB.Database, ic.Bridge.Log.With().Str("db_section", "meta").Logger())
 	ic.MsgConv = igconv.New(bridge, ic.DB)
 	ic.MsgConv.DisableViewOnce = ic.Config.DisableViewOnce
+	ic.MsgConv.SuppressXMA = ic.Config.SuppressXMA
 }
 
 func (ic *IGConnector) Start(ctx context.Context) error {

pkg/igconnector/example-config.yaml

@@ -23,6 +23,10 @@ cache_connection_state: true
 disable_xma_backfill: true
 # Disable fetching XMA media entirely.
 disable_xma_always: false
+# Don't bridge the preview image of XMA media at all, only the caption and a link to the original.
+# The disable_xma_* options above only stop the extra requests for the full quality media,
+# the preview is still reuploaded. This drops it too, so nothing is stored on the homeserver.
+suppress_xma: false
 # Should typing notification bridging be disabled?
 # Bridging typing notifications requires 2 extra connections per user (one for receiving and one for sending),
 # because Meta only employs professional software engineers who make excellent architecture decisions.

pkg/instameow/client.go

@@ -212,6 +212,8 @@ func (c *Client) LoadIndex(ctx context.Context) (*types.PolarisViewer, *slidetyp
 	mailbox, err := c.GetMailbox(ctx)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to get mailbox: %w", err)
+	} else if mailbox.GetMailbox() == nil {
+		return nil, nil, ErrMailboxMissing
 	}
 	c.seqID = mailbox.Mailbox.UQSeqID
 	c.seqIDTS = time.Now()
@@ -271,6 +273,7 @@ const MaxCachedStateAge = 24 * time.Hour
 
 var ErrCachedStateTooOld = errors.New("cached state is too old")
 var ErrClientIsNil = errors.New("client is nil")
+var ErrMailboxMissing = errors.New("mailbox missing from response")
 
 func (c *Client) LoadState(state json.RawMessage) error {
 	if c == nil {

pkg/instameow/login_account_manager.go

@@ -266,7 +266,7 @@ func (c *Client) switchInstagramAccountManagerProfile(
 		return fmt.Errorf("failed to prepare the primary Instagram web session: %w", err)
 	}
 	primaryCookies := c.cookies.GetAll()
-	primaryWWWClaim := c.cookies.IGWWWClaim
+	primaryWWWClaim := c.cookies.GetWWWClaim()
 
 	if err := c.switchInstagramAccountManagerMobileAccount(ctx, state, account); err != nil {
 		return err
@@ -276,7 +276,7 @@ func (c *Client) switchInstagramAccountManagerProfile(
 	// provisioned separately. Switch the already authenticated primary web session
 	// through Instagram's matching FXCAL endpoint before persisting the selection.
 	c.cookies.UpdateValues(primaryCookies)
-	c.cookies.IGWWWClaim = primaryWWWClaim
+	c.cookies.SetWWWClaim(primaryWWWClaim)
 	if err := c.switchInstagramAccountManagerWebAccount(ctx, account.Username); err != nil {
 		return err
 	}

pkg/instameow/login_web.go

@@ -159,10 +159,10 @@ func (c *Client) addInstagramWebLoginHeaders(headers http.Header) error {
 	}
 	headers.Set("x-instagram-ajax", config.InstagramWebPushInfo.RolloutHash)
 	headers.Set("x-web-session-id", c.configs.WebSessionID)
-	if c.cookies.IGWWWClaim == "" {
+	if wwwClaim := c.cookies.GetWWWClaim(); wwwClaim == "" {
 		headers.Set("x-ig-www-claim", "0")
 	} else {
-		headers.Set("x-ig-www-claim", c.cookies.IGWWWClaim)
+		headers.Set("x-ig-www-claim", wwwClaim)
 	}
 	if config.PolarisSiteData.SendDeviceIDHeader {
 		if config.PolarisSiteData.DeviceID == "" {
@@ -329,7 +329,7 @@ func (c *Client) CreateInstagramWebSession(
 		cookies.IGCookieMachineID: c.cookies.Get(cookies.IGCookieMachineID),
 		cookies.IGCookieDeviceID:  c.cookies.Get(cookies.IGCookieDeviceID),
 	})
-	c.cookies.IGWWWClaim = ""
+	c.cookies.SetWWWClaim("")
 
 	c.configs = httpclient.NewConfigs(c)
 	c.http.SetConfigs(c.configs)

pkg/messagix/cookies/cookies.go

@@ -54,7 +54,7 @@ type Cookies struct {
 	values   map[MetaCookieName]string
 	lock     sync.RWMutex
 
-	IGWWWClaim string
+	igWWWClaim string
 }
 
 func (c *Cookies) UpdateValues(newValues map[MetaCookieName]string) {
@@ -67,10 +67,14 @@ func (c *Cookies) UpdateValues(newValues map[MetaCookieName]string) {
 }
 
 func (c *Cookies) MarshalJSON() ([]byte, error) {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
 	return json.Marshal(c.values)
 }
 
 func (c *Cookies) UnmarshalJSON(data []byte) error {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	return json.Unmarshal(data, &c.values)
 }
 
@@ -158,6 +162,21 @@ func (c *Cookies) Set(key MetaCookieName, value string) {
 	c.values[key] = value
 }
 
+func (c *Cookies) GetWWWClaim() string {
+	if c == nil {
+		return ""
+	}
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	return c.igWWWClaim
+}
+
+func (c *Cookies) SetWWWClaim(claim string) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.igWWWClaim = claim
+}
+
 func (c *Cookies) UpdateFromResponse(r *http.Response) {
 	if c == nil || r == nil {
 		return
@@ -181,7 +200,7 @@ func (c *Cookies) UpdateFromResponse(r *http.Response) {
 		}
 	}
 	if wwwClaim := r.Header.Get("x-ig-set-www-claim"); wwwClaim != "" {
-		c.IGWWWClaim = wwwClaim
+		c.igWWWClaim = wwwClaim
 	}
 }
 

pkg/messagix/cookies/cookies_test.go

@@ -0,0 +1,70 @@
+package cookies
+
+import (
+	"encoding/json"
+	"strconv"
+	"sync"
+	"testing"
+)
+
+const concurrencyIterations = 500
+
+// Cookies is stored in the user login metadata, so it gets marshaled from
+// whichever goroutine calls UserLogin.Save while HTTP responses keep updating
+// it. Run both at once to make sure the map is never touched without the lock.
+func TestCookiesConcurrentMarshalAndUpdate(t *testing.T) {
+	c := &Cookies{}
+	c.UpdateValues(map[MetaCookieName]string{IGCookieSessionID: "session"})
+
+	errs := make(chan error, concurrencyIterations)
+	var wg sync.WaitGroup
+	wg.Add(3)
+	go func() {
+		defer wg.Done()
+		for i := range concurrencyIterations {
+			c.Set(IGCookieRUR, strconv.Itoa(i))
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for range concurrencyIterations {
+			if _, err := json.Marshal(c); err != nil {
+				errs <- err
+			}
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for range concurrencyIterations {
+			if err := json.Unmarshal([]byte(`{"sessionid":"other"}`), c); err != nil {
+				errs <- err
+			}
+		}
+	}()
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		t.Errorf("Unexpected error: %v", err)
+	}
+}
+
+func TestCookiesConcurrentWWWClaim(t *testing.T) {
+	c := &Cookies{}
+	c.UpdateValues(nil)
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		for i := range concurrencyIterations {
+			c.SetWWWClaim(strconv.Itoa(i))
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for range concurrencyIterations {
+			c.GetWWWClaim()
+		}
+	}()
+	wg.Wait()
+}

pkg/messagix/httpclient/graphql.go

@@ -254,7 +254,7 @@ func (c *HTTPClient) MakeGraphQLRequest(ctx context.Context, name string, variab
 
 	reqUrl := c.parent.GetEndpoint("graphql")
 	//c.Logger.Info().Any("url", reqUrl).Any("payload", string(payloadBytes)).Any("headers", headers).Msg("Sending graphQL request.")
-	resp, respData, err := c.makeRequest(ctx, reqUrl, "POST", headers, payloadBytes, types.FORM, func(e *zerolog.Event) *zerolog.Event {
+	resp, respData, err := c.makeRequest(ctx, reqUrl, "POST", headers, payloadBytes, types.FORM, c.HTTP, func(e *zerolog.Event) *zerolog.Event {
 		return e.Str("graphql_method", name)
 	})
 	if err == nil && resp != nil {