import com.cajunsystems.*;
import com.cajunsystems.handler.Handler;

public class HelloWorld {
    
    // Define your message types with explicit replyTo
    public record HelloMessage(String name, Pid replyTo) {}
    public record GreetingResponse(String greeting) {}
    
    // Greeter actor that processes requests
    public static class GreeterHandler implements Handler<HelloMessage> {
        @Override
        public void receive(HelloMessage message, ActorContext context) {
            context.getLogger().info("Received greeting request for: {}", message.name());
            
            // Process and reply
            String greeting = "Hello, " + message.name() + "!";
            context.tell(message.replyTo(), new GreetingResponse(greeting));
        }
    }
    
    // Receiver actor that handles responses
    public static class ReceiverHandler implements Handler<GreetingResponse> {
        @Override
        public void receive(GreetingResponse message, ActorContext context) {
            System.out.println("Received: " + message.greeting());
        }
    }
    
    public static void main(String[] args) throws Exception {
        // 1. Create the ActorSystem
        ActorSystem system = new ActorSystem();
        
        // 2. Spawn actors
        Pid greeter = system.actorOf(GreeterHandler.class)
            .withId("greeter")
            .spawn();
        
        Pid receiver = system.actorOf(ReceiverHandler.class)
            .withId("receiver")
            .spawn();
        
        // 3. Send a message with explicit replyTo
        greeter.tell(new HelloMessage("World", receiver));
        
        // Wait a bit for async processing
        Thread.sleep(100);
        
        // 4. Shutdown the system when done
        system.shutdown();
    }
}

public static void main(String[] args) {
    ActorSystem system = new ActorSystem();
    
    Pid actor = system.actorOf(new Handler<String>() {
        @Override
        public void receive(String message, ActorContext context) {
            System.out.println("Received: " + message);
        }
    }).withId("demo-actor").spawn();
    
    actor.tell("Hello");
    
    // Main thread exits here, but JVM stays alive
    System.out.println("Main exiting - JVM will continue running");
    
    // To allow JVM to exit, you must explicitly shutdown:
    // system.shutdown();
}

// Graceful shutdown - waits for actors to complete current work
system.shutdown();

// Or stop individual actors
system.stopActor(actorPid);

public sealed interface GreetingMessage permits HelloMessage, ByeMessage, GetHelloCount, Shutdown {
}

public record HelloMessage() implements GreetingMessage {
}

public record ByeMessage() implements GreetingMessage {
}

public record Shutdown() implements GreetingMessage {
}

public record GetHelloCount(Pid replyTo) implements GreetingMessage {
}

public record HelloCount(int count) {
}

public class GreetingHandler implements Handler<GreetingMessage> {
    private int helloCount = 0;
    
    @Override
    public void receive(GreetingMessage message, ActorContext context) {
        switch (message) {
            case HelloMessage ignored -> {
                // Updating state of the handler
                helloCount++;
            }
            case GetHelloCount ghc -> {
                // Replying back to calling actor
                context.tell(ghc.replyTo(), new HelloCount(helloCount));
            }
            case ByeMessage ignored -> {
                // Sending a message to self
                context.tellSelf(new Shutdown());
            }
            case Shutdown ignored -> {
                // Stopping actor
                context.stop();
            }
        }
    }
    
    // Optional lifecycle methods
    @Override
    public void preStart(ActorContext context) {
        // Initialization logic
    }
    
    @Override
    public void postStop(ActorContext context) {
        // Cleanup logic
    }
    
    @Override
    public boolean onError(GreetingMessage message, Throwable exception, ActorContext context) {
        // Custom error handling
        return false; // Return true to reprocess the message
    }
}

// Create an actor with a handler class (instantiated automatically)
Pid actorPid = system.actorOf(GreetingHandler.class)
    .withId("greeter-1")  // Optional: specify ID (otherwise auto-generated)
    .spawn();

// Or create with a handler instance
GreetingHandler handler = new GreetingHandler();
Pid actorPid = system.actorOf(handler).spawn();

// Send messages
actorPid.tell(new HelloMessage());

public class CounterHandler implements StatefulHandler<Integer, CounterMessage> {
    
    @Override
    public Integer receive(CounterMessage message, Integer state, ActorContext context) {
        return switch (message) {
            case Increment ignored -> state + 1;
            case Decrement ignored -> state - 1;
            case Reset ignored -> 0;
            case GetCount gc -> {
                context.tell(gc.replyTo(), new CountResult(state));
                yield state; // Return unchanged state
            }
        };
    }
    
    @Override
    public Integer preStart(Integer state, ActorContext context) {
        // Optional initialization logic
        return state;
    }
    
    @Override
    public void postStop(Integer state, ActorContext context) {
        // Optional cleanup logic
    }
    
    @Override
    public boolean onError(CounterMessage message, Integer state, Throwable exception, ActorContext context) {
        // Custom error handling
        return false; // Return true to reprocess the message
    }
}

// Create a stateful actor with a handler class and initial state
Pid counterPid = system.statefulActorOf(CounterHandler.class, 0)
    .withId("counter-1")  // Optional: specify ID (otherwise auto-generated)
    .spawn();

// Or create with a handler instance
CounterHandler handler = new CounterHandler();
Pid counterPid = system.statefulActorOf(handler, 0).spawn();

