npm install @siglo/core @siglo/postgres pg
npm install --save-dev @types/pg

-- 1. Checkpoint Storage Table (with OCC version tracking)
CREATE TABLE IF NOT EXISTS siglo_checkpoints (
    stack_id VARCHAR(255) PRIMARY KEY,
    namespace VARCHAR(255),
    state JSONB NOT NULL,
    version BIGINT NOT NULL DEFAULT 1,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
ALTER TABLE siglo_checkpoints ADD COLUMN IF NOT EXISTS version BIGINT NOT NULL DEFAULT 1;
CREATE INDEX IF NOT EXISTS idx_siglo_checkpoints_namespace ON siglo_checkpoints(namespace);

-- 2. Step Context Mutation Audit Event Log
CREATE TABLE IF NOT EXISTS siglo_context_events (
    id BIGSERIAL PRIMARY KEY,
    stack_id VARCHAR(255) NOT NULL,
    step_id VARCHAR(255) NOT NULL,
    namespace VARCHAR(255),
    field VARCHAR(255) NOT NULL,
    value JSONB,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_siglo_context_events_stack ON siglo_context_events(stack_id);

-- 3. Durable Wait and Continuation Records Table
CREATE TABLE IF NOT EXISTS siglo_wait_records (
    correlation_key VARCHAR(255) PRIMARY KEY,
    stack_id VARCHAR(255) NOT NULL,
    namespace VARCHAR(255),
    event_name VARCHAR(255),
    status VARCHAR(50) NOT NULL DEFAULT 'WAITING',
    continuation_step_index INT NOT NULL DEFAULT 0,
    version VARCHAR(50),
    data JSONB,
    resumed_by VARCHAR(255),
    resumed_at TIMESTAMPTZ,
    expires_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
ALTER TABLE siglo_wait_records ADD COLUMN IF NOT EXISTS resume_payload JSONB;
ALTER TABLE siglo_wait_records ADD COLUMN IF NOT EXISTS locked_by TEXT;
ALTER TABLE siglo_wait_records ADD COLUMN IF NOT EXISTS locked_at TIMESTAMPTZ;
CREATE INDEX IF NOT EXISTS idx_siglo_wait_records_stack ON siglo_wait_records(stack_id);
CREATE INDEX IF NOT EXISTS idx_siglo_wait_records_status ON siglo_wait_records(status);

-- 4. Distributed Idempotency Table
CREATE TABLE IF NOT EXISTS siglo_idempotency (
    key VARCHAR(512) PRIMARY KEY,
    status VARCHAR(50) NOT NULL,
    result JSONB,
    expires_at TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_siglo_idempotency_expires ON siglo_idempotency(expires_at);

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

SELECT pg_try_advisory_lock(hashtextextended('siglo:workflow:' || stackId, 0)) AS locked

UPDATE siglo_wait_records
SET status = 'RESUMING', locked_at = NOW(), locked_by = $1,
    resume_payload = COALESCE(resume_payload, $3::jsonb), updated_at = NOW()
WHERE correlation_key = $2
  AND (status = 'WAITING' OR (status = 'RESUMING' AND $5))
  AND (resume_payload IS NOT NULL OR $4 OR expires_at IS NULL OR expires_at > NOW())
RETURNING *

import { PostgresIdempotencyStore } from '@siglo/postgres';

{
    status: string;           // 'in-progress' | 'completed'
    result?: any;             // Parsed JSON result
    expiresAt?: number;       // Expiry timestamp (ms)
    claimedAt: number;        // Created timestamp (ms)
}

import { OptimisticLockException, DuplicateEventException, ExpiredWaitException } from '@siglo/core';

import { Pool } from 'pg';
import { Step, flow, OptimisticLockException, DuplicateEventException } from '@siglo/core';
import { PostgresStorageAdapter, PostgresIdempotencyStore } from '@siglo/postgres';

// 1. Initialize PostgreSQL Connection Pool
const pool = new Pool({
    connectionString: process.env.DATABASE_URL || 'postgresql://postgres:postgres@localhost:5432/orders_db',
    max: 20,
    idleTimeoutMillis: 30000,
    connectionTimeoutMillis: 5000,
});

pool.on('error', err => {
    console.error('Unexpected idle PostgreSQL client error:', err);
});

// 2. Initialize and Register Storage Adapter
const storageAdapter = new PostgresStorageAdapter({
    client: pool,
    checkpointsTable: 'siglo_checkpoints',
    eventsTable: 'siglo_context_events',
    waitRecordsTable: 'siglo_wait_records',
    recordContextChanges: true,
});

// 3. Initialize Idempotency Store
const idempotencyStore = new PostgresIdempotencyStore({
    client: pool,
    tableName: 'siglo_idempotency',
    defaultTtlMs: 86400000, // 24 hours
});

// 4. Schema Initialization on Application Startup
export async function initializeDatabase() {
    await storageAdapter.ensureTables();
    await idempotencyStore.ensureTable();
    console.log('PostgreSQL workflow schemas initialized successfully.');
}

// Register global storage adapter
Step.useStorage(storageAdapter);

// 5. Define Production Workflow Steps
interface OrderInput {
    orderId: string;
    accountId: string;
    amountCents: number;
}

const debitAccount = Step.create('debitAccount', async (param: OrderInput) => {
    // Execute debit operation
    return {
        debitReference: `dr_${param.orderId}`,
        debitedAmount: param.amountCents,
    };
});

const emitReceipt = Step.create('emitReceipt', async (param: any) => {
    return {
        receiptNumber: `rcpt_${param.orderId}`,
        completedAt: new Date().toISOString(),
    };
});

const paymentWorkflow = flow(debitAccount, emitReceipt);

// 6. Execute Workflow with Distributed Idempotency and Workflow Lock
export async function processOrder(input: OrderInput) {
    const idempKey = `order:payment:${input.orderId}`;

    // Attempt 60-second atomic claim lock
    const claimStatus = await idempotencyStore.claim(idempKey, 60000);

    if (claimStatus === 'completed') {
        console.log(`Order ${input.orderId} already completed. Returning cached result.`);
        return await idempotencyStore.getResult(idempKey);
    }

    if (claimStatus === 'in-progress') {
        throw new Error(`Order ${input.orderId} is currently being processed by another worker.`);
    }

    try {
        // Execute inside distributed advisory workflow lock
        const stackId = `stack_${input.orderId}`;
        const output = await storageAdapter.withWorkflowLock(stackId, async () => {
            return await paymentWorkflow.run(input, {
                namespace: 'ecommerce-prod',
            });
        });

        const finalResult = {
            orderId: input.orderId,
            debitReference: output.debitReference,
            receiptNumber: output.receiptNumber,
            completedAt: output.completedAt,
        };

        // Complete idempotency record with final payload
        await idempotencyStore.complete(idempKey, finalResult);
        return finalResult;
    } catch (err) {
        // Clear in-flight idempotency claim so retries are permitted
        await idempotencyStore.fail(idempKey, err);

        if (err instanceof OptimisticLockException) {
            console.error(
                `OCC Conflict on stack ${err.stackId}: expected v${err.expectedVersion}, got v${err.currentVersion}`,
            );
        } else if (err instanceof DuplicateEventException) {
            console.warn(`Concurrent execution conflict on lock ${err.correlationKey}`);
        }

        throw err;
    }
}

// 7. Scheduled Maintenance & Retention Cleanup
export async function runNightlyRetentionCleanup() {
    console.log('Starting automated PostgreSQL workflow data retention cleanup...');

    const prunedCheckpoints = await storageAdapter.cleanupOldCheckpoints(14); // 14-day checkpoint retention
    const prunedEvents = await storageAdapter.cleanupOldEvents(30); // 30-day audit log retention
    const prunedWaits = await storageAdapter.cleanupOldWaitRecords(14); // 14-day wait record retention
    const prunedExpiredCheckpoints = await storageAdapter.cleanExpiredCheckpoints();
    const prunedIdemp = await idempotencyStore.cleanupExpired();

    console.log(`Maintenance Summary:
      - Deleted Checkpoints: ${prunedCheckpoints}
      - Expired Checkpoints: ${prunedExpiredCheckpoints}
      - Deleted Context Events: ${prunedEvents}
      - Deleted Wait Records: ${prunedWaits}
      - Expired Idempotency Keys: ${prunedIdemp}`);
}

// 8. Graceful Shutdown
export async function shutdown() {
    await pool.end();
    console.log('PostgreSQL connection pool closed.');
}