diff --git a/kvrocks.conf b/kvrocks.conf
index d13a13cd120..3f85b1f8282 100644
--- a/kvrocks.conf
+++ b/kvrocks.conf
@@ -610,6 +610,23 @@ lua-strict-key-accessing no
 #
 # tls-replication yes
 
+############################# KEYSPACE NOTIFICATIONS ##########################
+
+# Keyspace notifications publish key changes to SUBSCRIBE and PSUBSCRIBE clients.
+# Supported flags:
+#   K  keyspace channels
+#   E  keyevent channels
+#   g  generic events, currently del
+#   $  string events, currently set
+#   A  same as g$, without K or E
+#
+# Default namespace uses db 0. Redis database namespaces use db indexes.
+# Other namespaces use ns:encoded-name.
+# Notifications are emitted only when at least one channel flag (K or E) and one event class are enabled.
+#
+# Default: "" disabled
+notify-keyspace-events ""
+
 ################################## SLOW LOG ###################################
 
 # The Kvrocks Slow Log is a mechanism to log queries that exceeded a specified
diff --git a/src/commands/cmd_key.cc b/src/commands/cmd_key.cc
index 584f412937e..1f53d234794 100644
--- a/src/commands/cmd_key.cc
+++ b/src/commands/cmd_key.cc
@@ -19,9 +19,11 @@
  */
 
 #include <cstdint>
+#include <string_view>
 
 #include "commander.h"
 #include "commands/ttl_util.h"
+#include "common/keyspace_events.h"
 #include "error_constants.h"
 #include "server/redis_reply.h"
 #include "server/server.h"
@@ -372,9 +374,17 @@ class CommandDel : public Commander {
     uint64_t cnt = 0;
     redis::Database redis(srv->storage, conn->GetNamespace());
 
-    auto s = redis.MDel(ctx, keys, &cnt);
+    const bool notify_del = GetAttributes()->name == "del" && conn->IsKeyspaceEventEnabled(kNotifyGeneric);
+    std::vector<rocksdb::Slice> deleted_keys;
+    if (notify_del) deleted_keys.reserve(keys.size());
+
+    auto s = redis.MDel(ctx, keys, &cnt, notify_del ? &deleted_keys : nullptr);
     if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
 
+    for (const auto &key : deleted_keys) {
+      conn->AddKeyspaceEvent(kNotifyGeneric, "del", std::string_view(key.data(), key.size()));
+    }
+
     *output = redis::Integer(cnt);
     return Status::OK();
   }
diff --git a/src/commands/cmd_string.cc b/src/commands/cmd_string.cc
index 71d538a1da9..6ffeb68feb7 100644
--- a/src/commands/cmd_string.cc
+++ b/src/commands/cmd_string.cc
@@ -24,6 +24,8 @@
 
 #include "commander.h"
 #include "commands/command_parser.h"
+#include "common/keyspace_events.h"
+#include "common/string_util.h"
 #include "error_constants.h"
 #include "server/redis_reply.h"
 #include "server/redis_request.h"
