feat(rune): documentation, serialization

Author: dasein108

docs/backend.md

@@ -2,52 +2,196 @@
 
 > TODO: update
 
+# Cyb local backend
+
+Cyb plays singinficat role in cyber infrastructure. The app reconstruct self-sufficient backend+frontend pattern inside the browser.
+In big view app consist from 3 parts:
+
 ```mermaid
 graph TD;
+        App["Frontend\n(main thread)"]-.proxy.->Backend["Backend\n(shared worker)"];
+        App-.proxy.->Db["Graph Db\n(shared worker)"];
+        Backend-.proxy.->Db;
+        App<-.message\nchannels.->Backend;
+```
 
-    subgraph frontend["frontend(main thread)"]
-        App["Frontend"]-->Hook["useBackend()"];
-        Hook-->methods("startSync()\nloadIpfs()\n...\nisReady\nipfsError\n...");
-        Hook-.broadcast channel\n(any worker).->reducer["redux(state)"]
-        Hook-.save history from app.->defferedDbApiFront[/"DefferedDbApi(proxy)"/]
-        Hook--osenseApi["senseApi"];
-        Hook--oipfsApiFront[/"ipfsApi(proxy)"/];
-        senseApi--odbApi[/"dbApi(proxy)"/];
-    end
+To reduce overload of main thread we have created 2 separate shared workers, where all the stuff is hosted. Bi-interraction between all layers occurs using proxy(comlink abstraction) or directly using broadcast channels.
 
-    dbApi<-.message channel.->dbWorker["dbApi"];
-    subgraph dbWorkerGraph["cyb~db(worker)"]
-        dbWorker<-.bindings(webApi).->cozodb{{"CozoDb(wasm)"}}
-    end
+## Db
 
