Resolve conflicts together, live 5e6d6484c28e

//! Live collaboration on the conflict resolver (hub spec §6.10): everyone

//! looking at the same conflict sees who else is there, the choices each

//! makes region by region (the last choice for a region wins), and the

//! resolution when one of them commits it. Rooms live in memory: they are

//! for people working together now, not a record.

use std::collections::{BTreeMap, HashMap};

use std::sync::{Arc, Mutex};

use tokio::sync::broadcast;

/// One conflict being resolved: repository, bookmark, path.

#[derive(Clone, Debug, PartialEq, Eq, Hash)]

pub struct RoomKey {

    pub repo_id: i64,

    pub bookmark: String,

    pub path: String,

}

/// A region's current choice: `side1`…, `all`, `none` or `text`, with the

/// text for `text`, and who made it.

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]

pub struct Choice {

    pub choice: String,

    pub text: String,

    pub by: String,

}

/// What a room tells its viewers.

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]

#[serde(tag = "kind", rename_all = "lowercase")]

pub enum LiveEvent {

    /// Everything so far, sent to a viewer who joins.

    Snapshot { present: Vec<String>, choices: BTreeMap<String, Choice>, keep: Option<Choice> },

    /// Who is looking now.

    Presence { present: Vec<String> },

    /// A viewer chose for region `region` (`keep` for a candidate page).

    Choice { region: String, choice: Choice, client: String },

    /// A resolution was committed; `url` is its change.

    Landed { url: String, by: String },

}

struct Room {

    tx: broadcast::Sender<LiveEvent>,

    /// Connected viewers by client id: their user names.

    present: HashMap<String, String>,

    choices: BTreeMap<String, Choice>,

}

impl Room {

    fn names(&self) -> Vec<String> {

        let mut v: Vec<String> = self.present.values().cloned().collect();

        v.sort();

        v.dedup();

        v

    }

}

#[derive(Default)]

pub struct Rooms {

    rooms: Mutex<HashMap<RoomKey, Room>>,

}

/// A viewer's place in a room; leaving happens when it is dropped (the

/// viewer's event stream closed).

pub struct Seat {

    rooms: Arc<Rooms>,

    key: RoomKey,

    client: String,

}

impl Drop for Seat {

    fn drop(&mut self) {

        self.rooms.leave(&self.key, &self.client);

    }

}

impl Rooms {

    /// Join `key` as `client` (one browser tab) of `user`: the events to

    /// follow, starting with a snapshot, and the seat that leaves on drop.

    pub fn join(&self, key: RoomKey, client: &str, user: &str) -> (broadcast::Receiver<LiveEvent>, LiveEvent) {

        let mut rooms = self.rooms.lock().unwrap();

        let room =

            rooms.entry(key).or_insert_with(|| Room { tx: broadcast::channel(256).0, present: HashMap::new(), choices: BTreeMap::new() });

        room.present.insert(client.to_string(), user.to_string());

        let rx = room.tx.subscribe();

        let present = room.names();

        let _ = room.tx.send(LiveEvent::Presence { present: present.clone() });

        let mut choices = room.choices.clone();

        let keep = choices.remove("keep");

        (rx, LiveEvent::Snapshot { present, choices, keep })

    }

    /// A seat for `client` in `key`, to hold while its stream is open.

    pub fn seat(self: &Arc<Self>, key: RoomKey, client: &str) -> Seat {

        Seat { rooms: Arc::clone(self), key, client: client.to_string() }

    }

    fn leave(&self, key: &RoomKey, client: &str) {

        let mut rooms = self.rooms.lock().unwrap();

        let Some(room) = rooms.get_mut(key) else { return };

        room.present.remove(client);

        if room.present.is_empty() {

            // Nobody is looking: forget the room and its choices.

            rooms.remove(key);

        } else {

            let _ = room.tx.send(LiveEvent::Presence { present: room.names() });

        }

    }

    /// Record and share a choice; `false` when nobody is in the room.

    pub fn choose(&self, key: &RoomKey, client: &str, region: &str, choice: Choice) -> bool {

        let mut rooms = self.rooms.lock().unwrap();

        let Some(room) = rooms.get_mut(key) else { return false };

        room.choices.insert(region.to_string(), choice.clone());

        let _ = room.tx.send(LiveEvent::Choice { region: region.to_string(), choice, client: client.to_string() });

        true

    }

