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:

// Default: activities use workflow's task queue
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. Explicit Activity Registration

Register activity workers explicitly before workflows start:

// Register shared activity pool for interceptors
await MemFlow.registerActivityWorker({
  connection: {
    class: Postgres,
    options: { connectionString: 'postgresql://usr:pwd@localhost:5432/db' }
  },
  taskQueue: 'shared-activities'
}, sharedActivities, 'shared-activities');

// Register custom activity pool for specific use cases
await MemFlow.registerActivityWorker({
  connection: {
    class: Postgres,
    options: { connectionString: 'postgresql://usr:pwd@localhost:5432/db' }
  },
  taskQueue: 'priority-activities'
}, priorityActivities, 'priority-activities');
Copy

5. 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

6. Workflow Interceptors

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

// First register shared activity worker for interceptors
await MemFlow.registerActivityWorker({
  connection: {
    class: Postgres,
    options: { connectionString: 'postgresql://usr:pwd@localhost:5432/db' }
  },
  taskQueue: 'interceptor-activities'
}, { auditLog }, 'interceptor-activities');

// Add audit interceptor that uses activities with explicit taskQueue
MemFlow.registerInterceptor({
  async execute(ctx, next) {
    try {
      // Interceptors use explicit taskQueue to prevent per-workflow queues
      const { auditLog } = MemFlow.workflow.proxyActivities<typeof activities>({
        activities: { auditLog },
        taskQueue: 'interceptor-activities', // Explicit shared queue
        retryPolicy: { maximumAttempts: 3 }
      });

      await auditLog(ctx.get('workflowId'), 'started');

      const result = await next();

      await auditLog(ctx.get('workflowId'), 'completed');

      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';
import * as activities from './activities';

// (Optional) Register shared activity workers for interceptors
await MemFlow.registerActivityWorker({
  connection: {
    class: Postgres,
    options: { connectionString: 'postgresql://usr:pwd@localhost:5432/db' }
  },
  taskQueue: 'shared-activities'
}, sharedActivities, 'shared-activities');

// 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