-    defferedDbApiFront-.->defferedDbApi;
-    ipfsApiFront<-.->ipfsApi;
-    subgraph backgroundWorker["cyb~backend(worker)"]
-        subgraph sync["sync service"]
-            ipfsNode["ipfs node"];
-            links;
-            transactions;
+Db worker is pretty simple it it's host only local relational-graph-vector database - [[cozo]].
+Cozo provide bazing fast access to brain and ipfs data in relational form and also in vector format, processing by [ml]embedder.
+
+```mermaid
+graph TD;
+        subgraph rune["cozo db"]
+            db_meta_orm[["meta orm"]]-.->db;
         end
-        sync--oparticleResolver[["Particle resolver"]]
-        particleResolver--oqueue;
-        particleResolver--odbProxyWorker;
-        sync--oipfsApi;
-        sync--odbProxyWorker[/"dbApi(proxy)"/];
-        defferedDbApi[["defferedDbApi"]]-->dbProxyWorker;
-        queue-->defferedDbApi;
-        ipfsApi--oqueue[["queue"]];
-        ipfsApi--onode["node"];
-        queue--balancer-->node;
-        node--embedded-->helia;
-        node--rpc-->kubo;
-        node--embedded-->js-ipfs;
+```
+
+### Entities
+
+- brain:
+  - particles
+    - embeddings
+  - links
+  - transactions
+  - community
+- sense:
+
+  - sync items + update status
+
+- system:
+  - config
+  - queue messages
+
+## Backend
+
+Backend worker is more complicated it contains significant elements of cyb architecture:
+
+```mermaid
+graph TD;
+    subgraph Backend["backend(shared worker)"]
+
         subgraph ipfs["ipfs implementations"]
             helia;
             kubo;
             js-ipfs;
         end
 
-        dbProxyWorker<-.message channel.->dbWorker
+        subgraph queues["message brokers"]
+            ipfs_queue["ipfs load balancer"];
+            queue["data processing queue aka bus"];
+        end
+
+        subgraph rune["rune"]
+            vm["virtual machine"]--ovm_bingen{{"cyb bindings"}};
+        end
+
+        subgraph sense["sense"]
+            link_sync["link sync"];
+            msg_sync["message sync"];
+            swarm_sync["swarm sync"];
+        end
+
+        subgraph ml["ML transformers"]
+            feature_extractor["embedder"];
+        end
+
+    end
+```
+
+### Ipfs module
+
+Represented with IpfsApi at frontend layer, but also have direct access for some edge cases
+
+- Uses module that encapsulate different Ipfs implementations(kubo, helia, js-ipfs(obsolete))
+  - cache content(local storage & cozo)
+  - preserve redundancy
+- Ipfs queue, process all requests to ipfs, prioritize, cancel non-actual requests and organize content pipeline
+  - responsible for:
+    - ipfs load balancing(limit of requests)
+    - request prioritizing(actual requests first)
+    - fault processing(switch fetch policy)
+    - post processing(**inline rune vm** into pipeline)
+
+```mermaid
+graph LR
+user(ipfsApi\nenqueue particle) --> q[["queue\n(balancer)"]] --> node[/"ipfs"/] -- found --> rune[rune vm] -- mutation | content --> cache["cache"] --> app(app\ncontent)
+node -. not found\n(retry | error) .-> q
+```
+
+## Bus
+
+Represented with some helpers and used for cases when blaancer is needed, some services not initialized yet(deffered actions), or long calculations is requered(ml inference, ipfs requests):
+
+- particle, request ipfs, save; calc embedding
+- link, deffered save
+- message persistence is protected by db store
+
+```mermaid
+graph TD;
+        sender{{"enqueue(...)"}} -.message bus.-> bus
+        subgraph task["task manager"]
+            bus[["queue listener"]];
+
+            bus-.task.->db("store\ndata")--odb1["dbApi"];
+            bus-.task.->ml("calculate\nembedding")--oml1["mlApi"];
+            bus-.task.->ipfs("request ipfs\nlow-priority")--oi["ipfsApi"]
+        end
+```
+
+## Sense
+
+Represented by SenseApi + subscription to broadcast channel at fronted layer. Provide continious update of cyberlinks related to my brain and my swarm, recieving on chain messages etc.:
+
+- Particles service (pooling)
+- Transactions service (pooling + websocket)
+- My friends service (pooling)
+- Ipfs service(pooling)
+
+All data and update status is stored into db, when some new data is recieved that triggers notification for frontendю
+
+```mermaid
+graph TD;
+    db[["dbApi"]];
+    bus[["particle queue"]];
+
+    subgraph sense["sync service"]
+        notification("notification service")
+
+        particles[["particle service"]]--onotification;
+        transactions[["transaction service"]]--onotification;
+        myfriend[["my friends service"]]--onotification;
+
+        particles -.loop.-> particles;
+        transactions -.loop.-> transactions;
+        myfriend -.loop.-> myfriend;
+    end
+
+
+    subgraph blockchain["blockchain"]
+        lcd[["lcd"]]
+        websockets("websockets")
+        indexer[["indexer"]]
+    end
+
+    subgraph app["frontend"]
+        redux["redux"]
+        sender{{"senseApi"}};
+    end
+
+    notification -.message.-> redux;
+    sender -.proxy.-> db;
+    sense -.proxy.-> db;
+    sense -.message.-> bus;
+    bus -.proxy.-> db;
+
+    sense <-.request\nsubscriptin.->blockchain;
+
+```
+
+## Rune
+
+Rune VM execution is pipelined thru special abstraction called entrypoints. VM have bindings to all app parts: DB, transformers, signer, blockchain api, ipfs and also includes context of the entrypoint.(see. [[scripting]] for detailed description).
+
+## ML transformers
+
+Represented my mlApi. Uses inference from local ML models hosted inside browser.
+
+- future extractor. BERT-like model to trnsform text-to-embeddings.
+
+```mermaid
+graph TD;
+    subgraph ml["transformers"]
+        embedder["embedder"];
+    end
+
+    subgraph dbApi["dbApi"]
+        db[["DB"]];
     end
+    mlApi["mlApi"];
 
+    mlApi--odb;
+    mlApi--oembedder;
 ```

src/pages/robot/Soul/Soul.tsx