    /// Tell a room its conflict was resolved.

    pub fn landed(&self, key: &RoomKey, url: &str, by: &str) {

        let rooms = self.rooms.lock().unwrap();

        if let Some(room) = rooms.get(key) {

            let _ = room.tx.send(LiveEvent::Landed { url: url.to_string(), by: by.to_string() });

        }

    }

    /// How many rooms are open (for tests and the admin page).

    pub fn open(&self) -> usize {

        self.rooms.lock().unwrap().len()

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    fn key() -> RoomKey {

        RoomKey { repo_id: 1, bookmark: "main".into(), path: "a.txt".into() }

    }

    fn pick(choice: &str, by: &str) -> Choice {

        Choice { choice: choice.into(), text: String::new(), by: by.into() }

    }

    #[test]

    fn viewers_see_each_other_and_each_others_choices() {

        let rooms = Arc::new(Rooms::default());

        let (mut a_rx, a_snap) = rooms.join(key(), "tab-a", "andy");

        assert_eq!(a_snap, LiveEvent::Snapshot { present: vec!["andy".into()], choices: BTreeMap::new(), keep: None });

        let seat_a = rooms.seat(key(), "tab-a");

        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Presence { .. })), "a's own arrival");

        assert!(rooms.choose(&key(), "tab-a", "1", pick("side2", "andy")));

        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Choice { region, .. }) if region == "1"));

        // A second viewer gets the choices so far and is announced.

        let (mut b_rx, b_snap) = rooms.join(key(), "tab-b", "bea");

        let seat_b = rooms.seat(key(), "tab-b");

        let LiveEvent::Snapshot { present, choices, .. } = b_snap else { panic!() };

        assert_eq!(present, vec!["andy", "bea"]);

        assert_eq!(choices["1"].choice, "side2");

        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into(), "bea".into()] });

        // The last choice for a region wins; keep is apart for candidate pages.

        rooms.choose(&key(), "tab-b", "1", pick("all", "bea"));

        rooms.choose(&key(), "tab-b", "keep", pick("1", "bea"));

        let _ = b_rx.try_recv();

        drop(seat_b);

        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Choice { region: "1".into(), choice: pick("all", "bea"), client: "tab-b".into() });

        let _ = a_rx.try_recv();

        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into()] }, "bea left");

        rooms.landed(&key(), "/andy/demo/change/abc", "andy");

        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Landed { .. })));

        // The last viewer leaving closes the room.

        drop(seat_a);

        assert_eq!(rooms.open(), 0);

        assert!(!rooms.choose(&key(), "tab-a", "1", pick("none", "andy")));

    }

}

  // Live conflict resolution (hub spec §6.10): who else is resolving this

  // conflict, their choices region by region (the last one wins), and the

  // resolution when someone commits it.

  var live = document.querySelector("[data-resolve-live]");

  if (live && window.EventSource && window.fetch) {

    var url = live.getAttribute("data-resolve-live"), csrf = live.getAttribute("data-csrf");

    var client = Math.random().toString(36).slice(2) + Date.now().toString(36);

    var present = document.querySelector("[data-live-present]"), landed = document.querySelector("[data-live-landed]");

    var applying = false;

    var fieldsOf = function (region) {

      return region === "keep"

        ? { radios: live.querySelectorAll('input[name="keep"]'), text: null }

        : { radios: live.querySelectorAll('input[name="r' + region + '"]'), text: live.querySelector('textarea[name="t' + region + '"]') };

    };

    var apply = function (region, c) {

      var f = fieldsOf(region);

      applying = true;

      f.radios.forEach(function (r) { r.checked = r.value === c.choice; });

      if (f.text && c.choice === "text") f.text.value = c.text;

      applying = false;

      var sec = f.radios.length ? f.radios[0].closest("section") : null;

      if (sec) sec.setAttribute("title", "chosen by " + c.by);

    };

    var send = function (region, choice, text) {

      fetch(url, { method: "POST", headers: { "content-type": "application/json" }, credentials: "same-origin",

        body: JSON.stringify({ _csrf: csrf, client: client, region: region, choice: choice, text: text || "" }) });

    };

    var timers = {};

    live.addEventListener("change", function (ev) {

      var t = ev.target;

      if (applying || !t.name || t.type !== "radio") return;

      var region = t.name === "keep" ? "keep" : t.name.slice(1);

      var ta = live.querySelector('textarea[name="t' + region + '"]');

      send(region, t.value, t.value === "text" && ta ? ta.value : "");

    });

    live.addEventListener("input", function (ev) {

      var t = ev.target;

      if (applying || t.tagName !== "TEXTAREA") return;

      var region = t.name.slice(1);

      live.querySelectorAll('input[name="r' + region + '"]').forEach(function (r) { r.checked = r.value === "text"; });

      clearTimeout(timers[region]);

      timers[region] = setTimeout(function () { send(region, "text", t.value); }, 400);

    });

    var es = new EventSource(url + "?client=" + encodeURIComponent(client));

    var showPresent = function (names) {

      present.hidden = names.length < 2;

      present.textContent = "Resolving together: " + names.join(", ") + ". Choices are shared as you make them; the last choice for a conflict wins.";

    };

    es.addEventListener("snapshot", function (e) {

      var d = JSON.parse(e.data);

      showPresent(d.present);

      Object.keys(d.choices).forEach(function (r) { apply(r, d.choices[r]); });

      if (d.keep) apply("keep", d.keep);

    });

    es.addEventListener("presence", function (e) { showPresent(JSON.parse(e.data).present); });

    es.addEventListener("choice", function (e) {

      var d = JSON.parse(e.data);

      if (d.client !== client) apply(d.region, d.choice);

    });

    es.addEventListener("landed", function (e) {

      var d = JSON.parse(e.data);

      es.close();

      landed.hidden = false;

      landed.textContent = "";

      landed.appendChild(document.createTextNode(d.by + " committed the resolution: "));

      var a = document.createElement("a");

      a.href = d.url;

      a.textContent = "see the change";

      landed.appendChild(a);

      live.querySelectorAll("input, textarea, button").forEach(function (x) { x.disabled = true; });

    });

  }

