a

lib/index.d.ts

@@ -41,6 +41,7 @@ export declare class BankClient {
         timestamp: string | null;
         hash: string;
     };
+    runAgent(address: string, topic: string, storage: Storage, itemProcessCallback: (item: ContentItem | ContactItem) => Promise<any>): Promise<void>;
     private connectWebsocket;
     private getPriv;
     private makePlaintextPayload;

lib/index.js

@@ -341,6 +341,41 @@ class BankClient {
         }
         return { type, timestamp, hash };
     }
+    runAgent(address, topic, storage, itemProcessCallback) {
+        return __awaiter(this, void 0, void 0, function* () {
+            yield this.subscribePrivate(address, topic, () => {
+                // console.log('websocket connected');
+            }, (msg) => __awaiter(this, void 0, void 0, function* () {
+                console.log('Got new item notification', msg);
+                yield agentUpdate();
+            }));
+            setTimeout(() => __awaiter(this, void 0, void 0, function* () {
+                yield agentUpdate();
+            }), 0);
+            const agentUpdate = () => __awaiter(this, void 0, void 0, function* () {
+                const agentConfig = (yield storage.get('config')) || {};
+                const items = yield this.retrievePrivate(address, topic);
+                const itemsList = items.split(',').filter((x) => x.trim() !== '');
+                for (const itemId of itemsList) {
+                    const processed = agentConfig.processed || [];
+                    if (processed.includes(itemId)) {
+                        continue;
+                    }
+                    try {
+                        const item = yield this.getContentItemByHash(itemId);
+                        yield itemProcessCallback(item);
+                        processed.push(itemId);
+                        agentConfig.processed = processed;
+                        yield storage.set('config', agentConfig);
+                    }
+                    catch (e) {
+                        console.error('error processing item', itemId, e);
+                    }
+                }
+                console.log('Finished checking for items');
+            });
+        });
+    }
     connectWebsocket(peerAddr, topic, connectCallback, messageCallback) {
         return new Promise((resolve, reject) => __awaiter(this, void 0, void 0, function* () {
             const nonce = yield this.getNonce();

src/index.ts

@@ -309,28 +309,69 @@ export class BankClient {
       return items;
     }
 
-    public parseItemHash(itemHash: string) {
-      let type: string | null = null;
-      let timestamp: string | null = null;
-      let hash: string | null = null;
+  public parseItemHash(itemHash: string) {
+    let type: string | null = null;
+    let timestamp: string | null = null;
+    let hash: string | null = null;
 
-      if (itemHash.startsWith('/ipfs/')) {
-        itemHash = itemHash.substring(6);
-      }
-      const matched = itemHash.match(/^([0-9]*)_(..)_(.*)$/);
-      if (matched) {
-        timestamp = matched[1];
-        type = matched[2];
-        hash = matched[3];
-      }
-      if (!type) {
-        type = 'CO';
+    if (itemHash.startsWith('/ipfs/')) {
+      itemHash = itemHash.substring(6);
+    }
+    const matched = itemHash.match(/^([0-9]*)_(..)_(.*)$/);
+    if (matched) {
+      timestamp = matched[1];
+      type = matched[2];
+      hash = matched[3];
+    }
+    if (!type) {
+      type = 'CO';
+    }
+    if (!hash) {
+      hash = itemHash;
+    }
+    return { type, timestamp, hash };
+  }
+
+  public async runAgent(address: string, topic: string, storage: Storage, itemProcessCallback: (item: ContentItem | ContactItem) => Promise<any>) {
+    await this.subscribePrivate(
+      address,
+      topic,
+      () => {
+        // console.log('websocket connected');
+      },
+      async (msg) => {
+        console.log('Got new item notification', msg);
+        await agentUpdate();
       }
-      if (!hash) {
-        hash = itemHash;
+    );
+
+    setTimeout(async () => {
+      await agentUpdate();
+    }, 0);
+
+    const agentUpdate = async () => {
+      const agentConfig = (await storage.get('config')) || {};
+      const items = await this.retrievePrivate(address, topic);
+      const itemsList = items.split(',').filter((x) => x.trim() !== '');
+
+      for (const itemId of itemsList) {
+        const processed = agentConfig.processed || [];
+        if (processed.includes(itemId)) {
+          continue;
+        }
+        try {
+          const item = await this.getContentItemByHash(itemId);
+          await itemProcessCallback(item);
+          processed.push(itemId);
+          agentConfig.processed = processed;
+          await storage.set('config', agentConfig);
+        } catch (e) {
+          console.error('error processing item', itemId, e);
+        }
       }
-      return { type, timestamp, hash };
+      console.log('Finished checking for items');
     }
+  }
 
   private connectWebsocket(peerAddr: string, topic: string, connectCallback: () => void, messageCallback: (data: any) => void) {
     return new Promise(async (resolve, reject) => {
@@ -378,40 +419,40 @@ export class BankClient {
     
   }
 
-    private getPriv(): RsaPrivateKey {
-      if (!this.privateKey) {
-        throw new Error('missing private key');
-      }
-      return this.privateKey;
+  private getPriv(): RsaPrivateKey {
+    if (!this.privateKey) {
+      throw new Error('missing private key');
     }
+    return this.privateKey;
+  }
 
-    private makePlaintextPayload(message: string) {
-      const messageBytes = Buffer.from(message, 'utf-8');
-  
-      return new Promise(async (resolve, reject) => {
-        await this.bootstrap();
-        this.privateKey.sign(messageBytes, async (signErr, signatureBytes) => {
-          if (signErr) {
-            reject(signErr);
+  private makePlaintextPayload(message: string) {
+    const messageBytes = Buffer.from(message, 'utf-8');
+
+    return new Promise(async (resolve, reject) => {
+      await this.bootstrap();
+      this.privateKey.sign(messageBytes, async (signErr, signatureBytes) => {
+        if (signErr) {
+          reject(signErr);
+          return;
+        }
+        const publicDERBytes = this.privateKey.public.bytes;
+        this.privateKey.id((idErr, pubHash) => {
+          if (idErr) {
+            reject(idErr);
             return;
           }
-          const publicDERBytes = this.privateKey.public.bytes;
-          this.privateKey.id((idErr, pubHash) => {
-            if (idErr) {
-              reject(idErr);
-              return;
-            }
-            const result = {
-              date: new Date().toISOString(),
-              msg: encodeHex(messageBytes),
-              pub: encodeHex(publicDERBytes),
-              pubHash,
-              sig: encodeHex(signatureBytes),
-            };
-            // console.log('result', result, signatureBytes);
-            resolve(result);
-          });
+          const result = {
+            date: new Date().toISOString(),
+            msg: encodeHex(messageBytes),
+            pub: encodeHex(publicDERBytes),
+            pubHash,
+            sig: encodeHex(signatureBytes),
+          };
+          // console.log('result', result, signatureBytes);
+          resolve(result);
         });
       });
-    }
+    });
+  }
 }