Transaction Operations

/**
 * Run a transactional function with automatic retry on retryable errors.
 * Commits transaction automatically after function completes successfully.
 * Blocks until transaction commits.
 * 
 * Type Parameters:
 * - T: Return type of the transactional function
 * 
 * Parameters:
 * - retryable: Function<? super Transaction, T> - Function to execute transactionally
 * 
 * Returns:
 * T - Result of the transactional function
 * 
 * Throws:
 * FDBException - If transaction fails with non-retryable error
 */
<T> T Database.run(Function<? super Transaction, T> retryable);

/**
 * Run a transactional function with custom executor.
 * 
 * Type Parameters:
 * - T: Return type of the transactional function
 * 
 * Parameters:
 * - retryable: Function<? super Transaction, T> - Function to execute transactionally
 * - e: Executor - Executor for asynchronous operations
 * 
 * Returns:
 * T - Result of the transactional function
 */
<T> T Database.run(Function<? super Transaction, T> retryable, Executor e);

/**
 * Run a transactional function asynchronously with automatic retry.
 * Returns immediately with CompletableFuture.
 * 
 * Type Parameters:
 * - T: Return type of the transactional function
 * 
 * Parameters:
 * - retryable: Function<? super Transaction, ? extends CompletableFuture<T>>
 *              Async function to execute transactionally
 * 
 * Returns:
 * CompletableFuture<T> - Future that completes with transaction result
 */
<T> CompletableFuture<T> Database.runAsync(
    Function<? super Transaction, ? extends CompletableFuture<T>> retryable
);

/**
 * Run a transactional function asynchronously with custom executor.
 * 
 * Type Parameters:
 * - T: Return type of the transactional function
 * 
 * Parameters:
 * - retryable: Function<? super Transaction, ? extends CompletableFuture<T>>
 *              Async function to execute transactionally
 * - e: Executor - Executor for asynchronous operations
 * 
 * Returns:
 * CompletableFuture<T> - Future that completes with transaction result
 */
<T> CompletableFuture<T> Database.runAsync(
    Function<? super Transaction, ? extends CompletableFuture<T>> retryable, Executor e
);

import com.apple.foundationdb.*;

// Blocking transaction with automatic retry
String result = db.run(tr -> {
    tr.set("key".getBytes(), "value".getBytes());
    byte[] value = tr.get("key".getBytes()).join();
    return new String(value);
});

// Async transaction with automatic retry
CompletableFuture<Void> future = db.runAsync(tr -> {
    return tr.get("key".getBytes())
        .thenCompose(value -> {
            tr.set("another_key".getBytes(), value);
            return tr.commit();
        });
});

// Multiple operations in one transaction
db.run(tr -> {
    // Read
    byte[] value1 = tr.get("key1".getBytes()).join();
    
    // Conditional write
    if (value1 != null) {
        tr.set("key2".getBytes(), value1);
    }
    
    // Atomic operation
    tr.mutate(MutationType.ADD, "counter".getBytes(), 
              ByteBuffer.allocate(8).putLong(1).array());
    
    return null;
});

/**
 * Run a read-only function with automatic retry.
 * Provides snapshot view of database.
 * 
 * Type Parameters:
 * - T: Return type of the read function
 * 
 * Parameters:
 * - retryable: Function<? super ReadTransaction, T> - Read function to execute
 * 
 * Returns:
 * T - Result of the read function
 */
<T> T Database.read(Function<? super ReadTransaction, T> retryable);

/**
 * Run a read-only function with custom executor.
 * 
 * Type Parameters:
 * - T: Return type of the read function
 * 
 * Parameters:
 * - retryable: Function<? super ReadTransaction, T> - Read function to execute
 * - e: Executor - Executor for asynchronous operations
 * 
 * Returns:
 * T - Result of the read function
 */
<T> T Database.read(Function<? super ReadTransaction, T> retryable, Executor e);

/**
 * Run a read-only function asynchronously with automatic retry.
 * 
 * Type Parameters:
 * - T: Return type of the read function
 * 
 * Parameters:
 * - retryable: Function<? super ReadTransaction, ? extends CompletableFuture<T>>
 *              Async read function to execute
 * 
 * Returns:
 * CompletableFuture<T> - Future that completes with read result
 */
