Added sending of presence state, typing state, and read receipts.

Author: Nordskog

src/Client.ts

@@ -16,6 +16,7 @@ import EventEmitter from 'events'
 import { AttachmentNotFoundError, AttachmentURLMissingError } from './types/Errors'
 import StrictEventEmitter from 'strict-event-emitter-types'
 import ClientEvents from './ClientEvents'
+import * as Payloads from './mqtt/payloads'
 
 const debugLog = debug('fblib')
 
@@ -135,6 +136,32 @@ export default class Client extends (EventEmitter as { new(): ClientEmitter }) {
     return this.mqttApi.sendMessage(threadId, message, options)
   }
 
+  /**
+   * Indicate that the user is currently present in the conversation.
+   * Only relevant for non-group conversations
+   */
+  sendPresenceState = async (recipientUserId: string, present: boolean) => {
+    const payload = new Payloads.PresenceState(recipientUserId, present)
+    this.mqttApi.sendPublish(payload.getTopic(), await Payloads.encodePayload(payload))
+  }
+
+  /**
+   * Send "User is typing" message.
+   * In a non-group conversation, sendPresenceState() must be called first.
+   */
+  sendTypingState = async (threadOrRecipientUserId: string, present: boolean) => {
+    const payload = new Payloads.TypingState(this.session.tokens.uid, present, threadOrRecipientUserId)
+    this.mqttApi.sendPublish(payload.getTopic(), await Payloads.encodePayload(payload))
+  }
+
+  /**
+   * Mark a message as read.
+   */
+  sendReadReceipt = async (message: Message) => {
+    const payload = new Payloads.ReadReceipt(message)
+    this.mqttApi.sendPublish(payload.getTopic(), await Payloads.encodePayload(payload))
+  }
+
   getThreadList = async (count: number): Promise<Thread[]> => {
     const threads = await this.httpApi.threadListQuery(count)
     return threads.viewer.message_threads.nodes.map(parseThread)

src/mqtt/payloads/PresenceStatePayload.ts

@@ -0,0 +1,50 @@
+import Payload from './Payload'
+import { Thrift, TCompactProtocol, Int64 } from 'thrift'
+import Long from 'long'
+
+export default class PresenceStatePayload extends Payload {
+  recipient: string
+  sender: number // Number because always 0
+  state: boolean
+
+  constructor (recipientId: string, state: boolean) {
+    super()
+    this.recipient = recipientId.toString()
+    this.sender = 0
+    this.state = state
+  }
+
+  encode (proto: TCompactProtocol): Promise<void> {
+    /////////////////////
+    // Tracking info
+    /////////////////////
+
+    proto.writeFieldBegin('traceInfo', Thrift.Type.STRING, 1)
+    proto.writeString('')  // Empty string in our case
+
+    // Note lack of end-of-object null
+
+    /////////////////////
+    // Typing presence
+    /////////////////////
+
+    proto.writeFieldBegin('recipient', Thrift.Type.I64, 1)
+    proto.writeI64(new Int64(Buffer.from(Long.fromString(this.recipient).toBytes())))
+
+         // sender always 0
+    proto.writeFieldBegin('sender', Thrift.Type.I64, 2)
+    proto.writeI64(this.sender)
+
+    proto.writeFieldBegin('state', Thrift.Type.I32, 3)
+    proto.writeI32(this.state ? 1 : 0)
+
+    // denotes end of object
+    proto.writeByte(0)
+
+    return null
+  }
+
+  getTopic (): string {
+    return '/t_stp'
+  }
+}

src/mqtt/payloads/ReadReceiptPayload.ts

@@ -0,0 +1,73 @@
+import Payload from './Payload'
+import { Thrift, TCompactProtocol, Int64 } from 'thrift'
+import Message from '../../types/message'
+import RandomIntGenerator from '../../RandomIntGenerator'
+import Long from 'long'
+
+export default class ReadReceiptPayload extends Payload {
+
+  otherUserFbId: string     // Used for individuals
+  threadFbId: string        // Used for groups, same as threadId ( usually ? )
+  watermarkTimestamp: number// Must match timestamp of a message
+
+  public constructor (message: Message) {
+    super()
+
+    if (message.authorId.toString() === message.threadId.toString()) {
+      // If author id equals thread id, then this is not a group convo
+      this.otherUserFbId = message.authorId.toString()
+    } else {
+      // Otherwise it is
+      this.threadFbId = message.threadId.toString()
+    }
+
+    this.watermarkTimestamp = message.timestamp
+
+  }
+
+  encode (proto: TCompactProtocol): Promise<void> {
+    /////////////////////
+    // thrift header?
+    /////////////////////
+
+    // Always completely empty.
+    proto.writeByte(0)
+
+    // Note lack of end-of-object null
+
+    /////////////////////
+    // Typing presence
+    /////////////////////
+
+    proto.writeFieldBegin('mark', Thrift.Type.STRING, 1)
+    proto.writeString('read')
+
+    proto.writeFieldBegin('state', Thrift.Type.BOOL, 2)
+    proto.writeBool(true)
+
+    if (this.threadFbId != null) {
+      proto.writeFieldBegin('threadFbId', Thrift.Type.I64, 6)
+      proto.writeI64(new Int64(Buffer.from(Long.fromString(this.threadFbId).toBytes())))
+    } else if (this.otherUserFbId != null) {
+      proto.writeFieldBegin('otherUserFbId', Thrift.Type.I64, 7)
+      proto.writeI64(new Int64(Buffer.from(Long.fromString(this.otherUserFbId).toBytes())))
+    } else {
+        // Throw?
+    }
+
+    proto.writeFieldBegin('watermarkTimestamp', Thrift.Type.I64, 9)
+    proto.writeI64(this.watermarkTimestamp)
+
+    proto.writeFieldBegin('attemptId', Thrift.Type.I64, 13)
+    proto.writeI64(new Int64(Buffer.from(RandomIntGenerator.getAttemptId().toBytes())))
+
+    // denotes end of object
+    proto.writeByte(0)
+
+    return null
+  }
+
+  getTopic (): string {
+    return '/t_mt_req'
+  }
+}

src/mqtt/payloads/ThreadKey.ts

@@ -0,0 +1,31 @@
+import Payload from './Payload'
+import { Thrift, TCompactProtocol } from 'thrift'
+
+export default class ThreadKeyPayload extends Payload {
+  threadId: string
+  threadType: number    // 0: CANONICAL, 1: GROUP
+
+  constructor (threadId: string, threadType: 0 | 1) {
+    super()
+    this.threadId = threadId.toString()
+    this.threadType = threadType
+  }
+
+  encode (proto: TCompactProtocol): Promise<void> {
+
+    proto.writeFieldBegin('threadId', Thrift.Type.STRING, 1)
+    proto.writeString(this.threadId)
+
+    proto.writeFieldBegin('threadType', Thrift.Type.I32, 2)
+    proto.writeI64(this.threadType)
+
+    // denotes end of object
+    proto.writeByte(0)
+
+    return null
+  }
+
+  getTopic (): string {
+    return null
+  }
+}

src/mqtt/payloads/TypingStatePayload.ts

@@ -0,0 +1,57 @@
+import Payload from './Payload'
+import ThreadKey from './ThreadKey'
+import { Thrift, TCompactProtocol, Int64 } from 'thrift'
+import Long from 'long'
+
+export default class TypingStatePayload extends Payload {
+  recipient: string
+  sender: string
+  state: boolean
+  threadKey: ThreadKey
+
+  constructor (senderUserId: string, state: boolean, recipientUserOrThreadId: string) {
+    super()
+
+    this.sender = senderUserId.toString()
+    this.state = state
+
+    if (recipientUserOrThreadId.toString().startsWith('100')) {
+      this.recipient = recipientUserOrThreadId.toString()
+    } else {
+      this.threadKey = new ThreadKey(recipientUserOrThreadId.toString(), 1)
+    }
+  }
+
+  encode (proto: TCompactProtocol): Promise<void> {
+    // Added explicitly outside of thrift code
+    proto.writeByte(0)
+
+    if (this.recipient != null) {
+      proto.writeFieldBegin('recipient', Thrift.Type.I64, 1)
+      proto.writeI64(new Int64(Buffer.from(Long.fromString(this.recipient).toBytes())))
+    }
+
+    proto.writeFieldBegin('sender', Thrift.Type.I64, 2)
+    proto.writeI64(new Int64(Buffer.from(Long.fromString(this.sender).toBytes())))
+
+    proto.writeFieldBegin('state', Thrift.Type.I32, 3)
+    proto.writeI32(this.state ? 1 : 0)
+
+    if (this.threadKey != null) {
+      proto.writeFieldBegin('threadKey', Thrift.Type.STRUCT, 5)
+      proto.writeStructBegin('threadKey')
+      this.threadKey.encode(proto)
+      proto.writeStructEnd()
+      proto.writeFieldEnd()
+    }
+
+    // denotes end of object
+    proto.writeByte(0)
+
+    return null
+  }
+
+  getTopic (): string {
+    return '/t_st'
+  }
+}

src/mqtt/payloads/index.ts

@@ -1,5 +1,8 @@
 import Payload from './Payload'
 
 export { Payload }
+export { default as PresenceState } from './PresenceStatePayload'
+export { default as ReadReceipt } from './ReadReceiptPayload'
+export { default as TypingState } from './TypingStatePayload'
 
 export const encodePayload = Payload.encodePayload