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 (create an initial post before starting realtime too)', 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('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} posts from bob`)

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

test('create 100 messages in parallel that still manage to sync realtime (without creating an initial post before starting 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',
  })

  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, and 1 for the 'post' moot
  assert.equal(bobMsgs.length, n + 2, "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, `alice has ${arr.length} posts from bob`)

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

test('create 200 messages that manage to replicate with low "newest" goals', async (t) => {
  const n = 200

  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',
  })

  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')
  }

  const confirmed = []
  // for keeping track of which msgs have arrived
  for (let i = 0; i < n; i++) {
    confirmed.push(false)
  }

  alice.db.onRecordAdded(rec => {
    if (rec.msg.data?.text) {
      const num = Number.parseInt(rec.msg.data.text)
      confirmed[num] = true
    }
  })

  bob.goals.set(bobPostsID, 'newest-50')
  alice.goals.set(bobPostsID, 'newest-50')

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

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

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

  let tries = 30
  // just waiting for them to arrive
  do {
    await p(setTimeout)(100)
  } while (!confirmed.every(v => v === true) && tries-- > 0)
  console.log({tries})

  assert.equal(confirmed.filter(v => v === true).length, n, `alice has all of bob's posts`)

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