feat: add archiving data to nodes

backend/migrations/20241115171659-add-node-archiving-data.js

@@ -0,0 +1,53 @@
+'use strict';
+
+var dbm;
+var type;
+var seed;
+var fs = require('fs');
+var path = require('path');
+var Promise;
+
+/**
+  * We receive the dbmigrate dependency from dbmigrate initially.
+  * This enables us to not have to rely on NODE_PATH.
+  */
+exports.setup = function(options, seedLink) {
+  dbm = options.dbmigrate;
+  type = dbm.dataType;
+  seed = seedLink;
+  Promise = options.Promise;
+};
+
+exports.up = function(db) {
+  var filePath = path.join(__dirname, 'sqls', '20241115171659-add-node-archiving-data-up.sql');
+  return new Promise( function( resolve, reject ) {
+    fs.readFile(filePath, {encoding: 'utf-8'}, function(err,data){
+      if (err) return reject(err);
+      console.log('received data: ' + data);
+
+      resolve(data);
+    });
+  })
+  .then(function(data) {
+    return db.runSql(data);
+  });
+};
+
+exports.down = function(db) {
+  var filePath = path.join(__dirname, 'sqls', '20241115171659-add-node-archiving-data-down.sql');
+  return new Promise( function( resolve, reject ) {
+    fs.readFile(filePath, {encoding: 'utf-8'}, function(err,data){
+      if (err) return reject(err);
+      console.log('received data: ' + data);
+
+      resolve(data);
+    });
+  })
+  .then(function(data) {
+    return db.runSql(data);
+  });
+};
+
+exports._meta = {
+  "version": 1
+};

backend/migrations/sqls/20241115171659-add-node-archiving-data-down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE public.nodes DROP COLUMN piece_index;
+ALTER TABLE public.nodes DROP COLUMN piece_offset;

backend/migrations/sqls/20241115171659-add-node-archiving-data-up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE public.nodes ADD COLUMN piece_index INTEGER;
+ALTER TABLE public.nodes ADD COLUMN piece_offset INTEGER;

backend/package.json

