riverctl: Add `riverctl dm stream` subcommand for real-time DM events

[ofansifkapital]

Problem

Bridge authors (XMPP, Matrix, Discord, etc.) cannot deliver DMs in
real-time through riverctl. The only mechanism to read DMs is
riverctl dm list — a blocking fetch-and-exit command. There is no
streaming equivalent of riverctl message stream for the
direct_messages sub-state.

This means any bridge built on riverctl must either:

• Poll dm list on a timer (wasteful, adds latency, complexity
  of task lifecycle management)
• Require the user to manually trigger a fetch (poor UX — the
  user must explicitly "check for DMs" instead of receiving them as
  they arrive)

Current workaround

The riverjabber XMPP gateway (a slidge-based bridge) currently
requires the user to run a check-dms chat command to fetch DMs on
demand. This works but is cumbersome — it feels like "checking your
inbox" rather than chat.

Proposed solution

A riverctl dm stream subcommand that outputs JSONL events when new
DMs arrive in a room, following the same pattern as riverctl message stream:

riverctl dm stream <ROOM_ID> [--subscribe] [--format json] [--initial-messages N]

Output format: JSONL, one event per DM, with fields matching the
existing dm list --format json output:

{
  "counterparty": "MAOKOLK5",
  "counterparty_nickname": "alice",
  "direction": "incoming",
  "body": "hello there",
  "purge_token": "aabbccdd11223344aabbccdd11223344",
  "timestamp": "2026-07-31T14:22:00+00:00",
  "timestamp_unix": 1785511320,
  "is_invitation": false
}

The --initial-messages N flag (matching message stream's
equivalent) would emit the last N DMs on startup, so bridges can
backfill recent history when they connect.

This is a CLI-only change — no protocol or node changes needed

The subscription mechanism in subscribe_and_stream()
(cli/src/api.rs) already receives UpdateNotification events for
the entire room contract state, which includes
direct_messages. On each notification, it re-fetches the full
ChatRoomStateV1 via get_room() — this state already contains
direct_messages. The current code simply processes only
recent_messages from that state and ignores direct_messages.

A dm stream command would use the same ContractRequest::Subscribe
mechanism, the same UpdateNotification handling, and the same
get_room() re-fetch — it would just process
room_state.direct_messages instead of
room_state.recent_messages. The node, the contract, and the
protocol are unaffected.

Impact

Every bridge built on riverctl would benefit. Without this, DMs on
any CLI-bridged client require manual fetches or polling. With it,
DMs flow naturally as they arrive — same experience as the River
webUI, which already gets real-time DM updates via the same
subscription mechanism.

References

• subscribe_and_stream() — cli/src/api.rs:5284 (subscribes to
  room contract, re-fetches full state on each UpdateNotification,
  currently processes only recent_messages)
• dm list JSON output — cli/src/commands/dm.rs:759-787
• direct_messages in ChatRoomStateV1 —
  common/src/room_state/direct_messages.rs

[sanity]

Issue Priority Assessment

Generated: 2026-08-14T01:24:22.128819418+00:00
Issue hash: b201923a

Value Assessment

Confirmed: message stream exists (cli/src/commands/message.rs) using api.subscribe_and_stream() + api.stream_messages(), but there is no dm stream subcommand — DmCommands only has Send/List/Invite/Purge/Accept. The issue's description of the gap and its proposed reuse of subscribe_and_stream() matches the current code exactly.

Relevance: Fully current. subscribe_and_stream() at cli/src/api.rs:5284 still re-fetches the whole ChatRoomStateV1 (including direct_messages) on every UpdateNotification, and message.rs's Stream variant is the only consumer — it only pulls recent_messages. DmCommands (cli/src/commands/dm.rs:31) has no Stream arm. Nothing about this has been fixed since the issue was filed; it's a real, still-open gap, not stale.

User impact: Narrow but real — this affects developers building bridges/integrations on riverctl (XMPP/Matrix/Discord gateways), not end users of the River UI directly (the UI already gets real-time DMs via its own subscription path). Today those bridge authors must poll or require manual "check-dms" triggers, which is a genuine UX regression for anyone using such a bridge, as documented by the real-world riverjabber project cited in the issue.

Strategic alignment: Moderate-positive. Growing a bridge/integration ecosystem around riverctl supports River's broader adoption goal, and this is exactly the kind of CLI ergonomics work that unlocks third-party clients without touching the protocol, contract, or delegate WASM (no migration risk, no wire-format risk — a "Skip"/"Light" review tier by the project's own risk-tiering rules).

