Feat/upload wallet messages to hb

servers/mu/src/domain/api/pushResultToHb.js

@@ -0,0 +1,72 @@
+import { Rejected, Resolved, fromPromise, of } from 'hyper-async'
+
+import { getCuAddressWith } from '../lib/get-cu-address.js'
+import { pullResultWith } from '../lib/pull-result.js'
+
+export function pushResultToHbWith ({
+  selectNode,
+  fetchResult,
+  buildAndSign,
+  logger,
+  HB_GRAPHQL_URL,
+  ENABLE_PUSH,
+  fetch
+}) {
+  const getCuAddress = getCuAddressWith({ selectNode, logger })
+  const pullResult = pullResultWith({ fetchResult, logger })
+  const buildAndSignAsync = fromPromise(buildAndSign)
+
+  const uploadToHb = async ({ signedDataItem, processId, messageId, logId }) => {
+    const url = `${HB_GRAPHQL_URL}/id?codec-device=ans104@1.0`
+    logger({ log: `[pushResultToHb] Uploading signed data item to HB: ${url} processId=${processId} messageId=${messageId} logId=${logId}` })
+    const res = await fetch(url, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/octet-stream' },
+      body: signedDataItem
+    })
+    const text = await res.text()
+    if (!res.ok) {
+      throw new Error(`[pushResultToHb] HB upload failed: ${res.status} ${text}`)
+    }
+    logger({ log: `[pushResultToHb] HB upload succeeded: ${text}` })
+    return text
+  }
+
+  const uploadToHbAsync = fromPromise(uploadToHb)
+
+  return (ctx) => {
+    return of(ctx)
+      .chain(getCuAddress)
+      .chain(pullResult)
+      .chain((res) => {
+        if(!ENABLE_PUSH) {
+          return Rejected(new Error('Repush not enabled on this MU.', { cause: ctx }))
+        }
+        const { msgs, number } = res
+        if (msgs.length <= number) {
+          return Rejected(new Error('Message number does not exist in the result.', { cause: ctx }))
+        }
+        return Resolved(res)
+      })
+      .chain((res) => {
+        const { msgs, number } = res
+        const targetMsg = msgs[number].msg
+        logger({ log: `[pushResultToHb] Building and signing result message ${number} for ${ctx.tx.id} -> target=${targetMsg.Target}` })
+        console.dir({ targetMsg }, { depth: null })
+        return buildAndSignAsync({
+          processId: targetMsg.Target,
+          tags: targetMsg.Tags,
+          anchor: targetMsg.Anchor,
+          data: targetMsg.Data
+        }).chain((tx) => {
+          logger({ log: `[pushResultToHb] Signed data item id=${tx.id}, uploading to HB` })
+          return uploadToHbAsync({
+            signedDataItem: tx.data,
+            processId: ctx.tx.processId,
+            messageId: ctx.tx.id,
+            logId: ctx.logId
+          }).map((hbRes) => ({ ...res, hbRes, txId: tx.id }))
+        })
+      })
+  }
+}

servers/mu/src/domain/clients/uploader.js

@@ -11,9 +11,10 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ
     }),
     logger
   })