<T> CompletableFuture<T> Database.readAsync(
    Function<? super ReadTransaction, ? extends CompletableFuture<T>> retryable
);

/**
 * Run a read-only function asynchronously with custom executor.
 * 
 * Type Parameters:
 * - T: Return type of the read function
 * 
 * Parameters:
 * - retryable: Function<? super ReadTransaction, ? extends CompletableFuture<T>>
 *              Async read function to execute
 * - e: Executor - Executor for asynchronous operations
 * 
 * Returns:
 * CompletableFuture<T> - Future that completes with read result
 */
<T> CompletableFuture<T> Database.readAsync(
    Function<? super ReadTransaction, ? extends CompletableFuture<T>> retryable, Executor e
);

// Read-only transaction
List<String> keys = db.read(tr -> {
    List<String> result = new ArrayList<>();
    for (KeyValue kv : tr.getRange("prefix".getBytes(), "prefiy".getBytes())) {
        result.add(new String(kv.getKey()));
    }
    return result;
});

// Async read-only transaction
CompletableFuture<byte[]> valueFuture = db.readAsync(tr -> {
    return tr.get("key".getBytes());
});

/**
 * Create a new transaction for manual management.
 * Caller responsible for commit and error handling.
 * 
 * Returns:
 * Transaction - New transaction object (must be closed)
 */
Transaction Database.createTransaction();

/**
 * Create a new transaction with custom executor.
 * 
 * Parameters:
 * - e: Executor - Executor for asynchronous operations
 * 
 * Returns:
 * Transaction - New transaction object (must be closed)
 */
Transaction Database.createTransaction(Executor e);

/**
 * Create a new transaction with custom executor and event tracking.
 * 
 * Parameters:
 * - e: Executor - Executor for asynchronous operations
 * - eventKeeper: EventKeeper - Instrumentation for tracking operations
 * 
 * Returns:
 * Transaction - New transaction object (must be closed)
 */
Transaction Database.createTransaction(Executor e, EventKeeper eventKeeper);

// Manual transaction with retry logic
while (true) {
    try (Transaction tr = db.createTransaction()) {
        // Perform operations
        tr.set("key".getBytes(), "value".getBytes());
        byte[] value = tr.get("another_key".getBytes()).join();
        
        // Commit
        tr.commit().join();
        break; // Success
        
    } catch (FDBException e) {
        // Check if retryable
        if (!e.isRetryable()) {
            throw e; // Non-retryable error
        }
        // Retry on next iteration
    }
}

/**
 * Commit the transaction, making all changes durable.
 * Must be called on all transactions, including read-only.
 * 
 * Returns:
 * CompletableFuture<Void> - Completes when commit succeeds
 * 
 * Throws:
 * FDBException - If commit fails due to conflicts or other errors
 */
CompletableFuture<Void> Transaction.commit();

/**
 * Get the version at which the transaction was committed.
 * Only valid after successful commit.
 * 
 * Returns:
 * Long - Commit version, or null if not yet committed
 */
Long Transaction.getCommittedVersion();

/**
 * Get the versionstamp used by versionstamped operations in this transaction.
 * Only valid after successful commit for transactions using versionstamps.
 * 
 * Returns:
 * CompletableFuture<byte[]> - 10-byte transaction versionstamp
 */
CompletableFuture<byte[]> Transaction.getVersionstamp();

/**
 * Get approximate transaction size in bytes before commit.
 * Useful for staying under transaction size limits.
 * 
 * Returns:
 * CompletableFuture<Long> - Approximate size in bytes
 */
CompletableFuture<Long> Transaction.getApproximateSize();

/**
 * Cancel the transaction, abandoning all operations.
 * Transaction cannot be used after cancellation.
 */
void Transaction.cancel();

/**
 * Close transaction and release resources.
 * Does not commit - must call commit() explicitly.
 */
void Transaction.close();

// Get commit version
try (Transaction tr = db.createTransaction()) {
    tr.set("key".getBytes(), "value".getBytes());
    tr.commit().join();
    
    Long version = tr.getCommittedVersion();
    System.out.println("Committed at version: " + version);
}

