// 1. Open the log with in-memory storage
SharedLogConfig config = SharedLogConfig.builder()
    .persistenceAdapter(new InMemoryPersistenceAdapter())
    .build();

try (SharedLogService log = SharedLogService.open(config)) {

    LogTag orders = LogTag.of("orders");

    // 2. Append
    AppendResult r = log.append(AppendRequest.to(orders, "order-placed".getBytes())).join();
    System.out.println("seqnum=" + r.seqnum() + " localId=" + r.localId());

    // 3. Read
    List<LogEntry> entries = log.readAll(orders).join();

    // 4. Subscribe (runs listener on a virtual thread)
    SharedLog.Subscription sub = log.subscribe(orders, LogPosition.BEGINNING, entry ->
        System.out.println("received: " + new String(entry.data())));

    // 5. Use a view
    LogView view = log.getView(orders);
    view.append("order-confirmed".getBytes()).join();

    sub.close();
}

SharedLogConfig config = SharedLogConfig.builder()
    .persistenceAdapter(new FileBasedPersistenceAdapter("/var/data/myapp"))
    .build();

LogTag allOrders   = LogTag.of("orders");           // namespace-wide
LogTag singleOrder = LogTag.of("orders", "order-42"); // instance-scoped

AppendRequest.to(
    Set.of(LogTag.of("orders"), LogTag.of("orders", "order-42")),
    payload);

SharedLogService log = SharedLogService.open(config);

// Append entries
AppendResult result = log.append(AppendRequest.to(tag, data)).join();

// Read entries
List<LogEntry> all = log.readAll(tag).join();
List<LogEntry> from = log.readFrom(tag, position, maxEntries).join();

// Subscribe to new entries
Subscription sub = log.subscribe(tag, position, entry -> {
    // Process entry on virtual thread
});

// Batch operations
List<AppendResult> results = log.appendBatch(requests).join();

// Trim old entries
log.trim(upToSeqnum).join();

LogView view = log.getView(LogTag.of("orders"));

// Append (automatically includes the view's tag)
view.append(data).join();

// Read operations
List<LogEntry> entries = view.readAll().join();
LogEntry next = view.readNextAfter(seqnum).join();
LogEntry prev = view.readPrevBefore(seqnum).join();
long tail = view.checkTail().join();

// Subscribe to new entries only (no backlog)
Subscription sub = view.subscribeTail(entry -> process(entry));

// Durable KV storage (for checkpoints)
view.setValue("checkpoint", bytes).join();
byte[] saved = view.getValue("checkpoint").join();

LogSerializer<OrderEvent> serializer = new KryoLogSerializer<>(OrderEvent.class);
TypedLogView<OrderEvent> view = log.getTypedView(tag, serializer);

// Append domain objects directly
view.append(new OrderEvent("order-42", "placed")).join();

// Read typed objects
List<OrderEvent> events = view.readAll().join();

// Typed subscriptions
Subscription sub = view.subscribeTail(event -> 
    System.out.println("Order: " + event.orderId()));

// Tag: identifies a logical stream
LogTag tag = LogTag.of("orders");              // namespace-wide
LogTag tag = LogTag.of("orders", "order-42");  // instance-scoped

// Entry: immutable log record
record LogEntry(
    long seqnum,        // global sequence number
    long localId,       // per-tag sequence number
    Set<LogTag> tags,   // tags this entry belongs to
    byte[] data,        // payload
    long timestamp      // append timestamp
)

// Position: read cursor
LogPosition pos = new LogPosition(seqnum);
LogPosition.BEGINNING  // start from seqnum 0
LogPosition.END        // start from latest

// Append request
AppendRequest req = AppendRequest.to(tag, data);
AppendRequest req = AppendRequest.to(Set.of(tag1, tag2), data);  // multi-tag

// Append result
record AppendResult(
    long seqnum,
    long localId,
    LogTag primaryTag,
    long timestamp
)

PersistenceAdapter base     = new FileBasedPersistenceAdapter("/var/data");
PersistenceAdapter batching = BatchingPersistenceAdapter.of(base); // 64 entries / 10 ms defaults

// or with explicit tuning:
PersistenceAdapter batching = new BatchingPersistenceAdapter(base, 128, 5 /* ms */);