pub mod live;

    /// Who is resolving which conflict together (hub spec §6.10).

    pub rooms: Arc<live::Rooms>,

        rooms: Arc::new(live::Rooms::default()),

        .route("/{owner}/{name}/resolve-live/{bookmark}/{*path}", get(resolve::live_events).post(resolve::live_choose))

            let url = format!("/{owner}/{name}/change/{}", done.change.to_hex());

            app.rooms.landed(&room(&row, &bookmark, &path), &url, &me.account.username);

            return Ok(Redirect::to(&url).into_response());

            let url = format!("/{owner}/{name}/change/{}", done.change.to_hex());

            app.rooms.landed(&room(&row, &bookmark, &path), &url, &me.account.username);

            return Ok(Redirect::to(&url).into_response());

    let url = format!("/{owner}/{name}/change/{}", done.change.to_hex());

    app.rooms.landed(&room(&row, &bookmark, &path), &url, &me.account.username);

    Ok(Redirect::to(&url).into_response())

    }

}

// -------------------------------------------------------------- live (P1b)

fn room(row: &crate::db::repos::RepoRow, bookmark: &str, path: &str) -> crate::live::RoomKey {

    crate::live::RoomKey { repo_id: row.id, bookmark: bookmark.to_string(), path: path.to_string() }

}

#[derive(serde::Deserialize)]

pub struct LiveQuery {

    /// This browser tab's id, so a tab ignores its own choices.

    client: String,

}

/// `GET /{owner}/{name}/resolve-live/{bookmark}/{path}?client=…`: server-sent

/// events for everyone resolving this conflict (hub spec §6.10): a snapshot

/// of who is here and what they chose, then presence, choices, and the

/// resolution when it lands. Writers only, as the resolver.

