/** * Comprehensive Transaction Example * * Demonstrates distributed transactions with ACID-like semantics */ import { createConfig, ElasticsearchConnectionManager, LogLevel, createTransactionManager, type TransactionCallbacks, type ConflictInfo, } from '../../index.js'; interface BankAccount { accountId: string; balance: number; currency: string; lastUpdated: Date; } interface Order { orderId: string; customerId: string; items: Array<{ productId: string; quantity: number; price: number }>; total: number; status: 'pending' | 'confirmed' | 'cancelled'; createdAt: Date; } interface Inventory { productId: string; quantity: number; reserved: number; lastUpdated: Date; } async function main() { console.log('=== Transaction System Example ===\n'); // ============================================================================ // Step 1: Configuration // ============================================================================ console.log('Step 1: Configuring Elasticsearch connection...'); const config = createConfig() .fromEnv() .nodes(process.env.ELASTICSEARCH_URL || 'http://localhost:9200') .basicAuth( process.env.ELASTICSEARCH_USERNAME || 'elastic', process.env.ELASTICSEARCH_PASSWORD || 'changeme' ) .timeout(30000) .retries(3) .logLevel(LogLevel.INFO) .enableMetrics(true) .build(); // ============================================================================ // Step 2: Initialize Connection and Transaction Manager // ============================================================================ console.log('Step 2: Initializing connection and transaction manager...'); const connectionManager = ElasticsearchConnectionManager.getInstance(config); await connectionManager.initialize(); const transactionManager = createTransactionManager({ defaultIsolationLevel: 'read_committed', defaultLockingStrategy: 'optimistic', defaultTimeout: 30000, maxConcurrentTransactions: 100, conflictResolution: 'retry', enableLogging: true, enableMetrics: true, }); await transactionManager.initialize(); console.log('✓ Connection and transaction manager initialized\n'); // ============================================================================ // Step 3: Setup Test Data // ============================================================================ console.log('Step 3: Setting up test data...'); const client = connectionManager.getClient(); // Create test indices for (const index of ['accounts', 'orders', 'inventory']) { try { await client.indices.create({ index }); } catch (error) { // Index might already exist } } // Create test accounts await client.index({ index: 'accounts', id: 'acc-001', document: { accountId: 'acc-001', balance: 1000, currency: 'USD', lastUpdated: new Date(), }, }); await client.index({ index: 'accounts', id: 'acc-002', document: { accountId: 'acc-002', balance: 500, currency: 'USD', lastUpdated: new Date(), }, }); // Create test inventory await client.index({ index: 'inventory', id: 'prod-001', document: { productId: 'prod-001', quantity: 100, reserved: 0, lastUpdated: new Date(), }, }); console.log('✓ Test data created\n'); // ============================================================================ // Step 4: Simple Transaction - Money Transfer // ============================================================================ console.log('Step 4: Simple transaction - money transfer...'); const transferTxn = await transactionManager.begin({ isolationLevel: 'read_committed', autoRollback: true, }); try { // Read source account const sourceAccount = await transferTxn.read('accounts', 'acc-001'); console.log(` Source balance before: $${sourceAccount?.balance}`); // Read destination account const destAccount = await transferTxn.read('accounts', 'acc-002'); console.log(` Destination balance before: $${destAccount?.balance}`); // Transfer amount const transferAmount = 200; if (!sourceAccount || sourceAccount.balance < transferAmount) { throw new Error('Insufficient funds'); } // Update source account await transferTxn.update('accounts', 'acc-001', { balance: sourceAccount.balance - transferAmount, lastUpdated: new Date(), }); // Update destination account await transferTxn.update('accounts', 'acc-002', { balance: destAccount!.balance + transferAmount, lastUpdated: new Date(), }); // Commit transaction const result = await transferTxn.commit(); console.log(` ✓ Transfer completed`); console.log(` Operations: ${result.operationsExecuted}`); console.log(` Duration: ${result.duration}ms`); } catch (error: any) { console.log(` ✗ Transfer failed: ${error.message}`); } console.log(); // ============================================================================ // Step 5: Transaction with Rollback // ============================================================================ console.log('Step 5: Transaction with rollback...'); const rollbackTxn = await transactionManager.begin({ autoRollback: true, }); try { const account = await rollbackTxn.read('accounts', 'acc-001'); console.log(` Balance before: $${account?.balance}`); // Update account await rollbackTxn.update('accounts', 'acc-001', { balance: account!.balance + 500, lastUpdated: new Date(), }); // Simulate error throw new Error('Simulated error - transaction will rollback'); } catch (error: any) { console.log(` ✗ Error occurred: ${error.message}`); const result = await rollbackTxn.rollback(); console.log(` ✓ Transaction rolled back`); console.log(` Operations rolled back: ${result.operationsRolledBack}`); } // Verify balance unchanged const accountAfter = await client.get({ index: 'accounts', id: 'acc-001' }); console.log(` Balance after rollback: $${(accountAfter._source as BankAccount).balance}`); console.log(); // ============================================================================ // Step 6: Transaction with Savepoints // ============================================================================ console.log('Step 6: Transaction with savepoints...'); const savepointTxn = await transactionManager.begin(); try { const account = await savepointTxn.read('accounts', 'acc-001'); console.log(` Initial balance: $${account?.balance}`); // First operation await savepointTxn.update('accounts', 'acc-001', { balance: account!.balance + 100, }); console.log(' Operation 1: +$100'); // Create savepoint savepointTxn.savepoint('after_first_op'); // Second operation await savepointTxn.update('accounts', 'acc-001', { balance: account!.balance + 200, }); console.log(' Operation 2: +$200'); // Rollback to savepoint (removes operation 2) savepointTxn.rollbackTo('after_first_op'); console.log(' Rolled back to savepoint (operation 2 removed)'); // Commit transaction (only operation 1 will be committed) await savepointTxn.commit(); console.log(' ✓ Transaction committed (only operation 1)'); } catch (error: any) { console.log(` ✗ Error: ${error.message}`); await savepointTxn.rollback(); } console.log(); // ============================================================================ // Step 7: Concurrent Transactions with Conflict // ============================================================================ console.log('Step 7: Concurrent transactions with conflict handling...'); let conflictsDetected = 0; const callbacks: TransactionCallbacks = { onConflict: async (conflict: ConflictInfo) => { conflictsDetected++; console.log(` ⚠ Conflict detected on ${conflict.operation.index}/${conflict.operation.id}`); return 'retry'; // Automatically retry }, }; // Start two concurrent transactions modifying the same document const txn1 = transactionManager.begin({ maxRetries: 5 }, callbacks); const txn2 = transactionManager.begin({ maxRetries: 5 }, callbacks); const [transaction1, transaction2] = await Promise.all([txn1, txn2]); try { // Both read the same account const [account1, account2] = await Promise.all([ transaction1.read('accounts', 'acc-001'), transaction2.read('accounts', 'acc-001'), ]); console.log(` Initial balance (txn1): $${account1?.balance}`); console.log(` Initial balance (txn2): $${account2?.balance}`); // Both try to update await transaction1.update('accounts', 'acc-001', { balance: account1!.balance + 50, }); await transaction2.update('accounts', 'acc-001', { balance: account2!.balance + 75, }); // Commit both (one will conflict and retry) const [result1, result2] = await Promise.all([ transaction1.commit(), transaction2.commit(), ]); console.log(` ✓ Transaction 1: ${result1.success ? 'committed' : 'failed'}`); console.log(` ✓ Transaction 2: ${result2.success ? 'committed' : 'failed'}`); console.log(` Conflicts detected and resolved: ${conflictsDetected}`); } catch (error: any) { console.log(` ✗ Error: ${error.message}`); } console.log(); // ============================================================================ // Step 8: Complex Multi-Document Transaction - Order Processing // ============================================================================ console.log('Step 8: Complex multi-document transaction - order processing...'); const orderTxn = await transactionManager.begin({ isolationLevel: 'repeatable_read', autoRollback: true, }); try { // Create order const order: Order = { orderId: 'ord-001', customerId: 'cust-001', items: [ { productId: 'prod-001', quantity: 5, price: 10 }, ], total: 50, status: 'pending', createdAt: new Date(), }; await orderTxn.create('orders', order.orderId, order); console.log(' Created order'); // Check and reserve inventory const inventory = await orderTxn.read('inventory', 'prod-001'); console.log(` Available inventory: ${inventory?.quantity}`); if (!inventory || inventory.quantity < 5) { throw new Error('Insufficient inventory'); } await orderTxn.update('inventory', 'prod-001', { quantity: inventory.quantity - 5, reserved: inventory.reserved + 5, lastUpdated: new Date(), }); console.log(' Reserved inventory: 5 units'); // Charge customer account const customerAccount = await orderTxn.read('accounts', 'acc-001'); if (!customerAccount || customerAccount.balance < order.total) { throw new Error('Insufficient funds'); } await orderTxn.update('accounts', 'acc-001', { balance: customerAccount.balance - order.total, lastUpdated: new Date(), }); console.log(` Charged customer: $${order.total}`); // Update order status await orderTxn.update('orders', order.orderId, { status: 'confirmed', }); console.log(' Order confirmed'); // Commit all operations atomically const result = await orderTxn.commit(); console.log(` ✓ Order processed successfully`); console.log(` Operations: ${result.operationsExecuted}`); console.log(` Duration: ${result.duration}ms`); } catch (error: any) { console.log(` ✗ Order processing failed: ${error.message}`); console.log(' All changes rolled back'); } console.log(); // ============================================================================ // Step 9: Transaction Statistics // ============================================================================ console.log('Step 9: Transaction statistics...\n'); const stats = transactionManager.getStats(); console.log('Transaction Manager Statistics:'); console.log(` Total started: ${stats.totalStarted}`); console.log(` Total committed: ${stats.totalCommitted}`); console.log(` Total rolled back: ${stats.totalRolledBack}`); console.log(` Total failed: ${stats.totalFailed}`); console.log(` Total operations: ${stats.totalOperations}`); console.log(` Total conflicts: ${stats.totalConflicts}`); console.log(` Total retries: ${stats.totalRetries}`); console.log(` Success rate: ${(stats.successRate * 100).toFixed(2)}%`); console.log(` Avg duration: ${stats.avgDuration.toFixed(2)}ms`); console.log(` Avg operations/txn: ${stats.avgOperationsPerTransaction.toFixed(2)}`); console.log(` Active transactions: ${stats.activeTransactions}`); console.log(); // ============================================================================ // Step 10: Cleanup // ============================================================================ console.log('Step 10: Cleanup...'); await transactionManager.destroy(); await connectionManager.destroy(); console.log('✓ Cleanup complete\n'); console.log('=== Transaction System Example Complete ==='); console.log('\nKey Features Demonstrated:'); console.log(' ✓ ACID-like transaction semantics'); console.log(' ✓ Optimistic concurrency control'); console.log(' ✓ Automatic rollback on error'); console.log(' ✓ Compensation-based rollback'); console.log(' ✓ Savepoints for partial rollback'); console.log(' ✓ Conflict detection and retry'); console.log(' ✓ Multi-document transactions'); console.log(' ✓ Isolation levels (read_committed, repeatable_read)'); console.log(' ✓ Transaction callbacks and hooks'); console.log(' ✓ Comprehensive statistics'); } // Run the example main().catch((error) => { console.error('Example failed:', error); process.exit(1); });