ANS Caching and Verification (#72)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@arweave/gateway",
-  "version": "0.9.1",
+  "version": "0.10.1",
   "main": "dist/src/Gateway.js",
   "repository": "git@github.com:ArweaveTeam/gateway.git",
   "author": "Arweave <hello@arweave.org>",
@@ -15,6 +15,7 @@
     "dev:restart": "npm run migrate:down && npm run migrate:latest && npm run dev:start",
     "import:transaction": "npm run dev:build && node dist/src/import/transaction.import.js",
     "import:tags": "npm run dev:build && node dist/src/import/tags.import.ts",
+    "verify:ans": "npm run dev:build && node dist/src/caching/ans.verify.js",
     "verify:prune": "npm run dev:build && node dist/src/verify/prune.verify.js",
     "verify:transaction": "npm run dev:build && node dist/src/verify/transaction.verify.js",
     "migrate:down": "knex migrate:down",

src/caching/ans.caching.ts

@@ -3,6 +3,7 @@ import {DataItemJson} from 'arweave-bundles';
 import {ansBundles} from '../utility/ans.utility';
 import {getDataFromChunks} from '../query/node.query';
 import {tagToUTF8} from '../query/transaction.query';
+import {cacheANSEntries} from './ans.entry.caching';
 
 export async function streamAndCacheAns(id: string): Promise<boolean> {
   try {
@@ -24,12 +25,14 @@ export async function streamAndCacheAns(id: string): Promise<boolean> {
         signature: ansTx.signature,
         tags: tagToUTF8(ansTx.tags),
       };
-      
+
       ansTxsConverted.push(newAnsTx);
     }
 
     write(`${process.cwd()}/cache/tx/${id}`, JSON.stringify(ansTxsConverted, null, 2));
 
+    await cacheANSEntries(ansTxs);
+
     return true;
   } catch (error) {
     remove(`${process.cwd()}/cache/tx/${id}`);

src/caching/ans.entry.caching.ts

@@ -0,0 +1,17 @@
+import {exists, write} from 'fs-jetpack';
+import {DataItemJson} from 'arweave-bundles';
+import {b64UrlToBuffer} from '../utility/encoding.utility';
+
+export async function cacheANSEntries(entries: Array<DataItemJson>) {
+    for (let i = 0; i < entries.length; i++) {
+        const entry = entries[i];
+        const id = entry.id;
+        const data = entry.data;
+
+        const bufferData = Buffer.from(b64UrlToBuffer(data));
+
+        if (exists(`${process.cwd()}/cache/tx/${id}`) === false) {
+            write(`${process.cwd()}/cache/tx/${id}`, bufferData.toString('utf-8'));
+        }
+    }
+}

src/caching/ans.verify.ts

@@ -0,0 +1,40 @@
+import 'colors';
+import {connection} from '../database/connection.database';
+import {toB64url} from '../query/transaction.query';
+import {cacheAnsFile} from './file.caching';
+
+const name = toB64url('Bundle-Type');
+const value = toB64url('ANS-102');
+
+export async function ansVerify(offset: number = 0) {
+    const query = await connection
+      .queryBuilder()
+      .select('*')
+      .from('tags')
+      .where('name', name)
+      .where('value', value)
+      .orderByRaw('created_at ASC')
+      .limit(10)
+      .offset(offset);
+
+    for (let i = 0; i < query.length; i++) {
+        const item = query[i];
+
+        try {
+            await cacheAnsFile(item.tx_id);
+        } catch (error) {
+            console.log(`Could not cache ${item.tx_id}`.red);
+        }
+    }
+
+    console.log(`Successfully cached ANS bundles at offset ${offset}`.green);
+
+    if (query.length === 0) {
+        console.log(`Successfully cached all ANS bundles. Good work!`.green.bold);
+        process.exit();
+    } else {
+        ansVerify(offset + 10);
+    }
+}
+
+(async () => await ansVerify())();

src/database/sync.database.ts

@@ -15,6 +15,7 @@ import {transaction, tagValue, Tag} from '../query/transaction.query';
 import {getDataFromChunks} from '../query/node.query';
 import {importBlocks, importTransactions, importTags} from './import.database';
 import {DatabaseTag} from './transaction.database';
+import {cacheANSEntries} from '../caching/ans.entry.caching';
 
 config();
 mkdir('snapshot');
@@ -202,6 +203,7 @@ export async function processAns(id: string, height: number, retry: boolean = tr
     const ansPayload = await getDataFromChunks(id);
     const ansTxs = await ansBundles.unbundleData(ansPayload.toString('utf-8'));
 
+    await cacheANSEntries(ansTxs);
     await processANSTransaction(ansTxs, height);
   } catch (error) {
     if (retry) {

src/query/node.query.ts

@@ -62,7 +62,7 @@ export async function getDataFromChunks(id: string, retry: boolean = true): Prom
   } catch (error) {
     if (retry) {
       console.error(`error retrieving data from ${id}, please note that this may be a cancelled transaction`.red.bold);
-      return await getDataFromChunks(id, true);
+      return await getDataFromChunks(id, false);
     } else {
       throw error;
     }

src/route/data.route.ts

@@ -32,23 +32,27 @@ export async function dataRoute(req: Request, res: Response) {
     }
   }
 
-  const metadata = await getTransaction(transaction);
-  const contentType = tagValue(metadata.tags, 'Content-Type');
-  const ans102 = tagValue(metadata.tags, 'Bundle-Type') === 'ANS-102';
-
-  res.setHeader('content-type', contentType);
-
-  if (ans102) {
-    await cacheAnsFile(transaction);
-  } else {
-    await cacheFile(transaction);
-  }  
+  try {
+    const metadata = await getTransaction(transaction);
+    const contentType = tagValue(metadata.tags, 'Content-Type');
+    const ans102 = tagValue(metadata.tags, 'Bundle-Type') === 'ANS-102';
+
+    res.setHeader('content-type', contentType);
+
+    if (ans102) {
+      await cacheAnsFile(transaction);
+    } else {
+      await cacheFile(transaction);
+    }  
+  } catch (error) {
+
+  }
 
   if (exists(`${process.cwd()}/cache/tx/${transaction}`)) {
     res.status(200);
     res.sendFile(`${process.cwd()}/cache/tx/${transaction}`);
   } else {
     res.status(500);
-    res.json({status: 'ERROR', message: 'Could not retrieve tx, please retry'});
+    res.json({status: 'ERROR', message: 'Could not retrieve transaction'});
   }
 }