feat(readonly): reject writes in gRPC handlers + implement SetReadOnly/GetReadOnlyStatus

Every write handler (Insert/Update/Upsert/Delete/Patch/Batch/CreateCollection/
DropCollection/CreateView/DropView/ConfigureCollection/MigrateCollectionTimestamps/
SetAdd/SetRemove/UploadFile/DeleteFile/RestoreVersion/RestoreToDate) now checks
service.isReadOnly() at the top and rejects with FAILED_PRECONDITION if set.

Read handlers (Get/Find/Count/ListCollections/Subscribe/etc.) unaffected —
read-only means read-ONLY.

New SetReadOnly RPC lets clients toggle state at runtime.
New GetReadOnlyStatus RPC returns current state + recovery outcome details.

service/src/database_grpc_impl.cpp

@@ -1,4 +1,5 @@
 #include "database_grpc_impl.hpp"
+#include "database_service.hpp"
 #include "persistence/wal.hpp"
 #include "views/projection.hpp"
 
@@ -10,9 +11,19 @@ namespace smartbotic::database {
 // Namespace alias for proto types
 namespace pb = smartbotic::databasepb;
 
+namespace {
+// Returns a Status that should be returned when the DB is read-only.
+// Used by write handlers whose responses don't have set_success/set_error fields.
+grpc::Status readOnlyStatus(const std::string& rpcName, const DatabaseService& svc) {
+    return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
+        rpcName + " rejected: database is in read-only mode. " + svc.readOnlyReason());
+}
+} // anonymous namespace
+
 // ===== DatabaseGrpcImpl =====
 
 DatabaseGrpcImpl::DatabaseGrpcImpl(
+    DatabaseService& service,
     MemoryStore& store,
     PersistenceManager& persistence,
     EventManager& events,
@@ -20,7 +31,8 @@ DatabaseGrpcImpl::DatabaseGrpcImpl(
     EncryptionManager& encryption,
     ViewManager& view_manager,
     CollectionConfigManager& config_manager
-) : store_(store)
+) : service_(service)
+  , store_(store)
   , persistence_(persistence)
   , events_(events)
   , files_(files)