// Check transaction size
try (Transaction tr = db.createTransaction()) {
    for (int i = 0; i < 1000; i++) {
        tr.set(("key" + i).getBytes(), new byte[1000]);
    }
    
    long size = tr.getApproximateSize().join();
    if (size > 9_000_000) { // Near 10MB limit
        System.out.println("Warning: Transaction size approaching limit");
    }
    
    tr.commit().join();
}

// Get versionstamp after commit
try (Transaction tr = db.createTransaction()) {
    Tuple key = Tuple.from("log", Versionstamp.incomplete());
    tr.mutate(MutationType.SET_VERSIONSTAMPED_KEY,
              key.packWithVersionstamp(), "data".getBytes());
    tr.commit().join();
    
    byte[] versionstamp = tr.getVersionstamp().join();
    System.out.println("Transaction versionstamp: " + 
                      ByteArrayUtil.printable(versionstamp));
}

/**
 * Handle a transaction error and prepare for retry.
 * Returns reset transaction if error is retryable.
 * Throws if error is not retryable.
 * 
 * Original transaction object is invalid after this call.
 * 
 * Parameters:
 * - e: Throwable - Error that occurred during transaction
 * 
 * Returns:
 * CompletableFuture<Transaction> - Reset transaction for retry
 * 
 * Throws:
 * FDBException - If error is not retryable
 */
CompletableFuture<Transaction> Transaction.onError(Throwable e);

// Manual retry loop with onError
Transaction tr = db.createTransaction();
try {
    while (true) {
        try {
            // Perform operations
            tr.set("key".getBytes(), "value".getBytes());
            tr.commit().join();
            break; // Success
            
        } catch (Throwable e) {
            // Let onError determine if retryable
            tr = tr.onError(e).join(); // Returns reset transaction
        }
    }
} finally {
    tr.close();
}

/**
 * Create transaction scoped to tenant key-space.
 * 
 * Returns:
 * Transaction - New transaction in tenant context
 */
Transaction Tenant.createTransaction();

/**
 * Create transaction with custom executor.
 * 
 * Parameters:
 * - e: Executor - Executor for asynchronous operations
 * 
 * Returns:
 * Transaction - New transaction in tenant context
 */
Transaction Tenant.createTransaction(Executor e);

/**
 * Create transaction with custom executor and event tracking.
 * 
 * Parameters:
 * - e: Executor - Executor for asynchronous operations
 * - eventKeeper: EventKeeper - Instrumentation for tracking operations
 * 
 * Returns:
 * Transaction - New transaction in tenant context
 */
Transaction Tenant.createTransaction(Executor e, EventKeeper eventKeeper);

/**
 * Run transactional function with automatic retry in tenant context.
 * 
 * Type Parameters:
 * - T: Return type
 * 
 * Parameters:
 * - retryable: Function<? super Transaction, T> - Function to execute
 * 
 * Returns:
 * T - Result of function
 */
<T> T Tenant.run(Function<? super Transaction, T> retryable);

/**
 * Run async transactional function with automatic retry in tenant context.
 * 
 * Type Parameters:
 * - T: Return type
 * 
 * Parameters:
 * - retryable: Function<? super Transaction, ? extends CompletableFuture<T>>
 * 
 * Returns:
 * CompletableFuture<T> - Future completing with result
 */
<T> CompletableFuture<T> Tenant.runAsync(
    Function<? super Transaction, ? extends CompletableFuture<T>> retryable
);

// Open tenant
Tenant tenant = db.openTenant("tenant1".getBytes());

// Run transaction in tenant
tenant.run(tr -> {
    // These operations are scoped to tenant key-space
    tr.set("key".getBytes(), "value".getBytes());
    return null;
});

/**
 * Run transactional function in this transaction's context.
 * Does not create new transaction - uses this transaction.
 * Does not automatically retry.
 *
 * Type Parameters:
 * - T: Return type
 *
 * Parameters:
 * - retryable: Function<? super Transaction, T> - Function to execute
 *
 * Returns:
 * T - Result of function
 */
<T> T Transaction.run(Function<? super Transaction, T> retryable);

/**
 * Run async transactional function in this transaction's context.
 *
 * Type Parameters:
 * - T: Return type
 *
 * Parameters:
 * - retryable: Function<? super Transaction, ? extends CompletableFuture<T>>
 *
 * Returns:
 * CompletableFuture<T> - Future completing with result
 */
