WIP: implement messenger DGW sockets

go.mod

@@ -8,6 +8,7 @@ tool go.mau.fi/util/cmd/maubuild
 
 require (
 	github.com/beeper/poly1305 v0.0.0-20250815183548-d4eede7bbf3c
+	github.com/coder/websocket v1.8.14
 	github.com/gabriel-vasile/mimetype v1.4.13
 	github.com/google/go-querystring v1.2.0
 	github.com/google/uuid v1.6.0
@@ -31,7 +32,6 @@ require (
 require (
 	filippo.io/edwards25519 v1.2.0 // indirect
 	github.com/beeper/argo-go v1.1.2 // indirect
-	github.com/coder/websocket v1.8.14 // indirect
 	github.com/coreos/go-systemd/v22 v22.7.0 // indirect
 	github.com/elliotchance/orderedmap/v3 v3.1.0 // indirect
 	github.com/lib/pq v1.12.3 // indirect

pkg/connector/client.go

@@ -76,6 +76,7 @@ type MetaClient struct {
 func (m *MetaConnector) getMessagixConfig() *messagix.Config {
 	return &messagix.Config{
 		MayConnectToDGW:          m.Config.ReceiveInstagramTypingIndicators,
+		UseMessengerDGWRealtime:  m.Config.UseMessengerDGWRealtime,
 		ClientSettings:           m.Bridge.GetHTTPClientSettings(),
 		LogRedactedBloksPayloads: m.Config.LogRedactedBloksPayloads,
 	}
@@ -335,7 +336,7 @@ func (m *MetaClient) connectWithRetry(retryCtx, ctx context.Context, attempts in
 }
 
 func (m *MetaClient) connectWithTable(ctx context.Context, initialTable *table.LSTable, currentUser types.UserInfo) {
-	zerolog.Ctx(ctx).Debug().Msg("Loaded messages page, connecting to MQTT with initial table")
+	zerolog.Ctx(ctx).Debug().Msg("Loaded messages page, connecting to Meta realtime with initial table")
 	go m.handleTableLoop(ctx)
 
 	var err error

pkg/connector/config.go

@@ -43,6 +43,7 @@ type Config struct {
 	// Only affects E2EE chats right now.
 	SendPresenceOnTyping             bool `yaml:"send_presence_on_typing"`
 	ReceiveInstagramTypingIndicators bool `yaml:"receive_instagram_typing_indicators"`
+	UseMessengerDGWRealtime          bool `yaml:"use_messenger_dgw_realtime"`
 	DisableViewOnce                  bool `yaml:"disable_view_once"`
 	MarketplaceSpace                 bool `yaml:"marketplace_space"`
 	LogRedactedBloksPayloads         bool `yaml:"log_redacted_bloks_payloads"`
@@ -98,6 +99,7 @@ func upgradeConfig(helper up.Helper) {
 	helper.Copy(up.Bool, "disable_xma_always")
 	helper.Copy(up.Bool, "send_presence_on_typing")
 	helper.Copy(up.Bool, "receive_instagram_typing_indicators")
+	helper.Copy(up.Bool, "use_messenger_dgw_realtime")
 	helper.Copy(up.Bool, "disable_view_once")
 	helper.Copy(up.Bool, "marketplace_space")
 	helper.Copy(up.Bool, "log_redacted_bloks_payloads")

pkg/connector/example-config.yaml

@@ -48,6 +48,11 @@ send_presence_on_typing: false
 # typing indicators; the existing connection(s) are sufficient to send
 # and receive typing indicators in all other cases.
 receive_instagram_typing_indicators: true
+# Experimental: for facebook.com/messenger.com logins, use the Messenger
+# Web DGW Lightspeed websocket instead of MQTT edge-chat for realtime
+# Lightspeed requests. Leave disabled unless testing browser-parity
+# realtime behavior.
+use_messenger_dgw_realtime: false
 # Should view-once messages be disabled entirely?
 disable_view_once: false
 # Should FB marketplace chats have a separate space inside the main Facebook space?

pkg/messagix/client.go

@@ -36,6 +36,7 @@ type EventHandler func(ctx context.Context, evt any)
 
 type Config struct {
 	MayConnectToDGW          bool
+	UseMessengerDGWRealtime  bool
 	ClientSettings           exhttp.ClientSettings
 	LogRedactedBloksPayloads bool
 }
@@ -47,20 +48,22 @@ type Client struct {
 	Logger        zerolog.Logger
 	Platform      types.Platform
 
-	http         *http.Client
-	httpSettings exhttp.ClientSettings
-	proxyAddr    string
-	socket       *Socket
-	dgwSocket    *dgw.Socket
-	eventHandler EventHandler
-	configs      *Configs
-	syncManager  *SyncManager
-
-	cookies         *cookies.Cookies
-	httpProxy       func(*http.Request) (*url.URL, error)
-	socksProxy      proxy.Dialer
-	GetNewProxy     func(reason string) (string, error)
-	mayConnectToDGW bool
+	http          *http.Client
+	httpSettings  exhttp.ClientSettings
+	proxyAddr     string
+	socket        *Socket
+	dgwSocket     *dgw.Socket
+	dgwLightSpeed *DGWLightSpeedSocket
+	eventHandler  EventHandler
+	configs       *Configs
+	syncManager   *SyncManager
+
+	cookies                 *cookies.Cookies
+	httpProxy               func(*http.Request) (*url.URL, error)
+	socksProxy              proxy.Dialer
+	GetNewProxy             func(reason string) (string, error)
+	mayConnectToDGW         bool
+	useMessengerDGWRealtime bool
 
 	device *store.Device
 
@@ -114,7 +117,9 @@ func NewClient(cookies *cookies.Cookies, logger zerolog.Logger, cfg *Config) *Cl
 		CSRBitmap:          crypto.NewBitmap(),
 	}
 	cli.socket = cli.newSocketClient()
+	cli.dgwLightSpeed = cli.newDGWLightSpeedSocket()
 	cli.mayConnectToDGW = cfg.MayConnectToDGW
+	cli.useMessengerDGWRealtime = cfg.UseMessengerDGWRealtime
 
 	return cli
 }
@@ -124,15 +129,32 @@ func (c *Client) GetCookies() *cookies.Cookies {
 }
 
 type dumpedState struct {
-	Configs     *Configs
-	SyncStore   map[int64]*socket.QueryMetadata
-	PacketsSent uint16
-	SessionID   int64
-	Timestamp   time.Time
+	Configs        *Configs
+	SyncStore      map[int64]*socket.QueryMetadata
+	PacketsSent    uint16
+	DGWStreamsSent uint16
+	SessionID      int64
+	Timestamp      time.Time
 }
 
 func (c *Client) DumpState() (json.RawMessage, error) {
-	if c == nil || c.configs == nil || c.syncManager == nil || c.socket == nil || c.socket.packetsSent == 0 || !c.socket.previouslyConnected {
+	if c == nil || c.configs == nil || c.syncManager == nil {
+		return nil, nil
+	}
+	if c.useDGWLightSpeedRealtime() {
+		if c.dgwLightSpeed == nil || c.dgwLightSpeed.packetsSent == 0 || !c.dgwLightSpeed.previouslyConnected {
+			return nil, nil
+		}
+		return json.Marshal(&dumpedState{
+			Configs:        c.configs,
+			SyncStore:      c.syncManager.store,
+			PacketsSent:    c.dgwLightSpeed.packetsSent,
+			DGWStreamsSent: c.dgwLightSpeed.streamsSent,
+			SessionID:      c.socket.sessionID,
+			Timestamp:      time.Now(),
+		})
+	}
+	if c.socket == nil || c.socket.packetsSent == 0 || !c.socket.previouslyConnected {
 		return nil, nil
 	}
 	return json.Marshal(&dumpedState{
@@ -162,8 +184,17 @@ func (c *Client) LoadState(state json.RawMessage) error {
 	c.configs = dumped.Configs
 	c.syncManager = c.newSyncManager()
 	c.syncManager.store = dumped.SyncStore
+	if c.useDGWLightSpeedRealtime() {
+		c.dgwLightSpeed.packetsSent = dumped.PacketsSent
+		c.dgwLightSpeed.streamsSent = dumped.DGWStreamsSent
+		c.dgwLightSpeed.previouslyConnected = true
+		c.Logger.Info().Str("transport", "dgw_lightspeed").Msg("Loaded state")
+		return nil
+	}
 	c.socket.packetsSent = dumped.PacketsSent
-	c.socket.sessionID = dumped.SessionID
+	if dumped.SessionID != 0 {
+		c.socket.sessionID = dumped.SessionID
+	}
 	if c.Platform == types.Instagram {
 		c.socket.broker = "wss://edge-chat.instagram.com/chat?"
 	} else {
@@ -308,8 +339,15 @@ func (c *Client) UpdateProxy(reason string) bool {
 func (c *Client) Connect(ctx context.Context) error {
 	if c == nil {
 		return ErrClientIsNil
-	} else if err := c.socket.CanConnect(); err != nil {
-		return err
+	}
+	if c.useDGWLightSpeedRealtime() {
+		if err := c.dgwLightSpeed.CanConnect(); err != nil {
+			return err
+		}
+	} else {
+		if err := c.socket.CanConnect(); err != nil {
+			return err
+		}
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	oldCancel := c.stopCurrentConnections.Swap(&cancel)
@@ -328,7 +366,7 @@ func (c *Client) Connect(ctx context.Context) error {
 		for {
 			c.canSendMessages.Clear() // In case we're reconnecting from a normal network error
 			connectStart := time.Now()
-			err := c.socket.Connect(ctx)
+			err := c.connectMainRealtime(ctx)
 			c.canSendMessages.Clear()
 			if ctx.Err() != nil {
 				zerolog.Ctx(ctx).Warn().
@@ -382,6 +420,17 @@ func (c *Client) Connect(ctx context.Context) error {
 	return nil
 }
 
+func (c *Client) connectMainRealtime(ctx context.Context) error {
+	if c.useDGWLightSpeedRealtime() {
+		return c.dgwLightSpeed.Connect(ctx)
+	}
+	return c.socket.Connect(ctx)
+}
+
+func (c *Client) useDGWLightSpeedRealtime() bool {
+	return c != nil && c.useMessengerDGWRealtime && (c.Platform == types.Facebook || c.Platform == types.Messenger)
+}
+
 func (c *Client) connectDGW(ctx context.Context) error {
 	reconnectIn := 2 * time.Second
 	for {
@@ -431,7 +480,12 @@ func (c *Client) Disconnect() {
 	if fn := c.stopCurrentConnections.Load(); fn != nil {
 		(*fn)()
 	}
-	c.socket.Disconnect()
+	if c.socket != nil {
+		c.socket.Disconnect()
+	}
+	if c.dgwLightSpeed != nil {
+		c.dgwLightSpeed.Disconnect()
+	}
 	if c.dgwSocket != nil {
 		c.dgwSocket.Disconnect()
 	}
@@ -441,7 +495,13 @@ func (c *Client) Disconnect() {
 }
 
 func (c *Client) IsConnected() bool {
-	return c != nil && c.socket.conn != nil
+	if c == nil {
+		return false
+	}
+	if c.useDGWLightSpeedRealtime() {
+		return c.dgwLightSpeed != nil && c.dgwLightSpeed.conn != nil
+	}
+	return c.socket != nil && c.socket.conn != nil
 }
 
 func (c *Client) GetEndpoint(name string) string {
@@ -512,8 +572,15 @@ func (c *Client) ForceReconnect() {
 	if c == nil {
 		return
 	}
-	c.socket.Disconnect()
-	c.dgwSocket.Disconnect()
+	if c.socket != nil {
+		c.socket.Disconnect()
+	}
+	if c.dgwLightSpeed != nil {
+		c.dgwLightSpeed.Disconnect()
+	}
+	if c.dgwSocket != nil {
+		c.dgwSocket.Disconnect()
+	}
 }
 
 func (c *Client) FetchMoreThreads(ctx context.Context, syncGroup int64) (*socket.KeyStoreData, *table.LSTable, error) {
@@ -542,13 +609,13 @@ func (c *Client) FetchMoreThreads(ctx context.Context, syncGroup int64) (*socket
 		return nil, nil, err
 	}
 
-	resp, err := c.socket.makeLSRequest(ctx, payload, 3)
+	resp, err := c.makeRealtimeLSRequest(ctx, payload, 3)
 	if err != nil {
 		return nil, nil, err
 	}
 
 	resp.Finish()
-	c.socket.postHandlePublishResponse(resp.Table)
+	c.PostHandlePublishResponse(resp.Table)
 
 	return keyStore, resp.Table, nil
 }

pkg/messagix/configs.go

@@ -34,6 +34,9 @@ func (c *Configs) SetupConfigs(ctx context.Context, ls *table.LSTable) (*table.L
 	if c.client.socket != nil {
 		c.client.socket.previouslyConnected = false
 	}
+	if c.client.dgwLightSpeed != nil {
+		c.client.dgwLightSpeed.previouslyConnected = false
+	}
 	authenticated := c.client.IsAuthenticated()
 	c.WebSessionID = methods.GenerateWebsessionID(authenticated)
 	c.LSDToken = c.BrowserConfigTable.LSD.Token
@@ -58,7 +61,7 @@ func (c *Configs) SetupConfigs(ctx context.Context, ls *table.LSTable) (*table.L
 			936619743392459,
 		)
 	} else {
-		if c.BrowserConfigTable.MqttWebConfig.Endpoint == "" {
+		if c.BrowserConfigTable.MqttWebConfig.Endpoint == "" && !c.client.useDGWLightSpeedRealtime() {
 			return ls, fmt.Errorf("MQTT broker endpoint not found in page response (MqttWebConfig.Endpoint is empty)")
 		}
 		c.client.socket.broker = c.BrowserConfigTable.MqttWebConfig.Endpoint