// Send messages
counterPid.tell(new Increment());

// Configure backpressure, mailbox, and persistence
Pid actorPid = system.actorOf(GreetingHandler.class)
    .withBackpressureConfig(new BackpressureConfig())
    .withMailboxConfig(new ResizableMailboxConfig())
    .spawn();

// Configure stateful actor with persistence
Pid counterPid = system.statefulActorOf(CounterHandler.class, 0)
    .withPersistence(
        PersistenceFactory.createBatchedFileMessageJournal(),
        PersistenceFactory.createFileSnapshotStore()
    )
    .spawn();

// Configure actor with custom thread pool for CPU-intensive work
ThreadPoolFactory cpuFactory = new ThreadPoolFactory()
    .optimizeFor(ThreadPoolFactory.WorkloadType.CPU_BOUND);

Pid computeActor = system.actorOf(ComputationHandler.class)
    .withThreadPoolFactory(cpuFactory)
    .spawn();

// Create a parent actor
Pid parentPid = system.actorOf(ParentHandler.class).spawn();

// Create a child actor through the parent
Pid childPid = system.actorOf(ChildHandler.class)
    .withParent(system.getActor(parentPid))
    .spawn();

// Or create a child directly from another handler
public class ParentHandler implements Handler<ParentMessage> {
    @Override
    public void receive(ParentMessage message, ActorContext context) {
        if (message instanceof CreateChild) {
            // Create a child actor
            Pid childPid = context.createChild(ChildHandler.class, "child-1");
            // Send message to the child
            context.tell(childPid, new ChildMessage());
        }
    }
}

// Need advanced configuration (e.g., supervision) when creating children?
// Use the new childBuilder() API exposed on ActorContext.
public class SupervisedParentHandler implements Handler<ParentMessage> {
    @Override
    public void receive(ParentMessage message, ActorContext context) {
        if (message instanceof CreateChild) {
            Pid childPid = context.childBuilder(ChildHandler.class)
                .withSupervisionStrategy(SupervisionStrategy.RESTART)
                .withId("supervised-child")
                .spawn();

            context.tell(childPid, new ChildMessage());
        }
    }
}

// Priority 1: Explicit ID wins
Pid actor = system.actorOf(Handler.class)
    .withId("my-actor")           // ← This is used
    .withIdTemplate("user-{seq}") // ← Ignored
    .withIdStrategy(IdStrategy.UUID) // ← Ignored
    .spawn();
// Result: "my-actor"

// Simple explicit ID
Pid userService = system.actorOf(UserServiceHandler.class)
    .withId("user-service")
    .spawn();

// Unicode characters supported
Pid actor = system.actorOf(MyHandler.class)
    .withId("actor-测试-🎭")
    .spawn();

// Simple sequence
Pid actor = system.actorOf(MyHandler.class)
    .withIdTemplate("user-{seq}")
    .spawn();
// Result: "user-1", "user-2", "user-3", ...

// Multiple placeholders
Pid actor = system.actorOf(MyHandler.class)
    .withIdTemplate("{class}-{seq}-{short-uuid}")
    .spawn();
// Result: "myhandler-1-a1b2c3d4"

// With timestamp
Pid session = system.actorOf(SessionHandler.class)
    .withIdTemplate("session-{timestamp}-{seq}")
    .spawn();
// Result: "session-1732956789123-1"

// First run: Create stateful actors with sequential IDs
Pid user1 = system.statefulActorOf(UserHandler.class, initialState)
    .withIdStrategy(IdStrategy.CLASS_BASED_SEQUENTIAL)
    .withPersistence(journal, snapshot)
    .spawn();
// Result: "userhandler:1"

Pid user2 = system.statefulActorOf(UserHandler.class, initialState)
    .withIdStrategy(IdStrategy.CLASS_BASED_SEQUENTIAL)
    .withPersistence(journal, snapshot)
    .spawn();
// Result: "userhandler:2"

// After restart: Counters resume from persisted state
Pid user3 = system.statefulActorOf(UserHandler.class, initialState)
    .withIdStrategy(IdStrategy.CLASS_BASED_SEQUENTIAL)
    .withPersistence(journal, snapshot)
    .spawn();
// Result: "userhandler:3" (not "userhandler:1"!)
// Existing actors "userhandler:1" and "userhandler:2" are restored

// UUID (Default)
Pid actor = system.actorOf(MyHandler.class)
    .withIdStrategy(IdStrategy.UUID)
    .spawn();
// Result: "a1b2c3d4-e5f6-7890-abcd-ef1234567890"

// CLASS_BASED_UUID: {class}:{uuid}
Pid actor = system.actorOf(MyHandler.class)
    .withIdStrategy(IdStrategy.CLASS_BASED_UUID)
    .spawn();
// Result: "myhandler:a1b2c3d4-e5f6-7890-abcd-ef1234567890"

// CLASS_BASED_SEQUENTIAL: {class}:{seq}
Pid actor = system.actorOf(MyHandler.class)
    .withIdStrategy(IdStrategy.CLASS_BASED_SEQUENTIAL)
    .spawn();
// Result: "myhandler:1", "myhandler:2", "myhandler:3", ...

