diff --git a/batch_bench_test.go b/batch_bench_test.go
new file mode 100644
index 0000000..d46486c
--- /dev/null
+++ b/batch_bench_test.go
@@ -0,0 +1,100 @@
+package clset_test
+
+import (
+	"fmt"
+	"testing"
+
+	"example.com/clset"
+	badgerds "github.com/ipfs/go-ds-badger4"
+)
+
+func BenchmarkPutSequential_10(b *testing.B)   { benchmarkPutSequential(b, 10) }
+func BenchmarkPutSequential_100(b *testing.B)  { benchmarkPutSequential(b, 100) }
+func BenchmarkPutSequential_1000(b *testing.B) { benchmarkPutSequential(b, 1000) }
+
+func benchmarkPutSequential(b *testing.B, numOps int) {
+	crdt := createBenchCRDT(b, "peer1")
+	ops := make(map[string][]byte, numOps)
+	for i := 0; i < numOps; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	for i := 0; i < b.N; i++ {
+		for key, value := range ops {
+			if err := crdt.Set(key, value); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+}
+
+func BenchmarkPutBatch_10(b *testing.B)   { benchmarkPutBatch(b, 10) }
+func BenchmarkPutBatch_100(b *testing.B)  { benchmarkPutBatch(b, 100) }
+func BenchmarkPutBatch_1000(b *testing.B) { benchmarkPutBatch(b, 1000) }
+
+func benchmarkPutBatch(b *testing.B, numOps int) {
+	crdt := createBenchCRDT(b, "peer1")
+	ops := make(map[string][]byte, numOps)
+	for i := 0; i < numOps; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	for i := 0; i < b.N; i++ {
+		if err := crdt.PutBatch(ops); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkChangeLogSize_Sequential(b *testing.B) {
+	crdt := createBenchCRDT(b, "peer1")
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < 10; j++ {
+			key := fmt.Sprintf("key%d", j)
+			if err := crdt.Set(key, []byte(fmt.Sprintf("value%d", j))); err != nil {
+				b.Fatal(err)
+			}
+		}
+	}
+
+	b.StopTimer()
+	b.ReportMetric(float64(crdt.PeerSeq), "changes")
+}
+
+func BenchmarkChangeLogSize_Batch(b *testing.B) {
+	crdt := createBenchCRDT(b, "peer1")
+	ops := make(map[string][]byte, 10)
+	for j := 0; j < 10; j++ {
+		ops[fmt.Sprintf("key%d", j)] = []byte(fmt.Sprintf("value%d", j))
+	}
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		if err := crdt.PutBatch(ops); err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	b.StopTimer()
+	b.ReportMetric(float64(crdt.PeerSeq), "changes")
+}
+
+func createBenchCRDT(tb testing.TB, peerID string) *clset.CRDT {
+	opts := &badgerds.DefaultOptions
+	opts.InMemory = true
+	ds, err := badgerds.NewDatastore("", opts)
+	if err != nil {
+		tb.Fatalf("Failed to create datastore: %v", err)
+	}
+	tb.Cleanup(func() { ds.Close() })
+	return clset.New(peerID, ds)
+}
diff --git a/batch_test.go b/batch_test.go
new file mode 100644
index 0000000..c0f6b2a
--- /dev/null
+++ b/batch_test.go
@@ -0,0 +1,147 @@
+package clset_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPutBatch_Basic(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	ops := map[string][]byte{
+		"key1": []byte("value1"),
+		"key2": []byte("value2"),
+		"key3": []byte("value3"),
+	}
+
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// Verify all keys were set
+	for key, expectedValue := range ops {
+		val, exists, err := crdt.Get(key)
+		require.NoError(t, err)
+		assert.True(t, exists, "key %s should exist", key)
+		assert.Equal(t, expectedValue, val, "value for key %s should match", key)
+	}
+
+	// Verify only one sequence number was used
+	assert.Equal(t, uint64(1), crdt.PeerSeq, "PeerSeq should be 1 after batch")
+}
+
+func TestPutBatch_Empty(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	require.NoError(t, crdt.PutBatch(map[string][]byte{}))
+	assert.Equal(t, uint64(0), crdt.PeerSeq, "PeerSeq should remain 0 for empty batch")
+}
+
+func TestPutBatch_SingleSequenceNumber(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	// Put 10 keys in a batch
+	ops := make(map[string][]byte)
+	for i := 0; i < 10; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// All keys should have the same PeerSeq
+	for key := range ops {
+		meta, found, err := crdt.GetKeyMeta(key)
+		require.NoError(t, err)
+		require.True(t, found)
+		assert.Equal(t, uint64(1), meta.PeerSeq, "all keys should have PeerSeq=1")
+		assert.Equal(t, "peer1", meta.PeerID)
+	}
+
+	// Only one sequence number should be consumed
+	assert.Equal(t, uint64(1), crdt.PeerSeq)
+}
+
+func TestDeleteBatch_Basic(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	// First, create some keys
+	ops := map[string][]byte{
+		"key1": []byte("value1"),
+		"key2": []byte("value2"),
+		"key3": []byte("value3"),
+	}
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// Delete them in a batch
+	keys := []string{"key1", "key2", "key3"}
+	require.NoError(t, crdt.DeleteBatch(keys))
+
+	// Verify all keys were deleted
+	for _, key := range keys {
+		_, exists, err := crdt.Get(key)
+		require.NoError(t, err)
+		assert.False(t, exists, "key %s should not exist", key)
+	}
+
+	// Verify only one additional sequence number was used
+	assert.Equal(t, uint64(2), crdt.PeerSeq)
+}
+
+func TestBatch_LargeOperations(t *testing.T) {
+	crdt := createTestCRDT(t, "peer1")
+
+	// Test with 1000 keys
+	ops := make(map[string][]byte)
+	for i := 0; i < 1000; i++ {
+		ops[fmt.Sprintf("key%d", i)] = []byte(fmt.Sprintf("value%d", i))
+	}
+
+	require.NoError(t, crdt.PutBatch(ops))
+
+	// Verify all keys exist
+	count, err := crdt.KeyCount()
+	require.NoError(t, err)
+	assert.Equal(t, uint64(1000), count)
+
+	// Delete half of them
+	keysToDelete := make([]string, 500)
+	for i := 0; i < 500; i++ {
+		keysToDelete[i] = fmt.Sprintf("key%d", i)
+	}
+	require.NoError(t, crdt.DeleteBatch(keysToDelete))
+
+	// Verify count is correct
+	count, err = crdt.KeyCount()
+	require.NoError(t, err)
+	assert.Equal(t, uint64(500), count)
+
+	// Only 2 sequence numbers should be used
+	assert.Equal(t, uint64(2), crdt.PeerSeq)
+}
+
+func TestBatch_P2PSync(t *testing.T) {
+	crdt1 := createTestCRDT(t, "peer1")
+	crdt2 := createTestCRDT(t, "peer2")
+
+	// Peer1 does a batch put
+	ops := map[string][]byte{
+		"key1": []byte("value1"),
+		"key2": []byte("value2"),
+		"key3": []byte("value3"),
+	}
+	require.NoError(t, crdt1.PutBatch(ops))
+
+	// Sync from peer1 to peer2
+	changes, tracked, err := crdt1.GetLatestChanges("peer2", crdt2.GetTrackedPeers())
+	require.NoError(t, err)
+	require.NoError(t, crdt2.MergeChanges("peer1", changes, tracked))
+
+	// Verify all keys made it to peer2
+	for key, expectedValue := range ops {
+		val, exists, err := crdt2.Get(key)
+		require.NoError(t, err)
+		assert.True(t, exists, "key %s should exist on peer2", key)
+		assert.Equal(t, expectedValue, val, "value for key %s should match", key)
+	}
+}
diff --git a/crdt.go b/crdt.go
index ae242a5..dee717f 100644
--- a/crdt.go
+++ b/crdt.go
@@ -4,16 +4,39 @@ import (
 	"bytes"
 	"context"
 	"encoding/gob"
+	"encoding/json"
 	"fmt"
 	"maps"
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/query"
 )
 
+// PeerMetadata represents metadata about a peer in the cluster
+type PeerMetadata struct {
+	// BestBefore is the Unix timestamp after which this peer
+	// should be considered offline if not refreshed
+	BestBefore uint64 `json:"best_before"`
+
+	// Metadata contains user-defined key-value pairs for this peer
+	// Examples: {"name": "Neelix", "role": "write"}
+	Metadata map[string]string `json:"metadata"`
+
+	// Sequence is the last known sequence number for this peer
+	Sequence uint64 `json:"sequence"`
+}
+
+// MembershipState represents the current cluster membership view
+type MembershipState struct {
+	// Members maps peer ID to their metadata
+	// Key format: libp2p peer ID as string
+	Members map[string]*PeerMetadata `json:"members"`
+}
+
 type CRDT struct {
 	PeerID       string
 	PeerSeq      uint64
@@ -22,6 +45,12 @@ type CRDT struct {
 	mu           sync.Mutex
 	hooksMu      sync.RWMutex
 
+	// Namespace is an optional prefix for all keys to allow multiple
+	// independent CRDT instances to coexist in the same datastore.
+	// If empty, keys are stored at the root (e.g., "/data/key").
+	// If set (e.g., "ipservice"), keys are namespaced (e.g., "/ipservice/data/key").
+	namespace string
+
 	// Transactional hooks (called inside transaction, can abort)
 	insertTxnHook func(ctx context.Context, txn datastore.Write, key string, val []byte, meta CRDTKeyMeta) error
 	updateTxnHook func(ctx context.Context, txn datastore.Write, key string, oldVal []byte, oldMeta CRDTKeyMeta, newVal []byte, newMeta CRDTKeyMeta) error
@@ -31,6 +60,84 @@ type CRDT struct {
 	insertHooks []func(key string, val []byte, meta CRDTKeyMeta)
 	updateHooks []func(key string, oldVal []byte, oldMeta CRDTKeyMeta, newVal []byte, newMeta CRDTKeyMeta)
 	deleteHooks []func(key string, oldVal []byte, oldMeta CRDTKeyMeta)
+
+	// Metadata support
+	localMetadata   map[string]string
+	peerMetadata    map[string]*PeerMetadata // peerID -> metadata
+	metadataMu      sync.RWMutex
+	membershipHooks []func(members map[string]*PeerMetadata)
+	peerTTL         time.Duration
+}
+
+// Key construction helpers - these ensure namespace is consistently applied
+// across all internal storage. If namespace is empty, keys are at root level.
+
+func (c *CRDT) dataKey(userKey string) datastore.Key {
+	if c.namespace == "" {
+		return datastore.NewKey("data").ChildString(userKey)
+	}
+	return datastore.NewKey(c.namespace).ChildString("data").ChildString(userKey)
+}
+
+func (c *CRDT) changeKey(peerID string, seq uint64) datastore.Key {
+	seqStr := fmt.Sprintf("%010d", seq)
+	if c.namespace == "" {
+		return datastore.NewKey("change").ChildString(peerID).ChildString(seqStr)
+	}
+	return datastore.NewKey(c.namespace).ChildString("change").ChildString(peerID).ChildString(seqStr)
+}
+
+// changeKeyWithUserKey creates a change log key that includes the user key.
+// This is used for batch operations where multiple keys share the same sequence number.
+// Format: /change/peerID/seqNumber/userKey
+func (c *CRDT) changeKeyWithUserKey(peerID string, seq uint64, userKey string) datastore.Key {
+	seqStr := fmt.Sprintf("%010d", seq)
+	if c.namespace == "" {
+		return datastore.NewKey("change").ChildString(peerID).ChildString(seqStr).ChildString(userKey)
+	}
+	return datastore.NewKey(c.namespace).ChildString("change").ChildString(peerID).ChildString(seqStr).ChildString(userKey)
+}
+
+func (c *CRDT) trackedKey(peerID string) datastore.Key {
+	if c.namespace == "" {
+		return datastore.NewKey("tracked").ChildString(peerID)
+	}
+	return datastore.NewKey(c.namespace).ChildString("tracked").ChildString(peerID)
+}
+
+func (c *CRDT) metadataKey(peerID string) datastore.Key {
+	if c.namespace == "" {
+		return datastore.NewKey("metadata").ChildString(peerID)
+	}
+	return datastore.NewKey(c.namespace).ChildString("metadata").ChildString(peerID)
+}
+
+func (c *CRDT) dataPrefix() string {
+	if c.namespace == "" {
+		return "/data/"
+	}
+	return datastore.NewKey(c.namespace).ChildString("data").String() + "/"
+}
+
+func (c *CRDT) changePrefix(peerID string) string {
+	if c.namespace == "" {
+		return "/change/" + peerID + "/"
+	}
+	return datastore.NewKey(c.namespace).ChildString("change").ChildString(peerID).String() + "/"
+}
+
+func (c *CRDT) trackedPrefix() string {
+	if c.namespace == "" {
+		return "/tracked/"
+	}
+	return datastore.NewKey(c.namespace).ChildString("tracked").String() + "/"
+}
+
+func (c *CRDT) metadataPrefix() string {
+	if c.namespace == "" {
+		return "/metadata/"
+	}
+	return datastore.NewKey(c.namespace).ChildString("metadata").String() + "/"
 }
 
 func (c *CRDT) AddInsertHook(h func(key string, val []byte, meta CRDTKeyMeta)) {
@@ -92,9 +199,12 @@ func (c *CRDT) runDeleteHooks(key string, oldVal []byte, oldMeta CRDTKeyMeta, ho
 
 func New(peerID string, ds datastore.Datastore, opts ...func(*CRDT)) *CRDT {
 	crdt := &CRDT{
-		PeerID:       peerID,
-		trackedPeers: make(map[string]uint64),
-		ds:           ds,
+		PeerID:        peerID,
+		trackedPeers:  make(map[string]uint64),
+		ds:            ds,
+		localMetadata: make(map[string]string),
+		peerMetadata:  make(map[string]*PeerMetadata),
+		peerTTL:       7 * 24 * time.Hour, // default 7 days, matching go-ds-crdt
 	}
 
 	for _, opt := range opts {
@@ -102,6 +212,7 @@ func New(peerID string, ds datastore.Datastore, opts ...func(*CRDT)) *CRDT {
 	}
 
 	_ = crdt.loadTrackedPeers()
+	_ = crdt.loadPeerMetadata(context.Background())
 	crdt.PeerSeq = crdt.trackedPeers[peerID]
 	return crdt
 }
@@ -161,16 +272,60 @@ func WithDeleteTxnHook(h func(ctx context.Context, txn datastore.Write, key stri
 	return func(c *CRDT) { c.deleteTxnHook = h }
 }
 
+// WithMembershipHook sets a callback that is invoked whenever
+// cluster membership changes. This includes:
+// - New peer joining the cluster
+// - Existing peer updating their metadata
+// - Peer TTL expiring (peer removed from membership)
+//
+// The hook receives a map of all current members with their metadata.
+// The hook is called asynchronously (in a goroutine) and should not block.
+func WithMembershipHook(hook func(members map[string]*PeerMetadata)) func(*CRDT) {
+	return func(c *CRDT) {
+		c.hooksMu.Lock()
+		defer c.hooksMu.Unlock()
+		c.membershipHooks = append(c.membershipHooks, hook)
+	}
+}
+
+// WithPeerTTL sets the duration that a peer's membership remains valid
+// without refresh. Default is 7 days (matching go-ds-crdt default).
+//
+// The TTL is automatically refreshed on each rebroadcast interval.
+func WithPeerTTL(ttl time.Duration) func(*CRDT) {
+	return func(c *CRDT) {
+		c.peerTTL = ttl
+	}
+}
+
+// WithNamespace sets a namespace prefix for all keys in the datastore.
+// This allows multiple independent CRDT instances to coexist in the same
+// underlying datastore without conflicts.
+//
+// Example:
+//
+//	crdt := clset.New(peerID, ds, clset.WithNamespace("ipservice"))
+//	// User keys stored as: /ipservice/data/key
+//	// Internal keys stored as: /ipservice/change/..., /ipservice/tracked/..., etc.
+//
+// If not set, keys are stored at the root level (e.g., /data/key) for
+// backward compatibility with existing datastores.
+func WithNamespace(namespace string) func(*CRDT) {
+	return func(c *CRDT) {
+		c.namespace = namespace
+	}
+}
+
 func (c *CRDT) loadTrackedPeers() error {
 	ctx := context.Background()
 
 	// Use QueryIter for efficient prefix iteration
-	q := query.Query{Prefix: "/tracked/"}
+	q := query.Query{Prefix: c.trackedPrefix()}
 	for entry, err := range datastore.QueryIter(ctx, c.ds, q) {
 		if err != nil {
 			return err
 		}
-		peerID := strings.TrimPrefix(entry.Key, "/tracked/")
+		peerID := strings.TrimPrefix(entry.Key, c.trackedPrefix())
 		seq, err := strconv.ParseUint(string(entry.Value), 10, 64)
 		if err != nil {
 			return err
@@ -181,7 +336,7 @@ func (c *CRDT) loadTrackedPeers() error {
 }
 
 func (c *CRDT) saveTrackedPeer(ctx context.Context, writer datastore.Write, peerID string, seq uint64) error {
-	key := datastore.NewKey("/tracked/" + peerID)
+	key := c.trackedKey(peerID)
 	val := []byte(strconv.FormatUint(seq, 10))
 	return writer.Put(ctx, key, val)
 }
@@ -197,7 +352,7 @@ func decodeEntry(data []byte, entry *KeyEntry) error {
 }
 
 func (c *CRDT) getEntry(ctx context.Context, reader datastore.Read, key string) (KeyEntry, error) {
-	dsKey := datastore.NewKey("/data/" + key)
+	dsKey := c.dataKey(key)
 	val, err := reader.Get(ctx, dsKey)
 	if err != nil {
 		return KeyEntry{}, err
@@ -273,7 +428,7 @@ func (c *CRDT) setWithTransaction(ctx context.Context, txnDs datastore.TxnDatast
 
 		// clean up old index entry
 		if meta.PeerID != "" {
-			oldIndexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", meta.PeerID, meta.PeerSeq))
+			oldIndexKey := c.changeKey(meta.PeerID, meta.PeerSeq)
 			_ = txn.Delete(ctx, oldIndexKey) // ignore error if missing
 		}
 
@@ -282,11 +437,11 @@ func (c *CRDT) setWithTransaction(ctx context.Context, txnDs datastore.TxnDatast
 		newEntry := KeyEntry{Key: key, Value: value, Meta: meta}
 
 		encoded, _ := encodeEntry(newEntry)
-		if err := txn.Put(ctx, datastore.NewKey("/data/"+key), encoded); err != nil {
+		if err := txn.Put(ctx, c.dataKey(key), encoded); err != nil {
 			return 0, err
 		}
 
-		indexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", c.PeerID, peerSeq))
+		indexKey := c.changeKey(c.PeerID, peerSeq)
 		if err := txn.Put(ctx, indexKey, encoded); err != nil {
 			return 0, err
 		}
@@ -355,16 +510,16 @@ func (c *CRDT) setWithBatch(ctx context.Context, batchDs datastore.Batching, key
 
 	// clean up old index entry
 	if entry.Meta.PeerID != "" {
-		oldIndexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", entry.Meta.PeerID, entry.Meta.PeerSeq))
+		oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
 		_ = batch.Delete(ctx, oldIndexKey) // ignore error if missing
 	}
 
 	encoded, _ := encodeEntry(newEntry)
-	if err := batch.Put(ctx, datastore.NewKey("/data/"+key), encoded); err != nil {
+	if err := batch.Put(ctx, c.dataKey(key), encoded); err != nil {
 		return err
 	}
 
-	indexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", c.PeerID, peerSeq))
+	indexKey := c.changeKey(c.PeerID, peerSeq)
 	if err := batch.Put(ctx, indexKey, encoded); err != nil {
 		return err
 	}
@@ -425,16 +580,16 @@ func (c *CRDT) setDirect(ctx context.Context, ds datastore.Datastore, key string
 
 	// clean up old index entry
 	if entry.Meta.PeerID != "" {
-		oldIndexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", entry.Meta.PeerID, entry.Meta.PeerSeq))
+		oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
 		_ = ds.Delete(ctx, oldIndexKey) // ignore error if missing
 	}
 
 	encoded, _ := encodeEntry(newEntry)
-	if err := ds.Put(ctx, datastore.NewKey("/data/"+key), encoded); err != nil {
+	if err := ds.Put(ctx, c.dataKey(key), encoded); err != nil {
 		return err
 	}
 
-	indexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", c.PeerID, peerSeq))
+	indexKey := c.changeKey(c.PeerID, peerSeq)
 	if err := ds.Put(ctx, indexKey, encoded); err != nil {
 		return err
 	}
@@ -509,7 +664,7 @@ func (c *CRDT) deleteWithTransaction(ctx context.Context, txnDs datastore.TxnDat
 
 			// clean up old index entry
 			if meta.PeerID != "" {
-				oldIndexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", meta.PeerID, meta.PeerSeq))
+				oldIndexKey := c.changeKey(meta.PeerID, meta.PeerSeq)
 				_ = txn.Delete(ctx, oldIndexKey) // ignore error if missing
 			}
 
@@ -519,11 +674,11 @@ func (c *CRDT) deleteWithTransaction(ctx context.Context, txnDs datastore.TxnDat
 			deletedEntry := KeyEntry{Key: key, Value: []byte{}, Meta: meta}
 			encoded, _ := encodeEntry(deletedEntry)
 
-			if err := txn.Put(ctx, datastore.NewKey("/data/"+key), encoded); err != nil {
+			if err := txn.Put(ctx, c.dataKey(key), encoded); err != nil {
 				return 0, err
 			}
 
-			indexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", c.PeerID, peerSeq))
+			indexKey := c.changeKey(c.PeerID, peerSeq)
 			if err := txn.Put(ctx, indexKey, encoded); err != nil {
 				return 0, err
 			}
@@ -577,15 +732,15 @@ func (c *CRDT) deleteWithBatch(ctx context.Context, batchDs datastore.Batching,
 
 		// clean up old index entry
 		if entry.Meta.PeerID != "" {
-			oldIndexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", entry.Meta.PeerID, entry.Meta.PeerSeq))
+			oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
 			_ = batch.Delete(ctx, oldIndexKey) // ignore error if missing
 		}
 
-		if err := batch.Put(ctx, datastore.NewKey("/data/"+key), encoded); err != nil {
+		if err := batch.Put(ctx, c.dataKey(key), encoded); err != nil {
 			return err
 		}
 
-		indexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", c.PeerID, peerSeq))
+		indexKey := c.changeKey(c.PeerID, peerSeq)
 		if err := batch.Put(ctx, indexKey, encoded); err != nil {
 			return err
 		}
@@ -632,15 +787,15 @@ func (c *CRDT) deleteDirect(ctx context.Context, ds datastore.Datastore, key str
 
 		// clean up old index entry
 		if entry.Meta.PeerID != "" {
-			oldIndexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", entry.Meta.PeerID, entry.Meta.PeerSeq))
+			oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
 			_ = ds.Delete(ctx, oldIndexKey) // ignore error if missing
 		}
 
-		if err := ds.Put(ctx, datastore.NewKey("/data/"+key), encoded); err != nil {
+		if err := ds.Put(ctx, c.dataKey(key), encoded); err != nil {
 			return err
 		}
 
-		indexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", c.PeerID, peerSeq))
+		indexKey := c.changeKey(c.PeerID, peerSeq)
 		if err := ds.Put(ctx, indexKey, encoded); err != nil {
 			return err
 		}
@@ -657,6 +812,605 @@ func (c *CRDT) deleteDirect(ctx context.Context, ds datastore.Datastore, key str
 	return nil
 }
 
+// PutBatch atomically commits multiple Put operations as a single CRDT change.
+// All operations succeed or fail together, creating only one change log entry.
+// If the map is empty, this is a no-op.
+func (c *CRDT) PutBatch(ops map[string][]byte) error {
+	if len(ops) == 0 {
+		return nil
+	}
+
+	c.mu.Lock()
+
+	var afterCommit []func()
+	peerSeq := c.PeerSeq
+	ctx := context.Background()
+
+	// Try TxnDatastore first (best: full transactional semantics)
+	if txnDs, ok := c.ds.(datastore.TxnDatastore); ok {
+		err := c.putBatchWithTransaction(ctx, txnDs, ops, peerSeq, &afterCommit)
+		if err != nil {
+			c.mu.Unlock()
+			return err
+		}
+	} else if batchDs, ok := c.ds.(datastore.Batching); ok {
+		// Fall back to Batching (good: atomic commit)
+		err := c.putBatchWithBatching(ctx, batchDs, ops, peerSeq, &afterCommit)
+		if err != nil {
+			c.mu.Unlock()
+			return err
+		}
+	} else {
+		// Fall back to direct operations (last resort: no atomicity)
+		err := c.putBatchDirect(ctx, c.ds, ops, peerSeq, &afterCommit)
+		if err != nil {
+			c.mu.Unlock()
+			return err
+		}
+	}
+
+	c.mu.Unlock()
+
+	// Execute post-commit hooks
+	for _, fn := range afterCommit {
+		fn()
+	}
+
+	return nil
+}
+
+func (c *CRDT) putBatchWithTransaction(ctx context.Context, txnDs datastore.TxnDatastore, ops map[string][]byte, peerSeq uint64, afterCommit *[]func()) error {
+	return c.withTransaction(ctx, txnDs, func(ctx context.Context, txn datastore.Txn) (uint64, error) {
+		// Check if any values are actually changing
+		hasChanges := false
+		entries := make(map[string]KeyEntry)
+
+		for key, value := range ops {
+			entry, err := c.getEntry(ctx, txn, key)
+			if err != nil && err != datastore.ErrNotFound {
+				return 0, err
+			}
+			entries[key] = entry
+
+			if !bytes.Equal(entry.Value, value) {
+				hasChanges = true
+			}
+		}
+
+		if !hasChanges {
+			return 0, nil // No changes needed
+		}
+
+		// All operations share the same sequence number
+		peerSeq++
+
+		// Track hooks to execute after commit
+		var hooks []func()
+
+		for key, value := range ops {
+			entry := entries[key]
+			wasPresent := (entry.Meta.CausalLength%2 == 1)
+
+			meta := entry.Meta
+			if entry.Meta.PeerID == "" || meta.CausalLength%2 == 0 {
+				meta.CausalLength++
+				meta.ValueVersion = 1
+			} else {
+				meta.ValueVersion++
+			}
+
+			// Clean up old index entry
+			if meta.PeerID != "" {
+				oldIndexKey := c.changeKey(meta.PeerID, meta.PeerSeq)
+				_ = txn.Delete(ctx, oldIndexKey)
+			}
+
+			meta.PeerID = c.PeerID
+			meta.PeerSeq = peerSeq
+
+			newEntry := KeyEntry{Key: key, Value: value, Meta: meta}
+			encoded, _ := encodeEntry(newEntry)
+
+			if err := txn.Put(ctx, c.dataKey(key), encoded); err != nil {
+				return 0, err
+			}
+
+			indexKey := c.changeKeyWithUserKey(c.PeerID, peerSeq, key)
+			if err := txn.Put(ctx, indexKey, encoded); err != nil {
+				return 0, err
+			}
+
+			// Transactional hooks
+			if !wasPresent && c.insertTxnHook != nil {
+				if err := c.insertTxnHook(ctx, txn, key, value, meta); err != nil {
+					return 0, err
+				}
+			} else if wasPresent && c.updateTxnHook != nil {
+				if err := c.updateTxnHook(ctx, txn, key, entry.Value, entry.Meta, value, meta); err != nil {
+					return 0, err
+				}
+			}
+
+			// Prepare post-commit hooks
+			if !wasPresent {
+				if insertHooks := c.getInsertHooks(); len(insertHooks) > 0 {
+					key, value, meta := key, value, meta // capture for closure
+					hooks = append(hooks, func() { c.runInsertHooks(key, value, meta, insertHooks) })
+				}
+			} else {
+				if updateHooks := c.getUpdateHooks(); len(updateHooks) > 0 {
+					key, oldVal, oldMeta, value, meta := key, entry.Value, entry.Meta, value, meta // capture
+					hooks = append(hooks, func() { c.runUpdateHooks(key, oldVal, oldMeta, value, meta, updateHooks) })
+				}
+			}
+		}
+
+		*afterCommit = hooks
+		return peerSeq, nil
+	})
+}
+
+func (c *CRDT) putBatchWithBatching(ctx context.Context, batchDs datastore.Batching, ops map[string][]byte, peerSeq uint64, afterCommit *[]func()) error {
+	// Check if any values are actually changing
+	hasChanges := false
+	entries := make(map[string]KeyEntry)
+
+	for key, value := range ops {
+		entry, err := c.getEntry(ctx, c.ds, key)
+		if err != nil && err != datastore.ErrNotFound {
+			return err
+		}
+		entries[key] = entry
+
+		if !bytes.Equal(entry.Value, value) {
+			hasChanges = true
+		}
+	}
+
+	if !hasChanges {
+		return nil
+	}
+
+	peerSeq++
+
+	// Claim the sequence number first to prevent reuse
+	if err := c.saveTrackedPeer(ctx, c.ds, c.PeerID, peerSeq); err != nil {
+		return err
+	}
+
+	batch, err := batchDs.Batch(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create batch: %w", err)
+	}
+
+	var hooks []func()
+
+	for key, value := range ops {
+		entry := entries[key]
+		wasPresent := (entry.Meta.CausalLength%2 == 1)
+
+		meta := entry.Meta
+		if entry.Meta.PeerID == "" || meta.CausalLength%2 == 0 {
+			meta.CausalLength++
+			meta.ValueVersion = 1
+		} else {
+			meta.ValueVersion++
+		}
+
+		// Clean up old index entry
+		if entry.Meta.PeerID != "" {
+			oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
+			_ = batch.Delete(ctx, oldIndexKey)
+		}
+
+		meta.PeerID = c.PeerID
+		meta.PeerSeq = peerSeq
+
+		newEntry := KeyEntry{Key: key, Value: value, Meta: meta}
+		encoded, _ := encodeEntry(newEntry)
+
+		if err := batch.Put(ctx, c.dataKey(key), encoded); err != nil {
+			return err
+		}
+
+		indexKey := c.changeKeyWithUserKey(c.PeerID, peerSeq, key)
+		if err := batch.Put(ctx, indexKey, encoded); err != nil {
+			return err
+		}
+
+		// Prepare post-commit hooks
+		if !wasPresent {
+			if insertHooks := c.getInsertHooks(); len(insertHooks) > 0 {
+				key, value, meta := key, value, meta
+				hooks = append(hooks, func() { c.runInsertHooks(key, value, meta, insertHooks) })
+			}
+		} else {
+			if updateHooks := c.getUpdateHooks(); len(updateHooks) > 0 {
+				key, oldVal, oldMeta, value, meta := key, entry.Value, entry.Meta, value, meta
+				hooks = append(hooks, func() { c.runUpdateHooks(key, oldVal, oldMeta, value, meta, updateHooks) })
+			}
+		}
+	}
+
+	if err := batch.Commit(ctx); err != nil {
+		return fmt.Errorf("batch commit error: %w", err)
+	}
+
+	// Update in-memory state
+	c.PeerSeq = peerSeq
+	c.trackedPeers[c.PeerID] = peerSeq
+
+	*afterCommit = hooks
+	return nil
+}
+
+func (c *CRDT) putBatchDirect(ctx context.Context, ds datastore.Datastore, ops map[string][]byte, peerSeq uint64, afterCommit *[]func()) error {
+	// Check if any values are actually changing
+	hasChanges := false
+	entries := make(map[string]KeyEntry)
+
+	for key, value := range ops {
+		entry, err := c.getEntry(ctx, ds, key)
+		if err != nil && err != datastore.ErrNotFound {
+			return err
+		}
+		entries[key] = entry
+
+		if !bytes.Equal(entry.Value, value) {
+			hasChanges = true
+		}
+	}
+
+	if !hasChanges {
+		return nil
+	}
+
+	peerSeq++
+
+	// Claim the sequence number first
+	if err := c.saveTrackedPeer(ctx, c.ds, c.PeerID, peerSeq); err != nil {
+		return err
+	}
+
+	var hooks []func()
+
+	for key, value := range ops {
+		entry := entries[key]
+		wasPresent := (entry.Meta.CausalLength%2 == 1)
+
+		meta := entry.Meta
+		if entry.Meta.PeerID == "" || meta.CausalLength%2 == 0 {
+			meta.CausalLength++
+			meta.ValueVersion = 1
+		} else {
+			meta.ValueVersion++
+		}
+
+		// Clean up old index entry
+		if entry.Meta.PeerID != "" {
+			oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
+			_ = ds.Delete(ctx, oldIndexKey)
+		}
+
+		meta.PeerID = c.PeerID
+		meta.PeerSeq = peerSeq
+
+		newEntry := KeyEntry{Key: key, Value: value, Meta: meta}
+		encoded, _ := encodeEntry(newEntry)
+
+		if err := ds.Put(ctx, c.dataKey(key), encoded); err != nil {
+			return err
+		}
+
+		indexKey := c.changeKeyWithUserKey(c.PeerID, peerSeq, key)
+		if err := ds.Put(ctx, indexKey, encoded); err != nil {
+			return err
+		}
+
+		// Prepare post-commit hooks
+		if !wasPresent {
+			if insertHooks := c.getInsertHooks(); len(insertHooks) > 0 {
+				key, value, meta := key, value, meta
+				hooks = append(hooks, func() { c.runInsertHooks(key, value, meta, insertHooks) })
+			}
+		} else {
+			if updateHooks := c.getUpdateHooks(); len(updateHooks) > 0 {
+				key, oldVal, oldMeta, value, meta := key, entry.Value, entry.Meta, value, meta
+				hooks = append(hooks, func() { c.runUpdateHooks(key, oldVal, oldMeta, value, meta, updateHooks) })
+			}
+		}
+	}
+
+	// Update in-memory state
+	c.PeerSeq = peerSeq
+	c.trackedPeers[c.PeerID] = peerSeq
+
+	*afterCommit = hooks
+	return nil
+}
+
+// DeleteBatch atomically commits multiple Delete operations as a single CRDT change.
+// All operations succeed or fail together, creating only one change log entry.
+// If the slice is empty, this is a no-op.
+func (c *CRDT) DeleteBatch(keys []string) error {
+	if len(keys) == 0 {
+		return nil
+	}
+
+	c.mu.Lock()
+
+	var afterCommit []func()
+	peerSeq := c.PeerSeq
+	ctx := context.Background()
+
+	// Try TxnDatastore first
+	if txnDs, ok := c.ds.(datastore.TxnDatastore); ok {
+		err := c.deleteBatchWithTransaction(ctx, txnDs, keys, peerSeq, &afterCommit)
+		if err != nil {
+			c.mu.Unlock()
+			return err
+		}
+	} else if batchDs, ok := c.ds.(datastore.Batching); ok {
+		// Fall back to Batching
+		err := c.deleteBatchWithBatching(ctx, batchDs, keys, peerSeq, &afterCommit)
+		if err != nil {
+			c.mu.Unlock()
+			return err
+		}
+	} else {
+		// Fall back to direct operations
+		err := c.deleteBatchDirect(ctx, c.ds, keys, peerSeq, &afterCommit)
+		if err != nil {
+			c.mu.Unlock()
+			return err
+		}
+	}
+
+	c.mu.Unlock()
+
+	// Execute post-commit hooks
+	for _, fn := range afterCommit {
+		fn()
+	}
+
+	return nil
+}
+
+func (c *CRDT) deleteBatchWithTransaction(ctx context.Context, txnDs datastore.TxnDatastore, keys []string, peerSeq uint64, afterCommit *[]func()) error {
+	return c.withTransaction(ctx, txnDs, func(ctx context.Context, txn datastore.Txn) (uint64, error) {
+		// Check which keys actually need deletion
+		entries := make(map[string]KeyEntry)
+		needsDeletion := false
+
+		for _, key := range keys {
+			entry, err := c.getEntry(ctx, txn, key)
+			if err != nil && err != datastore.ErrNotFound {
+				return 0, err
+			}
+			entries[key] = entry
+
+			// Only delete if key exists (CausalLength is odd)
+			if entry.Meta.CausalLength%2 == 1 {
+				needsDeletion = true
+			}
+		}
+
+		if !needsDeletion {
+			return 0, nil // Nothing to delete
+		}
+
+		peerSeq++
+		var hooks []func()
+
+		for _, key := range keys {
+			entry := entries[key]
+			meta := entry.Meta
+
+			// Only process if key currently exists
+			if meta.CausalLength%2 == 1 {
+				meta.CausalLength++
+				meta.ValueVersion = 0
+
+				// Clean up old index entry
+				if meta.PeerID != "" {
+					oldIndexKey := c.changeKey(meta.PeerID, meta.PeerSeq)
+					_ = txn.Delete(ctx, oldIndexKey)
+				}
+
+				meta.PeerID = c.PeerID
+				meta.PeerSeq = peerSeq
+
+				deletedEntry := KeyEntry{Key: key, Value: []byte{}, Meta: meta}
+				encoded, _ := encodeEntry(deletedEntry)
+
+				if err := txn.Put(ctx, c.dataKey(key), encoded); err != nil {
+					return 0, err
+				}
+
+				indexKey := c.changeKeyWithUserKey(c.PeerID, peerSeq, key)
+				if err := txn.Put(ctx, indexKey, encoded); err != nil {
+					return 0, err
+				}
+
+				// Transactional hook
+				if c.deleteTxnHook != nil {
+					if err := c.deleteTxnHook(ctx, txn, key, entry.Value, entry.Meta); err != nil {
+						return 0, err
+					}
+				}
+
+				// Prepare post-commit hooks
+				if deleteHooks := c.getDeleteHooks(); len(deleteHooks) > 0 {
+					key, oldVal, oldMeta := key, entry.Value, entry.Meta
+					hooks = append(hooks, func() { c.runDeleteHooks(key, oldVal, oldMeta, deleteHooks) })
+				}
+			}
+		}
+
+		*afterCommit = hooks
+		return peerSeq, nil
+	})
+}
+
+func (c *CRDT) deleteBatchWithBatching(ctx context.Context, batchDs datastore.Batching, keys []string, peerSeq uint64, afterCommit *[]func()) error {
+	// Check which keys actually need deletion
+	entries := make(map[string]KeyEntry)
+	needsDeletion := false
+
+	for _, key := range keys {
+		entry, err := c.getEntry(ctx, c.ds, key)
+		if err != nil && err != datastore.ErrNotFound {
+			return err
+		}
+		entries[key] = entry
+
+		if entry.Meta.CausalLength%2 == 1 {
+			needsDeletion = true
+		}
+	}
+
+	if !needsDeletion {
+		return nil
+	}
+
+	peerSeq++
+
+	// Claim the sequence number first
+	if err := c.saveTrackedPeer(ctx, c.ds, c.PeerID, peerSeq); err != nil {
+		return err
+	}
+
+	batch, err := batchDs.Batch(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create batch: %w", err)
+	}
+
+	var hooks []func()
+
+	for _, key := range keys {
+		entry := entries[key]
+		meta := entry.Meta
+
+		if meta.CausalLength%2 == 1 {
+			meta.CausalLength++
+			meta.ValueVersion = 0
+
+			// Clean up old index entry
+			if entry.Meta.PeerID != "" {
+				oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
+				_ = batch.Delete(ctx, oldIndexKey)
+			}
+
+			meta.PeerID = c.PeerID
+			meta.PeerSeq = peerSeq
+
+			deletedEntry := KeyEntry{Key: key, Value: []byte{}, Meta: meta}
+			encoded, _ := encodeEntry(deletedEntry)
+
+			if err := batch.Put(ctx, c.dataKey(key), encoded); err != nil {
+				return err
+			}
+
+			indexKey := c.changeKeyWithUserKey(c.PeerID, peerSeq, key)
+			if err := batch.Put(ctx, indexKey, encoded); err != nil {
+				return err
+			}
+
+			// Prepare post-commit hooks
+			if deleteHooks := c.getDeleteHooks(); len(deleteHooks) > 0 {
+				key, oldVal, oldMeta := key, entry.Value, entry.Meta
+				hooks = append(hooks, func() { c.runDeleteHooks(key, oldVal, oldMeta, deleteHooks) })
+			}
+		}
+	}
+
+	if err := batch.Commit(ctx); err != nil {
+		return fmt.Errorf("batch commit error: %w", err)
+	}
+
+	// Update in-memory state
+	c.PeerSeq = peerSeq
+	c.trackedPeers[c.PeerID] = peerSeq
+
+	*afterCommit = hooks
+	return nil
+}
+
+func (c *CRDT) deleteBatchDirect(ctx context.Context, ds datastore.Datastore, keys []string, peerSeq uint64, afterCommit *[]func()) error {
+	// Check which keys actually need deletion
+	entries := make(map[string]KeyEntry)
+	needsDeletion := false
+
+	for _, key := range keys {
+		entry, err := c.getEntry(ctx, ds, key)
+		if err != nil && err != datastore.ErrNotFound {
+			return err
+		}
+		entries[key] = entry
+
+		if entry.Meta.CausalLength%2 == 1 {
+			needsDeletion = true
+		}
+	}
+
+	if !needsDeletion {
+		return nil
+	}
+
+	peerSeq++
+
+	// Claim the sequence number first
+	if err := c.saveTrackedPeer(ctx, c.ds, c.PeerID, peerSeq); err != nil {
+		return err
+	}
+
+	var hooks []func()
+
+	for _, key := range keys {
+		entry := entries[key]
+		meta := entry.Meta
+
+		if meta.CausalLength%2 == 1 {
+			meta.CausalLength++
+			meta.ValueVersion = 0
+
+			// Clean up old index entry
+			if entry.Meta.PeerID != "" {
+				oldIndexKey := c.changeKey(entry.Meta.PeerID, entry.Meta.PeerSeq)
+				_ = ds.Delete(ctx, oldIndexKey)
+			}
+
+			meta.PeerID = c.PeerID
+			meta.PeerSeq = peerSeq
+
+			deletedEntry := KeyEntry{Key: key, Value: []byte{}, Meta: meta}
+			encoded, _ := encodeEntry(deletedEntry)
+
+			if err := ds.Put(ctx, c.dataKey(key), encoded); err != nil {
+				return err
+			}
+
+			indexKey := c.changeKeyWithUserKey(c.PeerID, peerSeq, key)
+			if err := ds.Put(ctx, indexKey, encoded); err != nil {
+				return err
+			}
+
+			// Prepare post-commit hooks
+			if deleteHooks := c.getDeleteHooks(); len(deleteHooks) > 0 {
+				key, oldVal, oldMeta := key, entry.Value, entry.Meta
+				hooks = append(hooks, func() { c.runDeleteHooks(key, oldVal, oldMeta, deleteHooks) })
+			}
+		}
+	}
+
+	// Update in-memory state
+	c.PeerSeq = peerSeq
+	c.trackedPeers[c.PeerID] = peerSeq
+
+	*afterCommit = hooks
+	return nil
+}
+
 func (c *CRDT) Get(key string) ([]byte, bool, error) {
 	ctx := context.Background()
 	entry, err := c.getEntry(ctx, c.ds, key)
@@ -693,7 +1447,7 @@ func (c *CRDT) KeyCount() (uint64, error) {
 	ctx := context.Background()
 
 	// Use QueryIter for efficient prefix iteration
-	q := query.Query{Prefix: "/data/"}
+	q := query.Query{Prefix: c.dataPrefix()}
 	for entry, err := range datastore.QueryIter(ctx, c.ds, q) {
 		if err != nil {
 			return 0, err
@@ -710,6 +1464,58 @@ func (c *CRDT) KeyCount() (uint64, error) {
 	return count, nil
 }
 
+// Query executes a query against the datastore.
+// This method exposes the underlying datastore's Query functionality,
+// allowing callers to perform operations like prefix-based key filtering.
+// Query executes a query against the CRDT's data.
+// The query prefix is automatically adjusted to include the namespace and /data/ prefix.
+// For example, if the namespace is "ipservice" and the query prefix is "/pool/",
+// the actual query will use "/ipservice/data/pool/".
+// Results have the namespace and /data/ prefix stripped from keys.
+func (c *CRDT) Query(ctx context.Context, q query.Query) (query.Results, error) {
+	// Adjust the query prefix to include namespace and /data/
+	if q.Prefix != "" {
+		// User provides prefixes like "/pool/" or "pool/"
+		// We need to convert to "/namespace/data/pool/" or "/data/pool/"
+		q.Prefix = c.dataPrefix() + strings.TrimPrefix(q.Prefix, "/")
+	}
+
+	results, err := c.ds.Query(ctx, q)
+	if err != nil {
+		return nil, err
+	}
+
+	// Wrap results to strip namespace+data prefix from keys
+	return &namespacedQueryResults{
+		Results: results,
+		prefix:  c.dataPrefix(),
+	}, nil
+}
+
+// namespacedQueryResults wraps query.Results to strip the namespace+data prefix from keys
+type namespacedQueryResults struct {
+	query.Results
+	prefix string
+}
+
+func (n *namespacedQueryResults) Next() <-chan query.Result {
+	originalCh := n.Results.Next()
+	strippedCh := make(chan query.Result)
+
+	go func() {
+		defer close(strippedCh)
+		for result := range originalCh {
+			// Strip the namespace+data prefix from the key
+			if strings.HasPrefix(result.Key, n.prefix) {
+				result.Key = result.Key[len(n.prefix):]
+			}
+			strippedCh <- result
+		}
+	}()
+
+	return strippedCh
+}
+
 // GetTrackedPeers returns a copy of the tracked peers map
 func (c *CRDT) GetTrackedPeers() map[string]uint64 {
 	c.mu.Lock()
@@ -731,7 +1537,7 @@ func (c *CRDT) GetLatestChanges(requestorID string, requestorTracked map[string]
 	count := 0
 	for peerID := range c.trackedPeers {
 		startSeq := requestorTracked[peerID] + 1
-		prefix := fmt.Sprintf("/change/%s/", peerID)
+		prefix := c.changePrefix(peerID)
 
 		// Create query for this peer's changes
 		q := query.Query{Prefix: prefix}
@@ -803,16 +1609,16 @@ func (c *CRDT) mergeChangeBatch(ctx context.Context, writer datastore.Write, cha
 		if isRemoteWinner(incoming, existing) {
 			// Clean up old index
 			if existing.Meta.PeerID != "" {
-				oldIndexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", existing.Meta.PeerID, existing.Meta.PeerSeq))
+				oldIndexKey := c.changeKey(existing.Meta.PeerID, existing.Meta.PeerSeq)
 				_ = writer.Delete(ctx, oldIndexKey)
 			}
 
 			encoded, _ := encodeEntry(incoming)
-			if err := writer.Put(ctx, datastore.NewKey("/data/"+incoming.Key), encoded); err != nil {
+			if err := writer.Put(ctx, c.dataKey(incoming.Key), encoded); err != nil {
 				return nil, err
 			}
 
-			indexKey := datastore.NewKey(fmt.Sprintf("/change/%s/%010d", incoming.Meta.PeerID, incoming.Meta.PeerSeq))
+			indexKey := c.changeKey(incoming.Meta.PeerID, incoming.Meta.PeerSeq)
 			if err := writer.Put(ctx, indexKey, encoded); err != nil {
 				return nil, err
 			}
@@ -961,3 +1767,159 @@ func (c *CRDT) MergeChanges(fromPeerID string, changes []KeyEntry, tracked map[s
 
 	return nil
 }
+
+// UpdateMeta sets or updates the local peer's metadata.
+// This metadata will be broadcast to all other peers via gossip.
+//
+// The TTL is automatically managed - each rebroadcast updates the
+// BestBefore timestamp to Now + TTL duration.
+//
+// Returns error if metadata cannot be persisted.
+func (c *CRDT) UpdateMeta(ctx context.Context, metadata map[string]string) error {
+	c.metadataMu.Lock()
+	c.localMetadata = make(map[string]string)
+	for k, v := range metadata {
+		c.localMetadata[k] = v
+	}
+	c.metadataMu.Unlock()
+
+	// Persist to datastore
+	key := c.metadataKey(c.PeerID)
+	data, err := json.Marshal(metadata)
+	if err != nil {
+		return fmt.Errorf("failed to marshal metadata: %w", err)
+	}
+
+	if err := c.ds.Put(ctx, key, data); err != nil {
+		return fmt.Errorf("failed to persist metadata: %w", err)
+	}
+
+	return nil
+}
+
+// GetState returns the current membership state of the cluster.
+// This includes all known peers with their metadata and TTL status.
+//
+// The returned state is a snapshot at the time of the call.
+// Expired peers are filtered out based on their BestBefore timestamp.
+func (c *CRDT) GetState(ctx context.Context) *MembershipState {
+	c.metadataMu.RLock()
+	defer c.metadataMu.RUnlock()
+
+	state := &MembershipState{
+		Members: make(map[string]*PeerMetadata),
+	}
+
+	now := uint64(time.Now().Unix())
+
+	for peerID, meta := range c.peerMetadata {
+		// Filter out expired peers
+		if meta.BestBefore >= now {
+			// Deep copy to prevent mutation
+			state.Members[peerID] = &PeerMetadata{
+				BestBefore: meta.BestBefore,
+				Metadata:   copyStringMap(meta.Metadata),
+				Sequence:   c.trackedPeers[peerID],
+			}
+		}
+	}
+
+	return state
+}
+
+// AddSelfToPeerMetadata manually adds the local peer to its own peer metadata.
+// This is useful for single-node scenarios or testing where the node needs to
+// see itself in the membership immediately without waiting for P2P gossip.
+//
+// In multi-node scenarios, peers learn about each other through P2P gossip, but
+// a node ignores its own gossip messages. This method ensures the local peer
+// can see itself in the membership state.
+func (c *CRDT) AddSelfToPeerMetadata() {
+	c.metadataMu.Lock()
+
+	// Get local metadata
+	metadata := make(map[string]string)
+	for k, v := range c.localMetadata {
+		metadata[k] = v
+	}
+
+	// Calculate TTL for local peer
+	bestBefore := uint64(time.Now().Add(c.peerTTL).Unix())
+
+	// Add self to peer metadata
+	c.peerMetadata[c.PeerID] = &PeerMetadata{
+		BestBefore: bestBefore,
+		Metadata:   metadata,
+		Sequence:   c.PeerSeq,
+	}
+	c.metadataMu.Unlock()
+
+	// Trigger membership hooks
+	c.runMembershipHooks()
+}
+
+// loadPeerMetadata loads peer metadata from the datastore on startup
+func (c *CRDT) loadPeerMetadata(ctx context.Context) error {
+	q := query.Query{Prefix: c.metadataPrefix()}
+	results, err := c.ds.Query(ctx, q)
+	if err != nil {
+		return err
+	}
+	defer results.Close()
+
+	c.metadataMu.Lock()
+	defer c.metadataMu.Unlock()
+
+	for result := range results.Next() {
+		if result.Error != nil {
+			return result.Error
+		}
+
+		peerID := strings.TrimPrefix(result.Key, c.metadataPrefix())
+
+		var metadata map[string]string
+		if err := json.Unmarshal(result.Value, &metadata); err != nil {
+			return err
+		}
+
+		// Initialize with expired TTL - will be updated by gossip
+		if c.peerMetadata[peerID] == nil {
+			c.peerMetadata[peerID] = &PeerMetadata{
+				BestBefore: 0,
+				Metadata:   metadata,
+			}
+		} else {
+			c.peerMetadata[peerID].Metadata = metadata
+		}
+	}
+
+	return nil
+}
+
+// runMembershipHooks calls all registered membership hooks with current state
+func (c *CRDT) runMembershipHooks() {
+	c.hooksMu.RLock()
+	hooks := c.membershipHooks
+	c.hooksMu.RUnlock()
+
+	if len(hooks) == 0 {
+		return
+	}
+
+	state := c.GetState(context.Background())
+
+	for _, hook := range hooks {
+		go hook(state.Members)
+	}
+}
+
+func copyStringMap(m map[string]string) map[string]string {
+	if m == nil {
+		return nil
+	}
+	result := make(map[string]string, len(m))
+	for k, v := range m {
+		result[k] = v
+	}
+	return result
+}
diff --git a/crdt_test.go b/crdt_test.go
index d1a90e0..e060708 100644
--- a/crdt_test.go
+++ b/crdt_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	badgerds "github.com/ipfs/go-ds-badger4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -489,3 +490,151 @@ func TestCRDT_RemoteHooks_OnMerge(t *testing.T) {
 	require.NoError(t, crdt2.MergeChanges("p1", changes, tracked))
 	assert.Contains(t, inserts, "foo:v3")
 }
+
+func TestCRDT_Query(t *testing.T) {
+	ds := createTestDatastore(t)
+	crdt := clset.New("peer1", ds)
+
+	// Setup: Create test data with different prefixes
+	testData := map[string][]byte{
+		"pool/pool1/subscriber/sub1": []byte("allocation1"),
+		"pool/pool1/subscriber/sub2": []byte("allocation2"),
+		"pool/pool2/subscriber/sub3": []byte("allocation3"),
+		"subscriber/sub1":            []byte("metadata1"),
+		"subscriber/sub2":            []byte("metadata2"),
+		"other/data":                 []byte("other"),
+	}
+
+	for key, value := range testData {
+		require.NoError(t, crdt.Set(key, value))
+	}
+
+	t.Run("query with prefix - pool1 allocations", func(t *testing.T) {
+		ctx := t.Context()
+		q := query.Query{Prefix: "/data/pool/pool1/"}
+
+		results, err := crdt.Query(ctx, q)
+		require.NoError(t, err)
+		require.NotNil(t, results)
+		defer results.Close()
+
+		// Collect results
+		var keys []string
+		for result := range results.Next() {
+			require.NoError(t, result.Error)
+			keys = append(keys, result.Key)
+		}
+
+		// Should find 2 allocations under pool1
+		assert.Len(t, keys, 2)
+		assert.Contains(t, keys, "/data/pool/pool1/subscriber/sub1")
+		assert.Contains(t, keys, "/data/pool/pool1/subscriber/sub2")
+	})
+
+	t.Run("query with prefix - subscriber metadata", func(t *testing.T) {
+		ctx := t.Context()
+		q := query.Query{Prefix: "/data/subscriber/"}
+
+		results, err := crdt.Query(ctx, q)
+		require.NoError(t, err)
+		require.NotNil(t, results)
+		defer results.Close()
+
+		// Collect results
+		var keys []string
+		for result := range results.Next() {
+			require.NoError(t, result.Error)
+			keys = append(keys, result.Key)
+		}
+
+		// Should find 2 subscriber records
+		assert.Len(t, keys, 2)
+		assert.Contains(t, keys, "/data/subscriber/sub1")
+		assert.Contains(t, keys, "/data/subscriber/sub2")
+	})
+
+	t.Run("query with prefix - all pools", func(t *testing.T) {
+		ctx := t.Context()
+		q := query.Query{Prefix: "/data/pool/"}
+
+		results, err := crdt.Query(ctx, q)
+		require.NoError(t, err)
+		require.NotNil(t, results)
+		defer results.Close()
+
+		// Collect results
+		var keys []string
+		for result := range results.Next() {
+			require.NoError(t, result.Error)
+			keys = append(keys, result.Key)
+		}
+
+		// Should find 3 pool allocations (2 from pool1, 1 from pool2)
+		assert.Len(t, keys, 3)
+	})
+
+	t.Run("query with non-matching prefix", func(t *testing.T) {
+		ctx := t.Context()
+		q := query.Query{Prefix: "/data/nonexistent/"}
+
+		results, err := crdt.Query(ctx, q)
+		require.NoError(t, err)
+		require.NotNil(t, results)
+		defer results.Close()
+
+		// Collect results
+		var keys []string
+		for result := range results.Next() {
+			require.NoError(t, result.Error)
+			keys = append(keys, result.Key)
+		}
+
+		// Should find no results
+		assert.Len(t, keys, 0)
+	})
+
+	t.Run("query all data keys", func(t *testing.T) {
+		ctx := t.Context()
+		q := query.Query{Prefix: "/data/"}
+
+		results, err := crdt.Query(ctx, q)
+		require.NoError(t, err)
+		require.NotNil(t, results)
+		defer results.Close()
+
+		// Collect results
+		var keys []string
+		for result := range results.Next() {
+			require.NoError(t, result.Error)
+			keys = append(keys, result.Key)
+		}
+
+		// Should find all 6 test keys
+		assert.Len(t, keys, 6)
+	})
+
+	t.Run("query returns values", func(t *testing.T) {
+		ctx := t.Context()
+		q := query.Query{Prefix: "/data/pool/pool1/"}
+
+		results, err := crdt.Query(ctx, q)
+		require.NoError(t, err)
+		require.NotNil(t, results)
+		defer results.Close()
+
+		// Collect at least one result
+		foundResults := false
+		for result := range results.Next() {
+			require.NoError(t, result.Error)
+
+			// Result should have key and value from the datastore
+			// Note: The value contains CRDT metadata, not the raw user value
+			assert.NotEmpty(t, result.Key)
+			assert.NotNil(t, result.Value)
+			foundResults = true
+		}
+
+		// Should have found at least one result
+		assert.True(t, foundResults, "Expected to find at least one result")
+	})
+}
diff --git a/metadata_test.go b/metadata_test.go
new file mode 100644
index 0000000..248bb1f
--- /dev/null
+++ b/metadata_test.go
@@ -0,0 +1,167 @@
+package clset
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/ipfs/go-datastore"
+	dssync "github.com/ipfs/go-datastore/sync"
+)
+
+func TestUpdateMeta(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	crdt := New("peer1", ds)
+
+	metadata := map[string]string{
+		"name": "TestApp",
+		"role": "write",
+	}
+
+	err := crdt.UpdateMeta(context.Background(), metadata)
+	if err != nil {
+		t.Fatalf("UpdateMeta failed: %v", err)
+	}
+
+	// Verify metadata was persisted
+	crdt2 := New("peer1", ds)
+	crdt2.metadataMu.RLock()
+	if len(crdt2.peerMetadata) == 0 {
+		t.Fatal("Expected metadata to be loaded from datastore")
+	}
+	crdt2.metadataMu.RUnlock()
+}
+
+func TestGetState(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	crdt := New("peer1", ds)
+
+	// Add some peer metadata manually
+	crdt.metadataMu.Lock()
+	crdt.peerMetadata["peer2"] = &PeerMetadata{
+		BestBefore: uint64(time.Now().Add(1 * time.Hour).Unix()),
+		Metadata: map[string]string{
+			"name": "TestApp",
+			"role": "write",
+		},
+		Sequence: 42,
+	}
+	// Add an expired peer
+	crdt.peerMetadata["peer3"] = &PeerMetadata{
+		BestBefore: uint64(time.Now().Add(-1 * time.Hour).Unix()), // expired
+		Metadata: map[string]string{
+			"name": "TestApp",
+			"role": "read",
+		},
+		Sequence: 10,
+	}
+	crdt.metadataMu.Unlock()
+
+	state := crdt.GetState(context.Background())
+
+	// Should only include non-expired peers
+	if len(state.Members) != 1 {
+		t.Fatalf("Expected 1 member (peer3 should be filtered as expired), got %d", len(state.Members))
+	}
+
+	peer2, exists := state.Members["peer2"]
+	if !exists {
+		t.Fatal("Expected peer2 to be in state")
+	}
+
+	if peer2.Metadata["name"] != "TestApp" {
+		t.Errorf("Expected name=TestApp, got %s", peer2.Metadata["name"])
+	}
+
+	if peer2.Metadata["role"] != "write" {
+		t.Errorf("Expected role=write, got %s", peer2.Metadata["role"])
+	}
+}
+
+func TestMembershipHook(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+
+	hookCalled := make(chan bool, 1)
+	var receivedMembers map[string]*PeerMetadata
+
+	crdt := New("peer1", ds, WithMembershipHook(func(members map[string]*PeerMetadata) {
+		receivedMembers = members
+		select {
+		case hookCalled <- true:
+		default:
+		}
+	}))
+
+	// Add a peer
+	crdt.metadataMu.Lock()
+	crdt.peerMetadata["peer2"] = &PeerMetadata{
+		BestBefore: uint64(time.Now().Add(1 * time.Hour).Unix()),
+		Metadata: map[string]string{
+			"name": "TestApp",
+		},
+	}
+	crdt.metadataMu.Unlock()
+
+	// Trigger hook
+	crdt.runMembershipHooks()
+
+	// Wait for hook to be called (with timeout)
+	select {
+	case <-hookCalled:
+		// Success
+	case <-time.After(1 * time.Second):
+		t.Fatal("Membership hook was not called within timeout")
+	}
+
+	// Verify received data
+	if len(receivedMembers) != 1 {
+		t.Fatalf("Expected 1 member in hook callback, got %d", len(receivedMembers))
+	}
+
+	if receivedMembers["peer2"].Metadata["name"] != "TestApp" {
+		t.Error("Hook received incorrect metadata")
+	}
+}
+
+func TestWithPeerTTL(t *testing.T) {
+	ds := dssync.MutexWrap(datastore.NewMapDatastore())
+	customTTL := 1 * time.Hour
+
+	crdt := New("peer1", ds, WithPeerTTL(customTTL))
+
+	if crdt.peerTTL != customTTL {
+		t.Errorf("Expected TTL to be %v, got %v", customTTL, crdt.peerTTL)
+	}
+}
+
+func TestCopyStringMap(t *testing.T) {
+	original := map[string]string{
+		"key1": "value1",
+		"key2": "value2",
+	}
+
+	copied := copyStringMap(original)
+
+	// Verify contents match
+	if len(copied) != len(original) {
+		t.Fatalf("Expected %d entries, got %d", len(original), len(copied))
+	}
+
+	for k, v := range original {
+		if copied[k] != v {
+			t.Errorf("Expected %s=%s, got %s", k, v, copied[k])
+		}
+	}
+
+	// Verify it's a deep copy (modifying copied doesn't affect original)
+	copied["key3"] = "value3"
+	if _, exists := original["key3"]; exists {
+		t.Error("Modifying copied map affected original map")
+	}
+
+	// Test nil map
+	nilCopy := copyStringMap(nil)
+	if nilCopy != nil {
+		t.Error("copyStringMap(nil) should return nil")
+	}
+}
diff --git a/p2p_sync.go b/p2p_sync.go
index d7c1252..a124a5b 100644
--- a/p2p_sync.go
+++ b/p2p_sync.go
@@ -59,8 +59,10 @@ type Peer struct {
 }
 
 type SummaryMessage struct {
-	PeerID  string            `json:"peer_id"`
-	Tracked map[string]uint64 `json:"tracked"`
+	PeerID   string            `json:"peer_id"`
+	Tracked  map[string]uint64 `json:"tracked"`
+	Metadata map[string]string `json:"metadata"` // NEW: Peer metadata
+	TTL      uint64            `json:"ttl"`      // NEW: BestBefore timestamp
 }
 
 type cachedSummary struct {
@@ -175,6 +177,9 @@ func NewPeer(crdt *CRDT, ctx context.Context, host host.Host, opts ...PeerOption
 	go p2p.broadcastSummaries()
 	go p2p.scheduledSync()
 
+	// Start TTL cleanup for expired peers
+	go p2p.cleanupExpiredPeers()
+
 	log.Printf("P2P CRDT node started")
 	log.Printf("Peer ID: %s", host.ID())
 	log.Printf("Listening on: %v", host.Addrs())
@@ -582,6 +587,18 @@ func (p *Peer) broadcastSummaries() {
 			return
 		case <-p.summaryCh:
 			tracked := p.crdt.GetTrackedPeers()
+
+			// Get local metadata
+			p.crdt.metadataMu.RLock()
+			metadata := make(map[string]string)
+			for k, v := range p.crdt.localMetadata {
+				metadata[k] = v
+			}
+			p.crdt.metadataMu.RUnlock()
+
+			// Calculate TTL (BestBefore timestamp)
+			ttl := uint64(time.Now().Add(p.crdt.peerTTL).Unix())
+
 			// Avoid re-sending identical state too often
 			if equalTracked(tracked, p.lastSummary) && time.Since(p.lastPub) < p.config.MinSummaryInterval {
 				continue
@@ -590,12 +607,15 @@ func (p *Peer) broadcastSummaries() {
 			p.lastPub = time.Now()
 
 			msg := SummaryMessage{
-				PeerID:  p.Host.ID().String(),
-				Tracked: tracked,
+				PeerID:   p.Host.ID().String(),
+				Tracked:  tracked,
+				Metadata: metadata,
+				TTL:      ttl,
 			}
 			data, err := json.Marshal(msg)
 			if err != nil {
 				log.Printf("Failed to marshal summary: %v", err)
+				continue
 			}
 			if err := p.topic.Publish(p.ctx, data); err != nil {
 				log.Printf("Failed to publish summary: %v", err)
@@ -652,6 +672,34 @@ func (p *Peer) readSummaries() {
 			continue
 		}
 
+		// Update peer metadata
+		p.crdt.metadataMu.Lock()
+		changed := false
+
+		existing := p.crdt.peerMetadata[msg.PeerID]
+		if existing == nil {
+			p.crdt.peerMetadata[msg.PeerID] = &PeerMetadata{
+				BestBefore: msg.TTL,
+				Metadata:   msg.Metadata,
+			}
+			changed = true
+		} else {
+			// Update if changed
+			if existing.BestBefore != msg.TTL ||
+				!mapsEqual(existing.Metadata, msg.Metadata) {
+				existing.BestBefore = msg.TTL
+				existing.Metadata = msg.Metadata
+				changed = true
+			}
+		}
+		p.crdt.metadataMu.Unlock()
+
+		// Trigger membership hook if metadata changed
+		if changed {
+			p.crdt.runMembershipHooks()
+		}
+
+		// Cache summary for sync scheduling
 		p.cacheMu.Lock()
 		p.summaryCache[msg.PeerID] = cachedSummary{
 			Msg:       msg,
@@ -661,6 +709,19 @@ func (p *Peer) readSummaries() {
 	}
 }
 
+// mapsEqual compares two string maps for equality
+func mapsEqual(a, b map[string]string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for k, v := range a {
+		if b[k] != v {
+			return false
+		}
+	}
+	return true
+}
+
 // scheduledSync periodically evaluates cached peer summaries and chooses a peer to synchronize with.
 // It selects the peer with the largest sequence number gap in tracked states.
 func (p *Peer) scheduledSync() {
@@ -733,3 +794,35 @@ func (p *Peer) cleanupSummaryCache() {
 		}
 	}
 }
+
+// cleanupExpiredPeers periodically removes peers whose TTL has expired.
+// This runs in a background goroutine and triggers the membership hook when peers are removed.
+func (p *Peer) cleanupExpiredPeers() {
+	ticker := time.NewTicker(1 * time.Minute)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-p.ctx.Done():
+			return
+		case <-ticker.C:
+			p.crdt.metadataMu.Lock()
+			now := uint64(time.Now().Unix())
+			removed := false
+
+			for peerID, meta := range p.crdt.peerMetadata {
+				if meta.BestBefore < now {
+					delete(p.crdt.peerMetadata, peerID)
+					removed = true
+					log.Printf("Removed expired peer: %s (TTL expired)", peerID)
+				}
+			}
+			p.crdt.metadataMu.Unlock()
+
+			// Trigger membership hook if any peers were removed
+			if removed {
+				p.crdt.runMembershipHooks()
+			}
+		}
+	}
+}
diff --git a/types.go b/types.go
index 2b30c05..c235e6f 100644
--- a/types.go
+++ b/types.go
@@ -12,3 +12,25 @@ type KeyEntry struct {
 	Value []byte
 	Meta  CRDTKeyMeta
 }
+
+// OpType represents the type of operation in a batch change
+type OpType int
+
+const (
+	OpTypePut OpType = iota
+	OpTypeDelete
+)
+
+// Operation represents a single operation within a batch change
+type Operation struct {
+	Type  OpType
+	Key   string
+	Value []byte // Only used for Put operations
+}
+
+// BatchChange represents multiple operations committed atomically as a single CRDT change
+type BatchChange struct {
+	PeerID string
+	Seq    uint64
+	Ops    []Operation
+}