diff --git a/src/chat/chat_service.cpp b/src/chat/chat_service.cpp index 5502f9b..83c6170 100644 --- a/src/chat/chat_service.cpp +++ b/src/chat/chat_service.cpp @@ -39,25 +39,14 @@ int ChatService::ingest(const std::vector& metadata int added = 0; std::vector 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; - } + + // Build, decrypt (Message) and store one piece of message metadata — a single-transaction message, + // or a tier-3 message reassembled from its slices below. Shared so both apply the same guards exactly. + auto processMessage = [&](const HushChatTransactionMetadata& meta) { // 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; - // A memo whose sender is our OWN identity is something we sent (only we hold our key). The local - // echo already records it as outgoing — ingesting it as incoming would duplicate it as a phantom - // "from peer" message. (This also collapses same-seed self-chat, where the "peer" wallet shares - // our identity, so every message would otherwise loop back.) - if (!myPubKey.empty() && meta.sender_public_key_hex == myPubKey) continue; + // decrypt for all but the new messages. For a reassembled tier-3 message this keys off its stable + // canonical txid, deduping it across scans. + if (store_.contains(meta.txid, meta.payload_position)) return; ChatMessage message; message.direction = ChatDirection::Incoming; @@ -86,13 +75,13 @@ int ChatService::ingest(const std::vector& metadata // is dropped outright (old + future messages) until unblocked; a tombstoned (txid,position) was // 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 (blocked_pred_ && blocked_pred_(message.conversation_id)) return; + if (conversationHiddenOnChain(message.conversation_id)) return; // hidden by an on-chain delete // Withhold a sensitive conversation's messages while its local vault key is absent (locked vault or // a fresh seed import). Fail-closed: if we can't store it under the vault key we don't surface it. // They re-import and persist under the vault key once it is unlocked and a scan runs. - if (isConversationSensitive(message.conversation_id) && !(db_ && db_->hasSensitiveKey())) continue; - if (db_ && db_->isTombstoned(message)) continue; + if (isConversationSensitive(message.conversation_id) && !(db_ && db_->hasSensitiveKey())) return; + if (db_ && db_->isTombstoned(message)) return; if (meta.type == HushChatHeaderType::ContactRequest) { message.kind = ChatMessageKind::ContactRequest; @@ -103,7 +92,7 @@ int ChatService::ingest(const std::vector& metadata // different header with this payload gets its own try. const std::string failKey = ChatStore::messageKey(meta.txid, meta.payload_position) + "|" + meta.sender_public_key_hex + "|" + meta.secretstream_header_hex; - if (undecryptable_.count(failKey)) continue; + if (undecryptable_.count(failKey)) return; ++decrypt_attempts_; std::string plaintext; const ChatCryptoStatus status = decryptIncoming( @@ -114,7 +103,7 @@ int ChatService::ingest(const std::vector& metadata if (undecryptable_.size() >= 4096) undecryptable_.clear(); undecryptable_.insert(failKey); } - continue; + return; } message.body = std::move(plaintext); } @@ -129,7 +118,84 @@ int ChatService::ingest(const std::vector& metadata // relying on a seen-watermark delta (which block-time vs wall-clock skew can swallow). if (newIncomingCids) newIncomingCids->push_back(message.conversation_id); } + }; + + 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; + } + // A memo whose sender is our OWN identity is something we sent (only we hold our key). The local + // echo already records it as outgoing — ingesting it as incoming would duplicate it as a phantom + // "from peer" message (this also collapses same-seed self-chat). Skips our own tier-3 slices too. + if (!myPubKey.empty() && meta.sender_public_key_hex == myPubKey) continue; + + // Larger-messages tier 3: a slice of a multi-transaction message. Accumulate it by message id; + // the whole message is reassembled below once every chunk has arrived (across txs and scans). + if (!meta.message_id.empty()) { + if (meta.total_parts <= 0 || meta.total_parts > kHushChatMaxMessageChunks) continue; // sanity + // Bound the buffer against a flood of never-completing partials (re-fed from chain, so a clear + // only delays, never loses, a genuine message). + if (tier3_pending_.size() > 256 && !tier3_pending_.count(meta.message_id)) tier3_pending_.clear(); + Tier3Pending& g = tier3_pending_[meta.message_id]; + if (g.chunks.empty()) { // first slice seen: capture the shared header fields + g.total_parts = meta.total_parts; + g.secretstream_header_hex = meta.secretstream_header_hex; + g.sender_public_key_hex = meta.sender_public_key_hex; + g.conversation_id = meta.conversation_id; + g.reply_zaddr = meta.reply_zaddr; + g.sent_at = meta.sent_at; + g.canonical_txid = meta.txid; + } + if (meta.txid < g.canonical_txid) g.canonical_txid = meta.txid; // deterministic stable key + for (const auto& c : meta.partial_chunks) + if (c.first >= 0 && c.first < g.total_parts) g.chunks[c.first] = c.second; + continue; + } + + processMessage(meta); } + + // Finalize every tier-3 message whose chunks have all arrived: reassemble the one ciphertext and run + // it through the shared message path (decrypt + the block/hide/sensitive guards). Retire the group + // afterwards unless it was withheld only because the sensitive vault is still locked. + { + std::vector retire; + for (auto& entry : tier3_pending_) { + Tier3Pending& g = entry.second; + if (g.total_parts <= 0 || static_cast(g.chunks.size()) != g.total_parts) continue; + bool complete = true; + for (int i = 0; i < g.total_parts; ++i) if (!g.chunks.count(i)) { complete = false; break; } + if (!complete) continue; + if (store_.contains(g.canonical_txid, 0)) { retire.push_back(entry.first); continue; } // already have it + + HushChatTransactionMetadata synth; + synth.type = HushChatHeaderType::Message; + synth.txid = g.canonical_txid; + synth.payload_position = 0; + synth.conversation_id = g.conversation_id; + synth.reply_zaddr = g.reply_zaddr; + synth.sender_public_key_hex = g.sender_public_key_hex; + synth.secretstream_header_hex = g.secretstream_header_hex; + synth.sent_at = g.sent_at; + std::string full; + for (int i = 0; i < g.total_parts; ++i) full += g.chunks[i]; + synth.payload_memo = std::move(full); + processMessage(synth); + + const bool withheldSensitive = + isConversationSensitive(g.conversation_id) && !(db_ && db_->hasSensitiveKey()); + if (!withheldSensitive) retire.push_back(entry.first); + } + for (const auto& mid : retire) tier3_pending_.erase(mid); + } + 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 diff --git a/src/chat/chat_service.h b/src/chat/chat_service.h index 1a78f77..ba0090e 100644 --- a/src/chat/chat_service.h +++ b/src/chat/chat_service.h @@ -153,6 +153,22 @@ private: }; std::unordered_map control_; // hide axis: del / res std::unordered_map sensitive_control_; // sensitive axis: sensitive / unsensitive + + // Larger-messages tier 3: a big message split across several transactions arrives one slice at a + // time (siblings share a message id). We accumulate the slices here until every chunk is present, + // then reassemble + decrypt + emit the whole message. Not persisted — a rescan re-sends every slice + // so the buffer re-derives from chain (like the control reducer); a completed group is retired. + struct Tier3Pending { + int total_parts = 0; + std::string secretstream_header_hex; // "e", shared by every slice + std::string sender_public_key_hex; // "p", shared + std::string conversation_id; + std::string reply_zaddr; + std::int64_t sent_at = 0; + std::string canonical_txid; // smallest sibling txid — the reassembled message's stable key + std::unordered_map chunks; // global index -> ciphertext-hex chunk + }; + std::unordered_map tier3_pending_; // message id -> accumulating slices // Monotonic reduce: route the op to its axis (hide vs sensitive), ignore unknown ops (forward-compat), // and within an axis a later (timestamp, txid) wins so a replay of an older op never flips the state. void applyControl(const std::string& cid, const std::string& op, const std::string& label, diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index df7306a..a5e6648 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -15827,6 +15827,65 @@ void testHushChatLargeMessageTier3() cid, tooHuge, over) == ChatComposeStatus::TooLong); } +// Tier 3 through the ChatService: slices arriving across separate ingests (separate scans) accumulate +// until the message is whole, then it surfaces once; a re-scan of every slice dedups it. +void testHushChatTier3Reassembly() +{ + using namespace dragonx::chat; + ChatKeyPair alice, bob; + ChatIdentityResult ra = deriveChatIdentityFromSecret("t3svc-alice", alice, true); + ChatIdentityResult rb = deriveChatIdentityFromSecret("t3svc-bob", bob, true); + EXPECT_TRUE(ra.status == ChatIdentityStatus::Ready && rb.status == ChatIdentityStatus::Ready); + const std::string cid = "conv-t3svc"; + + std::string big; + for (int i = 0; i < 9000; ++i) big += static_cast('a' + (i % 26)); + + std::vector txs; + EXPECT_TRUE(splitOutgoingMessage(alice, ra.public_key_hex, "zs-alice", rb.public_key_hex, "zs-bob", + cid, big, txs) == ChatComposeStatus::Ok); + EXPECT_TRUE((int)txs.size() >= 3); + + // Extract each transaction into partial metadata (outputs reversed to mimic the daemon shuffle). + std::vector all; + for (std::size_t t = 0; t < txs.size(); ++t) { + std::vector wire{ txs[t].headerMemo }; + for (const auto& pm : txs[t].payloadMemos) wire.push_back(pm); + std::reverse(wire.begin(), wire.end()); + HushChatTransactionInput tx; tx.txid = "t3svc-" + std::to_string(t); + for (std::size_t i = 0; i < wire.size(); ++i) tx.outputs.push_back({i, wire[i]}); + auto md = extractHushChatTransactionMetadata(tx, true).metadata; + for (auto& m : md) all.push_back(std::move(m)); + } + EXPECT_TRUE(all.size() == txs.size()); // one partial metadata per transaction + + ChatService svc; + svc.setIdentity(bob); + + // Stage 1: only the FIRST transaction's slice has arrived — the message is incomplete, nothing surfaces. + std::vector firstOnly(all.begin(), all.begin() + 1); + EXPECT_EQ(svc.ingest(firstOnly, {}, 1000), 0); + EXPECT_EQ((int)svc.store().conversation(cid).size(), 0); + + // Stage 2: the whole history re-arrives (every slice) — reassembled, decrypted and surfaced once. + EXPECT_EQ(svc.ingest(all, {}, 1000), 1); + std::vector conv = svc.store().conversation(cid); + EXPECT_EQ((int)conv.size(), 1); + EXPECT_EQ(conv[0].body, big); + EXPECT_TRUE(conv[0].direction == ChatDirection::Incoming); + EXPECT_EQ(conv[0].peer_public_key_hex, ra.public_key_hex); + + // A re-scan of the same slices does not duplicate the message. + EXPECT_EQ(svc.ingest(all, {}, 1000), 0); + EXPECT_EQ((int)svc.store().conversation(cid).size(), 1); + + // Our OWN tier-3 slices (sender == us) are never ingested as incoming — the echo records them. + ChatService selfSvc; + selfSvc.setIdentity(alice); + EXPECT_EQ(selfSvc.ingest(all, {}, 1000), 0); + EXPECT_EQ((int)selfSvc.store().conversation(cid).size(), 0); +} + void testHushChatOutgoing() { using namespace dragonx::chat; @@ -20438,6 +20497,7 @@ int main() testHushChatOutgoing(); testHushChatLargeMessage(); testHushChatLargeMessageTier3(); + testHushChatTier3Reassembly(); testHushChatTransport(); testHushChatShuffledReceive(); testWalletFileProbe();