// SEQUENTIAL: Simple counter
Pid actor = system.actorOf(MyHandler.class)
    .withIdStrategy(IdStrategy.SEQUENTIAL)
    .spawn();
// Result: "1", "2", "3", ...

// Create parent
Pid parent = system.actorOf(ParentHandler.class)
    .withId("parent")
    .spawn();

// Create child - ID is automatically prefixed
Pid child = system.actorOf(ChildHandler.class)
    .withId("child")
    .withParent(system.getActor(parent))
    .spawn();
// Result: "parent/child"

// Children with templates
Pid child1 = system.actorOf(ChildHandler.class)
    .withIdTemplate("child-{seq}")
    .withParent(system.getActor(parent))
    .spawn();
// Result: "parent/child-1"

// Deep hierarchies
Pid grandchild = system.actorOf(Handler.class)
    .withId("grandchild")
    .withParent(system.getActor(child))
    .spawn();
// Result: "parent/child/grandchild"

// ✅ Good: Explicit IDs for singletons
Pid service = system.actorOf(ServiceHandler.class)
    .withId("user-service")
    .spawn();

// ✅ Good: Templates for dynamic actors
Pid session = system.actorOf(SessionHandler.class)
    .withIdTemplate("session-{seq}")
    .spawn();

// ✅ Good: Strategies for consistency
Pid worker = system.actorOf(WorkerHandler.class)
    .withIdStrategy(IdStrategy.CLASS_BASED_SEQUENTIAL)
    .spawn();

// ✅ Good: Descriptive IDs
.withIdTemplate("user-session-{seq}")
.withIdTemplate("{class}-worker-{seq}")

// ❌ Bad: Generic IDs
.withIdTemplate("actor-{seq}")

// ✅ Good: Stable IDs for stateful actors
Pid counter = system.statefulActorOf(CounterHandler.class, 0)
    .withId("global-counter")  // Same ID after restart
    .withPersistence(...)
    .spawn();

public class TimerHandler implements Handler<TimerMessage> {
    @Override
    public void receive(TimerMessage message, ActorContext context) {
        switch (message) {
            case Start ignored -> {
                // Schedule a message to self after 1 second
                context.tellSelf(new Tick(), 1, TimeUnit.SECONDS);
            }
            case Tick ignored -> {
                context.getLogger().info("Tick received");
                // Schedule next tick
                context.tellSelf(new Tick(), 1, TimeUnit.SECONDS);
            }
        }
    }
}

public class LoggingHandler implements Handler<String> {
    @Override
    public void receive(String message, ActorContext context) {
        // Logger automatically includes actor ID in output
        context.getLogger().info("Processing message: {}", message);
        context.getLogger().debug("Debug info for actor {}", context.getActorId());
        context.getLogger().error("Error occurred", exception);
    }
}

// Define messages that require replies
public record GetUserRequest(String userId, Pid replyTo) implements ReplyingMessage {}
public record GetOrderRequest(String orderId, Pid replyTo) implements ReplyingMessage {}

public record UserResponse(String userId, String name) {}
public record OrderResponse(String orderId, double amount) {}

// In your handler
public class DatabaseHandler implements Handler<Object> {
    @Override
    public void receive(Object message, ActorContext context) {
        switch (message) {
            case GetUserRequest req -> {
                UserResponse user = fetchUser(req.userId());
                // Use the reply convenience method
                context.reply(req, user);
            }
            case GetOrderRequest req -> {
                OrderResponse order = fetchOrder(req.orderId());
                context.reply(req, order);
            }
        }
    }
}

public interface ReplyingMessage {
    Pid replyTo();
}

// Create a stateful actor with file-based persistence
Pid counterPid = system.statefulActorOf(CounterHandler.class, 0)
    .withPersistence(
        PersistenceFactory.createFileSnapshotStore("/path/to/snapshots"),
        PersistenceFactory.createBatchedFileMessageJournal("/path/to/journal")
    )
    .spawn();

// Configure recovery options
Pid counterPid = system.statefulActorOf(CounterHandler.class, 0)
    .withRecoveryConfig(RecoveryConfig.builder()
        .withRecoveryStrategy(RecoveryStrategy.SNAPSHOT_THEN_JOURNAL)
        .withMaxMessagesToRecover(1000)
        .build())
    .spawn();

// Configure snapshot strategy
Pid counterPid = system.statefulActorOf(CounterHandler.class, 0)
    .withSnapshotStrategy(SnapshotStrategy.builder()
        .withInterval(Duration.ofMinutes(5))
        .withThreshold(100) // Take snapshot every 100 state changes
        .build())
    .spawn();

import static com.cajunsystems.functional.ActorSystemEffectExtensions.*;

// Define messages
sealed interface CounterMsg {}
record Increment(int amount) implements CounterMsg {}
record Decrement(int amount) implements CounterMsg {}
record GetCount(Pid replyTo) implements CounterMsg {}

// Build behavior using effects - Note: Message type is at match level
Effect<Integer, Throwable, Void> counterBehavior = 
    Effect.<Integer, Throwable, Void, CounterMsg>match()
        .when(Increment.class, (state, msg, ctx) -> 
            Effect.modify(s -> s + msg.amount())
                .andThen(Effect.logState(s -> "Count: " + s)))
        
        .when(Decrement.class, (state, msg, ctx) ->
            Effect.modify(s -> s - msg.amount())
                .andThen(Effect.logState(s -> "Count: " + s)))
        
        .when(GetCount.class, (state, msg, ctx) ->
            Effect.tell(msg.replyTo(), state))
        
        .build();

