Storage and Adapters

export interface StorageAdapter {
    name: string;

    // 1. Context & Event Streaming
    onContextChange?(entry: ContextChangeEvent): Promise<void> | void;

    // 2. Checkpointing & Durability
    saveCheckpoint?(checkpoint: SerializedExecutionStack): Promise<void>;
    getCheckpoint?(stackId: string): Promise<SerializedExecutionStack | undefined>;
    cleanExpiredCheckpoints?(): Promise<number> | number | void;

    // 3. Durable Wait Persistence & Workflow Locking
    saveWaitRecord?(record: any): Promise<void> | void;
    getWaitRecord?(correlationKeyOrStackId: string): Promise<any | undefined>;
    claimWaitRecord?(
        correlationKey: string,
        claimantId?: string,
        options?: { allowExpired?: boolean; eventPayload?: any },
    ): Promise<any>;
    completeWaitRecord?(correlationKey: string, claimantId: string): Promise<void>;
    withWorkflowLock?<T>(stackId: string, action: () => Promise<T>): Promise<T>;
    cancelWaitRecord?(correlationKey: string): Promise<void>;
    cancelWorkflow?(stackId: string): Promise<void>;

    // 4. Distributed Caching
    getCachedStep?(key: string): Promise<any | undefined>;
    setCachedStep?(key: string, value: any, ttlMs: number): Promise<void>;

    // 5. Cache & Telemetry Events
    onCacheEvent?(event: CacheEvent): Promise<void> | void;
}

import { useStorage } from '@siglo/core';
import { PostgresStorageAdapter } from '@siglo/postgres';

const adapter = new PostgresStorageAdapter({ client: pgPool });
useStorage(adapter);

import { Pool } from 'pg';
import { PostgresStorageAdapter, PostgresIdempotencyStore } from '@siglo/postgres';

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const storage = new PostgresStorageAdapter({ client: pool });
await storage.ensureTables();
useStorage(storage);

import { Redis } from 'ioredis';
import { RedisStorageAdapter, RedisLock } from '@siglo/redis';

const redis = new Redis(process.env.REDIS_URL);
const storage = new RedisStorageAdapter({ client: redis, ttlMs: 86400000 });
useStorage(storage);

await workflow.run(input, {
    namespace: 'tenant_acme_corp',
    checkpoint: true, // Persists intermediate step snapshots to storage adapter
    durable: true, // Enables wait record persistence and suspension
});

import { Step, flow, useStorage, StorageAdapter } from '@siglo/core';

const customAuditAdapter: StorageAdapter = {
    name: 'console-audit',
    onContextChange: evt => {
        console.log(`[AUDIT] Step ${evt.stepId} changed field ${evt.field}`);
    },
};

useStorage(customAuditAdapter);

const s1 = Step.create('init', () => ({ initialized: true }));
const wf = flow(s1);
await wf.run();

import { setErrorMode } from '@siglo/core';

// In mission-critical environments: 'strict' rejects unhandled storage failures on awaited persistence
setErrorMode('strict');

// In resilient latency-sensitive pipelines: 'warn' logs warnings and continues
setErrorMode('warn');

import { expect } from 'chai';
import { InMemoryStorageFake, createWorkflowHarness } from '@siglo/testing';
import { Step, flow } from '@siglo/core';

const wf = flow(Step.create('test', () => ({ ok: true })));

describe('Storage Adapter Testing', () => {
    it('captures context changes in memory fake', async () => {
        const fakeStorage = new InMemoryStorageFake();
        const harness = createWorkflowHarness(wf, { storage: fakeStorage });

        await harness.run();
        expect(fakeStorage.contextChanges.length).to.be.greaterThan(0);
    });
});