Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src-tauri/crates/p2p
Submodule p2p updated 1 files
+153 −1 tests/beelay.rs
112 changes: 111 additions & 1 deletion src-tauri/crates/server/src/share_sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -329,6 +329,9 @@ fn touch(p: &Path) {
pub enum SyncRole {
Hoster,
Reader,
/// Received leg where the member's live role is editor/admin: it pushed its
/// SQLite edits into the doc as well as pulling the host's.
Editor,
}

/// `POST /api/share/{id}/sync` `reason` — why a pass skipped or came up short.
Expand Down Expand Up @@ -428,6 +431,20 @@ fn synced(role: SyncRole) -> SyncedReceipt {
}
}

/// This device's own live role on an e2ee share; `None` when keyhive is
/// unavailable or the query fails — treated as no push rights.
async fn self_member_role(node: &ShareNode, share_id: &str) -> Option<linxiv_share::Role> {
let me = node.self_member_id().ok()?;
tokio::time::timeout(
crate::route::share::SHARE_NET_TIMEOUT * 2,
node.query_role(share_id, me),
)
.await
.ok()?
.ok()
.flatten()
}

/// One sync pass for one share, honoring role (hoster/reader, by which doc file
/// exists) and settings (paused + direction). Route arm + interval task both call this.
pub async fn sync_share(
Expand Down Expand Up @@ -583,6 +600,20 @@ pub async fn sync_share(
let prior = load(&applied_dir(&e2ee_received_dir(&dir)), share_id)
.ok()
.or_else(|| ShareNode::e2ee_received(&dir, share_id).ok());
// This device's live role gates the push: editors/admins union their
// SQLite edits into the doc so sync_e2ee's flush uploads them; viewers
// keep the host→SQLite-only pull (the host serves viewer writes from a
// scratch core and drops them).
let editor = matches!(
self_member_role(&node, share_id).await,
Some(linxiv_share::Role::Edit | linxiv_share::Role::Admin)
);
if editor {
if let Some(fk) = state.with_conn(|c| project_svc::find_by_share_id(c, share_id))? {
let sp = state.with_conn(|c| build_shared_project(c, fk))?;
node.merge_received_edits(share_id, &sp).await?;
}
}
// Presence heartbeat: written before the dial so sync_e2ee's flush
// carries it. Viewer writes evaporate at the host (scratch core).
presence_heartbeat(&node, share_id).await;
Expand Down Expand Up @@ -653,7 +684,11 @@ pub async fn sync_share(
pending: pending.then_some(true),
reason,
undecryptable: (undecryptable > 0).then_some(undecryptable),
..synced(SyncRole::Reader)
..synced(if editor {
SyncRole::Editor
} else {
SyncRole::Reader
})
});
}

Expand Down Expand Up @@ -1131,4 +1166,79 @@ mod tests {

share.shutdown().await.unwrap();
}

// A member upgraded to editor pushes its SQLite edits back to the host: the
// reader leg unions them into the doc, uploads them, and reports
// SyncRole::Editor. A read-role member keeps the pull-only behavior.
#[tokio::test(flavor = "multi_thread")]
async fn e2ee_editor_reader_leg_pushes_local_edits() {
let a_dir = TempDir::new().unwrap();
let b_dir = TempDir::new().unwrap();
let node_a = ShareNode::bind_offline(a_dir.path(), &a_dir.path().join("p2p"))
.await
.unwrap();
let node_b = ShareNode::bind_offline(b_dir.path(), &b_dir.path().join("p2p"))
.await
.unwrap();
let share_a = ShareState::with_node(a_dir.path(), node_a);
let share_b = ShareState::with_node(b_dir.path(), node_b);
let state_a = mem_state();
let state_b = mem_state();

let sp = e2ee_sample();
state_a
.with_conn(|c| import_shared_project(c, &sp))
.unwrap();
let node_a = share_a.node().await.unwrap();
slow(node_a.publish_secure(&sp)).await.unwrap();

let node_b = share_b.node().await.unwrap();
let code = node_b.member_code().await.unwrap();
let (_member, invite) = node_a
.invite_member(E2EE_SID, &code, linxiv_share::Role::Edit)
.await
.unwrap();
assert_eq!(
slow(node_b.accept_invite(&invite)).await.unwrap().share_id,
E2EE_SID
);
let mirror = ShareNode::e2ee_received(b_dir.path(), E2EE_SID).unwrap();
state_b
.with_conn(|c| import_shared_project(c, &mirror))
.unwrap();

// B adds a paper to its linked project, then syncs as an editor.
let mut local = sp.clone();
local.papers.push(linxiv_share::SharedPaper {
source_id: "arxiv:local-edit".into(),
version: 1,
published: None,
title: "Edited".into(),
summary: "s".into(),
authors: vec![],
tags: vec![],
pdf_blob: None,
author_orcids: vec![],
});
state_b
.with_conn(|c| import_shared_project(c, &local))
.unwrap();

let v = slow(sync_share(&state_b, &share_b, E2EE_SID))
.await
.unwrap();
assert_eq!(v["synced"], json!(true));
assert_eq!(v["role"], json!("editor"));

// The host's live doc carries the pushed paper (the session's end ack is
// sent only after the host applied the upload).
let ids = slow(node_a.e2ee_paper_ids(E2EE_SID)).await.unwrap();
assert!(
ids.contains(&"arxiv:local-edit".to_string()),
"editor's SQLite edit must reach the host doc, got {ids:?}"
);

share_a.shutdown().await.unwrap();
share_b.shutdown().await.unwrap();
}
}
120 changes: 120 additions & 0 deletions src-tauri/crates/share/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -616,6 +616,71 @@ impl RemovalOutcome {
}
}

