feat: add query endpoint

lib/client/index.js

@@ -192,6 +192,67 @@ class Client {
     return () => this._supscriptions.delete(url, callback)
   }
 
+  /**
+   * Returns a list of items in a directory.
+   *
+   * @param {string} directory
+   * @param {object} [opts]
+   * @param {boolean} [opts.skipCache] - Skip the local cache and wait for the remote relay to respond with fresh data
+   *
+   * @returns {Promise<Array<string>>}
+   */
+  async list (directory, opts = {}) {
+    if (!directory.endsWith('/')) directory += '/'
+    if (!directory.startsWith('/')) directory = '/' + directory
+
+    if (opts.skipCache) {
+      return this._listRelay(directory)
+    }
+
+    return this._listLocal(directory)
+  }
+
+  /**
+   * Returns a list of items in a directory from local store.
+   *
+   * @param {string} directory
+   *
+   * @returns {Promise<Array<string>>}
+   */
+  async _listLocal (directory) {
+    const range = {
+      gte: PREFIXES.RECORDS + this.id + directory,
+      //
+      lte: PREFIXES.RECORDS + this.id + directory.slice(0, directory.length - 1) + '~'
+    }
+
+    const list = await this._store.iterator(range).all()
+
+    return list.map(r => r[0].split(PREFIXES.RECORDS + this.id + directory)[1])
+  }
+
+  /**
+   * Returns a list of items in a directory from the relay.
+   *
+   * @param {string} directory
+   *
+   * @returns {Promise<Array<string>>}
+   */
+  async _listRelay (directory) {
+    const path = this.id + directory
+    const url = this._relay + '/' + path
+
+    const response = await fetch(url)
+
+    if (!response.ok) {
+      throw new Error(`Failed to list director ${directory} from relay ${this._relay}, status: ${response.status}`)
+    }
+
+    const text = await response.text()
+
+    return text.split('\n').map(p => p.split(path)[1]).filter(Boolean)
+  }
+
   /**
      * Return a url that can be shared by others to acess a file.
      *

lib/server/index.js

@@ -216,16 +216,20 @@ class Relay {
       return
     }
 
+    const url = new URL('http://example.com' + req.url)
+
     switch (req.method) {
       case 'OPTIONS':
         this._OPTIONS(req, res)
         break
       case 'GET':
         if (req.url.startsWith('/subscribe/')) {
           this._SUBSCRIBE(req, res)
-          return
+        } else if (url.pathname.endsWith('/')) {
+          this._QUERY(req, res, url)
+        } else {
+          this._GET(req, res)
         }
-        this._GET(req, res)
         break
       case 'PUT':
         this._PUT(req, res)
@@ -491,6 +495,46 @@ class Relay {
     })
   }
 
+  /**
+   * @param {string} id
+   * @param {string} directory
+   * @param {object} [options]
+   * @param {number} [options.limit]
+   * @param {number} [options.offset]
+   *
+   * @returns {lmdb.RangeIterable}
+   */
+  _searchDirectory (id, directory, options = {}) {
+    const range = { start: '/' + id + directory, end: '/' + id + directory.slice(0, directory.length - 1) + '0', ...options }
+
+    return this._recordsDB.getRange(range)
+  }
+
+  /**
+   * @param {http.IncomingMessage} _req
+   * @param {http.ServerResponse} res
+   * @param {URL} url
+   */
+  async _QUERY (_req, res, url) {
+    const [id, ...rest] = url.pathname.slice(1).split('/')
+    const dir = '/' + rest.join('/')
+
+    const options = {
+      // Skip options until we really need them. currently, it should be cheap to return 1000s of keys at once.
+
+      // limit: parseInt(url.searchParams.get('limit')) || 1000,
+      // offset: parseInt(url.searchParams.get('offset')) || 0
+    }
+
+    const keys = this._searchDirectory(id, dir, options).asArray.map(r => {
+      this._updateLastQueried(r.key)
+      return r.key
+    })
+
+    res.writeHead(200)
+    res.end(keys.join('\n'))
+  }
+
   /**
    * Health check endpoint to provide server metrics.
    *

test/client.js

@@ -314,6 +314,40 @@ test('throw errors on server 4XX response if awaitRelaySync set to true', async
   relay.close()
 })
 
+test('list', async (t) => {
+  const relay = new Relay(tmpdir())
+
+  const address = await relay.listen()
+
+  const content = b4a.from(JSON.stringify({
+    name: 'Alice Bob Carl'
+  }))
+
+  const client = new Client({ storage: tmpdir(), relay: address })
+
+  for (let i = 0; i < 10; i++) {
+    await client.put(`/dir/subdir/foo${i}.txt`, content, { awaitRelaySync: true })
+    await client.put(`/dir/wrong/foo${i}.txt`, content, { awaitRelaySync: true })
+  }
+
+  const list = await client.list('/dir/subdir', { skipCache: true })
+
+  t.alike(list, [
+    'foo0.txt',
+    'foo1.txt',
+    'foo2.txt',
+    'foo3.txt',
+    'foo4.txt',
+    'foo5.txt',
+    'foo6.txt',
+    'foo7.txt',
+    'foo8.txt',
+    'foo9.txt'
+  ])
+
+  relay.close()
+})
+
 function tmpdir () {
   return path.join(os.tmpdir(), Math.random().toString(16).slice(2))
 }

test/server.js

@@ -535,6 +535,63 @@ test('save query dates to help prunning abandoned records later', async (t) => {
   relay.close()
 })
 
+test('query all in a directory', async (t) => {
+  const relay = new Relay(tmpdir(), { _writeInterval: 1 })
+  const address = await relay.listen()
+
+  const keyPair = createKeyPair(ZERO_SEED)
+
+  const LESSER_ID = '8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewn'
+  const GREATER_ID = '9pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo'
+
+  for (let i = 0; i < 10; i++) {
+    const path = '/' + ZERO_ID + `/dir/subdir/foo${i}.txt`
+    const content = Buffer.from('foo content')
+    const record = await Record.create(keyPair, path, content, { timestamp: 1000 })
+
+    const bytes = record.serialize()
+
+    await relay._recordsDB.put(path, bytes)
+
+    /// False positive subdirectory
+    await relay._recordsDB.put('/' + ZERO_ID + `/dir/wrong/foo${i}.txt`, bytes)
+
+    /// False positive IDs
+    await relay._recordsDB.put('/' + LESSER_ID + `/dir/subdir/foo${i}.txt`, bytes)
+    await relay._recordsDB.put('/' + GREATER_ID + `/dir/subdir/foo${i}.txt`, bytes)
+  }
+
+  const headers = {
+    [HEADERS.CONTENT_TYPE]: 'application/octet-stream'
+  }
+
+  const response = await fetch(address + '/' + ZERO_ID + '/dir/subdir/?something', {
+    method: 'GET',
+    headers
+  })
+
+  const keys = await response.text()
+
+  const list = keys.split('\n')
+
+  t.is(list.length, 10)
+
+  t.alike(list, [
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo0.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo1.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo2.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo3.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo4.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo5.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo6.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo7.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo8.txt',
+    '/8pinxxgqs41n4aididenw5apqp1urfmzdztr8jt4abrkdn435ewo/dir/subdir/foo9.txt'
+  ])
+
+  relay.close()
+})
+
 function tmpdir () {
   return path.join(os.tmpdir(), Math.random().toString(16).slice(2))
 }

types/lib/client/index.d.ts

@@ -100,6 +100,34 @@ declare class Client {
      * @returns {() => void}
      */
     subscribe(path: string, onupdate: (value: Uint8Array | null) => any): () => void;
