diff --git a/internal/pubsub/nats.go b/internal/pubsub/nats.go
index ed3e2ec..22e4cd0 100644
--- a/internal/pubsub/nats.go
+++ b/internal/pubsub/nats.go
@@ -51,7 +51,7 @@ func (s *subscriber) SubscribeChanges(ctx context.Context, topic string) (<-chan
 
 	opts := s.subscriptionOptions
 
-	opts = append(opts, nats.Durable(s.durableName(subject)))
+	opts = append(opts, nats.Durable(s.durableName(subject)), nats.AckExplicit(), nats.ManualAck())
 
 	msgCh := make(chan *changeEvent, subscriptionBufferSize)