Author SHA1 Message Date
mollusk adf7d1c0e2 Fix inline audio duration detection 2026-06-21 01:41:57 -04:00
mollusk bcb597a0ea Add inline chat audio player 2026-06-21 01:05:24 -04:00
molluskandClaude Opus 4.8 79b24fd567 deps: add rodio for inline chat audio playback (senior-vetted)
Pre-stages the playback dependency for the inline chat audio player task so
Codex can build it in its network-off sandbox.

rodio 0.22.2 decodes wav/mp3/ogg(vorbis)/flac (via bundled symphonia) and
handles output + play/pause/seek + resampling. It brings its own cpal 0.17
(the project's PipeWire/cpal-0.15 call path is untouched; rodio's output is a
separate stream on the system default device) and alsa on Linux.

Supply chain: cargo audit reports NO new advisories from this subtree -- the
only 2 warnings (audiopus_sys, paste) are pre-existing, unmaintained-only, and
already on the allow-list. Builds clean (release).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-21 00:52:55 -04:00
molluskandClaude Opus 4.8 efadc228eb fix(files): keep serve connection alive until fetcher has the bytes
cargo-deny / cargo-deny (push) Has been cancelled
windows-build / windows-build (push) Has been cancelled
Chat file fetches failed 100% of the time with "file fetch: read failed:
read error: connection lost" (both images and arbitrary files, both
directions). Root cause: the FileRouter serve handler called send.finish()
and immediately returned Ok(()), which dropped the Connection. In QUIC,
finish() only marks the stream's EOF -- it does not wait for the written
bytes to be delivered and acknowledged -- so the connection's
CONNECTION_CLOSE raced ahead of the still-in-flight stream data and the
fetcher's read_to_end aborted.

Fix: after finishing, wait on connection.closed() (bounded by
FILE_FETCH_TIMEOUT) so the link stays up until the fetcher has read
everything and closed the connection itself, which is the signal the
transfer landed.

Wire-compatible (no protocol change), so version stays 0.3.0; both peers
just need the rebuilt binary since either side can be the file server.

Adds tests/file_transfer_loopback.rs: a real two-endpoint serve->fetch
round-trip over FILES_ALPN with a 2 MiB multi-packet blob (deterministic
A/B: 0/20 pass without the fix, 20/20 with it) plus an unknown-id "gone"
case.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-21 00:07:23 -04:00
molluskandClaude Opus 4.8 2d067a2e41 packaging(win): update INSTALL.md + README.md for 0.3.0
cargo-deny / cargo-deny (push) Has been cancelled
windows-build / windows-build (push) Has been cancelled
- INSTALL.md: bump the setup filename to 0.3.0; add an end-user section
  on text chat + sending photos/files (inline images, file chips,
  Save/Download, 25 MB cap, session-only); note that both ends must run
  the same version under "won't connect".