pub async fn live_events(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, bookmark, path)): Path<(String, String, String, String)>,

    axum::extract::Query(q): axum::extract::Query<LiveQuery>,

) -> Result<axum::response::Sse<impl futures::Stream<Item = Result<axum::response::sse::Event, std::convert::Infallible>>>, Error> {

    use futures::StreamExt;

    let shell = Shell::new(&app, &format!("/{owner}/{name}/resolve/{bookmark}/{path}"), &viewer);

    let me = require(&viewer, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;

    let client: String = q.client.chars().filter(|c| c.is_ascii_alphanumeric() || *c == '-').take(64).collect();

    if client.is_empty() {

        return Err(Error::bad(shell, "A live view needs a client id."));

    let key = room(&row, &bookmark, &path);

    let (rx, snapshot) = app.rooms.join(key.clone(), &client, &me.account.username);

    let seat = app.rooms.seat(key, &client);

    let event = |e: &crate::live::LiveEvent| {

        let kind = match e {

            crate::live::LiveEvent::Snapshot { .. } => "snapshot",

            crate::live::LiveEvent::Presence { .. } => "presence",

            crate::live::LiveEvent::Choice { .. } => "choice",

            crate::live::LiveEvent::Landed { .. } => "landed",

        };

        Ok(axum::response::sse::Event::default().event(kind).data(serde_json::to_string(e).unwrap_or_default()))

    };

    let first = futures::stream::once(std::future::ready(event(&snapshot)));

    let rest = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(move |item| {

        // The seat lives as long as the stream: dropping it leaves the room.

        let _held = &seat;

        std::future::ready(item.ok().map(|e| event(&e)))

    });

    Ok(axum::response::Sse::new(first.chain(rest)).keep_alive(axum::response::sse::KeepAlive::default()))

}

#[derive(serde::Deserialize)]

pub struct LiveChoice {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    client: String,

    /// The region's number, or `keep` on a candidate page.

    region: String,

    choice: String,

    #[serde(default)]

    text: String,

}

/// `POST /{owner}/{name}/resolve-live/{bookmark}/{path}`: share a choice

/// with the others resolving this conflict. The last choice for a region

/// wins; nothing is committed until someone commits the resolution.

pub async fn live_choose(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, bookmark, path)): Path<(String, String, String, String)>,

    axum::Json(c): axum::Json<LiveChoice>,

) -> Result<axum::http::StatusCode, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/resolve/{bookmark}/{path}"), &viewer);

    let me = require(&viewer, &shell)?;

    csrf(&viewer, &c.csrf_token, &shell)?;

    let (_, row, _) = lookup_role(&app, &shell, &viewer, &owner, &name, Access::Write).await?;

    let valid_region = c.region == "keep" || c.region.parse::<usize>().is_ok();

    let valid_choice = matches!(c.choice.as_str(), "all" | "none" | "text")

        || c.choice.strip_prefix("side").is_some_and(|i| i.parse::<usize>().is_ok())

        || (c.region == "keep" && c.choice.parse::<usize>().is_ok());

    if !valid_region || !valid_choice || c.text.len() > 1 << 20 {

        return Err(Error::bad(shell, "That is not a choice."));

    }

    let choice = crate::live::Choice { choice: c.choice, text: c.text, by: me.account.username.clone() };

    Ok(if app.rooms.choose(&room(&row, &bookmark, &path), &c.client, &c.region, choice) {

        axum::http::StatusCode::NO_CONTENT

    } else {

        axum::http::StatusCode::GONE

    })

/// Read server-sent events from `body` until one of `kind` arrives.

