Notifications and live Work pages for work items without a repository

44	44		#[serde(default)]
45	45		pub expires_at: Option<String>,
46	46		/// Its scopes as permissions: each resource it may use, by name, at
47		−	/// the highest level, such as `{"issues": "write"}`. Every resource at
	47	+	/// the highest level, such as `{"work": "write"}`. Every resource at
48	48		/// its highest when `scopes` is null.
49	49		#[serde(default)]
50	50		pub permissions: std::collections::BTreeMap<String, String>,
⋯
162	162		pub ids: Vec<String>,
163	163		}
164	164
	165	+	/// `members_among`: services only. Which of these people (by username,
	166	+	/// at most 200) are members of `workspace` (its slug). Returns their
	167	+	/// usernames, lowercased.
	168	+	#[derive(Debug, Default, Serialize, Deserialize)]
	169	+	pub struct MembersAmongArgs {
	170	+	pub workspace: String,
	171	+	pub usernames: Vec<String>,
	172	+	}
	173	+
165	174		/// `display_usernames`: how each of these people (by lowercased username,
166	175		/// at most 200) wrote their username, for showing it beside the key.
167	176		/// Returns a map from the lowercased username to its chosen case; people
⋯
937	946		#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
938	947		pub struct AgentScope {
939	948		pub repo: crate::repos::RepoPath,
940		−	/// API and MCP operation names, such as `create_issue`.
	949	+	/// API and MCP operation names, such as `open_work_item`.
941	950		pub operations: Vec<String>,
942	951		/// Set on a run credential: the run it belongs to, and what it may do
943	952		/// with git. See [`crate::credentials`].

761	761		}
762	762
763	763		/// `notify_by_email` on the identity service: one item, emailed to the
764		−	/// person it is for, if they can still read its repository and their
765		−	/// address is confirmed. Returns whether it was sent.
	764	+	/// person it is for, if they can still see what it is about and their
	765	+	/// address is confirmed: they can read its repository, or, for a work
	766	+	/// item (`workspace`), they are a member of its workspace (one filed with
	767	+	/// no repository has only that). Returns whether it was sent.
766	768		#[derive(Clone, Debug, Serialize, Deserialize)]
767	769		#[serde(rename_all = "camelCase")]
768	770		pub struct NotifyByEmailArgs {
769	771		pub username: String,
770		−	pub repo_id: String,
	772	+	/// Its repository: a pull request's, or a work item's origin.
	773	+	#[serde(default)]
	774	+	pub repo_id: Option<String>,
	775	+	/// A work item's workspace.
	776	+	#[serde(default, skip_serializing_if = "Option::is_none")]
	777	+	pub workspace: Option<String>,
771	778		/// The subject line and the first paragraph.
772	779		pub subject: String,
773	780		pub intro: String,

11	11		//!
12	12		//! A change says only what kind of thing moved and where: never a title, a
13	13		//! body, a name or anything else written in the repository.
	14	+	//!
	15	+	//! Work items live in workspaces, and one may have no repository at all:
	16	+	//! what happens to a workspace's work items is also told to the people
	17	+	//! watching its Work pages (`workspace_changed`, with
	18	+	//! [`workspace_change_of`] for each; a tab asks with `watch_workspace`,
	19	+	//! through the site, which checks they are a member first).
14	20
15	21		use serde::{Deserialize, Serialize};
16	22
⋯
145	151		Some(change)
146	152		}
147	153
	154	+	/// The notify service's method that takes [`WorkspaceChangedArgs`].
	155	+	pub const WORKSPACE_CHANGED: &str = "workspace_changed";
	156	+
	157	+	/// Something that happened to a workspace's work items, as its Work pages
	158	+	/// hear of it: what kind of thing moved, and the work item's number.
	159	+	#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
	160	+	pub struct WorkspaceChange {
	161	+	/// Its slug, lowercase.
	162	+	pub workspace: String,
	163	+	/// The event's type (`work_item.updated`, `comment.created`, `work.claimed`, …).
	164	+	pub kind: String,
	165	+	/// The work item it is on.
	166	+	#[serde(default, skip_serializing_if = "Option::is_none")]
	167	+	pub number: Option<u32>,
	168	+	/// RFC 3339.
	169	+	pub at: String,
	170	+	}
	171	+
	172	+	/// `workspace_changed`.
	173	+	#[derive(Clone, Debug, Default, Serialize, Deserialize)]
	174	+	pub struct WorkspaceChangedArgs {
	175	+	pub changes: Vec<WorkspaceChange>,
	176	+	}
	177	+
	178	+	/// What a workspace's Work pages hear of an event, or None when it is not
	179	+	/// about one of its work items: a work item's own events, and the events
	180	+	/// that name one (its comments and reactions, claims on it, what it is
	181	+	/// blocked by).
	182	+	pub fn workspace_change_of(event: &Event) -> Option<WorkspaceChange> {
	183	+	if !is_live(&event.kind) {
	184	+	return None;
	185	+	}
	186	+	let data = &event.data;
	187	+	let workspace = text(&data["workspace"])?.to_lowercase();
	188	+	let about_item = event.kind.starts_with("work_item.")
	189	+	\|\| text(&data["itemId"]).is_some()
	190	+	\|\| matches!(event.kind.as_str(), "dependency.changed" \| "work.claimed" \| "work.released" \| "work.handoff_requested");
	191	+	if !about_item {
	192	+	return None;
	193	+	}
	194	+	Some(WorkspaceChange {
	195	+	workspace,
	196	+	kind: event.kind.clone(),
	197	+	number: number(&data["number"]).or_else(\|\| number(&data["item"])),
	198	+	at: event.time.clone(),
	199	+	})
	200	+	}
	201	+
	202	+	/// The workspace changes a batch of events makes, in order, in calls of at
	203	+	/// most [`MAX_CHANGES`].
	204	+	pub fn workspace_changes_of(events: &[Event]) -> Vec<Vec<WorkspaceChange>> {
	205	+	let changes: Vec<WorkspaceChange> = events.iter().filter_map(workspace_change_of).collect();
	206	+	changes.chunks(MAX_CHANGES).map(<[WorkspaceChange]>::to_vec).collect()
	207	+	}
	208	+
