feat(chat): reduce on-chain Control memos into per-conversation hide state

ingest() now routes Control memos (self-addressed, so processed ahead of the
"skip our own memos" guard) into a per-cid reducer that keeps the latest op by
(compose-time, txid) — monotonic, so a replayed older op never flips state. A
conversation whose latest op is "del" is hidden: its messages are dropped from
ingest and erased from the store view (even rows loaded from the DB before the
signal was reduced), on every device and on a fresh seed import, until a later
"res" restores it. The DB rows are left intact so restore (or a rescan) brings
the thread back. State is re-derived from chain each session; no local
persistence. Adds conversationHiddenOnChain() + onChainDeletedConversations()
(the restore picker's source), fully reducer-tested.

Phase A, step 3 (reducer) of on-chain chat delete/restore.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/chat/chat_service.cpp

@@ -40,6 +40,16 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
     int added = 0;
     std::vector<ChatMessage> fresh;   // persisted once after the loop (one transaction)
     for (const auto& meta : metadata) {
+        // A self-addressed control signal (delete / restore / ...). Reduce it and move on: it is never a
+        // message, and it must be processed even though WE are its sender — unlike ordinary self-sent memos,
+        // which the guard below skips. The label (peer hint) rides the payload.
+        if (meta.type == HushChatHeaderType::Control) {
+            const auto ctlTimeIt = txTimestamps.find(meta.txid);
+            const std::int64_t ctlRef = ctlTimeIt != txTimestamps.end() ? ctlTimeIt->second : fallbackTimestamp;
+            applyControl(meta.conversation_id, meta.op, meta.payload_memo,
+                         meta.sent_at > 0 ? meta.sent_at : ctlRef, meta.txid);
+            continue;
+        }
         // Already stored: append would refuse it. Every scan re-sends the whole history, so this skips the
         // decrypt for all but the new messages.
         if (store_.contains(meta.txid, meta.payload_position)) continue;
@@ -77,6 +87,7 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
         // deleted with "revive on new message", so only that exact message is skipped — a new message
         // in the same conversation has a different txid and flows through normally.
         if (blocked_pred_ && blocked_pred_(message.conversation_id)) continue;
+        if (conversationHiddenOnChain(message.conversation_id)) continue;  // hidden by an on-chain delete
         if (db_ && db_->isTombstoned(message)) continue;
 
         if (meta.type == HushChatHeaderType::ContactRequest) {
@@ -116,6 +127,13 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
         }
     }
     if (db_ && !fresh.empty()) db_->appendBatch(fresh);
+    // Enforce on-chain deletions against the store view: erase any conversation whose latest control op is
+    // "del". This catches rows appended earlier in this very batch (before their delete signal was seen in
+    // the metadata order) and rows rehydrated from the DB at startup before any control was reduced. The DB
+    // rows are left intact so a later "res" — or a rescan — can restore the thread.
+    for (const auto& entry : control_) {
+        if (entry.second.op == "del") store_.eraseConversation(entry.first);
+    }
     return added;
 }
 
@@ -146,6 +164,33 @@ bool ChatService::deleteConversation(const std::string& conversationId, bool blo
     return true;
 }
 
+void ChatService::applyControl(const std::string& cid, const std::string& op, const std::string& label,
+                               std::int64_t timestamp, const std::string& txid) {
+    auto it = control_.find(cid);
+    if (it != control_.end()) {
+        const ControlState& cur = it->second;
+        // A later signal wins; equal timestamps break by txid so the reduction is deterministic and a
+        // replay of an older op can never flip the resolved state.
+        if (timestamp < cur.timestamp || (timestamp == cur.timestamp && txid <= cur.txid)) return;
+        it->second = ControlState{op, label, timestamp, txid};
+    } else {
+        control_.emplace(cid, ControlState{op, label, timestamp, txid});
+    }
+}
+
+bool ChatService::conversationHiddenOnChain(const std::string& conversationId) const {
+    auto it = control_.find(conversationId);
+    return it != control_.end() && it->second.op == "del";
+}
+
+std::vector<std::pair<std::string, std::string>> ChatService::onChainDeletedConversations() const {
+    std::vector<std::pair<std::string, std::string>> out;
+    for (const auto& entry : control_) {
+        if (entry.second.op == "del") out.emplace_back(entry.first, entry.second.label);
+    }
+    return out;
+}
+
 std::string ChatService::identityPublicKeyHex() const {
     if (!has_identity_) return {};
     return chatIdentityPublicKeyHex(identity_);

src/chat/chat_service.h

@@ -15,6 +15,7 @@
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
+#include <utility>
 #include <vector>
 
 namespace dragonx::chat {
@@ -70,6 +71,16 @@ public:
     // persisted rows couldn't be removed, so the caller can avoid a store/DB divergence.
     bool deleteConversation(const std::string& conversationId, bool block);
 
+    // On-chain control signals (self-addressed Control memos) reduced per conversation: the latest op,
+    // by sender compose time with a txid tiebreak, wins. A conversation whose latest op is "del" is hidden
+    // on-chain — ingest() drops its messages and erases it from the store view, on every device and on a
+    // fresh seed import — until a later "res" restores it. The state is re-derived from chain each session
+    // (the control memos re-harvest on every scan), so it needs no local persistence.
+    bool conversationHiddenOnChain(const std::string& conversationId) const;
+    // Conversations whose latest on-chain control op is "del" (for the restore picker), each paired with
+    // the label carried in its control memo (the peer address / display name at delete time).
+    std::vector<std::pair<std::string, std::string>> onChainDeletedConversations() const;
+
     // --- Outgoing (compose) ---
     // My chat public key (hex), or "" without an identity — goes in an outgoing header's "p".
     std::string identityPublicKeyHex() const;
@@ -112,6 +123,18 @@ private:
     ChatStore store_;
     ChatDatabase* db_ = nullptr;  // optional; not owned
     std::function<bool(const std::string&)> blocked_pred_;  // true => cid is blocked (drop its messages)
+
+    // Reduced on-chain control state, cid -> latest signal. Not persisted (re-derived from chain).
+    struct ControlState {
+        std::string op;            // "del" / "res" / future ops
+        std::string label;         // the control memo's payload hint (peer address / display name)
+        std::int64_t timestamp = 0;
+        std::string txid;
+    };
+    std::unordered_map<std::string, ControlState> control_;
+    // Monotonic reduce: a later (timestamp, txid) wins; a replay of an older op never flips the state.
+    void applyControl(const std::string& cid, const std::string& op, const std::string& label,
+                      std::int64_t timestamp, const std::string& txid);
 };
 
 } // namespace dragonx::chat