Repository: Tochemey/goakt
Pull request #210 (2024-01-14 19:58:18)

refactor!: refactor the event streams subscription
BREAKING CHANGE: 

- Subscribe method does not take any argument any longer
- Unsubscribe method only take the consumer instance. 
- The following events are handled out of the box:
  - ActorStarted
  - ActorStopped
  - ActorRestarted
  - ActorChildCreated
  - ActorPassivated
  - Deadletter

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/210?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
Attention: `8 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`6bc65f7`)](https://app.codecov.io/gh/Tochemey/goakt/commit/6bc65f7ceecc1eab577c0230532a3c7a686cca9d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 88.74% compared to head [(`883b1b7`)](https://app.codecov.io/gh/Tochemey/goakt/pull/210?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 88.71%.

| [Files](https://app.codecov.io/gh/Tochemey/goakt/pull/210?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Patch % | Lines |
|---|---|---|
| [actors/pid.go](https://app.codecov.io/gh/Tochemey/goakt/pull/210?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-YWN0b3JzL3BpZC5nbw==) | 81.39% | [7 Missing and 1 partial :warning: ](https://app.codecov.io/gh/Tochemey/goakt/pull/210?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #210      +/-   ##
==========================================
- Coverage   88.74%   88.71%   -0.03%     
==========================================
  Files          44       44              
  Lines        4886     4920      +34     
==========================================
+ Hits         4336     4365      +29     
- Misses        434      439       +5     
  Partials      116      116              
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/210?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).


Patch:
diff --git a/actors/actor_system.go b/actors/actor_system.go
index 1b0df47..9ce6a46 100644
--- a/actors/actor_system.go
+++ b/actors/actor_system.go
@@ -45,7 +45,6 @@ import (
 	"github.com/tochemey/goakt/log"
 	"github.com/tochemey/goakt/metric"
 	addresspb "github.com/tochemey/goakt/pb/address/v1"
-	eventspb "github.com/tochemey/goakt/pb/events/v1"
 	"github.com/tochemey/goakt/pkg/types"
 	"github.com/tochemey/goakt/telemetry"
 	"go.opentelemetry.io/otel/codes"
@@ -94,9 +93,9 @@ type ActorSystem interface {
 	// GetPartition returns the partition where a given actor is located
 	GetPartition(actorName string) uint64
 	// Subscribe creates an event subscriber.
-	Subscribe(event eventspb.Event) (eventstream.Subscriber, error)
+	Subscribe() (eventstream.Subscriber, error)
 	// Unsubscribe unsubscribes a subscriber.
-	Unsubscribe(event eventspb.Event, subscriber eventstream.Subscriber) error
+	Unsubscribe(subscriber eventstream.Subscriber) error
 	// ScheduleOnce schedules a message that will be delivered to the receiver actor
 	// This will send the given message to the actor after the given interval specified.
 	// The message will be sent once
@@ -307,37 +306,28 @@ func (x *actorSystem) RemoteScheduleWithCron(ctx context.Context, message proto.
 }
 
 // Subscribe help receive dead letters whenever there are available
-func (x *actorSystem) Subscribe(event eventspb.Event) (eventstream.Subscriber, error) {
+func (x *actorSystem) Subscribe() (eventstream.Subscriber, error) {
 	// first check whether the actor system has started
 	if !x.hasStarted.Load() {
 		return nil, ErrActorSystemNotStarted
 	}
 	// create the consumer
 	subscriber := x.eventsStream.AddSubscriber()
-	// based upon the event we will subscribe to the various topic
-	switch event {
-	case eventspb.Event_DEAD_LETTER:
-		// subscribe the consumer to the deadletter topic
-		x.eventsStream.Subscribe(subscriber, deadlettersTopic)
-	}
+	// subscribe the consumer to the deadletter topic
+	x.eventsStream.Subscribe(subscriber, eventsTopic)
 	return subscriber, nil
 }
 
 // Unsubscribe unsubscribes a subscriber.
-func (x *actorSystem) Unsubscribe(event eventspb.Event, subscriber eventstream.Subscriber) error {
+func (x *actorSystem) Unsubscribe(subscriber eventstream.Subscriber) error {
 	// first check whether the actor system has started
 	if !x.hasStarted.Load() {
 		return ErrActorSystemNotStarted
 	}
-
-	// based upon the event we will unsubscribe to the various topic
-	switch event {
-	case eventspb.Event_DEAD_LETTER:
-		// subscribe the consumer to the deadletter topic
-		x.eventsStream.Unsubscribe(subscriber, deadlettersTopic)
-		// remove the subscriber from the events stream
-		x.eventsStream.RemoveSubscriber(subscriber)
-	}
+	// subscribe the consumer to the deadletter topic
+	x.eventsStream.Unsubscribe(subscriber, eventsTopic)
+	// remove the subscriber from the events stream
+	x.eventsStream.RemoveSubscriber(subscriber)
 	return nil
 }
 
diff --git a/actors/constants.go b/actors/constants.go
index 9cd817c..5e81082 100644
--- a/actors/constants.go
+++ b/actors/constants.go
@@ -55,7 +55,7 @@ const (
 	DefaultInitTimeout = time.Second
 
 	defaultMailboxSize = 4096
-	deadlettersTopic   = "topic.deadletters"
+	eventsTopic        = "topic.events"
 )
 
 // NoSender means that there is no sender
diff --git a/actors/pid.go b/actors/pid.go
index 629c300..812eea2 100644
--- a/actors/pid.go
+++ b/actors/pid.go
@@ -521,6 +521,14 @@ func (p *pid) Restart(ctx context.Context) error {
 	span.SetStatus(codes.Ok, "Restart")
 	// increment the restart count
 	p.restartCount.Inc()
+	// emit actor started event
+	if p.eventsStream != nil {
+		// send the event to the event streams
+		p.eventsStream.Publish(eventsTopic, &eventspb.ActorRestarted{
+			Address:     p.ActorPath().RemoteAddress(),
+			RestartedAt: timestamppb.Now(),
+		})
+	}
 	// return
 	return nil
 }
@@ -612,6 +620,16 @@ func (p *pid) SpawnChild(ctx context.Context, name string, actor Actor) (PID, er
 
 	// set the span status and return
 	span.SetStatus(codes.Ok, "SpawnChild")
+
+	// emit actor started event
+	if p.eventsStream != nil {
+		// send the event to the event streams
+		p.eventsStream.Publish(eventsTopic, &eventspb.ActorChildCreated{
+			Address:   cid.ActorPath().RemoteAddress(),
+			CreatedAt: timestamppb.Now(),
+			Parent:    p.ActorPath().RemoteAddress(),
+		})
+	}
 	return cid, nil
 }
 
@@ -1079,6 +1097,15 @@ func (p *pid) Shutdown(ctx context.Context) error {
 		return err
 	}
 
+	// emit actor stopped event
+	if p.eventsStream != nil {
+		// send the event to the event streams
+		p.eventsStream.Publish(eventsTopic, &eventspb.ActorStopped{
+			Address:   p.ActorPath().RemoteAddress(),
+			StoppedAt: timestamppb.Now(),
+		})
+	}
+
 	p.logger.Infof("Actor=%s successfully shutdown", p.ActorPath().String())
 	span.SetStatus(codes.Ok, "Shutdown")
 	return nil
@@ -1182,6 +1209,15 @@ func (p *pid) init(ctx context.Context) error {
 	// add some logging info
 	p.logger.Info("Initialization process successfully completed.")
 	span.SetStatus(codes.Ok, "Init")
+	// emit actor started event
+	if p.eventsStream != nil {
+		// send the event to the event streams
+		p.eventsStream.Publish(eventsTopic, &eventspb.ActorStarted{
+			Address:   p.ActorPath().RemoteAddress(),
+			StartedAt: timestamppb.Now(),
+		})
+	}
+	// return safely
 	return nil
 }
 
@@ -1403,6 +1439,17 @@ func (p *pid) passivationListener() {
 		p.logger.Errorf("failed to passivate actor=(%s): reason=(%v)", p.ActorPath().String(), err)
 		return
 	}
+
+	// emit actor passivated event
+	if p.eventsStream != nil {
+		event := &eventspb.ActorPassivated{
+			Address:      p.ActorPath().RemoteAddress(),
+			PassivatedAt: timestamppb.Now(),
+		}
+		// send the event to the event streams
+		p.eventsStream.Publish(eventsTopic, event)
+	}
+
 	p.logger.Infof("Actor=%s successfully passivated", p.ActorPath().String())
 }
 
@@ -1547,7 +1594,7 @@ func (p *pid) emitDeadletter(receiveCtx ReceiveContext, err error) {
 	// define the receiver
 	receiver := p.actorPath.RemoteAddress()
 	// create the deadletter
-	deadletter := &eventspb.DeadletterEvent{
+	deadletter := &eventspb.Deadletter{
 		Sender:   senderAddr,
 		Receiver: receiver,
 		Message:  msg,
@@ -1555,7 +1602,7 @@ func (p *pid) emitDeadletter(receiveCtx ReceiveContext, err error) {
 		Reason:   err.Error(),
 	}
 	// add to the events stream
-	p.eventsStream.Publish(deadlettersTopic, deadletter)
+	p.eventsStream.Publish(eventsTopic, deadletter)
 }
 
 // registerMetrics register the PID metrics with OTel instrumentation.
diff --git a/examples/protos/pb/v1/samplepbconnect/service.connect.go b/examples/protos/pb/v1/samplepbconnect/service.connect.go
index 50fcd20..115340e 100644
--- a/examples/protos/pb/v1/samplepbconnect/service.connect.go
+++ b/examples/protos/pb/v1/samplepbconnect/service.connect.go
@@ -18,7 +18,7 @@ import (
 // generated with a version of connect newer than the one compiled into your binary. You can fix the
 // problem by either regenerating this code with an older version of connect or updating the connect
 // version compiled into your binary.
-const _ = connect.IsAtLeastVersion0_1_0
+const _ = connect.IsAtLeastVersion1_13_0
 
 const (
 	// AccountServiceName is the fully-qualified name of the AccountService service.
@@ -44,6 +44,14 @@ const (
 	AccountServiceGetAccountProcedure = "/sample.v1.AccountService/GetAccount"
 )
 
+// These variables are the protoreflect.Descriptor objects for the RPCs defined in this package.
+var (
+	accountServiceServiceDescriptor             = v1.File_pb_v1_service_proto.Services().ByName("AccountService")
+	accountServiceCreateAccountMethodDescriptor = accountServiceServiceDescriptor.Methods().ByName("CreateAccount")
+	accountServiceCreditAccountMethodDescriptor = accountServiceServiceDescriptor.Methods().ByName("CreditAccount")
+	accountServiceGetAccountMethodDescriptor    = accountServiceServiceDescriptor.Methods().ByName("GetAccount")
+)
+
 // AccountServiceClient is a client for the sample.v1.AccountService service.
 type AccountServiceClient interface {
 	CreateAccount(context.Context, *connect.Request[v1.CreateAccountRequest]) (*connect.Response[v1.CreateAccountResponse], error)
@@ -64,17 +72,20 @@ func NewAccountServiceClient(httpClient connect.HTTPClient, baseURL string, opts
 		createAccount: connect.NewClient[v1.CreateAccountRequest, v1.CreateAccountResponse](
 			httpClient,
 			baseURL+AccountServiceCreateAccountProcedure,
-			opts...,
+			connect.WithSchema(accountServiceCreateAccountMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 		creditAccount: connect.NewClient[v1.CreditAccountRequest, v1.CreditAccountResponse](
 			httpClient,
 			baseURL+AccountServiceCreditAccountProcedure,
-			opts...,
+			connect.WithSchema(accountServiceCreditAccountMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 		getAccount: connect.NewClient[v1.GetAccountRequest, v1.GetAccountResponse](
 			httpClient,
 			baseURL+AccountServiceGetAccountProcedure,
-			opts...,
+			connect.WithSchema(accountServiceGetAccountMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 	}
 }
@@ -117,17 +128,20 @@ func NewAccountServiceHandler(svc AccountServiceHandler, opts ...connect.Handler
 	accountServiceCreateAccountHandler := connect.NewUnaryHandler(
 		AccountServiceCreateAccountProcedure,
 		svc.CreateAccount,
-		opts...,
+		connect.WithSchema(accountServiceCreateAccountMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	accountServiceCreditAccountHandler := connect.NewUnaryHandler(
 		AccountServiceCreditAccountProcedure,
 		svc.CreditAccount,
-		opts...,
+		connect.WithSchema(accountServiceCreditAccountMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	accountServiceGetAccountHandler := connect.NewUnaryHandler(
 		AccountServiceGetAccountProcedure,
 		svc.GetAccount,
-		opts...,
+		connect.WithSchema(accountServiceGetAccountMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	return "/sample.v1.AccountService/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch r.URL.Path {
diff --git a/internal/v1/internalpbconnect/remoting.connect.go b/internal/v1/internalpbconnect/remoting.connect.go
index 2f74096..75ad729 100644
--- a/internal/v1/internalpbconnect/remoting.connect.go
+++ b/internal/v1/internalpbconnect/remoting.connect.go
@@ -18,7 +18,7 @@ import (
 // generated with a version of connect newer than the one compiled into your binary. You can fix the
 // problem by either regenerating this code with an older version of connect or updating the connect
 // version compiled into your binary.
-const _ = connect.IsAtLeastVersion0_1_0
+const _ = connect.IsAtLeastVersion1_13_0
 
 const (
 	// RemotingServiceName is the fully-qualified name of the RemotingService service.
@@ -50,6 +50,16 @@ const (
 	RemotingServiceRemoteBatchAskProcedure = "/internal.v1.RemotingService/RemoteBatchAsk"
 )
 
+// These variables are the protoreflect.Descriptor objects for the RPCs defined in this package.
+var (
+	remotingServiceServiceDescriptor               = v1.File_internal_v1_remoting_proto.Services().ByName("RemotingService")
+	remotingServiceRemoteAskMethodDescriptor       = remotingServiceServiceDescriptor.Methods().ByName("RemoteAsk")
+	remotingServiceRemoteTellMethodDescriptor      = remotingServiceServiceDescriptor.Methods().ByName("RemoteTell")
+	remotingServiceRemoteLookupMethodDescriptor    = remotingServiceServiceDescriptor.Methods().ByName("RemoteLookup")
+	remotingServiceRemoteBatchTellMethodDescriptor = remotingServiceServiceDescriptor.Methods().ByName("RemoteBatchTell")
+	remotingServiceRemoteBatchAskMethodDescriptor  = remotingServiceServiceDescriptor.Methods().ByName("RemoteBatchAsk")
+)
+
 // RemotingServiceClient is a client for the internal.v1.RemotingService service.
 type RemotingServiceClient interface {
 	// RemoteAsk is used to send a message to an actor remotely and expect a response immediately.
@@ -79,27 +89,32 @@ func NewRemotingServiceClient(httpClient connect.HTTPClient, baseURL string, opt
 		remoteAsk: connect.NewClient[v1.RemoteAskRequest, v1.RemoteAskResponse](
 			httpClient,
 			baseURL+RemotingServiceRemoteAskProcedure,
-			opts...,
+			connect.WithSchema(remotingServiceRemoteAskMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 		remoteTell: connect.NewClient[v1.RemoteTellRequest, v1.RemoteTellResponse](
 			httpClient,
 			baseURL+RemotingServiceRemoteTellProcedure,
-			opts...,
+			connect.WithSchema(remotingServiceRemoteTellMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 		remoteLookup: connect.NewClient[v1.RemoteLookupRequest, v1.RemoteLookupResponse](
 			httpClient,
 			baseURL+RemotingServiceRemoteLookupProcedure,
-			opts...,
+			connect.WithSchema(remotingServiceRemoteLookupMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 		remoteBatchTell: connect.NewClient[v1.RemoteBatchTellRequest, v1.RemoteBatchTellResponse](
 			httpClient,
 			baseURL+RemotingServiceRemoteBatchTellProcedure,
-			opts...,
+			connect.WithSchema(remotingServiceRemoteBatchTellMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 		remoteBatchAsk: connect.NewClient[v1.RemoteBatchAskRequest, v1.RemoteBatchAskResponse](
 			httpClient,
 			baseURL+RemotingServiceRemoteBatchAskProcedure,
-			opts...,
+			connect.WithSchema(remotingServiceRemoteBatchAskMethodDescriptor),
+			connect.WithClientOptions(opts...),
 		),
 	}
 }
@@ -163,27 +178,32 @@ func NewRemotingServiceHandler(svc RemotingServiceHandler, opts ...connect.Handl
 	remotingServiceRemoteAskHandler := connect.NewUnaryHandler(
 		RemotingServiceRemoteAskProcedure,
 		svc.RemoteAsk,
-		opts...,
+		connect.WithSchema(remotingServiceRemoteAskMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	remotingServiceRemoteTellHandler := connect.NewUnaryHandler(
 		RemotingServiceRemoteTellProcedure,
 		svc.RemoteTell,
-		opts...,
+		connect.WithSchema(remotingServiceRemoteTellMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	remotingServiceRemoteLookupHandler := connect.NewUnaryHandler(
 		RemotingServiceRemoteLookupProcedure,
 		svc.RemoteLookup,
-		opts...,
+		connect.WithSchema(remotingServiceRemoteLookupMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	remotingServiceRemoteBatchTellHandler := connect.NewUnaryHandler(
 		RemotingServiceRemoteBatchTellProcedure,
 		svc.RemoteBatchTell,
-		opts...,
+		connect.WithSchema(remotingServiceRemoteBatchTellMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	remotingServiceRemoteBatchAskHandler := connect.NewUnaryHandler(
 		RemotingServiceRemoteBatchAskProcedure,
 		svc.RemoteBatchAsk,
-		opts...,
+		connect.WithSchema(remotingServiceRemoteBatchAskMethodDescriptor),
+		connect.WithHandlerOptions(opts...),
 	)
 	return "/internal.v1.RemotingService/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch r.URL.Path {
diff --git a/pb/events/v1/events.pb.go b/pb/events/v1/events.pb.go
index f8a6834..e62427b 100644
--- a/pb/events/v1/events.pb.go
+++ b/pb/events/v1/events.pb.go
@@ -23,54 +23,8 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-type Event int32
-
-const (
-	Event_UNSPECIFIED Event = 0
-	Event_DEAD_LETTER Event = 1
-)
-
-// Enum value maps for Event.
-var (
-	Event_name = map[int32]string{
-		0: "UNSPECIFIED",
-		1: "DEAD_LETTER",
-	}
-	Event_value = map[string]int32{
-		"UNSPECIFIED": 0,
-		"DEAD_LETTER": 1,
-	}
-)
-
-func (x Event) Enum() *Event {
-	p := new(Event)
-	*p = x
-	return p
-}
-
-func (x Event) String() string {
-	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
-}
-
-func (Event) Descriptor() protoreflect.EnumDescriptor {
-	return file_events_v1_events_proto_enumTypes[0].Descriptor()
-}
-
-func (Event) Type() protoreflect.EnumType {
-	return &file_events_v1_events_proto_enumTypes[0]
-}
-
-func (x Event) Number() protoreflect.EnumNumber {
-	return protoreflect.EnumNumber(x)
-}
-
-// Deprecated: Use Event.Descriptor instead.
-func (Event) EnumDescriptor() ([]byte, []int) {
-	return file_events_v1_events_proto_rawDescGZIP(), []int{0}
-}
-
-// DeadletterEvent defines the deadletter entry
-type DeadletterEvent struct {
+// Deadletter defines the deadletter entry
+type Deadletter struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
@@ -88,8 +42,8 @@ type DeadletterEvent struct {
 	Reason string `protobuf:"bytes,5,opt,name=reason,proto3" json:"reason,omitempty"`
 }
 
-func (x *DeadletterEvent) Reset() {
-	*x = DeadletterEvent{}
+func (x *Deadletter) Reset() {
+	*x = Deadletter{}
 	if protoimpl.UnsafeEnabled {
 		mi := &file_events_v1_events_proto_msgTypes[0]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -97,13 +51,13 @@ func (x *DeadletterEvent) Reset() {
 	}
 }
 
-func (x *DeadletterEvent) String() string {
+func (x *Deadletter) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*DeadletterEvent) ProtoMessage() {}
+func (*Deadletter) ProtoMessage() {}
 
-func (x *DeadletterEvent) ProtoReflect() protoreflect.Message {
+func (x *Deadletter) ProtoReflect() protoreflect.Message {
 	mi := &file_events_v1_events_proto_msgTypes[0]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -115,46 +69,343 @@ func (x *DeadletterEvent) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use DeadletterEvent.ProtoReflect.Descriptor instead.
-func (*DeadletterEvent) Descriptor() ([]byte, []int) {
+// Deprecated: Use Deadletter.ProtoReflect.Descriptor instead.
+func (*Deadletter) Descriptor() ([]byte, []int) {
 	return file_events_v1_events_proto_rawDescGZIP(), []int{0}
 }
 
-func (x *DeadletterEvent) GetSender() *v1.Address {
+func (x *Deadletter) GetSender() *v1.Address {
 	if x != nil {
 		return x.Sender
 	}
 	return nil
 }
 
-func (x *DeadletterEvent) GetReceiver() *v1.Address {
+func (x *Deadletter) GetReceiver() *v1.Address {
 	if x != nil {
 		return x.Receiver
 	}
 	return nil
 }
 
-func (x *DeadletterEvent) GetMessage() *anypb.Any {
+func (x *Deadletter) GetMessage() *anypb.Any {
 	if x != nil {
 		return x.Message
 	}
 	return nil
 }
 
-func (x *DeadletterEvent) GetSendTime() *timestamppb.Timestamp {
+func (x *Deadletter) GetSendTime() *timestamppb.Timestamp {
 	if x != nil {
 		return x.SendTime
 	}
 	return nil
 }
 
-func (x *DeadletterEvent) GetReason() string {
+func (x *Deadletter) GetReason() string {
 	if x != nil {
 		return x.Reason
 	}
 	return ""
 }
 
+// ActorStarted defines the actor started event
+type ActorStarted struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the actor address
+	Address *v1.Address `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
+	// Specifies the started time
+	StartedAt *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=started_at,json=startedAt,proto3" json:"started_at,omitempty"`
+}
+
+func (x *ActorStarted) Reset() {
+	*x = ActorStarted{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_events_v1_events_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ActorStarted) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ActorStarted) ProtoMessage() {}
+
+func (x *ActorStarted) ProtoReflect() protoreflect.Message {
+	mi := &file_events_v1_events_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ActorStarted.ProtoReflect.Descriptor instead.
+func (*ActorStarted) Descriptor() ([]byte, []int) {
+	return file_events_v1_events_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *ActorStarted) GetAddress() *v1.Address {
+	if x != nil {
+		return x.Address
+	}
+	return nil
+}
+
+func (x *ActorStarted) GetStartedAt() *timestamppb.Timestamp {
+	if x != nil {
+		return x.StartedAt
+	}
+	return nil
+}
+
+// ActorStopped defines the actor stopped events
+type ActorStopped struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the actor address
+	Address *v1.Address `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
+	// Specifies the stop time
+	StoppedAt *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=stopped_at,json=stoppedAt,proto3" json:"stopped_at,omitempty"`
+}
+
+func (x *ActorStopped) Reset() {
+	*x = ActorStopped{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_events_v1_events_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ActorStopped) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ActorStopped) ProtoMessage() {}
+
+func (x *ActorStopped) ProtoReflect() protoreflect.Message {
+	mi := &file_events_v1_events_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ActorStopped.ProtoReflect.Descriptor instead.
+func (*ActorStopped) Descriptor() ([]byte, []int) {
+	return file_events_v1_events_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *ActorStopped) GetAddress() *v1.Address {
+	if x != nil {
+		return x.Address
+	}
+	return nil
+}
+
+func (x *ActorStopped) GetStoppedAt() *timestamppb.Timestamp {
+	if x != nil {
+		return x.StoppedAt
+	}
+	return nil
+}
+
+type ActorPassivated struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the actor address
+	Address *v1.Address `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
+	// Specifies the passivation time
+	PassivatedAt *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=passivated_at,json=passivatedAt,proto3" json:"passivated_at,omitempty"`
+}
+
+func (x *ActorPassivated) Reset() {
+	*x = ActorPassivated{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_events_v1_events_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ActorPassivated) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ActorPassivated) ProtoMessage() {}
+
+func (x *ActorPassivated) ProtoReflect() protoreflect.Message {
+	mi := &file_events_v1_events_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ActorPassivated.ProtoReflect.Descriptor instead.
+func (*ActorPassivated) Descriptor() ([]byte, []int) {
+	return file_events_v1_events_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *ActorPassivated) GetAddress() *v1.Address {
+	if x != nil {
+		return x.Address
+	}
+	return nil
+}
+
+func (x *ActorPassivated) GetPassivatedAt() *timestamppb.Timestamp {
+	if x != nil {
+		return x.PassivatedAt
+	}
+	return nil
+}
+
+type ActorChildCreated struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the actor address
+	Address *v1.Address `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
+	// Specifies the parent address
+	Parent *v1.Address `protobuf:"bytes,2,opt,name=parent,proto3" json:"parent,omitempty"`
+	// Specifies the started time
+	CreatedAt *timestamppb.Timestamp `protobuf:"bytes,3,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"`
+}
+
+func (x *ActorChildCreated) Reset() {
+	*x = ActorChildCreated{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_events_v1_events_proto_msgTypes[4]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ActorChildCreated) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ActorChildCreated) ProtoMessage() {}
+
+func (x *ActorChildCreated) ProtoReflect() protoreflect.Message {
+	mi := &file_events_v1_events_proto_msgTypes[4]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ActorChildCreated.ProtoReflect.Descriptor instead.
+func (*ActorChildCreated) Descriptor() ([]byte, []int) {
+	return file_events_v1_events_proto_rawDescGZIP(), []int{4}
+}
+
+func (x *ActorChildCreated) GetAddress() *v1.Address {
+	if x != nil {
+		return x.Address
+	}
+	return nil
+}
+
+func (x *ActorChildCreated) GetParent() *v1.Address {
+	if x != nil {
+		return x.Parent
+	}
+	return nil
+}
+
+func (x *ActorChildCreated) GetCreatedAt() *timestamppb.Timestamp {
+	if x != nil {
+		return x.CreatedAt
+	}
+	return nil
+}
+
+// ActorRestarted defines the actor restarted events
+type ActorRestarted struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the actor address
+	Address *v1.Address `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
+	// Specifies the stop time
+	RestartedAt *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=restarted_at,json=restartedAt,proto3" json:"restarted_at,omitempty"`
+}
+
+func (x *ActorRestarted) Reset() {
+	*x = ActorRestarted{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_events_v1_events_proto_msgTypes[5]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ActorRestarted) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ActorRestarted) ProtoMessage() {}
+
+func (x *ActorRestarted) ProtoReflect() protoreflect.Message {
+	mi := &file_events_v1_events_proto_msgTypes[5]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ActorRestarted.ProtoReflect.Descriptor instead.
+func (*ActorRestarted) Descriptor() ([]byte, []int) {
+	return file_events_v1_events_proto_rawDescGZIP(), []int{5}
+}
+
+func (x *ActorRestarted) GetAddress() *v1.Address {
+	if x != nil {
+		return x.Address
+	}
+	return nil
+}
+
+func (x *ActorRestarted) GetRestartedAt() *timestamppb.Timestamp {
+	if x != nil {
+		return x.RestartedAt
+	}
+	return nil
+}
+
 var File_events_v1_events_proto protoreflect.FileDescriptor
 
 var file_events_v1_events_proto_rawDesc = []byte{
@@ -165,35 +416,74 @@ var file_events_v1_events_proto_rawDesc = []byte{
 	0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x61,
 	0x6e, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
 	0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74,
-	0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xf0, 0x01, 0x0a, 0x0f, 0x44, 0x65,
-	0x61, 0x64, 0x6c, 0x65, 0x74, 0x74, 0x65, 0x72, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x2b, 0x0a,
-	0x06, 0x73, 0x65, 0x6e, 0x64, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e,
-	0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65,
-	0x73, 0x73, 0x52, 0x06, 0x73, 0x65, 0x6e, 0x64, 0x65, 0x72, 0x12, 0x2f, 0x0a, 0x08, 0x72, 0x65,
-	0x63, 0x65, 0x69, 0x76, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x61,
-	0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73,
-	0x73, 0x52, 0x08, 0x72, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65, 0x72, 0x12, 0x2e, 0x0a, 0x07, 0x6d,
-	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67,
-	0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41,
-	0x6e, 0x79, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x37, 0x0a, 0x09, 0x73,
-	0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a,
-	0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66,
-	0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x08, 0x73, 0x65, 0x6e, 0x64,
-	0x54, 0x69, 0x6d, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e, 0x18, 0x05,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e, 0x2a, 0x29, 0x0a, 0x05,
-	0x45, 0x76, 0x65, 0x6e, 0x74, 0x12, 0x0f, 0x0a, 0x0b, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49,
-	0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0f, 0x0a, 0x0b, 0x44, 0x45, 0x41, 0x44, 0x5f, 0x4c,
-	0x45, 0x54, 0x54, 0x45, 0x52, 0x10, 0x01, 0x42, 0x94, 0x01, 0x0a, 0x0d, 0x63, 0x6f, 0x6d, 0x2e,
-	0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e, 0x76, 0x31, 0x42, 0x0b, 0x45, 0x76, 0x65, 0x6e, 0x74,
-	0x73, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x48, 0x02, 0x50, 0x01, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68,
-	0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x74, 0x6f, 0x63, 0x68, 0x65, 0x6d, 0x65, 0x79, 0x2f,
-	0x67, 0x6f, 0x61, 0x6b, 0x74, 0x2f, 0x70, 0x62, 0x2f, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2f,
-	0x76, 0x31, 0x3b, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x70, 0x62, 0xa2, 0x02, 0x03, 0x45, 0x58,
-	0x58, 0xaa, 0x02, 0x09, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e, 0x56, 0x31, 0xca, 0x02, 0x09,
-	0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x5c, 0x56, 0x31, 0xe2, 0x02, 0x15, 0x45, 0x76, 0x65, 0x6e,
-	0x74, 0x73, 0x5c, 0x56, 0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74,
-	0x61, 0xea, 0x02, 0x0a, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x3a, 0x3a, 0x56, 0x31, 0x62, 0x06,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xeb, 0x01, 0x0a, 0x0a, 0x44, 0x65,
+	0x61, 0x64, 0x6c, 0x65, 0x74, 0x74, 0x65, 0x72, 0x12, 0x2b, 0x0a, 0x06, 0x73, 0x65, 0x6e, 0x64,
+	0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65,
+	0x73, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x52, 0x06, 0x73,
+	0x65, 0x6e, 0x64, 0x65, 0x72, 0x12, 0x2f, 0x0a, 0x08, 0x72, 0x65, 0x63, 0x65, 0x69, 0x76, 0x65,
+	0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73,
+	0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x52, 0x08, 0x72, 0x65,
+	0x63, 0x65, 0x69, 0x76, 0x65, 0x72, 0x12, 0x2e, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67,
+	0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x41, 0x6e, 0x79, 0x52, 0x07, 0x6d,
+	0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x37, 0x0a, 0x09, 0x73, 0x65, 0x6e, 0x64, 0x5f, 0x74,
+	0x69, 0x6d, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67,
+	0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65,
+	0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x08, 0x73, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x12,
+	0x16, 0x0a, 0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x06, 0x72, 0x65, 0x61, 0x73, 0x6f, 0x6e, 0x22, 0x78, 0x0a, 0x0c, 0x41, 0x63, 0x74, 0x6f, 0x72,
+	0x53, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x12, 0x2d, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65,
+	0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65,
+	0x73, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x61,
+	0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x39, 0x0a, 0x0a, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65,
+	0x64, 0x5f, 0x61, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f,
+	0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d,
+	0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x41,
+	0x74, 0x22, 0x78, 0x0a, 0x0c, 0x41, 0x63, 0x74, 0x6f, 0x72, 0x53, 0x74, 0x6f, 0x70, 0x70, 0x65,
+	0x64, 0x12, 0x2d, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01,
+	0x28, 0x0b, 0x32, 0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2e, 0x76, 0x31, 0x2e,
+	0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73,
+	0x12, 0x39, 0x0a, 0x0a, 0x73, 0x74, 0x6f, 0x70, 0x70, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18, 0x02,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
+	0x52, 0x09, 0x73, 0x74, 0x6f, 0x70, 0x70, 0x65, 0x64, 0x41, 0x74, 0x22, 0x81, 0x01, 0x0a, 0x0f,
+	0x41, 0x63, 0x74, 0x6f, 0x72, 0x50, 0x61, 0x73, 0x73, 0x69, 0x76, 0x61, 0x74, 0x65, 0x64, 0x12,
+	0x2d, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64,
+	0x64, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x3f,
+	0x0a, 0x0d, 0x70, 0x61, 0x73, 0x73, 0x69, 0x76, 0x61, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
+	0x70, 0x52, 0x0c, 0x70, 0x61, 0x73, 0x73, 0x69, 0x76, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74, 0x22,
+	0xaa, 0x01, 0x0a, 0x11, 0x41, 0x63, 0x74, 0x6f, 0x72, 0x43, 0x68, 0x69, 0x6c, 0x64, 0x43, 0x72,
+	0x65, 0x61, 0x74, 0x65, 0x64, 0x12, 0x2d, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73,
+	0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x61, 0x64, 0x64,
+	0x72, 0x65, 0x73, 0x73, 0x12, 0x2b, 0x0a, 0x06, 0x70, 0x61, 0x72, 0x65, 0x6e, 0x74, 0x18, 0x02,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2e, 0x76,
+	0x31, 0x2e, 0x41, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x52, 0x06, 0x70, 0x61, 0x72, 0x65, 0x6e,
+	0x74, 0x12, 0x39, 0x0a, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
+	0x70, 0x52, 0x09, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74, 0x22, 0x7e, 0x0a, 0x0e,
+	0x41, 0x63, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x12, 0x2d,
+	0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32,
+	0x13, 0x2e, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x41, 0x64, 0x64,
+	0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, 0x3d, 0x0a,
+	0x0c, 0x72, 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52,
+	0x0b, 0x72, 0x65, 0x73, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64, 0x41, 0x74, 0x42, 0x94, 0x01, 0x0a,
+	0x0d, 0x63, 0x6f, 0x6d, 0x2e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e, 0x76, 0x31, 0x42, 0x0b,
+	0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x48, 0x02, 0x50, 0x01, 0x5a,
+	0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x74, 0x6f, 0x63, 0x68,
+	0x65, 0x6d, 0x65, 0x79, 0x2f, 0x67, 0x6f, 0x61, 0x6b, 0x74, 0x2f, 0x70, 0x62, 0x2f, 0x65, 0x76,
+	0x65, 0x6e, 0x74, 0x73, 0x2f, 0x76, 0x31, 0x3b, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x70, 0x62,
+	0xa2, 0x02, 0x03, 0x45, 0x58, 0x58, 0xaa, 0x02, 0x09, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x2e,
+	0x56, 0x31, 0xca, 0x02, 0x09, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x5c, 0x56, 0x31, 0xe2, 0x02,
+	0x15, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x5c, 0x56, 0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65,
+	0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x0a, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x3a,
+	0x3a, 0x56, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -208,25 +498,39 @@ func file_events_v1_events_proto_rawDescGZIP() []byte {
 	return file_events_v1_events_proto_rawDescData
 }
 
-var file_events_v1_events_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_events_v1_events_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_events_v1_events_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
 var file_events_v1_events_proto_goTypes = []interface{}{
-	(Event)(0),                    // 0: events.v1.Event
-	(*DeadletterEvent)(nil),       // 1: events.v1.DeadletterEvent
-	(*v1.Address)(nil),            // 2: address.v1.Address
-	(*anypb.Any)(nil),             // 3: google.protobuf.Any
-	(*timestamppb.Timestamp)(nil), // 4: google.protobuf.Timestamp
+	(*Deadletter)(nil),            // 0: events.v1.Deadletter
+	(*ActorStarted)(nil),          // 1: events.v1.ActorStarted
+	(*ActorStopped)(nil),          // 2: events.v1.ActorStopped
+	(*ActorPassivated)(nil),       // 3: events.v1.ActorPassivated
+	(*ActorChildCreated)(nil),     // 4: events.v1.ActorChildCreated
+	(*ActorRestarted)(nil),        // 5: events.v1.ActorRestarted
+	(*v1.Address)(nil),            // 6: address.v1.Address
+	(*anypb.Any)(nil),             // 7: google.protobuf.Any
+	(*timestamppb.Timestamp)(nil), // 8: google.protobuf.Timestamp
 }
 var file_events_v1_events_proto_depIdxs = []int32{
-	2, // 0: events.v1.DeadletterEvent.sender:type_name -> address.v1.Address
-	2, // 1: events.v1.DeadletterEvent.receiver:type_name -> address.v1.Address
-	3, // 2: events.v1.DeadletterEvent.message:type_name -> google.protobuf.Any
-	4, // 3: events.v1.DeadletterEvent.send_time:type_name -> google.protobuf.Timestamp
-	4, // [4:4] is the sub-list for method output_type
-	4, // [4:4] is the sub-list for method input_type
-	4, // [4:4] is the sub-list for extension type_name
-	4, // [4:4] is the sub-list for extension extendee
-	0, // [0:4] is the sub-list for field type_name
+	6,  // 0: events.v1.Deadletter.sender:type_name -> address.v1.Address
+	6,  // 1: events.v1.Deadletter.receiver:type_name -> address.v1.Address
+	7,  // 2: events.v1.Deadletter.message:type_name -> google.protobuf.Any
+	8,  // 3: events.v1.Deadletter.send_time:type_name -> google.protobuf.Timestamp
+	6,  // 4: events.v1.ActorStarted.address:type_name -> address.v1.Address
+	8,  // 5: events.v1.ActorStarted.started_at:type_name -> google.protobuf.Timestamp
+	6,  // 6: events.v1.ActorStopped.address:type_name -> address.v1.Address
+	8,  // 7: events.v1.ActorStopped.stopped_at:type_name -> google.protobuf.Timestamp
+	6,  // 8: events.v1.ActorPassivated.address:type_name -> address.v1.Address
+	8,  // 9: events.v1.ActorPassivated.passivated_at:type_name -> google.protobuf.Timestamp
+	6,  // 10: events.v1.ActorChildCreated.address:type_name -> address.v1.Address
+	6,  // 11: events.v1.ActorChildCreated.parent:type_name -> address.v1.Address
+	8,  // 12: events.v1.ActorChildCreated.created_at:type_name -> google.protobuf.Timestamp
+	6,  // 13: events.v1.ActorRestarted.address:type_name -> address.v1.Address
+	8,  // 14: events.v1.ActorRestarted.restarted_at:type_name -> google.protobuf.Timestamp
+	15, // [15:15] is the sub-list for method output_type
+	15, // [15:15] is the sub-list for method input_type
+	15, // [15:15] is the sub-list for extension type_name
+	15, // [15:15] is the sub-list for extension extendee
+	0,  // [0:15] is the sub-list for field type_name
 }
 
 func init() { file_events_v1_events_proto_init() }
@@ -236,7 +540,67 @@ func file_events_v1_events_proto_init() {
 	}
 	if !protoimpl.UnsafeEnabled {
 		file_events_v1_events_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DeadletterEvent); i {
+			switch v := v.(*Deadletter); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_events_v1_events_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ActorStarted); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_events_v1_events_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ActorStopped); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_events_v1_events_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ActorPassivated); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_events_v1_events_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ActorChildCreated); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_events_v1_events_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ActorRestarted); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -253,14 +617,13 @@ func file_events_v1_events_proto_init() {
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_events_v1_events_proto_rawDesc,
-			NumEnums:      1,
-			NumMessages:   1,
+			NumEnums:      0,
+			NumMessages:   6,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
 		GoTypes:           file_events_v1_events_proto_goTypes,
 		DependencyIndexes: file_events_v1_events_proto_depIdxs,
-		EnumInfos:         file_events_v1_events_proto_enumTypes,
 		MessageInfos:      file_events_v1_events_proto_msgTypes,
 	}.Build()
 	File_events_v1_events_proto = out.File
