Merge pull request #94 from ssb-ngi-pointer/just-live

support and(live()) operations

Author: arj03

operators.js

@@ -7,42 +7,28 @@ const pullAwaitable = require('pull-awaitable')
 const cat = require('pull-cat')
 const { safeFilename } = require('./files')
 
-function query(...cbs) {
-  let res = cbs[0]
-  for (let i = 1, n = cbs.length; i < n; i++) if (cbs[i]) res = cbs[i](res)
-  return res
-}
+//#region Helper functions and util operators
 
-function fromDB(db) {
-  return {
-    meta: { db },
+function copyMeta(orig, dest) {
+  if (orig.meta) {
+    dest.meta = orig.meta
   }
 }
 
-function toBufferOrFalsy(value) {
-  if (!value) return value
-  return Buffer.isBuffer(value) ? value : Buffer.from(value)
-}
-
-function seqs(values) {
-  return {
-    type: 'SEQS',
-    seqs: values,
-  }
+function updateMeta(orig, key, value) {
+  const res = Object.assign({}, orig)
+  res.meta[key] = value
+  return res
 }
 
-function liveSeqs(pullStream) {
-  return {
-    type: 'LIVESEQS',
-    stream: pullStream,
-  }
+function extractMeta(orig) {
+  const meta = orig.meta
+  return meta
 }
 
-function offsets(values) {
-  return {
-    type: 'OFFSETS',
-    offsets: values,
-  }
+function toBufferOrFalsy(value) {
+  if (!value) return value
+  return Buffer.isBuffer(value) ? value : Buffer.from(value)
 }
 
 function seekFromDesc(desc) {
@@ -58,6 +44,40 @@ function seekFromDesc(desc) {
   }
 }
 
+function query(...cbs) {
+  let res = cbs[0]
+  for (let i = 1, n = cbs.length; i < n; i++) if (cbs[i]) res = cbs[i](res)
+  return res
+}
+
+function debug() {
+  return (ops) => {
+    const meta = JSON.stringify(ops.meta, (key, val) =>
+      key === 'db' ? void 0 : val
+    )
+    console.log(
+      'debug',
+      JSON.stringify(
+        ops,
+        (key, val) => {
+          if (key === 'meta') return void 0
+          else if (key === 'task' && typeof val === 'function')
+            return '[Function]'
+          else if (key === 'value' && val.type === 'Buffer')
+            return Buffer.from(val.data).toString()
+          else return val
+        },
+        2
+      ),
+      meta === '{}' ? '' : 'meta: ' + meta
+    )
+    return ops
+  }
+}
+
+//#endregion
+//#region "Unit operators": they create objects that JITDB interprets
+
 function slowEqual(seekDesc, target, opts) {
   opts = opts || {}
   const seek = seekFromDesc(seekDesc)
@@ -150,13 +170,6 @@ function includes(seek, target, opts) {
   }
 }
 
-function not(ops) {
-  return {
-    type: 'NOT',
-    data: [ops],
-  }
-}
-
 function gt(value, indexName) {
   if (typeof value !== 'number') throw new Error('gt() needs a number arg')
   return {
@@ -201,62 +214,51 @@ function lte(value, indexName) {
   }
 }
 
-function deferred(task) {
+function seqs(values) {
   return {
-    type: 'DEFERRED',
-    task,
+    type: 'SEQS',
+    seqs: values,
   }
 }
 
-function debug() {
-  return (ops) => {
-    const meta = JSON.stringify(ops.meta, (key, val) =>
-      key === 'db' ? void 0 : val
-    )
-    console.log(
-      'debug',
-      JSON.stringify(
-        ops,
-        (key, val) => {
-          if (key === 'meta') return void 0
-          else if (key === 'task' && typeof val === 'function')
-            return '[Function]'
-          else if (key === 'value' && val.type === 'Buffer')
-            return Buffer.from(val.data).toString()
-          else return val
-        },
-        2
-      ),
-      meta === '{}' ? '' : 'meta: ' + meta
-    )
-    return ops
+function liveSeqs(pullStream) {
+  return {
+    type: 'LIVESEQS',
+    stream: pullStream,
   }
 }
 
-function copyMeta(orig, dest) {
-  if (orig.meta) {
-    dest.meta = orig.meta
+function offsets(values) {
+  return {
+    type: 'OFFSETS',
+    offsets: values,
   }
 }
 
-function updateMeta(orig, key, value) {
-  const res = Object.assign({}, orig)
-  res.meta[key] = value
-  return res
+function deferred(task) {
+  return {
+    type: 'DEFERRED',
+    task,
+  }
 }
 
-function extractMeta(orig) {
-  const meta = orig.meta
-  return meta
+//#endregion
+//#region "Combinator operators": they build composite operations
+
+function not(ops) {
+  return {
+    type: 'NOT',
+    data: [ops],
+  }
 }
 
 function and(...args) {
-  const rhs = args
-    .map((arg) => (typeof arg === 'function' ? arg() : arg))
-    .filter((arg) => !!arg)
-  return (ops) => {
+  return (ops, isSpecialOps) => {
+    const rhs = args
+      .map((arg) => (typeof arg === 'function' ? arg(ops, true) : arg))
+      .filter((arg) => !!arg)
     const res =
-      ops && ops.type
+      ops && ops.type && !isSpecialOps
         ? {
             type: 'AND',
             data: [ops, ...rhs],
@@ -273,12 +275,12 @@ function and(...args) {
 }
 
 function or(...args) {
-  const rhs = args
-    .map((arg) => (typeof arg === 'function' ? arg() : arg))
-    .filter((arg) => !!arg)
-  return (ops) => {
+  return (ops, isSpecialOps) => {
+    const rhs = args
+      .map((arg) => (typeof arg === 'function' ? arg(ops, true) : arg))
+      .filter((arg) => !!arg)
     const res =
-      ops && ops.type
+      ops && ops.type && !isSpecialOps
         ? {
             type: 'OR',
             data: [ops, ...rhs],
@@ -294,6 +296,15 @@ function or(...args) {
   }
 }
 
+//#endregion
+//#region "Special operators": they only update meta
+
+function fromDB(db) {
+  return {
+    meta: { db },
+  }
+}
+
 function live(opts) {
   if (opts && opts.old) return (ops) => updateMeta(ops, 'live', 'liveAndOld')
   else return (ops) => updateMeta(ops, 'live', 'liveOnly')
@@ -311,6 +322,9 @@ function paginate(pageSize) {
   return (ops) => updateMeta(ops, 'pageSize', pageSize)
 }
 
+//#endregion
+//#region "Consumer operators": they execute the query tree
+
 async function executeDeferredOps(ops, meta) {
   // Collect all deferred tasks and their object-traversal paths
   const allDeferred = []
@@ -420,6 +434,8 @@ function toAsyncIter() {
   }
 }
 
+//#endregion
+
 module.exports = {
   fromDB,
   query,

test/operators.js

@@ -1033,6 +1033,31 @@ prepareAndRunTest('support live operations', dir, (t, db, raf) => {
   })
 })
 
+prepareAndRunTest('support live inside and', dir, (t, db, raf) => {
+  const msg = { type: 'post', text: 'Testing!' }
+  let state = validate.initial()
+  state = validate.appendNew(state, null, alice, msg, Date.now())
+  state = validate.appendNew(state, null, bob, msg, Date.now() + 1)
+
+  addMsg(state.queue[0].value, raf, (e1, msg1) => {
+    let i = 0
+    query(
+      fromDB(db),
+      and(live({ old: true }), slowEqual('value.content.type', 'post')),
+      toPullStream(),
+      pull.drain((msg) => {
+        if (i++ == 0) {
+          t.equal(msg.value.author, alice.id)
+          addMsg(state.queue[1].value, raf, (e2, msg2) => {})
+        } else {
+          t.equal(msg.value.author, bob.id)
+          t.end()
+        }
+      })
+    )
+  })
+})
+
 prepareAndRunTest('support live with not', dir, (t, db, raf) => {
   const msg = { type: 'post', text: 'Testing!' }
   let state = validate.initial()