feat(prs-create) send to multiple relays

add tests but these currently don't work when run together

11 files changed, 1089 insertions, 59 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 15bc112..b8c0f04 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -863,6 +863,19 @@ dependencies = [
 ]
 
 [[package]]
+name = "flume"
+version = "0.10.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1657b4441c3403d9f7b3409e47575237dac27b1b5726df654a6ecbf92f0f7577"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "nanorand",
+ "pin-project",
+ "spin 0.9.8",
+]
+
+[[package]]
 name = "fnv"
 version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1219,6 +1232,19 @@ dependencies = [
 ]
 
 [[package]]
+name = "indicatif"
+version = "0.17.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fb28741c9db9a713d93deb3bb9515c20788cef5815265bee4980e87bde7e0f25"
+dependencies = [
+ "console",
+ "instant",
+ "number_prefix",
+ "portable-atomic",
+ "unicode-width",
+]
+
+[[package]]
 name = "inout"
 version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1481,6 +1507,15 @@ dependencies = [
 ]
 
 [[package]]
+name = "nanorand"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a51313c5820b0b02bd422f4b44776fbf47961755c74ce64afc73bfad10226c3"
+dependencies = [
+ "getrandom",
+]
+
+[[package]]
 name = "ngit"
 version = "0.0.1"
 dependencies = [
@@ -1489,11 +1524,13 @@ dependencies = [
  "async-trait",
  "chacha20poly1305",
  "clap",
+ "console",
  "dialoguer",
  "directories",
  "duplicate",
  "futures",
  "git2",
+ "indicatif",
  "keyring",
  "mockall",
  "nostr",
@@ -1591,7 +1628,7 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "tokio-socks",
- "tokio-tungstenite",
+ "tokio-tungstenite 0.20.1",
  "url-fork",
  "webpki-roots",
  "ws_stream_wasm",
@@ -1684,6 +1721,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "number_prefix"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3"
+
+[[package]]
 name = "object"
 version = "0.32.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1814,6 +1857,26 @@ dependencies = [
 ]
 
 [[package]]
+name = "pin-project"
+version = "1.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fda4ed1c6c173e3fc7a83629421152e01d7b1f9b7f65fb301e490e8cfc656422"
+dependencies = [
+ "pin-project-internal",
+]
+
+[[package]]
+name = "pin-project-internal"
+version = "1.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4359fd9c9171ec6e8c62926d6faaf553a8dc3f64e1507e76da7911b4f6a04405"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.38",
+]
+
+[[package]]
 name = "pin-project-lite"
 version = "0.2.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1870,6 +1933,12 @@ dependencies = [
 ]
 
 [[package]]
