Facebook: message queue initialization code

7 files changed, 99 insertions, 25 deletions

diff --git a/protocols/Facebook/src/db.h b/protocols/Facebook/src/db.h
index dd79f18faa..7f59ad2e74 100644
--- a/protocols/Facebook/src/db.h
+++ b/protocols/Facebook/src/db.h
@@ -27,6 +27,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 // Contact DB keys
 #define DBKEY_LOGIN           "Email"
 #define DBKEY_ID              "ID"
+#define DBKEY_SID             "SID"
 #define DBKEY_TID             "ThreadID"
 #define DBKEY_FIRST_NAME      "FirstName"
 #define DBKEY_SECOND_NAME     "SecondName"
@@ -36,12 +37,9 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #define DBKEY_CLIENT_ID       "ClientID"
 #define DBKEY_DEVICE_ID       "DeviceID"
 #define DBKEY_AVATAR          "Avatar"
-#define DBKEY_DELETED         "DeletedTS"
 #define DBKEY_CONTACT_TYPE    "ContactType"
-#define DBKEY_MESSAGE_ID      "LastMessageId"
-#define DBKEY_MESSAGE_READ    "LastMsgReadTime"
-#define DBKEY_MESSAGE_READERS "MessageReaders"
 #define DBKEY_TOKEN           "Token"
+#define DBKEY_SYNC_TOKEN      "SyncToken"
 
 // Thread specific DB keys 
 #define DBKEY_CHAT_CAN_REPLY  "CanReply"
diff --git a/protocols/Facebook/src/main.cpp b/protocols/Facebook/src/main.cpp
index a00c1cbe23..d521088028 100644
--- a/protocols/Facebook/src/main.cpp
+++ b/protocols/Facebook/src/main.cpp
@@ -19,7 +19,6 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
 
 #include "stdafx.h"
-#include "version.h"
 
 #pragma comment(lib, "Rpcrt4.lib")
 
diff --git a/protocols/Facebook/src/mqtt.cpp b/protocols/Facebook/src/mqtt.cpp
index f23ee010c8..7c76953836 100644
--- a/protocols/Facebook/src/mqtt.cpp
+++ b/protocols/Facebook/src/mqtt.cpp
@@ -110,9 +110,10 @@ bool FacebookProto::MqttParse(const MqttMessage &payload)
 
 	case FB_MQTT_MESSAGE_TYPE_PUBREL:
 		mid = ntohs(*(u_short *)pData);
+		pData += 2;
 		{
 			MqttMessage reply(FB_MQTT_MESSAGE_TYPE_PUBCOMP);
-			reply << mid;
+			reply.writeInt16(mid);
 			MqttSend(reply);
 		}
 		break;
@@ -122,9 +123,10 @@ bool FacebookProto::MqttParse(const MqttMessage &payload)
 
 		if ((flags & FB_MQTT_MESSAGE_FLAG_QOS1) || (flags & FB_MQTT_MESSAGE_FLAG_QOS2)) {
 			mid = ntohs(*(u_short *)pData);
+			pData += 2;
 
 			MqttMessage reply((flags & FB_MQTT_MESSAGE_FLAG_QOS1) ? FB_MQTT_MESSAGE_TYPE_PUBACK : FB_MQTT_MESSAGE_TYPE_PUBREC);
-			reply << mid;
+			reply.writeInt16(mid);
 			MqttSend(reply);
 		}
 
diff --git a/protocols/Facebook/src/proto.cpp b/protocols/Facebook/src/proto.cpp
index 599212932a..d5f153d3ba 100644
--- a/protocols/Facebook/src/proto.cpp
+++ b/protocols/Facebook/src/proto.cpp
@@ -72,6 +72,8 @@ FacebookProto::FacebookProto(const char *proto_name, const wchar_t *username) :
 	}
 
 	m_uid = _atoi64(getMStringA(DBKEY_ID));
+	m_sid = _atoi64(getMStringA(DBKEY_SID));
+	m_szSyncToken = getMStringA(DBKEY_SYNC_TOKEN);
 
 	// Create standard network connection
 	wchar_t descr[512];
diff --git a/protocols/Facebook/src/proto.h b/protocols/Facebook/src/proto.h
index 2eac313a05..5acca32c74 100644
--- a/protocols/Facebook/src/proto.h
+++ b/protocols/Facebook/src/proto.h
@@ -290,6 +290,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *   0: xma_id
  */
 #define FB_API_QUERY_XMA  10153919431161729