148	209		/// The changes a batch of events makes, in order, in calls of at most
149	210		/// [`MAX_CHANGES`].
150	211		pub fn changes_of(events: &[Event]) -> Vec<Vec<RepoChange>> {
⋯
191	252		}
192	253
193	254		#[test]
	255	+	fn a_workspaces_work_pages_hear_of_its_work_items_only() {
	256	+	let data = json!({ "itemId": "wki_1", "workspace": "Acme", "number": 40, "repoIds": [], "title": "secret plans" });
	257	+	let change = workspace_change_of(&event("work_item.updated", None, data)).unwrap();
	258	+	assert_eq!(
	259	+	change,
	260	+	WorkspaceChange { workspace: "acme".into(), kind: "work_item.updated".into(), number: Some(40), at: "2026-10-11T10:00:00Z".into() }
	261	+	);
	262	+	// A comment on it, and a claim (which names it as `item`).
	263	+	let comment = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "commentId": "cmt_1" });
	264	+	assert_eq!(workspace_change_of(&event("comment.created", None, comment)).unwrap().number, Some(40));
	265	+	let claim = json!({ "workspace": "acme", "number": 40, "item": 40, "claim": {} });
	266	+	assert!(workspace_change_of(&event("work.claimed", None, claim)).is_some());
	267	+	// A pull request's comment, and what names no workspace, are not.
	268	+	let on_pull = json!({ "repoId": "rep_1", "number": 4, "pullId": "pr_4" });
	269	+	assert!(workspace_change_of(&event("comment.created", Some("rep_1"), on_pull)).is_none());
	270	+	assert!(workspace_change_of(&event("work_item.updated", None, json!({ "number": 4 }))).is_none());
	271	+	assert!(workspace_change_of(&event("audit.recorded", None, json!({ "workspace": "acme", "itemId": "wki_1" }))).is_none());
	272	+	}
	273	+
	274	+	#[test]