- README.md: add a Version compatibility section (installer version
  tracks Cargo; a 0.x MINOR bump is a breaking wire change so everyone
  must reinstall; 0.3.0 can't talk to 0.2.x).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 23:41:22 -04:00
molluskandClaude Opus 4.8 8ea40f719c packaging(win): bump installer version to 0.3.0
cargo-deny / cargo-deny (push) Has been cancelled
windows-build / windows-build (push) Has been cancelled
Match the Cargo 0.3.0 release (chat file sharing + per-peer gate). The
installer payload is unchanged (single self-contained peerspeak.exe +
icon); only the version string / output filename move to 0.3.0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 23:39:30 -04:00
molluskandClaude Opus 4.8 60c1951567 Chat file attachments, stages 3-4: core wiring + chat UI
cargo-deny / cargo-deny (push) Has been cancelled
windows-build / windows-build (push) Has been cancelled
Wire the send/receive paths and the chat UI on top of the file plane.
(Committed together because the UI renders the state the core wiring
produces.)

Core:
- CoreCommand::SendChatFile {text, attachment, data}: serve the bytes on
  the file plane (serve_attachment) then broadcast the descriptor via
  send_chat. CoreCommand::FetchAttachment {from, attachment}: detached
  fetch -> AttachmentReady/AttachmentFailed.
- On an inbound Chat with an Image attachment, auto-fetch + defensively
  re-validate (decodable + within pixel limits) before delivering;
  non-images wait for an explicit fetch (the Save/Download chip).
- UiEvent::ChatMessage carries the attachment; new AttachmentReady /
  AttachmentFailed events keyed by attachment id.

App:
- 📎 attach button + native picker; reads the file, enforces the size
  cap, classifies image vs file, mints a random id, optimistically
  echoes the message + caches our own bytes (so we see our own image
  inline), and sends SendChatFile.
- Renders inline image thumbnails (handle cached by id to avoid the
  per-redraw re-upload flicker), file chips with Save/Download, a
  loading placeholder for in-flight images, and an error line on
  failure. Image messages with no caption still render.
- SaveAttachment: saves immediately if bytes are in hand, else fetches
  then saves when ready (pending_saves) via a native save dialog;
  filename defaulted from the sanitized descriptor.
- Session-only: attachment bytes/handles cleared on leave, never
  persisted.

Binary + clippy clean, 349 lib tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 23:15:06 -04:00
molluskandClaude Opus 4.8 f75760b14e Chat file attachments, stage 2: file-transfer network plane
Add the dedicated FILES_ALPN data plane that moves attachment bytes
off-gossip via direct QUIC streams.

- FileRouter (ProtocolHandler on the persistent router, mirroring
  AudioRouter): bound to the active session's Shared on join, cleared on
  leave. On an inbound stream it authenticates the peer via the ALPN
  handshake, gates on live room membership (reuses audio_sender_admitted,
  so a former member cannot pull files), reads exactly one 32-byte
  attachment id (bounded request read), and streams back the matching
  blob from the session serve store — or an empty body for an unknown id.
- Shared gains served_files (id -> bytes), populated by serve_attachment
  and cleared on leave.
- IrohTransport::serve_attachment + fetch_attachment (inherent methods;
  transport is used concretely). fetch dials the sender on FILES_ALPN
  (preferring a known full address), writes the id, and reads bounded by
  the descriptor's declared size, with a 30s connect/read timeout so a
  stalled sender can't hang the fetch.
- Register FILES_ALPN in the router; bind/clear file_router in lock-step
  with audio_router at every join/leave site.

Builds + clippy clean, 349 lib tests pass (plane is runtime I/O,
field-tested in stage 5).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 23:06:12 -04:00
molluskandClaude Opus 4.8 02cb46550e Chat file attachments, stage 1: protocol + data model + pure seams
First slice of in-chat file/photo sharing (dedicated file plane, images
inline + file chips, session-only). This stage adds the wire types and
the pure, unit-tested logic; no transport or UI yet.

- protocol: new FILES_ALPN / FILES_PROTO (peerspeak/files/1) for the
  dedicated file-transfer plane. Bump GOSSIP_PROTO 1->2 + sig domain v2
  (Chat gained an attachment field, so cross-version peers fail fast
  rather than half-work) and Cargo 0.2.0 -> 0.3.0 per VERSIONING.md.
  BREAKING wire change: all peers must run >= 0.3.0.
- new src/files.rs: ChatAttachment descriptor (name/size/kind/id; bytes
  travel off-gossip), AttachmentKind, plus pure seams — sanitize_filename
  (path-traversal/control-char/length-safe), size_within_cap, image
  magic-byte sniffing + defensive limited decode (decode-bomb guard),
  32-byte request parsing, human_size. 13 unit tests.
- GossipMessage::Chat and RoomEvent::ChatMessage carry an optional
  ChatAttachment; send_chat takes Option<ChatAttachment>. Untrusted
  inbound descriptors are filename-sanitized + size-validated on ingest.
  serde(default) keeps the field forward-compatible at the JSON layer;
  +round-trip and pre-v2 back-compat tests.

The attachment id is a random 32-byte handle (rand, already a dep), not
a content hash — the fetch is authenticated + encrypted + member-gated,
so no crypto-hash dep is needed.

349 lib tests pass, clippy clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 23:00:51 -04:00
molluskandClaude Opus 4.8 cbba4b644e Add per-peer listener-side noise gate
cargo-deny / cargo-deny (push) Has been cancelled
windows-build / windows-build (push) Has been cancelled
Let a listener apply a noise gate to an individual peer's incoming
audio — "fix this person's noisy mic / background hum on my end" —
which is only possible because full-mesh P2P keeps every peer's stream
unmixed locally (server-mixed apps can't do per-listener per-peer DSP).

The DSP is the existing mic NoiseGate reused verbatim: it already
processes i16 frames at a fixed rate with hysteresis/attack/release/
hangover and takes the threshold per-frame. Wiring mirrors per-peer EQ:
- AppConfig.peer_gate map (threshold per peer id; absent/0 = off),
  persisted, never sent over the wire
- CoreCommand::SetPeerGate + Arc<Mutex<HashMap>> shared into the mixer
- a live HashMap<EndpointId, NoiseGate> in the mixer task, created
  lazily and dropped when disabled (no rebuild needed — threshold is
  passed per frame)
- Gate row (threshold slider, "Off" at zero) in each participant card
  next to Vol/Pan/EQ, persisting on release

The gate runs on the raw decoded frame: after the clean multitrack stem
tap (recordings stay ungated) but before volume/EQ, so the threshold
tracks the peer's true signal level regardless of our volume setting.
Same 0..METER_MAX scale as the mic gate.

+2 unit tests (config helper); +1 config back-compat assertion.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 22:18:50 -04:00
molluskandClaude Opus 4.8 c5375e200a Make per-peer in-call volume continuous and persistent
cargo-deny / cargo-deny (push) Has been cancelled
windows-build / windows-build (push) Has been cancelled
The per-peer volume slider had no .step(), so iced's default step of
1.0 on a 0.0..=2.0 range meant it could only snap to 0%, 100%, or
200% — it felt like hard-left/hard-right only. Add .step(0.01) for
smooth 1%-increment control (matching the Pan slider below it, which
already set its own step).

Also persist per-peer volume across sessions, mirroring peer_pan/peer_eq:
- new AppConfig.peer_volume map (keyed by peer id string, serde default
  for back-compat; never sent over the wire)
- replace the in-memory peer_volumes map with config-backed storage via
  a new set_peer_volume_config helper (clamps to range, drops at-unity
  entries so the config stays tidy)
- replay saved volumes to core on startup alongside pan/eq
- the slider writes to disk on release (AppMessage::PersistConfig)

+1 unit test for the config helper; +1 config back-compat assertion.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-20 21:29:18 -04:00
20 changed files with 2169 additions and 43 deletions
+21
View File
@@ -0,0 +1,21 @@
# PeerSpeak Codebase Layout and Architecture Rules
When working in the PeerSpeak repository, adhere to the following architectural boundaries and layout:
## Code Layout
- `src/main.rs`: The application entry point (initializes Tokio and the Iced GUI).
- `src/app/`: The UI layer (Iced). Handles themes, views (Home, Room, Settings), and visual state. Must communicate with the core via message passing (`UiEvent`/`CoreCommand`), not direct function calls.
- `src/core/`: The central orchestrator.
- `mod.rs`: Manages the session lifecycle, ties together network and UI, and manages the async mixer tasks.
- `jitter.rs`: Houses the adaptive playout delay JitterBuffer and Packet Loss Concealment (PLC) logic.
- `src/network/`: The "Dual-Plane" transport layer.
- `gossip.rs` (Control Plane): Built on `iroh-gossip`. Manages room rosters, verified membership, presence, and chat via cryptographically signed envelopes.
- `iroh_impl.rs` (Data Plane): Manages raw QUIC endpoints and peer connections. Forwards UDP voice datagrams directly to peers for minimum latency.
- `src/audio/`: Hardware audio backends.
- Interfaces heavily with `cpal_impl.rs` (Windows/WASAPI) and `pipewire_impl.rs` (Linux).
- **CRITICAL RULE**: The RT audio callbacks are strictly lock-free. They communicate with the async core exclusively via Single-Producer Single-Consumer (SPSC) ring buffers (`HeapRb`). Never allocate memory, log to stdout, or lock Mutexes on the RT threads.
- `src/codec/`: Audio compression abstractions, standardizing on Opus at 48kHz mono (`opus_impl.rs`).
## General Directives
- **Security**: Audio admission is strictly derived from the verified gossip roster (S8). Never trust raw UDP sender IDs without validating against gossip.
- **Latency**: Preserve the deterministic dialer vs acceptor logic in the QUIC layer to prevent connection loops.
Generated
+314 -5
View File
@@ -117,6 +117,18 @@ dependencies = [
"libc", "libc",
] ]
[[package]]
name = "alsa"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7c88dbbce13b232b26250e1e2e6ac18b6a891a646b8148285036ebce260ac5c3"
dependencies = [
"alsa-sys",
"bitflags 2.11.1",
"cfg-if",
"libc",
]
[[package]] [[package]]
name = "alsa-sys" name = "alsa-sys"
version = "0.3.1" version = "0.3.1"
@@ -1041,6 +1053,20 @@ dependencies = [
"coreaudio-sys", "coreaudio-sys",
] ]
[[package]]
name = "coreaudio-rs"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1aae284fbaf7d27aa0e292f7677dfbe26503b0d555026f702940805a630eac17"
dependencies = [
"bitflags 1.3.2",
"libc",
"objc2-audio-toolbox",
"objc2-core-audio",
"objc2-core-audio-types",
"objc2-core-foundation",
]
[[package]] [[package]]
name = "coreaudio-sys" name = "coreaudio-sys"
version = "0.2.18" version = "0.2.18"
@@ -1080,14 +1106,14 @@ version = "0.15.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "873dab07c8f743075e57f524c583985fbaf745602acbe916a01539364369a779" checksum = "873dab07c8f743075e57f524c583985fbaf745602acbe916a01539364369a779"
dependencies = [ dependencies = [
"alsa", "alsa 0.9.1",
"core-foundation-sys", "core-foundation-sys",
"coreaudio-rs", "coreaudio-rs 0.11.3",
"dasp_sample", "dasp_sample",
"jni 0.21.1", "jni 0.21.1",
"js-sys", "js-sys",
"libc", "libc",
"mach2", "mach2 0.4.3",
"ndk 0.8.0", "ndk 0.8.0",
"ndk-context", "ndk-context",
"oboe", "oboe",
@@ -1097,6 +1123,36 @@ dependencies = [
"windows 0.54.0", "windows 0.54.0",
] ]
[[package]]
name = "cpal"
version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b1f9c7312f19fc2fa12fd7acaf38de54e8320ba10d1a02dcbe21038def51ccb"
dependencies = [
"alsa 0.10.0",
"coreaudio-rs 0.13.0",
"dasp_sample",
"jni 0.21.1",
"js-sys",
"libc",
"mach2 0.5.0",
"ndk 0.9.0",
"ndk-context",
"num-derive",
"num-traits",
"objc2 0.6.4",
"objc2-audio-toolbox",
"objc2-avf-audio",
"objc2-core-audio",
"objc2-core-audio-types",
"objc2-core-foundation",
"objc2-foundation 0.3.2",
"wasm-bindgen",
"wasm-bindgen-futures",
"web-sys",
"windows 0.62.2",
]
[[package]] [[package]]
name = "cpufeatures" name = "cpufeatures"
version = "0.2.17" version = "0.2.17"
@@ -1564,6 +1620,15 @@ version = "0.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d"
[[package]]
name = "encoding_rs"
version = "0.8.35"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3"
dependencies = [
"cfg-if",
]
[[package]] [[package]]
name = "endi" name = "endi"
version = "1.1.1" version = "1.1.1"
@@ -1699,6 +1764,12 @@ dependencies = [
"zune-inflate", "zune-inflate",
] ]
[[package]]
name = "extended"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "af9673d8203fcb076b19dfd17e38b3d4ae9f44959416ea532ce72415a6020365"
[[package]] [[package]]
name = "fastrand" name = "fastrand"
version = "2.4.1" version = "2.4.1"
@@ -2557,6 +2628,7 @@ dependencies = [
"iced_core", "iced_core",
"log", "log",
"rustc-hash 2.1.2", "rustc-hash 2.1.2",
"tokio",
"wasm-bindgen-futures", "wasm-bindgen-futures",
"wasmtimer", "wasmtimer",
] ]
@@ -3565,6 +3637,15 @@ dependencies = [
"libc", "libc",
] ]
[[package]]
name = "mach2"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a1b95cd5421ec55b445b5ae102f5ea0e768de1f82bd3001e11f426c269c3aea"
dependencies = [
"libc",
]
[[package]] [[package]]
name = "malloc_buf" name = "malloc_buf"
version = "0.0.6" version = "0.0.6"
@@ -4252,6 +4333,31 @@ dependencies = [
"objc2-quartz-core 0.3.2", "objc2-quartz-core 0.3.2",
] ]
[[package]]
name = "objc2-audio-toolbox"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6948501a91121d6399b79abaa33a8aa4ea7857fe019f341b8c23ad6e81b79b08"
dependencies = [
"bitflags 2.11.1",
"libc",
"objc2 0.6.4",
"objc2-core-audio",
"objc2-core-audio-types",
"objc2-core-foundation",
"objc2-foundation 0.3.2",
]
[[package]]
name = "objc2-avf-audio"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13a380031deed8e99db00065c45937da434ca987c034e13b87e4441f9e4090be"
dependencies = [
"objc2 0.6.4",
"objc2-foundation 0.3.2",
]
[[package]] [[package]]
name = "objc2-cloud-kit" name = "objc2-cloud-kit"
version = "0.2.2" version = "0.2.2"
@@ -4287,6 +4393,29 @@ dependencies = [
"objc2-foundation 0.2.2", "objc2-foundation 0.2.2",
] ]
[[package]]
name = "objc2-core-audio"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1eebcea8b0dbff5f7c8504f3107c68fc061a3eb44932051c8cf8a68d969c3b2"
dependencies = [
"dispatch2",
"objc2 0.6.4",
"objc2-core-audio-types",
"objc2-core-foundation",
"objc2-foundation 0.3.2",
]
[[package]]
name = "objc2-core-audio-types"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a89f2ec274a0cf4a32642b2991e8b351a404d290da87bb6a9a9d8632490bd1c"
dependencies = [
"bitflags 2.11.1",
"objc2 0.6.4",
]
[[package]] [[package]]
name = "objc2-core-data" name = "objc2-core-data"
version = "0.2.2" version = "0.2.2"
@@ -4742,13 +4871,13 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
[[package]] [[package]]
name = "peerspeak" name = "peerspeak"
version = "0.2.0" version = "0.3.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait", "async-trait",
"base64", "base64",
"bytes", "bytes",
"cpal", "cpal 0.15.3",
"dirs", "dirs",
"iced", "iced",
"image", "image",
@@ -4759,6 +4888,7 @@ dependencies = [
"rand 0.10.1", "rand 0.10.1",
"rfd", "rfd",
"ringbuf", "ringbuf",
"rodio",
"serde", "serde",
"serde_json", "serde_json",
"thiserror 2.0.18", "thiserror 2.0.18",
@@ -5198,6 +5328,16 @@ version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
[[package]]
name = "rand_distr"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4d431c2703ccf129de4d45253c03f49ebb22b97d6ad79ee3ecfc7e3f4862c1d8"
dependencies = [
"num-traits",
"rand 0.10.1",
]
[[package]] [[package]]
name = "rand_pcg" name = "rand_pcg"
version = "0.10.2" version = "0.10.2"
@@ -5487,12 +5627,34 @@ dependencies = [
"portable-atomic-util", "portable-atomic-util",
] ]
[[package]]
name = "rodio"
version = "0.22.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d0a536bb79db59098ef71a4dd4246c02eb87b316deceb1b68e0cde7167ec01eb"
dependencies = [
"cpal 0.17.1",
"dasp_sample",
"num-rational",
"rand 0.10.1",
"rand_distr",
"rtrb",
"symphonia",
"thiserror 2.0.18",
]
[[package]] [[package]]
name = "roxmltree" name = "roxmltree"
version = "0.20.0" version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c20b6793b5c2fa6553b250154b78d6d0db37e72700ae35fad9387a46f487c97" checksum = "6c20b6793b5c2fa6553b250154b78d6d0db37e72700ae35fad9387a46f487c97"
[[package]]
name = "rtrb"
version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ade083ccbb4bf536df69d1f6432cc23deb7acccff86b183f3923a6fd56a1153"
[[package]] [[package]]
name = "rustc-hash" name = "rustc-hash"
version = "1.1.0" version = "1.1.0"
@@ -6156,6 +6318,153 @@ dependencies = [
"zeno", "zeno",
] ]
[[package]]
name = "symphonia"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5773a4c030a19d9bfaa090f49746ff35c75dfddfa700df7a5939d5e076a57039"
dependencies = [
"lazy_static",
"symphonia-bundle-flac",
"symphonia-bundle-mp3",
"symphonia-codec-aac",
"symphonia-codec-pcm",
"symphonia-codec-vorbis",
"symphonia-core",
"symphonia-format-isomp4",
"symphonia-format-ogg",
"symphonia-format-riff",
"symphonia-metadata",
]
[[package]]
name = "symphonia-bundle-flac"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c91565e180aea25d9b80a910c546802526ffd0072d0b8974e3ebe59b686c9976"
dependencies = [
"log",
"symphonia-core",
"symphonia-metadata",
"symphonia-utils-xiph",
]
[[package]]
name = "symphonia-bundle-mp3"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4872dd6bb56bf5eac799e3e957aa1981086c3e613b27e0ac23b176054f7c57ed"
dependencies = [
"lazy_static",
"log",
"symphonia-core",
"symphonia-metadata",
]
[[package]]
name = "symphonia-codec-aac"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4c263845aa86881416849c1729a54c7f55164f8b96111dba59de46849e73a790"
dependencies = [
"lazy_static",
"log",
"symphonia-core",
]
[[package]]
name = "symphonia-codec-pcm"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e89d716c01541ad3ebe7c91ce4c8d38a7cf266a3f7b2f090b108fb0cb031d95"
dependencies = [
"log",
"symphonia-core",
]
[[package]]
name = "symphonia-codec-vorbis"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f025837c309cd69ffef572750b4a2257b59552c5399a5e49707cc5b1b85d1c73"
dependencies = [
"log",
"symphonia-core",
"symphonia-utils-xiph",
]
[[package]]
name = "symphonia-core"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ea00cc4f79b7f6bb7ff87eddc065a1066f3a43fe1875979056672c9ef948c2af"
dependencies = [
"arrayvec",
"bitflags 1.3.2",
"bytemuck",
"lazy_static",
"log",
]
[[package]]
name = "symphonia-format-isomp4"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "243739585d11f81daf8dac8d9f3d18cc7898f6c09a259675fc364b382c30e0a5"
dependencies = [
"encoding_rs",
"log",
"symphonia-core",
"symphonia-metadata",
"symphonia-utils-xiph",
]
[[package]]
name = "symphonia-format-ogg"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b4955c67c1ed3aa8ae8428d04ca8397fbef6a19b2b051e73b5da8b1435639cb"
dependencies = [
"log",
"symphonia-core",
"symphonia-metadata",
"symphonia-utils-xiph",
]
[[package]]
name = "symphonia-format-riff"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2d7c3df0e7d94efb68401d81906eae73c02b40d5ec1a141962c592d0f11a96f"
dependencies = [
"extended",
"log",
"symphonia-core",
"symphonia-metadata",
]
[[package]]
name = "symphonia-metadata"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "36306ff42b9ffe6e5afc99d49e121e0bd62fe79b9db7b9681d48e29fa19e6b16"
dependencies = [
"encoding_rs",
"lazy_static",
"log",
"symphonia-core",
]
[[package]]
name = "symphonia-utils-xiph"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee27c85ab799a338446b68eec77abf42e1a6f1bb490656e121c6e27bfbab9f16"
dependencies = [
"symphonia-core",
"symphonia-metadata",
]
[[package]] [[package]]
name = "syn" name = "syn"
version = "2.0.117" version = "2.0.117"
+3 -2
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "peerspeak" name = "peerspeak"
version = "0.2.0" version = "0.3.0"
edition = "2024" edition = "2024"
# Application crate, not a crates.io library — refuse `cargo publish` and let # Application crate, not a crates.io library — refuse `cargo publish` and let
# cargo-deny's [licenses.private] skip the missing-license check. # cargo-deny's [licenses.private] skip the missing-license check.
@@ -28,7 +28,7 @@ async-trait = "0.1.89"
base64 = "0.22.1" base64 = "0.22.1"
bytes = "1.11.1" bytes = "1.11.1"
dirs = "6.0.0" dirs = "6.0.0"
iced = { version = "0.14.0", features = ["canvas", "image"] } iced = { version = "0.14.0", features = ["canvas", "image", "tokio"] }
# W4 custom avatars: decode/resize an arbitrary user image (png/jpeg only to keep # W4 custom avatars: decode/resize an arbitrary user image (png/jpeg only to keep
# the codec surface small). The matching native file picker (`rfd`) is platform- # the codec surface small). The matching native file picker (`rfd`) is platform-
# gated below — its backend differs per OS (xdg-portal on Linux, Win32 on Windows). # gated below — its backend differs per OS (xdg-portal on Linux, Win32 on Windows).
@@ -38,6 +38,7 @@ iroh-gossip = "0.99.0"
opus = "0.3.1" opus = "0.3.1"
rand = "0.10.1" rand = "0.10.1"
ringbuf = "0.5.0" ringbuf = "0.5.0"
rodio = "0.22.2"
serde = { version = "1.0.228", features = ["derive"] } serde = { version = "1.0.228", features = ["derive"] }
serde_json = "1.0.150" serde_json = "1.0.150"
thiserror = "2.0.18" thiserror = "2.0.18"
+1 -1
View File
@@ -1,7 +1,7 @@
# Maintainer: mollusk <jitty+lc1iz0dc@protonmail.com> # Maintainer: mollusk <jitty+lc1iz0dc@protonmail.com>
pkgname=peerspeak-git pkgname=peerspeak-git
_pkgname=peerspeak _pkgname=peerspeak
pkgver=0.1.0 pkgver=0.2.0.r218.gcbba4b6
pkgrel=1 pkgrel=1
pkgdesc="Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)" pkgdesc="Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)"
arch=('x86_64') arch=('x86_64')
+28 -2
View File
@@ -8,7 +8,7 @@ it once, then you and I connect directly to each other.
## 1. Install it ## 1. Install it
1. Double-click **`peerspeak-0.2.0-setup.exe`** (the file I sent you). 1. Double-click **`peerspeak-0.3.0-setup.exe`** (the file I sent you).
2. **Windows will probably show a blue "Windows protected your PC" warning.** 2. **Windows will probably show a blue "Windows protected your PC" warning.**
This is normal — it shows up for any app that isn't from a big company with a This is normal — it shows up for any app that isn't from a big company with a
@@ -69,13 +69,39 @@ Either way works the same; it just depends on who makes the room.
--- ---
## 4. Chatting and sharing photos/files
There's a **text chat** box at the bottom of the call window — type a message
and press **Enter** to send it to everyone in the room.
You can also **send a photo or a file**:
1. Click the **attach button** (the small paperclip-style button) next to the
message box.
2. Pick a photo or file from your computer.
3. It sends to everyone in the room. **Photos show up right in the chat**;
other files appear as a small download chip with the file's name.
To **save** a file someone sent you, click the **Save** (or **Download**)
button next to it in the chat and choose where to put it.
A couple of notes:
- There's a size limit of about **25 MB** per file — bigger files are turned
away with a message.
- Shared files only last for the **current call**. They aren't saved anywhere
automatically, so save anything you want to keep before you leave the room.
---
## Troubleshooting ## Troubleshooting
- **"I don't hear anything."** Open Settings and pick the correct microphone and - **"I don't hear anything."** Open Settings and pick the correct microphone and
output device. Headphones are best — they prevent echo. output device. Headphones are best — they prevent echo.
- **"It won't connect."** Make sure you pasted the *entire* ticket (they're - **"It won't connect."** Make sure you pasted the *entire* ticket (they're
long and easy to cut off). If it still won't connect, we may just need a fresh long and easy to cut off). If it still won't connect, we may just need a fresh
ticket — they're meant to be used right away. ticket — they're meant to be used right away. Also make sure we're both on the
**same version** — if I've sent you an updated installer, install it (an old
version and a new one can't connect to each other).
- **The blue warning again.** Same as install: **More info → Run anyway**. It's - **The blue warning again.** Same as install: **More info → Run anyway**. It's
the unsigned-app warning, not malware. the unsigned-app warning, not malware.
+12
View File
@@ -9,6 +9,18 @@ notification chimes, and avatar presets are all embedded in the binary
runtime, so there are no extra DLLs to bundle. The installer payload is just the runtime, so there are no extra DLLs to bundle. The installer payload is just the
`.exe` plus an `.ico` for the Start-menu / desktop shortcuts. `.exe` plus an `.ico` for the Start-menu / desktop shortcuts.
## Version compatibility
The installer version tracks the crate version in `Cargo.toml` (currently
**0.3.0**) — keep `MyAppVersion` in `peerspeak.iss` in sync when it changes.
Per `VERSIONING.md`, a **MINOR** bump in `0.x` is a **breaking wire change**:
peers on different MINOR versions can't connect (they fail fast at the
handshake rather than misbehaving). So when you ship a new Windows build after
a MINOR bump, **everyone on the call must reinstall** — an old Windows build
and a newer Linux/Windows peer won't talk. (0.3.0 was the chat file-sharing +
per-peer noise-gate release; it cannot connect to a 0.2.x peer.)
## Files ## Files
| File | Tracked | Purpose | | File | Tracked | Purpose |
+1 -1
View File
@@ -12,7 +12,7 @@
; (x86_64-pc-windows-gnu, statically linked -- no extra DLLs needed). ; (x86_64-pc-windows-gnu, statically linked -- no extra DLLs needed).
#define MyAppName "PeerSpeak" #define MyAppName "PeerSpeak"
#define MyAppVersion "0.2.0" #define MyAppVersion "0.3.0"
#define MyAppPublisher "mollusk" #define MyAppPublisher "mollusk"
#define MyAppExeName "peerspeak.exe" #define MyAppExeName "peerspeak.exe"
+543 -9
View File
@@ -2,6 +2,10 @@ use crate::core::{CoreController, messages::{CoreCommand, UiEvent}};
use crate::network::PeerState; use crate::network::PeerState;
use crate::notify::{self, Sound}; use crate::notify::{self, Sound};
use crate::audio::eq::{EqSettings, EQ_GAIN_DB_MAX, EQ_GAIN_DB_MIN}; use crate::audio::eq::{EqSettings, EQ_GAIN_DB_MAX, EQ_GAIN_DB_MIN};
use crate::audio::clip_player::{
ClipPlayer, SharedClipStatus, format_time as format_clip_time, progress as clip_progress,
seek_target, status_snapshot,
};
use crate::audio::{AudioDevice, enumerate_audio_devices}; use crate::audio::{AudioDevice, enumerate_audio_devices};
use crate::config::{AppConfig, NetworkMode, RecordingMode, RoomLayout}; use crate::config::{AppConfig, NetworkMode, RecordingMode, RoomLayout};
use crate::hotkeys::{format_binding, HotkeyAction, HotkeyContext, KeyBinding}; use crate::hotkeys::{format_binding, HotkeyAction, HotkeyContext, KeyBinding};
@@ -114,6 +118,19 @@ struct ChatEntry {
/// Sender's node id string, used to key their avatar colour (W4). `None` only /// Sender's node id string, used to key their avatar colour (W4). `None` only
/// for any future system-generated lines. /// for any future system-generated lines.
from: Option<String>, from: Option<String>,
/// Optional file attachment descriptor. The bytes (if fetched) live in
/// `AppState.attachment_data` keyed by `attachment.id`; the entry only holds
/// the descriptor so history stays cheap.
attachment: Option<crate::files::ChatAttachment>,
}
/// Fetch state of a chat attachment's bytes (session-only).
#[derive(Debug, Clone)]
enum AttachmentState {
/// Bytes in hand (image decoded-valid, or a file ready to save).
Ready(Vec<u8>),
/// Fetch or decode failed; carries a short reason for the UI.
Failed(String),
} }
/// Cap on retained chat history so a long call can't grow it without bound. /// Cap on retained chat history so a long call can't grow it without bound.
@@ -208,6 +225,7 @@ pub enum AppMessage {
ClearHotkey(HotkeyAction), ClearHotkey(HotkeyAction),
PeerVolumeChanged(EndpointId, f32), PeerVolumeChanged(EndpointId, f32),
PeerPanChanged(EndpointId, f32), PeerPanChanged(EndpointId, f32),
PeerGateChanged(EndpointId, f32),
PeerEqChanged(EndpointId, EqBand, f32), PeerEqChanged(EndpointId, EqBand, f32),
/// Toggle local mute of a peer (silence them just for us). /// Toggle local mute of a peer (silence them just for us).
TogglePeerMute(EndpointId), TogglePeerMute(EndpointId),
@@ -263,6 +281,19 @@ pub enum AppMessage {
ToggleRecording, ToggleRecording,
/// Live edits to the chat input line. /// Live edits to the chat input line.
ChatInputChanged(String), ChatInputChanged(String),
/// Open the native picker to attach a file to the chat.
PickAttachmentFile,
/// Result of the attach picker: (filename, bytes), or None if cancelled.
AttachmentFilePicked(Option<(String, Vec<u8>)>),
/// Save (downloading first if needed) a received attachment to disk.
SaveAttachment(crate::files::AttachmentId),
/// Fetch (if needed) and start an inline audio attachment.
PlayAudio(crate::files::AttachmentId),
PauseAudio,
ResumeAudio,
SeekAudio(crate::files::AttachmentId, f32),
/// Redraw cadence while an inline clip is active.
AudioTick,
/// Send the current chat input line (Enter or the Send button). /// Send the current chat input line (Enter or the Send button).
ChatSubmit, ChatSubmit,
/// Open a clicked chat link in the system browser (A13). /// Open a clicked chat link in the system browser (A13).
@@ -349,7 +380,6 @@ pub struct AppState {
/// refreshed when the background is changed/removed. `None` = no custom bg. /// refreshed when the background is changed/removed. `None` = no custom bg.
background_image: Option<bytes::Bytes>, background_image: Option<bytes::Bytes>,
peers: HashMap<EndpointId, PeerState>, peers: HashMap<EndpointId, PeerState>,
peer_volumes: HashMap<EndpointId, f32>,
audio_levels: HashMap<EndpointId, f32>, audio_levels: HashMap<EndpointId, f32>,
/// Peers we've locally muted (their audio isn't mixed into our output). /// Peers we've locally muted (their audio isn't mixed into our output).
locally_muted: HashSet<EndpointId>, locally_muted: HashSet<EndpointId>,
@@ -362,6 +392,24 @@ pub struct AppState {
/// Room text-chat history (newest last) and the pending input line. /// Room text-chat history (newest last) and the pending input line.
chat_messages: Vec<ChatEntry>, chat_messages: Vec<ChatEntry>,
chat_input: String, chat_input: String,
/// Fetched/failed state for chat attachments, keyed by attachment id.
/// Session-only (cleared on leave); never persisted.
attachment_data: HashMap<crate::files::AttachmentId, AttachmentState>,
/// Cached iced image handles for ready image attachments, keyed by id, so we
/// don't re-upload to the GPU every redraw (the e917c53 avatar flicker fix).
image_handle_cache: HashMap<crate::files::AttachmentId, iced::widget::image::Handle>,
/// Attachment ids the user asked to save before the bytes arrived; when the
/// fetch completes a save dialog is opened for them.
pending_saves: std::collections::HashSet<crate::files::AttachmentId>,
/// Clip ids waiting for the existing attachment fetch path to return bytes.
pending_plays: HashSet<crate::files::AttachmentId>,
/// Filename-hinted audio whose fetched bytes or decoder validation failed;
/// these entries fall back to the normal file chip.
invalid_audio: HashSet<crate::files::AttachmentId>,
/// Independent system-default-device player for chat clips. It never enters
/// the call capture/mixer path.
clip_player: ClipPlayer,
clip_status: SharedClipStatus,
/// Last known window size, tracked so divider clamps stay valid on resize. /// Last known window size, tracked so divider clamps stay valid on resize.
/// (The divider positions themselves are persisted in `config`.) /// (The divider positions themselves are persisted in `config`.)
window_size: Size, window_size: Size,
@@ -474,6 +522,16 @@ impl Default for AppState {
let _ = controller.send(CoreCommand::SetPeerPan(id, *pan)); let _ = controller.send(CoreCommand::SetPeerPan(id, *pan));
} }
} }
for (peer, volume) in &config.peer_volume {
if let Ok(id) = peer.parse::<EndpointId>() {
let _ = controller.send(CoreCommand::SetPeerVolume(id, *volume));
}
}
for (peer, threshold) in &config.peer_gate {
if let Ok(id) = peer.parse::<EndpointId>() {
let _ = controller.send(CoreCommand::SetPeerGate(id, *threshold));
}
}
let pixelpass_available = let pixelpass_available =
crate::screenshare::is_available(config.pixelpass_path.as_deref()); crate::screenshare::is_available(config.pixelpass_path.as_deref());
let all_devices = enumerate_audio_devices(); let all_devices = enumerate_audio_devices();
@@ -484,6 +542,7 @@ impl Default for AppState {
let selected_output = output_devices.iter().find(|d| d.name == config.output_device).cloned(); let selected_output = output_devices.iter().find(|d| d.name == config.output_device).cloned();
let background_image = load_background_bytes(&config); let background_image = load_background_bytes(&config);
let (clip_player, clip_status) = ClipPlayer::new();
Self { Self {
// Pre-fill the nickname with the last one used (or "Peer" by default). // Pre-fill the nickname with the last one used (or "Peer" by default).
@@ -505,13 +564,19 @@ impl Default for AppState {
config, config,
background_image, background_image,
peers: HashMap::new(), peers: HashMap::new(),
peer_volumes: HashMap::new(),
audio_levels: HashMap::new(), audio_levels: HashMap::new(),
locally_muted: HashSet::new(), locally_muted: HashSet::new(),
call_started: None, call_started: None,
recording: false, recording: false,
recording_started: None, recording_started: None,
chat_messages: Vec::new(), chat_messages: Vec::new(),
attachment_data: HashMap::new(),
image_handle_cache: HashMap::new(),
pending_saves: std::collections::HashSet::new(),
pending_plays: HashSet::new(),
invalid_audio: HashSet::new(),
clip_player,
clip_status,
chat_input: String::new(), chat_input: String::new(),
window_size: Size::new(ww, wh), window_size: Size::new(ww, wh),
layout_picker_open: false, layout_picker_open: false,
@@ -636,10 +701,15 @@ fn initial_window_position(
} }
} }
fn subscription(_state: &AppState) -> Subscription<AppMessage> { fn subscription(state: &AppState) -> Subscription<AppMessage> {
let core_sub = Subscription::run(core_subscription).map(AppMessage::UiEventReceived); let core_sub = Subscription::run(core_subscription).map(AppMessage::UiEventReceived);
let event_sub = iced::event::listen().map(AppMessage::EventOccurred); let event_sub = iced::event::listen().map(AppMessage::EventOccurred);
Subscription::batch(vec![core_sub, event_sub]) let audio_sub = if status_snapshot(&state.clip_status).playing_id.is_some() {
iced::time::every(std::time::Duration::from_millis(250)).map(|_| AppMessage::AudioTick)
} else {
Subscription::none()
};
Subscription::batch(vec![core_sub, event_sub, audio_sub])
} }
fn shutdown_timeout_task() -> Task<AppMessage> { fn shutdown_timeout_task() -> Task<AppMessage> {
@@ -744,6 +814,34 @@ fn set_peer_pan_config(config: &mut AppConfig, id: EndpointId, pan: f32) -> f32
pan pan
} }
/// Store the per-peer listener volume, clamped to the slider range. Unity gain
/// (`1.0`) is the implicit default, so an at-unity entry is removed rather than
/// stored to keep the config tidy. Returns the clamped value.
fn set_peer_volume_config(config: &mut AppConfig, id: EndpointId, volume: f32) -> f32 {
let volume = volume.clamp(0.0, 2.0);
let key = id.to_string();
if (volume - 1.0).abs() <= 0.001 {
config.peer_volume.remove(&key);
} else {
config.peer_volume.insert(key, volume);
}
volume
}
/// Store the per-peer listener noise-gate threshold, clamped to the slider
/// range. `0.0` means the gate is off, so an at-zero entry is removed rather
/// than stored. Returns the clamped value.
fn set_peer_gate_config(config: &mut AppConfig, id: EndpointId, threshold: f32) -> f32 {
let threshold = threshold.clamp(0.0, METER_MAX);
let key = id.to_string();
if threshold <= 0.0 {
config.peer_gate.remove(&key);
} else {
config.peer_gate.insert(key, threshold);
}
threshold
}
fn peer_eq_settings(config: &AppConfig, id: &EndpointId) -> EqSettings { fn peer_eq_settings(config: &AppConfig, id: &EndpointId) -> EqSettings {
config config
.peer_eq .peer_eq
@@ -883,6 +981,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
notify::play(Sound::SelfJoin, state.config.custom_sound_self_join.as_deref()); notify::play(Sound::SelfJoin, state.config.custom_sound_self_join.as_deref());
} }
UiEvent::RoomLeft => { UiEvent::RoomLeft => {
state.clip_player.stop();
state.ticket = "".to_string(); state.ticket = "".to_string();
state.peers.clear(); state.peers.clear();
state.audio_levels.clear(); state.audio_levels.clear();
@@ -892,6 +991,11 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
state.recording_started = None; state.recording_started = None;
state.chat_messages.clear(); state.chat_messages.clear();
state.chat_input.clear(); state.chat_input.clear();
state.attachment_data.clear();
state.image_handle_cache.clear();
state.pending_saves.clear();
state.pending_plays.clear();
state.invalid_audio.clear();
state.connecting.clear(); state.connecting.clear();
state.ever_connected.clear(); state.ever_connected.clear();
state.status_message = "Ready to connect".to_string(); state.status_message = "Ready to connect".to_string();
@@ -959,19 +1063,50 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
state.recording_started = None; state.recording_started = None;
state.status_message = format!("Saved recording → {path}"); state.status_message = format!("Saved recording → {path}");
} }
UiEvent::ChatMessage { from, name, text } => { UiEvent::ChatMessage { from, name, text, attachment } => {
// Incoming peer content is untrusted — sanitize name + text. // Incoming peer content is untrusted — sanitize name + text.
// (The attachment filename was already sanitized in core.)
let text = sanitize_chat(&text); let text = sanitize_chat(&text);
if !text.is_empty() { // Keep the message if it has visible text OR an attachment (an
// image with no caption is still a real message).
if !text.is_empty() || attachment.is_some() {
let name = sanitize_chat(&name); let name = sanitize_chat(&name);
push_chat(&mut state.chat_messages, ChatEntry { push_chat(&mut state.chat_messages, ChatEntry {
name, name,
text, text,
mine: false, mine: false,
from: Some(from), from: Some(from),
attachment,
}); });
} }
} }
UiEvent::AttachmentReady { id, data } => {
// Bytes arrived. For images we can cache the iced handle now
// (built once, not per redraw). If the user was waiting to save
// this file, the save dialog is opened from update() below by
// checking pending_saves — done lazily so this arm stays simple.
if crate::files::validate_image_bytes(&data).is_some() {
state.image_handle_cache.insert(
id,
iced::widget::image::Handle::from_bytes(data.clone()),
);
}
let needs_save = state.pending_saves.remove(&id);
let needs_play = state.pending_plays.remove(&id);
state.attachment_data.insert(id, AttachmentState::Ready(data));
if needs_save {
save_attachment_to_disk(state, id);
}
if needs_play {
play_ready_audio(state, id);
}
}
UiEvent::AttachmentFailed { id, error } => {
state.pending_saves.remove(&id);
state.pending_plays.remove(&id);
state.attachment_data.insert(id, AttachmentState::Failed(error.clone()));
state.status_message = format!("Attachment failed: {error}");
}
UiEvent::ScreenShareStarted => { UiEvent::ScreenShareStarted => {
state.self_sharing = true; state.self_sharing = true;
state.status_message = "Sharing your screen".to_string(); state.status_message = "Sharing your screen".to_string();
@@ -1038,13 +1173,17 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
state.config.save(); state.config.save();
} }
AppMessage::PeerVolumeChanged(id, vol) => { AppMessage::PeerVolumeChanged(id, vol) => {
state.peer_volumes.insert(id, vol); let vol = set_peer_volume_config(&mut state.config, id, vol);
let _ = state.controller.send(CoreCommand::SetPeerVolume(id, vol)); let _ = state.controller.send(CoreCommand::SetPeerVolume(id, vol));
} }
AppMessage::PeerPanChanged(id, pan) => { AppMessage::PeerPanChanged(id, pan) => {
let pan = set_peer_pan_config(&mut state.config, id, pan); let pan = set_peer_pan_config(&mut state.config, id, pan);
let _ = state.controller.send(CoreCommand::SetPeerPan(id, pan)); let _ = state.controller.send(CoreCommand::SetPeerPan(id, pan));
} }
AppMessage::PeerGateChanged(id, threshold) => {
let threshold = set_peer_gate_config(&mut state.config, id, threshold);
let _ = state.controller.send(CoreCommand::SetPeerGate(id, threshold));
}
AppMessage::PeerEqChanged(id, band, gain_db) => { AppMessage::PeerEqChanged(id, band, gain_db) => {
let settings = set_peer_eq_config(&mut state.config, id, band, gain_db); let settings = set_peer_eq_config(&mut state.config, id, band, gain_db);
let _ = state.controller.send(CoreCommand::SetPeerEq(id, settings)); let _ = state.controller.send(CoreCommand::SetPeerEq(id, settings));
@@ -1450,11 +1589,132 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
text: text.clone(), text: text.clone(),
mine: true, mine: true,
from: Some(state.self_id.clone()), from: Some(state.self_id.clone()),
attachment: None,
}); });
let _ = state.controller.send(CoreCommand::SendChat(text)); let _ = state.controller.send(CoreCommand::SendChat(text));
state.chat_input.clear(); state.chat_input.clear();
} }
} }
AppMessage::PickAttachmentFile => {
// Native picker off the UI thread; returns (filename, bytes).
return Task::perform(
async {
let handle = rfd::AsyncFileDialog::new()
.set_title("Attach a file to the chat")
.pick_file()
.await;
match handle {
Some(h) => Some((h.file_name(), h.read().await)),
None => None,
}
},
AppMessage::AttachmentFilePicked,
);
}
AppMessage::AttachmentFilePicked(picked) => {
if let Some((name, bytes)) = picked {
let size = bytes.len() as u64;
if !crate::files::size_within_cap(size) {
state.status_message = format!(
"File too large — max {}.",
crate::files::human_size(crate::files::MAX_ATTACHMENT_BYTES)
);
return Task::none();
}
let kind = crate::files::classify(&bytes);
// Random 32-byte handle for this attachment.
let id: crate::files::AttachmentId = rand::random();
let att = crate::files::ChatAttachment {
name: crate::files::sanitize_filename(&name),
size,
kind,
id,
};
// Keep our own bytes locally so we see our own attachment inline
// immediately (others fetch it off the file plane).
if kind == crate::files::AttachmentKind::Image
&& crate::files::validate_image_bytes(&bytes).is_some()
{
state
.image_handle_cache
.insert(id, iced::widget::image::Handle::from_bytes(bytes.clone()));
}
state
.attachment_data
.insert(id, AttachmentState::Ready(bytes.clone()));
push_chat(
&mut state.chat_messages,
ChatEntry {
name: format!("{} (You)", state.name),
text: String::new(),
mine: true,
from: Some(state.self_id.clone()),
attachment: Some(att.clone()),
},
);
let _ = state.controller.send(CoreCommand::SendChatFile {
text: String::new(),
attachment: att,
data: bytes,
});
}
}
AppMessage::SaveAttachment(id) => {
// If we already have the bytes, save now; otherwise fetch from the
// sender and save when AttachmentReady arrives (pending_saves).
if matches!(state.attachment_data.get(&id), Some(AttachmentState::Ready(_))) {
save_attachment_to_disk(state, id);
} else if let Some((from, att)) = find_attachment_source(state, id) {
if let Ok(eid) = from.parse::<EndpointId>() {
state.pending_saves.insert(id);
state.status_message = format!("Downloading {}", att.name);
let _ = state
.controller
.send(CoreCommand::FetchAttachment { from: eid, attachment: att });
} else {
state.status_message = "Can't download: unknown sender.".to_string();
}
}
}
AppMessage::PlayAudio(id) => {
if matches!(state.attachment_data.get(&id), Some(AttachmentState::Ready(_))) {
play_ready_audio(state, id);
} else if let Some((from, att)) = find_attachment_source(state, id) {
if let Ok(eid) = from.parse::<EndpointId>() {
// Repeated clicks while the transfer is pending must not
// launch duplicate fetches.
if state.pending_plays.insert(id) {
state.status_message = format!("Loading {}", att.name);
let _ = state
.controller
.send(CoreCommand::FetchAttachment { from: eid, attachment: att });
}
} else {
state.status_message = "Can't play: unknown sender.".to_string();
}
}
}
AppMessage::PauseAudio => state.clip_player.pause(),
AppMessage::ResumeAudio => state.clip_player.resume(),
AppMessage::SeekAudio(id, fraction) => {
let clip = status_snapshot(&state.clip_status);
if clip.playing_id == Some(id)
&& let Some(total) = clip.total
{
state.clip_player.seek(seek_target(fraction, total));
}
}
AppMessage::AudioTick => {
let clip = status_snapshot(&state.clip_status);
if let Some(failure) = clip.failure {
if failure.invalid_audio {
state.invalid_audio.insert(failure.id);
}
state.pending_plays.remove(&failure.id);
state.status_message = format!("Audio playback failed: {}", failure.error);
state.clip_player.stop();
}
}
AppMessage::OpenUrl(url) => { AppMessage::OpenUrl(url) => {
// Defence in depth: only ever hand http(s) URLs to the opener. The // Defence in depth: only ever hand http(s) URLs to the opener. The
// link span's href came from `linkify`, which only emits http/https, // link span's href came from `linkify`, which only emits http/https,
@@ -1688,6 +1948,69 @@ fn push_chat(messages: &mut Vec<ChatEntry>, entry: ChatEntry) {
} }
} }
/// Find the sender id + descriptor for a received attachment by its id, so a
/// fetch can be addressed. Returns `None` for our own attachments or an unknown
/// id.
fn find_attachment_source(
state: &AppState,
id: crate::files::AttachmentId,
) -> Option<(String, crate::files::ChatAttachment)> {
state.chat_messages.iter().find_map(|m| {
let att = m.attachment.as_ref()?;
if att.id == id && !m.mine {
Some((m.from.clone()?, att.clone()))
} else {
None
}
})
}
/// Validate cached bytes and hand them to the independent clip player. A false
/// filename hint falls back to the generic file chip without reaching rodio.
fn play_ready_audio(state: &mut AppState, id: crate::files::AttachmentId) {
let Some(AttachmentState::Ready(data)) = state.attachment_data.get(&id) else {
return;
};
if crate::files::is_probably_audio(data) {
state.invalid_audio.remove(&id);
state.clip_player.play(id, data.clone());
} else {
state.invalid_audio.insert(id);
state.status_message = "This attachment is not valid supported audio.".to_string();
}
}
/// Write a ready attachment's bytes to a user-chosen location via a native save
/// dialog. The default filename comes from the (already-sanitized) descriptor.
/// No-op if the bytes aren't ready. Sync dialog: the brief block is acceptable
/// for a deliberate save action.
fn save_attachment_to_disk(state: &mut AppState, id: crate::files::AttachmentId) {
let Some(AttachmentState::Ready(data)) = state.attachment_data.get(&id) else {
return;
};
let data = data.clone();
let default_name = state
.chat_messages
.iter()
.find_map(|m| {
m.attachment
.as_ref()
.filter(|a| a.id == id)
.map(|a| a.name.clone())
})
.unwrap_or_else(|| "download".to_string());
if let Some(path) = rfd::FileDialog::new()
.set_file_name(default_name)
.set_title("Save attachment")
.save_file()
{
match std::fs::write(&path, &data) {
Ok(()) => state.status_message = format!("Saved {}", path.display()),
Err(e) => state.status_message = format!("Save failed: {e}"),
}
}
}
fn horizontal_space() -> iced::widget::Space { fn horizontal_space() -> iced::widget::Space {
iced::widget::Space::new().width(iced::Length::Fill) iced::widget::Space::new().width(iced::Length::Fill)
} }
@@ -3348,11 +3671,18 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
].spacing(8); ].spacing(8);
// Peer volume slider // Peer volume slider
let current_vol = state.peer_volumes.get(peer_id).copied().unwrap_or(1.0); let current_vol = state
.config
.peer_volume
.get(&peer_id.to_string())
.copied()
.unwrap_or(1.0);
card_content = card_content.push( card_content = card_content.push(
row![ row![
text("Vol:").size(12).color(color_subtext), text("Vol:").size(12).color(color_subtext),
slider(0.0..=2.0, current_vol, move |v| AppMessage::PeerVolumeChanged(peer_id_clone, v)) slider(0.0..=2.0, current_vol, move |v| AppMessage::PeerVolumeChanged(peer_id_clone, v))
.step(0.01)
.on_release(AppMessage::PersistConfig)
].spacing(8).align_y(iced::alignment::Vertical::Center) ].spacing(8).align_y(iced::alignment::Vertical::Center)
); );
@@ -3371,6 +3701,28 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
.align_y(iced::alignment::Vertical::Center), .align_y(iced::alignment::Vertical::Center),
); );
// Peer noise gate: suppress this peer's background noise on our end.
// Threshold is normalized RMS on the same 0..METER_MAX scale as the
// mic gate; 0 = off.
let current_gate = state.config.peer_gate.get(&peer_key).copied().unwrap_or(0.0);
let gate_label = if current_gate <= 0.0 {
"Off".to_string()
} else {
format!("{:.0}%", (current_gate / METER_MAX * 100.0).clamp(0.0, 100.0))
};
card_content = card_content.push(
row![
text("Gate:").size(12).color(color_subtext),
container(text(gate_label).size(11).color(color_subtext))
.width(iced::Length::Fixed(58.0)),
slider(0.0..=METER_MAX, current_gate, move |v| AppMessage::PeerGateChanged(peer_id_clone, v))
.step(0.001)
.on_release(AppMessage::PersistConfig),
]
.spacing(8)
.align_y(iced::alignment::Vertical::Center),
);
let eq = peer_eq_settings(&state.config, peer_id); let eq = peer_eq_settings(&state.config, peer_id);
let eq_row = |label: &'static str, band: EqBand, value: f32| -> Element<'_, AppMessage> { let eq_row = |label: &'static str, band: EqBand, value: f32| -> Element<'_, AppMessage> {
row![ row![
@@ -3576,6 +3928,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
.color(color_subtext), .color(color_subtext),
); );
} else { } else {
let clip_status = status_snapshot(&state.clip_status);
for m in &state.chat_messages { for m in &state.chat_messages {
let name_color = if m.mine { color_green } else { color_lavender }; let name_color = if m.mine { color_green } else { color_lavender };
// Split the (already-sanitized) message into text + URL spans so // Split the (already-sanitized) message into text + URL spans so
@@ -3623,6 +3976,140 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
.spacing(8) .spacing(8)
.align_y(iced::alignment::Vertical::Top), .align_y(iced::alignment::Vertical::Top),
); );
// Attachment row (indented under the message), if any.
if let Some(att) = &m.attachment {
let data = state.attachment_data.get(&att.id);
let elem: Element<'_, AppMessage> =
if let Some(AttachmentState::Failed(e)) = data {
text(format!("{}{e}", att.name))
.size(12)
.color(color_red)
.into()
} else if att.kind == crate::files::AttachmentKind::Image {
match state.image_handle_cache.get(&att.id) {
Some(handle) => iced::widget::image(handle.clone())
.width(iced::Length::Fixed(260.0))
.into(),
None => text(format!("🖼 {} — loading…", att.name))
.size(12)
.color(color_subtext)
.into(),
}
} else if crate::files::looks_like_audio_name(&att.name)
&& !state.invalid_audio.contains(&att.id)
{
let active = clip_status.playing_id == Some(att.id);
let loading = state.pending_plays.contains(&att.id)
&& !matches!(data, Some(AttachmentState::Ready(_)));
let position = if active {
clip_status.position
} else {
std::time::Duration::ZERO
};
let total = active.then_some(clip_status.total).flatten();
let play_button = if loading {
button(text("Loading…").size(12))
} else if active && clip_status.paused {
button(text("Play").size(12)).on_press(AppMessage::ResumeAudio)
} else if active {
button(text("Pause").size(12)).on_press(AppMessage::PauseAudio)
} else {
button(text("Play").size(12))
.on_press(AppMessage::PlayAudio(att.id))
}
.style(b_style(
color_blue,
color_lavender,
color_crust,
6.0,
))
.padding(6);
let elapsed = format_clip_time(position);
let duration = total
.map(format_clip_time)
.unwrap_or_else(|| "--:--".to_string());
column![
row![
text(format!(
"{} ({})",
att.name,
crate::files::human_size(att.size)
))
.size(12)
.color(color_text),
button(text(if matches!(data, Some(AttachmentState::Ready(_))) {
"Save"
} else {
"Download"
})
.size(12))
.on_press(AppMessage::SaveAttachment(att.id))
.style(b_style(
color_surface,
color_overlay,
color_text,
6.0,
))
.padding(6),
]
.spacing(8)
.align_y(iced::alignment::Vertical::Center),
row![
play_button,
slider(
0.0..=1.0,
if active {
clip_progress(position, total)
} else {
0.0
},
move |fraction| AppMessage::SeekAudio(att.id, fraction),
)
.step(0.001)
.width(iced::Length::Fixed(180.0)),
text(format!("{elapsed} / {duration}"))
.size(11)
.color(color_subtext),
]
.spacing(8)
.align_y(iced::alignment::Vertical::Center),
]
.spacing(4)
.into()
} else {
let ready =
matches!(data, Some(AttachmentState::Ready(_)));
let btn_label = if ready { "Save" } else { "Download" };
row![
text(format!(
"📎 {} ({})",
att.name,
crate::files::human_size(att.size)
))
.size(12)
.color(color_text),
button(text(btn_label).size(12))
.on_press(AppMessage::SaveAttachment(att.id))
.style(b_style(
color_blue,
color_lavender,
color_crust,
6.0,
))
.padding(6),
]
.spacing(8)
.align_y(iced::alignment::Vertical::Center)
.into()
};
chat_col = chat_col.push(
row![
iced::widget::Space::new().width(iced::Length::Fixed(30.0)),
elem
]
.spacing(0),
);
}
} }
} }
let chat_scroll = scrollable(chat_col) let chat_scroll = scrollable(chat_col)
@@ -3630,6 +4117,10 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
.height(iced::Length::Fill) .height(iced::Length::Fill)
.anchor_bottom(); .anchor_bottom();
let chat_input_row = row![ let chat_input_row = row![
button(text("📎").size(15))
.on_press(AppMessage::PickAttachmentFile)
.style(b_style(color_surface, color_overlay, color_text, 6.0))
.padding(8),
text_input("Message the room…", &state.chat_input) text_input("Message the room…", &state.chat_input)
.on_input(AppMessage::ChatInputChanged) .on_input(AppMessage::ChatInputChanged)
.on_submit(AppMessage::ChatSubmit) .on_submit(AppMessage::ChatSubmit)
@@ -4900,8 +5391,48 @@ impl Program<AppMessage> for Icon {
mod tests { mod tests {
use super::{ use super::{
format_duration, initial_window_position, reconnect_attempt_chime, reconnected_chime, format_duration, initial_window_position, reconnect_attempt_chime, reconnected_chime,
GateMeter, METER_MAX, set_peer_gate_config, set_peer_volume_config, AppConfig, GateMeter, METER_MAX,
}; };
use iroh::SecretKey;
#[test]
fn peer_gate_persists_when_on_and_clears_when_off() {
let mut config = AppConfig::default();
let id = SecretKey::generate().public();
// A positive threshold is stored, clamped to the slider's METER_MAX ceiling.
let stored = set_peer_gate_config(&mut config, id, 0.05);
assert_eq!(stored, 0.05);
assert_eq!(config.peer_gate.get(&id.to_string()).copied(), Some(0.05));
assert_eq!(set_peer_gate_config(&mut config, id, 99.0), METER_MAX);
// Zero (or negative) means "gate off" — the entry is removed so the
// config doesn't carry a disabled gate.
let stored = set_peer_gate_config(&mut config, id, 0.0);
assert_eq!(stored, 0.0);
assert!(!config.peer_gate.contains_key(&id.to_string()));
}
#[test]
fn peer_volume_persists_non_unity_and_clears_at_unity() {
let mut config = AppConfig::default();
let id = SecretKey::generate().public();
// A non-unity value is clamped into range and stored.
let stored = set_peer_volume_config(&mut config, id, 1.5);
assert_eq!(stored, 1.5);
assert_eq!(config.peer_volume.get(&id.to_string()).copied(), Some(1.5));
// Out-of-range values clamp to the slider bounds.
assert_eq!(set_peer_volume_config(&mut config, id, 5.0), 2.0);
assert_eq!(set_peer_volume_config(&mut config, id, -1.0), 0.0);
// Returning to unity removes the entry (unity is the implicit default),
// so the config doesn't accumulate no-op entries.
let stored = set_peer_volume_config(&mut config, id, 1.0);
assert_eq!(stored, 1.0);
assert!(!config.peer_volume.contains_key(&id.to_string()));
}
#[test] #[test]
fn x11_restores_saved_window_position() { fn x11_restores_saved_window_position() {
@@ -5176,6 +5707,7 @@ mod tests {
text: "Hello".to_string(), text: "Hello".to_string(),
mine: true, mine: true,
from: None, from: None,
attachment: None,
}; };
push_chat(&mut messages, entry); push_chat(&mut messages, entry);
assert_eq!(messages.len(), 1); assert_eq!(messages.len(), 1);
@@ -5196,6 +5728,7 @@ mod tests {
text: format!("Msg{}", i), text: format!("Msg{}", i),
mine: i % 2 == 0, mine: i % 2 == 0,
from: None, from: None,
attachment: None,
}, },
); );
} }
@@ -5219,6 +5752,7 @@ mod tests {
text: format!("Msg{}", i), text: format!("Msg{}", i),
mine: i % 2 == 0, mine: i % 2 == 0,
from: None, from: None,
attachment: None,
}, },
); );
} }
+306
View File
@@ -0,0 +1,306 @@
//! Independent playback engine for inline chat audio attachments.
//!
//! The rodio device sink stays on a dedicated OS thread and never enters iced
//! state or the call-audio pipeline. The GUI sends small commands and reads a
//! shared status snapshot at its redraw cadence.
use crate::files::AttachmentId;
use rodio::{Decoder, DeviceSinkBuilder, MixerDeviceSink, Player, Source, decoder::DecoderError};
use std::io::Cursor;
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;
/// State published by the playback thread for the GUI.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ClipStatus {
pub playing_id: Option<AttachmentId>,
pub position: Duration,
pub total: Option<Duration>,
pub paused: bool,
/// Set when output initialization or decoding rejects the requested clip.
/// The app consumes this as a signal to fall back to the normal file chip.
pub failure: Option<ClipFailure>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClipFailure {
pub id: AttachmentId,
pub error: String,
/// Decoder rejection means the filename hint should fall back to a file
/// chip. Output-device failures remain retryable as audio.
pub invalid_audio: bool,
}
pub type SharedClipStatus = Arc<Mutex<ClipStatus>>;
#[derive(Debug)]
enum ClipCommand {
Play(AttachmentId, Vec<u8>),
Pause,
Resume,
Seek(Duration),
Stop,
}
/// Cheap, `Send` command handle for the dedicated playback thread.
pub struct ClipPlayer {
command_tx: mpsc::Sender<ClipCommand>,
status: SharedClipStatus,
}
impl ClipPlayer {
/// Start the playback worker. The system output device is opened lazily on
/// first Play, so merely launching PeerSpeak never claims another stream.
pub fn new() -> (Self, SharedClipStatus) {
let (command_tx, command_rx) = mpsc::channel();
let status = Arc::new(Mutex::new(ClipStatus::default()));
let worker_status = Arc::clone(&status);
std::thread::Builder::new()
.name("peerspeak-clip-player".to_string())
.spawn(move || playback_worker(command_rx, worker_status))
.expect("failed to spawn clip playback thread");
(
Self {
command_tx,
status: Arc::clone(&status),
},
status,
)
}
pub fn play(&self, id: AttachmentId, bytes: Vec<u8>) {
update_status(&self.status, |status| {
status.playing_id = Some(id);
status.position = Duration::ZERO;
status.total = None;
status.paused = false;
status.failure = None;
});
let _ = self.command_tx.send(ClipCommand::Play(id, bytes));
}
pub fn pause(&self) {
let _ = self.command_tx.send(ClipCommand::Pause);
}
pub fn resume(&self) {
let _ = self.command_tx.send(ClipCommand::Resume);
}
pub fn seek(&self, position: Duration) {
let _ = self.command_tx.send(ClipCommand::Seek(position));
}
pub fn stop(&self) {
let _ = self.command_tx.send(ClipCommand::Stop);
}
}
fn playback_worker(command_rx: mpsc::Receiver<ClipCommand>, status: SharedClipStatus) {
let mut output: Option<MixerDeviceSink> = None;
let mut player: Option<Player> = None;
loop {
match command_rx.recv_timeout(Duration::from_millis(100)) {
Ok(ClipCommand::Play(id, bytes)) => {
// In-memory readers do not expose file metadata to rodio. Pass
// the known attachment length explicitly so formats without a
// duration in their headers (notably MP3 and Vorbis) can derive
// a total duration and support reliable seeking.
let source = match decode_clip(bytes) {
Ok(source) => source,
Err(error) => {
fail(
&status,
id,
format!("unsupported or invalid audio: {error}"),
true,
);
continue;
}
};
let total = source.total_duration();
if output.is_none() {
match DeviceSinkBuilder::open_default_sink() {
Ok(sink) => {
player = Some(Player::connect_new(sink.mixer()));
output = Some(sink);
}
Err(error) => {
fail(
&status,
id,
format!("audio output unavailable: {error}"),
false,
);
continue;
}
}
}
if let Some(player) = player.as_ref() {
player.clear();
player.append(source);
player.play();
update_status(&status, |s| {
s.playing_id = Some(id);
s.position = Duration::ZERO;
s.total = total;
s.paused = false;
s.failure = None;
});
}
}
Ok(ClipCommand::Pause) => {
if let Some(player) = player.as_ref() {
player.pause();
update_status(&status, |s| s.paused = true);
}
}
Ok(ClipCommand::Resume) => {
if let Some(player) = player.as_ref() {
player.play();
update_status(&status, |s| s.paused = false);
}
}
Ok(ClipCommand::Seek(position)) => {
if let Some(player) = player.as_ref()
&& player.try_seek(position).is_ok()
{
update_status(&status, |s| s.position = position);
}
}
Ok(ClipCommand::Stop) => {
if let Some(player) = player.as_ref() {
player.clear();
}
reset(&status);
}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
Err(mpsc::RecvTimeoutError::Timeout) => {}
}
if let Some(player) = player.as_ref() {
let (active, failed) = status
.lock()
.map(|s| (s.playing_id.is_some(), s.failure.is_some()))
.unwrap_or_default();
if active && !failed && player.empty() {
reset(&status);
} else if active && !failed {
update_status(&status, |s| {
s.position = player.get_pos();
s.paused = player.is_paused();
});
}
}
}
}
fn decode_clip(bytes: Vec<u8>) -> Result<Decoder<Cursor<Vec<u8>>>, DecoderError> {
let byte_len = bytes.len() as u64;
Decoder::builder()
.with_data(Cursor::new(bytes))
.with_byte_len(byte_len)
.build()
}
fn fail(status: &SharedClipStatus, id: AttachmentId, error: String, invalid_audio: bool) {
crate::log_msg(&format!("Inline audio playback failed: {error}"));
update_status(status, |s| {
// Keep the id active until the GUI observes the failure on its next
// tick. This guarantees the active-only timer cannot disappear in the
// small window between sending Play and decoder/output failure.
s.playing_id = Some(id);
s.position = Duration::ZERO;
s.total = None;
s.paused = false;
s.failure = Some(ClipFailure {
id,
error,
invalid_audio,
});
});
}
fn reset(status: &SharedClipStatus) {
update_status(status, |s| *s = ClipStatus::default());
}
fn update_status(status: &SharedClipStatus, update: impl FnOnce(&mut ClipStatus)) {
if let Ok(mut status) = status.lock() {
update(&mut status);
}
}
pub fn status_snapshot(status: &SharedClipStatus) -> ClipStatus {
status.lock().map(|s| s.clone()).unwrap_or_default()
}
/// Format clip time as `mm:ss` (hours are folded into minutes).
pub fn format_time(duration: Duration) -> String {
let seconds = duration.as_secs();
format!("{}:{:02}", seconds / 60, seconds % 60)
}
/// Playback progress in `0.0..=1.0`; unknown and zero durations report zero.
pub fn progress(position: Duration, total: Option<Duration>) -> f32 {
let Some(total) = total.filter(|duration| !duration.is_zero()) else {
return 0.0;
};
(position.as_secs_f64() / total.as_secs_f64()).clamp(0.0, 1.0) as f32
}
/// Convert a slider fraction into a clamped position within a clip.
pub fn seek_target(fraction: f32, total: Duration) -> Duration {
total.mul_f64(f64::from(fraction.clamp(0.0, 1.0)))
}
#[cfg(test)]
mod tests {
use super::*;
use base64::Engine;
#[test]
fn in_memory_mp3_reports_duration() {
// One headerless constant-bitrate MP3 frame repeated to model files
// that do not carry an Xing/VBR duration header.
let frame = base64::engine::general_purpose::STANDARD
.decode("//sQxAAABIQVWVRggDCqCKiDNlAAAAGgS4BgAmTT2AQAABCxOD5d7gQOfqBAEHS4Ph/EAIRI7//0A0KBNpABgMRIDCSI04PcIFdF0PJKFgzlUf5eAoF8BRIPfh4FTvUDQl+dUi5pc0w=")
.expect("valid test fixture");
let bytes = frame.repeat(20);
let decoder = decode_clip(bytes).expect("CBR MP3 should decode");
assert!(decoder.total_duration().is_some());
}
#[test]
fn formats_clip_time() {
assert_eq!(format_time(Duration::ZERO), "0:00");
assert_eq!(format_time(Duration::from_secs(65)), "1:05");
assert_eq!(format_time(Duration::from_secs(3_661)), "61:01");
}
#[test]
fn progress_handles_unknown_zero_and_clamps() {
assert_eq!(progress(Duration::from_secs(1), None), 0.0);
assert_eq!(progress(Duration::from_secs(1), Some(Duration::ZERO)), 0.0);
assert_eq!(
progress(Duration::from_secs(5), Some(Duration::from_secs(10))),
0.5
);
assert_eq!(
progress(Duration::from_secs(20), Some(Duration::from_secs(10))),
1.0
);
}
#[test]
fn seek_target_clamps_fraction() {
let total = Duration::from_secs(100);
assert_eq!(seek_target(0.25, total), Duration::from_secs(25));
assert_eq!(seek_target(-1.0, total), Duration::ZERO);
assert_eq!(seek_target(2.0, total), total);
}
}
+1
View File
@@ -56,6 +56,7 @@ pub trait AudioBackend: Send + Sync {
fn stop(&self) -> Result<(), AudioError>; fn stop(&self) -> Result<(), AudioError>;
} }
pub mod clip_player;
pub mod eq; pub mod eq;
pub mod gate; pub mod gate;
pub mod limiter; pub mod limiter;
+13
View File
@@ -255,6 +255,15 @@ pub struct AppConfig {
/// keyed by peer node id string. Local preference only. /// keyed by peer node id string. Local preference only.
#[serde(default)] #[serde(default)]
pub peer_pan: HashMap<String, f32>, pub peer_pan: HashMap<String, f32>,
/// Per-peer listener-side volume/gain (`1.0` = unity), keyed by peer node id
/// string. Local preference only; never sent to peers. Absent entry = unity.
#[serde(default)]
pub peer_volume: HashMap<String, f32>,
/// Per-peer listener-side noise-gate threshold (normalized RMS, `0.0` = off),
/// keyed by peer node id string. Local preference only; never sent to peers.
/// Absent entry = gate disabled (pass-through).
#[serde(default)]
pub peer_gate: HashMap<String, f32>,
/// Focused app-local keyboard shortcuts. /// Focused app-local keyboard shortcuts.
#[serde(default)] #[serde(default)]
pub hotkeys: crate::hotkeys::HotkeyMap, pub hotkeys: crate::hotkeys::HotkeyMap,
@@ -317,6 +326,8 @@ impl Default for AppConfig {
recents: Vec::new(), recents: Vec::new(),
peer_eq: HashMap::new(), peer_eq: HashMap::new(),
peer_pan: HashMap::new(), peer_pan: HashMap::new(),
peer_volume: HashMap::new(),
peer_gate: HashMap::new(),
hotkeys: crate::hotkeys::HotkeyMap::default(), hotkeys: crate::hotkeys::HotkeyMap::default(),
window_width: default_window_width(), window_width: default_window_width(),
window_height: default_window_height(), window_height: default_window_height(),
@@ -457,6 +468,8 @@ mod tests {
// shortcut settings. // shortcut settings.
assert!(deserialized.peer_eq.is_empty()); assert!(deserialized.peer_eq.is_empty());
assert!(deserialized.peer_pan.is_empty()); assert!(deserialized.peer_pan.is_empty());
assert!(deserialized.peer_volume.is_empty());
assert!(deserialized.peer_gate.is_empty());
assert_eq!( assert_eq!(
crate::hotkeys::format_binding( crate::hotkeys::format_binding(
deserialized deserialized
+19 -1
View File
@@ -26,6 +26,10 @@ pub enum CoreCommand {
SetPeerEq(EndpointId, crate::audio::eq::EqSettings), SetPeerEq(EndpointId, crate::audio::eq::EqSettings),
/// Listener-side per-peer pan. Local only; never leaves this app instance. /// Listener-side per-peer pan. Local only; never leaves this app instance.
SetPeerPan(EndpointId, f32), SetPeerPan(EndpointId, f32),
/// Listener-side per-peer noise gate threshold (normalized RMS, `0.0` = off).
/// Applies the same smooth gate as the mic path to a peer's incoming audio,
/// to suppress their background noise on our end. Local only.
SetPeerGate(EndpointId, f32),
/// Locally mute/unmute a peer: when muted, their audio is decoded (so levels /// Locally mute/unmute a peer: when muted, their audio is decoded (so levels
/// still show) but not mixed into our output. /// still show) but not mixed into our output.
SetPeerMuted(EndpointId, bool), SetPeerMuted(EndpointId, bool),
@@ -49,6 +53,14 @@ pub enum CoreCommand {
SetRecordingMode(RecordingMode), SetRecordingMode(RecordingMode),
/// Broadcast a room text-chat message. No-op when not in a call. /// Broadcast a room text-chat message. No-op when not in a call.
SendChat(String), SendChat(String),
/// Send a chat message carrying a file attachment. The app has already read +
/// capped the file and built the descriptor; core makes the bytes available
/// on the file plane and broadcasts the descriptor.
SendChatFile { text: String, attachment: crate::files::ChatAttachment, data: Vec<u8> },
/// Fetch a received attachment's bytes from its sender over the file plane
/// (used for on-demand file/chip downloads; images are auto-fetched on
/// receipt). Replies with `AttachmentReady`/`AttachmentFailed`.
FetchAttachment { from: EndpointId, attachment: crate::files::ChatAttachment },
/// Set the pixelpass binary location (config override, empty = use `$PATH`). /// Set the pixelpass binary location (config override, empty = use `$PATH`).
/// Sent at startup so screen-share can resolve the binary. /// Sent at startup so screen-share can resolve the binary.
SetPixelpassPath(Option<String>), SetPixelpassPath(Option<String>),
@@ -105,7 +117,13 @@ pub enum UiEvent {
/// A room text-chat message arrived from a peer (never our own — local /// A room text-chat message arrived from a peer (never our own — local
/// messages are echoed by the UI on send). `from` is the sender's node id /// messages are echoed by the UI on send). `from` is the sender's node id
/// string, used to key their avatar (W4). /// string, used to key their avatar (W4).
ChatMessage { from: String, name: String, text: String }, ChatMessage { from: String, name: String, text: String, attachment: Option<crate::files::ChatAttachment> },
/// An attachment's bytes are now available (auto-fetched for images, or
/// fetched on demand for files). Keyed by attachment id so the UI can match
/// it to the chat entry.
AttachmentReady { id: crate::files::AttachmentId, data: Vec<u8> },
/// An attachment fetch failed (sender gone, too large, decode error, etc.).
AttachmentFailed { id: crate::files::AttachmentId, error: String },
/// Our own screen share started; the UI flips the Share button to "Stop". /// Our own screen share started; the UI flips the Share button to "Stop".
ScreenShareStarted, ScreenShareStarted,
/// Our own screen share stopped (or failed to start). /// Our own screen share stopped (or failed to start).
+137 -3
View File
@@ -8,7 +8,7 @@ use crate::codec::{AudioEncoder, opus_impl::OpusEncoder};
use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES}; use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES};
use crate::network::{ use crate::network::{
NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket, NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket,
iroh_impl::{IrohTransport, AudioRouter}, iroh_impl::{IrohTransport, AudioRouter, FileRouter},
gossip::IrohGossipState, gossip::IrohGossipState,
}; };
use crate::core::messages::{CoreCommand, UiEvent}; use crate::core::messages::{CoreCommand, UiEvent};
@@ -578,6 +578,9 @@ struct NetStack {
/// The persistent inbound-audio handler on `router`; per-join we bind the /// The persistent inbound-audio handler on `router`; per-join we bind the
/// active session's transport into it, and clear it on leave. /// active session's transport into it, and clear it on leave.
audio_router: AudioRouter, audio_router: AudioRouter,
/// The persistent chat-file-transfer handler on `router`; bound/cleared in
/// lock-step with `audio_router` (same session lifecycle).
file_router: FileRouter,
/// In-memory address book (ticket + gossip fed), shared with every session. /// In-memory address book (ticket + gossip fed), shared with every session.
memory_lookup: iroh::address_lookup::memory::MemoryLookup, memory_lookup: iroh::address_lookup::memory::MemoryLookup,
} }
@@ -689,6 +692,7 @@ async fn build_net_stack(
.spawn(endpoint.clone()); .spawn(endpoint.clone());
let audio_router = AudioRouter::new(); let audio_router = AudioRouter::new();
let file_router = FileRouter::new();
// The friends presence listener (W7 B2) rides this same persistent router as a // The friends presence listener (W7 B2) rides this same persistent router as a
// third ALPN — it MUST be a handler here, not a standalone accept loop, since // third ALPN — it MUST be a handler here, not a standalone accept loop, since
// the router owns endpoint.accept(). Policy (who we answer / what room we // the router owns endpoint.accept(). Policy (who we answer / what room we
@@ -696,6 +700,7 @@ async fn build_net_stack(
let router = Router::builder(endpoint.clone()) let router = Router::builder(endpoint.clone())
.accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone()) .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
.accept(crate::protocol::AUDIO_ALPN, audio_router.clone()) .accept(crate::protocol::AUDIO_ALPN, audio_router.clone())
.accept(crate::protocol::FILES_ALPN, file_router.clone())
.accept( .accept(
crate::presence_net::FRIENDS_ALPN, crate::presence_net::FRIENDS_ALPN,
crate::presence_net::FriendsProtocol::new(friends_handler), crate::presence_net::FriendsProtocol::new(friends_handler),
@@ -707,10 +712,49 @@ async fn build_net_stack(
gossip, gossip,
router, router,
audio_router, audio_router,
file_router,
memory_lookup, memory_lookup,
}) })
} }
/// Fetch a chat attachment's bytes from `from` over the file plane in a detached
/// task, then report the result to the UI via [`UiEvent::AttachmentReady`] /
/// [`UiEvent::AttachmentFailed`] keyed by the attachment id. For images
/// (`is_image`) the bytes are defensively re-validated (decodable + within pixel
/// limits) before being handed to the renderer; an "Image" that doesn't decode is
/// reported as a failure rather than rendered.
fn spawn_attachment_fetch(
transport: Arc<IrohTransport>,
ui_tx: mpsc::Sender<UiEvent>,
from: EndpointId,
att: crate::files::ChatAttachment,
is_image: bool,
) {
tokio::spawn(async move {
match transport.fetch_attachment(from, &att).await {
Ok(data) => {
if is_image && crate::files::validate_image_bytes(&data).is_none() {
let _ = ui_tx
.send(UiEvent::AttachmentFailed {
id: att.id,
error: "received image failed to decode".to_string(),
})
.await;
return;
}
let _ = ui_tx
.send(UiEvent::AttachmentReady { id: att.id, data })
.await;
}
Err(e) => {
let _ = ui_tx
.send(UiEvent::AttachmentFailed { id: att.id, error: e.to_string() })
.await;
}
}
});
}
/// Finalize and clear the active recording, if any, emitting `RecordingStopped`. /// Finalize and clear the active recording, if any, emitting `RecordingStopped`.
/// No-op when not recording. Called on stop, room leave, and room switch so a /// No-op when not recording. Called on stop, room leave, and room switch so a
/// recording is always closed cleanly (its WAV size fields patched). /// recording is always closed cleanly (its WAV size fields patched).
@@ -869,6 +913,8 @@ async fn run_core_loop(
let peer_volumes = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new())); let peer_volumes = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
let peer_eq = Arc::new(Mutex::new(HashMap::<EndpointId, EqSettings>::new())); let peer_eq = Arc::new(Mutex::new(HashMap::<EndpointId, EqSettings>::new()));
let peer_pan = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new())); let peer_pan = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
// Listener-side per-peer noise-gate thresholds (normalized RMS, 0.0 = off).
let peer_gate = Arc::new(Mutex::new(HashMap::<EndpointId, f32>::new()));
// Peers locally muted by us: decoded for level metering but not mixed. // Peers locally muted by us: decoded for level metering but not mixed.
let locally_muted = Arc::new(Mutex::new(HashSet::<EndpointId>::new())); let locally_muted = Arc::new(Mutex::new(HashSet::<EndpointId>::new()));
let mut current_name = "Anonymous".to_string(); let mut current_name = "Anonymous".to_string();
@@ -1069,6 +1115,7 @@ async fn run_core_loop(
if let Some(session) = active_session.take() { if let Some(session) = active_session.take() {
session.shutdown(audio_backend.clone()).await; session.shutdown(audio_backend.clone()).await;
net.audio_router.clear(); net.audio_router.clear();
net.file_router.clear();
} }
*current_room.lock().unwrap() = None; *current_room.lock().unwrap() = None;
@@ -1091,6 +1138,7 @@ async fn run_core_loop(
crate::log_msg("Shutting down existing active session"); crate::log_msg("Shutting down existing active session");
session.shutdown(audio_backend.clone()).await; session.shutdown(audio_backend.clone()).await;
net.audio_router.clear(); net.audio_router.clear();
net.file_router.clear();
} }
// If a network-mode / identity change was deferred while a call was // If a network-mode / identity change was deferred while a call was
@@ -1147,6 +1195,7 @@ async fn run_core_loop(
// NetStack; the session just subscribes its topic below. // NetStack; the session just subscribes its topic below.
let transport = Arc::new(IrohTransport::new(endpoint.clone())); let transport = Arc::new(IrohTransport::new(endpoint.clone()));
net.audio_router.bind(&transport); net.audio_router.bind(&transport);
net.file_router.bind(&transport);
let room_state = Arc::new(IrohGossipState::new( let room_state = Arc::new(IrohGossipState::new(
endpoint.clone(), endpoint.clone(),
@@ -1191,6 +1240,7 @@ async fn run_core_loop(
crate::log_msg(&format!("Error room_state.join failed: {:?}", e)); crate::log_msg(&format!("Error room_state.join failed: {:?}", e));
let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await; let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await;
net.audio_router.clear(); net.audio_router.clear();
net.file_router.clear();
continue; continue;
} }
crate::log_msg("Joined room successfully via room_state"); crate::log_msg("Joined room successfully via room_state");
@@ -1243,6 +1293,7 @@ async fn run_core_loop(
let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await; let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
let _ = room_state.leave().await; let _ = room_state.leave().await;
net.audio_router.clear(); net.audio_router.clear();
net.file_router.clear();
continue; continue;
} }
@@ -1255,6 +1306,7 @@ async fn run_core_loop(
let _ = audio_backend.stop(); let _ = audio_backend.stop();
let _ = room_state.leave().await; let _ = room_state.leave().await;
net.audio_router.clear(); net.audio_router.clear();
net.file_router.clear();
continue; continue;
} }
@@ -1397,6 +1449,7 @@ async fn run_core_loop(
let peer_volumes_mixer = peer_volumes.clone(); let peer_volumes_mixer = peer_volumes.clone();
let peer_eq_mixer = peer_eq.clone(); let peer_eq_mixer = peer_eq.clone();
let peer_pan_mixer = peer_pan.clone(); let peer_pan_mixer = peer_pan.clone();
let peer_gate_mixer = peer_gate.clone();
let locally_muted_mixer = locally_muted.clone(); let locally_muted_mixer = locally_muted.clone();
let output_gain_mixer = output_gain.clone(); let output_gain_mixer = output_gain.clone();
let ui_tx_mixer = ui_tx.clone(); let ui_tx_mixer = ui_tx.clone();
@@ -1413,6 +1466,11 @@ async fn run_core_loop(
// Per-peer EQ filter state. Settings are live-cloned each // Per-peer EQ filter state. Settings are live-cloned each
// cycle; state is rebuilt only when a peer's EQ changes. // cycle; state is rebuilt only when a peer's EQ changes.
let mut peer_eqs: HashMap<EndpointId, Eq> = HashMap::new(); let mut peer_eqs: HashMap<EndpointId, Eq> = HashMap::new();
// Per-peer noise-gate envelope state. The threshold is passed
// per frame (live slider), so the gate is never rebuilt — only
// created once per peer and dropped when the peer leaves.
let mut peer_noise_gates: HashMap<EndpointId, crate::audio::gate::NoiseGate> =
HashMap::new();
// When the ring is at/above target we have nothing to do; nap // When the ring is at/above target we have nothing to do; nap
// briefly and re-check. Short enough (relative to the ~60ms // briefly and re-check. Short enough (relative to the ~60ms
// target and ~21ms device quantum) that we always refill well // target and ~21ms device quantum) that we always refill well
@@ -1438,6 +1496,7 @@ async fn run_core_loop(
let current_volumes = peer_volumes_mixer.lock().await.clone(); let current_volumes = peer_volumes_mixer.lock().await.clone();
let current_eq = peer_eq_mixer.lock().await.clone(); let current_eq = peer_eq_mixer.lock().await.clone();
let current_pans = peer_pan_mixer.lock().await.clone(); let current_pans = peer_pan_mixer.lock().await.clone();
let current_gates = peer_gate_mixer.lock().await.clone();
let muted_peers = locally_muted_mixer.lock().await.clone(); let muted_peers = locally_muted_mixer.lock().await.clone();
let mut peer_frames: Vec<(Vec<i16>, f32)> = Vec::new(); let mut peer_frames: Vec<(Vec<i16>, f32)> = Vec::new();
let mut peers_seen = HashSet::new(); let mut peers_seen = HashSet::new();
@@ -1463,6 +1522,26 @@ async fn run_core_loop(
stems.push((peer_id, frame.clone())); stems.push((peer_id, frame.clone()));
} }
// Listener-side per-peer noise gate, applied to the
// raw decoded frame (after the clean stem tap, before
// volume/EQ) so the threshold tracks the peer's true
// signal level regardless of our volume setting. The
// gate's "should transmit" return is irrelevant here —
// we only attenuate. Threshold 0 = off; the gate is
// created lazily and dropped when disabled.
let gate_threshold =
current_gates.get(&peer_id).copied().unwrap_or(0.0);
if gate_threshold > 0.0 {
peer_noise_gates
.entry(peer_id)
.or_insert_with(|| {
crate::audio::gate::NoiseGate::new(48_000)
})
.process(&mut frame, gate_threshold);
} else {
peer_noise_gates.remove(&peer_id);
}
let vol = current_volumes.get(&peer_id).copied().unwrap_or(1.0); let vol = current_volumes.get(&peer_id).copied().unwrap_or(1.0);
apply_volume(&mut frame, vol); apply_volume(&mut frame, vol);
@@ -1507,6 +1586,8 @@ async fn run_core_loop(
} }
} }
peer_eqs.retain(|id, _| peers_seen.contains(id) || current_eq.contains_key(id)); peer_eqs.retain(|id, _| peers_seen.contains(id) || current_eq.contains_key(id));
peer_noise_gates
.retain(|id, _| peers_seen.contains(id) || current_gates.contains_key(id));
// Lossless i32 sum, then the limiter applies the master // Lossless i32 sum, then the limiter applies the master
// output gain (in f32, so a boost past the ceiling is // output gain (in f32, so a boost past the ceiling is
@@ -1691,11 +1772,27 @@ async fn run_core_loop(
.insert(peer_id, state.addr.clone()); .insert(peer_id, state.addr.clone());
let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await; let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await;
} }
RoomEvent::ChatMessage { from, name, text, .. } => { RoomEvent::ChatMessage { from, name, text, ts: _, attachment } => {
// Auto-fetch image attachments so they render inline
// without a click; non-image files wait for an explicit
// FetchAttachment (the "Save" chip). The descriptor was
// already filename-sanitized + size-capped on ingest.
if let Some(att) = attachment.clone()
&& att.kind == crate::files::AttachmentKind::Image
{
spawn_attachment_fetch(
transport_events.clone(),
ui_tx_events.clone(),
from,
att,
true,
);
}
let _ = ui_tx_events.send(UiEvent::ChatMessage { let _ = ui_tx_events.send(UiEvent::ChatMessage {
from: from.to_string(), from: from.to_string(),
name, name,
text, text,
attachment,
}).await; }).await;
} }
RoomEvent::PeerConnectionLost(peer_id) => { RoomEvent::PeerConnectionLost(peer_id) => {
@@ -1798,6 +1895,7 @@ async fn run_core_loop(
session.shutdown(audio_backend.clone()).await; session.shutdown(audio_backend.clone()).await;
// Stop routing inbound audio links — the endpoint/router stay up. // Stop routing inbound audio links — the endpoint/router stay up.
net.audio_router.clear(); net.audio_router.clear();
net.file_router.clear();
// No longer in a gathering — friends who ping see us as just online. // No longer in a gathering — friends who ping see us as just online.
*current_room.lock().unwrap() = None; *current_room.lock().unwrap() = None;
let _ = ui_tx.send(UiEvent::RoomLeft).await; let _ = ui_tx.send(UiEvent::RoomLeft).await;
@@ -1884,6 +1982,16 @@ async fn run_core_loop(
} }
} }
CoreCommand::SetPeerGate(peer_id, threshold) => {
let threshold = threshold.clamp(0.0, 1.0);
let mut guard = peer_gate.lock().await;
if threshold <= 0.0 {
guard.remove(&peer_id);
} else {
guard.insert(peer_id, threshold);
}
}
CoreCommand::SetPeerMuted(peer_id, muted) => { CoreCommand::SetPeerMuted(peer_id, muted) => {
let mut guard = locally_muted.lock().await; let mut guard = locally_muted.lock().await;
if muted { if muted {
@@ -2167,12 +2275,38 @@ async fn run_core_loop(
CoreCommand::SendChat(text) => { CoreCommand::SendChat(text) => {
if let Some(session) = &active_session if let Some(session) = &active_session
&& let Err(e) = session.room_state.send_chat(text).await && let Err(e) = session.room_state.send_chat(text, None).await
{ {
crate::log_msg(&format!("Failed to send chat: {e}")); crate::log_msg(&format!("Failed to send chat: {e}"));
} }
} }
CoreCommand::SendChatFile { text, attachment, data } => {
if let Some(session) = &active_session {
// Make the bytes fetchable by room members, then broadcast the
// descriptor alongside the (possibly empty) caption text.
session
.transport
.serve_attachment(attachment.id, Arc::new(data));
if let Err(e) = session.room_state.send_chat(text, Some(attachment)).await {
crate::log_msg(&format!("Failed to send chat file: {e}"));
}
}
}
CoreCommand::FetchAttachment { from, attachment } => {
if let Some(session) = &active_session {
let is_image = attachment.kind == crate::files::AttachmentKind::Image;
spawn_attachment_fetch(
session.transport.clone(),
ui_tx.clone(),
from,
attachment,
is_image,
);
}
}
CoreCommand::SetPixelpassPath(path) => { CoreCommand::SetPixelpassPath(path) => {
pixelpass_override = path.filter(|p| !p.trim().is_empty()); pixelpass_override = path.filter(|p| !p.trim().is_empty());
} }
+368
View File
@@ -0,0 +1,368 @@
//! Chat file attachments: the compact descriptor that rides a gossip chat
//! message, plus the pure validation/sanitization seams for the file-transfer
//! plane.
//!
//! Attachment **bytes do not travel over gossip** — gossip is a small-frame
//! broadcast plane (see `avatar` for why image bytes there are hard-capped to
//! tens of KB). Instead a chat message carries a [`ChatAttachment`] *descriptor*
//! (name, size, kind, id); the sender serves the actual bytes over the dedicated
//! file ALPN (`protocol::FILES_ALPN`) via direct QUIC streams, and recipients
//! fetch them point-to-point. Everything in this module is dependency-light and
//! pure so it can be unit-tested away from the network and the GUI.
use serde::{Deserialize, Serialize};
/// Hard ceiling on a single attachment's byte size. Bounds the memory a peer can
/// make us hold (when fetching) or serve, and the time a transfer can take.
/// 25 MiB comfortably covers phone photos and ordinary documents.
pub const MAX_ATTACHMENT_BYTES: u64 = 25 * 1024 * 1024;
/// Max decoded pixels per side for an inline image preview. Defends against a
/// decode-bomb (a small file that expands to an enormous bitmap), independent of
/// the byte cap. Applied via `image::Limits` when validating/decoding.
pub const MAX_IMAGE_PX: u32 = 4096;
/// Longest filename we keep and display. Keeps the gossip descriptor compact and
/// the UI tidy; the real bytes are unaffected.
pub const MAX_FILENAME_LEN: usize = 96;
/// A 32-byte opaque id identifying one attachment for the fetch request. Minted
/// randomly per attachment by the sender (see core); the transfer itself is
/// authenticated + encrypted + room-member gated, so the id only needs to be a
/// hard-to-guess handle into the sender's serve store, not a content hash.
pub type AttachmentId = [u8; 32];
/// How the receiver should present an attachment. A *hint* derived from the
/// sender's content sniff — never trusted for a safety decision. The receiver
/// re-validates image bytes itself before decoding, and falls back to a file
/// chip if an "Image" doesn't actually decode.
#[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Debug)]
pub enum AttachmentKind {
Image,
File,
}
/// The descriptor carried inside a `GossipMessage::Chat`. Compact by design: it
/// holds no file bytes, only what the UI needs to render a placeholder/chip and
/// what a fetch needs to pull the bytes.
#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, Debug)]
pub struct ChatAttachment {
/// Sanitized display filename (already path-stripped — see
/// [`sanitize_filename`]). Never used as a filesystem path on receipt without
/// the user choosing a save location.
pub name: String,
/// Byte length of the file. Bounds the fetch read; must be
/// `<= MAX_ATTACHMENT_BYTES` (enforced by [`size_within_cap`]).
pub size: u64,
/// Presentation hint (image vs. generic file).
pub kind: AttachmentKind,
/// Opaque handle the receiver writes on the file plane to request the bytes.
pub id: AttachmentId,
}
/// Sanitize an arbitrary (possibly hostile) filename for display and as a
/// save-dialog default. Strips any directory component (both `/` and `\`),
/// removes control characters, collapses whitespace, trims, caps the length
/// while trying to preserve a short extension, and rejects the `.`/`..` traps.
/// Always returns a non-empty, path-component-free name (falls back to `file`).
pub fn sanitize_filename(raw: &str) -> String {
// Take only the final *non-empty* path component, defeating
// `../../etc/passwd`, `C:\foo\bar`, embedded separators, and trailing slashes
// (`a/b/c/` → `c`).
let base = raw
.rsplit(['/', '\\'])
.find(|s| !s.trim().is_empty())
.unwrap_or("")
.trim();
// Drop control chars; turn other whitespace into single spaces later.
let cleaned: String = base
.chars()
.filter(|c| !c.is_control())
.collect();
let collapsed = cleaned.split_whitespace().collect::<Vec<_>>().join(" ");
let collapsed = collapsed.trim_matches('.').trim();
if collapsed.is_empty() {
return "file".to_string();
}
if collapsed.chars().count() <= MAX_FILENAME_LEN {
return collapsed.to_string();
}
// Too long: keep the extension (if short + sane) and truncate the stem.
if let Some((stem, ext)) = collapsed.rsplit_once('.')
&& !ext.is_empty()
&& ext.chars().count() <= 8
&& ext.chars().all(|c| c.is_ascii_alphanumeric())
{
let keep = MAX_FILENAME_LEN.saturating_sub(ext.chars().count() + 1);
let truncated: String = stem.chars().take(keep).collect();
return format!("{truncated}.{ext}");
}
collapsed.chars().take(MAX_FILENAME_LEN).collect()
}
/// Whether a declared/observed size is within the transfer cap and non-zero.
/// Used both when sending (reject before serving) and when fetching (reject a
/// descriptor before opening a stream).
pub fn size_within_cap(size: u64) -> bool {
size > 0 && size <= MAX_ATTACHMENT_BYTES
}
/// Sniff the leading bytes for a known image container, to set the attachment
/// *kind* hint at send time. Recognizes PNG, JPEG, GIF, WebP, and BMP. This is a
/// presentation hint only — actual inline rendering still depends on the bytes
/// decoding (we only build image features for PNG/JPEG), with a file-chip
/// fallback otherwise.
pub fn is_probably_image(bytes: &[u8]) -> bool {
let b = bytes;
let png = b.starts_with(&[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A]);
let jpeg = b.starts_with(&[0xFF, 0xD8, 0xFF]);
let gif = b.starts_with(b"GIF87a") || b.starts_with(b"GIF89a");
let bmp = b.starts_with(b"BM");
let webp = b.len() >= 12 && b.starts_with(b"RIFF") && &b[8..12] == b"WEBP";
png || jpeg || gif || bmp || webp
}
/// Sniff the leading bytes for an audio container supported by the inline clip
/// player. Audio remains [`AttachmentKind::File`] on the wire; this receiver-side
/// check confirms that a filename-based player hint actually contains WAV, MP3,
/// Ogg Vorbis, or FLAC data before playback is attempted.
pub fn is_probably_audio(bytes: &[u8]) -> bool {
let flac = bytes.starts_with(b"fLaC");
let ogg = bytes.starts_with(b"OggS");
let wav = bytes.len() >= 12 && bytes.starts_with(b"RIFF") && &bytes[8..12] == b"WAVE";
let mp3_id3 = bytes.starts_with(b"ID3");
let mp3_frame = bytes.len() >= 2 && bytes[0] == 0xFF && bytes[1] & 0xE0 == 0xE0;
flac || ogg || wav || mp3_id3 || mp3_frame
}
/// Whether a sanitized attachment name has an extension supported by the
/// inline audio player. This is only a pre-fetch presentation hint; fetched
/// bytes are confirmed with [`is_probably_audio`] before being decoded.
pub fn looks_like_audio_name(name: &str) -> bool {
let Some((_, extension)) = name.rsplit_once('.') else {
return false;
};
matches!(extension.to_ascii_lowercase().as_str(), "wav" | "mp3" | "ogg" | "oga" | "flac")
}
/// The attachment kind for some file bytes: [`AttachmentKind::Image`] if it
/// sniffs as an image container, else [`AttachmentKind::File`].
pub fn classify(bytes: &[u8]) -> AttachmentKind {
if is_probably_image(bytes) {
AttachmentKind::Image
} else {
AttachmentKind::File
}
}
/// Defensively decode image bytes under strict pixel limits to confirm they're a
/// real, sane image before we hand them to the renderer. Returns the decoded
/// dimensions on success. Guards against decode-bombs (small file → huge bitmap)
/// regardless of [`MAX_ATTACHMENT_BYTES`]. Only PNG/JPEG are buildable in our
/// `image` feature set; anything else returns `None` and the caller shows a chip.
pub fn validate_image_bytes(bytes: &[u8]) -> Option<(u32, u32)> {
let mut limits = image::Limits::default();
limits.max_image_width = Some(MAX_IMAGE_PX);
limits.max_image_height = Some(MAX_IMAGE_PX);
let reader = image::ImageReader::new(std::io::Cursor::new(bytes))
.with_guessed_format()
.ok()?;
let mut reader = reader;
reader.limits(limits);
let img = reader.decode().ok()?;
let (w, h) = (img.width(), img.height());
if w == 0 || h == 0 || w > MAX_IMAGE_PX || h > MAX_IMAGE_PX {
return None;
}
Some((w, h))
}
/// Parse a file-plane request: it must be exactly one [`AttachmentId`] (32
/// bytes). Anything else is rejected so a peer can't send a malformed/oversized
/// request frame. Pure half of the serve handler.
pub fn parse_request(bytes: &[u8]) -> Option<AttachmentId> {
if bytes.len() != 32 {
return None;
}
let mut id = [0u8; 32];
id.copy_from_slice(bytes);
Some(id)
}
/// A human-readable size like `2.3 MB` / `812 KB` / `40 B` for the file chip.
pub fn human_size(bytes: u64) -> String {
const KB: u64 = 1024;
const MB: u64 = 1024 * KB;
if bytes >= MB {
format!("{:.1} MB", bytes as f64 / MB as f64)
} else if bytes >= KB {
format!("{:.0} KB", bytes as f64 / KB as f64)
} else {
format!("{bytes} B")
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sanitize_strips_directory_traversal() {
assert_eq!(sanitize_filename("../../etc/passwd"), "passwd");
assert_eq!(sanitize_filename("/abs/path/photo.png"), "photo.png");
assert_eq!(sanitize_filename(r"C:\Users\me\secret.doc"), "secret.doc");
assert_eq!(sanitize_filename("a/b/c/"), "c");
}
#[test]
fn sanitize_rejects_dot_traps_and_empty() {
assert_eq!(sanitize_filename(""), "file");
assert_eq!(sanitize_filename("."), "file");
assert_eq!(sanitize_filename(".."), "file");
assert_eq!(sanitize_filename(" "), "file");
assert_eq!(sanitize_filename("/"), "file");
}
#[test]
fn sanitize_removes_control_chars_and_collapses_ws() {
// Control chars (incl. tab/newline) are stripped entirely.
assert_eq!(sanitize_filename("my\tphoto\n.png"), "myphoto.png");
assert_eq!(sanitize_filename("a\u{0000}b.txt"), "ab.txt");
// Real spaces are collapsed but preserved.
assert_eq!(sanitize_filename("my photo .png"), "my photo .png");
}
#[test]
fn sanitize_caps_length_preserving_extension() {
let long_stem = "x".repeat(200);
let name = format!("{long_stem}.png");
let out = sanitize_filename(&name);
assert!(out.chars().count() <= MAX_FILENAME_LEN, "len was {}", out.chars().count());
assert!(out.ends_with(".png"), "extension preserved: {out}");
}
#[test]
fn size_cap_bounds() {
assert!(!size_within_cap(0));
assert!(size_within_cap(1));
assert!(size_within_cap(MAX_ATTACHMENT_BYTES));
assert!(!size_within_cap(MAX_ATTACHMENT_BYTES + 1));
}
#[test]
fn image_sniffing_recognizes_containers() {
assert!(is_probably_image(&[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A, 0, 0]));
assert!(is_probably_image(&[0xFF, 0xD8, 0xFF, 0xE0]));
assert!(is_probably_image(b"GIF89a...."));
let mut webp = b"RIFF".to_vec();
webp.extend_from_slice(&[0, 0, 0, 0]);
webp.extend_from_slice(b"WEBP");
assert!(is_probably_image(&webp));
assert!(!is_probably_image(b"%PDF-1.7"));
assert!(!is_probably_image(b""));
}
#[test]
fn audio_sniffing_recognizes_supported_containers() {
assert!(is_probably_audio(b"fLaC\0\0\0\x22"));
assert!(is_probably_audio(b"OggS\0\x02"));
let mut wav = b"RIFF".to_vec();
wav.extend_from_slice(&[0, 0, 0, 0]);
wav.extend_from_slice(b"WAVE");
assert!(is_probably_audio(&wav));
assert!(is_probably_audio(b"ID3\x04\0\0"));
assert!(is_probably_audio(&[0xFF, 0xFB, 0x90, 0x64]));
}
#[test]
fn audio_sniffing_disambiguates_wav_from_webp() {
let mut wav = b"RIFF".to_vec();
wav.extend_from_slice(&[0, 0, 0, 0]);
wav.extend_from_slice(b"WAVE");
assert!(is_probably_audio(&wav));
assert!(!is_probably_image(&wav));
let mut webp = b"RIFF".to_vec();
webp.extend_from_slice(&[0, 0, 0, 0]);
webp.extend_from_slice(b"WEBP");
assert!(is_probably_image(&webp));
assert!(!is_probably_audio(&webp));
}
#[test]
fn audio_sniffing_rejects_non_audio() {
assert!(!is_probably_audio(b"%PDF-1.7"));
assert!(!is_probably_audio(&[0x89, b'P', b'N', b'G']));
assert!(!is_probably_audio(&[]));
assert!(!is_probably_audio(&[0xFF]));
}
#[test]
fn audio_name_detection_is_case_insensitive() {
for name in ["clip.wav", "clip.mp3", "clip.ogg", "clip.oga", "clip.flac"] {
assert!(looks_like_audio_name(name), "{name}");
}
assert!(looks_like_audio_name("VOICE.MP3"));
assert!(looks_like_audio_name("mix.FlAc"));
assert!(!looks_like_audio_name("recording"));
assert!(!looks_like_audio_name("notes.pdf"));
assert!(!looks_like_audio_name("photo.webp"));
}
#[test]
fn classify_maps_sniff_to_kind() {
assert_eq!(classify(&[0xFF, 0xD8, 0xFF]), AttachmentKind::Image);
assert_eq!(classify(b"fLaC\0\0\0\x22"), AttachmentKind::File);
assert_eq!(classify(b"plain text"), AttachmentKind::File);
}
#[test]
fn parse_request_requires_exact_32_bytes() {
assert_eq!(parse_request(&[7u8; 32]), Some([7u8; 32]));
assert_eq!(parse_request(&[7u8; 31]), None);
assert_eq!(parse_request(&[7u8; 33]), None);
assert_eq!(parse_request(&[]), None);
}
#[test]
fn validate_image_rejects_garbage() {
assert_eq!(validate_image_bytes(b"not an image"), None);
assert_eq!(validate_image_bytes(&[]), None);
}
#[test]
fn validate_image_accepts_a_real_png() {
// Encode a tiny PNG in-memory, then validate it.
let img = image::RgbImage::from_pixel(4, 3, image::Rgb([10, 20, 30]));
let mut buf = std::io::Cursor::new(Vec::new());
image::DynamicImage::ImageRgb8(img)
.write_to(&mut buf, image::ImageFormat::Png)
.unwrap();
assert_eq!(validate_image_bytes(&buf.into_inner()), Some((4, 3)));
}
#[test]
fn human_size_units() {
assert_eq!(human_size(40), "40 B");
assert_eq!(human_size(2048), "2 KB");
assert_eq!(human_size(3 * 1024 * 1024 + 300 * 1024), "3.3 MB");
}
#[test]
fn attachment_descriptor_round_trips_json() {
let a = ChatAttachment {
name: "photo.png".to_string(),
size: 12345,
kind: AttachmentKind::Image,
id: [9u8; 32],
};
let bytes = serde_json::to_vec(&a).unwrap();
let back: ChatAttachment = serde_json::from_slice(&bytes).unwrap();
assert_eq!(a, back);
}
}
+1
View File
@@ -19,6 +19,7 @@ pub mod background;
pub mod recents; pub mod recents;
pub mod discovery; pub mod discovery;
pub mod hotkeys; pub mod hotkeys;
pub mod files;
use std::fs::File; use std::fs::File;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
+74 -13
View File
@@ -184,9 +184,16 @@ fn compute_bootstrap(
pub enum GossipMessage { pub enum GossipMessage {
Announce(PeerState), Announce(PeerState),
Leave, Leave,
/// A room text-chat message: the author's display name, the text, and a /// A room text-chat message: the author's display name, the text, a
/// sender-stamped millisecond timestamp. /// sender-stamped millisecond timestamp, and an optional file attachment
Chat { name: String, text: String, ts: u64 }, /// descriptor (the bytes are fetched off-gossip on the file plane).
Chat {
name: String,
text: String,
ts: u64,
#[serde(default)]
attachment: Option<crate::files::ChatAttachment>,
},
} }
pub struct IrohGossipState { pub struct IrohGossipState {
@@ -438,13 +445,24 @@ impl RoomState for IrohGossipState {
let _ = event_tx.send(RoomEvent::PeerLeft(payload.author)).await; let _ = event_tx.send(RoomEvent::PeerLeft(payload.author)).await;
} }
} }
GossipMessage::Chat { name, text, ts } => { GossipMessage::Chat { name, text, ts, attachment } => {
crate::log_msg(&format!("Gossip chat from author={:?}", payload.author)); crate::log_msg(&format!("Gossip chat from author={:?}", payload.author));
// Defensively normalize an untrusted attachment
// descriptor: sanitize the filename and drop it
// entirely if it declares an out-of-cap size.
let attachment = attachment.and_then(|mut a| {
if !crate::files::size_within_cap(a.size) {
return None;
}
a.name = crate::files::sanitize_filename(&a.name);
Some(a)
});
let _ = event_tx.send(RoomEvent::ChatMessage { let _ = event_tx.send(RoomEvent::ChatMessage {
from: payload.author, from: payload.author,
name, name,
text, text,
ts, ts,
attachment,
}).await; }).await;
} }
} }
@@ -565,7 +583,11 @@ impl RoomState for IrohGossipState {
} }
} }
async fn send_chat(&self, text: String) -> Result<(), NetError> { async fn send_chat(
&self,
text: String,
attachment: Option<crate::files::ChatAttachment>,
) -> Result<(), NetError> {
let name = { let name = {
let guard = self.self_state.lock().unwrap(); let guard = self.self_state.lock().unwrap();
match guard.as_ref() { match guard.as_ref() {
@@ -582,7 +604,7 @@ impl RoomState for IrohGossipState {
&self.secret_key, &self.secret_key,
&topic, &topic,
ts, ts,
GossipMessage::Chat { name, text, ts }, GossipMessage::Chat { name, text, ts, attachment },
); );
if let Ok(bytes) = serde_json::to_vec(&payload) { if let Ok(bytes) = serde_json::to_vec(&payload) {
sender.broadcast(bytes.into()).await sender.broadcast(bytes.into()).await
@@ -744,13 +766,15 @@ mod tests {
name: "Alice".to_string(), name: "Alice".to_string(),
text: "Hello".to_string(), text: "Hello".to_string(),
ts: 123456789, ts: 123456789,
attachment: None,
}; };
let serialized = serde_json::to_string(&original).unwrap(); let serialized = serde_json::to_string(&original).unwrap();
let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap(); let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
if let GossipMessage::Chat { name, text, ts } = deserialized { if let GossipMessage::Chat { name, text, ts, attachment } = deserialized {
assert_eq!(name, "Alice"); assert_eq!(name, "Alice");
assert_eq!(text, "Hello"); assert_eq!(text, "Hello");
assert_eq!(ts, 123456789); assert_eq!(ts, 123456789);
assert_eq!(attachment, None);
} else { } else {
panic!("Expected GossipMessage::Chat"); panic!("Expected GossipMessage::Chat");
} }
@@ -760,10 +784,11 @@ mod tests {
name: "".to_string(), name: "".to_string(),
text: "".to_string(), text: "".to_string(),
ts: u64::MAX, ts: u64::MAX,
attachment: None,
}; };
let serialized_empty = serde_json::to_string(&original_empty).unwrap(); let serialized_empty = serde_json::to_string(&original_empty).unwrap();
let deserialized_empty: GossipMessage = serde_json::from_str(&serialized_empty).unwrap(); let deserialized_empty: GossipMessage = serde_json::from_str(&serialized_empty).unwrap();
if let GossipMessage::Chat { name, text, ts } = deserialized_empty { if let GossipMessage::Chat { name, text, ts, .. } = deserialized_empty {
assert_eq!(name, ""); assert_eq!(name, "");
assert_eq!(text, ""); assert_eq!(text, "");
assert_eq!(ts, u64::MAX); assert_eq!(ts, u64::MAX);
@@ -772,6 +797,40 @@ mod tests {
} }
} }
#[test]
fn test_gossip_chat_attachment_round_trip_and_back_compat() {
let att = crate::files::ChatAttachment {
name: "photo.png".to_string(),
size: 4096,
kind: crate::files::AttachmentKind::Image,
id: [42u8; 32],
};
let original = GossipMessage::Chat {
name: "Alice".to_string(),
text: "look at this".to_string(),
ts: 1,
attachment: Some(att.clone()),
};
let serialized = serde_json::to_string(&original).unwrap();
let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
if let GossipMessage::Chat { attachment, .. } = deserialized {
assert_eq!(attachment, Some(att));
} else {
panic!("Expected GossipMessage::Chat");
}
// A pre-v2 chat payload (no `attachment` field) must still deserialize,
// defaulting the attachment to None (serde(default)).
let legacy = r#"{"Chat":{"name":"Old","text":"hi","ts":7}}"#;
let parsed: GossipMessage = serde_json::from_str(legacy).unwrap();
if let GossipMessage::Chat { name, attachment, .. } = parsed {
assert_eq!(name, "Old");
assert_eq!(attachment, None);
} else {
panic!("Expected GossipMessage::Chat");
}
}
#[test] #[test]
fn test_gossip_payload_chat_round_trip() { fn test_gossip_payload_chat_round_trip() {
let secret = SecretKey::generate(); let secret = SecretKey::generate();
@@ -784,6 +843,7 @@ mod tests {
name: "Bob".to_string(), name: "Bob".to_string(),
text: "Hi there".to_string(), text: "Hi there".to_string(),
ts: 987654321, ts: 987654321,
attachment: None,
}, },
); );
@@ -791,7 +851,7 @@ mod tests {
let deserialized: GossipPayload = serde_json::from_str(&serialized).unwrap(); let deserialized: GossipPayload = serde_json::from_str(&serialized).unwrap();
assert_eq!(deserialized.author, secret.public()); assert_eq!(deserialized.author, secret.public());
if let GossipMessage::Chat { name, text, ts } = deserialized.msg { if let GossipMessage::Chat { name, text, ts, .. } = deserialized.msg {
assert_eq!(name, "Bob"); assert_eq!(name, "Bob");
assert_eq!(text, "Hi there"); assert_eq!(text, "Hi there");
assert_eq!(ts, 987654321); assert_eq!(ts, 987654321);
@@ -806,10 +866,11 @@ mod tests {
name: "🎙 User".to_string(), name: "🎙 User".to_string(),
text: "héllo 🎙 世界".to_string(), text: "héllo 🎙 世界".to_string(),
ts: 1717171717, ts: 1717171717,
attachment: None,
}; };
let serialized = serde_json::to_string(&original).unwrap(); let serialized = serde_json::to_string(&original).unwrap();
let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap(); let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
if let GossipMessage::Chat { name, text, ts } = deserialized { if let GossipMessage::Chat { name, text, ts, .. } = deserialized {
assert_eq!(name, "🎙 User"); assert_eq!(name, "🎙 User");
assert_eq!(text, "héllo 🎙 世界"); assert_eq!(text, "héllo 🎙 世界");
assert_eq!(ts, 1717171717); assert_eq!(ts, 1717171717);
@@ -849,7 +910,7 @@ mod tests {
let secret = SecretKey::generate(); let secret = SecretKey::generate();
let topic = [4u8; 32]; let topic = [4u8; 32];
let mut p = sign_gossip(&secret, &topic, 5_000, GossipMessage::Leave); let mut p = sign_gossip(&secret, &topic, 5_000, GossipMessage::Leave);
p.msg = GossipMessage::Chat { name: "x".into(), text: "y".into(), ts: 5_000 }; p.msg = GossipMessage::Chat { name: "x".into(), text: "y".into(), ts: 5_000, attachment: None };
assert_eq!( assert_eq!(
verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS), verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
Err(GossipReject::BadSignature) Err(GossipReject::BadSignature)
@@ -922,8 +983,8 @@ mod tests {
fn state_mutation_replay_gate_leaves_chat_ordering_untouched() { fn state_mutation_replay_gate_leaves_chat_ordering_untouched() {
let author = fresh_id(); let author = fresh_id();
let mut seen = HashMap::new(); let mut seen = HashMap::new();
let later_chat = GossipMessage::Chat { name: "A".into(), text: "later".into(), ts: 200 }; let later_chat = GossipMessage::Chat { name: "A".into(), text: "later".into(), ts: 200, attachment: None };
let earlier_chat = GossipMessage::Chat { name: "A".into(), text: "earlier".into(), ts: 100 }; let earlier_chat = GossipMessage::Chat { name: "A".into(), text: "earlier".into(), ts: 100, attachment: None };
assert!(admit_state_mutation(&mut seen, author, &later_chat, 200)); assert!(admit_state_mutation(&mut seen, author, &later_chat, 200));
assert!(admit_state_mutation(&mut seen, author, &earlier_chat, 100)); assert!(admit_state_mutation(&mut seen, author, &earlier_chat, 100));
+157 -1
View File
@@ -9,7 +9,8 @@ use std::collections::{HashMap, HashSet};
use std::time::Duration; use std::time::Duration;
use async_trait::async_trait; use async_trait::async_trait;
use crate::protocol::AUDIO_ALPN; use crate::protocol::{AUDIO_ALPN, FILES_ALPN};
use crate::files::{AttachmentId, ChatAttachment};
/// Per-peer datagram send queue depth. Audio is real-time, so a backlog is /// Per-peer datagram send queue depth. Audio is real-time, so a backlog is
/// useless latency — keep it shallow and drop the oldest frame when full. /// useless latency — keep it shallow and drop the oldest frame when full.
@@ -31,6 +32,10 @@ const MAX_BACKOFF: Duration = Duration::from_secs(5);
/// already means "the peer closed this on purpose." /// already means "the peer closed this on purpose."
const GOODBYE_CODE: u32 = 1; const GOODBYE_CODE: u32 = 1;
/// Bound on each phase (connect, read) of a chat-attachment fetch, so a slow or
/// stalled sender can't hang the fetch indefinitely.
const FILE_FETCH_TIMEOUT: Duration = Duration::from_secs(30);
/// State shared between the transport, its protocol handler, and every per-peer /// State shared between the transport, its protocol handler, and every per-peer
/// supervisor task. One supervisor owns a peer's whole connection lifecycle. /// supervisor task. One supervisor owns a peer's whole connection lifecycle.
struct Shared { struct Shared {
@@ -60,6 +65,11 @@ struct Shared {
/// verified gossip roster plus peers still inside reconnect grace; transport /// verified gossip roster plus peers still inside reconnect grace; transport
/// connections alone never mutate this set. /// connections alone never mutate this set.
admitted_audio: StdMutex<HashSet<EndpointId>>, admitted_audio: StdMutex<HashSet<EndpointId>>,
/// Chat file attachments we're serving to room members this session, keyed by
/// the random attachment id. Populated when we send a chat file; read by the
/// file protocol handler to answer a member's fetch. Cleared on leave. Each
/// blob is already byte-capped at send time.
served_files: StdMutex<HashMap<crate::files::AttachmentId, Arc<Vec<u8>>>>,
incoming_tx: mpsc::Sender<(EndpointId, Bytes)>, incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
/// Best-effort link-state notifications for the UI (connecting / connected). /// Best-effort link-state notifications for the UI (connecting / connected).
conn_events_tx: mpsc::Sender<ConnEvent>, conn_events_tx: mpsc::Sender<ConnEvent>,
@@ -402,6 +412,97 @@ impl iroh::protocol::ProtocolHandler for AudioRouter {
} }
} }
/// Protocol handler for the file-transfer plane (`FILES_ALPN`). Mirrors
/// [`AudioRouter`]: it's persistent on the router and bound to the active
/// session's [`Shared`] on join. On an inbound stream it authenticates the peer
/// (iroh ALPN handshake gives us `remote_id`), gates on **live room membership**
/// (same invariant as audio admission, so a former member can't pull files),
/// reads a single 32-byte attachment id, and streams back the matching blob from
/// the session serve store — or nothing if the id is unknown.
#[derive(Clone, Default)]
pub struct FileRouter {
current: Arc<StdMutex<Option<Arc<Shared>>>>,
}
impl std::fmt::Debug for FileRouter {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FileRouter").finish_non_exhaustive()
}
}
impl FileRouter {
pub fn new() -> Self {
Self::default()
}
/// Route inbound file fetches to `transport`'s session (called on join).
pub fn bind(&self, transport: &IrohTransport) {
*self.current.lock().unwrap() = Some(transport.shared.clone());
}
/// Stop serving files until the next [`bind`](Self::bind) (called on leave).
pub fn clear(&self) {
*self.current.lock().unwrap() = None;
}
}
/// Max bytes we'll read for a fetch *request* frame. A request is exactly one
/// 32-byte id; this small ceiling rejects a peer trying to stream us a huge
/// "request" as a cheap DoS.
const FILE_REQUEST_MAX: usize = 64;
impl iroh::protocol::ProtocolHandler for FileRouter {
fn accept(
&self,
connection: Connection,
) -> impl std::future::Future<Output = Result<(), iroh::protocol::AcceptError>> + Send {
let peer_id = connection.remote_id();
let shared = self.current.lock().unwrap().clone();
async move {
// No active call → nothing to serve.
let Some(shared) = shared else {
return Ok(());
};
// Member gating: only current room members may fetch our files. Reuses
// the audio admission roster (the authoritative room membership set).
if !shared.audio_sender_admitted(peer_id) {
crate::log_msg(&format!(
"Transport: rejected file fetch from non-member {}",
crate::short_id(&peer_id.to_string())
));
return Ok(());
}
// Accept one bidirectional stream: read the id, write the bytes.
let Ok((mut send, mut recv)) = connection.accept_bi().await else {
return Ok(());
};
let Ok(req) = recv.read_to_end(FILE_REQUEST_MAX).await else {
return Ok(());
};
let Some(id) = crate::files::parse_request(&req) else {
return Ok(());
};
let blob = shared.served_files.lock().unwrap().get(&id).cloned();
if let Some(blob) = blob {
let _ = send.write_all(&blob).await;
}
// Finish either way: an unknown id closes with an empty body, which
// the fetcher reads as a zero-length result and treats as "gone".
let _ = send.finish();
// CRITICAL: `finish()` only marks the stream's EOF — it does NOT wait
// for the written bytes to be delivered and acknowledged. If we return
// here the `connection` drops, and its CONNECTION_CLOSE can race ahead
// of the still-in-flight stream data, so the fetcher's read aborts with
// "connection lost". Wait for the fetcher to receive everything and
// close the connection itself (it drops `conn` right after read_to_end);
// that close is our signal the transfer landed. Bounded so a fetcher
// that vanishes can't pin this task forever.
let _ = tokio::time::timeout(FILE_FETCH_TIMEOUT, connection.closed()).await;
Ok(())
}
}
}
pub struct IrohTransport { pub struct IrohTransport {
shared: Arc<Shared>, shared: Arc<Shared>,
incoming_rx: tokio::sync::Mutex<Option<mpsc::Receiver<(EndpointId, Bytes)>>>, incoming_rx: tokio::sync::Mutex<Option<mpsc::Receiver<(EndpointId, Bytes)>>>,
@@ -426,6 +527,7 @@ impl IrohTransport {
peers: tokio::sync::Mutex::new(HashMap::new()), peers: tokio::sync::Mutex::new(HashMap::new()),
live_conns: StdMutex::new(HashMap::new()), live_conns: StdMutex::new(HashMap::new()),
admitted_audio: StdMutex::new(HashSet::new()), admitted_audio: StdMutex::new(HashSet::new()),
served_files: StdMutex::new(HashMap::new()),
incoming_tx, incoming_tx,
conn_events_tx, conn_events_tx,
}); });
@@ -455,6 +557,7 @@ impl IrohTransport {
self.shared.senders.lock().unwrap().clear(); self.shared.senders.lock().unwrap().clear();
self.shared.addrs.lock().unwrap().clear(); self.shared.addrs.lock().unwrap().clear();
self.shared.admitted_audio.lock().unwrap().clear(); self.shared.admitted_audio.lock().unwrap().clear();
self.shared.served_files.lock().unwrap().clear();
// Give the CONNECTION_CLOSE frames a moment to flush before the caller // Give the CONNECTION_CLOSE frames a moment to flush before the caller
// shuts the endpoint/router down (the `conns` clones are still alive // shuts the endpoint/router down (the `conns` clones are still alive
// here, so the endpoint can still transmit them). // here, so the endpoint can still transmit them).
@@ -483,6 +586,59 @@ impl IrohTransport {
pub fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool { pub fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool {
self.shared.audio_sender_admitted(peer_id) self.shared.audio_sender_admitted(peer_id)
} }
/// Make `bytes` available to room members under `id` for the rest of this
/// session (served by the [`FileRouter`] handler). Called by core when we
/// send a chat file. The blob is cleared on leave.
pub fn serve_attachment(&self, id: AttachmentId, bytes: Arc<Vec<u8>>) {
self.shared.served_files.lock().unwrap().insert(id, bytes);
}
/// Fetch a chat attachment's bytes from its sender over the file plane. Dials
/// the sender on `FILES_ALPN` (preferring a known full address), writes the
/// 32-byte id, and reads the response bounded by the descriptor's declared
/// size (which the caller has already validated against the global cap). The
/// read limit means a malicious sender can't stream us more than advertised.
pub async fn fetch_attachment(
&self,
from: EndpointId,
att: &ChatAttachment,
) -> Result<Vec<u8>, NetError> {
if !crate::files::size_within_cap(att.size) {
return Err(NetError::Other("attachment size out of range".to_string()));
}
let addr = self.shared.addrs.lock().unwrap().get(&from).cloned();
let connect = async {
match addr {
Some(addr) => self.shared.endpoint.connect(addr, FILES_ALPN).await,
None => self.shared.endpoint.connect(from, FILES_ALPN).await,
}
};
let conn = tokio::time::timeout(FILE_FETCH_TIMEOUT, connect)
.await
.map_err(|_| NetError::Other("file fetch: connect timed out".to_string()))?
.map_err(|e| NetError::Other(format!("file fetch: connect failed: {e}")))?;
let (mut send, mut recv) = conn
.open_bi()
.await
.map_err(|e| NetError::Other(format!("file fetch: open stream failed: {e}")))?;
send.write_all(&att.id)
.await
.map_err(|e| NetError::Other(format!("file fetch: request write failed: {e}")))?;
send.finish()
.map_err(|e| NetError::Other(format!("file fetch: request finish failed: {e}")))?;
let read = recv.read_to_end(att.size as usize);
let bytes = tokio::time::timeout(FILE_FETCH_TIMEOUT, read)
.await
.map_err(|_| NetError::Other("file fetch: read timed out".to_string()))?
.map_err(|e| NetError::Other(format!("file fetch: read failed: {e}")))?;
if bytes.is_empty() {
return Err(NetError::Other("file fetch: sender no longer has the file".to_string()));
}
Ok(bytes)
}
} }
#[async_trait] #[async_trait]
+16 -3
View File
@@ -57,7 +57,15 @@ pub enum RoomEvent {
/// A peer sent a room text-chat message. Carries the sender's id, their /// A peer sent a room text-chat message. Carries the sender's id, their
/// display name (embedded so it shows even without a presence entry), the /// display name (embedded so it shows even without a presence entry), the
/// text, and a sender-stamped millisecond timestamp. /// text, and a sender-stamped millisecond timestamp.
ChatMessage { from: EndpointId, name: String, text: String, ts: u64 }, ChatMessage {
from: EndpointId,
name: String,
text: String,
ts: u64,
/// Optional file attachment descriptor; the bytes are fetched off-gossip
/// on the file plane. Already filename-sanitized + size-capped on ingest.
attachment: Option<crate::files::ChatAttachment>,
},
} }
/// Transport-level link state for a peer, surfaced so the UI can show when a /// Transport-level link state for a peer, surfaced so the UI can show when a
@@ -206,8 +214,13 @@ pub trait RoomState: Send + Sync {
fn mark_peer_disconnected(&self, peer_id: EndpointId); fn mark_peer_disconnected(&self, peer_id: EndpointId);
/// Broadcasts a room text-chat message authored by us (our display name is /// Broadcasts a room text-chat message authored by us (our display name is
/// taken from the current self-state). /// taken from the current self-state), optionally carrying a file attachment
async fn send_chat(&self, text: String) -> Result<(), NetError>; /// descriptor whose bytes are served separately on the file plane.
async fn send_chat(
&self,
text: String,
attachment: Option<crate::files::ChatAttachment>,
) -> Result<(), NetError>;
/// Leaves the room and announces departure. /// Leaves the room and announces departure.
async fn leave(&self) -> Result<(), NetError>; async fn leave(&self) -> Result<(), NetError>;
+13 -2
View File
@@ -22,16 +22,26 @@ pub const FRIENDS_PROTO: u32 = 1;
/// Gossip plane version (`GossipPayload`/`GossipMessage`/`PeerState`, signing, /// Gossip plane version (`GossipPayload`/`GossipMessage`/`PeerState`, signing,
/// freshness). Bump on any change. Mirrored in [`GOSSIP_SIG_DOMAIN`] and folded /// freshness). Bump on any change. Mirrored in [`GOSSIP_SIG_DOMAIN`] and folded
/// into [`versioned_topic`]. /// into [`versioned_topic`].
pub const GOSSIP_PROTO: u32 = 1; ///
/// v2 (0.3.0): `GossipMessage::Chat` gained an optional file attachment
/// (`ChatAttachment`), so a pre-v2 peer can't interpret/serve chat files — bumped
/// to fail fast rather than half-work.
pub const GOSSIP_PROTO: u32 = 2;
/// File-transfer plane version (chat attachment request/stream shape). Bump on
/// any change. Mirrored in [`FILES_ALPN`].
pub const FILES_PROTO: u32 = 1;
/// ALPN for the audio datagram plane: `peerspeak/audio/<AUDIO_PROTO>`. /// ALPN for the audio datagram plane: `peerspeak/audio/<AUDIO_PROTO>`.
pub const AUDIO_ALPN: &[u8] = b"peerspeak/audio/1"; pub const AUDIO_ALPN: &[u8] = b"peerspeak/audio/1";
/// ALPN for the friends/presence plane: `peerspeak/friends/<FRIENDS_PROTO>`. /// ALPN for the friends/presence plane: `peerspeak/friends/<FRIENDS_PROTO>`.
pub const FRIENDS_ALPN: &[u8] = b"peerspeak/friends/1"; pub const FRIENDS_ALPN: &[u8] = b"peerspeak/friends/1";
/// ALPN for the file-transfer plane: `peerspeak/files/<FILES_PROTO>`. Carries
/// chat attachment bytes via direct QUIC streams (not gossip).
pub const FILES_ALPN: &[u8] = b"peerspeak/files/1";
/// ed25519 gossip signature domain: `peerspeak-gossip-v<GOSSIP_PROTO>`. Carries /// ed25519 gossip signature domain: `peerspeak-gossip-v<GOSSIP_PROTO>`. Carries
/// the gossip protocol version into every signed payload — a version mismatch /// the gossip protocol version into every signed payload — a version mismatch
/// fails verification (cryptographic separation between gossip versions). /// fails verification (cryptographic separation between gossip versions).
pub const GOSSIP_SIG_DOMAIN: &str = "peerspeak-gossip-v1"; pub const GOSSIP_SIG_DOMAIN: &str = "peerspeak-gossip-v2";
/// Version-namespace a room topic so peers on different gossip protocol versions /// Version-namespace a room topic so peers on different gossip protocol versions
/// derive **different subscription topics from the same ticket** and therefore /// derive **different subscription topics from the same ticket** and therefore
@@ -62,6 +72,7 @@ mod tests {
fn alpns_match_their_proto_versions() { fn alpns_match_their_proto_versions() {
assert_eq!(AUDIO_ALPN, format!("peerspeak/audio/{AUDIO_PROTO}").as_bytes()); assert_eq!(AUDIO_ALPN, format!("peerspeak/audio/{AUDIO_PROTO}").as_bytes());
assert_eq!(FRIENDS_ALPN, format!("peerspeak/friends/{FRIENDS_PROTO}").as_bytes()); assert_eq!(FRIENDS_ALPN, format!("peerspeak/friends/{FRIENDS_PROTO}").as_bytes());
assert_eq!(FILES_ALPN, format!("peerspeak/files/{FILES_PROTO}").as_bytes());
assert_eq!(GOSSIP_SIG_DOMAIN, format!("peerspeak-gossip-v{GOSSIP_PROTO}")); assert_eq!(GOSSIP_SIG_DOMAIN, format!("peerspeak-gossip-v{GOSSIP_PROTO}"));
} }
+141
View File
@@ -0,0 +1,141 @@
//! End-to-end loopback test for the chat file-transfer plane (`FILES_ALPN`).
//!
//! Spins up two real iroh endpoints on localhost (relay disabled, addresses
//! exchanged directly), registers the production [`FileRouter`] on each, serves a
//! multi-megabyte blob on one side, and fetches it from the other through the
//! real `serve_attachment`/`fetch_attachment` path.
//!
//! This is the regression guard for the "file fetch: read failed: connection
//! lost" bug: the serve handler used to return (and drop the connection) the
//! instant it called `finish()`, so the CONNECTION_CLOSE raced ahead of the
//! still-in-flight stream data and the fetcher's `read_to_end` aborted. A blob
//! large enough to span many packets makes that race deterministic — the fix
//! (waiting on `connection.closed()` before returning) keeps the link up until
//! the fetcher has the bytes.
use std::sync::Arc;
use std::time::Duration;
use iroh::address_lookup::memory::MemoryLookup;
use iroh::endpoint::presets;
use iroh::protocol::Router;
use iroh::{Endpoint, RelayMode};
use peerspeak::files::{ChatAttachment, AttachmentKind};
use peerspeak::network::NetworkTransport;
use peerspeak::network::iroh_impl::{FileRouter, IrohTransport};
use peerspeak::protocol::FILES_ALPN;
struct Node {
endpoint: Endpoint,
transport: Arc<IrohTransport>,
_router: Router,
lookup: MemoryLookup,
}
async fn spawn_node() -> Node {
let lookup = MemoryLookup::new();
let endpoint = Endpoint::builder(presets::Minimal)
.secret_key(iroh::SecretKey::generate())
.relay_mode(RelayMode::Disabled)
.address_lookup(lookup.clone())
.bind()
.await
.expect("bind endpoint");
let transport = Arc::new(IrohTransport::new(endpoint.clone()));
// Mirror production: a persistent FileRouter bound to this session's transport
// is what the router accepts inbound file fetches on.
let file_router = FileRouter::new();
file_router.bind(&transport);
let router = Router::builder(endpoint.clone())
.accept(FILES_ALPN, file_router)
.spawn();
Node { endpoint, transport, _router: router, lookup }
}
/// A pseudo-random-ish multi-megabyte payload spanning many QUIC packets, so a
/// premature connection close on the serve side reliably corrupts/aborts the read.
fn big_blob() -> Vec<u8> {
(0..(2 * 1024 * 1024u32))
.map(|i| (i.wrapping_mul(2654435761) >> 13) as u8)
.collect()
}
#[tokio::test]
async fn loopback_attachment_round_trips_intact() {
let server = spawn_node().await;
let client = spawn_node().await;
// Seed each side with the other's full address so direct dialing works.
server.lookup.add_endpoint_info(client.endpoint.addr());
client.lookup.add_endpoint_info(server.endpoint.addr());
let server_id = server.endpoint.id();
let client_id = client.endpoint.id();
// The serve handler gates on room membership (the audio admission roster), so
// the server must admit the client before it will answer the fetch.
server.transport.admit_audio_sender(client_id);
client.transport.admit_audio_sender(server_id);
// The fetcher dials the retained full address; seed it so fetch_attachment
// doesn't have to fall back to a bare-id lookup.
client.transport.connect_peer(server.endpoint.addr()).await;
let blob = big_blob();
let id = [42u8; 32];
server.transport.serve_attachment(id, Arc::new(blob.clone()));
let att = ChatAttachment {
name: "exterior-landscape.jpg".to_string(),
size: blob.len() as u64,
kind: AttachmentKind::Image,
id,
};
let fetched = tokio::time::timeout(
Duration::from_secs(30),
client.transport.fetch_attachment(server_id, &att),
)
.await
.expect("fetch did not time out")
.expect("fetch succeeded");
assert_eq!(fetched.len(), blob.len(), "fetched the full blob");
assert_eq!(fetched, blob, "fetched bytes match served bytes exactly");
}
#[tokio::test]
async fn loopback_unknown_id_reports_gone() {
let server = spawn_node().await;
let client = spawn_node().await;
server.lookup.add_endpoint_info(client.endpoint.addr());
client.lookup.add_endpoint_info(server.endpoint.addr());
let server_id = server.endpoint.id();
let client_id = client.endpoint.id();
server.transport.admit_audio_sender(client_id);
client.transport.admit_audio_sender(server_id);
client.transport.connect_peer(server.endpoint.addr()).await;
// Never served — the handler closes with an empty body and the fetcher must
// surface that as an error, not hang or return empty bytes.
let att = ChatAttachment {
name: "missing.bin".to_string(),
size: 4096,
kind: AttachmentKind::File,
id: [7u8; 32],
};
let result = tokio::time::timeout(
Duration::from_secs(30),
client.transport.fetch_attachment(server_id, &att),
)
.await
.expect("fetch did not time out");
assert!(result.is_err(), "unknown id should error, got {result:?}");
}