DSN Integration

backend/migrations/20241114152906-dsn-intergration.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', '20241114152906-dsn-intergration-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', '20241114152906-dsn-intergration-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/20241114152906-dsn-intergration-down.sql

@@ -0,0 +1 @@
+/* Replace with your SQL commands */

backend/migrations/sqls/20241114152906-dsn-intergration-up.sql

@@ -0,0 +1,3 @@
+DROP TABLE nodes;
+
+ALTER TABLE uploads.blockstore ADD COLUMN "status" TEXT;

backend/package.json

@@ -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,92 @@
+import Websocket from 'websocket'
+
+type RPCMessage = {
+  jsonrpc: string
+  method: string
+  params: unknown
+  id: number
+}
+
+export type WS = {
+  send: (message: Omit<RPCMessage, 'id'>) => Promise<unknown>
+  on: (callback: (event: RPCMessage) => void) => void
+  off: (callback: (event: RPCMessage) => 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(JSON.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/uploads/upload.ts

@@ -15,11 +15,11 @@ export enum UploadType {
 }
 
 export enum UploadStatus {
-  PENDING = 'pending',
+  UPLOADING = 'uploading',
+  UPLOADED = 'uploaded',
+  PUBLISHING = 'publishing',
+  ARCHIVING = 'archiving',
   COMPLETED = 'completed',
-  MIGRATING = 'migrating',
-  CANCELLED = 'cancelled',
-  FAILED = 'failed',
 }
 
 export const uploadOptionsSchema = z.object({

backend/src/repositories/objects/index.ts

@@ -1,5 +1,4 @@
 export * from './interactions.js'
 export * from './metadata.js'
-export * from './nodes.js'
 export * from './ownership.js'
 export * from './transactionResults.js'

backend/src/repositories/objects/transactionResults.ts

@@ -1,6 +1,6 @@
 import { getDatabase } from '../../drivers/pg.js'
 import { TransactionResult } from '../../models/objects/transaction.js'
-import { Node } from './nodes.js'
+import { BlockstoreEntry } from '../uploads/blockstore.js'
 
 export interface TransactionResultEntry {
   cid: string
@@ -49,9 +49,9 @@ const getHeadTransactionResults = async (head_cid: string) => {
 const getPendingUploads = async (limit: number = 100) => {
   const db = await getDatabase()
   const result = await db
-    .query<Node>(
+    .query<BlockstoreEntry>(
       `
-    SELECT n.* FROM nodes n left join transaction_results tr on n.cid = tr.cid where tr.cid is null limit $1
+    SELECT n.* FROM uploads.blockstore n left join transaction_results tr on n.cid = tr.cid where tr.cid is null limit $1
   `,
       [limit],
     )
@@ -78,7 +78,7 @@ const getUploadedNodesByRootCid = async (rootCid: string) => {
   const db = await getDatabase()
   const result = await db
     .query<TransactionResultEntry>({
-      text: `select tr.* from transaction_results tr join nodes n on tr.cid = n.cid where n.root_cid = $1 and tr.transaction_result->>'blockNumber' is not null order by tr.transaction_result->>'blockNumber' asc
+      text: `select * from uploads.uploads u join uploads.blockstore b on u.id = b.upload_id and u.root_upload_id = u.id where b.cid = $1
   `,
       values: [rootCid],
     })