feat: Implement async props with incremental rendering for React Server Components

packages/react-on-rails-pro-node-renderer/src/shared/utils.ts

@@ -8,6 +8,7 @@ import * as errorReporter from './errorReporter.js';
 import { getConfig } from './configBuilder.js';
 import log from './log.js';
 import type { RenderResult } from '../worker/vm.js';
+import fileExistsAsync from './fileExistsAsync.js';
 
 export const TRUNCATION_FILLER = '\n... TRUNCATED ...\n';
 
@@ -169,3 +170,29 @@ export function getAssetPath(bundleTimestamp: string | number, filename: string)
   const bundleDirectory = getBundleDirectory(bundleTimestamp);
   return path.join(bundleDirectory, filename);
 }
+
+export async function validateBundlesExist(
+  bundleTimestamp: string | number,
+  dependencyBundleTimestamps?: (string | number)[],
+): Promise<ResponseResult | null> {
+  const missingBundles = (
+    await Promise.all(
+      [...(dependencyBundleTimestamps ?? []), bundleTimestamp].map(async (timestamp) => {
+        const bundleFilePath = getRequestBundleFilePath(timestamp);
+        const fileExists = await fileExistsAsync(bundleFilePath);
+        return fileExists ? null : timestamp;
+      }),
+    )
+  ).filter((timestamp) => timestamp !== null);
+
+  if (missingBundles.length > 0) {
+    const missingBundlesText = missingBundles.length > 1 ? 'bundles' : 'bundle';
+    log.info(`No saved ${missingBundlesText}: ${missingBundles.join(', ')}`);
+    return {
+      headers: { 'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate' },
+      status: 410,
+      data: 'No bundle uploaded',
+    };
+  }
+  return null;
+}

packages/react-on-rails-pro-node-renderer/src/worker.ts

@@ -13,11 +13,21 @@ import log, { sharedLoggerOptions } from './shared/log.js';
 import packageJson from './shared/packageJson.js';
 import { buildConfig, Config, getConfig } from './shared/configBuilder.js';
 import fileExistsAsync from './shared/fileExistsAsync.js';
-import type { FastifyInstance, FastifyReply, FastifyRequest } from './worker/types.js';
-import checkProtocolVersion from './worker/checkProtocolVersionHandler.js';
-import authenticate from './worker/authHandler.js';
-import { handleRenderRequest, type ProvidedNewBundle } from './worker/handleRenderRequest.js';
+import type { FastifyInstance, FastifyReply } from './worker/types.js';
+import { performRequestPrechecks } from './worker/requestPrechecks.js';
+import { type AuthBody, authenticate } from './worker/authHandler.js';
+import {
+  handleRenderRequest,
+  type ProvidedNewBundle,
+  handleNewBundlesProvided,
+} from './worker/handleRenderRequest.js';
 import handleGracefulShutdown from './worker/handleGracefulShutdown.js';
+import {
+  handleIncrementalRenderRequest,
+  type IncrementalRenderInitialRequest,
+  type IncrementalRenderSink,
+} from './worker/handleIncrementalRenderRequest.js';
+import { handleIncrementalRenderStream } from './worker/handleIncrementalRenderStream.js';
 import {
   errorResponseResult,
   formatExceptionMessage,
@@ -163,41 +173,11 @@ export default function run(config: Partial<Config>) {
     },
   });
 
-  const isProtocolVersionMatch = async (req: FastifyRequest, res: FastifyReply) => {
-    // Check protocol version
-    const protocolVersionCheckingResult = checkProtocolVersion(req);
-
-    if (typeof protocolVersionCheckingResult === 'object') {
-      await setResponse(protocolVersionCheckingResult, res);
-      return false;
-    }
-
-    return true;
-  };
-
-  const isAuthenticated = async (req: FastifyRequest, res: FastifyReply) => {
-    // Authenticate Ruby client
-    const authResult = authenticate(req);
-
-    if (typeof authResult === 'object') {
-      await setResponse(authResult, res);
-      return false;
-    }
-
-    return true;
-  };
-
-  const requestPrechecks = async (req: FastifyRequest, res: FastifyReply) => {
-    if (!(await isProtocolVersionMatch(req, res))) {
-      return false;
-    }
-
-    if (!(await isAuthenticated(req, res))) {
-      return false;
-    }
-
-    return true;
-  };
+  // Ensure NDJSON bodies are not buffered and are available as a stream immediately
+  app.addContentTypeParser('application/x-ndjson', (req, payload, done) => {
+    // Pass through the raw stream; the route will consume req.raw
+    done(null, payload);
+  });
 
   // See https://github.com/shakacode/react_on_rails_pro/issues/119 for why
   // the digest is part of the request URL. Yes, it's not used here, but the