@@ -104,15 +104,13 @@ function Soul() {
     outputRef.current?.scrollIntoView({ behavior: 'smooth' });
 
     addToLog([`🚧 Execute your '${funcName}'.`]);
-    console.log('---------test', code, funcName, funcParams);
     rune
       ?.run(code, {
         execute: true,
         funcName,
         funcParams,
       })
       .then((result) => {
-        console.log('---------test result', result);
         const isOk = !result.diagnosticsOutput && !result.error;
         highlightErrors(codeMirrorRef!.current, result.diagnostics, styles);
         if (!isOk) {

src/services/backend/channels/BackendQueueChannel/BackendQueueChannel.ts

@@ -57,8 +57,10 @@ class BackendQueueChannelListener {
   private async saveParticles(content: IPFSContent) {
     const dbApi = await this.getDeffredDbApi();
     const entity = mapParticleToEntity(content);
-    await dbApi.putParticles(entity);
-    await enqueueParticleEmbeddingMaybe(content);
+    const result = await dbApi.putParticles(entity);
+    if (result.ok) {
+      await enqueueParticleEmbeddingMaybe(content);
+    }
     // console.log('---saveParticles done', content);
   }
 

src/services/backend/services/DbApi/DbApi.ts

@@ -159,7 +159,7 @@ class DbApiWrapper {
     const entitites = dtoListToEntity(
       Array.isArray(particles) ? particles : [particles]
     );
-    await this.db!.executePutCommand('particle', entitites);
+    return this.db!.executePutCommand('particle', entitites);
   }
 
   public async getParticlesRaw(fields: string[]) {

src/services/backend/services/sync/services/ParticlesResolverQueue/ParticlesResolverQueue.ts

@@ -103,7 +103,6 @@ class ParticlesResolverQueue {
 
     deps.getEmbeddingInstance$?.subscribe((f) => {
       this.getEmbedding = f;
-      console.log('------PRQ getEmbedding SET', typeof f);
       // if embedding function is provided, retriger the queue
       if (this.queue.size > 0) {
         this._syncQueue$.next(this.queue);
@@ -136,23 +135,6 @@ class ParticlesResolverQueue {
         }
 
         await enqueueParticleEmbeddingMaybe(result);
-        // const [contentType, data] = await getContentToEmbed(result);
-
-        // if (contentType === 'text') {
-        //   if (this.canEmbed && data) {
-        //     this.saveEmbedding(id, data);
-        //   } else {
-        //     this.enqueue([
-        //       {
-        //         id,
-        //         data,
-        //         priority: QueuePriority.LOW,
-        //         jobType: SyncQueueJobType.embedding,
-        //       },
-        //     ]);
-        //   }
-        // }
-
         return true;
       })
       .catch(() => false);
@@ -166,7 +148,6 @@ class ParticlesResolverQueue {
         const vec = await this.getEmbedding!(text);
 
         const result = await this.db!.putEmbedding(cid, vec);
-        // console.log('------saveEmbedding ', vec, text, result.ok);
       }
 
       return true;

src/services/backend/workers/background/worker.ts

@@ -111,9 +111,9 @@ const createBackgroundWorkerApi = () => {
             const progressItem = {
               status,
               message,
-              done: ['done', 'ready', 'error'].some((s) => s === status),
+              done: ['ready', 'error'].some((s) => s === status),
             };
-            console.log('progress_callback', name, progressData);
+            // console.log('progress_callback', name, progressData);
 
             if (progress) {
               progressItem.progress = Math.round(progress);
@@ -232,7 +232,6 @@ const createBackgroundWorkerApi = () => {
   };
 
   const getEmbedding = async (text: string) => {
-    console.log('-------getEmbb', typeof mlInstances.featureExtractor);
     const output = await mlInstances.featureExtractor(text, {
       pooling: 'mean',
       normalize: true,

src/services/scripting/rune/default/particle.rn

@@ -100,6 +100,8 @@ pub async fn personal_processor(cid, content_type, content) {
         return content_result(content.replace(`${token_name}@NOW`, `Current ${token_name} price is ${json[token_name][vs_currency]} ${vs_currency}`))
     }
 
-    // anything else - pass as is
-    pass()
+    // anything else
+	content = content.replace("хуй", "🌽").replace("хуя", "🌽").replace("хуе", "🌽");
+
+    content_result(content)
 }

src/services/scripting/rune/runtime.rn

@@ -11,6 +11,7 @@ pub fn cid_result(cid) {
 }
 
 pub fn content_result(content) {
+    // #{ "content": json::to_string(content)?, "action": "content_result" }
     #{ "content": content, "action": "content_result" }
 }
 

src/utils/string.ts

@@ -9,17 +9,19 @@ export function replaceQuotes(string: string) {
 export function serializeString(input: string): string {
   return input
     .replace(/\\/g, '\\\\') // Escape backslashes
-    .replace(/"/g, '\\"') // Escape double quotes
+    .replace(/"/g, "\\''") // Escape double quotes
     .replace(/'/g, "\\'") // Escape single quotes
     .replace(/\n/g, '\\n') // Escape newlines
-    .replace(/\r/g, '\\r'); // Escape carriage returns
+    .replace(/\r/g, '\\r') // Escape carriage returns
+    .replace(/#/g, '\\!!'); // Escape  - that's comment in cozo
 }
 
 export function deserializeString(serialized: string): string {
   return serialized
     .replace(/\\r/g, '\r') // Unescape carriage returns
     .replace(/\\n/g, '\n') // Unescape newlines
     .replace(/\\'/g, "'") // Unescape single quotes
-    .replace(/\\"/g, '"') // Unescape double quotes
-    .replace(/\\\\/g, '\\'); // Unescape backslashes
+    .replace(/\\''/g, '"') // Unescape double quotes
+    .replace(/\\\\/g, '\\') // Unescape backslashes
+    .replace(/\\!!/g, '#'); // Unescape # cozo comment
 }