@@ -12,7 +12,7 @@
     "lint": "eslint . "
   },
   "dependencies": {
-    "@autonomys/auto-dag-data": "^1.0.5",
+    "@autonomys/auto-dag-data": "^1.0.7",
     "@autonomys/auto-drive": "^1.0.5",
     "@polkadot/api": "^12.3.1",
     "@polkadot/types": "^13.0.1",
@@ -30,6 +30,7 @@
     "pg-format": "^1.0.4",
     "pizzip": "^3.1.7",
     "uuid": "^10.0.0",
+    "websocket": "^1.0.35",
     "zod": "^3.23.8"
   },
   "devDependencies": {
@@ -40,6 +41,7 @@
     "@types/pg": "^8.11.10",
     "@types/pg-format": "^1",
     "@types/uuid": "^10",
+    "@types/websocket": "^1",
     "@typescript-eslint/eslint-plugin": "^8.13.0",
     "@typescript-eslint/parser": "^8.13.0",
     "blockstore-core": "^5.0.2",

backend/src/drivers/ws.ts

@@ -0,0 +1,93 @@
+import Websocket from 'websocket'
+import { stringify } from '../utils/misc.js'
+
+type RPCMessage = {
+  jsonrpc: string
+  method: string
+  params?: unknown
+  id: number
+}
+
+export type WS = {
+  send: (message: Omit<RPCMessage, 'id'>) => Promise<unknown>
+  on: (callback: (event: unknown) => void) => void
+  off: (callback: (event: unknown) => void) => void
+}
+
+export const createWS = (endpoint: string): WS => {
+  let ws: Websocket.w3cwebsocket
+  let onMessageCallbacks: ((event: RPCMessage) => void)[] = []
+
+  const handleConnection = () => {
+    ws = new Websocket.w3cwebsocket(endpoint)
+
+    ws.onerror = (event) => {
+      const errorDetails = {
+        readyState: ws.readyState,
+        url: endpoint,
+        message: event.message || 'Unknown error',
+      }
+
+      console.error('WebSocket connection error', errorDetails)
+
+      setTimeout(() => {
+        console.log(`Reconnecting to RPC Web Socket (${endpoint})`)
+        handleConnection()
+      }, 10_000)
+    }
+
+    ws.onmessage = (event) => {
+      console.debug(`Received message from WebSocket (${endpoint})`)
+      onMessageCallbacks.forEach((callback) =>
+        callback(JSON.parse(event.data.toString())),
+      )
+    }
+
+    ws.onclose = (event) => {
+      console.log(
+        `WebSocket connection closed (${event.code}) due to ${event.reason}.`,
+      )
+    }
+  }
+
+  handleConnection()
+
+  const connected: Promise<void> = new Promise((resolve) => {
+    ws.onopen = () => {
+      console.log(`Connected to RPC Web Socket (${endpoint})`)
+      resolve()
+    }
+  })
+
+  const send = async (message: Omit<RPCMessage, 'id'>) => {
+    await connected
+
+    const id = Math.floor(Math.random() * 65546)
+    const messageWithID = { ...message, id }
+
+    return new Promise((resolve, reject) => {
+      const cb = (event: RPCMessage) => {
+        try {
+          if (event.id === id) {
+            off(cb)
+            resolve(event)
+          }
+        } catch (error) {
+          reject(error)
+        }
+      }
+      on(cb)
+
+      ws.send(stringify(messageWithID))
+    })
+  }
+
+  const on = (callback: (event: RPCMessage) => void) => {
+    onMessageCallbacks.push(callback)
+  }
+  const off = (callback: (event: RPCMessage) => void) => {
+    onMessageCallbacks = onMessageCallbacks.filter((cb) => cb !== callback)
+  }
+
+  return { send, on, off }
+}

backend/src/models/objects/object.ts

@@ -10,6 +10,7 @@ export interface ObjectInformation {
 export interface UploadStatus {
   uploadedNodes: number | null
   totalNodes: number | null
+  archivedNodes: number | null
   minimumBlockDepth: number | null
   maximumBlockDepth: number | null
 }
@@ -32,7 +33,7 @@ export type ObjectSearchResult = {
 export type ObjectSummary = {
   headCid: string
   name?: string
-  size: number
+  size: string
   owners: Owner[]
   uploadStatus: UploadStatus
 } & (
@@ -54,7 +55,7 @@ export const getObjectSummary = (object: ObjectInformation): ObjectSummary => {
         headCid: object.metadata.dataCid,
         name: object.metadata.name,
         type: object.metadata.type,
-        size: object.metadata.totalSize,
+        size: object.metadata.totalSize.toString(),
         owners: object.owners,
         children: object.metadata.children,
         uploadStatus: object.uploadStatus,
@@ -63,7 +64,7 @@ export const getObjectSummary = (object: ObjectInformation): ObjectSummary => {
         headCid: object.metadata.dataCid,
         name: object.metadata.name,
         type: object.metadata.type,
-        size: object.metadata.totalSize,
+        size: object.metadata.totalSize.toString(),
         mimeType: object.metadata.mimeType,
         uploadStatus: object.uploadStatus,
         owners: object.owners,

backend/src/models/objects/objectMappings.ts

@@ -0,0 +1,20 @@
+import { z } from 'zod'
+
+export type ObjectMapping = [
+  hash: string,
+  pieceIndex: number,
+  pieceOffset: number,
+]
+
+export const ObjectMappingSchema = z.tuple([z.string(), z.number(), z.number()])
+
+export const ObjectMappingListEntrySchema = z.object({
+  blockNumber: z.number(),
+  v0: z.object({
+    objects: z.array(ObjectMappingSchema),
+  }),
+})
+
+export type ObjectMappingListEntry = z.infer<
+  typeof ObjectMappingListEntrySchema
+>

backend/src/repositories/objects/interactions.ts

@@ -23,13 +23,13 @@ const createInteraction = async (
   id: string,
   subscription_id: string,
   type: InteractionType,
-  size: number,
+  size: bigint,
 ): Promise<Interaction> => {
   const db = await getDatabase()
 
   const interaction = await db.query<DBInteraction>(
     'INSERT INTO interactions (id, subscription_id, type, size) VALUES ($1, $2, $3, $4) RETURNING *',
-    [id, subscription_id, type, size],
+    [id, subscription_id, type, size.toString()],
   )
 
   return mapRows(interaction.rows)[0]

backend/src/repositories/objects/metadata.ts

@@ -1,6 +1,7 @@
 import { OffchainMetadata } from '@autonomys/auto-dag-data'
 import { getDatabase } from '../../drivers/pg.js'
 import { PaginatedResult } from '../../useCases/objects/common.js'
+import { stringify } from '../../utils/misc.js'
 
 export interface MetadataEntry {
   root_cid: string
@@ -34,7 +35,7 @@ const setMetadata = async (
 
   return db.query({
     text: 'INSERT INTO metadata (root_cid, head_cid, metadata) VALUES ($1, $2, $3) ON CONFLICT (root_cid, head_cid) DO UPDATE SET metadata = EXCLUDED.metadata',
-    values: [rootCid, headCid, JSON.stringify(metadata)],
+    values: [rootCid, headCid, stringify(metadata)],
   })
 }
 

backend/src/repositories/objects/nodes.ts

@@ -66,7 +66,8 @@ const getNodeCount = async ({
 }) => {
   const db = await getDatabase()
 
-  let query = 'SELECT count(*) FROM nodes'
+  let query =
+    'SELECT count(*) as total_count, count(piece_index) as archived_count FROM nodes'
   const params = []
   const conditions = []
 
@@ -96,17 +97,50 @@ const getNodeCount = async ({
 
   return db
     .query<{
-      count: string
+      total_count: string
+      archived_count: string
     }>({
       text: query,
       values: params,
     })
-    .then((e) => Number(e.rows[0].count))
+    .then((e) => ({
+      totalCount: Number(e.rows[0].total_count),
+      archivedCount: Number(e.rows[0].archived_count),
+    }))
+}
+
+const getArchivingNodesCID = async () => {
+  const db = await getDatabase()
+
+  return db
+    .query<Node>({
+      text: 'SELECT cid FROM nodes WHERE piece_index IS NULL and piece_offset IS NULL',
+    })
+    .then((e) => e.rows.map((e) => e.cid))
+}
+
+const setNodeArchivingData = async ({
+  cid,
+  pieceIndex,
+  pieceOffset,
+}: {
+  cid: string
+  pieceIndex: number
+  pieceOffset: number
+}) => {
+  const db = await getDatabase()
+
+  return db.query({
+    text: 'UPDATE nodes SET piece_index = $1, piece_offset = $2 WHERE cid = $3',
+    values: [pieceIndex, pieceOffset, cid],
+  })
 }
 
 export const nodesRepository = {
   getNode,
   getNodeCount,
   saveNode,
   saveNodes,
+  getArchivingNodesCID,
+  setNodeArchivingData,
 }

backend/src/repositories/objects/transactionResults.ts

@@ -1,5 +1,6 @@
 import { getDatabase } from '../../drivers/pg.js'
 import { TransactionResult } from '../../models/objects/transaction.js'
+import { stringify } from '../../utils/misc.js'
 import { Node } from './nodes.js'
 
 export interface TransactionResultEntry {
@@ -15,7 +16,7 @@ const storeTransactionResult = async (
 
   await db.query({
     text: 'INSERT INTO transaction_results (cid, transaction_result) VALUES ($1, $2) ON CONFLICT (cid) DO UPDATE SET transaction_result = EXCLUDED.transaction_result',
-    values: [cid, JSON.stringify(transaction_result)],
+    values: [cid, stringify(transaction_result)],
   })
 }
 
@@ -87,11 +88,26 @@ const getUploadedNodesByRootCid = async (rootCid: string) => {
   return result
 }
 
+const getFirstNotArchivedNode = async () => {
+  const db = await getDatabase()
+
+  const result = await db
+    .query<{ block_number: number }>({
+      text: `SELECT MIN(transaction_results.transaction_result->>'blockNumber') as block_number FROM transaction_results inner join nodes 
+      on transaction_results.cid = nodes.cid 
+      where nodes.piece_index is null`,
+    })
+    .then(({ rows }) => rows.at(0)?.block_number ?? null)
+
+  return result
+}
+
 export const transactionResultsRepository = {
   storeTransactionResult,
   getTransactionResult,
   getPendingUploads,
   getHeadTransactionResults,
   getPendingUploadsByHeadCid,
   getUploadedNodesByRootCid,
+  getFirstNotArchivedNode,
 }

backend/src/repositories/uploads/blockstore.ts

@@ -5,22 +5,22 @@ interface BlockstoreEntry {
   upload_id: string
   cid: string
   node_type: MetadataType
-  node_size: number
+  node_size: bigint
   data: Buffer
 }
 
 const parseEntry = (entry: BlockstoreEntry) => {
   return {
     ...entry,
-    node_size: Number(entry.node_size),
+    node_size: entry.node_size.toString(),
   }
 }
 
 const addBlockstoreEntry = async (
   uploadId: string,
   cid: string,
   nodeType: MetadataType,
-  nodeSize: number,
+  nodeSize: bigint,
   data: Buffer,
 ) => {
   const db = await getDatabase()