diff --git a/src/features/agents/service.ts b/src/features/agents/service.ts index f7380fcaaf..6fe87efc06 100644 --- a/src/features/agents/service.ts +++ b/src/features/agents/service.ts @@ -54,7 +54,7 @@ import { removeRelayAgent, } from "./relay-removal"; import { communityDestination, relayOrigin } from "../communities/destination"; -import type { Communities } from "../communities/service"; +import type { ClientSnapshot, Communities } from "../communities/service"; import { prepareRemoteAgentAuthorization, type NipOaAuthorization, @@ -1359,10 +1359,11 @@ export { fromOwner } from "./owner-attestation"; /** What Agents needs of the communities: the viewer, the selected one, and * a connection to any joined one without selecting it. */ -export type AgentCommunities = Pick< - Communities, - "snapshot" | "subscribe" | "open" ->; +export type AgentCommunities = Readonly<{ + snapshot(): Readonly>; + subscribe: Communities["subscribe"]; + open: Communities["open"]; +}>; type Binding = { scope: string; origin: string; diff --git a/src/features/communities/page.test.ts b/src/features/communities/page.test.ts new file mode 100644 index 0000000000..80ae245356 --- /dev/null +++ b/src/features/communities/page.test.ts @@ -0,0 +1,103 @@ +import { Context } from "@deepseek-ai/cordis"; +import { afterEach, expect, it, vi } from "vitest"; +import type { ReadTransport } from "../relay/transport"; +import { keypair } from "../relay/testing"; +import { communityDestination } from "./destination"; +import { pageCommunities } from "./page"; + +const relays = vi.hoisted(() => ({ provide: vi.fn() })); +vi.mock("../relay/service", async (original) => { + const actual = await original(); + relays.provide.mockImplementation(actual.provideRelay); + return { ...actual, provideRelay: relays.provide }; +}); + +const roots: Context[] = []; +afterEach(async () => { + for (const ctx of roots.splice(0)) await ctx.fiber.dispose(); + relays.provide.mockClear(); +}); + +const alpha = communityDestination("wss://alpha.example").id; +const beta = communityDestination("wss://beta.example").id; +function setup() { + const viewer = keypair().pubkey; + const relay = keypair().pubkey; + const connected: string[] = []; + const ctx = new Context(); + roots.push(ctx); + const communities = pageCommunities(ctx, async (id) => { + connected.push(id); + return { + viewer, + relayAuthor: relay, + scope: id, + media: () => undefined, + query: async () => [], + } satisfies ReadTransport; + }); + const changes = vi.fn(); + communities.subscribe(changes); + return { viewer, communities, changes, connected }; +} + +it("opens nothing until the page that shows the communities sends them", () => { + const { communities } = setup(); + expect(communities.snapshot()).toEqual({ memberships: [], selected: null }); + expect(communities.open(alpha)).toBeUndefined(); + // Signed out there: no viewer to connect as. + communities.follow({ memberships: [alpha], selected: alpha }); + expect(communities.open(alpha)).toBeUndefined(); +}); + +it("connects a joined community for its agents with a session that keeps nothing on the device", async () => { + const { viewer, communities, changes, connected } = setup(); + communities.follow({ viewer, memberships: [alpha, beta], selected: beta }); + expect(changes).toHaveBeenCalledTimes(1); + expect(communities.snapshot()).toEqual({ + viewer, + memberships: [alpha, beta], + selected: beta, + }); + const relay = communities.open("wss://Alpha.example/"); + expect(relay).toBeDefined(); + expect(communities.open(alpha)).toBe(relay); + expect(communities.open("wss://gamma.example")).toBeUndefined(); + await vi.waitFor(() => expect(connected).toEqual([alpha])); + // No presence, nothing restored, and its sends and history in memory only. + const [, , presence, , , saved, , options] = + relays.provide.mock.calls[0] ?? []; + expect({ presence, saved, options }).toEqual({ + presence: undefined, + saved: undefined, + options: { ephemeral: true }, + }); +}); + +it("lets go of a community that is left, and of every one when the viewer changes", () => { + const { viewer, communities } = setup(); + communities.follow({ viewer, memberships: [alpha, beta], selected: alpha }); + const first = communities.open(alpha); + const second = communities.open(beta); + communities.follow({ viewer, memberships: [beta], selected: null }); + expect(communities.open(alpha)).toBeUndefined(); + expect(communities.open(beta)).toBe(second); + communities.follow({ viewer, memberships: [alpha, beta], selected: null }); + expect(communities.open(alpha)).not.toBe(first); + communities.follow({ + viewer: keypair().pubkey, + memberships: [alpha, beta], + selected: null, + }); + expect(communities.open(beta)).not.toBe(second); +}); + +it("tells its listeners only of what changed, and never selects a community it is not in", () => { + const { viewer, communities, changes } = setup(); + communities.follow({ viewer, memberships: [alpha], selected: alpha }); + communities.follow({ viewer, memberships: [alpha], selected: alpha }); + expect(changes).toHaveBeenCalledTimes(1); + communities.follow({ viewer, memberships: [alpha], selected: beta }); + expect(changes).toHaveBeenCalledTimes(2); + expect(communities.snapshot().selected).toBeNull(); +}); diff --git a/src/features/communities/page.ts b/src/features/communities/page.ts new file mode 100644 index 0000000000..07df206350 --- /dev/null +++ b/src/features/communities/page.ts @@ -0,0 +1,126 @@ +// FOUNDATION: Communities as a page that shows nothing follows them. The +// page that shows them owns the device's record: its memberships, selection +// and viewer. It sends them here each time they change. This page connects +// to the joined communities for its agents with sessions of its own that +// keep nothing on the device, so the other page stays the one owner of what +// the device saves, shows and clears. +import { Context } from "@deepseek-ai/cordis"; +import type { LocalAgentSource } from "../agents/local-activity"; +import type { LocalAgents } from "../agents/local"; +import type { IdentityNames } from "../identity-names/service"; +import { provideRelay, type RelayData } from "../relay/service"; +import type { ReadTransport } from "../relay/transport"; +import { communityDestination } from "./destination"; + +/** What this page follows of the communities. */ +export type FollowedCommunities = Readonly<{ + viewer?: string; + /** The joined communities' ids. */ + memberships: readonly string[]; + selected: string | null; +}>; + +export function pageCommunities( + ctx: Context, + connect: (id: string, signal: AbortSignal) => Promise, + options: Readonly<{ + identityNames?: IdentityNames; + agentChoices?: LocalAgents; + localAgents?: LocalAgentSource; + }> = {}, +) { + let state: FollowedCommunities = Object.freeze({ + memberships: [], + selected: null, + }); + let disposed = false; + const listeners = new Set<() => void>(); + // Each session owns a scope so leaving can dispose exactly that one. + const sessions = new Map< + string, + Readonly<{ relay: RelayData; scope: Context }> + >(); + const release = (id: string) => { + const session = sessions.get(id); + sessions.delete(id); + return session?.scope.fiber.dispose().catch((error: unknown) => { + console.warn(`Couldn't dispose the agents' session for ${id}`, error); + }); + }; + const idOf = (id: string) => { + try { + return communityDestination(id).id; + } catch { + return undefined; + } + }; + ctx.effect(() => () => { + disposed = true; + listeners.clear(); + return Promise.all([...sessions.keys()].map(release)).then(() => {}); + }); + return { + snapshot: () => state, + subscribe(listener: () => void) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + /** A joined community's session, for the viewer that joined it; it lives + * until the community is left or the viewer changes. */ + open(id: string): RelayData | undefined { + const key = idOf(id); + if (disposed || !key || !state.viewer || !state.memberships.includes(key)) + return undefined; + let session = sessions.get(key); + if (!session) { + const scope = new Context(); + session = Object.freeze({ + scope, + relay: provideRelay( + scope, + (signal) => connect(key, signal), + undefined, + options.identityNames, + options.agentChoices, + undefined, + options.localAgents, + { ephemeral: true }, + ), + }); + sessions.set(key, session); + } + return session.relay; + }, + /** What the page that shows the communities has now. */ + follow(next: FollowedCommunities) { + if (disposed) return; + const memberships = [ + ...new Set(next.memberships.flatMap((id) => idOf(id) ?? [])), + ]; + const selected = + next.selected === null ? null : (idOf(next.selected) ?? null); + const followed: FollowedCommunities = Object.freeze({ + ...(next.viewer !== undefined && { viewer: next.viewer }), + memberships: Object.freeze(memberships), + selected: selected && memberships.includes(selected) ? selected : null, + }); + if ( + followed.viewer === state.viewer && + followed.selected === state.selected && + followed.memberships.length === state.memberships.length && + followed.memberships.every( + (id, index) => state.memberships[index] === id, + ) + ) + return; + const viewerChanged = followed.viewer !== state.viewer; + state = followed; + for (const id of [...sessions.keys()]) + if (viewerChanged || !memberships.includes(id)) void release(id); + for (const listener of [...listeners]) listener(); + }, + }; +} +export type PageCommunities = ReturnType; diff --git a/src/features/relay/service.ts b/src/features/relay/service.ts index 88555afe75..4fbcf2dfb6 100644 --- a/src/features/relay/service.ts +++ b/src/features/relay/service.ts @@ -4,6 +4,7 @@ import type { LocalAgents } from "../agents/local"; import type { IdentityNames } from "../identity-names/service"; import type { PresenceActivity } from "../presence/activity"; import type { Context } from "@deepseek-ai/cordis"; +import type { OutboxStorage } from "./outbox"; import { createRelaySession, type RelaySession } from "./session"; import { createHeadPersistence } from "./persistence"; import { @@ -45,17 +46,26 @@ declare module "@deepseek-ai/cordis" { } } +/** Sends that live only as long as the page. */ +const PAGE_OUTBOX: OutboxStorage = { load: () => [], save() {} }; + /** App-owned session. Plugins receive read models, never a signer or transport. - * Reconnection replaces the session and invalidates all old async work. */ + * Reconnection replaces the session and invalidates all old async work. + * + * An `ephemeral` session is for a page that shows nothing, beside the page + * that does: it keeps its sends and history in memory only, and restores + * nothing, so it never adds to or clears what the other page saves. */ export function provideRelay( ctx: Context, connect?: (signal: AbortSignal) => Promise, presenceActivity?: PresenceActivity, identityNames?: IdentityNames, agentChoices?: LocalAgents, - resume?: { viewer: string; scope: string }, + saved?: { viewer: string; scope: string }, localAgents?: LocalAgentSource, + { ephemeral = false }: { ephemeral?: boolean } = {}, ) { + const resume = ephemeral ? undefined : saved; let disposed = false; let generation = 0; let controller: AbortController | undefined; @@ -246,10 +256,14 @@ export function provideRelay( "selected-channel", undefined, ), - persistence: createHeadPersistence( - transport.viewer, - transportOrigin(transport), - ), + ...(ephemeral + ? { outboxStorage: PAGE_OUTBOX } + : { + persistence: createHeadPersistence( + transport.viewer, + transportOrigin(transport), + ), + }), }); await successor.restore(); if ( diff --git a/src/features/relay/startup-service.test.ts b/src/features/relay/startup-service.test.ts index 6612ce1326..27ac847096 100644 --- a/src/features/relay/startup-service.test.ts +++ b/src/features/relay/startup-service.test.ts @@ -8,6 +8,11 @@ import { provideRelay } from "./service"; const storage = vi.hoisted(() => ({ create: vi.fn() })); vi.mock("./persistence", () => ({ createHeadPersistence: storage.create })); +const journal = vi.hoisted(() => ({ create: vi.fn() })); +vi.mock("./outbox", async (original) => ({ + ...(await original()), + browserOutboxStorage: journal.create, +})); const roots: Context[] = []; afterEach(async () => { for (const ctx of roots.splice(0)) await ctx.fiber.dispose(); @@ -238,3 +243,43 @@ it("disconnect during a coalesced cache clear does not reconnect", async () => { expect(data.snapshot().status).toBe("disconnected"); expect(connections).toHaveLength(1); }); +it("an ephemeral session saves, restores and clears nothing on the device", async () => { + const { source } = setup(); + storage.create.mockClear(); + journal.create.mockReset(); + journal.create.mockImplementation(() => ({ load: () => [], save() {} })); + const writing: ReadTransport = { + ...source, + writer: { + kinds: [9], + sign: async () => { + throw new Error("unused"); + }, + publish: async () => {}, + }, + }; + const open = (ephemeral: boolean) => { + const ctx = new Context(); + roots.push(ctx); + return provideRelay( + ctx, + async () => writing, + undefined, + undefined, + undefined, + { viewer: source.viewer, scope: "primary" }, + undefined, + { ephemeral }, + ); + }; + const page = open(true); + await vi.waitFor(() => expect(page.snapshot().status).toBe("ready")); + await page.clearCache(); + expect(storage.create).not.toHaveBeenCalled(); + expect(journal.create).not.toHaveBeenCalled(); + // The same session, kept on the device, uses both. + const kept = open(false); + await vi.waitFor(() => expect(kept.snapshot().status).toBe("ready")); + expect(storage.create).toHaveBeenCalled(); + expect(journal.create).toHaveBeenCalled(); +});