Merge pull request #1238 from permaweb/feat/compatibility

Feat/compatibility

Author: NickJ202

connect/package-lock.json

@@ -37,6 +37,7 @@
     "./dist"
   ],
   "scripts": {
+    "clean:install": "rm -rf node_modules package-lock.json || true && npm cache clean --force && npm install",
     "build": "npm run build:types && npm run build:src",
     "build:src": "node esbuild.js",
     "build:types": "tsc src/index.js --skipLibCheck --declaration --allowJs --emitDeclarationOnly --outDir dist",
@@ -47,6 +48,7 @@
   },
   "dependencies": {
     "@dha-team/arbundles": "1.0.3",
+    "@permaweb/ao-core-libs": "0.0.7",
     "@permaweb/ao-scheduler-utils": "~0.0.25",
     "@permaweb/protocol-tag-utils": "~0.0.2",
     "axios": "^1.7.9",

connect/package.json

@@ -0,0 +1,267 @@
+import { debugLog } from '../logger.js'
+
+function convertToLegacyOutput(jsonRes) {
+  let body = {}
+  try {
+    body = JSON.parse(jsonRes?.results?.json?.body)
+    debugLog('info', 'HyperBEAM Response Body:', body)
+  } catch (_) { }
+
+  debugLog('info', 'Parsed HyperBEAM Response Body:', body)
+
+  return {
+    Output: body?.Output || {},
+    Messages: body?.Messages || [],
+    Assignments: body?.Assignments || [],
+    Spawns: body?.Spawns || [],
+    Error: body?.Error,
+    ...(body ?? {})
+  }
+}
+
+const baseParams = {
+  method: 'POST',
+  'signing-format': 'ans104',
+  'accept-bundle': 'true',
+}
+
+const httpParams = { ...baseParams, 'accept-codec': '[email protected]' }
+const jsonParams = { ...baseParams, 'require-codec': 'application/json' }
+
+const getAOParams = (type) => ({
+  Type: type,
+  'Data-Protocol': 'ao',
+  Variant: 'ao.N.1'
+})
+
+const getTags = (args) =>
+  args.tags
+    ? Object.fromEntries(args.tags.map(tag => [tag.name, tag.value]))
+    : {}
+
+const getData = (args) => args.data ?? '1984'
+
+export function requestWith(deps) {
+  return async (args) => {
+    try {
+      return deps.aoCore.request(args);
+    }
+    catch (e) {
+      throw new Error(e.message ?? 'Error sending request')
+    }
+  }
+}
+
+export function spawnWith(deps) {
+  return async (args) => {
+    let scheduler = deps.scheduler
+
+    if (!scheduler && deps.url) {
+      const schedulerRes = await fetch(`${deps.url}/[email protected]/info/address`)
+      scheduler = await schedulerRes.text()
+    }
+
+    const module = process.env.MODULE || args.module
+
+    if (!scheduler) throw new Error('No scheduler provided')
+    if (!module) throw new Error('No module provided')
+
+    const authority = process.env.AUTHORITY || args.authority || scheduler
+
+    debugLog('info', 'Node URL:', deps.url)
+    debugLog('info', 'Scheduler:', scheduler)
+    debugLog('info', 'Authority:', authority)
+    debugLog('info', 'Module:', module)
+
+    try {
+      const params = {
+        path: '/push',
+        device: '[email protected]',
+        'scheduler-device': '[email protected]',
+        'push-device': '[email protected]',
+        'execution-device': '[email protected]',
+        Authority: authority,
+        Scheduler: scheduler,
+        Module: module,
+        data: getData(args),
+        ...getTags(args),
+        ...getAOParams('Process'),
+        ...httpParams
+      }
+
+      const response = await deps.aoCore.request(params)
+
+      try {
+        const processId = response.headers.get('process')
+        debugLog('info', 'Process ID:', processId)
+
+        if (processId) return processId;
+        else throw new Error('Error spawning process')
+      } catch (e) {
+        throw new Error(e.message ?? 'Error spawning process')
+      }
+    } catch (e) {
+      throw new Error(e.message ?? 'Error spawning process')
+    }
+  }
+}
+
+export function messageWith(deps) {
+  return async (args) => {
+    try {
+      const params = {
+        path: `/${args.process}[email protected]/push`,
+        target: args.process,
+        data: getData(args),
+        ...getTags(args),
+        ...getAOParams('Message'),
+        ...jsonParams,
+      }
+
+      const response = await deps.aoCore.request(params)
+      if (response.ok) {
+        const parsedResponse = await response.json()
+
+        if (args.opts?.fullResponse) return convertToLegacyOutput(parsedResponse)
+        else return parsedResponse.slot
+      }
+      else throw new Error('Error sending message')
+    } catch (e) {
+      throw new Error(e.message ?? 'Error sending message')
+    }
+  }
+}
+
+export function resultWith(deps) {
+  return async (args) => {
+    try {
+      const params = {
+        path: `/${args.process}[email protected]/compute=${args.slot ?? args.message}`,
+        target: args.process,
+        data: getData(args),
+        ...getTags(args),
+        ...jsonParams
+      }
+      const response = await deps.aoCore.request(params)
+      if (response.ok) {
+        const parsedResponse = await response.json()
+        return convertToLegacyOutput(parsedResponse)
+      }
+      else throw new Error('Error getting result')
+    } catch (e) {
+      throw new Error(e.message ?? 'Error getting result')
+    }
+  }
+}
+
+export function resultsWith(deps) {
+  return async (args) => {
+    try {
+      const slotParams = {
+        path: `/${args.process}/slot/current`,
+        ...httpParams
+      }
+
+      const slotResponse = await deps.aoCore.request(slotParams)
+      if (slotResponse.ok) {
+        try {
+          const currentSlot = await slotResponse.text();
+
+          const resultsParams = {
+            path: `/${args.process}/compute=${currentSlot}`,
+            ...jsonParams
+          }
+
+          const resultsResponse = await deps.aoCore.request(resultsParams)
+
+          if (resultsResponse.ok) {
+            let parsedResultsResponse = await resultsResponse.json();
+
+            return {
+              edges: [
+                {
+                  cursor: currentSlot,
+                  node: {
+                    ...convertToLegacyOutput(parsedResultsResponse)
+                  }
+                }
+              ]
+            }
+          }
+
+          else throw new Error('Error getting results')
+        } catch (e) {
+          throw new Error(e.message ?? 'Error getting current process slot')
+        }
+      }
+      else throw new Error('Error getting results')
+    } catch (e) {
+      throw new Error(e.message ?? 'Error getting results')
+    }
+  }
+}
+
+export function dryrunWith(deps) {
+  return async (args) => {
+    try {
+      const tags = getTags(args)
+      const tagsAsParams = Object.entries(tags)?.length > 0
+        ? `&${Object.entries(tags).map(([key, value]) => `${key}=${value}`).join('&')}`
+        : ''
+
+      const path = `/${args.process}[email protected]/as=execution/compute${tagsAsParams}`
+      const params = {
+        path,
+        target: args.process,
+        data: getData(args),
+        ...jsonParams
+      }
+
+      const response = await deps.aoCore.request(params)
+      if (response.ok) {
+        const parsedResponse = await response.json()
+        return convertToLegacyOutput(parsedResponse)
+      }
+      else throw new Error('Error running dryrun')
+    } catch (e) {
+      throw new Error(e.message ?? 'Error running dryrun')
+    }
+  }
+}
+
+async function _retryInitPush(deps, processId, maxAttempts = 10) {
+  const params = {
+    path: `/${processId}/push`,
+    target: processId,
+    Action: 'Eval',
+    data: 'require(\'.process\')._version',
+    ...getAOParams('Message'),
+    ...httpParams
+  }
+
+  let lastError = null
+
+  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+    try {
+      const initPush = await deps.aoCore.request(params)
+      if (initPush.ok) {
+        debugLog('info', `Init push succeeded on attempt ${attempt}`);
+        return initPush
+      } else {
+        debugLog('warn', `Init push attempt ${attempt} returned ok=false`, {
+          status: initPush.status,
+          body: initPush
+        })
+        lastError = new Error(`Init push returned ok=false (status=${initPush.status})`);
+        await new Promise((r) => setTimeout(r, 500));
+      }
+    } catch (err) {
+      debugLog('warn', `Init push attempt ${attempt} threw`, err)
+      lastError = err
+    }
+
+    if (attempt === maxAttempts) break
+  }
+
+  throw new Error(`Init push failed after ${maxAttempts} attempts: ${lastError?.message || 'unknown'}`)
+}