// Create actor from effect
Pid counter = fromEffect(system, counterBehavior, 0)
    .withId("counter")
    .spawn();

// Use like any actor
counter.tell(new Increment(5));

Effect.modify(count -> count + 1)              // Update state
Effect.setState(0)                             // Set to specific value

Effect.tell(otherActor, message)               // Send to another actor
Effect.tellSelf(message)                       // Send to self
Effect.ask(actor, request, Duration.ofSeconds(5))  // Request-response

Effect.modify(s -> s + 1)
    .andThen(Effect.logState(s -> "New state: " + s))
    .andThen(Effect.tell(monitor, new StateUpdate(s)))

Effect.attempt(() -> riskyOperation())
    .recover(error -> defaultValue)
    .orElse(Effect.of(fallbackValue))

// Write natural blocking code - Virtual Threads make it efficient!
Effect.attempt(() -> {
    var user = database.findUser(userId);           // Blocking - totally fine!
    var profile = httpClient.get("/api/profile");   // Also blocking - great!
    return new UserData(user, profile);
})
.recover(error -> UserData.empty());

public class CountResultHandler implements Handler<HelloCount> {
    @Override
    public void receive(HelloCount message, ActorContext context) {
        System.out.println("Count: " + message.count());
    }
}

public static void main(String[] args) {
    // Create the actor system
    var actorSystem = new ActorSystem();
    
    // Create a greeting actor
    var greetingPid = actorSystem.actorOf(GreetingHandler.class)
        .withId("greeting-actor-1")
        .spawn();
    
    // Create a receiver actor
    var receiverPid = actorSystem.actorOf(CountResultHandler.class)
        .withId("count-receiver")
        .spawn();
    
    // Send messages
    greetingPid.tell(new HelloMessage());
    greetingPid.tell(new GetHelloCount(receiverPid)); // Will print "Count: 1"
}

@Test
void testCounter() {
    try (TestKit testKit = TestKit.create()) {
        TestPid<Object> counter = testKit.spawnStateful(CounterHandler.class, 0);
        
        counter.tell(new Increment(5));
        
        // No Thread.sleep()! Wait for exact state
        AsyncAssertion.awaitValue(
            counter.stateInspector()::current,
            5,
            Duration.ofSeconds(1)
        );
    }
}

// Create an actor with custom batch size
Pid myActor = system.actorOf(MyHandler.class)
    .withId("high-throughput-actor")
    .withBatchSize(50)  // Process 50 messages at a time
    .spawn();

// For stateful actors
Pid statefulActor = system.statefulActorOf(MyStatefulHandler.class, initialState)
    .withBatchSize(100)  // Larger batches for high-throughput scenarios
    .spawn();

// Optimize for I/O-bound operations (high concurrency, virtual threads)
ThreadPoolFactory ioOptimized = new ThreadPoolFactory()
    .optimizeFor(ThreadPoolFactory.WorkloadType.IO_BOUND);

// Optimize for CPU-bound operations (fixed thread pool, platform threads)
ThreadPoolFactory cpuOptimized = new ThreadPoolFactory()
    .optimizeFor(ThreadPoolFactory.WorkloadType.CPU_BOUND);

// Optimize for mixed workloads (work-stealing pool)
ThreadPoolFactory mixedOptimized = new ThreadPoolFactory()
    .optimizeFor(ThreadPoolFactory.WorkloadType.MIXED);

// Create actors with optimized thread pools
Pid networkActor = system.actorOf(NetworkHandler.class)
    .withId("network-processor")
    .withThreadPoolFactory(ioOptimized)
    .spawn();

Pid computeActor = system.actorOf(ComputationHandler.class)
    .withId("compute-processor")
    .withThreadPoolFactory(cpuOptimized)
    .spawn();

// Stateful actors also support custom thread pools
Pid statefulActor = system.statefulActorOf(StateHandler.class, initialState)
    .withId("stateful-processor")
    .withThreadPoolFactory(mixedOptimized)
    .spawn();

// Custom configuration for specific requirements
ThreadPoolFactory customFactory = new ThreadPoolFactory()
    .setExecutorType(ThreadPoolFactory.ThreadPoolType.FIXED)
    .setFixedPoolSize(8)  // 8 platform threads
    .setPreferVirtualThreads(false)
    .setUseNamedThreads(true);

Pid customActor = system.actorOf(MyHandler.class)
    .withThreadPoolFactory(customFactory)
    .spawn();

// Default configuration
Pid actor = system.actorOf(MyHandler.class).spawn();

// Custom capacity
MailboxConfig config = new MailboxConfig(5000); // 5K capacity
Pid actor = system.actorOf(MyHandler.class)
    .withMailboxConfig(config)
    .spawn();

ResizableMailboxConfig config = new ResizableMailboxConfig(
    100,    // Initial capacity
    1000,   // Maximum capacity
    50,     // Minimum capacity
    0.8,    // Grow threshold (80% full)
    2.0,    // Growth factor (double size)
    0.2,    // Shrink threshold (20% full)
    0.5     // Shrink factor (halve size)
);