/// Union `local` (a SQLite projection) into `base` (a received doc's state) by
/// CRDT identity key — papers by source_id, notes/annotations by uuid, tags by
/// value — with `local` winning on collisions. `base` members absent from
/// `local` are kept: reconciling `local` alone would delete them, which for a
/// member pushing its edits means deleting host content it simply never
/// imported. Host-minted fields the projection can't reproduce (`pdf_blob`,
/// ORCIDs) survive a blank local value.
pub fn merge_shared_project(base: &mut SharedProject, local: &SharedProject) {
base.name = local.name.clone();
base.description = local.description.clone();
base.color = local.color;
for t in &local.tags {
if !base.tags.contains(t) {
base.tags.push(t.clone());
}
}
for lp in &local.papers {
match base.papers.iter_mut().find(|p| p.source_id == lp.source_id) {
Some(bp) => {
bp.version = lp.version;
bp.published = lp.published.clone();
bp.title = lp.title.clone();
bp.summary = lp.summary.clone();
bp.authors = lp.authors.clone();
bp.tags = lp.tags.clone();
if !lp.author_orcids.is_empty() {
bp.author_orcids = lp.author_orcids.clone();
}
if lp.pdf_blob.is_some() {
bp.pdf_blob = lp.pdf_blob.clone();
}
}
None => base.papers.push(lp.clone()),
}
}
for ln in &local.notes {
match base.notes.iter_mut().find(|n| n.uuid == ln.uuid) {
Some(bn) => {
bn.paper_source_id = ln.paper_source_id.clone().or(bn.paper_source_id.clone());
bn.title = ln.title.clone();
bn.body = ln.body.clone();
if ln.created_at.is_some() {
bn.created_at = ln.created_at.clone();
}
bn.updated_at = ln.updated_at.clone();
}
None => base.notes.push(ln.clone()),
}
}
for la in &local.annotations {
match base.annotations.iter_mut().find(|a| a.uuid == la.uuid) {
Some(ba) => {
ba.paper_source_id = la.paper_source_id.clone();
ba.anchor = la.anchor.clone();
ba.comment = la.comment.clone();
if la.created_at.is_some() {
ba.created_at = la.created_at.clone();
}
ba.updated_at = la.updated_at.clone();
}
None => base.annotations.push(la.clone()),
}
}
}

/// Destructive sibling of the additive import: remove what `fresh` lost relative
/// to `prior`. Project scope unlinks papers and subtracts project tags; library
/// scope (None) trashes them. Notes/annotations delete by uuid within the scope.
Expand Down Expand Up @@ -1383,6 +1448,61 @@ mod tests {
assert!(anns.is_empty());
}

// A reader's push unions its SQLite projection into the doc state: host
// items it never imported survive, its own edits win, and host-minted
// fields the projection can't reproduce (pdf_blob) are kept.
#[test]
fn merge_shared_project_unions_and_preserves_host_fields() {
let paper = |sid: &str, title: &str, pdf: Option<&str>| SharedPaper {
source_id: sid.into(),
version: 1,
published: None,
title: title.into(),
summary: String::new(),
authors: Vec::new(),
tags: Vec::new(),
pdf_blob: pdf.map(String::from),
author_orcids: Vec::new(),
};
let mut base = SharedProject {
share_id: "s".into(),
name: "Host".into(),
description: String::new(),
color: None,
tags: vec!["host-tag".into()],
papers: vec![paper("arxiv:1", "Host title", Some("ticket"))],
notes: vec![SharedNote {
uuid: "n1".into(),
paper_source_id: Some("arxiv:1".into()),
title: "host note".into(),
body: "b".into(),
created_at: None,
updated_at: None,
}],
annotations: vec![],
};
// Local knows arxiv:1 but not the host-only note; it added arxiv:3.
let mut local = base.clone();
local.name = "Local".into();
local.notes.clear();
local.papers = vec![
paper("arxiv:1", "Local title", None),
paper("arxiv:3", "Local new", None),
];

merge_shared_project(&mut base, &local);

assert_eq!(base.name, "Local");
let ids: Vec<&str> = base.papers.iter().map(|p| p.source_id.as_str()).collect();
assert!(ids.contains(&"arxiv:1") && ids.contains(&"arxiv:3"));
assert_eq!(base.papers[0].title, "Local title");
// host-minted ticket survives the projection's None
assert_eq!(base.papers[0].pdf_blob.as_deref(), Some("ticket"));
// host-only note kept; host tag kept
assert_eq!(base.notes.len(), 1);
assert!(base.tags.contains(&"host-tag".to_string()));
}