<T> CompletableFuture<T> Transaction.runAsync(
    Function<? super Transaction, ? extends CompletableFuture<T>> retryable
);

/**
 * Get parent database for this transaction.
 *
 * Returns:
 * Database - Parent database object
 */
Database Transaction.getDatabase();

/**
 * Add range of keys to transaction's read conflict ranges.
 * Other transactions that write keys in this range may cause conflict.
 *
 * Parameters:
 * - keyBegin: byte[] - First key in range (inclusive)
 * - keyEnd: byte[] - Ending key for range (exclusive)
 */
void Transaction.addReadConflictRange(byte[] keyBegin, byte[] keyEnd);

/**
 * Add key to transaction's read conflict ranges.
 * Other transactions that write this key may cause conflict.
 *
 * Parameters:
 * - key: byte[] - Key to add to read conflict range
 */
void Transaction.addReadConflictKey(byte[] key);

/**
 * Add range of keys to transaction's write conflict ranges.
 * Other transactions that read keys in this range may fail with conflict.
 *
 * Parameters:
 * - keyBegin: byte[] - First key in range (inclusive)
 * - keyEnd: byte[] - Ending key for range (exclusive)
 */
void Transaction.addWriteConflictRange(byte[] keyBegin, byte[] keyEnd);

/**
 * Add key to transaction's write conflict ranges.
 * Other transactions that read this key may fail with conflict.
 *
 * Parameters:
 * - key: byte[] - Key to add to write conflict range
 */
void Transaction.addWriteConflictKey(byte[] key);

/**
 * Add read conflict range if this is not a snapshot view.
 * Returns false if this is a snapshot (no conflict range added).
 *
 * Parameters:
 * - keyBegin: byte[] - First key in range (inclusive)
 * - keyEnd: byte[] - Ending key for range (exclusive)
 *
 * Returns:
 * boolean - true if conflict range added, false if snapshot
 */
boolean ReadTransaction.addReadConflictRangeIfNotSnapshot(byte[] keyBegin, byte[] keyEnd);

/**
 * Add read conflict key if this is not a snapshot view.
 * Returns false if this is a snapshot (no conflict range added).
 *
 * Parameters:
 * - key: byte[] - Key to add to read conflict range
 *
 * Returns:
 * boolean - true if conflict key added, false if snapshot
 */
boolean ReadTransaction.addReadConflictKeyIfNotSnapshot(byte[] key);

// Manually add conflict ranges for serializable isolation
try (Transaction tr = db.createTransaction()) {
    // Read from external source (not through transaction)
    byte[] externalValue = readFromExternalSystem();

    // Manually declare that we "read" this range
    // This ensures proper isolation even though read was external
    tr.addReadConflictRange("key1".getBytes(), "key2".getBytes());

    // Write based on external read
    tr.set("result".getBytes(), externalValue);
    tr.commit().join();
}

// Declare write conflicts for manual locking
try (Transaction tr = db.createTransaction()) {
    // Declare write conflict without actually writing
    // Other transactions reading this range will conflict
    tr.addWriteConflictRange("lock_start".getBytes(), "lock_end".getBytes());

    // Perform protected operations
    tr.set("protected_key".getBytes(), "value".getBytes());
    tr.commit().join();
}

/**
 * Create a watch that triggers when the value at the key changes.
 * The watch remains active after transaction commits until triggered or cancelled.
 * Must call commit() for watch to be registered.
 *
 * By default, maximum 10,000 active watches per database connection.
 * Watch automatically cancelled on key change, or can be cancelled manually.
 *
 * Parameters:
 * - key: byte[] - Key to watch for value changes
 *
 * Returns:
 * CompletableFuture<Void> - Completes when value changes or watch cancelled
 *
 * Throws:
 * FDBException - If too many watches exist (configurable with DatabaseOptions.setMaxWatches)
 */
CompletableFuture<Void> Transaction.watch(byte[] key);

// Watch a key for changes
CompletableFuture<Void> watchFuture;

try (Transaction tr = db.createTransaction()) {
    byte[] currentValue = tr.get("watched_key".getBytes()).join();

    // Create watch
    watchFuture = tr.watch("watched_key".getBytes());

    // Must commit for watch to be registered
    tr.commit().join();
}