+name = "portable-atomic"
+version = "1.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "31114a898e107c51bb1609ffaf55a0e011cf6a4d7f1170d0015a165082c0338b"
+
+[[package]]
 name = "ppv-lite86"
 version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2169,7 +2238,7 @@ dependencies = [
  "cc",
  "libc",
  "once_cell",
- "spin",
+ "spin 0.5.2",
  "untrusted",
  "web-sys",
  "winapi",
@@ -2483,6 +2552,18 @@ dependencies = [
 ]
 
 [[package]]
+name = "simple-websockets"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f38cc14717bb624d10e9bb4fff30344e8f540c0d2c0f876f8fb0111d808ee7c"
+dependencies = [
+ "flume",
+ "futures-util",
+ "tokio",
+ "tokio-tungstenite 0.19.0",
+]
+
+[[package]]
 name = "slab"
 version = "0.4.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2524,6 +2605,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d"
 
 [[package]]
+name = "spin"
+version = "0.9.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
+dependencies = [
+ "lock_api",
+]
+
+[[package]]
 name = "static_assertions"
 version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2625,6 +2715,7 @@ dependencies = [
  "once_cell",
  "rand",
  "rexpect 0.5.0 (git+https://github.com/phaer/rexpect.git?branch=skip-ansi-escape-codes)",
+ "simple-websockets",
  "strip-ansi-escapes",
 ]
 
@@ -2674,7 +2765,9 @@ dependencies = [
  "libc",
  "mio",
  "num_cpus",
+ "parking_lot",
  "pin-project-lite",
+ "signal-hook-registry",
  "socket2 0.5.4",
  "tokio-macros",
  "windows-sys 0.48.0",
@@ -2715,6 +2808,18 @@ dependencies = [
 
 [[package]]
 name = "tokio-tungstenite"
+version = "0.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec509ac96e9a0c43427c74f003127d953a265737636129424288d27cb5c4b12c"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite 0.19.0",
+]
+
+[[package]]
+name = "tokio-tungstenite"
 version = "0.20.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "212d5dcb2a1ce06d81107c3d0ffa3121fe974b73f068c8282cb1c32328113b6c"
@@ -2724,7 +2829,7 @@ dependencies = [
  "rustls",
  "tokio",
  "tokio-rustls",
- "tungstenite",
+ "tungstenite 0.20.1",
  "webpki-roots",
 ]
 
@@ -2805,6 +2910,25 @@ checksum = "3528ecfd12c466c6f163363caf2d02a71161dd5e1cc6ae7b34207ea2d42d81ed"
 
 [[package]]
 name = "tungstenite"
+version = "0.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15fba1a6d6bb030745759a9a2a588bfe8490fc8b4751a277db3a0be1c9ebbf67"
+dependencies = [
+ "byteorder",
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "rand",
+ "sha1",
+ "thiserror",
+ "url",
+ "utf-8",
+]
+
+[[package]]
+name = "tungstenite"
 version = "0.20.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9e3dac10fd62eaf6617d3a904ae222845979aec67c615d1c842b4002c7666fb9"
diff --git a/Cargo.toml b/Cargo.toml
index f7577a4..1571c02 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -16,10 +16,12 @@ anyhow = "1.0.75"
 async-trait = "0.1.73"
 chacha20poly1305 = "0.10.1"
 clap = { version = "4.3.19", features = ["derive"] }
+console = "0.15.7"
 dialoguer = "0.10.4"
 directories = "5.0.1"
 futures = "0.3.28"
 git2 = "0.18.1"
+indicatif = "0.17.7"
 keyring = "2.0.5"
 nostr = "0.24.0"
 nostr-sdk = "0.24.0"
diff --git a/src/client.rs b/src/client.rs
index a6f7dda..e0e0494 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -18,6 +18,7 @@ use nostr::Event;
 
 pub struct Client {
     client: nostr_sdk::Client,
+    pub fallback_relays: Vec<String>,
 }
 
 #[async_trait]
@@ -26,6 +27,7 @@ pub trait Connect {
     fn default() -> Self;
     fn new(opts: Params) -> Self;
     async fn connect(&self) -> Result<()>;
+    async fn disconnect(&self) -> Result<()>;
     async fn send_event_to(&self, url: &str, event: nostr::event::Event) -> Result<nostr::EventId>;
 }
 
@@ -34,19 +36,31 @@ impl Connect for Client {
     fn default() -> Self {
         Client {
             client: nostr_sdk::Client::new(&nostr::Keys::generate()),
+            fallback_relays: vec![
+                "ws://localhost:8080".to_string(),
+                "ws://localhost:8052".to_string(),
+            ],
         }
     }
     fn new(opts: Params) -> Self {
         Client {
             client: nostr_sdk::Client::new(&opts.keys.unwrap_or(nostr::Keys::generate())),
+            fallback_relays: opts.fallback_relays,
         }
     }
     async fn connect(&self) -> Result<()> {
-        self.client.add_relay("ws://localhost:8080", None).await?;
+        for relay in &self.fallback_relays {
+            self.client.add_relay(relay.as_str(), None).await?;
+        }
         self.client.connect().await;
-        // self.client.s
         Ok(())
     }
+
+    async fn disconnect(&self) -> Result<()> {
+        self.client.disconnect().await?;
+        Ok(())
+    }
+
     async fn send_event_to(&self, url: &str, event: Event) -> Result<nostr::EventId> {
         Ok(self.client.send_event_to(url, event).await?)
     }
@@ -55,6 +69,7 @@ impl Connect for Client {
 #[derive(Default)]
 pub struct Params {
     pub keys: Option<nostr::Keys>,
+    pub fallback_relays: Vec<String>,
 }
 
 impl Params {
@@ -62,4 +77,8 @@ impl Params {
         self.keys = Some(keys);
         self
     }
+    pub fn with_fallback_relays(mut self, fallback_relays: Vec<String>) -> Self {
+        self.fallback_relays = fallback_relays;
+        self
+    }
 }
diff --git a/src/git.rs b/src/git.rs
index ddbc646..337444a 100644
--- a/src/git.rs
+++ b/src/git.rs
@@ -245,6 +245,23 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn get_commit_parent() -> Result<()> {
+        let test_repo = GitTestRepo::default();
+        let parent_oid = test_repo.populate()?;
+        std::fs::write(test_repo.dir.join("t100.md"), "some content")?;
+        let child_oid = test_repo.stage_and_commit("add t100.md")?;
+
+        let git_repo = Repo::from_path(&test_repo.dir)?;
+
+        assert_eq!(
+            // Sha1Hash::from_byte_array("bla".to_string().as_bytes()),
+            oid_to_sha1(&parent_oid),
+            git_repo.get_commit_parent(&oid_to_sha1(&child_oid))?,
+        );
+        Ok(())
+    }
+
     mod make_patch_from_commit {
         use super::*;
         #[test]
diff --git a/src/login.rs b/src/login.rs
index a6ce76d..12fe76e 100644
--- a/src/login.rs
+++ b/src/login.rs
@@ -81,5 +81,8 @@ pub fn launch(nsec: &Option<String>, password: &Option<String>) -> Result<nostr:
             .to_bech32()
             .context("public key should always produce bech32")?
     );
+
+    // fetching metdata
+
     Ok(key)
 }
diff --git a/src/main.rs b/src/main.rs
index 9c37aa7..5c8518b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -24,6 +24,9 @@ pub struct Cli {
     /// password to decrypt nsec
     #[arg(short, long, global = true)]
     password: Option<String>,
+    /// disable spinner animations
+    #[arg(long, action)]
+    disable_cli_spinners: bool,
 }
 
 #[derive(Subcommand)]
diff --git a/src/sub_commands/prs/create.rs b/src/sub_commands/prs/create.rs
index 89ea652..3047e92 100644
--- a/src/sub_commands/prs/create.rs
+++ b/src/sub_commands/prs/create.rs
@@ -1,4 +1,9 @@
+use std::time::Duration;
+
 use anyhow::{bail, Context, Result};
+use console::Term;
+use futures::future::join_all;
+use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
 use nostr::{prelude::sha1::Hash as Sha1Hash, EventBuilder, Marker, Tag, TagKind};
 
 use crate::{
@@ -79,14 +84,245 @@ pub async fn launch(
             .input(PromptInputParms::default().with_prompt("description (Optional)"))?,
     };
 
-    let root_commit = git_repo
-        .get_root_commit(to_branch.as_str())
-        .context("failed to get root commit of the repository")?;
-
     // create PR event
 
     let keys = login::launch(&cli_args.nsec, &cli_args.password)?;
 
+    let events =
+        generate_pr_and_patch_events(&title, &description, &to_branch, &git_repo, &ahead, keys)?;
+
+    let my_write_relays: Vec<String> = vec![
+        "ws://localhost:8051".to_string(),
+        "ws://localhost:8052".to_string(),
+    ];
+
+    let repo_read_relays: Vec<String> = vec![
+        "ws://localhost:8051".to_string(),
+        "ws://localhost:8053".to_string(),
+    ];
+
+    send_events(
+        events,
+        keys,
+        my_write_relays,
+        repo_read_relays,
+        !cli_args.disable_cli_spinners,
+    )
+    .await?;
+    // TODO check if there is already a similarly named PR
+
+    // println!("failures:");
+    // println!("ws://relay.anon.io    Error: Payment Required");
+
+    // should we have a relays in Repository event?
+    // yes
+    //
+
+    // TODO connect to relays and post
+
+    Ok(())
+}
+
+async fn send_events(
+    events: Vec<nostr::Event>,
+    keys: nostr::Keys,
+    my_write_relays: Vec<String>,
+    repo_read_relays: Vec<String>,
+    animate: bool,
+) -> Result<()> {
+    let (_, _, _, all) = unique_and_duplicate_all(&my_write_relays, &repo_read_relays);
+
+    let client = Client::new(
+        ClientParams::default()
+            .with_keys(keys)
+            // .with_fallback_relays(vec!["ws://localhost:8080".to_string()]),
+            .with_fallback_relays(all.iter().map(std::string::ToString::to_string).collect()),
+    );
+
+    let term = Term::stdout();
+    term.write_line("connecting to relays...")?;
+    client.connect().await?;
+    term.clear_last_lines(1)?;
+
+    println!(
+        "posting 1 pull request with {} commits...",
+        events.len() - 1
+    );
+
+    let m = MultiProgress::new();
+    let pb_style = ProgressStyle::with_template(if animate {
+        " {spinner} {prefix} {bar} {pos}/{len} {msg}"
+    } else {
+        " - {prefix} {bar} {pos}/{len} {msg}"
+    })?
+    .progress_chars("##-");
+
+    let pb_after_style =
+        |symbol| ProgressStyle::with_template(format!(" {symbol} {}", "{prefix} {msg}",).as_str());
+    let pb_after_style_succeeded = pb_after_style(if animate {
+        console::style("✔".to_string())
+            .for_stderr()
+            .green()
+            .to_string()
+    } else {
+        "y".to_string()
+    })?;
+
+    let pb_after_style_failed = pb_after_style(if animate {
+        console::style("✘".to_string())
+            .for_stderr()
+            .red()
+            .to_string()
+    } else {
+        "x".to_string()
+    })?;
+
+    join_all(all.iter().map(|&relay| async {
+        let details = format!(
+            "{}{} {}",
+            if my_write_relays.iter().any(|r| relay.eq(r)) {
+                " [my-relay]"
+            } else {
+                ""
+            },
+            if repo_read_relays.iter().any(|r| relay.eq(r)) {
+                " [repo-relay]"
+            } else {
+                ""
+            },
+            *relay,
+        );
+        let pb = m.add(
+            ProgressBar::new(events.len() as u64)
+                .with_prefix(details.to_string())
+                .with_style(pb_style.clone()),
+        );
+        if animate {
+            pb.enable_steady_tick(Duration::from_millis(300));
+        }
+        pb.inc(0); // need to make pb display intially
+        let mut failed = false;
+        for event in &events {
+            match client.send_event_to(relay.as_str(), event.clone()).await {
+                Ok(_) => pb.inc(1),
+                Err(e) => {
+                    pb.set_style(pb_after_style_failed.clone());
+                    pb.finish_with_message(
+                        console::style(
+                            e.to_string()
+                                .replace("relay pool error:", "error:")
+                                .replace("event not published: ", ""),
+                        )
+                        .for_stderr()
+                        .red()
+                        .to_string(),
+                    );
+                    failed = true;
+                    break;
+                }
+            };
+        }
+        if !failed {
+            pb.set_style(pb_after_style_succeeded.clone());
+            pb.finish_with_message("");
+        }
+    }))
+    .await;
+    client.disconnect().await?;
+    Ok(())
+}
+
+/// returns `(unique_vec1, unique_vec2, duplicates, all)`
+fn unique_and_duplicate_all<'a, S>(
+    vec1: &'a Vec<S>,
+    vec2: &'a Vec<S>,
+) -> (Vec<&'a S>, Vec<&'a S>, Vec<&'a S>, Vec<&'a S>)
+where
+    S: PartialEq,
+{
+    let mut vec1_u = vec![];
+    let mut vec2_u = vec![];
+    let mut dup = vec![];
+    let mut all = vec![];
+    for s1 in vec1 {
+        if vec2.iter().any(|s2| s1.eq(s2)) {
+            dup.push(s1);
+        } else {
+            vec1_u.push(s1);
+        }
+    }
+    for s2 in vec2 {
+        if !vec1.iter().any(|s1| s2.eq(s1)) {
+            vec2_u.push(s2);
+        }
+    }
+    for a in [&dup, &vec1_u, &vec2_u] {
+        for e in a {
+            all.push(&**e);
+        }
+    }
+    (vec1_u, vec2_u, dup, all)
+}
+
+mod tests_unique_and_duplicate {
+
+    #[test]
+    fn correct_number_of_unique_and_duplicate_items() {
+        let v1 = vec![
+            "t1".to_string(),
+            "t2".to_string(),
+            "t3".to_string(),
+            "t4".to_string(),
+            "t5".to_string(),
+        ];
+        let v2 = vec![
+            "t3".to_string(),
+            "t4".to_string(),
+            "t5".to_string(),
+            "t6".to_string(),
+        ];
+
+        let (v1_u, v2_u, d, a) = super::unique_and_duplicate_all(&v1, &v2);
+
+        assert_eq!(v1_u.len(), 2);
+        assert_eq!(v2_u.len(), 1);
+        assert_eq!(d.len(), 3);
+        assert_eq!(a.len(), 6);
+    }
+    #[test]
+    fn all_begins_with_duplicates() {
+        let v1 = vec![
+            "t1".to_string(),
+            "t2".to_string(),
+            "t3".to_string(),
+            "t4".to_string(),
+            "t5".to_string(),
+        ];
+        let v2 = vec![
+            "t3".to_string(),
+            "t4".to_string(),
+            "t5".to_string(),
+            "t6".to_string(),
+        ];
+
+        let (_, _, d, a) = super::unique_and_duplicate_all(&v1, &v2);
+
+        assert_eq!(a[0], d[0]);
+    }
+}
+
+fn generate_pr_and_patch_events(
+    title: &String,
+    description: &String,
+    to_branch: &str,
+    git_repo: &Repo,
+    commits: &Vec<Sha1Hash>,
+    keys: nostr::Keys,
+) -> Result<Vec<nostr::Event>> {
+    let root_commit = git_repo
+        .get_root_commit(to_branch)
+        .context("failed to get root commit of the repository")?;
+
     let pr_event = EventBuilder::new(
         nostr::event::Kind::Custom(318),
         format!("{title}\r\n\r\n{description}"),
@@ -103,12 +339,14 @@ pub async fn launch(
     .to_event(&keys)
     .context("failed to create pr event")?;
 
-    let mut patch_events = vec![];
-    for commit in &ahead {
+    let pr_event_id = pr_event.id;
+
+    let mut events = vec![pr_event];
+    for commit in commits {
         let commit_parent = git_repo
             .get_commit_parent(commit)
             .context("failed to create patch event")?;
-        patch_events.push(
+        events.push(
             EventBuilder::new(
                 nostr::event::Kind::Custom(317),
                 git_repo
@@ -119,7 +357,7 @@ pub async fn launch(
                     Tag::Hashtag(commit.to_string()),
                     Tag::Hashtag(commit_parent.to_string()),
                     Tag::Event(
-                        pr_event.id,
+                        pr_event_id,
                         None, // TODO: add relay
                         Some(Marker::Root),
                     ),
@@ -136,32 +374,11 @@ pub async fn launch(
                     // TODO: add relay tags
                 ],
             )
-            .to_event(&keys),
+            .to_event(&keys)?,
         );
     }
-
-    let client = Client::new(ClientParams::default().with_keys(keys));
-
-    println!("connecting...");
-    client.connect().await?;
-    println!("connected...");
-
-    // TODO check if there is already a similarly named PR
-    let _ = client
-        .send_event_to("ws://localhost:8080", pr_event)
-        .await?;
-    // TODO post each PR
-    // TODO report
-    println!("posted successfully to 4/5 of your relays and 0/4 of maintainers relays");
-    // should we have a relays in Repository event?
-    // yes
-    //
-
-    // TODO connect to relays and post
-
-    Ok(())
+    Ok(events)
 }
-
 // TODO
 // - find profile
 // - file relays
diff --git a/test_utils/Cargo.toml b/test_utils/Cargo.toml
index 1773d93..2d3555b 100644
--- a/test_utils/Cargo.toml
+++ b/test_utils/Cargo.toml
@@ -13,4 +13,5 @@ nostr = "0.24.0"
 once_cell = "1.18.0"
 rand = "0.8"
 rexpect = { git = "https://github.com/phaer/rexpect.git", branch= "skip-ansi-escape-codes" }
+simple-websockets = "0.1.6"
 strip-ansi-escapes = "0.2.0"
diff --git a/test_utils/src/lib.rs b/test_utils/src/lib.rs
index 0f870f6..a9d818c 100644
--- a/test_utils/src/lib.rs
+++ b/test_utils/src/lib.rs
@@ -9,6 +9,7 @@ use rexpect::session::{Options, PtySession};
 use strip_ansi_escapes::strip_str;
 
 pub mod git;
+pub mod relay;
 
 pub static TEST_KEY_1_NSEC: &str =
     "nsec1ppsg5sm2aexq06juxmu9evtutr6jkwkhp98exxxvwamhru9lyx9s3rwseq";
@@ -353,7 +354,12 @@ impl CliTester {
     }
 
     /// returns what came before expected message
-    pub fn expect_eventually(&mut self, message: &str) -> Result<String> {
+    pub fn expect_eventually<S>(&mut self, message: S) -> Result<String>
+    where
+        S: Into<String>,
+    {
+        let message_string = message.into();
+        let message = message_string.as_str();
         let before = self
             .rexpect_session
             .exp_string(message)
@@ -361,8 +367,27 @@ impl CliTester {
         Ok(before)
     }
 
-    pub fn expect(&mut self, message: &str) -> Result<&mut Self> {
+    pub fn expect_after_whitespace<S>(&mut self, message: S) -> Result<&mut Self>
+    where
+        S: Into<String>,
+    {
+        assert_eq!("", self.expect_eventually(message)?.trim());
+        Ok(self)
+    }
+
+    pub fn expect<S>(&mut self, message: S) -> Result<&mut Self>
+    where
+        S: Into<String>,
+    {
+        let message_string = message.into();
+        let message = message_string.as_str();
         let before = self.expect_eventually(message)?;
+        if !before.is_empty() {
+            std::fs::write("aaaaaaaaaaaa.txt", before.clone())?;
+
+            // let mut output = std::fs::File::create("aaaaaaaaaaa.txt")?;
+            // write!(output, "{}", *before);
+        }
         ensure!(
             before.is_empty(),
             format!(
@@ -397,6 +422,16 @@ impl CliTester {
         assert_eq!(before, message);
         Ok(())
     }
+
+    pub fn expect_end_with_whitespace(&mut self) -> Result<()> {
+        let before = self
+            .rexpect_session
+            .exp_eof()
+            .context("expected immediate end but got timed out")?;
+        assert_eq!(before.trim(), "");
+        Ok(())
+    }
+
     pub fn expect_end_eventually(&mut self) -> Result<String> {
         self.rexpect_session
             .exp_eof()
diff --git a/test_utils/src/relay.rs b/test_utils/src/relay.rs
new file mode 100644
index 0000000..6de3618
--- /dev/null
+++ b/test_utils/src/relay.rs
@@ -0,0 +1,196 @@
+use std::collections::HashMap;
+
+use anyhow::{bail, Result};
+use nostr::{ClientMessage, RelayMessage};
+
+use crate::CliTester;
+
+type ListenerFunc<'a> = &'a dyn Fn(&mut Relay, u64, nostr::Event) -> Result<()>;
+
+pub struct Relay<'a> {
+    port: u16,
+    event_hub: simple_websockets::EventHub,
+    clients: HashMap<u64, simple_websockets::Responder>,
+    pub events: Vec<nostr::Event>,
+    event_listener: Option<ListenerFunc<'a>>,
+}
+
+impl<'a> Relay<'a> {
+    pub fn new(port: u16, event_listener: Option<ListenerFunc<'a>>) -> Self {
+        let event_hub = simple_websockets::launch(port)
+            .unwrap_or_else(|_| panic!("failed to listen on port {port}"));
+        Self {
+            port,
+            events: vec![],
+            event_hub,
+            clients: HashMap::new(),
+            event_listener,
+        }
+    }
+    pub fn respond_ok(
+        &self,
+        client_id: u64,
+        event: nostr::Event,
+        error: Option<&str>,
+    ) -> Result<bool> {
+        let responder = self.clients.get(&client_id).unwrap();
+
+        let ok_json = RelayMessage::Ok {
+            event_id: event.id,
+            status: error.is_none(),
+            message: error.unwrap_or("").to_string(),
+        }
+        .as_json();
+        // bail!(format!("{}", &ok_json));
+        Ok(responder.send(simple_websockets::Message::Text(ok_json)))
+    }
+    /// listen, collect events and responds with event_listener to events or
+    /// Ok(eventid) if event_listner is None
+    pub async fn listen_until_close(&mut self) -> Result<()> {
+        loop {
+            println!("polling");
+            match self.event_hub.poll_async().await {
+                simple_websockets::Event::Connect(client_id, responder) => {
+                    // add their Responder to our `clients` map:
+                    self.clients.insert(client_id, responder);
+                }
+                simple_websockets::Event::Disconnect(client_id) => {
+                    // remove the disconnected client from the clients map:
+                    println!("{} disconnected", self.port);
+                    self.clients.remove(&client_id);
+                    break;
+                }
+                simple_websockets::Event::Message(client_id, message) => {
+                    println!(
+                        "Received a message from client #{}: {:?}",
+                        client_id, message
+                    );
+
+                    if let Ok(event) = get_nevent(message) {
+                        self.events.push(event.clone());
+                        if let Some(listner) = self.event_listener {
+                            listner(self, client_id, event)?;
+                        } else {
+                            self.respond_ok(client_id, event, None)?;
+                        }
+                    }
+                }
+            }
+        }
+        println!("stop polling");
+        println!("we may not be polling but the tcplistner is still listening");
+        Ok(())
+    }
+}
+
+fn get_nevent(message: simple_websockets::Message) -> Result<nostr::Event> {
+    if let simple_websockets::Message::Text(s) = message.clone() {
+        let cm_result = ClientMessage::from_json(s);
+        if let Ok(ClientMessage::Event(event)) = cm_result {
+            let e = *event;
+            return Ok(e.clone());
+        }
+    }
+    bail!("not nostr event")
+}
+
+pub enum Message {
+    Event,
+    // Request,
+}
+
+/// leaves trailing whitespace and only compatible with --no-cli-spinners flag
+/// relays tuple: (title,successful,message)
+pub fn expect_send_with_progress(
+    p: &mut CliTester,
+    relays: Vec<(&str, bool, &str)>,
+    event_count: u16,
+) -> Result<()> {
+    p.expect(format!(
+        " - {} -------------------- 0/{event_count}",
+        &relays[0].0
+    ))?;
+    for relay in &relays {
+        // if successful
+        if relay.1 {
+            p.expect_eventually(format!(" y {}", relay.0))?;
+        } else {
+            p.expect_eventually(format!(" x {} {}", relay.0, relay.2))?;
+        }
+        // could check that before only contains titles:
+        // - # y x n/n and whitespace
+        // let before = p.expect_eventually(format!(" â {title}"))?;
+        // assert_eq!("", before.trim());
+    }
+    Ok(())
+}
+
+pub fn expect_send_with_progress_exact_interaction(
+    p: &mut CliTester,
+    titles: Vec<&str>,
+    count: u16,
+) -> Result<()> {
+    let whitespace_mid = " \r\n";
+    let whitespace_end = "                   \r\r\r";
+
+    p.expect(format!(
+        " - {} -------------------- 0/{count}        \r",
+        titles[0]
+    ))?;
+    p.expect(format!(
+        " - {} -------------------- 0/{count}{whitespace_mid}",
+        titles[0]
+    ))?;
+    p.expect(format!(
+        " - {} -------------------- 0/{count}                     \r\r",
+        titles[1]
+    ))?;
+
+    let generate_text = |title: &str, num: u16, confirmed_complete: bool| -> String {
+        let symbol = if confirmed_complete && num.eq(&count) {
+            "â"
+        } else {
+            "-"
+        };
+        let bar = match (num, count) {
+            (0, _) => "--------------------",
+            (1, 2) => "###########---------",
+            (x, y) => {
+                if x.eq(&y) {
+                    "####################"
+                } else {
+                    "--unknown--"
+                }
+            }
+        };
+        format!(
+            " {symbol} {title} {bar} {num}/{count}{}",
+            if (&title).eq(titles.last().unwrap()) {
+                whitespace_end
+            } else {
+                whitespace_mid
+            }
+        )
+    };
+    let mut nums: HashMap<&str, u16> = HashMap::new();
+    for title in &titles {
+        nums.insert(title, 0);
+        p.expect(generate_text(title, 0, false))?;
+    }
+    loop {
+        for selected_title in &titles {
+            for title in &titles {
+                if title.eq(selected_title) {
+                    let new_num = nums.get(title).unwrap() + 1;
+                    if new_num.gt(&count) {
+                        return Ok(());
+                    }
+                    nums.insert(title, new_num);
+                    p.expect(generate_text(title, *nums.get(title).unwrap(), false))?;
+                } else {
+                    p.expect(generate_text(title, *nums.get(title).unwrap(), true))?;
+                }
+            }
+        }
+    }
+}
diff --git a/tests/prs_create.rs b/tests/prs_create.rs
index d598e34..0863496 100644
--- a/tests/prs_create.rs
+++ b/tests/prs_create.rs
@@ -121,12 +121,36 @@ mod when_commits_behind_ask_to_proceed {
     }
 }
 
-mod when_no_commits_behind {
+#[test]
+#[serial]
+fn cli_message_creating_patches() -> Result<()> {
+    let test_repo = GitTestRepo::default();
+    test_repo.populate()?;
+    // create feature branch with 2 commit ahead
+    test_repo.create_branch("feature")?;
+    test_repo.checkout("feature")?;
+    std::fs::write(test_repo.dir.join("t3.md"), "some content")?;
+    test_repo.stage_and_commit("add t3.md")?;
+    std::fs::write(test_repo.dir.join("t4.md"), "some content")?;
+    test_repo.stage_and_commit("add t4.md")?;
+
+    let mut p = CliTester::new_from_dir(&test_repo.dir, ["prs", "create"]);
+
+    p.expect("creating patch for 2 commits from 'head' that can be merged into 'main'")?;
+    p.exit()?;
+    Ok(())
+}
+
+mod sends_pr_and_2_patches_to_3_relays {
+    use futures::join;
+    use test_utils::relay::Relay;
+
     use super::*;
 
-    #[test]
-    #[serial]
-    fn message_for_creating_patches() -> Result<()> {
+    static PR_KIND: u64 = 318;
+    static PATCH_KIND: u64 = 317;
+
+    fn prep_git_repo() -> Result<GitTestRepo> {
         let test_repo = GitTestRepo::default();
         test_repo.populate()?;
         // create feature branch with 2 commit ahead
@@ -136,28 +160,417 @@ mod when_no_commits_behind {
         test_repo.stage_and_commit("add t3.md")?;
         std::fs::write(test_repo.dir.join("t4.md"), "some content")?;
         test_repo.stage_and_commit("add t4.md")?;
+        Ok(test_repo)
+    }
 
-        let mut p = CliTester::new_from_dir(&test_repo.dir, ["prs", "create"]);
+    fn cli_tester_create_pr(git_repo: &GitTestRepo) -> CliTester {
+        CliTester::new_from_dir(
+            &git_repo.dir,
+            [
+                "--nsec",
+                TEST_KEY_1_NSEC,
+                "--disable-cli-spinners",
+                "prs",
+                "create",
+                "--title",
+                "example",
+                "--description",
+                "example",
+            ],
+        )
+    }
 
-        p.expect("creating patch for 2 commits from 'head' that can be merged into 'main'")?;
-        p.exit()?;
+    fn expect_msgs_first(p: &mut CliTester) -> Result<()> {
+        p.expect("creating patch for 2 commits from 'head' that can be merged into 'main'\r\n")?;
+        p.expect(
+            "logged in as npub175lyhnt6nn00qjw0v3navw9pxgv43txnku0tpxprl4h6mvpr6a5qlphudg\r\n",
+        )?;
+        p.expect("connecting to relays...\r\n")?;
+        p.expect("\r")?;
+        p.expect("posting 1 pull request with 2 commits...\r\n")?;
+        Ok(())
+    }
+
+    async fn prep_run_create_pr() -> Result<(Relay<'static>, Relay<'static>, Relay<'static>)> {
+        let git_repo = prep_git_repo()?;
+
+        let (mut r51, mut r52, mut r53) = (
+            Relay::new(8051, None),
+            Relay::new(8052, None),
+            Relay::new(8053, None),
+        );
+
+        // // check relay had the right number of events
+        let cli_tester_handle = std::thread::spawn(move || -> Result<()> {
+            let mut p = cli_tester_create_pr(&git_repo);
+            p.expect_end_eventually()?;
+            Ok(())
+        });
+
+        // launch relay
+        let _ = join!(
+            r51.listen_until_close(),
+            r52.listen_until_close(),
+            r53.listen_until_close(),
+        );
+        cli_tester_handle.join().unwrap()?;
+        Ok((r51, r52, r53))
+    }
+
+    #[test]
+    #[serial]
+    fn only_1_pr_kind_event_sent_to_each_relay() -> Result<()> {
+        let (r51, r52, r53) = futures::executor::block_on(prep_run_create_pr())?;
+        for relay in [&r51, &r52, &r53] {
+            assert_eq!(
+                relay
+                    .events
+                    .iter()
+                    .filter(|e| e.kind.as_u64().eq(&PR_KIND))
+                    .count(),
+                1,
+            );
+        }
+        Ok(())
+    }
+
+    #[test]
+    #[serial]
+    fn only_2_patch_kind_events_sent_to_each_relay() -> Result<()> {
+        let (r51, r52, r53) = futures::executor::block_on(prep_run_create_pr())?;
+        for relay in [&r51, &r52, &r53] {
+            assert_eq!(
+                relay
+                    .events
+                    .iter()
+                    .filter(|e| e.kind.as_u64().eq(&PATCH_KIND))
+                    .count(),
+                2,
+            );
+        }
         Ok(())
     }
-}
 
-// #[test]
-// #[serial]
-// fn succeeds_with_text_logged_in_as_npub() -> Result<()> {
-//     with_fresh_config(|| {
-//         let mut p = CliTester::new(["login"]);
+    #[test]
+    #[serial]
+    fn patch_content_contains_patch_in_email_format() -> Result<()> {
+        let (r51, r52, r53) = futures::executor::block_on(prep_run_create_pr())?;
+        for relay in [&r51, &r52, &r53] {
+            let patch_events: Vec<&nostr::Event> = relay
+                .events
+                .iter()
+                .filter(|e| e.kind.as_u64().eq(&PATCH_KIND))
+                .collect();
+
+            assert_eq!(
+                patch_events[0].content,
+                "\
+                    From fe973a840fba2a8ab37dd505c154854a69a6505c Mon Sep 17 00:00:00 2001\n\
+                    From: Joe Bloggs <joe.bloggs@pm.me>\n\
+                    Date: Thu, 1 Jan 1970 00:00:00 +0000\n\
+                    Subject: [PATCH] add t4.md\n\
+                    \n\
+                    ---\n \
+                    t4.md | 1 +\n \
+                    1 file changed, 1 insertion(+)\n \
+                    create mode 100644 t4.md\n\
+                    \n\
+                    diff --git a/t4.md b/t4.md\n\
+                    new file mode 100644\n\
+                    index 0000000..f0eec86\n\
+                    --- /dev/null\n\
+                    +++ b/t4.md\n\
+                    @@ -0,0 +1 @@\n\
+                    +some content\n\\ \
+                    No newline at end of file\n\
+                    --\n\
+                    libgit2 1.7.1\n\
+                    \n\
+                    ",
+            );
+            assert_eq!(
+                patch_events[1].content,
+                "\
+                    From 232efb37ebc67692c9e9ff58b83c0d3d63971a0a Mon Sep 17 00:00:00 2001\n\
+                    From: Joe Bloggs <joe.bloggs@pm.me>\n\
+                    Date: Thu, 1 Jan 1970 00:00:00 +0000\n\
+                    Subject: [PATCH] add t3.md\n\
+                    \n\
+                    ---\n \
+                    t3.md | 1 +\n \
+                    1 file changed, 1 insertion(+)\n \
+                    create mode 100644 t3.md\n\
+                    \n\
+                    diff --git a/t3.md b/t3.md\n\
+                    new file mode 100644\n\
+                    index 0000000..f0eec86\n\
+                    --- /dev/null\n\
+                    +++ b/t3.md\n\
+                    @@ -0,0 +1 @@\n\
+                    +some content\n\\ \
+                    No newline at end of file\n\
+                    --\n\
+                    libgit2 1.7.1\n\
+                    \n\
+                    ",
+            );
+        }
+        Ok(())
+    }
+
+    mod pr_tags {
+        use super::*;
+        #[test]
+        #[serial]
+        fn pr_tags_repo_commit() -> Result<()> {
+            let (r51, r52, r53) = futures::executor::block_on(prep_run_create_pr())?;
+            for relay in [&r51, &r52, &r53] {
+                let pr_event: &nostr::Event = relay
+                    .events
+                    .iter()
+                    .find(|e| e.kind.as_u64().eq(&PR_KIND))
+                    .unwrap();
+
+                // root commit 't' tag
+                assert!(pr_event.tags.iter().any(|t| t.as_vec()[0].eq("t")
+                    && t.as_vec()[1].eq("r-9ee507fc4357d7ee16a5d8901bedcd103f23c17d")));
+            }
+            Ok(())
+        }
+    }
+
+    mod patch_tags {
+        use super::*;
+        #[test]
+        #[serial]
+        fn patch_tags_correctly_formatted() -> Result<()> {
+            let (r51, r52, r53) = futures::executor::block_on(prep_run_create_pr())?;
+            for relay in [&r51, &r52, &r53] {
+                let patch_events: Vec<&nostr::Event> = relay
+                    .events
+                    .iter()
+                    .filter(|e| e.kind.as_u64().eq(&PATCH_KIND))
+                    .collect();
+
+                static COMMIT_ID: &str = "fe973a840fba2a8ab37dd505c154854a69a6505c";
+                let most_recent_patch = patch_events[0];
+
+                // commit 't' and 'commit' tag
+                assert!(
+                    most_recent_patch
+                        .tags
+                        .iter()
+                        .any(|t| t.as_vec()[0].eq("t") && t.as_vec()[1].eq(COMMIT_ID))
+                );
+                assert!(
+                    most_recent_patch
+                        .tags
+                        .iter()
+                        .any(|t| t.as_vec()[0].eq("commit") && t.as_vec()[1].eq(COMMIT_ID))
+                );
+
+                // commit parent 't' and 'parent-commit' tag
+                static COMMIT_PARENT_ID: &str = "232efb37ebc67692c9e9ff58b83c0d3d63971a0a";
+                assert!(
+                    most_recent_patch
+                        .tags
+                        .iter()
+                        .any(|t| t.as_vec()[0].eq("t") && t.as_vec()[1].eq(COMMIT_PARENT_ID))
+                );
+                assert!(most_recent_patch.tags.iter().any(
+                    |t| t.as_vec()[0].eq("parent-commit") && t.as_vec()[1].eq(COMMIT_PARENT_ID)
+                ));
+
+                // root commit 't' tag
+                assert!(most_recent_patch.tags.iter().any(|t| t.as_vec()[0].eq("t")
+                    && t.as_vec()[1].eq("r-9ee507fc4357d7ee16a5d8901bedcd103f23c17d")));
+            }
+            Ok(())
+        }
+
+        #[test]
+        #[serial]
+        fn patch_tags_pr_event_as_root() -> Result<()> {
+            let (r51, r52, r53) = futures::executor::block_on(prep_run_create_pr())?;
+            for relay in [&r51, &r52, &r53] {
+                let patch_events: Vec<&nostr::Event> = relay
+                    .events
+                    .iter()
+                    .filter(|e| e.kind.as_u64().eq(&PATCH_KIND))
+                    .collect();
+
+                let most_recent_patch = patch_events[0];
+                let pr_event = relay
+                    .events
+                    .iter()
+                    .find(|e| e.kind.as_u64().eq(&PR_KIND))
+                    .unwrap();
+
+                let root_event_tag = most_recent_patch
+                    .tags
+                    .iter()
+                    .find(|t| {
+                        t.as_vec()[0].eq("e") && t.as_vec().len().eq(&4) && t.as_vec()[3].eq("root")
+                    })
+                    .unwrap();
+
+                assert_eq!(root_event_tag.as_vec()[1], pr_event.id.to_string());
+            }
+            Ok(())
+        }
+    }
+
+    mod cli_ouput {
+        use super::*;
+
+        async fn run_test_async() -> Result<()> {
+            let git_repo = prep_git_repo()?;
 
-//         p.expect_input(EXPECTED_NSEC_PROMPT)?
-//             .succeeds_with(TEST_KEY_1_NSEC)?;
+            let (mut r51, mut r52, mut r53) = (
+                Relay::new(8051, None),
+                Relay::new(8052, None),
+                Relay::new(8053, None),
+            );
 
-//         p.expect_password(EXPECTED_SET_PASSWORD_PROMPT)?
-//             .with_confirmation(EXPECTED_SET_PASSWORD_CONFIRM_PROMPT)?
-//             .succeeds_with(TEST_PASSWORD)?;
+            // // check relay had the right number of events
+            let cli_tester_handle = std::thread::spawn(move || -> Result<()> {
+                let mut p = cli_tester_create_pr(&git_repo);
+                expect_msgs_first(&mut p)?;
+                relay::expect_send_with_progress(
+                    &mut p,
+                    vec![
+                        (" [my-relay] [repo-relay] ws://localhost:8051", true, ""),
+                        (" [my-relay] ws://localhost:8052", true, ""),
+                        (" [repo-relay] ws://localhost:8053", true, ""),
+                    ],
+                    3,
+                )?;
+                p.expect_end_with_whitespace()?;
+                Ok(())
+            });
 
-//         p.expect_end_with(format!("logged in as {}\r\n",
-// TEST_KEY_1_NPUB).as_str())     })
-// }
+            // launch relay
+            let _ = join!(
+                r51.listen_until_close(),
+                r52.listen_until_close(),
+                r53.listen_until_close(),
+            );
+            cli_tester_handle.join().unwrap()?;
+            Ok(())
+        }
+
+        #[test]
+        #[serial]
+        fn check_cli_output() -> Result<()> {
+            futures::executor::block_on(run_test_async())?;
+            Ok(())
+        }
+    }
+
+    mod first_event_rejected_by_1_relay {
+        use super::*;
+
+        mod only_first_rejected_event_sent_to_relay {
+            use super::*;
+
+            async fn run_test_async() -> Result<()> {
+                let git_repo = prep_git_repo()?;
+
+                let (mut r51, mut r52, mut r53) = (
+                    Relay::new(8051, None),
+                    Relay::new(
+                        8052,
+                        Some(&|relay, client_id, event| -> Result<()> {
+                            relay.respond_ok(client_id, event, Some("Payment Required"))?;
+                            Ok(())
+                        }),
+                    ),
+                    Relay::new(8053, None),
+                );
+
+                // // check relay had the right number of events
+                let cli_tester_handle = std::thread::spawn(move || -> Result<()> {
+                    let mut p = cli_tester_create_pr(&git_repo);
+                    p.expect_end_eventually()?;
+                    Ok(())
+                });
+
+                // launch relay
+                let _ = join!(
+                    r51.listen_until_close(),
+                    r52.listen_until_close(),
+                    r53.listen_until_close(),
+                );
+                cli_tester_handle.join().unwrap()?;
+
+                assert_eq!(r52.events.len(), 1);
+
+                Ok(())
+            }
+
+            #[test]
+            #[serial]
+            fn only_first_rejected_event_sent_to_relay() -> Result<()> {
+                futures::executor::block_on(run_test_async())?;
+                Ok(())
+            }
+        }
+
+        mod cli_show_rejection_with_comment {
+            use super::*;
+
+            async fn run_test_async() -> Result<(Relay<'static>, Relay<'static>, Relay<'static>)> {
+                let git_repo = prep_git_repo()?;
+
+                let (mut r51, mut r52, mut r53) = (
+                    Relay::new(8051, None),
+                    Relay::new(
+                        8052,
+                        Some(&|relay, client_id, event| -> Result<()> {
+                            relay.respond_ok(client_id, event, Some("Payment Required"))?;
+                            Ok(())
+                        }),
+                    ),
+                    Relay::new(8053, None),
+                );
+
+                // // check relay had the right number of events
+                let cli_tester_handle = std::thread::spawn(move || -> Result<()> {
+                    let mut p = cli_tester_create_pr(&git_repo);
+                    expect_msgs_first(&mut p)?;
+                    relay::expect_send_with_progress(
+                        &mut p,
+                        vec![
+                            (" [my-relay] [repo-relay] ws://localhost:8051", true, ""),
+                            (
+                                " [my-relay] ws://localhost:8052",
+                                false,
+                                "error: Payment Required",
+                            ),
+                            (" [repo-relay] ws://localhost:8053", true, ""),
+                        ],
+                        3,
+                    )?;
+                    p.expect_end_with_whitespace()?;
+                    Ok(())
+                });
+
+                // launch relay
+                let _ = join!(
+                    r51.listen_until_close(),
+                    r52.listen_until_close(),
+                    r53.listen_until_close(),
+                );
+                cli_tester_handle.join().unwrap()?;
+                Ok((r51, r52, r53))
+            }
+
+            #[test]
+            #[serial]
+            fn cli_show_rejection_with_comment() -> Result<()> {
+                futures::executor::block_on(run_test_async())?;
+                Ok(())
+            }
+        }
+    }
+}