@@ -212,7 +192,9 @@ export default function run(config: Partial<Config>) {
     // Can't infer from the route like Express can
     Params: { bundleTimestamp: string; renderRequestDigest: string };
   }>('/bundles/:bundleTimestamp/render/:renderRequestDigest', async (req, res) => {
-    if (!(await requestPrechecks(req, res))) {
+    const precheckResult = performRequestPrechecks(req.body);
+    if (precheckResult) {
+      await setResponse(precheckResult, res);
       return;
     }
 
@@ -254,7 +236,7 @@ export default function run(config: Partial<Config>) {
             providedNewBundles,
             assetsToCopy,
           });
-          await setResponse(result, res);
+          await setResponse(result.response, res);
         } catch (err) {
           const exceptionMessage = formatExceptionMessage(
             renderingRequest,
@@ -272,17 +254,129 @@ export default function run(config: Partial<Config>) {
     }
   });
 
+  // Streaming NDJSON incremental render endpoint
+  app.post<{
+    Params: { bundleTimestamp: string; renderRequestDigest: string };
+  }>('/bundles/:bundleTimestamp/incremental-render/:renderRequestDigest', async (req, res) => {
+    const { bundleTimestamp } = req.params;
+
+    // Stream parser state
+    let incrementalSink: IncrementalRenderSink | undefined;
+
+    try {
+      // Handle the incremental render stream
+      await handleIncrementalRenderStream({
+        request: req,
+        onRenderRequestReceived: async (obj: unknown) => {
+          // Build a temporary FastifyRequest shape for protocol/auth check
+          const tempReqBody = typeof obj === 'object' && obj !== null ? (obj as Record<string, unknown>) : {};
+
+          // Perform request prechecks
+          const precheckResult = performRequestPrechecks(tempReqBody);
+          if (precheckResult) {
+            return {
+              response: precheckResult,
+              shouldContinue: false,
+            };
+          }
+
+          // Extract data for incremental render request
+          const dependencyBundleTimestamps = extractBodyArrayField(
+            tempReqBody as WithBodyArrayField<Record<string, unknown>, 'dependencyBundleTimestamps'>,
+            'dependencyBundleTimestamps',
+          );
+
+          const initial: IncrementalRenderInitialRequest = {
+            firstRequestChunk: obj,
+            bundleTimestamp,
+            dependencyBundleTimestamps,
+          };
+
+          try {
+            const { response, sink } = await handleIncrementalRenderRequest(initial);
+            incrementalSink = sink;
+
+            return {
+              response,
+              shouldContinue: !!incrementalSink,
+            };
+          } catch (err) {
+            const errorResponse = errorResponseResult(
+              formatExceptionMessage(
+                'IncrementalRender',
+                err,
+                'Error while handling incremental render request',
+              ),
+            );
+            return {
+              response: errorResponse,
+              shouldContinue: false,
+            };
+          }
+        },
+
+        onUpdateReceived: async (obj: unknown) => {
+          if (!incrementalSink) {
+            log.error({ msg: 'Unexpected update chunk received after rendering was aborted', obj });
+            return;
+          }
+
+          try {
+            log.info(`Received a new update chunk ${JSON.stringify(obj)}`);
+            await incrementalSink.add(obj);
+          } catch (err) {
+            // Log error but don't stop processing
+            log.error({ err, msg: 'Error processing update chunk' });
+          }
+        },
+
+        onResponseStart: async (response: ResponseResult) => {
+          await setResponse(response, res);
+        },
+
+        onRequestEnded: () => {
+          if (!incrementalSink) {
+            return;
+          }
+
+          incrementalSink.handleRequestClosed();
+        },
+      });
+    } catch (err) {
+      // If an error occurred during stream processing, send error response
+      const errorResponse = errorResponseResult(
+        formatExceptionMessage('IncrementalRender', err, 'Error while processing incremental render stream'),
+      );
+      await setResponse(errorResponse, res);
+    }
+  });
+
   // There can be additional files that might be required at the runtime.
   // Since the remote renderer doesn't contain any assets, they must be uploaded manually.
   app.post<{
     Body: WithBodyArrayField<Record<string, Asset>, 'targetBundles'>;
   }>('/upload-assets', async (req, res) => {
-    if (!(await requestPrechecks(req, res))) {
+    const precheckResult = performRequestPrechecks(req.body);
+    if (precheckResult) {
+      await setResponse(precheckResult, res);
       return;
     }
     let lockAcquired = false;
     let lockfileName: string | undefined;
-    const assets: Asset[] = Object.values(req.body).filter(isAsset);
+    const assets: Asset[] = [];
+
+    // Extract bundles that start with 'bundle_' prefix
+    const bundles: Array<{ timestamp: string; bundle: Asset }> = [];
+    Object.entries(req.body).forEach(([key, value]) => {
+      if (isAsset(value)) {
+        if (key.startsWith('bundle_')) {
+          const timestamp = key.replace('bundle_', '');
+          bundles.push({ timestamp, bundle: value });
+        } else {
+          assets.push(value);
+        }
+      }
+    });
 
     // Handle targetBundles as either a string or an array
     const targetBundles = extractBodyArrayField(req.body, 'targetBundles');
@@ -294,7 +388,9 @@ export default function run(config: Partial<Config>) {
     }
 
     const assetsDescription = JSON.stringify(assets.map((asset) => asset.filename));
-    const taskDescription = `Uploading files ${assetsDescription} to bundle directories: ${targetBundles.join(', ')}`;
+    const bundlesDescription =
+      bundles.length > 0 ? ` and bundles ${JSON.stringify(bundles.map((b) => b.bundle.filename))}` : '';
+    const taskDescription = `Uploading files ${assetsDescription}${bundlesDescription} to bundle directories: ${targetBundles.join(', ')}`;
 
     try {
       const { lockfileName: name, wasLockAcquired, errorMessage } = await lock('transferring-assets');
@@ -333,7 +429,24 @@ export default function run(config: Partial<Config>) {
 
           await Promise.all(assetCopyPromises);
 
-          // Delete assets from uploads directory
+          // Handle bundles using the existing logic from handleRenderRequest
+          if (bundles.length > 0) {
+            const providedNewBundles = bundles.map(({ timestamp, bundle }) => ({
+              timestamp,
+              bundle,
+            }));
+
+            // Use the existing bundle handling logic
+            // Note: handleNewBundlesProvided will handle deleting the uploaded bundle files
+            // Pass null for assetsToCopy since we handle assets separately in this endpoint
+            const bundleResult = await handleNewBundlesProvided('upload-assets', providedNewBundles, null);
+            if (bundleResult) {
+              await setResponse(bundleResult, res);
+              return;
+            }
+          }
+
+          // Delete assets from uploads directory (bundles are already handled by handleNewBundlesProvided)
           await deleteUploadedAssets(assets);
 
           await setResponse(
@@ -344,7 +457,7 @@ export default function run(config: Partial<Config>) {
             res,
           );
         } catch (err) {
-          const msg = 'ERROR when trying to copy assets';
+          const msg = 'ERROR when trying to copy assets and bundles';
           const message = `${msg}. ${err}. Task: ${taskDescription}`;
           log.error({
             msg,
@@ -376,7 +489,9 @@ export default function run(config: Partial<Config>) {
     Querystring: { filename: string };
     Body: WithBodyArrayField<Record<string, unknown>, 'targetBundles'>;
   }>('/asset-exists', async (req, res) => {
-    if (!(await isAuthenticated(req, res))) {
+    const authResult = authenticate(req.body as AuthBody);
+    if (authResult) {
+      await setResponse(authResult, res);
       return;
     }
 

packages/react-on-rails-pro-node-renderer/src/worker/authHandler.ts

@@ -10,11 +10,15 @@ import { timingSafeEqual } from 'crypto';
 import type { FastifyRequest } from './types.js';
 import { getConfig } from '../shared/configBuilder.js';
 
-export default function authenticate(req: FastifyRequest) {
+export interface AuthBody {
+  password?: string;
+}
+
+export function authenticate(body: AuthBody) {
   const { password } = getConfig();
 
   if (password) {
-    const reqPassword = (req.body as { password?: string }).password || '';
+    const reqPassword = body.password || '';
 
     // Use timing-safe comparison to prevent timing attacks
     // Both strings must be converted to buffers of the same length

packages/react-on-rails-pro-node-renderer/src/worker/checkProtocolVersionHandler.ts

@@ -2,7 +2,6 @@
  * Logic for checking protocol version.
  * @module worker/checkProtocVersionHandler
  */
-import type { FastifyRequest } from './types.js';
 import packageJson from '../shared/packageJson.js';
 import log from '../shared/log.js';
 
@@ -35,14 +34,14 @@ function normalizeVersion(version: string): string {
   return normalized;
 }
 
-interface RequestBody {
+export interface RequestBody {
   protocolVersion?: string;
   gemVersion?: string;
   railsEnv?: string;
 }
 
-export default function checkProtocolVersion(req: FastifyRequest) {
-  const { protocolVersion: reqProtocolVersion, gemVersion, railsEnv } = req.body as RequestBody;
+export function checkProtocolVersion(body: RequestBody) {
+  const { protocolVersion: reqProtocolVersion, gemVersion, railsEnv } = body;
 
   // Check protocol version
   if (reqProtocolVersion !== packageJson.protocolVersion) {
@@ -52,7 +51,7 @@ export default function checkProtocolVersion(req: FastifyRequest) {
       data: `Unsupported renderer protocol version ${
         reqProtocolVersion
           ? `request protocol ${reqProtocolVersion}`
-          : `MISSING with body ${JSON.stringify(req.body)}`
+          : `MISSING with body ${JSON.stringify(body)}`
       } does not match installed renderer protocol ${packageJson.protocolVersion} for version ${packageJson.version}.
 Update either the renderer or the Rails server`,
     };