// The device actor pins every change `save` writes, changes carry real
// timestamps, and `doc_history`/`snapshot_at` read the log back.
#[test]
Expand Down
20 changes: 20 additions & 0 deletions src-tauri/crates/share/src/transport.rs
Original file line number Diff line number Diff line change
Expand Up @@ -912,6 +912,26 @@ impl ShareNode {
.await
.ok_or_else(|| ShareError::NotFound(share_id.to_string()))?
}

/// Push this device's SQLite edits into a received e2ee doc: union `local`
/// into the live doc and reconcile. Only an editor/admin should call this —
/// the host drops a viewer's writes from its scratch core. The next
/// [`Self::sync_e2ee`] flushes the merged doc to the host.
pub async fn merge_received_edits(&self, share_id: &str, local: &SharedProject) -> Result<()> {
self.with_e2ee_doc(share_id, |doc| {
let mut merged = if doc.get_heads().is_empty() {
local.clone()
} else {
autosurgeon::hydrate(doc).map_err(super::crdt)?
};
crate::merge_shared_project(&mut merged, local);
let mut tx = doc.transaction();
autosurgeon::reconcile(&mut tx, &merged).map_err(super::crdt)?;
tx.commit();
Ok(())
})
.await
}
}

/// Minimum gap between two heartbeat commits for one member; a reading
Expand Down
20 changes: 16 additions & 4 deletions src/components/share/ShareCard.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
importReceived,
listReceivedPapers,
syncShare,
type MemberRole,
type SharedSummary,
type SyncReceipt,
shareErrText,
Expand All @@ -24,7 +25,18 @@ import { SHARE_SYNC_MUTATION_KEY } from "../../lib/syncPill";
import { Button } from "../ui/button";
import { Spinner } from "../ui/spinner";

export type ShareRoleLabel = "Hoster" | "Reader";
/** A card's role: "Hoster" for a share this device published, otherwise the
* member's own live capability on a received e2ee share. Undefined while the
* role is unknown (pending join, node offline, plain mirror). */
export type ShareRoleLabel = "Hoster" | MemberRole;

const ROLE_DISPLAY: Record<ShareRoleLabel, string> = {
Hoster: "Hoster",
admin: "Admin",
"co-admin": "Co-admin",
editor: "Editor",
viewer: "Viewer",
};

function RolePill({ role }: { role: ShareRoleLabel }) {
const hosted = role === "Hoster";
Expand All @@ -39,7 +51,7 @@ function RolePill({ role }: { role: ShareRoleLabel }) {
: "var(--color-surface-2)",
}}
>
{role}
{ROLE_DISPLAY[role]}
</span>
);
}
Expand Down Expand Up @@ -152,7 +164,7 @@ export function ShareCard({
onSettings,
}: {
share: SharedSummary;
role: ShareRoleLabel;
role?: ShareRoleLabel;
onSettings: () => void;
}) {
const hosted = role === "Hoster";
Expand Down Expand Up @@ -261,7 +273,7 @@ export function ShareCard({
</span>
)}
<SyncBadge syncing={syncingThis} synced={share.synced_at != null} />
<RolePill role={role} />
{role && <RolePill role={role} />}
</div>
{share.description && (
<p
Expand Down
2 changes: 1 addition & 1 deletion src/components/share/ShareSettingsDialog.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ export function ShareSettingsDialog({
onClose,
}: {
share: SharedSummary;
role: ShareRoleLabel;
role?: ShareRoleLabel;
onClose: () => void;
}) {
const hosted = role === "Hoster";
Expand Down
8 changes: 5 additions & 3 deletions src/pages/SharePage.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ export default function SharePage() {
const [dialogOpen, setDialogOpen] = useState(false);
const [settingsFor, setSettingsFor] = useState<{
shareId: string;
role: ShareRoleLabel;
role?: ShareRoleLabel;
} | null>(null);
const [joinInput, setJoinInput] = useState("");
const [joining, setJoining] = useState(false);
Expand Down Expand Up @@ -135,9 +135,11 @@ export default function SharePage() {
}

const loading = published.isLoading || received.isLoading;
const cards: { share: SharedSummary; role: ShareRoleLabel }[] = [
// Published shares are hosted here; a received share carries the member's
// own live role (undefined while pending / node offline).
const cards: { share: SharedSummary; role?: ShareRoleLabel }[] = [
...(published.data ?? []).map((s) => ({ share: s, role: "Hoster" as const })),
...(received.data ?? []).map((s) => ({ share: s, role: "Reader" as const })),
...(received.data ?? []).map((s) => ({ share: s, role: s.role })),
];

return (
Expand Down
Loading
Loading