Blocking/urgency: Not blocking core roadmap work, and no urgency — it's an additive CLI feature with an existing workaround (polling dm list). Reasonable priority as a "nice-to-have" community/ecosystem enabler, implementable in an afternoon by mirroring MessageCommands::Stream, but not time-sensitive.

Cost Assessment

I have enough to write the assessment.

Cost Assessment: Issue #579 — riverctl dm stream

Relevance: The issue's premise still holds. subscribe_and_stream() (cli/src/api.rs:5284) is exactly as described — it subscribes via ContractRequest::Subscribe, handles the SUBSCRIBE-ack race, drains queued UpdateNotifications, and on each notification re-fetches full state via get_room(). It currently only walks room_state.recent_messages (via emit_new_and_edited/emit_deletions/emit_reaction_changes), never room_state.direct_messages. This function has grown substantially since the issue was filed (SUBSCRIBE-ack race handling, reaction-change tracking, deletion suppression, private-room secret decryption via room_display_secrets) — none of that invalidates the approach, but it means a new dm stream mode can't just piggyback trivially; it has to either parallel this now-fairly-elaborate state machine or (better) factor the subscribe/notification-loop plumbing so both message and DM streaming share it, since duplicating ~250 lines of SUBSCRIBE-race/shutdown/timeout handling would be its own maintenance liability.

Scope: Primarily cli/src/api.rs (new streaming method or generalized existing one) and cli/src/commands/dm.rs (new Stream variant in DmCommands, wired in cli/src/main.rs alongside the existing Dm command dispatch, following the pattern already used for MessageCommands::Stream in cli/src/commands/message.rs:44,411). No changes needed to common/, contracts, delegates, or the protocol — this is CLI-only as the issue states. Realistically 2-3 files touched, with the bulk of new logic being a "seen DM" tracking structure analogous to seen_messages/deleted_emitted/seen_reactions, but keyed on (counterparty, purge_token) or similar rather than message id.

Complexity: Moderate, not high. The tricky parts: (1) determining "new since last check" for DMs — unlike recent_messages, direct_messages state doesn't have an obvious ordering/id scheme surfaced yet in the CLI, so the new code needs to build a seen-set keyed by sender+recipient+purge_token+timestamp, decrypt via open_direct_message (ECIES, feature-gated), and diff against previous state on every notification, mirroring execute in dm.rs's List path (cli/src/commands/dm.rs:759+) which already does the decrypt-and-group logic and is the natural extraction target for a shared helper; (2) purge/tombstone handling — DMs can be purged (advance_recipient_purges), so the stream needs to decide whether purged DMs should be suppressed like deleted messages are (deleted_emitted analog), which the issue's spec doesn't address explicitly; (3) private-room secret availability timing, same as the existing message stream already handles via room_display_secrets. None of this is architecturally novel — it's essentially "do what dm list --format json does, but incrementally, driven by the same UpdateNotification loop subscribe_and_stream uses."

External dependencies: None. No third-party signups, no protocol/wire changes, no coordination needed — confirmed by the issue itself and by inspection.

Risk: Low to existing behavior since this is purely additive (new subcommand, no changes to existing message-stream code paths) unless the implementer chooses to refactor subscribe_and_stream into a shared core to avoid duplication, which would touch code with known historical fragility (the SUBSCRIBE-ack race fixed after a documented production incident, freenet-core#4970). That refactor is the main regression risk vector — any change to the shared subscribe/reconnect logic needs to preserve the pending-queue draining semantics carefully.

Testing effort: Needs both a fresh CLI integration test (spin up two identities, send a DM, assert the streaming command emits it, similar to existing message stream tests if present) and coverage for edge cases: purge/tombstone suppression, private-room decryption-not-yet-available, and --initial-messages backfill semantics matching dm list. Given the CLI's existing test patterns for dm and message stream, this is straightforward but non-trivial — probably a half-day of test-writing to match the bar set by the surrounding code's pinned-behavior test style (many _pinned/_wiring_pinned tests exist across this codebase as a convention).

Overall: Small-to-medium cost, a well-scoped, self-contained CLI feature with no external blockers. Most of the effort is in careful state-diffing (seen/purged tracking) and matching the JSON schema exactly to dm list, not in novel architecture.

---

Generated by issue-prioritizer