+    /**
+     * Returns a list of items in a directory.
+     *
+     * @param {string} directory
+     * @param {object} [opts]
+     * @param {boolean} [opts.skipCache] - Skip the local cache and wait for the remote relay to respond with fresh data
+     *
+     * @returns {Promise<Array<string>>}
+     */
+    list(directory: string, opts?: {
+        skipCache?: boolean;
+    }): Promise<Array<string>>;
+    /**
+     * Returns a list of items in a directory from local store.
+     *
+     * @param {string} directory
+     *
+     * @returns {Promise<Array<string>>}
+     */
+    _listLocal(directory: string): Promise<Array<string>>;
+    /**
+     * Returns a list of items in a directory from the relay.
+     *
+     * @param {string} directory
+     *
+     * @returns {Promise<Array<string>>}
+     */
+    _listRelay(directory: string): Promise<Array<string>>;
     /**
        * Return a url that can be shared by others to acess a file.
        *

types/lib/server/index.d.ts

@@ -1,4 +1,5 @@
 /// <reference types="node" />
+/// <reference types="node" />
 export = Relay;
 declare class Relay {
     static SERVER_SIDE_RECORDS_METADATA: string;
@@ -105,6 +106,25 @@ declare class Relay {
      * @param {http.ServerResponse} res
      */
     _SUBSCRIBE(req: http.IncomingMessage, res: http.ServerResponse): Promise<void>;
+    /**
+     * @param {string} id
+     * @param {string} directory
+     * @param {object} [options]
+     * @param {number} [options.limit]
+     * @param {number} [options.offset]
+     *
+     * @returns {lmdb.RangeIterable}
+     */
+    _searchDirectory(id: string, directory: string, options?: {
+        limit?: number;
+        offset?: number;
+    }): lmdb.RangeIterable<any>;
+    /**
+     * @param {http.IncomingMessage} _req
+     * @param {http.ServerResponse} res
+     * @param {URL} url
+     */
+    _QUERY(_req: http.IncomingMessage, res: http.ServerResponse, url: URL): Promise<void>;
     /**
      * Health check endpoint to provide server metrics.
      *
@@ -121,4 +141,6 @@ declare class Relay {
 import http = require("http");
 import Record = require("../record.js");
 import Prometheus = require("prom-client");
+import lmdb = require("lmdb");
+import { URL } from "url";
 //# sourceMappingURL=index.d.ts.map