LogoDaniel Wise

Integrating Azure Web PubSub with Inngest for Real-Time React Updates

Daniel Wise

Daniel Wise

@Daniel Wise

Integrating Azure Web PubSub with Inngest for Real-Time React Updates

Why Use Azure Web PubSub?

WebSockets are an efficient way to push real-time updates to clients without the need for constant polling. Azure Web PubSub makes it easy to manage WebSocket connections and broadcast messages to groups of clients.

The Core Approach

The key to this integration is structuring each step in my Inngest workflow so that it can return a publish object. This object contains all the required arguments to send an update.

1. Handling Real-Time Updates in an Inngest Workflow

The first step is defining how the parsed Excel sheets are processed in the workflow. Here's the code snippet:

const stepParsedSheets = await step.run('parse-sheets', async () => {
  const { data: _data, publish: _publish } = stepFileRecord;
  const stepNumber = _data.stepNumber + 1;
 
  const { data: fileBuffer, error } = await fileService.downloadFile(
    _data.fileId
  );
 
  if (error || !fileBuffer) {
    logger.error(
      `Failed to download file: ${error?.message || 'Unknown error'}`
    );
    throw new Error(`Failed to download file`);
  }
 
  const parsedSheets = parseExcelWithCorrectHeaders(fileBuffer);
  const sheetCount = Object.keys(parsedSheets).length;
 
  if (sheetCount < 1) {
    logger.error('No sheets found in the file');
    throw new Error('No sheets found in the file');
  }
 
  const publish: PublishArgs<Data> = {
    ..._publish,
    payload: {
      ..._publish.payload,
      status: FileStatus.Parsed,
    },
    message: `Parsed ${sheetCount} sheets`,
    progress: (100 / TOTAL_STEPS) * stepNumber,
  };
 
  return {
    publish,
    data: {
      ..._data,
      parsedSheets,
      stepNumber,
    },
  };
});

This ensures that once the Excel file is parsed, a structured message containing the progress is created and sent.

2. Using Inngest Middleware to Handle Publishing

The next step was figuring out where to inject the real-time updates inside the Inngest middleware. Initially, I tried using afterExecution, but that led to inconsistent step ordering and duplicate messages.

Instead, I found transformOutput to be the best place for handling real-time updates:

export const pubsubMiddleware = new InngestMiddleware({
  name: 'PubSub Middleware',
  init() {
    return {
      onFunctionRun({ fn, steps }) {
        return {
          transformInput() {
            return { ctx: { publish } };
          },
          async transformOutput({ result, step }) {
            if (step && result.data) {
              const publishArgs = (
                result.data as { publish: PublishArgs<unknown> }
              ).publish;
 
              logger.info(
                `${fn.id()} | transformOutput | ${step.displayName} | Step Count: ${steps.length} | Message: ${publishArgs?.message}`
              );
 
              await publish(publishArgs);
            }
          },
        };
      },
    };
  },
});

With this setup, every workflow step that includes a publish object will automatically trigger a real-time update via Azure Web PubSub.

3. Implementing the WebSocket Client in React

On the React frontend, I used a simple custom hook to listen for incoming updates via WebSockets:

export function useFileUpdates(fileId: number, orgId: string) {
  const [status, setStatus] = useState<FileUpdateStatus>({
    label: startCase(FileStatus.Pending),
    progress: 0,
    state: FileStatus.Pending,
  });
 
  useEffect(() => {
    let ws: WebSocket | null = null;
    let mounted = true;
 
    const connect = async () => {
      try {
        const [url] = await pubsubClientUrlAction({
          permission: FilesPermission.Read,
        });
        if (!url) throw new Error('No Web PubSub URL found');
 
        ws = new WebSocket(url);
        ws.onopen = () => logger.info('Connected to Web PubSub');
 
        ws.onmessage = (event) => {
          try {
            const data: FileUpdateMessage = JSON.parse(event.data);
            if (data && data.payload.id === fileId) {
              const newStatus = mapFileStatus(
                data.payload.status as FileStatus,
                data.progress
              );
              if (mounted) {
                setStatus({ ...newStatus, progress: data.progress });
              }
            }
          } catch (err) {
            logger.error('Error processing message:', err);
          }
        };
 
        ws.onerror = (event) => {
          logger.error('WebSocket error:', event);
        };
 
        ws.onclose = () => {
          logger.info('WebSocket connection closed');
          if (mounted) setTimeout(connect, 5000);
        };
      } catch (err) {
        logger.error('WebSocket connection error:', err);
      }
    };
 
    connect();
    return () => {
      mounted = false;
      ws?.close();
    };
  }, [fileId, orgId]);
 
  return { status };
}

References

See all Blogs