Pid actor = system.actorOf(MyHandler.class)
    .withMailboxConfig(config)
    .spawn();

// Requires custom MailboxProvider for ArrayBlockingQueue
public class ArrayBlockingQueueProvider<M> implements MailboxProvider<M> {
    @Override
    public BlockingQueue<M> createMailbox(MailboxConfig config, ThreadPoolFactory.WorkloadType workloadTypeHint) {
        return new ArrayBlockingQueue<>(config.getCapacity());
    }
}

ActorSystem system = ActorSystem.create("my-system")
    .withMailboxProvider(new ArrayBlockingQueueProvider<>())
    .build();

public class SynchronousQueueProvider<M> implements MailboxProvider<M> {
    @Override
    public BlockingQueue<M> createMailbox(MailboxConfig config, ThreadPoolFactory.WorkloadType workloadTypeHint) {
        return new SynchronousQueue<>();
    }
}

ActorSystem system = ActorSystem.create("my-system")
    .withMailboxProvider(new SynchronousQueueProvider<>())
    .build();

// Combine bounded mailbox with backpressure for flow control
BackpressureConfig bpConfig = new BackpressureConfig()
    .setStrategy(BackpressureStrategy.BLOCK)
    .setCriticalThreshold(0.9f);

MailboxConfig mailboxConfig = new MailboxConfig(1000); // Bounded

Pid actor = system.actorOf(MyHandler.class)
    .withMailboxConfig(mailboxConfig)
    .withBackpressureConfig(bpConfig)
    .spawn();

// Tier 1: Simple - just get the value
String name = userActor.ask(new GetName(), Duration.ofSeconds(5)).get();

// Tier 2: Safe - pattern matching with Result
switch (userActor.ask(new GetProfile(), Duration.ofSeconds(5)).await()) {
    case Result.Success(var profile) -> handleSuccess(profile);
    case Result.Failure(var error) -> handleError(error);
}

// Tier 3: Advanced - full CompletableFuture power
CompletableFuture<Combined> result = userReply.future()
    .thenCombine(ordersReply.future(), (user, orders) -> combine(user, orders));

// Send a request to an actor and get a future response
CompletableFuture<String> future = actorSystem.ask(
    targetActorPid,       // The actor to ask
    "ping",               // The message to send
    Duration.ofSeconds(3) // Timeout for the response
);

// Process the response when it arrives
future.thenAccept(response -> {
    System.out.println("Received response: " + response);
});

// Or wait for the response (blocking)
try {
    String response = future.get(5, TimeUnit.SECONDS);
    System.out.println("Received response: " + response);
} catch (ExecutionException | InterruptedException | TimeoutException e) {
    System.err.println("Error getting response: " + e.getMessage());
}

public class ResponderHandler implements Handler<String> {
    
    @Override
    public void receive(String message, ActorContext context) {
        // Process the message naturally
        String response = processMessage(message);
        
        // Reply to sender if present (will be present for ask requests)
        context.getSender().ifPresent(sender -> 
            context.tell(sender, response)
        );
    }
    
    private String processMessage(String message) {
        if ("ping".equals(message)) {
            return "pong";
        }
        return "unknown command";
    }
}

try {
    String response = actorSystem.ask(actorPid, message, Duration.ofSeconds(2)).get();
    // Process successful response
} catch (ExecutionException ex) {
    Throwable cause = ex.getCause();
    if (cause instanceof TimeoutException) {
        // Handle timeout
    } else if (cause instanceof RuntimeException && cause.getCause() instanceof ClassCastException) {
        // Handle type mismatch
    } else {
        // Handle other errors
    }
} catch (InterruptedException e) {
    // Handle interruption
}

public class AskPatternExample {
    
    public record PingMessage() {}
    public record PongMessage() {}
    
    public static class PingPongHandler implements Handler<PingMessage> {
        @Override
        public void receive(PingMessage message, ActorContext context) {
            // Automatically reply to the sender
            context.getSender().ifPresent(sender -> 
                context.tell(sender, new PongMessage())
            );
        }
    }
    
    public static void main(String[] args) throws Exception {
        ActorSystem system = new ActorSystem();
        
        // Create the responder actor
        Pid responder = system.actorOf(PingPongHandler.class).spawn();
        
        // Send an ask request
        CompletableFuture<PongMessage> future = system.ask(
            responder,
            new PingMessage(),
            Duration.ofSeconds(3)
        );
        
        // Wait for and process the response
        PongMessage response = future.get();
        System.out.println("Received pong!");
        
        system.shutdown();
    }
}

public class ProcessorHandler implements Handler<Request> {
    @Override
    public void receive(Request message, ActorContext context) {
        // Check if there's a sender (e.g., from ask pattern)
        context.getSender().ifPresent(sender -> {
            // Reply to the sender
            context.tell(sender, new Response("processed"));
        });
    }
}

public class RouterHandler implements Handler<RoutableRequest> {
    @Override
    public void receive(RoutableRequest message, ActorContext context) {
        Pid targetHandler = selectHandler(message);
        
        // Forward preserves the original sender context
        context.forward(targetHandler, message);
        
        // The target handler can now reply directly to the original requester
    }
}

