@flowcreate/core

Core library for distributed CRDT-backed finite state machines.

Installation

npm install @flowcreate/core

Features

• Type-Safe FSM Engine - Define and validate state machines with TypeScript
• CRDT Operation Layer - Automatic conflict resolution using last-write-wins
• Storage Abstraction - Pluggable storage backends (in-memory, Supabase)
• HTTP Middleware - Express and Fastify middleware included
• Comprehensive Testing - Unit tests and property-based tests

Quick Start

Basic Example

import express from 'express';
import { createExpressMiddleware, InMemoryStorage } from '@flowcreate/core';

const app = express();
app.use(express.json());

// Define your state machine
const orderFlow = {
  initial: 'pending',
  states: {
    pending: { events: { pay: 'paid', cancel: 'cancelled' } },
    paid: { events: { ship: 'shipped', refund: 'refunded' } },
    shipped: { events: { deliver: 'delivered' } },
    delivered: { events: {} },
    cancelled: { events: {} },
    refunded: { events: {} }
  }
};

// Create middleware
app.use('/flows', createExpressMiddleware({
  flows: { orders: orderFlow },
  storage: new InMemoryStorage()
}));

app.listen(3000, () => {
  console.log('Server running on http://localhost:3000');
});

API Usage

# Apply a transition
curl -X POST http://localhost:3000/flows/orders/order-123/events/pay \
  -H "Content-Type: application/json" \
  -d '{"actorId":"user-1"}'

# Response:
# {
#   "resourceId": "order-123",
#   "state": "paid",
#   "event": "pay"
# }

# Get current state
curl http://localhost:3000/flows/orders/order-123/state

# Response:
# {
#   "resourceId": "order-123",
#   "state": "paid"
# }

API Reference

State Machine Definition

Define your state machine using the StateMachineDefinition type:

import type { StateMachineDefinition } from '@flowcreate/core';

const myFlow: StateMachineDefinition = {
  initial: 'start',  // Initial state
  states: {
    start: {
      events: {
        next: 'middle',    // Event 'next' transitions to 'middle'
        skip: 'end'        // Event 'skip' transitions to 'end'
      }
    },
    middle: {
      events: {
        next: 'end',
        back: 'start'
      }
    },
    end: {
      events: {}  // Terminal state (no outgoing transitions)
    }
  }
};

FSM Engine

The FSM engine validates and executes state transitions:

import { FSMEngine } from '@flowcreate/core';

const engine = new FSMEngine(myFlow);

// Check if a transition is valid
const canTransition = engine.canTransition('start', 'next');  // true

// Execute a transition
const nextState = engine.transition('start', 'next');  // 'middle'

// Get initial state
const initial = engine.getInitialState();  // 'start'

CRDT Engine

The CRDT engine handles distributed state with conflict resolution:

import { CRDTEngine, FSMEngine } from '@flowcreate/core';
import { InMemoryStorage } from '@flowcreate/core';

const fsmEngine = new FSMEngine(myFlow);
const storage = new InMemoryStorage();
const crdtEngine = new CRDTEngine(fsmEngine, storage);

// Apply a transition
const newState = await crdtEngine.applyTransition(
  'resource-123',  // Resource ID
  'next',          // Event
  'user-1',        // Actor ID
  Date.now()       // Timestamp (optional, defaults to Date.now())
);

// Get current state
const currentState = await crdtEngine.getCurrentState('resource-123');

// Get operation history
const operations = await crdtEngine.getOperations('resource-123');

Storage Adapters

In-Memory Storage

import { InMemoryStorage } from '@flowcreate/core';

const storage = new InMemoryStorage();

Supabase Storage

import { SupabaseStorage } from '@flowcreate/core';
import { createClient } from '@supabase/supabase-js';

const supabase = createClient(
  process.env.SUPABASE_URL!,
  process.env.SUPABASE_KEY!
);

const storage = new SupabaseStorage(supabase, 'flowcreate_operations');

Custom Storage Adapter

Implement the StorageAdapter interface:

