fix(rune): serialize response

Author: dasein108

docs/backend.md

@@ -1,35 +1,32 @@
-# Backend Architecture
-
-> TODO: update
-
-# Cyb local backend
+# CYB local backend(in-browser)
 
 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)"];
+        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;
 ```
 
 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.
 
-## Db
+## Db layer
 
-Db worker is pretty simple it it's host only local relational-graph-vector database - [[cozo]].
+Db worker is pretty simple it it's host only local relational-graph-vector database - [[cozo]]. It's represented with DbApi in frontend and backend layers.
 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;
+        dbApi["dbApi"]--odb_meta_orm;
         subgraph rune["cozo db"]
             db_meta_orm[["meta orm"]]-.->db;
         end
 ```
 
-### Entities
+### Db entities
 
 - brain:
   - particles
@@ -45,7 +42,7 @@ graph TD;
   - config
   - queue messages
 
-## Backend
+## Backend layer
 
 Backend worker is more complicated it contains significant elements of cyb architecture:
 

src/services/CozoDb/cozoDb.ts

@@ -16,6 +16,7 @@ import { clearIndexedDBStore, toListOfObjects } from './utils';
 import { createCozoDbCommandFactory } from './cozoDbCommandFactory';
 
 import initializeScript from './migrations/schema.cozo';
+import { fetchInitialEmbeddings } from './migrations/migrations';
 
 export const DB_NAME = 'cyb-cozo-idb';
 
@@ -25,6 +26,8 @@ export const DB_VERSION = 1.2;
 
 type OnWrite = (writesCount: number) => void;
 
+let shouldInitialize = false;
+
 function createCozoDb() {
   let db: CozoDb | undefined;
 
@@ -82,7 +85,7 @@ function createCozoDb() {
   const initDbSchema = async (): Promise<void> => {
     let relations = await getRelations();
 
-    const shouldInitialize = relations.length === 0;
+    shouldInitialize = relations.length === 0;
     if (shouldInitialize) {
       cyblogCh.info('CozoDb: apply DB schema', initializeScript);
       const result = await runCommand(initializeScript);
@@ -107,6 +110,10 @@ function createCozoDb() {
     if (shouldInitialize) {
       // if initialized set initial version
       await setDbVersion(DB_VERSION);
+      await fetchInitialEmbeddings(async (items: Partial<DbEntity>[]) => {
+        console.log(' [initial]save initial particles...');
+        await put('sync_queue', items);
+      });
     }
   };
 

src/services/CozoDb/migrations/migrations.ts

@@ -1,4 +1,28 @@
+import { getRelevance } from 'src/utils/search/utils';
 import { DB_VERSION, type CybCozoDb } from '../cozoDb';
+import { DbEntity, SyncQueueJobType } from '../types/entities';
+import { QueuePriority } from 'src/services/QueueManager/types';
+import { SyncQueueDto } from '../types/dto';
+import { dtoListToEntity } from 'src/utils/dto';
+import { ParticleCid } from 'src/types/base';
+
+export const fetchInitialEmbeddings = async (
+  saveSyncQueue: (syncItems: Partial<DbEntity>[]) => Promise<void>
+) => {
+  console.log(' [initial]fetch initial particles...');
+  const relevancePaticles = await getRelevance(0, 400);
+
+  const items = relevancePaticles.result.map(
+    ({ particle }: { particle: ParticleCid }) => ({
+      id: particle,
+      data: '',
+      jobType: SyncQueueJobType.particle,
+      priority: QueuePriority.LOW,
+    })
+  ) as SyncQueueDto[];
+
+  await saveSyncQueue(dtoListToEntity(items));
+};
 
 const migrate = async (db: CybCozoDb) => {
   const version = await db.getDbVersion();

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

@@ -55,13 +55,17 @@ class BackendQueueChannelListener {
   }
 
   private async saveParticles(content: IPFSContent) {
-    const dbApi = await this.getDeffredDbApi();
-    const entity = mapParticleToEntity(content);
-    const result = await dbApi.putParticles(entity);
-    if (result.ok) {
-      await enqueueParticleEmbeddingMaybe(content);
+    try {
+      const dbApi = await this.getDeffredDbApi();
+      const entity = mapParticleToEntity(content);
+      const result = await dbApi.putParticles(entity);
+      if (result.ok) {
+        await enqueueParticleEmbeddingMaybe(content);
+      }
+    } catch (e) {
+      console.log('---saveParticle e', content, e);
+      throw e;
     }
-    // console.log('---saveParticles done', content);
   }
 
   private async enquueSync(data: SyncQueueItem | SyncQueueItem[]) {

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

@@ -36,6 +36,7 @@ import { SyncQueueItem } from './types';
 import { MAX_DATABASE_PUT_SIZE } from '../consts';
 
 import DbApi from '../../../DbApi/DbApi';
+import { PATTERN_COSMOS, PATTERN_CYBER } from 'src/constants/patterns';
 
 const QUEUE_BATCH_SIZE = 100;
 
@@ -60,7 +61,11 @@ export const getTextContentIfShouldEmbed = async (
 ) => {
   const [contentType, data] = await getContentToEmbed(content);
 
-  const shouldEmbed = contentType === 'text' && !!data;
+  let shouldEmbed = contentType === 'text' && !!data;
+
+  shouldEmbed =
+    shouldEmbed &&
+    (!data!.match(PATTERN_COSMOS) || !data!.match(PATTERN_CYBER));
 
   return shouldEmbed ? data : undefined;
 };

src/services/scripting/engine.ts

@@ -29,6 +29,7 @@ import {
 } from './types';
 
 import runtimeScript from './rune/runtime.rn';
+import { deserializeString, serializeString } from 'src/utils/string';
 
 const compileConfig = {
   budget: 1_000_000,
@@ -232,16 +233,23 @@ function enigine(): RuneEngine {
     const { cid, contentType, content } = params;
     const output = await run(script, {
       funcName: 'personal_processor',
-      funcParams: [cid, contentType, content], //params as EntrypointParams,
+      funcParams: [cid, contentType, serializeString(content || '')], //params as EntrypointParams,
     });
 
-    if (output.result.action === 'error') {
+    const { action, content: outputContent } = output.result;
+
+    if (action === 'error') {
       console.error(
         `[rune].personalProcessor error: ${params.cid}`,
         params,
         output
       );
     }
+
+    if (outputContent) {
+      return { ...output.result, content: deserializeString(outputContent) };
+    }
+
     return output.result;
   };
 

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

@@ -80,8 +80,8 @@ pub async fn personal_processor(cid, content_type, content) {
 
     // example of content modification
     // replaces cyber with cyber❤
-    let highlight_text = "cyber";
-    let highlight_with = "❤";
+    let highlight_text = "cyber ";
+    let highlight_with = "❤ ";
 
     if content.contains(highlight_text) {
         cyb::log(`Update ${cid} content, highlight ${highlight_text}${highlight_with}`);