feat(host,panel): stream a workspace's pending invitations to owners' and admins' windows

packages/features/src/tasks-view.ts

@@ -1264,6 +1264,19 @@ export const tasksWorkspaceInvitationsAnswerSchema = EffectSchema.Union([
 
 export type TasksWorkspaceInvitationsAnswer = typeof tasksWorkspaceInvitationsAnswerSchema.Type;
 
+/**
+ * A followed workspace's pending invitations as they now stand, told by the
+ * host to an owner's or an admin's windows each time they move, or why
+ * there are none. A follow names the workspace as a people's follow does
+ * (`workspacePeopleFollowSchema`).
+ */
+export const workspaceInvitationsFollowedSchema = EffectSchema.Struct({
+  workspaceId: workspaceIdSchema,
+  view: tasksWorkspaceInvitationsAnswerSchema,
+});
+
+export type WorkspaceInvitationsFollowed = typeof workspaceInvitationsFollowedSchema.Type;
+
 /** A page about the account's GitHub connection, as the window hears it: the address to open in the browser, or that the service did not answer. */
 export const tasksGitHubPageAnswerSchema = EffectSchema.Union([
   EffectSchema.Struct({ url: EffectSchema.String }),

packages/features/src/workspace-client.ts

@@ -42,7 +42,11 @@ import {
   workspaceInvitationCancelSchema,
   workspaceInviteRequestSchema,
 } from "./workspace-invitation-wire.js";
-import type { WorkspaceMembersLiveAnswer, WorkspacesLiveAnswer } from "./workspace-live.js";
+import type {
+  WorkspaceInvitationsLiveAnswer,
+  WorkspaceMembersLiveAnswer,
+  WorkspacesLiveAnswer,
+} from "./workspace-live.js";
 import {
   type FormerMember,
   type LinearIssuesRequest,
@@ -239,6 +243,32 @@ export class HostedWorkspaceClient {
     );
   }
 
+  /**
+   * A workspace's pending invitations sealed for following over Neon
+   * Realtime, to an owner or an admin of it and for the account named, or
+   * with their rows inline where `poll` asks for them; signed out where the
+   * session is another account's, or why not.
+   */
+  invitationsLive(
+    workspaceId: string,
+    userId: string,
+    poll = false,
+    after?: string,
+  ): Effect.Effect<
+    WorkspaceCallResult<WorkspaceInvitationsLiveAnswer> | typeof LIVE_SIGNED_OUT,
+    never,
+    HttpClient.HttpClient
+  > {
+    return liveResult(
+      askLive(
+        this.#hosted.live.workspaceInvitations,
+        { workspaceId, [LIVE_USER_QUERY]: userId },
+        poll,
+        after,
+      ),
+    );
+  }
+
   /** A workspace created with the name given, the account its owner; a name the service would refuse by shape never travels. */
   create(
     name: string,

packages/features/src/workspace-live.ts

@@ -224,6 +224,8 @@ export const workspaceInvitationsLiveAnswerSchema = EffectSchema.Struct({
   queries: EffectSchema.Struct({ invitations: sealedLiveQuerySchema }),
 });
 
+export type WorkspaceInvitationsLiveAnswer = typeof workspaceInvitationsLiveAnswerSchema.Type;
+
 /** A workspace's pending invitations as their rows deliver them, expired ones among them. */
 export const workspaceInvitationsLiveRowsSchema = EffectSchema.Struct({
   invitations: EffectSchema.Array(

packages/gateway/src/protocol.ts

@@ -140,6 +140,15 @@ export const GATEWAY_METHOD = {
   WORKSPACES_PEOPLE_FOLLOW: "workspaces.peopleFollow",
   /** A followed workspace's people let go: no more `WORKSPACE_PEOPLE_CHANGED` events of it until it is followed again. */
   WORKSPACES_PEOPLE_UNFOLLOW: "workspaces.peopleUnfollow",
+  /**
+   * A workspace's pending invitations, followed by an owner or an admin from
+   * now until let go, over Neon Realtime: heard as
+   * `WORKSPACE_INVITATIONS_CHANGED` events, never answered; following again
+   * starts over.
+   */
+  WORKSPACES_INVITATIONS_FOLLOW: "workspaces.invitationsFollow",
+  /** A followed workspace's invitations let go: no more `WORKSPACE_INVITATIONS_CHANGED` events of it until it is followed again. */
+  WORKSPACES_INVITATIONS_UNFOLLOW: "workspaces.invitationsUnfollow",
   /** One workspace's photo, read from the service now as its bytes in base64; answered to the caller, never held in the view. */
   WORKSPACES_PHOTO: "workspaces.photo",
   /** The repositories members added to one workspace on their own GitHub tokens, read from the service now; answered to the caller, never held in the view. */
@@ -678,6 +687,8 @@ export const GATEWAY_EVENT = {
   AUTOMATIONS_CHANGED: "automations.changed",
   /** A followed workspace's people as they now stand, each time they move while followed. */
   WORKSPACE_PEOPLE_CHANGED: "workspacePeople.changed",
+  /** A followed workspace's pending invitations as they now stand, each time they move while followed. */
+  WORKSPACE_INVITATIONS_CHANGED: "workspaceInvitations.changed",
 } as const;
 
 export type GatewayEventKind = (typeof GATEWAY_EVENT)[keyof typeof GATEWAY_EVENT];

packages/host/src/act-port/workspace-procedures.test.ts

@@ -5,6 +5,7 @@ import { GITHUB_AUTHOR_ASSOCIATION, ISSUE_STATE } from "@luke/features/github-re
 import {
   TASK_CALL_FAILURE,
   type TasksWorkspaceGitHubConnectAnswer,
+  type WorkspaceInvitationsFollowed,
   type WorkspacePeopleFollowed,
 } from "@luke/features/tasks-view";
 import { WORKSPACE_PHOTO_TYPE, WORKSPACE_ROLE } from "@luke/features/workspace-wire";
@@ -203,6 +204,7 @@ function windowPort(actor: ActSender, options: { trusted?: boolean } = {}) {
   const browsed: string[] = [];
   /** Whoever hears the host tell a workspace's people, as the gateway's event reaches them. */
   const peopleListeners = new Set<(changed: WorkspacePeopleFollowed) => void>();
+  const invitationListeners = new Set<(changed: WorkspaceInvitationsFollowed) => void>();
   const script: Script = {
     notPermitted: false,
     alreadyInvited: false,
@@ -324,6 +326,14 @@ function windowPort(actor: ActSender, options: { trusted?: boolean } = {}) {
       peopleListeners.add(listener);
       return () => peopleListeners.delete(listener);
     },
+    workspaceInvitationsFollow: (scope) =>
+      ask(`invitationsFollow:${scope.workspaceId}`, () => undefined),
+    workspaceInvitationsUnfollow: (scope) =>
+      ask(`invitationsUnfollow:${scope.workspaceId}`, () => undefined),
+    onWorkspaceInvitationsChanged: (listener) => {
+      invitationListeners.add(listener);
+      return () => invitationListeners.delete(listener);
+    },
   };
   const port = testActPort(
     workspaceProcedures({
@@ -340,7 +350,11 @@ function windowPort(actor: ActSender, options: { trusted?: boolean } = {}) {
   const tellPeople = (changed: WorkspacePeopleFollowed) => {
     for (const listener of [...peopleListeners]) listener(changed);
   };
-  return { client, asked, browsed, script, tellPeople, peopleListeners };
+  /** The host telling a workspace's invitations. */
+  const tellInvitations = (changed: WorkspaceInvitationsFollowed) => {
+    for (const listener of [...invitationListeners]) listener(changed);
+  };
+  return { client, asked, browsed, script, tellPeople, peopleListeners, tellInvitations };
 }
 
 /** Any port kind's procedure, erased to the union of what every one takes and answers. */
@@ -679,3 +693,18 @@ it("the voice window is refused a workspace's people, and the host is not asked"
   }
   assert.deepEqual(asked, []);
 });
+
+it("a workspace's invitations stream apart from its people, and the host lets them go with the last stream", async () => {
+  const { client, asked, tellInvitations, tellPeople } = windowPort(PANEL);
+  const stream = await client[WORKSPACE_ACT_KIND.WORKSPACE_INVITATIONS](WORKSPACE);
+  const next = stream.next();
+  await until(() => asked.length === 1);
+
+  tellPeople({ ...WORKSPACE, view: { members: [GRACE] } });
+  tellInvitations({ ...WORKSPACE, view: { invitations: [INVITATION] } });
+
+  assert.deepEqual((await next).value, { invitations: [INVITATION] });
+  await stream.return(undefined);
+  await until(() => asked.length === 2);
+  assert.deepEqual(asked, [`invitationsFollow:${ACME.id}`, `invitationsUnfollow:${ACME.id}`]);
+});

packages/host/src/act-port/workspace-procedures.ts

@@ -26,6 +26,7 @@ import type {
   TasksWorkspaceTokenRepositoriesParams,
   TasksWorkspaceTokenRepositorySetParams,
   TasksWorkspaceTransferParams,
+  WorkspaceInvitationsFollowed,
   WorkspacePeopleFollow,
   WorkspacePeopleFollowed,
 } from "@luke/features/tasks-view";
@@ -71,6 +72,12 @@ export interface WorkspaceProceduresDependencies {
     workspacePeopleFollow(scope: WorkspacePeopleFollow): Effect.Effect<void>;
     workspacePeopleUnfollow(scope: WorkspacePeopleFollow): Effect.Effect<void>;
     onWorkspacePeopleChanged(listener: (followed: WorkspacePeopleFollowed) => void): () => void;
+    /** A workspace's pending invitations followed at the host, heard through `onWorkspaceInvitationsChanged`; following again starts over. */
+    workspaceInvitationsFollow(scope: WorkspacePeopleFollow): Effect.Effect<void>;
+    workspaceInvitationsUnfollow(scope: WorkspacePeopleFollow): Effect.Effect<void>;
+    onWorkspaceInvitationsChanged(
+      listener: (followed: WorkspaceInvitationsFollowed) => void,
+    ): () => void;
     workspacePhoto(params: TasksWorkspacePhotoParams): Effect.Effect<TasksWorkspacePhotoAnswer>;
     workspaceMemberRemove(
       params: TasksWorkspaceMemberRemoveParams,
@@ -159,24 +166,33 @@ function streamed<A>(iterable: AsyncIterable<A>): AsyncIteratorObject<A, undefin
   return self;
 }
 
+/** One kind of a workspace's follow, as a stream of it reads the host: its follow, its let-go, and its words. */
+interface WorkspaceFollowSide<View> {
+  readonly follow: (scope: WorkspacePeopleFollow) => Effect.Effect<void>;
+  readonly unfollow: (scope: WorkspacePeopleFollow) => Effect.Effect<void>;
+  readonly onChanged: (
+    listener: (changed: { readonly workspaceId: string; readonly view: View }) => void,
+  ) => () => void;
+}
+
 /**
- * A workspace's people for one stream a window holds open: the host follows
- * them afresh as the stream opens, so the window hears where they stand,
- * and lets them go once the last stream of them closes. Every word the host
- * tells of them reaches every stream of them.
+ * A workspace's followed read for one stream a window holds open: the host
+ * follows it afresh as the stream opens, so the window hears where it
+ * stands, and lets it go once the last stream of it closes. Every word the
+ * host tells of the workspace reaches every stream of it.
  */
-function workspacePeople(
-  host: WorkspaceProceduresDependencies["host"],
+function followedStream<View>(
+  side: WorkspaceFollowSide<View>,
   open: MutableHashMap.MutableHashMap<string, number>,
   turn: Semaphore.Semaphore,
   scope: WorkspacePeopleFollow,
-): Stream.Stream<TasksWorkspaceMembersAnswer> {
+): Stream.Stream<View> {
   const { workspaceId } = scope;
-  return Stream.callback<TasksWorkspaceMembersAnswer>((queue) =>
+  return Stream.callback<View>((queue) =>
     Effect.gen(function* () {
       yield* Effect.acquireRelease(
         Effect.sync(() =>
-          host.onWorkspacePeopleChanged((changed) => {
+          side.onChanged((changed) => {
             if (changed.workspaceId === workspaceId) Queue.offerUnsafe(queue, changed.view);
           }),
         ),
@@ -194,7 +210,7 @@ function workspacePeople(
                 Option.getOrElse(MutableHashMap.get(open, workspaceId), () => 0) + 1,
               ),
             ),
-            host.workspacePeopleFollow(scope),
+            side.follow(scope),
           ),
         ),
         () =>
@@ -206,7 +222,7 @@ function workspacePeople(
                 return Effect.void;
               }
               MutableHashMap.remove(open, workspaceId);
-              return host.workspacePeopleUnfollow(scope);
+              return side.unfollow(scope);
             }),
           ),
       );
@@ -216,16 +232,33 @@ function workspacePeople(
 
 export function workspaceProcedures(dependencies: WorkspaceProceduresDependencies) {
   const { host, openExternal } = dependencies;
-  /** How many streams stand open of each workspace's people, so the host lets them go only with the last. */
+  /** How many streams stand open of each workspace's people, and of its invitations, so the host lets each go only with the last. */
   const openPeople = MutableHashMap.empty<string, number>();
-  /** One turn per workspace whose people a window has followed, so the host hears its follows and unfollows in the order its streams opened and closed. */
-  const peopleTurns = new Map<string, Semaphore.Semaphore>();
-  const peopleTurnOf = (workspaceId: string): Semaphore.Semaphore => {
-    const standing = peopleTurns.get(workspaceId);
+  const openInvitations = MutableHashMap.empty<string, number>();
+  const people: WorkspaceFollowSide<TasksWorkspaceMembersAnswer> = {
+    follow: host.workspacePeopleFollow,
+    unfollow: host.workspacePeopleUnfollow,
+    onChanged: host.onWorkspacePeopleChanged,
+  };
+  const invitations: WorkspaceFollowSide<TasksWorkspaceInvitationsAnswer> = {
+    follow: host.workspaceInvitationsFollow,
+    unfollow: host.workspaceInvitationsUnfollow,
+    onChanged: host.onWorkspaceInvitationsChanged,
+  };
+  /** One turn per workspace a window has followed, of its people and of its invitations, so the host hears each one's follows and unfollows in the order its streams opened and closed. */
+  const turns = {
+    people: new Map<string, Semaphore.Semaphore>(),
+    invitations: new Map<string, Semaphore.Semaphore>(),
+  };
+  const turnOf = (
+    made: Map<string, Semaphore.Semaphore>,
+    workspaceId: string,
+  ): Semaphore.Semaphore => {
+    const standing = made.get(workspaceId);
     if (standing !== undefined) return standing;
-    const made = Semaphore.makeUnsafe(1);
-    peopleTurns.set(workspaceId, made);
-    return made;
+    const turn = Semaphore.makeUnsafe(1);
+    made.set(workspaceId, turn);
+    return turn;
   };
   return {
     [WORKSPACE_ACT_KIND.WORKSPACE_CREATE]: actPort[WORKSPACE_ACT_KIND.WORKSPACE_CREATE].effect(
@@ -343,11 +376,27 @@ export function workspaceProcedures(dependencies: WorkspaceProceduresDependencie
         yield* refuseUnlessPanel(context.actor, () => errors.REFUSED());
         return streamed(
           yield* Stream.toAsyncIterableEffect(
-            workspacePeople(host, openPeople, peopleTurnOf(input.workspaceId), input),
+            followedStream(people, openPeople, turnOf(turns.people, input.workspaceId), input),
           ),
         );
       },
     ),
+    // A workspace's pending invitations, pushed to an owner's or an admin's stream while it stands open.
+    [WORKSPACE_ACT_KIND.WORKSPACE_INVITATIONS]: actPort[
+      WORKSPACE_ACT_KIND.WORKSPACE_INVITATIONS
+    ].effect(function* ({ input, context, errors }) {
+      yield* refuseUnlessPanel(context.actor, () => errors.REFUSED());
+      return streamed(
+        yield* Stream.toAsyncIterableEffect(
+          followedStream(
+            invitations,
+            openInvitations,
+            turnOf(turns.invitations, input.workspaceId),
+            input,
+          ),
+        ),
+      );
+    }),
     // Who left is read on the same read as the members, so a service that names nobody answers none.
     [WORKSPACE_ACT_KIND.WORKSPACE_FORMER_MEMBERS_READ]: actPort[
       WORKSPACE_ACT_KIND.WORKSPACE_FORMER_MEMBERS_READ