Class MemFlow

The MemFlow service provides a Temporal-compatible workflow framework backed by Postgres. It offers durable execution, entity-based memory management, and composable workflows.

Core Features

1. Entity-Based Memory Model

Each workflow has a durable JSONB entity that serves as its memory:

export async function researchAgent(query: string) {
  const agent = await MemFlow.workflow.entity();

  // Initialize entity state
  await agent.set({
    query,
    findings: [],
    status: 'researching'
  });

  // Update state atomically
  await agent.merge({ status: 'analyzing' });
  await agent.append('findings', newFinding);
}
Copy

2. Hook Functions & Workflow Coordination

Spawn and coordinate multiple perspectives/phases:

// Launch parallel research perspectives
await MemFlow.workflow.execHook({
  taskQueue: 'research',
  workflowName: 'optimisticView',
  args: [query],
  signalId: 'optimistic-complete'
});

await MemFlow.workflow.execHook({
  taskQueue: 'research',
  workflowName: 'skepticalView',
  args: [query],
  signalId: 'skeptical-complete'
});

// Wait for both perspectives
await Promise.all([
  MemFlow.workflow.waitFor('optimistic-complete'),
  MemFlow.workflow.waitFor('skeptical-complete')
]);
Copy

3. Durable Activities & Proxies

Define and execute durable activities with automatic retry:

const activities = MemFlow.workflow.proxyActivities<{
  analyzeDocument: typeof analyzeDocument;
  validateFindings: typeof validateFindings;
}>({
  activities: { analyzeDocument, validateFindings },
  retryPolicy: {
    maximumAttempts: 3,
    backoffCoefficient: 2
  }
});

// Activities are durable and automatically retried
const analysis = await activities.analyzeDocument(data);
const validation = await activities.validateFindings(analysis);
Copy

4. Workflow Composition

Build complex workflows through composition:

// Start a child workflow
const childResult = await MemFlow.workflow.execChild({
  taskQueue: 'analysis',
  workflowName: 'detailedAnalysis',
  args: [data],
  // Child workflow config
  config: {
    maximumAttempts: 5,
    backoffCoefficient: 2
  }
});

// Fire-and-forget child workflow
await MemFlow.workflow.startChild({
  taskQueue: 'notifications',
  workflowName: 'sendUpdates',
  args: [updates]
});
Copy

5. Workflow Interceptors

Add cross-cutting concerns through interceptors that run as durable functions:

// Add audit interceptor that uses MemFlow functions
MemFlow.registerInterceptor({
  async execute(ctx, next) {
    try {
      // Interceptors can use MemFlow functions and participate in replay
      const entity = await MemFlow.workflow.entity();
      await entity.append('auditLog', {
        action: 'started',
        timestamp: new Date().toISOString()
      });

      // Rate limiting with durable sleep
      await MemFlow.workflow.sleepFor('100 milliseconds');

      const result = await next();

      await entity.append('auditLog', {
        action: 'completed',
        timestamp: new Date().toISOString()
      });

      return result;
    } catch (err) {
      // CRITICAL: Always check for HotMesh interruptions
      if (MemFlow.didInterrupt(err)) {
        throw err; // Rethrow for replay system
      }
      throw err;
    }
  }
});
Copy

Basic Usage Example

import { Client, Worker, MemFlow } from '@hotmeshio/hotmesh';
import { Client as Postgres } from 'pg';

// Initialize worker
await Worker.create({
  connection: {
    class: Postgres,
    options: { connectionString: 'postgresql://usr:pwd@localhost:5432/db' }
  },
  taskQueue: 'default',
  workflow: workflows.example
});

// Initialize client
const client = new Client({
  connection: {
    class: Postgres,
    options: { connectionString: 'postgresql://usr:pwd@localhost:5432/db' }
  }
});

// Start workflow
const handle = await client.workflow.start({
  args: ['input data'],
  taskQueue: 'default',
  workflowName: 'example',
  workflowId: MemFlow.guid()
});

// Get result
const result = await handle.result();

// Cleanup
await MemFlow.shutdown();
Copy