// Wait for value to change (in another thread/context)
watchFuture.thenRun(() -> {
    System.out.println("Key value changed!");

    // Read new value
    db.run(tr -> {
        byte[] newValue = tr.get("watched_key".getBytes()).join();
        System.out.println("New value: " + new String(newValue));
        return null;
    });
});

// Another transaction changes the value, triggering watch
db.run(tr -> {
    tr.set("watched_key".getBytes(), "new_value".getBytes());
    return null;
});

/**
 * Get the version at which reads will access the database.
 *
 * Returns:
 * CompletableFuture<Long> - Database version for reads
 */
CompletableFuture<Long> ReadTransaction.getReadVersion();

/**
 * Set the version at which to execute reads.
 * Overrides normal version determination.
 * Setting version too far in past causes transaction_too_old errors.
 *
 * Parameters:
 * - version: long - Database version for reads
 */
void ReadTransaction.setReadVersion(long version);

// Read at specific version
long targetVersion = 12345678L;

db.read(tr -> {
    // Set specific read version
    tr.setReadVersion(targetVersion);

    // All reads use this version
    byte[] value = tr.get("key".getBytes()).join();
    return value;
});

// Read at same version as another transaction
long version1 = db.run(tr -> {
    tr.set("key".getBytes(), "value".getBytes());
    tr.commit().join();
    return tr.getCommittedVersion();
});

db.read(tr -> {
    // Read at the committed version of previous transaction
    tr.setReadVersion(version1);
    byte[] value = tr.get("key".getBytes()).join();
    return null;
});

/**
 * Get whether this is a snapshot view with relaxed isolation.
 * Snapshot reads don't add read conflict ranges.
 *
 * Returns:
 * boolean - true if snapshot view, false if normal transaction
 */
boolean ReadTransaction.isSnapshot();

/**
 * Return read-only snapshot view of the database.
 * Snapshot reads reduce conflicts but require careful reasoning about concurrency.
 *
 * Returns:
 * ReadTransaction - Snapshot view with relaxed isolation
 */
ReadTransaction ReadTransaction.snapshot();

// Use snapshot reads to avoid conflicts on frequently updated keys
db.run(tr -> {
    // Normal read - adds conflict range
    byte[] criticalData = tr.get("critical".getBytes()).join();

    // Snapshot read - no conflict range, won't conflict with other transactions
    ReadTransaction snapshot = tr.snapshot();
    byte[] statsData = snapshot.get("statistics".getBytes()).join();

    // Write based on critical data only
    // Won't conflict even if statistics key is updated by other transactions
    tr.set("result".getBytes(), criticalData);
    return null;
});

// Check if snapshot view
db.read(tr -> {
    boolean isSnap = tr.isSnapshot(); // false

    ReadTransaction snapshot = tr.snapshot();
    boolean isSnap2 = snapshot.isSnapshot(); // true

    return null;
});

// Snapshot iteration over large datasets
db.run(tr -> {
    ReadTransaction snapshot = tr.snapshot();

    // Iterate without conflicts
    for (KeyValue kv : snapshot.getRange("data_".getBytes(), "data`".getBytes())) {
        // Process data without adding conflict ranges
        processData(kv.getKey(), kv.getValue());
    }

    return null;
});

interface Transaction extends AutoCloseable, ReadTransaction, TransactionContext {
    // All methods documented above
}

interface ReadTransaction extends ReadTransactionContext {
    // Read operations documented in data-operations.md
}

interface TransactionContext extends ReadTransactionContext {
    <T> T run(Function<? super Transaction, T> retryable);
    <T> CompletableFuture<T> runAsync(
        Function<? super Transaction, ? extends CompletableFuture<T>> retryable
    );
}

interface ReadTransactionContext {
    <T> T read(Function<? super ReadTransaction, T> retryable);
    <T> CompletableFuture<T> readAsync(
        Function<? super ReadTransaction, ? extends CompletableFuture<T>> retryable
    );
    Executor getExecutor();
}

interface Tenant extends AutoCloseable, TransactionContext {
    // Tenant methods documented in tenant-management.md
}

class FDBException extends RuntimeException {
    int getCode();
    String getMessage();
    boolean isRetryable();
    boolean isRetryableNotCommitted();
    boolean isSuccess();
}