SharedLogConfig config = SharedLogConfig.builder()
    .persistenceAdapter(batching)
    .build();

// Minimal — uses the default fdb.cluster file
PersistenceAdapter fdb = new FoundationDBPersistenceAdapter();

SharedLogConfig config = SharedLogConfig.builder()
    .persistenceAdapter(fdb)
    .build();

FDB      fdb = FDB.selectAPIVersion(730);
Database db  = fdb.open("/etc/foundationdb/fdb.cluster");

// Both components share the same connection
FoundationDBSequencer        seq     = new FoundationDBSequencer(db, "myapp");
FoundationDBPersistenceAdapter store = new FoundationDBPersistenceAdapter(db, "myapp");
seq.open();
store.open();

SharedLogConfig config = SharedLogConfig.builder()
    .sequencer(store)        // seqnums assigned by FDB — safe across nodes
    .persistenceAdapter(store)
    .build();

public class RocksDbPersistenceAdapter implements PersistenceAdapter {
    @Override public void open()   throws IOException { /* ... */ }
    @Override public void close()  throws IOException { /* ... */ }
    @Override public void append(LogEntry entry) throws IOException { /* ... */ }
    @Override public List<LogEntry> readAll() throws IOException { /* ... */ }
    @Override public List<LogEntry> readFrom(long fromSeqnum) throws IOException { /* ... */ }
    @Override public List<LogEntry> readByTag(LogTag tag, long fromSeqnum) throws IOException { /* ... */ }
    @Override public void trim(long upToSeqnum) throws IOException { /* ... */ }
    @Override public long getLatestSeqnum() { /* ... */ }
    @Override public long getLocalIdCountForTag(LogTag tag) { /* ... */ }
}

SharedLogConfig config = SharedLogConfig.builder()
    .persistenceAdapter(new RocksDbPersistenceAdapter(path))
    .build();

// SharedLogService does this automatically on startup
sequencer.advanceTo(persistenceAdapter.getLatestSeqnum() + 1);

FoundationDBSequencer seq = new FoundationDBSequencer("/etc/foundationdb/fdb.cluster", "myapp");
seq.open();

// next() — claims one seqnum, blocks for one FDB round-trip (~1–5 ms)
long seqnum = seq.next();

// currentGlobal() — reads the authoritative global counter without claiming
long highWatermark = seq.currentGlobal();

// 1. Pick a serializer
LogSerializer<OrderEvent> s = new KryoLogSerializer<>(OrderEvent.class);

// 2. Get a typed view
TypedLogView<OrderEvent> view = service.getTypedView(LogTag.of("orders"), s);

// 3. Append domain objects directly — no .getBytes()
view.append(new OrderEvent("ord-42", "placed")).join();

// 4. Read back typed objects — no new String(e.data())
List<OrderEvent> events = view.readAll().join();

SharedLog.Subscription sub = view.subscribeTail(event ->
        System.out.println("received: " + event.orderId()));

public interface LogSerializer<T> {
    byte[] serialize(T value);
    T      deserialize(byte[] data);
}

// Default (no registration required — works for most types and records)
new KryoLogSerializer<>(MyClass.class)

// Pre-registered (smaller wire format, faster)
new KryoLogSerializer<>(MyClass.class, kryo -> {
    kryo.register(MyClass.class, 10);
    kryo.register(MyEnum.class,  11);
})

LogView inbox = service.getView(LogTag.of("actor", "orders"));

// 1. Read saved checkpoint (null on first start)
byte[] saved = inbox.getValue("checkpoint").join();
long resumeFrom = saved != null
    ? ByteBuffer.wrap(saved).getLong() + 1  // resume after last processed seqnum
    : 0L;

// 2. Replay the backlog from checkpoint
List<LogEntry> backlog = inbox.readFrom(new LogPosition(resumeFrom), 1000).join();
for (LogEntry e : backlog) {
    process(e);
    inbox.setValue("checkpoint", ByteBuffer.allocate(8).putLong(e.seqnum()).array()).join();
}

// 3. Switch to live delivery (no backlog, no polling)
SharedLog.Subscription sub = inbox.subscribeTail(e -> {
    process(e);
    inbox.setValue("checkpoint", ByteBuffer.allocate(8).putLong(e.seqnum()).array()).join();
});