194	275		fn pull_requests_work_items_comments_and_checks_name_their_number() {
195	276		for kind in ["pull.updated", "pull.merged", "work_item.closed", "comment.created", "review.completed"] {
196	277		let change = change_of(&event(kind, Some("rep_1"), json!({ "number": 7, "title": "secret plans" }))).unwrap();

66	66		inbox_thread?: string \| null;
67	67		/**
68	68		* The chat card it is about, when that card has something to press,
69		−	* such as a draft issue (File issue, Discard). The toast and the notifications panel show these
	69	+	* such as a draft work item (File it, Discard). The toast and the notifications panel show these
70	70		* actions, and pressing one goes to the site's `card_action`, exactly as
71	71		* on the card itself.
72	72		*/
⋯
216	216		kind: string;
217	217		/** The full ref a push or deletion moved (`refs/heads/main`), or a run's. */
218	218		ref?: string \| null;
219		−	/** The issue or pull request it is on, or the pull request a run ran for. */
	219	+	/** The work item or pull request it is on, or the pull request a run ran for. */
220	220		number?: number \| null;
221	221		/** The commit: where a push left its ref, the one checked or reported on. */
222	222		sha?: string \| null;
⋯
228	228		at: string;
229	229		};
230	230
231		−	/** How long one `watchRepo` lasts: a page open longer asks again before it runs out. */
	231	+	/**
	232	+	* Something that happened to a workspace's work items, as its Work pages
	233	+	* hear of it (`watchWorkspace`): what kind of thing moved and the work
	234	+	* item's number, never anything written in it. A work item may have no
	235	+	* repository, so its pages hear of it here rather than per repository.
	236	+	*/
	237	+	export type WorkspaceChange = {
	238	+	/** Its slug, lowercase. */
	239	+	workspace: string;
	240	+	/** The event's type (`work_item.updated`, `comment.created`, `work.claimed`, …). */
	241	+	kind: string;
	242	+	/** The work item it is on. */
	243	+	number?: number \| null;
	244	+	/** RFC 3339. */
	245	+	at: string;
	246	+	};
	247	+
	248	+	/** How long one `watchRepo` (or `watchWorkspace`) lasts: a page open longer asks again before it runs out. */
232	249		export const REPO_WATCH_MS = 15 * 60_000;
233	250
234	251		/** What the feed socket sends a tab. */
⋯
252	269		\| { type: "me"; me: OwnPresence }
253	270		/** What changed in repositories you are watching the pages of (`watchRepo`). */
254	271		\| { type: "repo"; changes: RepoChange[] }
	272	+	/** What changed to the work items of workspaces whose Work pages you are watching (`watchWorkspace`). */
	273	+	\| { type: "workspace"; changes: WorkspaceChange[] }
255	274		/**
256	275		* The person cleared a conversation (their direct message with @g1t):
257	276		* the notifications from it are gone, so tabs drop them too.
⋯
361	380		/** For services: what changed in repositories (`crates/contracts/src/live.rs`), told to whoever watches each. */
362	381		repoChanged(changes: RepoChange[]): Promise<{ ok: boolean }>;
363	382		/**
	383	+	* The person's tabs hear what changes to the workspace's work items, for
	384	+	* `REPO_WATCH_MS`. The caller must have checked they are a member (who
	385	+	* sees all of its work items): notify checks nothing.
	386	+	*/
	387	+	watchWorkspace(user: Pick<User, "id">, workspace: string): Promise<{ ok: boolean; until: string \| null }>;
	388	+	/** For services: what changed to workspaces' work items (`crates/contracts/src/live.rs`), told to whoever watches each. */
	389	+	workspaceChanged(changes: WorkspaceChange[]): Promise<{ ok: boolean }>;
	390	+	/**
364	391		* For chat, after the person cleared a conversation: the notifications
365	392		* kept from it are deleted, its counts go to nothing, and a
366	393		* notification of a message from before the clear that arrives late is
⋯
398	425		markVisit: (user, workspace, at) => call("mark_visit", { user_id: user.id, workspace, at }),
399	426		watchRepo: (user, repoId, repo) => call("watch_repo", { user_id: user.id, repo_id: repoId, repo }),
400	427		repoChanged: (changes) => call("repo_changed", { changes }),
	428	+	watchWorkspace: (user, workspace) => call("watch_workspace", { user_id: user.id, workspace }),
	429	+	workspaceChanged: (changes) => call("workspace_changed", { changes }),
401	430		forgetConversation: (user, workspace, channelId) => call("forget_conversation", { user_id: user.id, workspace, channel_id: channelId }),
402	431		};
403	432		}

56	56
57	57		use g1t_contracts::credentials::Principal;
58	58		use g1t_contracts::events::{Event, WorkspaceRenamed};
59		−	use g1t_contracts::identity::{AGENT_ID, UsernamesArgs};
	59	+	use g1t_contracts::identity::{AGENT_ID, MembersAmongArgs, UsernamesArgs};
60	60		use g1t_contracts::inbox::*;
61	61		use g1t_contracts::repos::{PathByIdArgs, ReadableArgs, Repo, RepoPath};
62	62		use g1t_contracts::security::SeverityCounts;
⋯
1383	1383		/// may be emailed.
1384	1384		struct Written {
1385	1385		notice: Notice,
1386		−	/// The repository its email is checked against: none for a work item
1387		−	/// without an origin, which is not emailed.
	1386	+	/// What its email is checked against: its repository (a work item's
	1387	+	/// origin, which it may not have), and a work item's workspace, whose
	1388	+	/// members may see it.
1388	1389		repo_id: Option<String>,
	1390	+	workspace: Option<String>,
1389	1391		url: String,
1390	1392		event_id: String,
1391	1393		/// Its event leaves nothing to do (`CLOSES`): settled once written.
1392	1394		closing: bool,
1393	1395		/// Pushed live: not for someone who hears of a work item only by
1394		−	/// watching a repository it is linked to other than its origin.
	1396	+	/// watching a repository it is linked to other than its origin, unless
	1397	+	/// they are a member of its workspace, who may see it.
1395	1398		live: bool,
1396	1399		/// The thread it is on, which notify keeps with what it replays.
1397	1400		thread: String,
⋯
1845	1848		news.extend(rows.into_iter().map(\|row\| row.username));
1846	1849		}
1847	1850		}
	1851	+	// Who hears of a work item only by watching another of its
	1852	+	// repositories is pushed it live when they are a member of its
	1853	+	// workspace, who may see it (all one filed with no repository has).
	1854	+	let members: HashSet<String> = match &wanted.item {
	1855	+	Some(item) => {
	1856	+	let usernames: Vec<String> = told
	1857	+	.iter()
	1858	+	.filter(\|notice\| notice.reason == Reason::Subscribed && elsewhere.contains(&notice.username) && news.contains(&notice.username))
	1859	+	.map(\|notice\| notice.username.clone())
	1860	+	.collect();
	1861	+	if usernames.is_empty() {
	1862	+	HashSet::new()
	1863	+	} else {
	1864	+	let asked = MembersAmongArgs { workspace: item.workspace.clone(), usernames };
	1865	+	g1t_kit::call::<_, Vec<String>>(sources.identity, "members_among", &asked)
	1866	+	.await
	1867	+	.unwrap_or_else(\|error\| {
	1868	+	worker::console_error!("inbox: could not tell who is a member of {}: {error}", item.workspace);
	1869	+	Vec::new()
	1870	+	})
	1871	+	.into_iter()
	1872	+	.collect()
	1873	+	}
	1874	+	}
	1875	+	None => HashSet::new(),
	1876	+	};
1848	1877		Ok(told
1849	1878		.into_iter()
1850	1879		.filter(\|notice\| news.contains(&notice.username))
1851	1880		.map(\|notice\| Written {
1852		−	live: !(notice.reason == Reason::Subscribed && elsewhere.contains(&notice.username)),
	1881	+	live: !(notice.reason == Reason::Subscribed
	1882	+	&& elsewhere.contains(&notice.username)
	1883	+	&& !members.contains(&notice.username)),
1853	1884		notice,
1854	1885		repo_id: wanted.repo_id.clone(),
	1886	+	workspace: wanted.item.as_ref().map(\|item\| item.workspace.clone()),
1855	1887		url: item_url.clone(),
1856	1888		event_id: event.id.clone(),
1857	1889		closing: CLOSES.contains(&event.kind.as_str()),
⋯
1992	2024
1993	2025		/// Emails what was news to people who asked to be emailed for its reason.
1994	2026		/// Identity sends each, only to a confirmed address and only if the person
1995		−	/// can still read the repository (a work item's origin; one without is
1996		−	/// not emailed).
	2027	+	/// can still see it: they can read its repository, or are a member of a
	2028	+	/// work item's workspace (all one filed with no repository has).
1997	2029		async fn email(db: &D1Database, identity: &Fetcher, written: Vec<Written>) -> Result<()> {
1998		−	let written: Vec<Written> = written.into_iter().filter(\|item\| item.repo_id.is_some()).collect();
	2030	+	let written: Vec<Written> = written.into_iter().filter(\|item\| item.repo_id.is_some() \|\| item.workspace.is_some()).collect();
1999	2031		if written.is_empty() {
2000	2032		return Ok(());
2001	2033		}
⋯
2016	2048		"notify_by_email",
2017	2049		&NotifyByEmailArgs {
2018	2050		username: item.notice.username.clone(),
2019		−	repo_id: item.repo_id.clone().unwrap_or_default(),
	2051	+	repo_id: item.repo_id.clone(),
	2052	+	workspace: item.workspace.clone(),
2020	2053		subject: item.notice.title.clone(),
2021	2054		intro: item.notice.body.clone(),
2022	2055		quote: None,

6	6		//! repository only after the site checked its person may read it.
7	7		//!
8	8		//! A work item's events and comments go to every repository it is linked
9		−	//! to (`repoIds`), not only its origin, which it may not have.
	9	+	//! to (`repoIds`), not only its origin, which it may not have, and to its
	10	+	//! workspace's Work pages (`workspace_changed`), which hear of it with or
	11	+	//! without a repository.
10	12
11	13		use g1t_contracts::events::Event;
12		−	use g1t_contracts::live::{MAX_CHANGES, REPO_CHANGED, RepoChange, RepoChangedArgs, change_of};
	14	+	use g1t_contracts::live::{
	15	+	MAX_CHANGES, REPO_CHANGED, RepoChange, RepoChangedArgs, WORKSPACE_CHANGED, WorkspaceChangedArgs, change_of, workspace_changes_of,
	16	+	};
13	17		use worker::{Fetcher, Result};
14	18
15	19		/// The changes an event makes: one for its repository, and one for each
⋯
41	45		worker::console_error!("events: live pages not told: {error}");
42	46		}
43	47		}
	48	+	for changes in workspace_changes_of(events) {
	49	+	let told: Result<serde_json::Value> = g1t_kit::call(notify, WORKSPACE_CHANGED, &WorkspaceChangedArgs { changes }).await;
	50	+	if let Err(error) = told {
	51	+	worker::console_error!("events: live Work pages not told: {error}");
	52	+	}
	53	+	}
44	54		}
45	55
46	56		#[cfg(test)]

249	249		"notify_by_email",
250	250		&g1t_contracts::inbox::NotifyByEmailArgs {
251	251		username: username.clone(),
252		−	repo_id: first.repo_id.clone(),
	252	+	repo_id: Some(first.repo_id.clone()),
	253	+	workspace: None,
253	254		subject: notice.title.clone(),
254	255		intro: notice.body.clone(),
255	256		quote: None,

474	474		fn an_inbox_item_says_what_happened_and_why_it_was_sent() {
475	475		let a = g1t_contracts::inbox::NotifyByEmailArgs {
476	476		username: "ana".into(),
477		−	repo_id: "rep_1".into(),
	477	+	repo_id: Some("rep_1".into()),
	478	+	workspace: None,
478	479		subject: "bo asked you to review acme/rocket#7".into(),
479	480		intro: "Add the inbox".into(),
480	481		quote: None,

603	603		}
604	604
605	605		/// `notify_by_email`: an inbox item, emailed to the person it is for,
606		−	/// only at a confirmed address and only while they can still read the
607		−	/// repository it is about. Returns whether it was sent.
	606	+	/// only at a confirmed address and only while they can still see what
	607	+	/// it is about: a member of a work item's workspace, or someone who can
	608	+	/// read its repository. Returns whether it was sent.
608	609		async fn notify_by_email(&self, a: g1t_contracts::inbox::NotifyByEmailArgs) -> Result<bool> {
609	610		#[derive(Deserialize)]
610	611		struct Address {
⋯
619	620		let Some(user) = user.filter(\|user\| user.verified) else {
620	621		return Ok(false);
621	622		};
622		−	let readable: Vec<g1t_contracts::repos::Repo> = g1t_kit::call(
623		−	&self.env.service("REPOS")?,
624		−	"readable",
625		−	&g1t_contracts::repos::ReadableArgs {
626		−	ids: vec![a.repo_id.clone()],
627		−	viewer: Some(user.clone()),
628		−	},
629		−	)
630		−	.await?;
631		−	if readable.is_empty() {
632		−	return Ok(false);
	623	+	// A work item is its workspace members' to see, under the
	624	+	// workspace's policy; anything else, its repository's readers'.
	625	+	let member = a.workspace.as_deref().is_some_and(\|workspace\| g1t_contracts::access::workspace_role(&user, workspace).is_some());
	626	+	if !member {
	627	+	let Some(repo_id) = a.repo_id.clone().filter(\|id\| !id.is_empty()) else {
	628	+	return Ok(false);
	629	+	};
	630	+	let readable: Vec<g1t_contracts::repos::Repo> = g1t_kit::call(
	631	+	&self.env.service("REPOS")?,
	632	+	"readable",
	633	+	&g1t_contracts::repos::ReadableArgs { ids: vec![repo_id], viewer: Some(user.clone()) },
	634	+	)
	635	+	.await?;
	636	+	if readable.is_empty() {
	637	+	return Ok(false);
	638	+	}
633	639		}
634	640		let address = self
635	641		.db
⋯
646	652		Ok(true)
647	653		}
648	654
	655	+	/// `members_among`: which of these people are members of the
	656	+	/// workspace, by lowercased username.
	657	+	async fn members_among(&self, a: g1t_contracts::identity::MembersAmongArgs) -> Result<Vec<String>> {
	658	+	#[derive(Deserialize)]
	659	+	struct Named {
	660	+	username: String,
	661	+	}
	662	+	let usernames: Vec<String> = a.usernames.iter().take(200).map(\|name\| name.to_lowercase()).collect();
	663	+	if usernames.is_empty() {
	664	+	return Ok(Vec::new());
	665	+	}
	666	+	Ok(self
	667	+	.db
	668	+	.prepare(
	669	+	"SELECT lower(users.username) AS username FROM workspace_members
	670	+	JOIN workspaces ON workspaces.id = workspace_members.workspace_id
	671	+	JOIN users ON users.id = workspace_members.user_id
	672	+	WHERE workspaces.slug = lower(?1) AND workspaces.deleted_at IS NULL AND users.deleted_at IS NULL
	673	+	AND lower(users.username) IN (SELECT value FROM json_each(?2))",
	674	+	)
	675	+	.bind(&[a.workspace.trim().into(), serde_json::to_string(&usernames)?.into()])?
	676	+	.all()
	677	+	.await?
	678	+	.results::<Named>()?
	679	+	.into_iter()
	680	+	.map(\|row\| row.username)
	681	+	.collect())
	682	+	}
	683	+
649	684		async fn usernames(&self, a: UsernamesArgs) -> Result<std::collections::HashMap<String, String>> {
650	685		#[derive(serde::Deserialize)]
651	686		struct Named {
⋯
1033	1068		"user_by_username" => reply(&identity.user_by_username(args(body)?).await?),
1034	1069		"user_by_id" => reply(&identity.current_user(args(body)?).await?),
1035	1070		"usernames" => reply(&identity.usernames(args(body)?).await?),
	1071	+	"members_among" => reply(&identity.members_among(args(body)?).await?),
1036	1072		// The API: each username's chosen case, shown beside it.
1037	1073		"display_usernames" => reply(&identity.display_usernames(args(body)?).await?),
1038	1074		"accounts" => reply(&identity.accounts(args(body)?).await?),

24	24		* Live repository pages: a tab on a repository's page has the site ask
25	25		* the feed to watch it (`watchRepo`), which tells the repository's topic
26	26		* (src/topic.ts); the topic hands back what changes there (`repoChanged`),
27		−	* and the feed sends it to the tabs as a `repo` event.
	27	+	* and the feed sends it to the tabs as a `repo` event. A workspace's Work
	28	+	* pages the same way, on the workspace's topic (`watchWorkspace`,
	29	+	* `workspaceChanged`, a `workspace` event).
28	30		*
29	31		* The feed authorizes nothing: the Worker reaches it only for the person
30	32		* the site checked (src/index.ts).
⋯
42	44		type FeedNotification,
43	45		type FeedSeed,
44	46		type RepoChange,
	47	+	type WorkspaceChange,
45	48		type NotifyPreferences,
46	49		type NotifyStatus,
47	50		type PushDevice,
⋯
69	72		import type { Room } from "./room.ts";
70	73		import type { Topic } from "./topic.ts";
71	74		import { type Visit, lastVisit, markVisit } from "./visits.ts";
72		−	import { MAX_WATCHING, stillWatching, watchUntil } from "./watch.ts";
	75	+	import { MAX_WATCHING, stillWatching, watchUntil, workspaceTopic } from "./watch.ts";
73	76		import { sendPush, type Vapid } from "./webpush.ts";
74	77
75	78		export type FeedEnv = {
⋯
809	812		}
810	813
811	814		/**
	815	+	* The person's tabs hear what changes to the work items of `workspace`
	816	+	* (its Work pages) for REPO_WATCH_MS, on the workspace's topic, kept
	817	+	* with the repositories they watch. The site checked they are a member.
	818	+	*/
	819	+	async watchWorkspace(userId: string, workspace: string): Promise<{ ok: boolean; until: string \| null }> {
	820	+	return this.watchRepo(userId, workspaceTopic(workspace), workspace);
	821	+	}
	822	+
	823	+	/**
812	824		* What changed in a repository, from its topic: sent to the tabs, which
813	825		* each keep what is about the page they show. False when nobody here is
814	826		* watching it any more, so the topic drops this feed.
815	827		*/
816	828		async repoChanged(repoId: string, changes: RepoChange[]): Promise<boolean> {
817		−	const row = this.sql.exec<{ until: number }>("SELECT until FROM watching WHERE repo_id = ?", repoId).toArray()[0];
	829	+	return this.passOn(repoId, { type: "repo", changes });
	830	+	}
	831	+
	832	+	/** What changed to a workspace's work items, from its topic (`key`), as `repoChanged`. */
	833	+	async workspaceChanged(key: string, changes: WorkspaceChange[]): Promise<boolean> {
	834	+	return this.passOn(key, { type: "workspace", changes });
	835	+	}
	836	+
	837	+	private passOn(key: string, event: FeedEvent): boolean {
	838	+	const row = this.sql.exec<{ until: number }>("SELECT until FROM watching WHERE repo_id = ?", key).toArray()[0];
818	839		if (!stillWatching(row?.until, Date.now())) return false;
819	840		const sockets = this.openSockets();
820	841		if (!sockets.length) return false;
821		−	const text = JSON.stringify({ type: "repo", changes } satisfies FeedEvent);
	842	+	const text = JSON.stringify(event);
822	843		for (const socket of sockets) {
823	844		try {
824	845		socket.send(text);

27	27		* `calendar` or `integration`, for the person they act for;
28	28		* - events and actions: `repo_changed`, what changed in repositories,
29	29		* told to whoever watches their pages; and the site `watch_repo`, for a
30		−	* tab on a repository page whose person it checked may read it.
	30	+	* tab on a repository page whose person it checked may read it;
	31	+	* - events: `workspace_changed`, what changed to workspaces' work items,
	32	+	* told to whoever watches their Work pages; and the site
	33	+	* `watch_workspace`, for a tab on a Work page whose person it checked is
	34	+	* a member.
31	35		*
32	36		* Presence: each feed works out its person's from their tabs and tells
33	37		* one room per workspace (src/room.ts), which tells everyone there who is
⋯
50	54		import { onlyPeople } from "./presence.ts";
51	55		import { visitWorkspace } from "./visits.ts";
52	56		import { settledThreads } from "./prefs.ts";
53		−	import { byRepo, cleanRepoId, cleanRepoPath } from "./watch.ts";
	57	+	import { byRepo, byWorkspace, cleanRepoId, cleanRepoPath, cleanWorkspace, workspaceTopic } from "./watch.ts";
54	58
55	59		export { Feed } from "./feed.ts";
56	60		export { Room } from "./room.ts";
⋯
120	124		for (const result of results) if (result.status === "rejected") console.error("notify: a repository's change was not told", result.reason);
121	125		return Response.json({ ok: true });
122	126		}
	127	+	if (method === "workspace_changed") {
	128	+	// For services: each workspace's work item changes to its topic, which tells whoever watches its Work pages.
	129	+	const topics = env.TOPICS;
	130	+	if (!topics) return Response.json({ ok: true });
	131	+	const grouped = byWorkspace(args?.changes);
	132	+	const results = await Promise.allSettled(
	133	+	[...grouped].map(([slug, changes]) => {
	134	+	const key = workspaceTopic(slug);
	135	+	return topics.get(topics.idFromName(key)).workspaceChanged(key, changes);
	136	+	}),
	137	+	);
	138	+	for (const result of results) if (result.status === "rejected") console.error("notify: a workspace's change was not told", result.reason);
	139	+	return Response.json({ ok: true });
	140	+	}
123	141		if (method === "deliver") {
124	142		const items = (Array.isArray(args?.items) ? args.items : []).slice(0, MAX_DELIVERIES) as FeedDelivery[];
125	143		const byUser = new Map<string, FeedDelivery[]>();
⋯
163	181		if (!repoId \|\| !repo) return Response.json({ ok: false, until: null });
164	182		return Response.json(await stub.watchRepo(userId, repoId, repo));
165	183		}
	184	+	case "watch_workspace": {
	185	+	// The site checked the person is a member: members see all of its work items.
	186	+	const workspace = cleanWorkspace(args.workspace);
	187	+	if (!workspace) return Response.json({ ok: false, until: null });
	188	+	return Response.json(await stub.watchWorkspace(userId, workspace));
	189	+	}
166	190		case "forget_conversation": {
167	191		// For chat, after the person cleared their direct message with @g1t (src/forget.ts).
168	192		const channelId = cleanChannelId(args.channel_id);

1	1		/**
2	2		* One repository's live topic, as a Durable Object named by its id: who
3		−	* is watching its pages now.
	3	+	* is watching its pages now. A workspace's Work pages have one too, named
	4	+	* `workspace:<slug>` (src/watch.ts `workspaceTopic`).
4	5		*
5	6		* A feed (src/feed.ts) adds its person when one of their tabs opens a page
6	7		* of the repository, and again before the watch runs out (REPO_WATCH_MS);
⋯
15	16		*/
16	17		import { DurableObject } from "cloudflare:workers";
17	18
18		−	import type { RepoChange } from "@g1t/contracts";
	19	+	import type { RepoChange, WorkspaceChange } from "@g1t/contracts";
19	20
20	21		import type { Feed } from "./feed.ts";
21	22		import { MAX_WATCHERS, named } from "./watch.ts";
⋯
51	52
52	53		/** Hands `changes` to everyone watching now; returns how many feeds passed them on. */
53	54		async changed(repoId: string, changes: RepoChange[]): Promise<number> {
	55	+	return this.tell(changes.length, (feed, watcher) => feed.repoChanged(repoId, named(changes, watcher.repo)));
	56	+	}
	57	+
	58	+	/** A workspace's topic (`key`): hands its work items' `changes` to everyone watching its Work pages. */
	59	+	async workspaceChanged(key: string, changes: WorkspaceChange[]): Promise<number> {
	60	+	return this.tell(changes.length, (feed) => feed.workspaceChanged(key, changes));
	61	+	}
	62	+
	63	+	private async tell(
	64	+	count: number,
	65	+	send: (feed: DurableObjectStub<Feed>, watcher: { user_id: string; repo: string }) => Promise<boolean>,
	66	+	): Promise<number> {
54	67		const feeds = this.env.FEEDS;
55		−	if (!feeds \|\| !changes.length) return 0;
	68	+	if (!feeds \|\| !count) return 0;
56	69		const now = Date.now();
57	70		this.sql.exec("DELETE FROM watchers WHERE until <= ?", now);
58	71		const watchers = this.sql.exec<{ user_id: string; repo: string }>("SELECT user_id, repo FROM watchers").toArray();
59	72		if (!watchers.length) return 0;
60		−	const results = await Promise.allSettled(
61		−	watchers.map((watcher) => feeds.get(feeds.idFromName(watcher.user_id)).repoChanged(repoId, named(changes, watcher.repo))),
62		−	);
	73	+	const results = await Promise.allSettled(watchers.map((watcher) => send(feeds.get(feeds.idFromName(watcher.user_id)), watcher)));
63	74		let told = 0;
64	75		results.forEach((result, i) => {
65	76		if (result.status === "rejected") {

1	1		import assert from "node:assert/strict";
2	2		import { test } from "node:test";
3	3
4		−	import { MAX_CHANGES, WATCH_MS, byRepo, cleanChange, cleanRepoId, cleanRepoPath, named, stillWatching, watchUntil } from "./watch.ts";
	4	+	import {
	5	+	MAX_CHANGES,
	6	+	WATCH_MS,
	7	+	byRepo,
	8	+	byWorkspace,
	9	+	cleanChange,
	10	+	cleanRepoId,
	11	+	cleanRepoPath,
	12	+	cleanWorkspace,
	13	+	cleanWorkspaceChange,
	14	+	named,
	15	+	stillWatching,
	16	+	topicWorkspace,
	17	+	watchUntil,
	18	+	workspaceTopic,
	19	+	} from "./watch.ts";
5	20
6	21		const AT = "2026-10-11T10:00:00Z";
7	22
⋯
79	94		assert.equal(stillWatching(until, now + WATCH_MS), false);
80	95		assert.equal(stillWatching(null, now), false);
81	96		});
	97	+
	98	+	test("a workspace's changes keep only what one carries, by workspace", () => {
	99	+	const grouped = byWorkspace([
	100	+	{ workspace: "Acme", kind: "work_item.updated", number: 40, at: AT, title: "a secret title" },
	101	+	{ workspace: "acme", kind: "comment.created", number: 40, at: AT },
	102	+	{ workspace: "globex", kind: "work.claimed", at: AT },
	103	+	{ workspace: "not a slug!", kind: "work_item.updated", at: AT },
	104	+	{ workspace: "acme", kind: "<script>", at: AT },
	105	+	]);
	106	+	assert.deepEqual([...grouped.keys()], ["acme", "globex"]);
	107	+	assert.deepEqual(grouped.get("acme"), [
	108	+	{ workspace: "acme", kind: "work_item.updated", number: 40, at: AT },
	109	+	{ workspace: "acme", kind: "comment.created", number: 40, at: AT },
	110	+	]);
	111	+	assert.equal(cleanWorkspaceChange(null), null);
	112	+	assert.equal(cleanWorkspace(" Acme "), "acme");
	113	+	// Its topic sits beside the repositories', and never takes a repository's id.
	114	+	assert.equal(topicWorkspace(workspaceTopic("acme")), "acme");
	115	+	assert.equal(topicWorkspace("rep_1"), null);
	116	+	assert.equal(cleanRepoId(workspaceTopic("acme")), null);
	117	+	});

6	6		* A change says what kind of thing moved and where, never anything
7	7		* written in the repository; anything else sent along is dropped here.
8	8		*/
9		−	import type { RepoChange } from "@g1t/contracts";
	9	+	import type { RepoChange, WorkspaceChange } from "@g1t/contracts";
10	10
11	11		/** How long one watch lasts: REPO_WATCH_MS in @g1t/contracts, which the pages renew before. */
12	12		export const WATCH_MS = 15 * 60_000;
⋯
86	86		export function stillWatching(until: number \| null \| undefined, now: number): boolean {
87	87		return typeof until === "number" && until > now;
88	88		}
	89	+
	90	+	/** The topic a workspace's Work pages are watched on: one per workspace, beside the repositories'. */
	91	+	export function workspaceTopic(workspace: string): string {
	92	+	return `workspace:${workspace}`;
	93	+	}
	94	+
	95	+	/** The workspace a topic key names, or null when it is a repository's. */
	96	+	export function topicWorkspace(key: string): string \| null {
	97	+	return key.startsWith("workspace:") ? key.slice("workspace:".length) : null;
	98	+	}
	99	+
	100	+	/** A workspace's slug as services and the site send it, lowercase, or null. */
	101	+	export function cleanWorkspace(value: unknown): string \| null {
	102	+	if (typeof value !== "string") return null;
	103	+	const slug = value.trim().toLowerCase();
	104	+	return /^[a-z0-9][a-z0-9-]{0,99}$/.test(slug) ? slug : null;
	105	+	}
	106	+
	107	+	/** A workspace change as a service sent it, with only what one carries; null when it is not one. */
	108	+	export function cleanWorkspaceChange(value: unknown): WorkspaceChange \| null {
	109	+	if (!value \|\| typeof value !== "object") return null;
	110	+	const given = value as Record<string, unknown>;
	111	+	const workspace = cleanWorkspace(given.workspace);
	112	+	const kind = typeof given.kind === "string" && /^[a-z_]{1,40}\.[a-z_]{1,40}$/.test(given.kind) ? given.kind : null;
	113	+	if (!workspace \|\| !kind) return null;
	114	+	const change: WorkspaceChange = { workspace, kind, at: text(given.at, 40) ?? new Date().toISOString() };
	115	+	const number = count(given.number);
	116	+	if (number != null) change.number = number;
	117	+	return change;
	118	+	}
	119	+
	120	+	/** The workspace changes of a call, cleaned, by workspace, in the order they came. */
	121	+	export function byWorkspace(values: unknown): Map<string, WorkspaceChange[]> {
	122	+	const grouped = new Map<string, WorkspaceChange[]>();
	123	+	const list = Array.isArray(values) ? values.slice(0, MAX_CHANGES) : [];
	124	+	for (const value of list) {
	125	+	const change = cleanWorkspaceChange(value);
	126	+	if (!change) continue;
	127	+	grouped.set(change.workspace, [...(grouped.get(change.workspace) ?? []), change]);
	128	+	}
	129	+	return grouped;
	130	+	}

flagon-io/g1t

Comments