Merge pull request #496 from xmtp/np/group-streaming-every-persons-group

Bug streaming groups multi client

example/src/tests/groupTests.ts

@@ -1247,6 +1247,29 @@ test('can stream all groups and conversations', async () => {
   return true
 })
 
+test('can stream groups and messages', async () => {
+  const [alixClient, boClient] = await createClients(2)
+
+  // Start streaming groups
+  const groups: Group<any>[] = []
+  await alixClient.conversations.streamGroups(async (group: Group<any>) => {
+    groups.push(group)
+  })
+  // Stream messages twice
+  await alixClient.conversations.streamAllMessages(async (message) => {}, true)
+  await alixClient.conversations.streamAllMessages(async (message) => {}, true)
+
+  // bo creates a group with alix so a stream callback is fired
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  await boClient.conversations.newGroup([alixClient.address])
+  await delayToPropogate()
+  if ((groups.length as number) !== 1) {
+    throw Error(`Unexpected num groups (should be 1): ${groups.length}`)
+  }
+
+  return true
+})
+
 test('canMessage', async () => {
   const [bo, alix, caro] = await createClients(3)
 
@@ -2261,6 +2284,43 @@ test('can sync all groups', async () => {
   return true
 })
 
+test('only streams groups that can be decrypted', async () => {
+  // Create three MLS enabled Clients
+  const [alixClient, boClient, caroClient] = await createClients(3)
+  const alixGroups: Group<any>[] = []
+  const boGroups: Group<any>[] = []
+  const caroGroups: Group<any>[] = []
+
+  await alixClient.conversations.streamGroups(async (group: Group<any>) => {
+    alixGroups.push(group)
+  })
+  await boClient.conversations.streamGroups(async (group: Group<any>) => {
+    boGroups.push(group)
+  })
+  await caroClient.conversations.streamGroups(async (group: Group<any>) => {
+    caroGroups.push(group)
+  })
+
+  await alixClient.conversations.newGroup([boClient.address])
+
+  assert(
+    alixGroups.length === 1,
+    `alix group length should be 1 but was ${alixGroups.length}`
+  )
+
+  assert(
+    boGroups.length === 1,
+    `bo group length should be 1 but was ${boGroups.length}`
+  )
+
+  assert(
+    caroGroups.length !== 1,
+    `caro group length should be 0 but was ${caroGroups.length}`
+  )
+
+  return true
+})
+
 // Commenting this out so it doesn't block people, but nice to have?
 // test('can stream messages for a long time', async () => {
 //   const bo = await Client.createRandom({ env: 'local', enableV3: true })

src/lib/Conversations.ts

@@ -146,7 +146,7 @@ export default class Conversations<
     const groupsSubscription = XMTPModule.emitter.addListener(
       EventTypes.Group,
       async ({ inboxId, group }: { inboxId: string; group: GroupParams }) => {
-        if (this.known[group.id]) {
+        if (this.known[group.id] || this.client.inboxId !== inboxId) {
           return
         }
         this.known[group.id] = true