public class HandlerActor implements Handler<RoutableRequest> {
    @Override
    public void receive(RoutableRequest message, ActorContext context) {
        Response response = process(message);
        
        // Reply goes to original requester, not the router
        context.getSender().ifPresent(requester -> 
            context.tell(requester, response)
        );
    }
}

// Grandparent initiates request
CompletableFuture<Result> future = system.ask(
    parentPid, 
    new ProcessRequest("data"),
    Duration.ofSeconds(3)
);

// Parent forwards to child (preserving grandparent as sender)
public class ParentHandler implements Handler<ProcessRequest> {
    @Override
    public void receive(ProcessRequest msg, ActorContext context) {
        ProcessRequest enhanced = preprocess(msg);
        context.forward(childPid, enhanced); // Sender preserved
    }
}

// Child processes and replies to grandparent
public class ChildHandler implements Handler<ProcessRequest> {
    @Override
    public void receive(ProcessRequest msg, ActorContext context) {
        Result result = process(msg);
        
        // Reply goes to grandparent (original requester)
        context.getSender().ifPresent(requester -> 
            context.tell(requester, result)
        );
    }
}

// Configure an actor with a specific supervision strategy
MyActor actor = new MyActor(system, "my-actor");
actor.withSupervisionStrategy(SupervisionStrategy.RESTART);

// Method chaining for configuration
MyActor actor = new MyActor(system, "my-actor")
    .withSupervisionStrategy(SupervisionStrategy.RESTART)
    .withErrorHook(ex -> logger.error("Actor error", ex));

public class ResilientActor extends Actor<String> {
    
    public ResilientActor(ActorSystem system, String actorId) {
        super(system, actorId);
    }
    
    @Override
    protected void preStart() {
        // Initialize resources
        logger.info("Actor starting: {}", self().id());
    }
    
    @Override
    protected void postStop() {
        // Clean up resources
        logger.info("Actor stopping: {}", self().id());
    }
    
    @Override
    protected void onError(Throwable error) {
        // Custom error handling
        logger.error("Error in actor: {}", self().id(), error);
    }
    
    @Override
    protected void receive(String message) {
        // Message processing logic
    }
}

@Override
protected SupervisionStrategy handleException(Throwable exception) {
    if (exception instanceof TemporaryException) {
        // Log and continue for temporary issues
        logger.warn("Temporary error, resuming", exception);
        return SupervisionStrategy.RESUME;
    } else if (exception instanceof RecoverableException) {
        // Restart for recoverable errors
        logger.error("Recoverable error, restarting", exception);
        return SupervisionStrategy.RESTART;
    } else {
        // Stop for critical errors
        logger.error("Critical error, stopping", exception);
        return SupervisionStrategy.STOP;
    }
}

try {
    // If the actor throws an exception while processing this message,
    // it will be propagated to the future based on the supervision strategy
    String result = actorSystem.ask(actorPid, "risky-operation", Duration.ofSeconds(5)).get();
    System.out.println("Success: " + result);
} catch (ExecutionException ex) {
    // The original exception is wrapped in an ExecutionException
    Throwable cause = ex.getCause();
    System.err.println("Actor error: " + cause.getMessage());
}

// Configure logging in your application
private static final Logger logger = LoggerFactory.getLogger(MyActor.class);

// Errors are automatically logged with appropriate levels
@Override
protected void receive(Message msg) {
    try {
        // Process message
    } catch (Exception e) {
        // This will be logged and handled according to the supervision strategy
        throw new ActorException("Failed to process message", e);
    }
}

// Define a stateful handler for the counter
public class CounterHandler implements StatefulHandler<Integer, CounterMessage> {
    @Override
    public Integer processMessage(Integer count, CounterMessage message) {
        if (message instanceof IncrementMessage) {
            return count + 1;
        } else if (message instanceof GetCountMessage getCountMsg) {
            // Send the current count back to the sender
            getCountMsg.getSender().tell(count);
            return count;
        }
        return count;
    }
}

// Create a stateful actor with an initial state using the handler pattern
Pid counterPid = system.statefulActor("counter", 0, new CounterHandler());

// Send messages to the actor
counterPid.tell(new IncrementMessage());
counterPid.tell(new GetCountMessage(myPid));

// Define a stateful handler with custom persistence (legacy approach)
public class MyStatefulHandler implements StatefulHandler<MyState, MyMessage> {
    @Override
    public MyState processMessage(MyState state, MyMessage message) {
        // Process the message and return the new state
        return newState;
    }
}

// Create the actor using the stateful handler
Pid actorPid = system.statefulActor(
    "my-actor",
    initialState,
    new MyStatefulHandler(),
    PersistenceFactory.createBatchedFileMessageJournal(),
    PersistenceFactory.createFileSnapshotStore()
);

// Register a custom persistence provider for the entire actor system
PersistenceProvider customProvider = new CustomPersistenceProvider();
ActorSystemPersistenceHelper.setPersistenceProvider(actorSystem, customProvider);

// Or use the fluent API
ActorSystemPersistenceHelper.persistence(actorSystem)
    .withPersistenceProvider(customProvider);