@@ -37,6 +49,9 @@ grpc::Status DatabaseGrpcImpl::Insert(
     const pb::InsertRequest* request,
     pb::InsertResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("Insert", service_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -138,6 +153,11 @@ grpc::Status DatabaseGrpcImpl::Update(
     const pb::UpdateRequest* request,
     pb::UpdateResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
     if (view_manager_.isView(request->collection())) {
         response->set_success(false);
         response->set_error("cannot write to view '" + request->collection() + "': views are read-only");
@@ -212,6 +232,11 @@ grpc::Status DatabaseGrpcImpl::PatchDocument(
     const pb::PatchDocumentRequest* request,
     pb::PatchDocumentResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
     if (view_manager_.isView(request->collection())) {
         response->set_success(false);
         response->set_error("cannot write to view '" + request->collection() + "': views are read-only");
@@ -262,6 +287,9 @@ grpc::Status DatabaseGrpcImpl::Upsert(
     const pb::UpsertRequest* request,
     pb::UpsertResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("Upsert", service_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -333,6 +361,9 @@ grpc::Status DatabaseGrpcImpl::Delete(
     const pb::DeleteRequest* request,
     pb::DeleteResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("Delete", service_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -410,6 +441,11 @@ grpc::Status DatabaseGrpcImpl::RestoreVersion(
     const pb::RestoreVersionRequest* request,
     pb::RestoreVersionResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
     uint64_t newVersion = store_.restoreToVersion(
         request->collection(), request->id(),
         request->version(), request->actor());
@@ -430,6 +466,11 @@ grpc::Status DatabaseGrpcImpl::RestoreToDate(
     const pb::RestoreToDateRequest* request,
     pb::RestoreToDateResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
     auto [restoredVersion, newVersion] = store_.restoreToDate(
         request->collection(), request->id(),
         request->timestamp(), request->actor());
@@ -453,6 +494,9 @@ grpc::Status DatabaseGrpcImpl::BatchInsert(
     const pb::BatchInsertRequest* request,
     pb::BatchInsertResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("BatchInsert", service_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -518,6 +562,9 @@ grpc::Status DatabaseGrpcImpl::BatchDelete(
     const pb::BatchDeleteRequest* request,
     pb::BatchDeleteResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("BatchDelete", service_);
+    }
     if (view_manager_.isView(request->collection())) {
         return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
             "cannot write to view '" + request->collection() + "': views are read-only");
@@ -661,6 +708,9 @@ grpc::Status DatabaseGrpcImpl::SetAdd(
     const pb::SetAddRequest* request,
     pb::SetAddResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("SetAdd", service_);
+    }
     bool added = store_.setAdd(request->collection(), request->set_id(), request->member());
     response->set_added(added);
     return grpc::Status::OK;
@@ -671,6 +721,9 @@ grpc::Status DatabaseGrpcImpl::SetRemove(
     const pb::SetRemoveRequest* request,
     pb::SetRemoveResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("SetRemove", service_);
+    }
     bool removed = store_.setRemove(request->collection(), request->set_id(), request->member());
     response->set_removed(removed);
     return grpc::Status::OK;
@@ -705,6 +758,9 @@ grpc::Status DatabaseGrpcImpl::CreateCollection(
     const pb::CreateCollectionRequest* request,
     pb::CreateCollectionResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("CreateCollection", service_);
+    }
     CollectionOptions opts;
     // Proto3 defaults bools to false, but we want autoCreateId to default to true
     // Since proto3 can't distinguish "not set" from "false", we always default to true
@@ -733,6 +789,9 @@ grpc::Status DatabaseGrpcImpl::DropCollection(
     const pb::DropCollectionRequest* request,
     pb::DropCollectionResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("DropCollection", service_);
+    }
     bool dropped = store_.dropCollection(request->name());
     response->set_dropped(dropped);
 
@@ -779,6 +838,9 @@ grpc::Status DatabaseGrpcImpl::UploadFile(
     grpc::ServerReader<pb::FileChunk>* reader,
     pb::UploadFileResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("UploadFile", service_);
+    }
     pb::FileChunk chunk;
     std::vector<uint8_t> fileData;
     pb::FileMetadata metadata;
@@ -865,6 +927,9 @@ grpc::Status DatabaseGrpcImpl::DeleteFile(
     const pb::DeleteFileRequest* request,
     pb::DeleteFileResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        return readOnlyStatus("DeleteFile", service_);
+    }
     bool deleted = files_.deleteFile(request->id());
     response->set_deleted(deleted);
     return grpc::Status::OK;
@@ -1143,6 +1208,11 @@ grpc::Status DatabaseGrpcImpl::CreateView(
     const pb::CreateViewRequest* request,
     pb::CreateViewResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
     try {
         ViewInfo v;
         v.name = request->name();
@@ -1188,6 +1258,11 @@ grpc::Status DatabaseGrpcImpl::DropView(
     const pb::DropViewRequest* request,
     pb::DropViewResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
     try {
         std::string err;
         bool ok = view_manager_.dropView(request->name(), err);
@@ -1387,6 +1462,11 @@ grpc::Status DatabaseGrpcImpl::ConfigureCollection(
     const pb::ConfigureCollectionRequest* request,
     pb::ConfigureCollectionResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        return grpc::Status::OK;
+    }
     try {
         CollectionCfg cfg;
         cfg.timestampPrecision = request->config().timestamp_precision();
@@ -1423,6 +1503,13 @@ grpc::Status DatabaseGrpcImpl::MigrateCollectionTimestamps(
     const pb::MigrateCollectionTimestampsRequest* request,
     pb::MigrateCollectionTimestampsResponse* response
 ) {
+    if (service_.isReadOnly()) {
+        response->set_success(false);
+        response->set_error("database is in read-only mode: " + service_.readOnlyReason());
+        response->set_rows_migrated(0);
+        response->set_rows_skipped(0);
+        return grpc::Status::OK;
+    }
     try {
         const std::string& collection = request->collection();
         const std::string& from = request->from_precision();
@@ -1505,4 +1592,48 @@ grpc::Status DatabaseGrpcImpl::MigrateCollectionTimestamps(
     }
 }
 
+// ===== Read-Only Control =====
+
+grpc::Status DatabaseGrpcImpl::SetReadOnly(
+    grpc::ServerContext* /*context*/,
+    const pb::SetReadOnlyRequest* request,
+    pb::SetReadOnlyResponse* response
+) {
+    bool was_readonly = service_.isReadOnly();
+    std::string reason = request->read_only()
+        ? "manually locked via SetReadOnly RPC"
+        : "";
+    service_.setReadOnly(request->read_only(), reason);
+    response->set_success(true);
+    response->set_was_read_only(was_readonly);
+    return grpc::Status::OK;
+}
+
+grpc::Status DatabaseGrpcImpl::GetReadOnlyStatus(
+    grpc::ServerContext* /*context*/,
+    const pb::GetReadOnlyStatusRequest* /*request*/,
+    pb::GetReadOnlyStatusResponse* response
+) {
+    response->set_read_only(service_.isReadOnly());
+    response->set_reason(service_.readOnlyReason());
+
+    const auto& out = service_.recoveryOutcome();
+    const char* kindStr = "trivial_success";
+    switch (out.kind) {
+        case RecoveryOutcome::Kind::TrivialSuccess:   kindStr = "trivial_success"; break;
+        case RecoveryOutcome::Kind::FreshInstall:     kindStr = "fresh_install"; break;
+        case RecoveryOutcome::Kind::SnapshotFellBack: kindStr = "snapshot_fell_back"; break;
+        case RecoveryOutcome::Kind::WalOnlyReplay:    kindStr = "wal_only_replay"; break;
+        case RecoveryOutcome::Kind::ForcedEmpty:      kindStr = "forced_empty"; break;
+        case RecoveryOutcome::Kind::Failed:           kindStr = "failed"; break;
+    }
+    response->set_recovery_outcome(kindStr);
+    response->set_expected_snapshot(out.expectedSnapshot.string());
+    response->set_snapshot_used(out.snapshotUsed.string());
+    response->set_failure_reason(out.failureReason);
+    response->set_wal_entries_replayed(out.walEntriesReplayed);
+    response->set_snapshots_attempted(static_cast<uint32_t>(out.snapshotsAttempted));
+    return grpc::Status::OK;
+}
+
 } // namespace smartbotic::database

service/src/database_grpc_impl.hpp

@@ -21,12 +21,16 @@ namespace smartbotic::database {
 // Namespace alias for proto types
 namespace pb = smartbotic::databasepb;
 
+// Forward declaration (DatabaseService and DatabaseGrpcImpl have circular references).
+class DatabaseService;
+
 /**
  * gRPC service implementation for the Storage service.
  */
 class DatabaseGrpcImpl final : public pb::DatabaseService::Service {
 public:
     DatabaseGrpcImpl(
+        DatabaseService& service,
         MemoryStore& store,
         PersistenceManager& persistence,
         EventManager& events,
@@ -300,6 +304,20 @@ public:
         pb::GetStatsResponse* response
     ) override;
 
+    // ===== Read-Only Control =====
+
+    grpc::Status SetReadOnly(
+        grpc::ServerContext* context,
+        const pb::SetReadOnlyRequest* request,
+        pb::SetReadOnlyResponse* response
+    ) override;
+
+    grpc::Status GetReadOnlyStatus(
+        grpc::ServerContext* context,
+        const pb::GetReadOnlyStatusRequest* request,
+        pb::GetReadOnlyStatusResponse* response
+    ) override;
+
 private:
     // Convert internal types to proto types
     static pb::Document toProto(const Document& doc);
@@ -309,6 +327,7 @@ private:
     static std::vector<Filter> fromProtoFilters(const google::protobuf::RepeatedPtrField<pb::Filter>& filters);
     static Query fromProtoQuery(const pb::FindRequest& request);
 
+    DatabaseService& service_;
     MemoryStore& store_;
     PersistenceManager& persistence_;
     EventManager& events_;

service/src/database_service.cpp

@@ -525,7 +525,7 @@ void DatabaseService::setupComponents() {
 
     // Create gRPC implementations
     storageImpl_ = std::make_unique<DatabaseGrpcImpl>(
-        *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_
+        *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_
     );
     replicationImpl_ = std::make_unique<DatabaseReplicationGrpcImpl>(
         *store_, *replication_, *persistence_