+
   /**
    * uploadDataItem
-   * Upload a Data Item directly to Arweave
+   * Upload a Data Item directly to Arweave, and fire-and-forget to HB.
    *
    * @param data - the Data Item to upload
    *
@@ -22,52 +23,45 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ
    * timestamp
    * signature
    * owner
-   *
    */
   return async (data) => {
     return of(data)
-      .map(logger.tap({ log: `Forwarding message to uploader ${UPLOADER_URL}` }))
-      .chain(
-        fromPromise((body) =>
-          dataItemFetch(`${UPLOADER_URL}/tx/arweave`, {
-            method: 'POST',
-            headers: {
-              'Content-Type': 'application/octet-stream',
-              Accept: 'application/json'
-            },
-            body
-          }).then((res) => { return { body, res } })
-        )
-      )
       .chain(
-        fromPromise(({ body, res }) =>
-          dataItemFetch(`${HB_GRAPHQL_URL}/~arweave@2.9-pre/tx?codec-device=ans104@1.0`, {
+        fromPromise(async (body) => {
+          // Fire HB upload in parallel — never blocks or fails the main flow
+          const hbUrl = `${HB_GRAPHQL_URL}/id?codec-device=ans104@1.0`
+          logger.tap({ log: `[uploader] Forwarding to HB: ${hbUrl}` })()
+          dataItemFetch(hbUrl, { method: 'POST', body })
+            .then((res) => res.text())
+            .then((text) => logger.tap({ log: `[uploader] HB response: ${text}` })())
+            .catch((err) => logger.tap({ log: `[uploader] HB upload error (non-fatal): ${err.message}` })())
+
+          // Arweave upload — this is the one we await and return
+          logger.tap({ log: `[uploader] Forwarding to Arweave: ${UPLOADER_URL}/tx/arweave` })()
+          return dataItemFetch(`${UPLOADER_URL}/tx/arweave`, {
             method: 'POST',
             headers: {
               'Content-Type': 'application/octet-stream',
               Accept: 'application/json'
             },
             body
           })
-            .then(() => res)
-            .catch((err) => {
-              logger.tap({ log: 'Error while communicating with HB uploader:' })(err)
-              return res
-            })
-        )
+        })
       )
-      .bimap(logger.tap({ log: 'Error while communicating with uploader:' }), identity)
+      .bimap(logger.tap({ log: '[uploader] Error communicating with Arweave uploader:' }), identity)
       .bichain(
         (err) => Rejected(JSON.stringify(err)),
         fromPromise(async (res) => {
           if (!res?.ok) {
             const text = await res.text()
+            logger.tap({ log: `[uploader] Arweave upload failed: ${res.status} ${text}` })()
             throw new Error(`${res.status}: ${text}`)
           }
-          return res.json()
+          const json = await res.json()
+          logger.tap({ log: `[uploader] Arweave upload succeeded: ${json.id}` })()
+          return json
         })
       )
-      .map(logger.tap({ log: 'Successfully forwarded DataItem to uploader' }))
       .toPromise()
   }
 }

servers/mu/src/domain/index.js

@@ -29,6 +29,7 @@ import { sendDataItemWith, startMessageRecoveryCronWith } from './api/sendDataIt
 import { sendAssignWith } from './api/sendAssign.js'
 import { processAssignWith } from './api/processAssign.js'
 import { pushMsgWith } from './api/pushMsg.js'
+import { pushResultToHbWith } from './api/pushResultToHb.js'
 
 import { createLogger } from './logger.js'
 import { cuFetchWithCache } from './lib/cu-fetch-with-cache.js'
@@ -410,6 +411,17 @@ export const createApis = async (ctx) => {
 
   const traceMsgs = fromPromise(readTracesWith({ db: traceDb, TRACE_DB_URL: ctx.TRACE_DB_URL, DISABLE_TRACE: ctx.DISABLE_TRACE }))
 
+  const pushResultToHbLogger = logger.child('pushResultToHb')
+  const pushResultToHb = pushResultToHbWith({
+    selectNode: cuClient.selectNodeWith({ CU_URL, logger: pushResultToHbLogger }),
+    fetchResult: cuClient.resultWith({ fetch: fetchWithCache, histogram, CU_URL, logger: pushResultToHbLogger }),
+    buildAndSign: signerClient.buildAndSignWith({ MU_WALLET, logger: pushResultToHbLogger }),
+    logger: pushResultToHbLogger,
+    HB_GRAPHQL_URL,
+    ENABLE_PUSH: ctx.ENABLE_PUSH,
+    fetch
+  })
+
   const pushMsgItemLogger = logger.child('pushMsg')
   const pushMsg = pushMsgWith({
     selectNode: cuClient.selectNodeWith({ CU_URL, logger: sendDataItemLogger }),
@@ -447,6 +459,7 @@ export const createApis = async (ctx) => {
     sendAssign,
     fetchCron,
     pushMsg,
+    pushResultToHb,
     traceMsgs,
     initCronProcs: cronClient.initCronProcsWith({
       startMonitoredProcess: startProcessMonitor,

servers/mu/src/domain/lib/cu-fetch-with-cache.js

@@ -34,7 +34,7 @@ export function cuFetchWithCache ({ fetch, cache, logger }) {
       logger({ log: ['found redirect url in cache for process: %s redirect: %s', processId, foundRedirectUrl], logId })
       // only sets the host, the protocol will be reused from the passed in url
       // this is a safe assumption because all CUs are implementing the same APIs over the same protocols
-      requestUrl.host = foundRedirectUrl
+      // requestUrl.host = foundRedirectUrl
     }
     return runFetch(requestUrl.toString(), opts, logId)
   }

servers/mu/src/routes/index.js

@@ -4,10 +4,12 @@ import { withRootRoutes } from './root.js'
 import { withMonitorRoutes } from './monitor.js'
 import { withMetricRoutes } from './metrics.js'
 import { withPushRoutes } from './push.js'
+import { withPushResultToHbRoutes } from './pushResultToHb.js'
 
 export const withRoutes = pipe(
   withMonitorRoutes,
   withRootRoutes,
   withMetricRoutes,
-  withPushRoutes
+  withPushRoutes,
+  withPushResultToHbRoutes
 )

servers/mu/src/routes/pushResultToHb.js

@@ -0,0 +1,58 @@
+import { always, compose, pipe } from 'ramda'
+import { of } from 'hyper-async'
+import { randomBytes } from 'node:crypto'
+
+import { withMetrics, withMiddleware } from './middleware/index.js'
+
+const withPushResultToHbRoute = (app) => {
+  app.post(
+    '/push-result/:id/:number',
+    compose(
+      withMiddleware,
+      withMetrics(),
+      always(async (req, res) => {
+        const {
+          logger: _logger,
+          domain: { apis: { pushResultToHb } },
+          params: { id, number },
+          query: {
+            'process-id': processId
+          }
+        } = req
+
+        const logger = _logger.child('POST_push_result_to_hb')
+        const logId = randomBytes(8).toString('hex')
+
+        if (isNaN(Number(number))) {
+          return res.status(400).send({ error: "'number' parameter must be a valid number" })
+        }
+
+        await of({
+          tx: { id, processId },
+          number: Number(number),
+          logId,
+          messageId: id,
+          initialTxId: id
+        })
+          .chain(pushResultToHb)
+          .bimap(
+            (e) => {
+              logger({ log: `[push-result] Failed: ${e}`, end: true }, e.cause)
+              res.status(500).send({ error: String(e) })
+            },
+            ({ hbRes }) => {
+              logger({ log: `[push-result] Success for ${id}/${number}`, end: true })
+              res.status(200).send({ message: 'Result uploaded to HB', id, number: Number(number), hbRes })
+            }
+          )
+          .toPromise()
+      })
+    )()
+  )
+
+  return app
+}
+
+export const withPushResultToHbRoutes = pipe(
+  withPushResultToHbRoute
+)