feat(app): one window at a time, share link as an argument

A second launch hands its request to the open window and exits: over a
Unix socket in the user's runtime directory, or a named pipe on Windows
whose name carries a random key from the user's config directory, so
another account cannot take the name first. The open window comes forward,
and a share link given as the first such argument opens the Add form with
it filled in. Setup failures are logged and the app runs without the lock.

README.md

@@ -35,8 +35,12 @@ Its toolchain, emulator, signing, and release instructions are in [`docs/ANDROID
 
 ```bash
 cargo run --release -p pstr-app          # the app
+proton-stream 'https://drive.proton.me/urls/TOKEN#fragment'   # open the Add form with a link
 ```
 
+One window at a time: a second launch brings the open window forward and
+hands it any share link it was given, then exits.
+
 The `pstr` CLI does the same work headlessly, and is what tells you which layer
 is broken:
 
@@ -106,6 +110,9 @@ see [`docs/ANDROID.md`](docs/ANDROID.md) for exact setup and release signing.
 - The block cache holds decrypted content. It lives in your cache directory,
   can be deleted at any time the app is not running, and can be sized or
   emptied from Settings → Storage.
+- A share link given as a command-line argument is visible to other users of
+  the machine in the process list while that launch runs, and may land in
+  shell history. Pasting it into the window (Ctrl+V) does neither.
 - The desktop's media widget is handed the title, the episode and a `file://`
   path to artwork already on disk — never a provider URL, so the shell never
   contacts a metadata provider on its own.

crates/pstr-app/Cargo.toml

@@ -38,5 +38,7 @@ souvlaki.workspace = true
 dbus.workspace = true
 
 [target.'cfg(windows)'.dependencies]
+# Named pipes, for handing a second launch's request to the open window.
+tokio = { workspace = true, features = ["net"] }
 windows-sys = { workspace = true, features = ["Win32_System_Power"] }
 raw-window-handle = "0.6"

crates/pstr-app/src/app.rs

@@ -417,6 +417,8 @@ pub struct ShareForm {
 pub struct App {
     engine: Engine,
     events: Receiver<Event>,
+    /// What later launches asked for. See [`crate::instance`].
+    requests: Receiver<crate::instance::Request>,
     pub page: Page,
     pub library: Library,
     pub shares: Vec<Share>,
@@ -498,7 +500,10 @@ impl App {
         cc: &eframe::CreationContext<'_>,
         runtime: std::sync::Arc<tokio::runtime::Runtime>,
         dirs: AppDirs,
+        instance: crate::instance::Listener,
+        request: crate::instance::Request,
     ) -> anyhow::Result<Self> {
+        let requests = instance.serve(&runtime, &cc.egui_ctx, request);
         // After the engine, not before: the engine is what reads the stored
         // theme, and a window that paints one frame in the default palette
         // before switching is a window that flashes on every launch.
@@ -518,6 +523,7 @@ impl App {
         Ok(Self {
             engine,
             events,
+            requests,
             page: Page::Library,
             library: Library::default(),
             shares: Vec::new(),
@@ -1571,6 +1577,15 @@ impl eframe::App for App {
         self.window.observe(ctx);
 
         let mut actions: Vec<Action> = Vec::new();
+        while let Ok(request) = self.requests.try_recv() {
+            // A second launch is someone looking for this window, so it comes
+            // forward even when there is nothing else to do.
+            ctx.send_viewport_cmd(egui::ViewportCommand::Minimized(false));
+            ctx.send_viewport_cmd(egui::ViewportCommand::Focus);
+            if let crate::instance::Request::AddShare(url) = request {
+                actions.push(Action::PasteShare(url));
+            }
+        }
         self.media.drain(self.playback.as_ref(), &mut actions);
         let cover = self.media_cover();
         self.media.show(self.playback.as_ref(), cover.as_deref());

crates/pstr-app/src/instance.rs

@@ -0,0 +1,350 @@
+//! One window at a time.
+//!
+//! A second launch hands what it was started with — a share link, or nothing
+//! — to the window already open, and exits. Without this, a second click on
+//! the launcher opened a second window on the same catalog: two engines
+//! writing the same watch state, two crawls of the same shares, and two films
+//! playing over each other.
+//!
+//! The channel between the two is a Unix socket in the user's runtime
+//! directory, or a named pipe on Windows. Anything that goes wrong setting it
+//! up is logged and the app runs on without it: a second window is a nuisance,
+//! and a launch that refuses to start because of this is a bug.
+
+use std::sync::mpsc::{Receiver, Sender, channel};
+
+use pstr_core::config::AppDirs;
+
+/// What a later launch asks of the window already open.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Request {
+    /// Come to the front.
+    Raise,
+    /// Come to the front, with this share link in the Add form.
+    AddShare(String),
+}
+
+impl Request {
+    /// What a launch was started with: the first argument that is a share
+    /// link, so `proton-stream 'https://drive.proton.me/urls/…#…'` adds it.
+    pub fn from_args(args: impl IntoIterator<Item = String>) -> Self {
+        args.into_iter()
+            .map(|arg| arg.trim().to_owned())
+            .find(|arg| pstr_core::shares::share_token(arg).is_ok())
+            .map_or(Self::Raise, Self::AddShare)
+    }
+
+    /// One line on the wire.
+    fn encode(&self) -> String {
+        match self {
+            Self::Raise => "raise\n".to_owned(),
+            Self::AddShare(url) => format!("add {url}\n"),
+        }
+    }
+
+    /// The other end of [`Request::encode`]. Anything else is ignored rather
+    /// than guessed at — and a link that is not a share link is not one to
+    /// put in the form.
+    fn decode(line: &str) -> Option<Self> {
+        let line = line.trim();
+        if line == "raise" {
+            return Some(Self::Raise);
+        }
+        let url = line.strip_prefix("add ")?.trim();
+        pstr_core::shares::share_token(url)
+            .is_ok()
+            .then(|| Self::AddShare(url.to_owned()))
+    }
+}
+
+/// How long a launch spends handing its request over before giving up on it.
+const HAND_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
+/// The most a request may be. A share link is a few hundred bytes.
+const MOST: u64 = 8 * 1024;
+
+/// The outcome of [`claim`].
+pub enum Claim {
+    /// This is the window. The listener answers later launches.
+    First(Listener),
+    /// A window was already open and has the request. This launch is done.
+    Handed,
+}
+
+/// Become the one window, or hand `request` to the one there is.
+pub fn claim(runtime: &tokio::runtime::Runtime, dirs: &AppDirs, request: &Request) -> Claim {
+    platform::claim(runtime, dirs, request)
+}
+
+/// Requests from later launches, until the process ends.
+pub struct Listener {
+    inner: Option<platform::Inner>,
+}
+
+impl Listener {
+    /// Start answering. `first` is this launch's own request, which arrives
+    /// through the same channel as the rest so there is one place to act on it.
+    pub fn serve(
+        self,
+        runtime: &tokio::runtime::Runtime,
+        ctx: &egui::Context,
+        first: Request,
+    ) -> Receiver<Request> {
+        let (sender, receiver) = channel();
+        if first != Request::Raise {
+            let _ = sender.send(first);
+        }
+        if let Some(inner) = self.inner {
+            platform::serve(
+                inner,
+                runtime,
+                Deliver {
+                    sender,
+                    ctx: ctx.clone(),
+                },
+            );
+        }
+        receiver
+    }
+}
+
+/// Where a received request goes: the app's queue, and a repaint so it is
+/// acted on now rather than when the pointer next moves.
+struct Deliver {
+    sender: Sender<Request>,
+    ctx: egui::Context,
+}
+
+impl Deliver {
+    fn line(&self, text: &str) {
+        let Some(request) = Request::decode(text) else {
+            tracing::debug!("instance: ignored a request that did not parse");
+            return;
+        };
+        let _ = self.sender.send(request);
+        self.ctx.request_repaint();
+    }
+}
+
+#[cfg(unix)]
+mod platform {
+    use std::io::{Read, Write};
+    use std::os::unix::fs::PermissionsExt;
+    use std::os::unix::net::{UnixListener, UnixStream};
+    use std::path::PathBuf;
+
+    use super::{AppDirs, Claim, Deliver, HAND_TIMEOUT, Listener, MOST, Request};
+
+    pub struct Inner(UnixListener);
+
+    /// In the runtime directory where there is one: it is the user's alone and
+    /// emptied at logout, which is exactly the life a socket should have.
+    fn socket(dirs: &AppDirs) -> PathBuf {
+        directories::BaseDirs::new()
+            .and_then(|base| base.runtime_dir().map(std::path::Path::to_path_buf))
+            .unwrap_or_else(|| dirs.cache.clone())
+            .join("io.narl.proton-stream.sock")
+    }
+
+    pub fn claim(_: &tokio::runtime::Runtime, dirs: &AppDirs, request: &Request) -> Claim {
+        let path = socket(dirs);
+        if let Ok(mut stream) = UnixStream::connect(&path) {
+            let _ = stream.set_write_timeout(Some(HAND_TIMEOUT));
+            match stream.write_all(request.encode().as_bytes()) {
+                Ok(()) => return Claim::Handed,
+                Err(error) => tracing::warn!("instance: hand over to the open window: {error}"),
+            }
+        }
+        // Nobody answered, so a socket file still there is left over from a
+        // window that did not exit cleanly.
+        let _ = std::fs::remove_file(&path);
+        let listener = match UnixListener::bind(&path) {
+            Ok(listener) => listener,
+            Err(error) => {
+                tracing::warn!("instance: listen at {}: {error}", path.display());
+                return Claim::First(Listener { inner: None });
+            }
+        };
+        // The runtime directory is private already; the cache fallback is
+        // not necessarily, and a share link is a secret.
+        let _ = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600));
+        Claim::First(Listener {
+            inner: Some(Inner(listener)),
+        })
+    }
+
+    pub fn serve(inner: Inner, _: &tokio::runtime::Runtime, deliver: Deliver) {
+        let spawned = std::thread::Builder::new()
+            .name("instance".into())
+            .spawn(move || {
+                for stream in inner.0.incoming() {
+                    let Ok(stream) = stream else { continue };
+                    let _ = stream.set_read_timeout(Some(HAND_TIMEOUT));
+                    let mut text = String::new();
+                    if stream.take(MOST).read_to_string(&mut text).is_ok() {
+                        deliver.line(&text);
+                    }
+                }
+            });
+        if let Err(error) = spawned {
+            tracing::warn!("instance thread: {error}");
+        }
+    }
+}
+
+#[cfg(windows)]
+mod platform {
+    use std::hash::{BuildHasher, Hasher};
+    use std::io::Write;
+
+    use tokio::io::AsyncReadExt;
+    use tokio::net::windows::named_pipe::{NamedPipeServer, ServerOptions};
+
+    use super::{AppDirs, Claim, Deliver, HAND_TIMEOUT, Listener, MOST, Request};
+
+    pub struct Inner {
+        name: String,
+        first: NamedPipeServer,
+    }
+
+    /// The pipe's name. Pipe names are machine-wide, so a fixed one could be
+    /// taken first by another account on the same machine, which would then
+    /// be handed this user's share links. The name carries a random part kept
+    /// in this user's own config directory, which nobody else can read.
+    fn pipe(dirs: &AppDirs) -> String {
+        let path = dirs.config.join("instance.key");
+        let key = std::fs::read_to_string(&path)
+            .ok()
+            .map(|key| key.trim().to_owned())
+            .filter(|key| key.len() == 32 && key.bytes().all(|b| b.is_ascii_hexdigit()))
+            .unwrap_or_else(|| {
+                // `RandomState` is seeded from the OS's random source, once
+                // per process and stepped per instance after that.
+                let mut key = String::new();
+                for _ in 0..2 {
+                    let word = std::collections::hash_map::RandomState::new()
+                        .build_hasher()
+                        .finish();
+                    key.push_str(&format!("{word:016x}"));
+                }
+                if let Err(error) = std::fs::write(&path, &key) {
+                    tracing::warn!("instance: store the pipe key: {error}");
+                }
+                key
+            });
+        format!(r"\\.\pipe\io.narl.proton-stream.{key}")
+    }
+
+    fn server(name: &str, first: bool) -> std::io::Result<NamedPipeServer> {
+        ServerOptions::new()
+            .first_pipe_instance(first)
+            .reject_remote_clients(true)
+            .access_outbound(false)
+            .create(name)
+    }
+
+    pub fn claim(runtime: &tokio::runtime::Runtime, dirs: &AppDirs, request: &Request) -> Claim {
+        let name = pipe(dirs);
+        // Registering a pipe with tokio needs its reactor.
+        let _entered = runtime.enter();
+        match server(&name, true) {
+            Ok(first) => Claim::First(Listener {
+                inner: Some(Inner { name, first }),
+            }),
+            // Taken: there is a window. Hand it the request.
+            Err(_) => {
+                let started = std::time::Instant::now();
+                loop {
+                    match std::fs::OpenOptions::new().write(true).open(&name) {
+                        Ok(mut pipe) => {
+                            if let Err(error) = pipe.write_all(request.encode().as_bytes()) {
+                                tracing::warn!("instance: hand over to the open window: {error}");
+                            }
+                            return Claim::Handed;
+                        }
+                        // Every instance busy with another launch: wait a turn.
+                        Err(_) if started.elapsed() < HAND_TIMEOUT => {
+                            std::thread::sleep(std::time::Duration::from_millis(50));
+                        }
+                        Err(error) => {
+                            tracing::warn!("instance: reach the open window: {error}");
+                            return Claim::First(Listener { inner: None });
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    pub fn serve(inner: Inner, runtime: &tokio::runtime::Runtime, deliver: Deliver) {
+        runtime.spawn(async move {
+            let Inner { name, mut first } = inner;
+            loop {
+                if first.connect().await.is_err() {
+                    continue;
+                }
+                let connected = first;
+                // The next instance is up before this one is read, so a launch
+                // right behind this one finds a pipe to connect to.
+                first = match server(&name, false) {
+                    Ok(next) => next,
+                    Err(error) => {
+                        tracing::warn!("instance: reopen the pipe: {error}");
+                        return;
+                    }
+                };
+                let mut text = String::new();
+                let mut limited = connected.take(MOST);
+                let read = limited.read_to_string(&mut text);
+                if let Ok(Ok(_)) = tokio::time::timeout(HAND_TIMEOUT, read).await {
+                    deliver.line(&text);
+                }
+            }
+        });
+    }
+}
+
+#[cfg(not(any(unix, windows)))]
+mod platform {
+    use super::{AppDirs, Claim, Deliver, Listener, Request};
+
+    pub struct Inner;
+
+    pub fn claim(_: &tokio::runtime::Runtime, _: &AppDirs, _: &Request) -> Claim {
+        Claim::First(Listener { inner: None })
+    }
+
+    pub fn serve(_: Inner, _: &tokio::runtime::Runtime, _: Deliver) {}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const LINK: &str = "https://drive.proton.me/urls/ABCDEFGHIJ#fragmentsecret";
+
+    #[test]
+    fn a_launch_with_a_share_link_asks_for_it_to_be_added() {
+        let args = ["--verbose".to_owned(), format!("  {LINK} ")];
+        assert_eq!(Request::from_args(args), Request::AddShare(LINK.to_owned()));
+    }
+
+    #[test]
+    fn a_launch_with_no_share_link_only_raises_the_window() {
+        let args = ["https://example.com/".to_owned()];
+        assert_eq!(Request::from_args(args), Request::Raise);
+        assert_eq!(Request::from_args(Vec::new()), Request::Raise);
+    }
+
+    #[test]
+    fn a_request_survives_the_wire() {
+        for request in [Request::Raise, Request::AddShare(LINK.to_owned())] {
+            assert_eq!(Request::decode(&request.encode()), Some(request));
+        }
+    }
+
+    #[test]
+    fn a_line_that_is_not_a_request_is_ignored() {
+        assert_eq!(Request::decode("add https://example.com/"), None);
+        assert_eq!(Request::decode("open the pod bay doors"), None);
+    }
+}

crates/pstr-app/src/lib.rs

@@ -23,6 +23,7 @@ pub mod app;
 pub mod desktop_prefs;
 pub mod engine;
 pub mod inhibit;
+pub mod instance;
 pub mod media;
 pub mod notify;
 pub mod pacing;

crates/pstr-app/src/main.rs

@@ -31,6 +31,17 @@ fn main() -> anyhow::Result<()> {
     );
     let dirs = AppDirs::ensure().context("resolve app directories")?;
 
+    // Before the window: a second launch hands its request to the first and
+    // never opens one.
+    let request = pstr_app::instance::Request::from_args(std::env::args().skip(1));
+    let instance = match pstr_app::instance::claim(&runtime, &dirs, &request) {
+        pstr_app::instance::Claim::First(listener) => listener,
+        pstr_app::instance::Claim::Handed => {
+            tracing::info!("handed over to the window already open");
+            return Ok(());
+        }
+    };
+
     let window = pstr_app::window::WindowState::load(&dirs.window_file());
     let mut viewport = egui::ViewportBuilder::default();
     // X11 and Windows take the icon from the window. Wayland ignores this and
@@ -57,7 +68,7 @@ fn main() -> anyhow::Result<()> {
         "proton-stream",
         options,
         Box::new(move |cc| {
-            pstr_app::app::App::new(cc, runtime, dirs)
+            pstr_app::app::App::new(cc, runtime, dirs, instance, request)
                 .map(|app| Box::new(app) as Box<dyn eframe::App>)
                 .map_err(|error| error.into())
         }),

docs/ARCHITECTURE.md

@@ -124,3 +124,4 @@ an edge case.
 | Provider artwork | cache dir, `posters/` | Yes — refetched by URL; stored scaled to 640 px |
 | Desktop-only settings | config dir, `desktop.json` | Defaults if missing — hardware decoding, cache budget; never overwritten when unparseable |
 | Window size | config dir, `window.json` | Defaults if missing or implausible |
+| Single-instance pipe key (Windows) | config dir, `instance.key` | Yes — regenerated if missing |