feat: is feed retrievable support (#641)

src/bee.ts

@@ -7,10 +7,10 @@ import * as chunk from './modules/chunk'
 import * as pss from './modules/pss'
 import * as status from './modules/status'
 
-import { BeeArgumentError, BeeError } from './utils/error'
+import { BeeArgumentError, BeeError, BeeResponseError } from './utils/error'
 import { prepareWebsocketData } from './utils/data'
 import { fileArrayBuffer, isFile } from './utils/file'
-import { makeFeedReader, makeFeedWriter } from './feed'
+import { Index, IndexBytes, makeFeedReader, makeFeedWriter } from './feed'
 import { makeSigner } from './chunk/signer'
 import { assertFeedType, DEFAULT_FEED_TYPE, FeedType } from './feed/type'
 import { downloadSingleOwnerChunk, uploadSingleOwnerChunkData } from './chunk/soc'
@@ -67,6 +67,7 @@ import type {
 } from './types'
 import { makeDefaultKy, wrapRequestClosure, wrapResponseClosure } from './utils/http'
 import { isReadable } from './utils/stream'
+import { areAllSequentialFeedsUpdateRetrievable } from './feed/retrievable'
 
 /**
  * The main component that abstracts operations available on the main Bee API.
@@ -610,6 +611,55 @@ export class Bee {
     return stewardship.isRetrievable(this.getKy(options), reference)
   }
 
+  /**
+   * Functions that validates if feed is retrievable in the network.
+   *
+   * If no index is passed then it check for "latest" update, which is a weaker guarantee as nobody can be really
+   * sure what is the "latest" update.
+   *
+   * If index is passed then it validates all previous sequence index chunks if they are available as they are required
+   * to correctly resolve the feed upto the given index update.
+   *
+   * @param type
+   * @param owner
+   * @param topic
+   * @param index
+   * @param options
+   */
+  async isFeedRetrievable(
+    type: FeedType,
+    owner: EthAddress | Uint8Array | string,
+    topic: Topic | Uint8Array | string,
+    index?: Index | number | IndexBytes | string,
+    options?: RequestOptions,
+  ): Promise<boolean> {
+    const canonicalOwner = makeEthAddress(owner)
+    const canonicalTopic = makeTopic(topic)
+
+    if (!index) {
+      try {
+        await this.makeFeedReader(type, canonicalTopic, canonicalOwner).download()
+
+        return true
+      } catch (e) {
+        const err = e as BeeResponseError
+
+        // Only if the error is "not-found" then we return false otherwise we re-throw the error
+        if (err?.status === 404) {
+          return false
+        }
+
+        throw e
+      }
+    }
+
+    if (type !== 'sequence') {
+      throw new BeeError('Only Sequence type of Feeds is supported at the moment')
+    }
+
+    return areAllSequentialFeedsUpdateRetrievable(this, canonicalOwner, canonicalTopic, index, options)
+  }
+
   /**
    * Send data to recipient or target with Postal Service for Swarm.
    *

src/feed/index.ts

@@ -37,6 +37,10 @@ export interface Epoch {
   level: number
 }
 
+/**
+ * Bytes of Feed's Index.
+ * For Sequential Feeds this is numeric value in big-endian.
+ */
 export type IndexBytes = Bytes<8>
 export type Index = number | Epoch | IndexBytes | string
 

src/feed/retrievable.ts

@@ -0,0 +1,79 @@
+import { Bee } from '../bee'
+import { EthAddress } from '../utils/eth'
+import { Reference, RequestOptions, Topic } from '../types'
+import { getFeedUpdateChunkReference, Index } from './index'
+import { readUint64BigEndian } from '../utils/uint64'
+import { bytesToHex } from '../utils/hex'
+import { BeeResponseError } from '../utils/error'
+
+function makeNumericIndex(index: Index): number {
+  if (index instanceof Uint8Array) {
+    return readUint64BigEndian(index)
+  }
+
+  if (typeof index === 'string') {
+    return parseInt(index)
+  }
+
+  if (typeof index === 'number') {
+    return index
+  }
+
+  throw new TypeError('Unknown type of index!')
+}
+
+/**
+ * Function that checks if a chunk is retrievable by actually downloading it.
+ * The /stewardship/{reference} endpoint does not support verification of chunks, but only manifest's references.
+ *
+ * @param bee
+ * @param ref
+ * @param options
+ */
+async function isChunkRetrievable(bee: Bee, ref: Reference, options?: RequestOptions): Promise<boolean> {
+  try {
+    await bee.downloadChunk(ref, options)
+
+    return true
+  } catch (e) {
+    const err = e as BeeResponseError
+
+    if (err.status === 404) {
+      return false
+    }
+
+    throw e
+  }
+}
+
+/**
+ * Creates array of references for all sequence updates chunk up to the given index.
+ *
+ * @param owner
+ * @param topic
+ * @param index
+ */
+function getAllSequenceUpdateReferences(owner: EthAddress, topic: Topic, index: Index): Reference[] {
+  const numIndex = makeNumericIndex(index)
+  const updateReferences: Reference[] = new Array(numIndex + 1)
+
+  for (let i = 0; i <= numIndex; i++) {
+    updateReferences[i] = bytesToHex(getFeedUpdateChunkReference(owner, topic, i))
+  }
+
+  return updateReferences
+}
+
+export async function areAllSequentialFeedsUpdateRetrievable(
+  bee: Bee,
+  owner: EthAddress,
+  topic: Topic,
+  index: Index,
+  options?: RequestOptions,
+): Promise<boolean> {
+  const chunkRetrievablePromises = getAllSequenceUpdateReferences(owner, topic, index).map(async ref =>
+    isChunkRetrievable(bee, ref, options),
+  )
+
+  return (await Promise.all(chunkRetrievablePromises)).every(result => result)
+}

test/integration/bee-class.spec.ts

@@ -562,6 +562,63 @@ describe('Bee class', () => {
       FEED_TIMEOUT,
     )
 
+    describe('isFeedRetrievable', () => {
+      const existingTopic = randomByteArray(32, Date.now())
+      const updates: { index: string; reference: BytesReference }[] = [
+        { index: '0000000000000000', reference: makeBytes(32) },
+        { index: '0000000000000001', reference: Uint8Array.from([1, ...makeBytes(31)]) as BytesReference },
+        { index: '0000000000000002', reference: Uint8Array.from([1, 1, ...makeBytes(30)]) as BytesReference },
+      ]
+
+      beforeAll(async () => {
+        const feed = bee.makeFeedWriter('sequence', existingTopic, signer)
+
+        await feed.upload(getPostageBatch(), updates[0].reference)
+        await feed.upload(getPostageBatch(), updates[1].reference)
+        await feed.upload(getPostageBatch(), updates[2].reference)
+      }, FEED_TIMEOUT)
+
+      it('should return false if no feed updates', async () => {
+        const nonExistingTopic = randomByteArray(32, Date.now())
+
+        await expect(bee.isFeedRetrievable('sequence', owner, nonExistingTopic)).resolves.toEqual(false)
+      })
+
+      it(
+        'should return true for latest query for existing topic',
+        async () => {
+          await expect(bee.isFeedRetrievable('sequence', owner, existingTopic)).resolves.toEqual(true)
+        },
+        FEED_TIMEOUT,
+      )
+
+      it(
+        'should return true for index based query for existing topic',
+        async () => {
+          await expect(bee.isFeedRetrievable('sequence', owner, existingTopic, '0000000000000000')).resolves.toEqual(
+            true,
+          )
+          await expect(bee.isFeedRetrievable('sequence', owner, existingTopic, '0000000000000001')).resolves.toEqual(
+            true,
+          )
+          await expect(bee.isFeedRetrievable('sequence', owner, existingTopic, '0000000000000002')).resolves.toEqual(
+            true,
+          )
+        },
+        FEED_TIMEOUT,
+      )
+
+      it(
+        'should return false for index based query for existing topic but non-existing index',
+        async () => {
+          await expect(bee.isFeedRetrievable('sequence', owner, existingTopic, '0000000000000005')).resolves.toEqual(
+            false,
+          )
+        },
+        FEED_TIMEOUT,
+      )
+    })
+
     describe('topic', () => {
       it('create feed topic', () => {
         const topic = bee.makeFeedTopic('swarm.eth:application:handshake')