First stuff for contacting specific nodes for each contact. Hard coded to hit the same bootstrap node for now plus doesn't handle unreachable nodes/errors well yet

app/sql.js

@@ -81,6 +81,8 @@ module.exports = {
   removeSessionsByNumber,
   removeAllSessions,
 
+  getSwarmNodesByPubkey,
+
   getConversationCount,
   saveConversation,
   saveConversations,
@@ -385,6 +387,7 @@ async function updateToSchemaVersion4(currentVersion, instance) {
       type STRING,
       members TEXT,
       name TEXT,
+      swarmNodes TEXT,
       profileName TEXT
     );`
   );
@@ -1025,6 +1028,18 @@ async function removeAllFromTable(table) {
 
 // Conversations
 
+async function getSwarmNodesByPubkey(pubkey) {
+  const row = await db.get('SELECT * FROM conversations WHERE id = $pubkey;', {
+    $pubkey: pubkey,
+  });
+
+  if (!row) {
+    return null;
+  }
+
+  return jsonToObject(row.json).swarmNodes;
+}
+
 async function getConversationCount() {
   const row = await db.get('SELECT count(*) from conversations;');
 
@@ -1037,7 +1052,7 @@ async function getConversationCount() {
 
 async function saveConversation(data) {
   // eslint-disable-next-line camelcase
-  const { id, active_at, type, members, name, friendRequestStatus, profileName } = data;
+  const { id, active_at, type, members, name, friendRequestStatus, swarmNodes, profileName } = data;
 
   await db.run(
     `INSERT INTO conversations (
@@ -1049,6 +1064,7 @@ async function saveConversation(data) {
     members,
     name,
     friendRequestStatus,
+    swarmNodes,
     profileName
   ) values (
     $id,
@@ -1059,6 +1075,7 @@ async function saveConversation(data) {
     $members,
     $name,
     $friendRequestStatus,
+    $swarmNodes,
     $profileName
   );`,
     {
@@ -1070,6 +1087,7 @@ async function saveConversation(data) {
       $members: members ? members.join(' ') : null,
       $name: name,
       $friendRequestStatus: friendRequestStatus,
+      $swarmNodes: swarmNodes ? swarmNodes.join(' ') : null,
       $profileName: profileName,
     }
   );
@@ -1093,7 +1111,7 @@ async function saveConversations(arrayOfConversations) {
 
 async function updateConversation(data) {
   // eslint-disable-next-line camelcase
-  const { id, active_at, type, members, name, friendRequestStatus, profileName } = data;
+  const { id, active_at, type, members, name, friendRequestStatus, swarmNodes, profileName } = data;
 
   await db.run(
     `UPDATE conversations SET
@@ -1104,6 +1122,7 @@ async function updateConversation(data) {
     members = $members,
     name = $name,
     friendRequestStatus = $friendRequestStatus,
+    swarmNodes = $swarmNodes,
     profileName = $profileName
   WHERE id = $id;`,
     {
@@ -1115,6 +1134,7 @@ async function updateConversation(data) {
       $members: members ? members.join(' ') : null,
       $name: name,
       $friendRequestStatus: friendRequestStatus,
+      $swarmNodes: swarmNodes ? swarmNodes.join(' ') : null,
       $profileName: profileName,
     }
   );

config/default.json

@@ -1,6 +1,8 @@
 {
-  "serverUrl": "http://localhost:8080",
-  "cdnUrl": "http://localhost",
+  "serverUrl": "http://qp994mrc8z7fqmsynzdumd35b5918q599gno46br86e537f7qzzy.snode",
+  "cdnUrl": "http://qp994mrc8z7fqmsynzdumd35b5918q599gno46br86e537f7qzzy.snode",
+  "messageServerPort": ":8080",
+  "swarmServerPort": ":8079",
   "disableAutoUpdate": false,
   "openDevTools": false,
   "buildExpiration": 0,

js/conversation_controller.js

@@ -174,6 +174,7 @@
           return conversation;
         }
 
+        conversation.refreshSwarmNodes();
         try {
           await window.Signal.Data.saveConversation(conversation.attributes, {
             Conversation: Whisper.Conversation,

js/models/conversations.js

@@ -86,6 +86,8 @@
         friendRequestStatus: FriendRequestStatusEnum.none,
         unlockTimestamp: null, // Timestamp used for expiring friend requests.
         sessionResetStatus: SessionResetEnum.none,
+        swarmNodes: [],
+        refreshPromise: null,
       };
     },
 
@@ -579,6 +581,22 @@
           throw new Error('Invalid friend request state');
       }
     },
+    async refreshSwarmNodes() {
+      // Refresh promise to ensure that we are only doing a single query at a time
+      let refreshPromise = this.get('refreshPromise');
+      if (refreshPromise === null) {
+        refreshPromise = (async () => {
+          const newSwarmNodes = await window.LokiAPI.getSwarmNodes(this.id);
+          this.set({ swarmNodes: _.union(this.get('swarmNodes'), newSwarmNodes) });
+          await window.Signal.Data.updateConversation(this.id, this.attributes, {
+            Conversation: Whisper.Conversation,
+          });
+        })();
+        this.set({ refreshPromise });
+      }
+      await refreshPromise;
+      this.set({ refreshPromise: null });
+    },
     async setFriendRequestStatus(newStatus) {
       // Ensure that the new status is a valid FriendStatusEnum value
       if (!(newStatus in Object.values(FriendRequestStatusEnum)))
@@ -1198,7 +1216,10 @@
         options.messageType = message.get('type');
 
         // Add the message sending on another queue so that our UI doesn't get blocked
-        this.queueMessageSend(async () =>
+        this.queueMessageSend(async () => {
+          if (this.get('swarmNodes').length === 0) {
+            await this.refreshSwarmNodes();
+          }
           message.send(
             this.wrapSend(
               sendFunction(
@@ -1213,7 +1234,7 @@
               )
             )
           )
-        );
+        });
 
         return true;
       });

js/modules/data.js

@@ -108,6 +108,8 @@ module.exports = {
   removeSessionsByNumber,
   removeAllSessions,
 
+  getSwarmNodesByPubkey,
+
   getConversationCount,
   saveConversation,
   saveConversations,
@@ -654,6 +656,10 @@ async function removeAllSessions(id) {
 
 // Conversation
 
+async function getSwarmNodesByPubkey(pubkey) {
+  return channels.getSwarmNodesByPubkey(pubkey);
+}
+
 async function getConversationCount() {
   return channels.getConversationCount();
 }

js/modules/loki_message_api.js

@@ -5,8 +5,10 @@ const is = require('@sindresorhus/is');
 
 class LokiServer {
 
-  constructor({ urls }) {
+  constructor({ urls, messageServerPort, swarmServerPort }) {
     this.nodes = [];
+    this.messageServerPort = messageServerPort;
+    this.swarmServerPort = swarmServerPort;
     urls.forEach(url => {
       if (!is.string(url)) {
         throw new Error('WebAPI.initialize: Invalid server url');
@@ -15,11 +17,78 @@ class LokiServer {
     });
   }
 
-  async sendMessage(pubKey, data, messageTimeStamp, ttl) {
-    const data64 = dcodeIO.ByteBuffer.wrap(data).toString('base64');
-    // Hardcoded to use a single node/server for now
+  async loadOurSwarm() {
+    const ourKey = window.textsecure.storage.user.getNumber();
+    const nodeAddresses = await this.getSwarmNodes(ourKey);
+    this.ourSwarmNodes = [];
+    nodeAddresses.forEach(url => {
+      this.ourSwarmNodes.push({ url });
+    })
+  }
+
+  async getSwarmNodes(pubKey) {
     const currentNode = this.nodes[0];
 
+    const options = {
+      url: `${currentNode.url}${this.swarmServerPort}/json_rpc`,
+      type: 'POST',
+      responseType: 'json',
+      timeout: undefined,
+    };
+
+    const body = {
+      jsonrpc: '2.0',
+      id: '0',
+      method: 'get_swarm_list_for_messenger_pubkey',
+      params: {
+        pubkey: pubKey,
+      },
+    }
+
+    const fetchOptions = {
+      method: options.type,
+      body: JSON.stringify(body),
+      headers: {
+        'Content-Type': 'application/json',
+      },
+      timeout: options.timeout,
+    };
+
+    let response;
+    try {
+      response = await fetch(options.url, fetchOptions);
+    } catch (e) {
+      log.error(options.type, options.url, 0, 'Error');
+      throw HTTPError('fetch error', 0, e.toString());
+    }
+
+    let result;
+    if (
+      options.responseType === 'json' &&
+      response.headers.get('Content-Type') === 'application/json'
+    ) {
+      result = await response.json();
+    } else if (options.responseType === 'arraybuffer') {
+      result = await response.buffer();
+    } else {
+      result = await response.text();
+    }
+
+    if (response.status >= 0 && response.status < 400) {
+      return result.nodes;
+    }
+    log.error(options.type, options.url, response.status, 'Error');
+    throw HTTPError('sendMessage: error response', response.status, result);
+  }
+
+  async sendMessage(pubKey, data, messageTimeStamp, ttl) {
+    const swarmNodes = await window.Signal.Data.getSwarmNodesByPubkey(pubKey);
+    if (!swarmNodes || swarmNodes.length === 0) {
+      // TODO: Refresh the swarm nodes list
+      throw Error('No swarm nodes to query!');
+    }
+
+    const data64 = dcodeIO.ByteBuffer.wrap(data).toString('base64');
     const timestamp = Math.floor(Date.now() / 1000);
     // Nonce is returned as a base64 string to include in header
     let nonce;
@@ -37,7 +106,7 @@ class LokiServer {
     }
 
     const options = {
-      url: `${currentNode.url}/store`,
+      url: `${swarmNodes[0]}${this.messageServerPort}/store`,
       type: 'POST',
       responseType: undefined,
       timeout: undefined,
@@ -83,11 +152,12 @@ class LokiServer {
   }
 
   async retrieveMessages(pubKey) {
-    // Hardcoded to use a single node/server for now
-    const currentNode = this.nodes[0];
-
+    if (!this.ourSwarmNodes || this.ourSwarmNodes.length === 0) {
+      await this.loadOurSwarm();
+    }
+    const currentNode = this.ourSwarmNodes[0];
     const options = {
-      url: `${currentNode.url}/retrieve`,
+      url: `${currentNode.url}${this.messageServerPort}/retrieve`,
       type: 'GET',
       responseType: 'json',
       timeout: undefined,

main.js

@@ -144,6 +144,8 @@ function prepareURL(pathSegments, moreKeys) {
       buildExpiration: config.get('buildExpiration'),
       serverUrl: config.get('serverUrl'),
       cdnUrl: config.get('cdnUrl'),
+      messageServerPort: config.get('messageServerPort'),
+      swarmServerPort: config.get('swarmServerPort'),
       certificateAuthority: config.get('certificateAuthority'),
       environment: config.environment,
       node_version: process.versions.node,

preload.js

@@ -269,6 +269,8 @@ const { LokiServer } = require('./js/modules/loki_message_api');
 
 window.LokiAPI = new LokiServer({
   urls: [config.serverUrl],
+  messageServerPort: config.messageServerPort,
+  swarmServerPort: config.swarmServerPort,
 });
 
 window.mnemonic = require('./libloki/mnemonic');