async fn next_event(body: &mut Body, kind: &str) -> String {

    let mut buf = String::new();

    loop {

        let frame = tokio::time::timeout(std::time::Duration::from_secs(5), body.frame())

            .await

            .unwrap_or_else(|_| panic!("no {kind} event; got {buf}"))

            .expect("the stream ended")

            .unwrap();

        if let Ok(data) = frame.into_data() {

            buf.push_str(&String::from_utf8_lossy(&data));

        }

        while let Some(end) = buf.find("\n\n") {

            let event: String = buf.drain(..end + 2).collect();

            if event.contains(&format!("event: {kind}\n")) {

                return event;

            }

        }

    }

}

    // Live: one tab follows the room, another shares a choice (P1b).

    let live_req = Request::builder()

        .uri("/andy/demo/resolve-live/main/a.txt?client=tab-a")

        .header(header::COOKIE, format!("oxyhub={andy}"))

        .body(Body::empty())

        .unwrap();

    let res = oxygenhub::router(app.clone()).oneshot(live_req).await.unwrap();

    assert_eq!(res.status(), StatusCode::OK);

    let mut stream = res.into_body();

    let snapshot = next_event(&mut stream, "snapshot").await;

    assert!(snapshot.contains("\"present\":[\"andy\"]"), "{snapshot}");

    let choose = |region: &str, choice: &str| {

        let body = serde_json::json!({ "_csrf": csrf, "client": "tab-b", "region": region, "choice": choice, "text": "" }).to_string();

        Request::builder()

            .method("POST")

            .uri("/andy/demo/resolve-live/main/a.txt")

            .header(header::COOKIE, format!("oxyhub={andy}"))

            .header(header::CONTENT_TYPE, "application/json")

            .body(Body::from(body))

            .unwrap()

    };

    let r = oxygenhub::router(app.clone()).oneshot(choose("1", "side2")).await.unwrap();

    assert_eq!(r.status(), StatusCode::NO_CONTENT);

    let shared = next_event(&mut stream, "choice").await;

    assert!(shared.contains("\"side2\"") && shared.contains("\"tab-b\""), "{shared}");

    assert_ne!(oxygenhub::router(app.clone()).oneshot(choose("1", "rm -rf")).await.unwrap().status(), StatusCode::NO_CONTENT);

    let landed = next_event(&mut stream, "landed").await;

    assert!(landed.contains(&change), "the other tab hears the resolution land: {landed}");

    drop(stream);

For each bookmark whose state has conflicts: the file (or tree entry), the conflict kind, and each side with the change that produced it and its content, rendered with the same marker structure `oxy checkout` writes. For a text file, **resolve region by region** (`/{owner}/{name}/resolve/{bookmark}/{path}`, writers) shows the lines no one disagrees on and each conflict region with its sides and the change that wrote each; per region the user keeps one side, every side in order, none, or their own text (a side that records a deletion keeps the deletion). Regions are linkable (`#region-N`). The file is put back together and committed through the delegate as a web edit (§6.9), which makes it a `kind = resolution` change depending on every contributor; the page carries the bookmark's state and refuses to commit if the bookmark moved meanwhile. Malformed markers fall back to the editor, which also stays available: "edit the marked file" opens the markers in it. A name conflict (several files want one name, shown as `NAME.oxy-conflict-N`) resolves on the same URL by keeping one candidate: it takes the name as a move, keeping its identity, and the other candidates are removed. A replace conflict (a binary file replaced two ways) resolves the same way by keeping one version, each labelled with the change that wrote it; the commit replaces every version with it (Oxygen spec §9.3). Conflicts between directories or links are resolved from a client. Both resolver pages offer **Resolve in VS Code**, a `vscode://oxygen.oxygen-vscode/resolve?url=…&bookmark=…&path=…` link that oxygen-vscode handles: it finds the open checkout of that repository (or offers to clone it), switches to the bookmark after asking, pulls it, and opens the file. Resolving is live: everyone on the same resolver page (a room per repository, bookmark and path, over server-sent events at `/{owner}/{name}/resolve-live/{bookmark}/{path}`) sees who else is there and each other's choices as they are made, region by region, the last choice for a region winning (free text is shared whole, a moment after typing stops); a newcomer gets the choices so far; and when one of them commits, the others are told the resolution landed, with a link to it, and their form is disabled. Rooms are in memory and end when the last viewer leaves. The change page tags resolution changes. Otherwise the page gives the two-line client recipe.

<p class="small" data-live-present hidden></p>

<div class="strip" data-live-landed hidden></div>

<form method="post" action="{{ repo.url }}/resolve/{{ bookmark }}/{{ path }}" data-resolve-live="{{ repo.url }}/resolve-live/{{ bookmark }}/{{ path }}" data-csrf="{{ shell.csrf }}">

<p class="small" data-live-present hidden></p>

<div class="strip" data-live-landed hidden></div>

<form method="post" action="{{ repo.url }}/resolve/{{ bookmark }}/{{ path }}" data-resolve-live="{{ repo.url }}/resolve-live/{{ bookmark }}/{{ path }}" data-csrf="{{ shell.csrf }}">