record OrderState(List<Order> pending, List<Order> fulfilled) {}

public class FulfilmentExecutor implements Executor<OrderState> {

    @Override
    public LogTag getInputTag() { return LogTag.of("orders"); }

    @Override
    public OrderState initialState() {
        return new OrderState(List.of(), List.of());
    }

    @Override
    public OrderState apply(OrderState state, LogEntry entry) {
        var event = OrderEvent.parse(entry.data());
        return switch (event.type()) {
            case PLACED    -> state.withPending(append(state.pending(), event.order()));
            case FULFILLED -> state.moveToPending(event.orderId());
            default        -> state;
        };
    }

    @Override
    public List<AppendRequest> execute(OrderState state, ExecutorContext ctx) {
        // Fulfil the first pending order, if any
        return state.pending().stream()
            .limit(1)
            .map(o -> AppendRequest.to(
                Set.of(LogTag.of("orders"), LogTag.of("orders", o.id())),
                OrderEvent.fulfilled(o).toBytes()))
            .toList();
    }
}

ExecutorEngine engine = new ExecutorEngine(service);
engine.register(new FulfilmentExecutor());
engine.start();

static class WorkflowSubmitExecutor implements Executor<List<String>> {

    private final Set<String> emitted = ConcurrentHashMap.newKeySet();

    @Override public LogTag getInputTag() { return LogTag.of("workflow.commands"); }
    @Override public String getName()     { return "WorkflowSubmit"; }

    @Override public List<String> initialState() { return List.of(); }

    @Override
    public List<String> apply(List<String> pending, LogEntry entry) {
        String msg = new String(entry.data());
        if (msg.startsWith("submit:")) {
            String id = msg.substring("submit:".length());
            List<String> next = new ArrayList<>(pending);
            next.add(id);
            return next;
        }
        return pending;
    }

    @Override
    public List<AppendRequest> execute(List<String> pending, ExecutorContext ctx) {
        return pending.stream()
            .filter(emitted::add)           // idempotency: only emit each id once
            .map(id -> AppendRequest.to(
                LogTag.of("workflow.events"),
                ("processing:" + id).getBytes()))
            .toList();
    }
}

static class WorkflowProcessExecutor implements Executor<List<String>> {

    private final Set<String> emitted = ConcurrentHashMap.newKeySet();

    @Override public LogTag getInputTag() { return LogTag.of("workflow.events"); }
    @Override public String getName()     { return "WorkflowProcess"; }

    @Override public List<String> initialState() { return List.of(); }

    @Override
    public List<String> apply(List<String> processing, LogEntry entry) {
        String msg = new String(entry.data());
        if (msg.startsWith("processing:")) {
            String id = msg.substring("processing:".length());
            List<String> next = new ArrayList<>(processing);
            next.add(id);
            return next;
        }
        return processing;
    }

    @Override
    public List<AppendRequest> execute(List<String> processing, ExecutorContext ctx) {
        return processing.stream()
            .filter(emitted::add)
            .map(id -> AppendRequest.to(
                LogTag.of("workflow.events"),
                ("complete:" + id).getBytes()))
            .toList();
    }
}

ExecutorEngine engine = new ExecutorEngine(service);
engine.register(new WorkflowSubmitExecutor());
engine.register(new WorkflowProcessExecutor());
engine.start();

// Trigger the pipeline
service.append(AppendRequest.to(LogTag.of("workflow.commands"), "submit:order-99".getBytes())).join();
// → WorkflowSubmitExecutor emits "processing:order-99" to workflow.events
// → WorkflowProcessExecutor picks it up and emits "complete:order-99"

@Override
public List<AppendRequest> execute(OrderState state, ExecutorContext ctx) {
    LogView auditLog = ctx.openView(LogTag.of("audit"));
    auditLog.append(("processed " + state.pending().size() + " orders").getBytes()).join();
    return List.of();
}

// From ExecutorEngine — each runner is one virtual thread
Thread.ofVirtual()
    .name("executor-" + executor.getName())
    .start(runner::run);

Executors.newVirtualThreadPerTaskExecutor()