// Create stateful actors using the handler pattern with the configured provider
// No need to specify persistence components explicitly
public class MyStatefulHandler implements StatefulHandler<MyState, MyMessage> {
    @Override
    public MyState processMessage(MyState state, MyMessage message) {
        // Process the message and return the new state
        return newState;
    }
}

// The system will use the configured persistence provider automatically
Pid actorPid = system.statefulActor("my-actor", initialState, new MyStatefulHandler());

public class CustomPersistenceProvider implements PersistenceProvider {
    @Override
    public <M> BatchedMessageJournal<M> createBatchedMessageJournal(String actorId) {
        // Implement custom message journaling
        return new CustomBatchedMessageJournal<>(actorId);
    }
    
    @Override
    public <S> SnapshotStore<S> createSnapshotStore(String actorId) {
        // Implement custom state snapshot storage
        return new CustomSnapshotStore<>(actorId);
    }
    
    // Implement other required methods
}

// Configure LMDB persistence provider
Path lmdbPath = Paths.get("/var/cajun/lmdb");
long mapSize = 10L * 1024 * 1024 * 1024; // 10GB
LmdbPersistenceProvider lmdbProvider = new LmdbPersistenceProvider(lmdbPath, mapSize);

// Register as system-wide provider
ActorSystemPersistenceHelper.setPersistenceProvider(actorSystem, lmdbProvider);

// Or use fluent API
ActorSystemPersistenceHelper.persistence(actorSystem)
    .withPersistenceProvider(lmdbProvider);

// Create stateful actors with LMDB persistence
Pid actor = system.statefulActorOf(MyHandler.class, initialState)
    .withPersistence(
        lmdbProvider.createMessageJournal("my-actor"),
        lmdbProvider.createSnapshotStore("my-actor")
    )
    .spawn();

// For high-throughput scenarios, use the native batched journal
BatchedMessageJournal<MyEvent> batchedJournal =
    lmdbProvider.createBatchedMessageJournalSerializable("my-actor", 5000, 10);

// Configure snapshot strategy (time-based and change-count-based)
statefulActor.configureSnapshotStrategy(
    30000,    // Take snapshot every 30 seconds
    500       // Or after 500 state changes, whichever comes first
);

// Force an immediate snapshot
statefulActor.forceSnapshot().join();

import com.cajunsystems.persistence.PersistenceTruncationConfig;
import com.cajunsystems.persistence.PersistenceTruncationMode;

// Option 1: Synchronous truncation (default)
// Journals are truncated during snapshot lifecycle
PersistenceTruncationConfig syncConfig = PersistenceTruncationConfig.builder()
    .mode(PersistenceTruncationMode.SYNC_ON_SNAPSHOT)
    .retainMessagesBehindSnapshot(500)    // Keep 500 messages before latest snapshot
    .retainLastMessagesPerActor(5000)     // Always keep last 5000 messages minimum
    .build();

// Option 2: Asynchronous truncation with background daemon
// Non-blocking truncation runs periodically
PersistenceTruncationConfig asyncConfig = PersistenceTruncationConfig.builder()
    .mode(PersistenceTruncationMode.ASYNC_DAEMON)
    .retainMessagesBehindSnapshot(500)
    .retainLastMessagesPerActor(5000)
    .daemonInterval(Duration.ofMinutes(5)) // Run every 5 minutes
    .build();

// Option 3: Disable truncation
PersistenceTruncationConfig offConfig = PersistenceTruncationConfig.builder()
    .mode(PersistenceTruncationMode.OFF)
    .build();

// Apply to stateful actor builder
Pid actor = system.statefulActorOf(MyHandler.class, initialState)
    .withTruncationConfig(asyncConfig)
    .spawn();

// Define backpressure settings using BackpressureConfig
BackpressureConfig systemBpConfig = new BackpressureConfig()
    .setStrategy(BackpressureStrategy.BLOCK)      // Default strategy
    .setWarningThreshold(0.7f)                 // 70% mailbox capacity
    .setCriticalThreshold(0.9f)                // 90% mailbox capacity
    .setRecoveryThreshold(0.5f);                // 50% mailbox capacity

// Create ActorSystem with this configuration
// This also requires a ThreadPoolFactory
ActorSystem system = new ActorSystem(new ThreadPoolFactory(), systemBpConfig);

// Actors created in this system will now use these backpressure settings by default.

// Direct actor configuration with type safety
BackpressureBuilder<MyMessage> builder = new BackpressureBuilder<>(myActor)
    .withStrategy(BackpressureStrategy.DROP_OLDEST)
    .withWarningThreshold(0.7f)
    .withCriticalThreshold(0.9f)
    .withRecoveryThreshold(0.5f);

// Apply the configuration
builder.apply();

// PID-based configuration through ActorSystem
BackpressureBuilder<MyMessage> builder = system.getBackpressureMonitor()
    .configureBackpressure(actorPid)
    .withStrategy(BackpressureStrategy.DROP_OLDEST)
    .withWarningThreshold(0.7f)
    .withCriticalThreshold(0.9f);

builder.apply();

// Using preset configurations for common scenarios
BackpressureBuilder<MyMessage> timeCriticalBuilder = new BackpressureBuilder<>(myActor)
    .presetTimeCritical()
    .apply();

