fix(mu): resolve merge conflict

Author: VinceJuliano

servers/mu/src/config.js

@@ -85,7 +85,9 @@ export const domainConfigSchema = z.object({
   RELAY_MAP: jsonObjectSchema,
   ENABLE_PUSH: z.boolean(),
   ENABLE_CUSTOM_PUSH: z.boolean(),
+  ENABLE_FILE_PUSH: z.boolean(),
   CUSTOM_CU_MAP_FILE_PATH: z.string(),
+  RESULT_FILE_DIRECTORY: z.string(),
   IP_WALLET_RATE_LIMIT: positiveIntSchema,
   IP_WALLET_RATE_LIMIT_INTERVAL: positiveIntSchema,
   STALE_CURSOR_RANGE: positiveIntSchema,
@@ -149,7 +151,9 @@ const CONFIG_ENVS = {
     RELAY_MAP: process.env.RELAY_MAP || '',
     ENABLE_PUSH: process.env.ENABLE_PUSH === 'true',
     ENABLE_CUSTOM_PUSH: process.env.ENABLE_CUSTOM_PUSH === 'true',
+    ENABLE_FILE_PUSH: process.env.ENABLE_FILE_PUSH === 'true',
     CUSTOM_CU_MAP_FILE_PATH: process.env.CUSTOM_CU_MAP_FILE_PATH || 'custom-cu-map.json',
+    RESULT_FILE_DIRECTORY: process.env.RESULT_FILE_DIRECTORY || 'results',
     IP_WALLET_RATE_LIMIT: process.env.IP_WALLET_RATE_LIMIT || 2000,
     IP_WALLET_RATE_LIMIT_INTERVAL: process.env.IP_WALLET_RATE_LIMIT_INTERVAL || 1000 * 60 * 60,
     STALE_CURSOR_RANGE: process.env.STALE_CURSOR_RANGE || 1 * 24 * 60 * 60 * 1000,
@@ -190,7 +194,9 @@ const CONFIG_ENVS = {
     RELAY_MAP: process.env.RELAY_MAP || '',
     ENABLE_PUSH: process.env.ENABLE_PUSH === 'true',
     ENABLE_CUSTOM_PUSH: process.env.ENABLE_CUSTOM_PUSH === 'true',
+    ENABLE_FILE_PUSH: process.env.ENABLE_FILE_PUSH === 'true',
     CUSTOM_CU_MAP_FILE_PATH: process.env.CUSTOM_CU_MAP_FILE_PATH || 'custom-cu-map.json',
+    RESULT_FILE_DIRECTORY: process.env.RESULT_FILE_DIRECTORY || 'results',
     IP_WALLET_RATE_LIMIT: process.env.IP_WALLET_RATE_LIMIT || 2000,
     IP_WALLET_RATE_LIMIT_INTERVAL: process.env.IP_WALLET_RATE_LIMIT_INTERVAL || 1000 * 60 * 60,
     STALE_CURSOR_RANGE: process.env.STALE_CURSOR_RANGE || 1 * 24 * 60 * 60 * 1000,

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

@@ -3,6 +3,7 @@ import { __, assoc, identity } from 'ramda'
 
 import { getCuAddressWith } from '../lib/get-cu-address.js'
 import { pullResultWith } from '../lib/pull-result.js'
+import { pullResultFromFileWith } from '../lib/pull-result-file.js'
 import { getCustomCuAddressWith } from '../lib/get-custom-cu-address.js'
 
 export function pushMsgWith ({
@@ -14,11 +15,14 @@ export function pushMsgWith ({
   ALLOW_PUSHES_AFTER,
   ENABLE_PUSH,
   ENABLE_CUSTOM_PUSH,
-  CUSTOM_CU_MAP_FILE_PATH
+  CUSTOM_CU_MAP_FILE_PATH,
+  readResultFromFile,
+  ENABLE_FILE_PUSH
 }) {
   const getCuAddress = getCuAddressWith({ selectNode, logger })
   const getCustomCuAddress = getCustomCuAddressWith({ CUSTOM_CU_MAP_FILE_PATH, logger })
   const pullResult = pullResultWith({ fetchResult, logger })
+  const pullResultFromFile = pullResultFromFileWith({ readResultFromFile, logger })
   const fetchTransactionsAsync = fromPromise(fetchTransactions)
 
   return (ctx) => {
@@ -48,7 +52,12 @@ export function pushMsgWith ({
         }
         return getCuAddress(ctx)
       })
-      .chain(pullResult)
+      .chain((ctx) => { 
+        if (ENABLE_FILE_PUSH && ctx.resultFile) {
+          return pullResultFromFile(ctx)
+        }
+        return pullResult(ctx)
+      })
       .chain((res) => {
         const { msgs, number } = res
         if(msgs.length <= number) {

servers/mu/src/domain/index.js

@@ -3,6 +3,8 @@ import * as B64js from "base64-js"
 import { randomBytes } from 'node:crypto'
 import { BroadcastChannel } from 'node:worker_threads'
 import cron from 'node-cron'
+import fs from 'fs'
+
 import { apply } from 'ramda'
 import warpArBundles from 'warp-arbundles'
 import { connect as schedulerUtilsConnect } from '@permaweb/ao-scheduler-utils'
@@ -405,6 +407,12 @@ export const createApis = async (ctx) => {
 
   const traceMsgs = fromPromise(readTracesWith({ db: traceDb, TRACE_DB_URL: ctx.TRACE_DB_URL, DISABLE_TRACE: ctx.DISABLE_TRACE }))
 
+  const readResultFromFile = async (pipelineCtx) => {
+    let resultsDir = ctx.RESULT_FILE_DIRECTORY
+    let messageId = pipelineCtx.tx.id
+    return JSON.parse(fs.readFileSync(`${resultsDir}/${messageId}.json`))
+  }
+
   const pushMsgItemLogger = logger.child('pushMsg')
   const pushMsg = pushMsgWith({
     selectNode: cuClient.selectNodeWith({ CU_URL, logger: sendDataItemLogger }),
@@ -415,7 +423,9 @@ export const createApis = async (ctx) => {
     ALLOW_PUSHES_AFTER,
     ENABLE_PUSH: ctx.ENABLE_PUSH,
     ENABLE_CUSTOM_PUSH: ctx.ENABLE_CUSTOM_PUSH,
-    CUSTOM_CU_MAP_FILE_PATH: ctx.CUSTOM_CU_MAP_FILE_PATH
+    ENABLE_FILE_PUSH: ctx.ENABLE_FILE_PUSH,
+    CUSTOM_CU_MAP_FILE_PATH: ctx.CUSTOM_CU_MAP_FILE_PATH,
+    readResultFromFile
   })
 
   const startMessageRecoveryCronLogger = logger.child('messageRecoveryCron')

servers/mu/src/domain/lib/pull-result-file.js

@@ -0,0 +1,86 @@
+import { of, fromPromise, Resolved } from 'hyper-async'
+import z from 'zod'
+import { checkStage } from '../utils.js'
+import { resultSchema } from '../dal.js'
+
+const ctxSchema = z.object({
+  msgs: z.any(),
+  spawns: z.any(),
+  assigns: z.any(),
+  initialTxId: z.any()
+}).passthrough()
+
+function readResultFromFileWith ({ readResultFromFile }) {
+  const readResultAsync = fromPromise(readResultFromFile)
+
+  return (ctx) => {
+    return of(ctx)
+      .chain(() => {
+        return readResultAsync(ctx)
+      })
+      .chain(fetchedResult => {
+        const msgs = fetchedResult.Messages
+          .filter(msg =>
+            msg.Target !== undefined && msg.Anchor !== undefined && msg.Tags !== undefined
+          )
+          .map(msg => {
+            return {
+              msg,
+              processId: msg.Target,
+              initialTxId: ctx.initialTxId,
+              fromProcessId: ctx.tx.processId,
+              parentId: ctx.messageId ?? ctx.initialTxId,
+              wallet: ctx.wallet ?? ctx.dataItem?.owner
+            }
+          })
+
+        const spawns = fetchedResult.Spawns.map(spawn => {
+          return {
+            spawn,
+            processId: ctx.tx.processId,
+            initialTxId: ctx.initialTxId,
+            fromProcessId: ctx.processId,
+            parentId: ctx.messageId ?? ctx.initialTxId,
+            wallet: ctx.wallet ?? ctx.dataItem?.owner
+          }
+        })
+
+        /*
+          we have to concat on any assignments that
+          come from the Assignments tag, so they get
+          returned in the final result and picked up
+          by the crank
+        */
+        const assigns = ctx.tagAssignments
+          ? fetchedResult.Assignments
+            .concat(ctx.tagAssignments)
+          : fetchedResult.Assignments
+
+        return of({ ...ctx, msgs, spawns, assigns })
+      })
+      .toPromise()
+  }
+}
+
+export function pullResultFromFileWith (env) {
+  const { logger } = env
+
+  const readResultFromFile = readResultFromFileWith(env)
+
+  return (ctx) => {
+    if (!checkStage('pull-result')(ctx) && !checkStage('pull-initial-result')(ctx)) return Resolved(ctx)
+    return of(ctx)
+      .map((ctx) => ({ ...ctx, tx: ctx.tx }))
+      .chain((ctx) => {
+        return of(ctx)
+          .chain(fromPromise(readResultFromFile))
+      })
+      .map(ctxSchema.parse)
+      .bimap(
+        (e) => {
+          return new Error(e, { cause: ctx })
+        },
+        logger.tap({ log: 'Added msgs, spawns, and assigns to ctx' })
+      )
+  }
+}

servers/mu/src/routes/push.js

@@ -17,7 +17,8 @@ const withPushRoute = (app) => {
           params: { id, number },
           query: {
             'process-id': processId,
-            'custom-cu': customCu
+            'custom-cu': customCu,
+            'result-file': resultFile
           }
         } = req
 
@@ -28,7 +29,7 @@ const withPushRoute = (app) => {
           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, customCu })
+        await of({ tx: { id, processId }, number: Number(number), logId, messageId: id, initialTxId: id, customCu, resultFile })
           .chain(pushMsg)
           .bimap(
             (e) => {