pzp-sync/test/realtime.test.js

const test = require('node:test')
const assert = require('node:assert')
const p = require('node:util').promisify
const { createPeer } = require('./util')

async function flatten(iter) {
  const ary = []
  for await (const it of iter) {
    ary.push(it)
  }
  return ary
}

test('sync feed msgs in realtime after the 9 rounds', async (t) => {
  const alice = createPeer({ name: 'alice' })
  const bob = createPeer({ name: 'bob' })

  await alice.db.loaded()
  await bob.db.loaded()

  const bobID = await p(bob.db.account.create)({
    subdomain: 'account',
    _nonce: 'bob',
  })

  await p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: 'm0' },
  })
  assert('bob published post 0')

  const bobPostsID = bob.db.feed.getID(bobID, 'post')

  {
    const arr = (await flatten(alice.db.msgs()))
      .filter((msg) => msg.metadata.account === bobID && msg.data)
      .map((msg) => msg.data.text)
    assert.deepEqual(arr, [], 'alice has no posts from bob')
  }

  bob.goals.set(bobPostsID, 'all')
  alice.goals.set(bobPostsID, 'all')

  const remoteAlice = await p(bob.connect)(alice.getAddress())
  assert('bob connected to alice')

  bob.sync.start()
  await p(setTimeout)(1000)
  assert('sync!')

  {
    const arr = (await flatten(alice.db.msgs()))
      .filter((msg) => msg.metadata.account === bobID && msg.data)
      .map((msg) => msg.data.text)
    assert.deepEqual(arr, ['m0'], 'alice has post 0 from bob')
  }

  await p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: 'm1' },
  })
  assert('bob published post 1')

  await p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: 'm2' },
  })
  assert('bob published post 2')

  {
    let arr
    for (let i = 0; i < 100; i++) {
      arr = (await flatten(alice.db.msgs()))
        .filter((msg) => msg.metadata.account === bobID && msg.data)
        .map((msg) => msg.data.text)
      if (arr.length < 3) {
        await p(setTimeout)(200)
        continue
      }
    }
    assert.deepEqual(arr, ['m0', 'm1', 'm2'], 'alice has posts 0..2 from bob')
  }

  await p(remoteAlice.close)(true)
  await p(alice.close)(true)
  await p(bob.close)(true)
})

test('sync feed msgs in realtime after the 9 rounds, reverse', async (t) => {
  const alice = createPeer({ name: 'alice' })
  const bob = createPeer({ name: 'bob' })

  await alice.db.loaded()
  await bob.db.loaded()

  const bobID = await p(bob.db.account.create)({
    subdomain: 'account',
    _nonce: 'bob',
  })

  await p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: 'm0' },
  })
  assert('bob published post 0')

  const bobPostsID = bob.db.feed.getID(bobID, 'post')

  {
    const arr = (await flatten(alice.db.msgs()))
      .filter((msg) => msg.metadata.account === bobID && msg.data)
      .map((msg) => msg.data.text)
    assert.deepEqual(arr, [], 'alice has no posts from bob')
  }

  bob.goals.set(bobPostsID, 'all')
  alice.goals.set(bobPostsID, 'all')

  const remoteBob = await p(alice.connect)(bob.getAddress())
  assert('bob connected to alice')

  // Reverse direction of who "starts"
  alice.sync.start()
  await p(setTimeout)(1000)
  assert('sync!')

  {
    const arr = (await flatten(alice.db.msgs()))
      .filter((msg) => msg.metadata.account === bobID && msg.data)
      .map((msg) => msg.data.text)
    assert.deepEqual(arr, ['m0'], 'alice has post 0 from bob')
  }

  await p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: 'm1' },
  })
  assert('bob published post 1')

  await p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: 'm2' },
  })
  assert('bob published post 2')

  {
    let arr
    for (let i = 0; i < 100; i++) {
      arr = (await flatten(alice.db.msgs()))
        .filter((msg) => msg.metadata.account === bobID && msg.data)
        .map((msg) => msg.data.text)
      if (arr.length < 3) {
        await p(setTimeout)(200)
        continue
      }
    }
    assert.deepEqual(arr, ['m0', 'm1', 'm2'], 'alice has posts 0..2 from bob')
  }

  await p(remoteBob.close)(true)
  await p(alice.close)(true)
  await p(bob.close)(true)
})

test('create 100 messages in parallel that still manage to sync realtime', async (t) => {
  const alice = createPeer({ name: 'alice' })
  const bob = createPeer({ name: 'bob' })

  await alice.db.loaded()
  await bob.db.loaded()

  const bobID = await p(bob.db.account.create)({
    subdomain: 'account',
    _nonce: 'bob',
  })

  // TODO: for some reason realtime doesn't work unless we post this initial one before starting replication. fix the bug and remove this, or create an issue to resolve it later
  await p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: 'm0' },
  })
  assert('bob published post 0')

  const bobPostsID = bob.db.feed.getID(bobID, 'post')

  {
    const arr = (await flatten(alice.db.msgs()))
      .filter((msg) => msg.metadata.account === bobID && msg.data)
      .map((msg) => msg.data.text)
    assert.deepEqual(arr, [], 'alice has no posts from bob')
  }

  bob.goals.set(bobPostsID, 'all')
  alice.goals.set(bobPostsID, 'all')

  const remoteAlice = await p(bob.connect)(alice.getAddress())
  assert('alice and bob connected')

  bob.sync.start()
  await p(setTimeout)(1000)
  assert('sync!')

  const n = 100
  const hundred = []
  for (let i = 0; i < n; i++) {
      hundred.push(i)
  }
  await Promise.all(hundred.map(i => p(bob.db.feed.publish)({
    account: bobID,
    domain: 'post',
    data: { text: `post nr ${i}` },
  })))
  assert('bob published 100 posts in parallel')

  const bobMsgs = await flatten(bob.db.msgs())
  // 1 for creating bob's account, 1 for the 'post' moot, and 1 for first post
  assert.equal(bobMsgs.length, n + 3, "bob has all of his own messages")

  let arr
  // just waiting for them to arrive
  for (let i = 0; i < 100; i++) {
    arr = (await flatten(alice.db.msgs()))
      // moot doesn't have msg.data
      .filter((msg) => msg.metadata.account === bobID && msg.data)
      .filter(msg => msg.metadata.domain === 'post')
      .map((msg) => msg.data.text)
    if (arr.length < n) {
      await p(setTimeout)(100)
    } else {
      break
    }
  }
  assert.equal(arr.length, n + 1, `alice has ${arr.length + 1} posts from bob`)

  await p(remoteAlice.close)(true)
  await p(alice.close)(true)
  await p(bob.close)(true)
})