BackpressureBuilder<MyMessage> reliableBuilder = new BackpressureBuilder<>(myActor)
    .presetReliable()
    .apply();

BackpressureBuilder<MyMessage> highThroughputBuilder = new BackpressureBuilder<>(myActor)
    .presetHighThroughput()
    .apply();

// Check backpressure status
BackpressureStatus status = actor.getBackpressureStatus();
BackpressureState currentState = status.getCurrentState();
float fillRatio = status.getFillRatio();

CustomBackpressureHandler<MyMessage> handler = new CustomBackpressureHandler<>() {
    @Override
    public boolean handleMessage(Actor<MyMessage> actor, MyMessage message, BackpressureSendOptions options) {
        // Custom logic to decide whether to accept the message
        if (message.isPriority()) {
            return true; // Always accept priority messages
        }
        return actor.getCurrentSize() < actor.getCapacity() * 0.9;
    }
    
    @Override
    public boolean makeRoom(Actor<MyMessage> actor) {
        // Custom logic to make room in the mailbox
        // Return true if room was successfully made
        return actor.dropOldestMessage();
    }
};

// Configure with custom handler
new BackpressureBuilder<>(myActor)
    .withStrategy(BackpressureStrategy.CUSTOM)
    .withCustomHandler(handler)
    .apply();

// Register for backpressure event notifications using the builder
new BackpressureBuilder<>(myActor)
    .withStrategy(BackpressureStrategy.DROP_OLDEST)
    .withWarningThreshold(0.7f)
    .withCriticalThreshold(0.9f)
    .withCallback(event -> {
        logger.info("Backpressure event: {} state, fill ratio: {}", 
                    event.getState(), event.getFillRatio());
        
        // Take action based on backpressure events
        if (event.isBackpressureActive()) {
            // Notify monitoring system, scale resources, etc.
        }
    })
    .apply();

// Access detailed backpressure metrics and history
BackpressureStatus status = actor.getBackpressureStatus();
List<BackpressureEvent> recentEvents = status.getRecentEvents();
List<StateTransition> stateTransitions = status.getStateTransitions();

// Monitor state transition history
for (StateTransition transition : stateTransitions) {
    logger.debug("Transition from {} to {} at {} due to: {}", 
                transition.getFromState(), 
                transition.getToState(),
                transition.getTimestamp(),
                transition.getReason());
}

// Create options for high priority messages that bypass backpressure
BackpressureSendOptions highPriority = new BackpressureSendOptions()
    .setHighPriority(true)
    .setTimeout(Duration.ofSeconds(5));

// Send with high priority
actor.tell(urgentMessage, highPriority);

// Or use the system to send with options
boolean accepted = system.tellWithOptions(actorPid, message, highPriority);

// Block until message is accepted or timeout occurs
BackpressureSendOptions blockingOptions = new BackpressureSendOptions()
    .setBlockUntilAccepted(true)
    .setTimeout(Duration.ofSeconds(3));

// Create a cluster configuration
ClusterConfig config = ClusterConfig.builder()
    .nodeName("node1")
    .port(2551)
    .seedNodes(List.of("127.0.0.1:2551", "127.0.0.1:2552"))
    .build();

// Create a clustered actor system
ActorSystem system = ActorSystem.createClustered(config);

// Create and register actors as usual
Pid actorPid = system.register(MyActor.class, "my-actor");

// Send a message to an actor (works the same whether the actor is local or remote)

// Node 1
MetadataStore metadataStore1 = new EtcdMetadataStore("http://etcd-host:2379");
DirectMessagingSystem messagingSystem1 = new DirectMessagingSystem("node1", 8080);
messagingSystem1.addNode("node2", "node2-host", 8080);
ClusterActorSystem system1 = new ClusterActorSystem("node1", metadataStore1, messagingSystem1);
system1.start().get();

// Node 2
MetadataStore metadataStore2 = new EtcdMetadataStore("http://etcd-host:2379");
DirectMessagingSystem messagingSystem2 = new DirectMessagingSystem("node2", 8080);
messagingSystem2.addNode("node1", "node1-host", 8080);
ClusterActorSystem system2 = new ClusterActorSystem("node2", metadataStore2, messagingSystem2);
system2.start().get();

public class CustomMetadataStore implements MetadataStore {
    // Implement the required methods
}

public class CustomMessagingSystem implements MessagingSystem {
    // Implement the required methods
}

class OrderServiceActor {
    void receive(CreateOrder order) {
        User user = userDB.find(order.userId);        // 5ms I/O
        Inventory inv = inventoryAPI.check(order);    // 20ms I/O
        Payment pay = paymentGateway.process(order);  // 15ms I/O
        orderDB.save(order);                          // 3ms I/O
        
        // Total: 43ms I/O
        // Actor overhead: 0.002ms (0.005%)
    }
}

class RequestHandlerActor {
    void receive(HttpRequest request) {
        Session session = sessionStore.get(request.token);  // 2ms
        Data data = database.query(request.params);         // 30ms
        String html = templateEngine.render(data);          // 8ms
        
        // Total: 40ms, Actor overhead: 0.002ms (0.005%)
    }
}

// For 100 independent parallel computations, use threads
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
List<Future<Result>> futures = items.parallelStream()
    .map(item -> executor.submit(() -> compute(item)))
    .toList();