@@ -383,13 +385,17 @@ class CommandSet : public Commander {
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     std::optional<std::string> ret;
     redis::String string_db(srv->storage, conn->GetNamespace());
+    const StringSetArgs set_args{expire_, set_flag_, get_, keep_ttl_, cmp_value_};
+    bool set_applied = false;
 
-    rocksdb::Status s = string_db.Set(ctx, args_[1], args_[2], {expire_, set_flag_, get_, keep_ttl_, cmp_value_}, ret);
+    rocksdb::Status s = string_db.Set(ctx, args_[1], args_[2], set_args, ret, &set_applied);
 
     if (!s.ok()) {
       return {Status::RedisExecErr, s.ToString()};
     }
 
+    if (set_applied) conn->AddKeyspaceEvent(kNotifyString, "set", args_[1]);
+
     if (get_) {
       if (ret.has_value()) {
         *output = redis::BulkString(ret.value());
diff --git a/src/commands/cmd_txn.cc b/src/commands/cmd_txn.cc
index 16d25a58ea7..0a25d7e72d0 100644
--- a/src/commands/cmd_txn.cc
+++ b/src/commands/cmd_txn.cc
@@ -90,6 +90,11 @@ class CommandExec : public Commander {
       s = storage->CommitTxn();
     }
 
+    // Publish queued notifications after a successful commit.
+    if (s.IsOK()) {
+      conn->FlushKeyspaceEvents();
+    }
+
     conn->ResetMultiExec();
     reset_multiexec.Disable();
 
diff --git a/src/common/keyspace_events.cc b/src/common/keyspace_events.cc
new file mode 100644
index 00000000000..02b97729866
--- /dev/null
+++ b/src/common/keyspace_events.cc
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "keyspace_events.h"
+
+#include <cstring>
+#include <utility>
+
+#include "config/config.h"
+#include "fmt/format.h"
+
+bool ShouldNotifyKeyspaceEvent(int notify_flags, int type_flag) {
+  return (notify_flags & type_flag) != 0 && (notify_flags & (kNotifyKeyspace | kNotifyKeyevent)) != 0;
+}
+
+KeyspaceEventCollector::KeyspaceEventCollector(std::string ns, int notify_flags)
+    : notify_flags_(notify_flags), ns_(std::move(ns)) {}
+
+void KeyspaceEventCollector::Add(int type_flag, std::string_view event, std::string_view key) {
+  if (!ShouldNotifyKeyspaceEvent(notify_flags_, type_flag)) return;
+  const int channel_flags = notify_flags_ & (kNotifyKeyspace | kNotifyKeyevent);
+  events_.emplace_back(KeyspaceEvent{channel_flags, std::string(event), ns_, std::string(key)});
+}
+
+std::vector<KeyspaceEvent> KeyspaceEventCollector::Take() { return std::move(events_); }
+
+Status ParseNotifyKeyspaceEventsFlags(const std::string &input, int *flags) {
+  int result = 0;
+  for (const char c : input) {
+    switch (c) {
+      case 'K':
+        result |= kNotifyKeyspace;
+        break;
+      case 'E':
+        result |= kNotifyKeyevent;
+        break;
+      case 'A':
+        result |= kNotifyAll;
+        break;
+      case 'g':
+        result |= kNotifyGeneric;
+        break;
+      case '$':
+        result |= kNotifyString;
+        break;
+      default:
+        return {Status::NotOK, fmt::format("unsupported notify-keyspace-events flag: '{}'", c)};
+    }
+  }
+
+  *flags = result;
+  return Status::OK();
+}
+
+namespace {
+std::string PercentEncode(const std::string &input) {
+  std::string output;
+  output.reserve(input.size());
+  for (const unsigned char c : input) {
+    if ((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_' || c == '.' ||
+        c == '-') {
+      output += static_cast<char>(c);
+    } else {
+      output += fmt::format("%{:02X}", c);
+    }
+  }
+  return output;
+}
+}  // namespace
+
+std::string MapNamespaceToKeyspaceDB(const std::string &ns, int redis_databases) {
+  if (ns == kDefaultNamespace) {
+    return "0";
+  }
+  if (redis_databases > 0 && ns.rfind(kDatabaseNamespacePrefix, 0) == 0) {
+    return ns.substr(strlen(kDatabaseNamespacePrefix));
+  }
+  return "ns:" + PercentEncode(ns);
+}
diff --git a/src/common/keyspace_events.h b/src/common/keyspace_events.h
new file mode 100644
index 00000000000..e0de54deec5
--- /dev/null
+++ b/src/common/keyspace_events.h
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#pragma once
+
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "status.h"
+
+// Flags for notify-keyspace-events, separate from RedisType.
+enum NotifyKeyspaceEventFlag {
+  kNotifyKeyspace = 1 << 0,  // K, keyspace channels
+  kNotifyKeyevent = 1 << 1,  // E, keyevent channels
+  kNotifyGeneric = 1 << 2,   // g, emits del
+  kNotifyString = 1 << 3,    // $, emits set
+  // A, supported data classes without K or E.
+  kNotifyAll = kNotifyGeneric | kNotifyString,
+};
+
+bool ShouldNotifyKeyspaceEvent(int notify_flags, int type_flag);
+
+struct KeyspaceEvent {
+  int channel_flags;
+  std::string event;
+  std::string ns;
+  std::string key;
+};
+
+// Collects semantic keyspace events for one command; Connection owns publish timing.
+class KeyspaceEventCollector {
+ public:
+  KeyspaceEventCollector(std::string ns, int notify_flags);
+  void Add(int type_flag, std::string_view event, std::string_view key);
+  // Moves out events collected during Execute.
+  std::vector<KeyspaceEvent> Take();
+
+ private:
+  int notify_flags_;
+  std::string ns_;
+  std::vector<KeyspaceEvent> events_;
+};
+
+// Parses notify-keyspace-events flags.
+Status ParseNotifyKeyspaceEventsFlags(const std::string &input, int *flags);
+
+// Maps namespaces to notification db names.
+// Default namespace maps to 0; database namespaces map back to db indexes when redis-databases is enabled.
+// Other namespaces map to ns:encoded-name.
+std::string MapNamespaceToKeyspaceDB(const std::string &ns, int redis_databases);
diff --git a/src/config/config.cc b/src/config/config.cc
index 0de6a2c6f47..5ab5c6ed5c4 100644
--- a/src/config/config.cc
+++ b/src/config/config.cc
@@ -35,6 +35,7 @@
 #include <utility>
 #include <vector>
 
+#include "common/keyspace_events.h"
 #include "common/string_util.h"
 #include "config_type.h"
 #include "config_util.h"
@@ -192,6 +193,7 @@ Config::Config() {
       {"compact-cron", false, new StringField(&compact_cron_str_, "")},
       {"bgsave-cron", false, new StringField(&bgsave_cron_str_, "")},
       {"dbsize-scan-cron", false, new StringField(&dbsize_scan_cron_str_, "")},
+      {"notify-keyspace-events", false, new StringField(&notify_keyspace_events_str_, "")},
       {"replica-announce-ip", false, new StringField(&replica_announce_ip, "")},
       {"replica-announce-port", false, new UInt32Field(&replica_announce_port, 0, 0, PORT_LIMIT)},
       {"compaction-checker-range", false, new StringField(&compaction_checker_range_str_, "")},
@@ -379,6 +381,11 @@ void Config::initFieldValidator() {
          }
          return Status::OK();
        }},
+      {"notify-keyspace-events",
+       []([[maybe_unused]] const std::string &k, const std::string &v) -> Status {
+         int flags = 0;
+         return ParseNotifyKeyspaceEventsFlags(v, &flags);
+       }},
       {"compact-cron",
        [this]([[maybe_unused]] const std::string &k, const std::string &v) -> Status {
          std::vector<std::string> args = util::Split(v, " \t");
@@ -556,6 +563,10 @@ void Config::initFieldCallback() {
              srv->AdjustWorkerThreads();
              return Status::OK();
            }},
+          {"notify-keyspace-events",
+           [this]([[maybe_unused]] Server *srv, [[maybe_unused]] const std::string &k, const std::string &v) -> Status {
+             return ParseNotifyKeyspaceEventsFlags(v, &notify_keyspace_events);
+           }},
           {"dir",
            [this]([[maybe_unused]] Server *srv, [[maybe_unused]] const std::string &k,
                   [[maybe_unused]] const std::string &v) -> Status {
diff --git a/src/config/config.h b/src/config/config.h
index a6dd19f213c..34c19325444 100644
--- a/src/config/config.h
+++ b/src/config/config.h
@@ -219,6 +219,9 @@ struct Config {
   // Enable transactional mode in engine::Context
   bool txn_context_enabled = false;
 
+  // Parsed notify-keyspace-events flags.
+  int notify_keyspace_events = 0;
+
   bool skip_block_cache_deallocation_on_close = false;
 
   bool lua_strict_key_accessing = false;
@@ -315,6 +318,7 @@ struct Config {
   std::string compaction_checker_cron_str_;
   std::string profiling_sample_commands_str_;
   std::string client_output_buffer_limit_str_;
+  std::string notify_keyspace_events_str_;
   std::map<std::string, std::unique_ptr<ConfigField>> fields_;
   std::vector<std::string> rename_command_;
   std::string histogram_bucket_boundaries_str_;
diff --git a/src/server/redis_connection.cc b/src/server/redis_connection.cc
index 65061c01707..5ed2176cace 100644
--- a/src/server/redis_connection.cc
+++ b/src/server/redis_connection.cc
@@ -23,6 +23,7 @@
 
 #include <mutex>
 #include <nonstd/span.hpp>
+#include <optional>
 #include <shared_mutex>
 
 #include "commands/commander.h"
@@ -466,7 +467,17 @@ Status Connection::ExecuteCommand(engine::Context &ctx, const std::string &cmd_n
 
   auto start = std::chrono::high_resolution_clock::now();
   bool is_profiling = IsProfilingEnabled(cmd_name);
+
+  keyspace_event_notify_flags_ = srv_->GetConfig()->notify_keyspace_events;
+  active_keyspace_event_collector_.reset();
+
   auto s = current_cmd->Execute(ctx, srv_, this, reply);
+  if (s.IsOK() && active_keyspace_event_collector_) {
+    queueOrPublishKeyspaceEvents(active_keyspace_event_collector_->Take());
+  }
+  active_keyspace_event_collector_.reset();
+  keyspace_event_notify_flags_ = 0;
+
   auto end = std::chrono::high_resolution_clock::now();
   uint64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
   if (is_profiling) RecordProfilingSampleIfNeed(cmd_name, duration);
@@ -476,6 +487,20 @@ Status Connection::ExecuteCommand(engine::Context &ctx, const std::string &cmd_n
   return s;
 }
 
+bool Connection::IsKeyspaceEventEnabled(int type_flag) const {
+  return ShouldNotifyKeyspaceEvent(keyspace_event_notify_flags_, type_flag);
+}
+
+void Connection::AddKeyspaceEvent(int type_flag, std::string_view event, std::string_view key) {
+  if (!IsKeyspaceEventEnabled(type_flag)) return;
+
+  if (!active_keyspace_event_collector_) {
+    active_keyspace_event_collector_ =
+        std::make_unique<KeyspaceEventCollector>(GetNamespace(), keyspace_event_notify_flags_);
+  }
+  active_keyspace_event_collector_->Add(type_flag, event, key);
+}
+
 static bool IsCmdForIndexing(uint64_t cmd_flags, CommandCategory cmd_cat) {
   return (cmd_flags & redis::kCmdWrite) &&
          (cmd_cat == CommandCategory::Hash || cmd_cat == CommandCategory::JSON || cmd_cat == CommandCategory::Key ||
@@ -709,10 +734,40 @@ void Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) {
   }
 }
 
+void Connection::queueOrPublishKeyspaceEvents(std::vector<KeyspaceEvent> &&events) {
+  if (events.empty()) return;
+
+  if (in_exec_) {
+    // Queue transaction events until commit.
+    for (auto &event : events) {
+      pending_keyspace_events_.emplace_back(std::move(event));
+    }
+    return;
+  }
+
+  for (const auto &event : events) {
+    srv_->NotifyKeyspaceEvent(event.channel_flags, event.event, event.ns, event.key);
+  }
+}
+
+void Connection::FlushKeyspaceEvents() {
+  for (const auto &e : pending_keyspace_events_) {
+    srv_->NotifyKeyspaceEvent(e.channel_flags, e.event, e.ns, e.key);
+  }
+  pending_keyspace_events_.clear();
+}
+
 void Connection::ResetMultiExec() {
   in_exec_ = false;
   multi_error_ = false;
   multi_cmds_.clear();
+  // Drop events from failed or aborted transactions.
+  pending_keyspace_events_.clear();
+  // Retain capacity for typical transactions, but request releasing unusually large buffers.
+  constexpr std::size_t kMaxRetainedKeyspaceEvents = 1024;
+  if (pending_keyspace_events_.capacity() > kMaxRetainedKeyspaceEvents) {
+    pending_keyspace_events_.shrink_to_fit();
+  }
   DisableFlag(Connection::kMultiExec);
 }
 
diff --git a/src/server/redis_connection.h b/src/server/redis_connection.h
index c4bbd630cc4..5f18b2271e4 100644
--- a/src/server/redis_connection.h
+++ b/src/server/redis_connection.h
@@ -26,10 +26,12 @@
 #include <memory>
 #include <set>
 #include <string>
+#include <string_view>
 #include <utility>
 #include <vector>
 
 #include "commands/commander.h"
+#include "common/keyspace_events.h"
 #include "event_util.h"
 #include "redis_request.h"
 #include "server/redis_reply.h"
@@ -208,6 +210,10 @@ class Connection : public EvbufCallbackBase<Connection> {
   void ResetMultiExec();
   std::deque<redis::CommandTokens> *GetMultiExecCommands() { return &multi_cmds_; }
 
+  void FlushKeyspaceEvents();
+  bool IsKeyspaceEventEnabled(int type_flag) const;
+  void AddKeyspaceEvent(int type_flag, std::string_view event, std::string_view key);
+
   std::function<void(int)> close_cb = nullptr;
 
   std::set<std::string> watched_keys;
@@ -218,6 +224,9 @@ class Connection : public EvbufCallbackBase<Connection> {
   ReplyMode GetReplyMode() const { return reply_mode_; }
 
  private:
+  // Queues events while EXEC is running; publishes them otherwise.
+  void queueOrPublishKeyspaceEvents(std::vector<KeyspaceEvent> &&events);
+
   uint64_t id_ = 0;
   std::atomic<int> flags_ = 0;
   std::string ns_;
@@ -248,6 +257,10 @@ class Connection : public EvbufCallbackBase<Connection> {
   bool multi_error_ = false;
   std::atomic<bool> is_running_ = false;
   std::deque<redis::CommandTokens> multi_cmds_;
+
+  int keyspace_event_notify_flags_ = 0;
+  std::unique_ptr<KeyspaceEventCollector> active_keyspace_event_collector_;
+  std::vector<KeyspaceEvent> pending_keyspace_events_;
   bool in_script_ = false;
 
   bool importing_ = false;
diff --git a/src/server/server.cc b/src/server/server.cc
index 48f6177d8a1..d5dcb27c740 100644
--- a/src/server/server.cc
+++ b/src/server/server.cc
@@ -41,6 +41,7 @@
 
 #include "commands/command_parser.h"
 #include "commands/commander.h"
+#include "common/keyspace_events.h"
 #include "common/string_util.h"
 #include "config/config.h"
 #include "fmt/format.h"
@@ -478,6 +479,17 @@ int Server::PublishMessage(const std::string &channel, const std::string &msg) {
   return cnt;
 }
 
+void Server::NotifyKeyspaceEvent(int flags, const std::string &event, const std::string &ns, const std::string &key) {
+  const std::string db = MapNamespaceToKeyspaceDB(ns, GetConfig()->redis_databases);
+  // Publish keyspace before keyevent for each key.
+  if (flags & kNotifyKeyspace) {
+    PublishMessage("__keyspace@" + db + "__:" + key, event);
+  }
+  if (flags & kNotifyKeyevent) {
+    PublishMessage("__keyevent@" + db + "__:" + event, key);
+  }
+}
+
 void Server::SubscribeChannel(const std::string &channel, redis::Connection *conn) {
   std::lock_guard<std::mutex> guard(pubsub_channels_mu_);
 
diff --git a/src/server/server.h b/src/server/server.h
index 4214cecc53b..ce592627094 100644
--- a/src/server/server.h
+++ b/src/server/server.h
@@ -218,6 +218,9 @@ class Server {
   int GetFetchFileThreadNum() const { return fetch_file_threads_num_; }
 
   int PublishMessage(const std::string &channel, const std::string &msg);
+
+  // Publishes a keyspace event through the channels selected when it was collected.
+  void NotifyKeyspaceEvent(int flags, const std::string &event, const std::string &ns, const std::string &key);
   void SubscribeChannel(const std::string &channel, redis::Connection *conn);
   void UnsubscribeChannel(const std::string &channel, redis::Connection *conn);
   void GetChannelsByPattern(const std::string &pattern, std::vector<std::string> *channels);
diff --git a/src/storage/redis_db.cc b/src/storage/redis_db.cc
index 99b3d96a35a..3d7e175ca18 100644
--- a/src/storage/redis_db.cc
+++ b/src/storage/redis_db.cc
@@ -21,6 +21,7 @@
 #include "redis_db.h"
 
 #include <ctime>
+#include <unordered_set>
 #include <utility>
 
 #include "cluster/redis_slot.h"
@@ -158,8 +159,10 @@ rocksdb::Status Database::Del(engine::Context &ctx, const Slice &user_key) {
   return storage_->Delete(ctx, storage_->DefaultWriteOptions(), metadata_cf_handle_, ns_key);
 }
 
-rocksdb::Status Database::MDel(engine::Context &ctx, const std::vector<Slice> &keys, uint64_t *deleted_cnt) {
+rocksdb::Status Database::MDel(engine::Context &ctx, const std::vector<Slice> &keys, uint64_t *deleted_cnt,
+                               std::vector<Slice> *deleted_user_keys) {
   *deleted_cnt = 0;
+  if (deleted_user_keys != nullptr) deleted_user_keys->clear();
 
   std::vector<std::string> ns_keys;
   ns_keys.reserve(keys.size());
@@ -186,6 +189,8 @@ rocksdb::Status Database::MDel(engine::Context &ctx, const std::vector<Slice> &k
   storage_->MultiGet(ctx, ctx.DefaultMultiGetOptions(), metadata_cf_handle_, slice_keys.size(), slice_keys.data(),
                      pin_values.data(), statuses.data());
 
+  std::unordered_set<std::string> deleted_ns_keys;
+  deleted_ns_keys.reserve(keys.size());
   for (size_t i = 0; i < slice_keys.size(); i++) {
     if (!statuses[i].ok() && !statuses[i].IsNotFound()) return statuses[i];
     if (statuses[i].IsNotFound()) continue;
@@ -196,10 +201,14 @@ rocksdb::Status Database::MDel(engine::Context &ctx, const std::vector<Slice> &k
     auto s = metadata.Decode(rocksdb::Slice(pin_values[i].data(), pin_values[i].size()));
     if (!s.ok()) continue;
     if (metadata.Expired()) continue;
+    if (!deleted_ns_keys.emplace(ns_keys[i]).second) continue;
 
     s = batch->Delete(metadata_cf_handle_, ns_keys[i]);
     if (!s.ok()) return s;
     *deleted_cnt += 1;
+    if (deleted_user_keys != nullptr) {
+      deleted_user_keys->emplace_back(keys[i]);
+    }
   }
 
   if (*deleted_cnt == 0) return rocksdb::Status::OK();
diff --git a/src/storage/redis_db.h b/src/storage/redis_db.h
index 506dc250bae..b9af91d74ad 100644
--- a/src/storage/redis_db.h
+++ b/src/storage/redis_db.h
@@ -110,7 +110,8 @@ class Database {
   [[nodiscard]] rocksdb::Status GetRawMetadata(engine::Context &ctx, const Slice &ns_key, std::string *bytes);
   [[nodiscard]] rocksdb::Status Expire(engine::Context &ctx, const Slice &user_key, uint64_t timestamp);
   [[nodiscard]] rocksdb::Status Del(engine::Context &ctx, const Slice &user_key);
-  [[nodiscard]] rocksdb::Status MDel(engine::Context &ctx, const std::vector<Slice> &keys, uint64_t *deleted_cnt);
+  [[nodiscard]] rocksdb::Status MDel(engine::Context &ctx, const std::vector<Slice> &keys, uint64_t *deleted_cnt,
+                                     std::vector<Slice> *deleted_user_keys = nullptr);
   [[nodiscard]] rocksdb::Status Exists(engine::Context &ctx, const std::vector<Slice> &keys, uint32_t *ret);
   [[nodiscard]] rocksdb::Status TTL(engine::Context &ctx, const Slice &user_key, int64_t *ttl);
   [[nodiscard]] rocksdb::Status GetExpireTime(engine::Context &ctx, const Slice &user_key, uint64_t *timestamp);
diff --git a/src/types/redis_string.cc b/src/types/redis_string.cc
index 8563275bf44..00768fe4f0f 100644
--- a/src/types/redis_string.cc
+++ b/src/types/redis_string.cc
@@ -239,7 +239,9 @@ rocksdb::Status String::Set(engine::Context &ctx, const std::string &user_key, c
 }
 
 rocksdb::Status String::Set(engine::Context &ctx, const std::string &user_key, const std::string &value,
-                            const StringSetArgs &args, std::optional<std::string> &ret) {
+                            const StringSetArgs &args, std::optional<std::string> &ret, bool *applied) {
+  if (applied != nullptr) *applied = false;
+
   uint64_t expire = 0;
   std::string ns_key = AppendNamespacePrefix(user_key);
 
@@ -345,7 +347,9 @@ rocksdb::Status String::Set(engine::Context &ctx, const std::string &user_key, c
   metadata.expire = expire;
   metadata.Encode(&new_raw_value);
   new_raw_value.append(value);
-  return updateRawValue(ctx, ns_key, new_raw_value);
+  auto s = updateRawValue(ctx, ns_key, new_raw_value);
+  if (s.ok() && applied != nullptr) *applied = true;
+  return s;
 }
 
 rocksdb::Status String::SetEX(engine::Context &ctx, const std::string &user_key, const std::string &value,
diff --git a/src/types/redis_string.h b/src/types/redis_string.h
index 57fb309da5c..c0b32c69a16 100644
--- a/src/types/redis_string.h
+++ b/src/types/redis_string.h
@@ -105,7 +105,7 @@ class String : public Database {
   rocksdb::Status GetDel(engine::Context &ctx, const std::string &user_key, std::string *value);
   rocksdb::Status Set(engine::Context &ctx, const std::string &user_key, const std::string &value);
   rocksdb::Status Set(engine::Context &ctx, const std::string &user_key, const std::string &value,
-                      const StringSetArgs &args, std::optional<std::string> &ret);
+                      const StringSetArgs &args, std::optional<std::string> &ret, bool *applied = nullptr);
   rocksdb::Status SetEX(engine::Context &ctx, const std::string &user_key, const std::string &value,
                         uint64_t expire_ms);
   rocksdb::Status SetNX(engine::Context &ctx, const std::string &user_key, const std::string &value, uint64_t expire_ms,
diff --git a/tests/cppunit/keyspace_events_test.cc b/tests/cppunit/keyspace_events_test.cc
new file mode 100644
index 00000000000..d809fc85584
--- /dev/null
+++ b/tests/cppunit/keyspace_events_test.cc
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "common/keyspace_events.h"
+
+#include <gtest/gtest.h>
+
+#include "config/config.h"
+
+TEST(KeyspaceEvents, CollectorRequiresEventClassAndChannel) {
+  KeyspaceEventCollector no_channel("tenant", kNotifyString);
+  EXPECT_FALSE(ShouldNotifyKeyspaceEvent(kNotifyString, kNotifyString));
+  no_channel.Add(kNotifyString, "set", "key");
+  EXPECT_TRUE(no_channel.Take().empty());
+
+  KeyspaceEventCollector no_event_class("tenant", kNotifyKeyspace);
+  EXPECT_FALSE(ShouldNotifyKeyspaceEvent(kNotifyKeyspace, kNotifyString));
+  no_event_class.Add(kNotifyString, "set", "key");
+  EXPECT_TRUE(no_event_class.Take().empty());
+}
+
+TEST(KeyspaceEvents, CollectorFiltersAndCapturesEvent) {
+  KeyspaceEventCollector collector("tenant", kNotifyKeyspace | kNotifyString);
+  EXPECT_TRUE(ShouldNotifyKeyspaceEvent(kNotifyKeyspace | kNotifyString, kNotifyString));
+  EXPECT_FALSE(ShouldNotifyKeyspaceEvent(kNotifyKeyspace | kNotifyString, kNotifyGeneric));
+
+  collector.Add(kNotifyGeneric, "del", "ignored");
+  collector.Add(kNotifyString, "set", "key");
+
+  auto events = collector.Take();
+  ASSERT_EQ(events.size(), 1);
+  EXPECT_EQ(events[0].channel_flags, kNotifyKeyspace);
+  EXPECT_EQ(events[0].event, "set");
+  EXPECT_EQ(events[0].ns, "tenant");
+  EXPECT_EQ(events[0].key, "key");
+}
+
+TEST(KeyspaceEvents, CollectorPreservesEventOrder) {
+  KeyspaceEventCollector collector("tenant", kNotifyKeyspace | kNotifyKeyevent | kNotifyAll);
+  collector.Add(kNotifyString, "set", "first");
+  collector.Add(kNotifyGeneric, "del", "second");
+
+  auto events = collector.Take();
+  ASSERT_EQ(events.size(), 2);
+  EXPECT_EQ(events[0].channel_flags, kNotifyKeyspace | kNotifyKeyevent);
+  EXPECT_EQ(events[0].event, "set");
+  EXPECT_EQ(events[0].key, "first");
+  EXPECT_EQ(events[1].channel_flags, kNotifyKeyspace | kNotifyKeyevent);
+  EXPECT_EQ(events[1].event, "del");
+  EXPECT_EQ(events[1].key, "second");
+}
+
+TEST(KeyspaceEvents, ParseFlags) {
+  int flags = 0;
+
+  // Empty disables notifications.
+  ASSERT_TRUE(ParseNotifyKeyspaceEventsFlags("", &flags).IsOK());
+  ASSERT_EQ(flags, 0);
+
+  // Each flag maps to one bit.
+  ASSERT_TRUE(ParseNotifyKeyspaceEventsFlags("K", &flags).IsOK());
+  ASSERT_EQ(flags, kNotifyKeyspace);
+  ASSERT_TRUE(ParseNotifyKeyspaceEventsFlags("E", &flags).IsOK());
+  ASSERT_EQ(flags, kNotifyKeyevent);
+  ASSERT_TRUE(ParseNotifyKeyspaceEventsFlags("g", &flags).IsOK());
+  ASSERT_EQ(flags, kNotifyGeneric);
+  ASSERT_TRUE(ParseNotifyKeyspaceEventsFlags("$", &flags).IsOK());
+  ASSERT_EQ(flags, kNotifyString);
+
+  // KEA enables both channels and set or del.
+  ASSERT_TRUE(ParseNotifyKeyspaceEventsFlags("KEA", &flags).IsOK());
+  ASSERT_TRUE(flags & kNotifyKeyspace);
+  ASSERT_TRUE(flags & kNotifyKeyevent);
+  ASSERT_TRUE(flags & kNotifyGeneric);  // del
+  ASSERT_TRUE(flags & kNotifyString);   // set
+}
+
+TEST(KeyspaceEvents, ParseFlagsAExpansion) {
+  int flags = 0;
+  ASSERT_TRUE(ParseNotifyKeyspaceEventsFlags("A", &flags).IsOK());
+  // A expands to all supported event classes without K or E.
+  ASSERT_EQ(flags, kNotifyAll);
+  ASSERT_FALSE(flags & kNotifyKeyspace);
+  ASSERT_FALSE(flags & kNotifyKeyevent);
+  ASSERT_TRUE(flags & kNotifyGeneric);
+  ASSERT_TRUE(flags & kNotifyString);
+}
+
+TEST(KeyspaceEvents, ParseFlagsRejectsUnsupported) {
+  int flags = 0;
+  // Unsupported flags are rejected.
+  for (const auto *bad : {"a", "d", "x", "e", "m", "n", "o", "c", "l", "s", "h", "z", "t", "Kx", "KEl", "?"}) {
+    ASSERT_FALSE(ParseNotifyKeyspaceEventsFlags(bad, &flags).IsOK()) << "should reject: " << bad;
+  }
+}
+
+TEST(KeyspaceEvents, MapNamespaceToKeyspaceDB) {
+  // Default namespace maps to db 0.
+  EXPECT_EQ(MapNamespaceToKeyspaceDB(kDefaultNamespace, 0), "0");
+  EXPECT_EQ(MapNamespaceToKeyspaceDB(kDefaultNamespace, 16), "0");
+
+  // Non-default namespaces are encoded and prefixed.
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("0", 0), "ns:0");
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("tenantA", 0), "ns:tenantA");
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("a.b-c_d", 0), "ns:a.b-c_d");
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("db1", 0), "ns:db1");
+  // Unsafe bytes are escaped.
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("a b", 0), "ns:a%20b");
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("a:b", 0), "ns:a%3Ab");
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("100%", 0), "ns:100%25");
+
+  // Redis database namespaces map back to numeric database names when redis-databases is enabled.
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("db1", 16), "1");
+  EXPECT_EQ(MapNamespaceToKeyspaceDB("db15", 16), "15");
+}
diff --git a/tests/gocase/unit/keyspace/keyspace_test.go b/tests/gocase/unit/keyspace/keyspace_test.go
index 37b86afd1d8..d67f1f434b8 100644
--- a/tests/gocase/unit/keyspace/keyspace_test.go
+++ b/tests/gocase/unit/keyspace/keyspace_test.go
@@ -53,6 +53,9 @@ func TestKeyspace(t *testing.T) {
 		require.NoError(t, rdb.Set(ctx, "foo3", "c", 0).Err())
 		require.EqualValues(t, 3, rdb.Del(ctx, "foo1", "foo2", "foo3").Val())
 		require.Equal(t, []interface{}{nil, nil, nil}, rdb.MGet(ctx, "foo1", "foo2", "foo3").Val())
+
+		require.NoError(t, rdb.Set(ctx, "foo-dup", "a", 0).Err())
+		require.EqualValues(t, 1, rdb.Del(ctx, "foo-dup", "foo-dup").Val())
 	})
 
 	t.Run("KEYS with pattern", func(t *testing.T) {
diff --git a/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go b/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go
new file mode 100644
index 00000000000..612200b90b1
--- /dev/null
+++ b/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package keyspacenotify
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/apache/kvrocks/tests/gocase/util"
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/require"
+)
+
+// drainSubscribeConfirms waits for subscription confirmations.
+func drainSubscribeConfirms(t *testing.T, ctx context.Context, pubsub *redis.PubSub, n int) {
+	t.Helper()
+	for range n {
+		msg, err := pubsub.ReceiveTimeout(ctx, 2*time.Second)
+		require.NoError(t, err)
+		require.IsType(t, &redis.Subscription{}, msg)
+	}
+}
+
+// expectMessage checks the next pubsub message.
+func expectMessage(t *testing.T, ctx context.Context, pubsub *redis.PubSub, channel, payload string) {
+	t.Helper()
+	msg, err := pubsub.ReceiveTimeout(ctx, 2*time.Second)
+	require.NoError(t, err)
+	m, ok := msg.(*redis.Message)
+	require.Truef(t, ok, "expected *redis.Message, got %T", msg)
+	require.Equal(t, channel, m.Channel)
+	require.Equal(t, payload, m.Payload)
+}
+
+// expectNoMessage checks that no message arrives soon.
+func expectNoMessage(t *testing.T, ctx context.Context, pubsub *redis.PubSub) {
+	t.Helper()
+	msg, err := pubsub.ReceiveTimeout(ctx, 300*time.Millisecond)
+	require.Errorf(t, err, "expected no message, got %v", msg)
+}
+
+func TestKeyspaceNotify(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{"notify-keyspace-events": "KEA"})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	// Subscribe to both channel forms.
+	pubsub := rdb.PSubscribe(ctx, "__keyspace@0__:*", "__keyevent@0__:*")
+	defer func() { require.NoError(t, pubsub.Close()) }()
+	drainSubscribeConfirms(t, ctx, pubsub, 2)
+
+	t.Run("SET publishes set", func(t *testing.T) {
+		require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+		// Keyspace is published before keyevent.
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:foo", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "foo")
+	})
+
+	t.Run("SET NX on existing key publishes nothing", func(t *testing.T) {
+		require.NoError(t, rdb.Set(ctx, "nxkey", "v1", 0).Err())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:nxkey", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "nxkey")
+
+		// NX fails, so nothing is published.
+		require.NoError(t, rdb.SetNX(ctx, "nxkey", "v2", 0).Err())
+		expectNoMessage(t, ctx, pubsub)
+	})
+
+	t.Run("SET GET with conditions publishes only on write", func(t *testing.T) {
+		require.NoError(t, rdb.Set(ctx, "getnx", "v1", 0).Err())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:getnx", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "getnx")
+
+		cmd := rdb.Do(ctx, "SET", "getnx", "v2", "GET", "NX")
+		require.NoError(t, cmd.Err())
+		require.Equal(t, "v1", cmd.Val())
+		expectNoMessage(t, ctx, pubsub)
+
+		cmd = rdb.Do(ctx, "SET", "getnx-missing", "v1", "GET", "NX")
+		require.ErrorIs(t, cmd.Err(), redis.Nil)
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:getnx-missing", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "getnx-missing")
+
+		cmd = rdb.Do(ctx, "SET", "getxx-missing", "v1", "GET", "XX")
+		require.ErrorIs(t, cmd.Err(), redis.Nil)
+		expectNoMessage(t, ctx, pubsub)
+
+		require.NoError(t, rdb.Set(ctx, "ifeq", "v1", 0).Err())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:ifeq", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "ifeq")
+
+		cmd = rdb.Do(ctx, "SET", "ifeq", "v2", "GET", "IFEQ", "other")
+		require.NoError(t, cmd.Err())
+		require.Equal(t, "v1", cmd.Val())
+		expectNoMessage(t, ctx, pubsub)
+
+		cmd = rdb.Do(ctx, "SET", "ifeq", "v2", "GET", "IFEQ", "v1")
+		require.NoError(t, cmd.Err())
+		require.Equal(t, "v1", cmd.Val())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:ifeq", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "ifeq")
+	})
+
+	t.Run("DEL publishes one del per deleted key", func(t *testing.T) {
+		require.NoError(t, rdb.Set(ctx, "d1", "x", 0).Err())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:d1", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "d1")
+
+		// Only d1 is deleted.
+		require.EqualValues(t, 1, rdb.Del(ctx, "d1", "d2").Val())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:d1", "del")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "d1")
+		expectNoMessage(t, ctx, pubsub)
+
+		require.NoError(t, rdb.Set(ctx, "ddup", "x", 0).Err())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:ddup", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "ddup")
+
+		require.EqualValues(t, 1, rdb.Del(ctx, "ddup", "ddup").Val())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:ddup", "del")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "ddup")
+		expectNoMessage(t, ctx, pubsub)
+	})
+
+	t.Run("UNLINK does not publish del", func(t *testing.T) {
+		require.NoError(t, rdb.Set(ctx, "unlink-key", "x", 0).Err())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:unlink-key", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "unlink-key")
+
+		require.EqualValues(t, 1, rdb.Unlink(ctx, "unlink-key").Val())
+		expectNoMessage(t, ctx, pubsub)
+	})
+
+	t.Run("Lua nested commands publish events", func(t *testing.T) {
+		script := `
+			redis.call("SET", KEYS[1], "v")
+			redis.call("DEL", KEYS[1])
+			return 1
+		`
+		require.NoError(t, rdb.Eval(ctx, script, []string{"lua-key"}).Err())
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:lua-key", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "lua-key")
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:lua-key", "del")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "lua-key")
+		expectNoMessage(t, ctx, pubsub)
+	})
+
+	t.Run("MULTI/EXEC publishes queued events after commit", func(t *testing.T) {
+		_, err := rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+			pipe.Set(ctx, "m1", "v", 0)
+			pipe.Del(ctx, "m1")
+			return nil
+		})
+		require.NoError(t, err)
+		// Events publish after commit.
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:m1", "set")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "m1")
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:m1", "del")
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:del", "m1")
+		expectNoMessage(t, ctx, pubsub)
+	})
+
+	t.Run("MULTI/EXEC preserves per-command notification config", func(t *testing.T) {
+		require.NoError(t, rdb.ConfigSet(ctx, "notify-keyspace-events", "E$").Err())
+		_, err := rdb.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+			pipe.Set(ctx, "event-before-disable", "v", 0)
+			pipe.ConfigSet(ctx, "notify-keyspace-events", "")
+			pipe.Set(ctx, "event-while-disabled", "v", 0)
+			pipe.ConfigSet(ctx, "notify-keyspace-events", "K$")
+			pipe.Set(ctx, "event-after-enable", "v", 0)
+			return nil
+		})
+		require.NoError(t, err)
+
+		expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "event-before-disable")
+		expectMessage(t, ctx, pubsub, "__keyspace@0__:event-after-enable", "set")
+		expectNoMessage(t, ctx, pubsub)
+	})
+}
+
+func TestKeyspaceNotifyDisabled(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	pubsub := rdb.PSubscribe(ctx, "__keyspace@0__:*", "__keyevent@0__:*")
+	defer func() { require.NoError(t, pubsub.Close()) }()
+	drainSubscribeConfirms(t, ctx, pubsub, 2)
+
+	require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+	require.NoError(t, rdb.Del(ctx, "foo").Err())
+	expectNoMessage(t, ctx, pubsub)
+
+	// CONFIG SET enables notifications at runtime.
+	require.NoError(t, rdb.ConfigSet(ctx, "notify-keyspace-events", "KEA").Err())
+	require.NoError(t, rdb.Set(ctx, "foo", "bar", 0).Err())
+	expectMessage(t, ctx, pubsub, "__keyspace@0__:foo", "set")
+	expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "foo")
+
+	// CONFIG SET rejects unsupported flags.
+	require.Error(t, rdb.ConfigSet(ctx, "notify-keyspace-events", "Kx").Err())
+	require.Error(t, rdb.ConfigSet(ctx, "notify-keyspace-events", "KEl").Err())
+}
+
+func TestKeyspaceNotifyRedisDatabases(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{"notify-keyspace-events": "KEA", "redis-databases": "16"})
+	defer srv.Close()
+
+	ctx := context.Background()
+	sub := srv.NewClient()
+	defer func() { require.NoError(t, sub.Close()) }()
+	writer := srv.NewClient()
+	defer func() { require.NoError(t, writer.Close()) }()
+
+	pubsub := sub.PSubscribe(ctx, "__keyspace@1__:*", "__keyevent@1__:*")
+	defer func() { require.NoError(t, pubsub.Close()) }()
+	drainSubscribeConfirms(t, ctx, pubsub, 2)
+
+	require.NoError(t, writer.Do(ctx, "SELECT", 1).Err())
+	require.NoError(t, writer.Set(ctx, "db-key", "v", 0).Err())
+	expectMessage(t, ctx, pubsub, "__keyspace@1__:db-key", "set")
+	expectMessage(t, ctx, pubsub, "__keyevent@1__:set", "db-key")
+
+	require.EqualValues(t, 1, writer.Del(ctx, "db-key").Val())
+	expectMessage(t, ctx, pubsub, "__keyspace@1__:db-key", "del")
+	expectMessage(t, ctx, pubsub, "__keyevent@1__:del", "db-key")
+	expectNoMessage(t, ctx, pubsub)
+}