import type { StorageAdapter, TransitionOperation } from '@flowcreate/core';

class MyStorage implements StorageAdapter {
  async getState(resourceId: string): Promise<string | null> {
    // Return current state or null if resource doesn't exist
  }
  
  async saveOp(operation: TransitionOperation): Promise<void> {
    // Persist the operation
  }
  
  async listOps(resourceId: string): Promise<TransitionOperation[]> {
    // Return all operations for the resource, sorted by timestamp
  }
}

Express Middleware

import { createExpressMiddleware } from '@flowcreate/core';
import type { Request } from 'express';

const middleware = createExpressMiddleware({
  flows: {
    orders: orderFlow,
    tickets: ticketFlow,
    // ... more flows
  },
  storage: new InMemoryStorage(),
  
  // Optional: Custom actor ID extraction
  actorIdExtractor: (req: Request) => {
    return req.user?.id || req.headers['x-user-id'] || 'anonymous';
  }
});

app.use('/flows', middleware);

Endpoints:

• POST /flows/:resource/:id/events/:event - Apply a transition
• GET /flows/:resource/:id/state - Get current state

Fastify Middleware

import Fastify from 'fastify';
import { createFastifyMiddleware } from '@flowcreate/core';

const fastify = Fastify();

await fastify.register(createFastifyMiddleware({
  flows: { orders: orderFlow },
  storage: new InMemoryStorage(),
  actorIdExtractor: (req) => req.headers['x-user-id'] || 'anonymous'
}), { prefix: '/flows' });

await fastify.listen({ port: 3000 });

Error Handling

FlowCreate provides specific error types for different failure scenarios:

import {
  ValidationError,
  TransitionError,
  StorageError,
  ResourceNotFoundError
} from '@flowcreate/core';

try {
  await crdtEngine.applyTransition('order-123', 'invalid-event', 'user-1');
} catch (error) {
  if (error instanceof TransitionError) {
    console.log('Invalid transition:', error.message);
    console.log('Resource:', error.resourceId);
    console.log('Current state:', error.currentState);
    console.log('Attempted event:', error.event);
  }
}

Error Types

• ValidationError - State machine definition is invalid
• TransitionError - Attempted transition is not allowed
• StorageError - Storage operation failed
• ResourceNotFoundError - Requested resource doesn't exist

CRDT Conflict Resolution

FlowCreate uses a last-write-wins (LWW) strategy for conflict resolution:

1. Each transition is stored as an operation with a timestamp
2. Operations are sorted by timestamp (ascending)
3. Ties are broken using actor ID (lexicographic order)
4. Current state is computed by replaying operations in order
5. Invalid operations (that violate FSM rules) are skipped during replay

This ensures:

• Deterministic - Same operations always produce same state
• Commutative - Order of arrival doesn't matter
• Eventually Consistent - All nodes converge to the same state

Resource Auto-Initialization

Resources are automatically initialized when first accessed:

// First transition on a new resource
await crdtEngine.applyTransition('new-order', 'pay', 'user-1');
// Resource is created in initial state ('pending') and transition is applied

TypeScript Support

FlowCreate is written in TypeScript and provides full type definitions:

import type {
  StateMachineDefinition,
  TransitionOperation,
  StorageAdapter,
  MiddlewareConfig
} from '@flowcreate/core';

// Generic types for custom state and event names
type OrderState = 'pending' | 'paid' | 'shipped' | 'delivered';
type OrderEvent = 'pay' | 'ship' | 'deliver';

const orderFlow: StateMachineDefinition<OrderState, OrderEvent> = {
  initial: 'pending',
  states: {
    pending: { events: { pay: 'paid' } },
    paid: { events: { ship: 'shipped' } },
    shipped: { events: { deliver: 'delivered' } },
    delivered: { events: {} }
  }
};

Testing

FlowCreate includes comprehensive tests:

# Run all tests
npm test

# Run tests in watch mode
npm run test:watch

# Run specific test file
npm test -- fsm.test.ts

License

MIT © FlowCreate Contributors