feat(mu): add push msg to hb route

Author: jfrain99

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

@@ -0,0 +1,68 @@
+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,
+  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) => {
+        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/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,16 @@ 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,
+    fetch
+  })
+
   const pushMsgItemLogger = logger.child('pushMsg')
   const pushMsg = pushMsgWith({
     selectNode: cuClient.selectNodeWith({ CU_URL, logger: sendDataItemLogger }),
@@ -447,6 +458,7 @@ export const createApis = async (ctx) => {
     sendAssign,
     fetchCron,
     pushMsg,
+    pushResultToHb,
     traceMsgs,
     initCronProcs: cronClient.initCronProcsWith({
       startMonitoredProcess: startProcessMonitor,

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
+)