+#define FB_API_QUERY_XMA_STR (#FB_API_QUERY_XMA)
 
 /**
  * FB_API_CONTACTS_COUNT:
@@ -353,11 +354,13 @@ class FacebookProto : public PROTO<FacebookProto>
    uint8_t *doZip(size_t cbData, const void *pData, size_t &cbRes);
    uint8_t *doUnzip(size_t cbData, const void *pData, size_t &cbRes);
 
+   void ConnectionFailed();
+
 	AsyncHttpRequest *CreateRequest(const char *szName, const char *szMethod);
 	AsyncHttpRequest *CreateRequestGQL(int64_t id);
 	NETLIBHTTPREQUEST *ExecuteRequest(AsyncHttpRequest *pReq);
 
-	// MQTT functions
+	// MQTT
    void MqttLogin();
    
    void MqttPing();
@@ -371,17 +374,22 @@ class FacebookProto : public PROTO<FacebookProto>
 
    void OnPublish(const char *str, const uint8_t *payLoad, size_t cbLen);
    void OnPublishPresence(FbThriftReader &rdr);
+   void OnPublishQueueCreated(FbThriftReader &rdr);
    void OnPublishUtn(FbThriftReader &rdr);
 
 	HNETLIBCONN m_mqttConn;
+   __int64     m_iMqttId;
+   int16_t     m_mid;        // MQTT message id
 
 	// internal data
-	CMStringA m_szDeviceID; // stored, GUID that identifies this miranda's account
-	CMStringA m_szClientID; // stored, random alphanumeric string of 20 chars
-	__int64   m_uid;        // stored, Facebook user id
-	__int64   m_iMqttId;
-   int16_t   m_mid;
+	CMStringA m_szDeviceID;   // stored, GUID that identifies this miranda's account
+	CMStringA m_szClientID;   // stored, random alphanumeric string of 20 chars
+   __int64   m_uid;          // stored, Facebook user id
+
+   CMStringA m_szSyncToken;  // stored, sequence query token
+   __int64   m_sid;          // stored, Facebook sequence id
 
+   int       m_iUnread;
 	bool      m_invisible;
 	bool      m_bOnline;
 
diff --git a/protocols/Facebook/src/server.cpp b/protocols/Facebook/src/server.cpp
index ee2820a521..8710263dcd 100644
--- a/protocols/Facebook/src/server.cpp
+++ b/protocols/Facebook/src/server.cpp
@@ -20,18 +20,66 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 #include "stdafx.h"
 
+void FacebookProto::ConnectionFailed()
+{
+	ProtoBroadcastAck(0, ACKTYPE_STATUS, ACKRESULT_FAILED, (HANDLE)m_iStatus, m_iDesiredStatus);
+
+	m_iStatus = m_iDesiredStatus = ID_STATUS_OFFLINE;
+	ProtoBroadcastAck(0, ACKTYPE_STATUS, ACKRESULT_SUCCESS, (HANDLE)m_iStatus, m_iDesiredStatus);
+	OnLoggedOut();
+}
+
 void FacebookProto::OnLoggedIn()
 {
 	m_bOnline = true;
 	m_mid = 0;
 
-	ProtoBroadcastAck(0, ACKTYPE_STATUS, ACKRESULT_SUCCESS, (HANDLE)m_iStatus, m_iDesiredStatus);
-	m_iStatus = m_iDesiredStatus;
-
 	MqttPublish("/foreground_state", "{\"foreground\":true, \"keepalive_timeout\":60}");
 
 	MqttSubscribe("/inbox", "/mercury", "/messaging_events", "/orca_presence", "/orca_typing_notifications", "/pp", "/t_ms", "/t_p", "/t_rtc", "/webrtc", "/webrtc_response", 0);
 	MqttUnsubscribe("/orca_message_notifications", 0);
+
+	// if sequence is not initialized, request SID from the server
+	if (m_sid == 0) {
+		auto *pReq = CreateRequestGQL(FB_API_QUERY_SEQ_ID);
+		pReq << CHAR_PARAM("query_params", "{\"1\":\"0\"}");
+		pReq->CalcSig();
+
+		JsonReply reply(ExecuteRequest(pReq));
+		if (reply.error()) {
+			ConnectionFailed();
+			return;
+		}
+
+		auto &n = reply.data()["viewer"]["message_threads"];
+		CMStringW wszSid(n["sync_sequence_id"].as_mstring());
+		setWString(DBKEY_SID, wszSid);
+		m_sid = _wtoi64(wszSid);
+		m_iUnread = n["unread_count"].as_int();
+	}
+
+	ProtoBroadcastAck(0, ACKTYPE_STATUS, ACKRESULT_SUCCESS, (HANDLE)m_iStatus, m_iDesiredStatus);
+	m_iStatus = m_iDesiredStatus;
+
+	// connect message queue
+	JSONNode query;
+	query <<	INT_PARAM("delta_batch_size", 125) << INT_PARAM("max_deltas_able_to_process", 1000) << INT_PARAM("sync_api_version", 3) <<	CHAR_PARAM("encoding", "JSON");
+	if (m_szSyncToken.IsEmpty()) {
+		JSONNode hashes; hashes.set_name("graphql_query_hashes"); hashes << CHAR_PARAM("xma_query_id", __STRINGIFY(FB_API_QUERY_XMA));
+
+		JSONNode xma; xma.set_name(__STRINGIFY(FB_API_QUERY_XMA)); xma << CHAR_PARAM("xma_id", "<ID>");
+		JSONNode hql; hql.set_name("graphql_query_params"); hql << xma;
+
+		JSONNode params; params.set_name("queue_params");
+		params << CHAR_PARAM("buzz_on_deltas_enabled", "false") << hashes << hql;
+
+		query << INT64_PARAM("initial_titan_sequence_id", m_sid) << CHAR_PARAM("device_id", m_szDeviceID) << INT64_PARAM("entity_fbid", m_uid) << params;
+		MqttPublish("/messenger_sync_create_queue", query.write().c_str());
+	}
+	else {
+		query << INT64_PARAM("last_seq_id", m_sid) << CHAR_PARAM("sync_token", m_szSyncToken);
+		MqttPublish("/messenger_sync_get_diffs", query.write().c_str());
+	}
 }
 
 void FacebookProto::OnLoggedOut()
@@ -127,17 +175,15 @@ void FacebookProto::ServerThread(void *)
 
 	if (m_szAuthToken.IsEmpty()) {
 		if (!RefreshToken()) {
-FAIL:
-			ProtoBroadcastAck(0, ACKTYPE_STATUS, ACKRESULT_FAILED, (HANDLE)m_iStatus, m_iDesiredStatus);
-
-			m_iStatus = m_iDesiredStatus = ID_STATUS_OFFLINE;
-			ProtoBroadcastAck(0, ACKTYPE_STATUS, ACKRESULT_SUCCESS, (HANDLE)m_iStatus, m_iDesiredStatus);
+			ConnectionFailed();
 			return;
 		}
 	}
 
-	if (!RefreshContacts())
-		goto FAIL;
+	if (!RefreshContacts()) {
+		ConnectionFailed();
+		return;
+	}
 
 	// connect to MQTT server
 	NETLIBOPENCONNECTION nloc = {};
@@ -147,7 +193,8 @@ FAIL:
 	m_mqttConn = Netlib_OpenConnection(m_hNetlibUser, &nloc);
 	if (m_mqttConn == nullptr) {
 		debugLogA("connection failed, exiting");
-		goto FAIL;
+		ConnectionFailed();
+		return;
 	}
 
 	// send initial packet
@@ -221,6 +268,8 @@ void FacebookProto::OnPublish(const char *topic, const uint8_t *p, size_t cbLen)
 
 	if (!strcmp(topic, "/t_p"))
 		OnPublishPresence(rdr);
+	else if (!strcmp(topic, "/t_ms"))
+		OnPublishQueueCreated(rdr);
 	else if (!strcmp(topic, "/orca_typing_notifications"))
 		OnPublishUtn(rdr);
 }
@@ -276,7 +325,7 @@ void FacebookProto::OnPublishPresence(FbThriftReader &rdr)
 
 		while (!rdr.isStop()) {
 			rdr.readField(fieldType, fieldId);
-			assert(fieldType == FB_THRIFT_TYPE_I64 || fieldType == FB_THRIFT_TYPE_I16 || fieldType == FB_THRIFT_TYPE_I32);
+			assert(fieldType == FB_THRIFT_TYPE_I64 || fieldType == FB_THRIFT_TYPE_I16 || fieldType == FB_THRIFT_TYPE_I32);																									
 			rdr.readIntV(voipBits);
 		}
 
@@ -288,6 +337,21 @@ void FacebookProto::OnPublishPresence(FbThriftReader &rdr)
 	assert(fieldType == FB_THRIFT_TYPE_STOP);
 }
 
+void FacebookProto::OnPublishQueueCreated(FbThriftReader &rdr)
+{
+	auto *pszJson = (const char *)rdr.data();
+	if (*pszJson == 0)
+		pszJson++;
+
+	JSONNode root = JSONNode::parse(pszJson);
+	m_szSyncToken = root["syncToken"].as_mstring();
+	setString(DBKEY_SYNC_TOKEN, m_szSyncToken);
+
+	CMStringW wszSid = root["firstDeltaSeqId"].as_mstring();
+	setWString(DBKEY_SID, wszSid);
+	m_sid = _wtoi64(wszSid);
+}
+
 void FacebookProto::OnPublishUtn(FbThriftReader &rdr)
 {
 	JSONNode root = JSONNode::parse((const char *)rdr.data());
diff --git a/protocols/Facebook/src/stdafx.h b/protocols/Facebook/src/stdafx.h
index f24195c746..6aef7f959f 100644
--- a/protocols/Facebook/src/stdafx.h
+++ b/protocols/Facebook/src/stdafx.h
@@ -61,5 +61,6 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #include "mqtt.h"
 #include "proto.h"
 #include "resource.h"
+#include "version.h"
 
 extern bool g_bMessageState;