diff --git a/protos/goakt/events/v1/events.proto b/protos/goakt/events/v1/events.proto
index 26c37ec..ffad2ed 100644
--- a/protos/goakt/events/v1/events.proto
+++ b/protos/goakt/events/v1/events.proto
@@ -8,8 +8,8 @@ import "google/protobuf/timestamp.proto";
 
 option go_package = "github.com/tochemey/goakt/pb/events/v1;eventspb";
 
-// DeadletterEvent defines the deadletter entry
-message DeadletterEvent {
+// Deadletter defines the deadletter entry
+message Deadletter {
   // Specifies the sender's address
   address.v1.Address sender = 1;
   // Specifies the actor address
@@ -23,7 +23,42 @@ message DeadletterEvent {
   string reason = 5;
 }
 
-enum Event {
-  UNSPECIFIED = 0;
-  DEAD_LETTER = 1;
+// ActorStarted defines the actor started event
+message ActorStarted {
+  // Specifies the actor address
+  address.v1.Address address = 1;
+  // Specifies the started time
+  google.protobuf.Timestamp started_at = 2;
+}
+
+// ActorStopped defines the actor stopped events
+message ActorStopped {
+  // Specifies the actor address
+  address.v1.Address address = 1;
+  // Specifies the stop time
+  google.protobuf.Timestamp stopped_at = 2;
+}
+
+message ActorPassivated {
+  // Specifies the actor address
+  address.v1.Address address = 1;
+  // Specifies the passivation time
+  google.protobuf.Timestamp passivated_at = 2;
+}
+
+message ActorChildCreated {
+  // Specifies the actor address
+  address.v1.Address address = 1;
+  // Specifies the parent address
+  address.v1.Address parent = 2;
+  // Specifies the started time
+  google.protobuf.Timestamp created_at = 3;
+}
+
+// ActorRestarted defines the actor restarted events
+message ActorRestarted {
+  // Specifies the actor address
+  address.v1.Address address = 1;
+  // Specifies the stop time
+  google.protobuf.Timestamp restarted_at = 2;
 }


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index 4cb9565..c7e2423 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -874,7 +874,7 @@ func TestActorSystem(t *testing.T) {
 		time.Sleep(time.Second)
 
 		// create a deadletter subscriber
-		consumer, err := sys.Subscribe(eventspb.Event_DEAD_LETTER)
+		consumer, err := sys.Subscribe()
 		require.NoError(t, err)
 		require.NotNil(t, consumer)
 
@@ -894,17 +894,20 @@ func TestActorSystem(t *testing.T) {
 
 		time.Sleep(time.Second)
 
-		var items []*eventspb.DeadletterEvent
+		var items []*eventspb.Deadletter
 		for message := range consumer.Iterator() {
 			payload := message.Payload()
-			deadletter := payload.(*eventspb.DeadletterEvent)
-			items = append(items, deadletter)
+			// only listening to deadletters
+			deadletter, ok := payload.(*eventspb.Deadletter)
+			if ok {
+				items = append(items, deadletter)
+			}
 		}
 
 		require.Len(t, items, 5)
 
 		// unsubscribe the consumer
-		err = sys.Unsubscribe(eventspb.Event_DEAD_LETTER, consumer)
+		err = sys.Unsubscribe(consumer)
 		require.NoError(t, err)
 
 		t.Cleanup(func() {
@@ -916,7 +919,7 @@ func TestActorSystem(t *testing.T) {
 		sys, _ := NewActorSystem("testSys", WithLogger(log.DiscardLogger))
 
 		// create a deadletter subscriber
-		consumer, err := sys.Subscribe(eventspb.Event_DEAD_LETTER)
+		consumer, err := sys.Subscribe()
 		require.Error(t, err)
 		require.Nil(t, consumer)
 	})
@@ -928,7 +931,7 @@ func TestActorSystem(t *testing.T) {
 		err := sys.Start(ctx)
 		assert.NoError(t, err)
 
-		consumer, err := sys.Subscribe(eventspb.Event_DEAD_LETTER)
+		consumer, err := sys.Subscribe()
 		require.NoError(t, err)
 		require.NotNil(t, consumer)
 
@@ -938,7 +941,7 @@ func TestActorSystem(t *testing.T) {
 		time.Sleep(time.Second)
 
 		// create a deadletter subscriber
-		err = sys.Unsubscribe(eventspb.Event_DEAD_LETTER, consumer)
+		err = sys.Unsubscribe(consumer)
 		require.Error(t, err)
 	})
 	t.Run("With Passivation with clustering enabled", func(t *testing.T) {
diff --git a/actors/context_test.go b/actors/context_test.go
index c441d91..5e2d9f7 100644
--- a/actors/context_test.go
+++ b/actors/context_test.go
@@ -1158,7 +1158,7 @@ func TestReceiveContext(t *testing.T) {
 
 		// create a consumer
 		consumer := eventsStream.AddSubscriber()
-		eventsStream.Subscribe(consumer, deadlettersTopic)
+		eventsStream.Subscribe(consumer, eventsTopic)
 
 		// create a Ping actor
 		opts := []pidOption{
@@ -1192,12 +1192,14 @@ func TestReceiveContext(t *testing.T) {
 		// wait for messages to be published
 		time.Sleep(time.Second)
 
-		var items []*eventspb.DeadletterEvent
+		var items []*eventspb.Deadletter
 		for message := range consumer.Iterator() {
 			payload := message.Payload()
-			assert.Equal(t, deadlettersTopic, message.Topic())
-			deadletter := payload.(*eventspb.DeadletterEvent)
-			items = append(items, deadletter)
+			assert.Equal(t, eventsTopic, message.Topic())
+			deadletter, ok := payload.(*eventspb.Deadletter)
+			if ok {
+				items = append(items, deadletter)
+			}
 		}
 
 		require.Len(t, items, 1)
@@ -1224,7 +1226,7 @@ func TestReceiveContext(t *testing.T) {
 
 		// create a consumer
 		consumer := eventsStream.AddSubscriber()
-		eventsStream.Subscribe(consumer, deadlettersTopic)
+		eventsStream.Subscribe(consumer, eventsTopic)
 
 		// create a Ping actor
 		opts := []pidOption{
@@ -1262,11 +1264,13 @@ func TestReceiveContext(t *testing.T) {
 		// wait for messages to be published
 		time.Sleep(time.Second)
 
-		var items []*eventspb.DeadletterEvent
+		var items []*eventspb.Deadletter
 		for message := range consumer.Iterator() {
 			payload := message.Payload()
-			deadletter := payload.(*eventspb.DeadletterEvent)
-			items = append(items, deadletter)
+			deadletter, ok := payload.(*eventspb.Deadletter)
+			if ok {
+				items = append(items, deadletter)
+			}
 		}
 
 		require.Len(t, items, 1)
diff --git a/testkit/cluster/interface.go b/testkit/cluster/interface.go
index e25be35..71e2d49 100644
--- a/testkit/cluster/interface.go
+++ b/testkit/cluster/interface.go
@@ -120,6 +120,59 @@ func (_c *Interface_GetPartition_Call) RunAndReturn(run func(string) int) *Inter
 	return _c
 }
 
+// KeyExists provides a mock function with given fields: ctx, key
+func (_m *Interface) KeyExists(ctx context.Context, key string) (bool, error) {
+	ret := _m.Called(ctx, key)
+
+	var r0 bool
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, string) (bool, error)); ok {
+		return rf(ctx, key)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, string) bool); ok {
+		r0 = rf(ctx, key)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, key)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Interface_KeyExists_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'KeyExists'
+type Interface_KeyExists_Call struct {
+	*mock.Call
+}
+
+// KeyExists is a helper method to define mock.On call
+//   - ctx context.Context
+//   - key string
+func (_e *Interface_Expecter) KeyExists(ctx interface{}, key interface{}) *Interface_KeyExists_Call {
+	return &Interface_KeyExists_Call{Call: _e.mock.On("KeyExists", ctx, key)}
+}
+
+func (_c *Interface_KeyExists_Call) Run(run func(ctx context.Context, key string)) *Interface_KeyExists_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(string))
+	})
+	return _c
+}
+
+func (_c *Interface_KeyExists_Call) Return(_a0 bool, _a1 error) *Interface_KeyExists_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *Interface_KeyExists_Call) RunAndReturn(run func(context.Context, string) (bool, error)) *Interface_KeyExists_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // NodeHost provides a mock function with given fields:
 func (_m *Interface) NodeHost() string {
 	ret := _m.Called()
@@ -245,6 +298,92 @@ func (_c *Interface_PutActor_Call) RunAndReturn(run func(context.Context, *inter
 	return _c
 }
 
+// RemoveActor provides a mock function with given fields: ctx, actorName
+func (_m *Interface) RemoveActor(ctx context.Context, actorName string) error {
+	ret := _m.Called(ctx, actorName)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, string) error); ok {
+		r0 = rf(ctx, actorName)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Interface_RemoveActor_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RemoveActor'
+type Interface_RemoveActor_Call struct {
+	*mock.Call
+}
+
+// RemoveActor is a helper method to define mock.On call
+//   - ctx context.Context
+//   - actorName string
+func (_e *Interface_Expecter) RemoveActor(ctx interface{}, actorName interface{}) *Interface_RemoveActor_Call {
+	return &Interface_RemoveActor_Call{Call: _e.mock.On("RemoveActor", ctx, actorName)}
+}
+
+func (_c *Interface_RemoveActor_Call) Run(run func(ctx context.Context, actorName string)) *Interface_RemoveActor_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(string))
+	})
+	return _c
+}
+
+func (_c *Interface_RemoveActor_Call) Return(_a0 error) *Interface_RemoveActor_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *Interface_RemoveActor_Call) RunAndReturn(run func(context.Context, string) error) *Interface_RemoveActor_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// SetKey provides a mock function with given fields: ctx, key
+func (_m *Interface) SetKey(ctx context.Context, key string) error {
+	ret := _m.Called(ctx, key)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, string) error); ok {
+		r0 = rf(ctx, key)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Interface_SetKey_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetKey'
+type Interface_SetKey_Call struct {
+	*mock.Call
+}
+
+// SetKey is a helper method to define mock.On call
+//   - ctx context.Context
+//   - key string
+func (_e *Interface_Expecter) SetKey(ctx interface{}, key interface{}) *Interface_SetKey_Call {
+	return &Interface_SetKey_Call{Call: _e.mock.On("SetKey", ctx, key)}
+}
+
+func (_c *Interface_SetKey_Call) Run(run func(ctx context.Context, key string)) *Interface_SetKey_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(string))
+	})
+	return _c
+}
+
+func (_c *Interface_SetKey_Call) Return(_a0 error) *Interface_SetKey_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *Interface_SetKey_Call) RunAndReturn(run func(context.Context, string) error) *Interface_SetKey_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Start provides a mock function with given fields: ctx
 func (_m *Interface) Start(ctx context.Context) error {
 	ret := _m.Called(ctx)
diff --git a/testkit/discovery/nats/option.go b/testkit/discovery/nats/option.go
new file mode 100644
index 0000000..c9dfd87
--- /dev/null
+++ b/testkit/discovery/nats/option.go
@@ -0,0 +1,68 @@
+// Code generated by mockery v2.32.4. DO NOT EDIT.
+
+package nats
+
+import (
+	mock "github.com/stretchr/testify/mock"
+	discoverynats "github.com/tochemey/goakt/discovery/nats"
+)
+
+// Option is an autogenerated mock type for the Option type
+type Option struct {
+	mock.Mock
+}
+
+type Option_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *Option) EXPECT() *Option_Expecter {
+	return &Option_Expecter{mock: &_m.Mock}
+}
+
+// Apply provides a mock function with given fields: disco
+func (_m *Option) Apply(disco *discoverynats.Discovery) {
+	_m.Called(disco)
+}
+
+// Option_Apply_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Apply'
+type Option_Apply_Call struct {
+	*mock.Call
+}
+
+// Apply is a helper method to define mock.On call
+//   - disco *discoverynats.Discovery
+func (_e *Option_Expecter) Apply(disco interface{}) *Option_Apply_Call {
+	return &Option_Apply_Call{Call: _e.mock.On("Apply", disco)}
+}
+
+func (_c *Option_Apply_Call) Run(run func(disco *discoverynats.Discovery)) *Option_Apply_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(*discoverynats.Discovery))
+	})
+	return _c
+}
+
+func (_c *Option_Apply_Call) Return() *Option_Apply_Call {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *Option_Apply_Call) RunAndReturn(run func(*discoverynats.Discovery)) *Option_Apply_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// NewOption creates a new instance of Option. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewOption(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *Option {
+	mock := &Option{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/testkit/discovery/nats/option_func.go b/testkit/discovery/nats/option_func.go
new file mode 100644
index 0000000..6325a7d
--- /dev/null
+++ b/testkit/discovery/nats/option_func.go
@@ -0,0 +1,68 @@
+// Code generated by mockery v2.32.4. DO NOT EDIT.
+
+package nats
+
+import (
+	mock "github.com/stretchr/testify/mock"
+	discoverynats "github.com/tochemey/goakt/discovery/nats"
+)
+
+// OptionFunc is an autogenerated mock type for the OptionFunc type
+type OptionFunc struct {
+	mock.Mock
+}
+
+type OptionFunc_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *OptionFunc) EXPECT() *OptionFunc_Expecter {
+	return &OptionFunc_Expecter{mock: &_m.Mock}
+}
+
+// Execute provides a mock function with given fields: disco
+func (_m *OptionFunc) Execute(disco *discoverynats.Discovery) {
+	_m.Called(disco)
+}
+
+// OptionFunc_Execute_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Execute'
+type OptionFunc_Execute_Call struct {
+	*mock.Call
+}
+
+// Execute is a helper method to define mock.On call
+//   - disco *discoverynats.Discovery
+func (_e *OptionFunc_Expecter) Execute(disco interface{}) *OptionFunc_Execute_Call {
+	return &OptionFunc_Execute_Call{Call: _e.mock.On("Execute", disco)}
+}
+
+func (_c *OptionFunc_Execute_Call) Run(run func(disco *discoverynats.Discovery)) *OptionFunc_Execute_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(*discoverynats.Discovery))
+	})
+	return _c
+}
+
+func (_c *OptionFunc_Execute_Call) Return() *OptionFunc_Execute_Call {
+	_c.Call.Return()
+	return _c
+}
+
+func (_c *OptionFunc_Execute_Call) RunAndReturn(run func(*discoverynats.Discovery)) *OptionFunc_Execute_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// NewOptionFunc creates a new instance of OptionFunc. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewOptionFunc(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *OptionFunc {
+	mock := &OptionFunc{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/testkit/internal/v1/internalpbconnect/remoting_service_client.go b/testkit/internal/v1/internalpbconnect/remoting_service_client.go
index 0108102..3ff43cb 100644
--- a/testkit/internal/v1/internalpbconnect/remoting_service_client.go
+++ b/testkit/internal/v1/internalpbconnect/remoting_service_client.go
@@ -80,6 +80,116 @@ func (_c *RemotingServiceClient_RemoteAsk_Call) RunAndReturn(run func(context.Co
 	return _c
 }
 
+// RemoteBatchAsk provides a mock function with given fields: _a0, _a1
+func (_m *RemotingServiceClient) RemoteBatchAsk(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchAskRequest]) (*connect.Response[internalpb.RemoteBatchAskResponse], error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *connect.Response[internalpb.RemoteBatchAskResponse]
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) (*connect.Response[internalpb.RemoteBatchAskResponse], error)); ok {
+		return rf(_a0, _a1)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) *connect.Response[internalpb.RemoteBatchAskResponse]); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*connect.Response[internalpb.RemoteBatchAskResponse])
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// RemotingServiceClient_RemoteBatchAsk_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RemoteBatchAsk'
+type RemotingServiceClient_RemoteBatchAsk_Call struct {
+	*mock.Call
+}
+
+// RemoteBatchAsk is a helper method to define mock.On call
+//   - _a0 context.Context
+//   - _a1 *connect.Request[internalpb.RemoteBatchAskRequest]
+func (_e *RemotingServiceClient_Expecter) RemoteBatchAsk(_a0 interface{}, _a1 interface{}) *RemotingServiceClient_RemoteBatchAsk_Call {
+	return &RemotingServiceClient_RemoteBatchAsk_Call{Call: _e.mock.On("RemoteBatchAsk", _a0, _a1)}
+}
+
+func (_c *RemotingServiceClient_RemoteBatchAsk_Call) Run(run func(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchAskRequest])) *RemotingServiceClient_RemoteBatchAsk_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*connect.Request[internalpb.RemoteBatchAskRequest]))
+	})
+	return _c
+}
+
+func (_c *RemotingServiceClient_RemoteBatchAsk_Call) Return(_a0 *connect.Response[internalpb.RemoteBatchAskResponse], _a1 error) *RemotingServiceClient_RemoteBatchAsk_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *RemotingServiceClient_RemoteBatchAsk_Call) RunAndReturn(run func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) (*connect.Response[internalpb.RemoteBatchAskResponse], error)) *RemotingServiceClient_RemoteBatchAsk_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// RemoteBatchTell provides a mock function with given fields: _a0, _a1
+func (_m *RemotingServiceClient) RemoteBatchTell(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchTellRequest]) (*connect.Response[internalpb.RemoteBatchTellResponse], error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *connect.Response[internalpb.RemoteBatchTellResponse]
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) (*connect.Response[internalpb.RemoteBatchTellResponse], error)); ok {
+		return rf(_a0, _a1)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) *connect.Response[internalpb.RemoteBatchTellResponse]); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*connect.Response[internalpb.RemoteBatchTellResponse])
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// RemotingServiceClient_RemoteBatchTell_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RemoteBatchTell'
+type RemotingServiceClient_RemoteBatchTell_Call struct {
+	*mock.Call
+}
+
+// RemoteBatchTell is a helper method to define mock.On call
+//   - _a0 context.Context
+//   - _a1 *connect.Request[internalpb.RemoteBatchTellRequest]
+func (_e *RemotingServiceClient_Expecter) RemoteBatchTell(_a0 interface{}, _a1 interface{}) *RemotingServiceClient_RemoteBatchTell_Call {
+	return &RemotingServiceClient_RemoteBatchTell_Call{Call: _e.mock.On("RemoteBatchTell", _a0, _a1)}
+}
+
+func (_c *RemotingServiceClient_RemoteBatchTell_Call) Run(run func(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchTellRequest])) *RemotingServiceClient_RemoteBatchTell_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*connect.Request[internalpb.RemoteBatchTellRequest]))
+	})
+	return _c
+}
+
+func (_c *RemotingServiceClient_RemoteBatchTell_Call) Return(_a0 *connect.Response[internalpb.RemoteBatchTellResponse], _a1 error) *RemotingServiceClient_RemoteBatchTell_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *RemotingServiceClient_RemoteBatchTell_Call) RunAndReturn(run func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) (*connect.Response[internalpb.RemoteBatchTellResponse], error)) *RemotingServiceClient_RemoteBatchTell_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // RemoteLookup provides a mock function with given fields: _a0, _a1
 func (_m *RemotingServiceClient) RemoteLookup(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteLookupRequest]) (*connect.Response[internalpb.RemoteLookupResponse], error) {
 	ret := _m.Called(_a0, _a1)
diff --git a/testkit/internal/v1/internalpbconnect/remoting_service_handler.go b/testkit/internal/v1/internalpbconnect/remoting_service_handler.go
index 26cb5fa..6e8dd57 100644
--- a/testkit/internal/v1/internalpbconnect/remoting_service_handler.go
+++ b/testkit/internal/v1/internalpbconnect/remoting_service_handler.go
@@ -80,6 +80,116 @@ func (_c *RemotingServiceHandler_RemoteAsk_Call) RunAndReturn(run func(context.C
 	return _c
 }
 
+// RemoteBatchAsk provides a mock function with given fields: _a0, _a1
+func (_m *RemotingServiceHandler) RemoteBatchAsk(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchAskRequest]) (*connect.Response[internalpb.RemoteBatchAskResponse], error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *connect.Response[internalpb.RemoteBatchAskResponse]
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) (*connect.Response[internalpb.RemoteBatchAskResponse], error)); ok {
+		return rf(_a0, _a1)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) *connect.Response[internalpb.RemoteBatchAskResponse]); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*connect.Response[internalpb.RemoteBatchAskResponse])
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// RemotingServiceHandler_RemoteBatchAsk_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RemoteBatchAsk'
+type RemotingServiceHandler_RemoteBatchAsk_Call struct {
+	*mock.Call
+}
+
+// RemoteBatchAsk is a helper method to define mock.On call
+//   - _a0 context.Context
+//   - _a1 *connect.Request[internalpb.RemoteBatchAskRequest]
+func (_e *RemotingServiceHandler_Expecter) RemoteBatchAsk(_a0 interface{}, _a1 interface{}) *RemotingServiceHandler_RemoteBatchAsk_Call {
+	return &RemotingServiceHandler_RemoteBatchAsk_Call{Call: _e.mock.On("RemoteBatchAsk", _a0, _a1)}
+}
+
+func (_c *RemotingServiceHandler_RemoteBatchAsk_Call) Run(run func(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchAskRequest])) *RemotingServiceHandler_RemoteBatchAsk_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*connect.Request[internalpb.RemoteBatchAskRequest]))
+	})
+	return _c
+}
+
+func (_c *RemotingServiceHandler_RemoteBatchAsk_Call) Return(_a0 *connect.Response[internalpb.RemoteBatchAskResponse], _a1 error) *RemotingServiceHandler_RemoteBatchAsk_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *RemotingServiceHandler_RemoteBatchAsk_Call) RunAndReturn(run func(context.Context, *connect.Request[internalpb.RemoteBatchAskRequest]) (*connect.Response[internalpb.RemoteBatchAskResponse], error)) *RemotingServiceHandler_RemoteBatchAsk_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// RemoteBatchTell provides a mock function with given fields: _a0, _a1
+func (_m *RemotingServiceHandler) RemoteBatchTell(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchTellRequest]) (*connect.Response[internalpb.RemoteBatchTellResponse], error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 *connect.Response[internalpb.RemoteBatchTellResponse]
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) (*connect.Response[internalpb.RemoteBatchTellResponse], error)); ok {
+		return rf(_a0, _a1)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) *connect.Response[internalpb.RemoteBatchTellResponse]); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*connect.Response[internalpb.RemoteBatchTellResponse])
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// RemotingServiceHandler_RemoteBatchTell_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RemoteBatchTell'
+type RemotingServiceHandler_RemoteBatchTell_Call struct {
+	*mock.Call
+}
+
+// RemoteBatchTell is a helper method to define mock.On call
+//   - _a0 context.Context
+//   - _a1 *connect.Request[internalpb.RemoteBatchTellRequest]
+func (_e *RemotingServiceHandler_Expecter) RemoteBatchTell(_a0 interface{}, _a1 interface{}) *RemotingServiceHandler_RemoteBatchTell_Call {
+	return &RemotingServiceHandler_RemoteBatchTell_Call{Call: _e.mock.On("RemoteBatchTell", _a0, _a1)}
+}
+
+func (_c *RemotingServiceHandler_RemoteBatchTell_Call) Run(run func(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteBatchTellRequest])) *RemotingServiceHandler_RemoteBatchTell_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context), args[1].(*connect.Request[internalpb.RemoteBatchTellRequest]))
+	})
+	return _c
+}
+
+func (_c *RemotingServiceHandler_RemoteBatchTell_Call) Return(_a0 *connect.Response[internalpb.RemoteBatchTellResponse], _a1 error) *RemotingServiceHandler_RemoteBatchTell_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *RemotingServiceHandler_RemoteBatchTell_Call) RunAndReturn(run func(context.Context, *connect.Request[internalpb.RemoteBatchTellRequest]) (*connect.Response[internalpb.RemoteBatchTellResponse], error)) *RemotingServiceHandler_RemoteBatchTell_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // RemoteLookup provides a mock function with given fields: _a0, _a1
 func (_m *RemotingServiceHandler) RemoteLookup(_a0 context.Context, _a1 *connect.Request[internalpb.RemoteLookupRequest]) (*connect.Response[internalpb.RemoteLookupResponse], error) {
 	ret := _m.Called(_a0, _a1)