feat: add project search notifications (#1117)

Adds feedback in form of notifications for cloned and timed out project
searches. Additionally keeps the following and requested lists
up-to-date when in the Following tab of the Profile to give realtime
feedback.

Part of #984

proxy/api/src/http/notification.rs

@@ -51,20 +51,12 @@ mod handler {
         )]);
         let filter = |notification: Notification| async move {
             match notification.clone() {
-                Notification::LocalPeer(event) => Some(map_to_event(event)),
+                Notification::LocalPeer(event) => Some(Ok::<_, Infallible>(sse::json(event))),
             }
         };
 
         Ok(sse::reply(
             sse::keep_alive().stream(initial.chain(subscriber).filter_map(filter)),
         ))
     }
-
-    /// Helper for mapping [`Notification::LocalPeer`] events onto
-    /// [`sse::ServerSentEvent`]s.
-    fn map_to_event(
-        event: notification::LocalPeer,
-    ) -> Result<impl sse::ServerSentEvent, Infallible> {
-        Ok((sse::event(event.to_string()), sse::json(event)))
-    }
 }

proxy/api/src/notification.rs

@@ -2,7 +2,6 @@
 
 use std::{
     collections::HashMap,
-    fmt,
     sync::{
         atomic::{AtomicUsize, Ordering},
         Arc,
@@ -25,6 +24,26 @@ pub enum Notification {
 #[derive(Clone, Debug, Serialize)]
 #[serde(rename_all = "camelCase", tag = "type")]
 pub enum LocalPeer {
+    /// A request for a project was cloned successfully.
+    #[serde(rename_all = "camelCase")]
+    RequestCloned {
+        /// Origin the project was cloned from.
+        peer: coco::PeerId,
+        /// Urn of the cloned project.
+        urn: coco::Urn,
+    },
+    /// A request for a project was queried on the network.
+    #[serde(rename_all = "camelCase")]
+    RequestQueried {
+        /// Urn of the queried project.
+        urn: coco::Urn,
+    },
+    /// A request for a project timed out.
+    #[serde(rename_all = "camelCase")]
+    RequestTimedOut {
+        /// Urn of the timed out project.
+        urn: coco::Urn,
+    },
     /// Transition between two statuses occurred.
     #[serde(rename_all = "camelCase")]
     StatusChanged {
@@ -35,18 +54,20 @@ pub enum LocalPeer {
     },
 }
 
-impl fmt::Display for LocalPeer {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            Self::StatusChanged { .. } => write!(f, "LOCAL_PEER_STATUS_CHANGED"),
-        }
-    }
-}
-
 #[allow(clippy::wildcard_enum_match_arm)]
 impl MaybeFrom<PeerEvent> for Notification {
     fn maybe_from(event: PeerEvent) -> Option<Self> {
         match event {
+            PeerEvent::RequestCloned(url) => Some(Self::LocalPeer(LocalPeer::RequestCloned {
+                peer: url.authority,
+                urn: url.urn,
+            })),
+            PeerEvent::RequestQueried(urn) => {
+                Some(Self::LocalPeer(LocalPeer::RequestQueried { urn }))
+            },
+            PeerEvent::RequestTimedOut(urn) => {
+                Some(Self::LocalPeer(LocalPeer::RequestTimedOut { urn }))
+            },
             PeerEvent::StatusChanged(old, new) => {
                 Some(Self::LocalPeer(LocalPeer::StatusChanged { old, new }))
             },

proxy/coco/src/peer/run_state.rs

@@ -98,6 +98,8 @@ pub enum Event {
     PeerSynced(PeerId),
     /// Request fullfilled with a successful clone.
     RequestCloned(RadUrl),
+    /// Request is being cloned from a peer.
+    RequestCloning(RadUrl),
     /// Requested urn was queried on the network.
     RequestQueried(RadUrn),
     /// Waiting room interval ticked.
@@ -117,6 +119,7 @@ impl MaybeFrom<&Input> for Event {
             Input::PeerSync(SyncInput::Succeeded(peer_id)) => Some(Self::PeerSynced(*peer_id)),
             Input::Protocol(event) => Some(Self::Protocol(event.clone())),
             Input::Request(RequestInput::Cloned(url)) => Some(Self::RequestCloned(url.clone())),
+            Input::Request(RequestInput::Cloning(url)) => Some(Self::RequestCloning(url.clone())),
             Input::Request(RequestInput::Queried(urn)) => Some(Self::RequestQueried(urn.clone())),
             Input::Request(RequestInput::Tick) => Some(Self::RequestTick),
             Input::Request(RequestInput::TimedOut(urn)) => Some(Self::RequestTimedOut(urn.clone())),

ui/App.svelte

@@ -1,11 +1,11 @@
 <script>
   import Router, { push, location } from "svelte-spa-router";
 
+  import * as hotkeys from "./src/hotkeys.ts";
+  import "./src/localPeer.ts";
   import * as notification from "./src/notification.ts";
   import * as path from "./src/path.ts";
   import * as remote from "./src/remote.ts";
-  import * as hotkeys from "./src/hotkeys.ts";
-
   import { fetch, session as store } from "./src/session.ts";
 
   import {

ui/Modal/Search.svelte

@@ -26,12 +26,9 @@
   const urnValidation = urnValidationStore();
 
   const navigateToProject = (project: Project) => {
-    dispatch("hide");
+    reset();
     push(path.projectSource(project.id));
-  };
-  const navigateToUntracked = () => {
     dispatch("hide");
-    push(path.projectUntracked(value));
   };
   const onKeydown = (event: KeyboardEvent) => {
     switch (event.code) {
@@ -45,6 +42,7 @@
         }
         break;
       case "Escape":
+        reset();
         dispatch("hide");
         break;
     }
@@ -70,15 +68,10 @@
   }
   // Fire notification when a request has been created.
   $: if ($request.status === remote.Status.Success) {
-    notification.info(
-      "You’ll be notified on your profile when this project has been found.",
-      false,
-      "View profile",
-      () => {
-        dispatch("hide");
-        push(path.profileFollowing());
-      }
-    );
+    reset();
+    push(path.profileFollowing());
+    notification.info("You’ll be notified when this project has been found.");
+    dispatch("hide");
   }
 
   $: tracked = $store.status === remote.Status.Success;
@@ -157,7 +150,7 @@
 
       <div slot="error" style="padding: 1.5rem;">
         <div class="header typo-header-3">
-          <span class="id" on:click={navigateToUntracked}>{id}</span>
+          <span class="id">{id}</span>
           <FollowToggle on:follow={follow} style="margin-left: 1rem;" />
         </div>
 

ui/Screen/Profile/Following.svelte

@@ -11,6 +11,7 @@
   import { cancelRequest } from "../../src/project";
   import type { Project } from "../../src/project";
   import type { Authenticated } from "../../src/session";
+  import type { Urn } from "../../src/urn";
 
   import {
     EmptyState,
@@ -23,6 +24,9 @@
   } from "../../DesignSystem/Component";
 
   const session: Authenticated = getContext("session");
+  const onCancel = (urn: Urn): void => {
+    cancelRequest(urn).then(fetchFollowing);
+  };
   const onSelect = ({ detail: project }: { detail: Project }) => {
     push(path.projectSource(project.id));
   };
@@ -87,7 +91,7 @@
                     expanded
                     warning
                     following
-                    on:unfollow={() => cancelRequest(request.urn)} />
+                    on:unfollow={() => onCancel(request.urn)} />
                 </div>
               {/if}
             </div>

ui/src/localPeer.ts

@@ -1,4 +1,11 @@
+import { derived, writable, Readable } from "svelte/store";
+import { push } from "svelte-spa-router";
+
+import * as identity from "./identity";
+import * as notifiation from "./notification";
+import * as path from "./path";
 import * as remote from "./remote";
+import * as urn from "./urn";
 
 // TYPES
 export enum StatusType {
@@ -33,31 +40,102 @@ interface Online {
 
 type Status = Stopped | Offline | Started | Syncing | Online;
 
-enum EventKind {
-  StatusChanged = "LOCAL_PEER_STATUS_CHANGED",
+enum EventType {
+  RequestCloned = "requestCloned",
+  RequestQueried = "requestQueried",
+  RequestTimedOut = "requestTimedOut",
+  StatusChanged = "statusChanged",
 }
 
-interface StatusChanged {
-  type: EventKind.StatusChanged;
-  old: Status;
-  new: Status;
+interface RequestCloned {
+  type: EventType.RequestCloned;
+  peer: identity.PeerId;
+  urn: urn.Urn;
 }
 
-export type PeerEvent = StatusChanged;
+interface RequestQueried {
+  type: EventType.RequestQueried;
+  urn: urn.Urn;
+}
 
-// STATE
-const statusStore = remote.createStore<Status>();
-export const status = statusStore.readable;
+interface RequestTimedOut {
+  type: EventType.RequestTimedOut;
+  urn: urn.Urn;
+}
+
+export type Event =
+  | RequestCloned
+  | RequestQueried
+  | RequestTimedOut
+  | { type: EventType.StatusChanged; old: Status; new: Status };
 
-statusStore.start(() => {
+// STATE
+const eventStore = writable<Event | null>(null, set => {
   const source = new EventSource(
     "http://localhost:8080/v1/notifications/local_peer_events"
   );
 
-  source.addEventListener(EventKind.StatusChanged, (event: Event): void => {
-    const changed = JSON.parse((event as MessageEvent).data);
-    statusStore.success(changed.new);
+  source.addEventListener("message", (msg: MessageEvent): void => {
+    const event: Event = JSON.parse(msg.data);
+    set(event);
   });
 
   return (): void => source.close();
 });
+
+// Event handling.
+// FIXME(xla): Formalise event handling.
+eventStore.subscribe((event: Event | null): void => {
+  if (!event) {
+    return;
+  }
+
+  switch (event.type) {
+    case EventType.RequestCloned:
+      notifiation.info(
+        `Project for "${event.urn}" found and cloned.`,
+        false,
+        "Show Project",
+        () => push(path.projectSource(event.urn))
+      );
+
+      break;
+
+    case EventType.RequestTimedOut:
+      notifiation.error(`Search for "${event.urn}" failed.`);
+
+      break;
+  }
+});
+
+export const requestEvents: Readable<
+  RequestCloned | RequestQueried | RequestTimedOut | null
+> = derived(eventStore, (event: Event | null):
+  | RequestCloned
+  | RequestTimedOut
+  | RequestQueried
+  | null => {
+  if (!event) {
+    return null;
+  }
+
+  switch (event.type) {
+    case EventType.RequestCloned:
+    case EventType.RequestQueried:
+    case EventType.RequestTimedOut:
+      return event;
+
+    default:
+      return null;
+  }
+});
+
+export const status: Readable<remote.Data<Status>> = derived(
+  eventStore,
+  (event: Event | null, set: (status: remote.Data<Status>) => void): void => {
+    if (event && event.type === EventType.StatusChanged) {
+      set({ status: remote.Status.Success, data: event.new });
+    }
+  },
+  { status: remote.Status.Loading }
+);

ui/src/notification.ts

@@ -1,4 +1,4 @@
-import { Writable, writable } from "svelte/store";
+import { Readable, derived, get, writable } from "svelte/store";
 
 import * as config from "./config";
 import * as event from "./event";
@@ -25,8 +25,11 @@ interface Notification {
 type Notifications = Notification[];
 
 // STATE
-let notifications: Notifications = [];
-export const store: Writable<Notifications> = writable([]);
+const notificationsStore = writable([]);
+export const store: Readable<Notifications> = derived(
+  notificationsStore,
+  (state: Notifications) => state
+);
 
 // EVENTS
 enum Kind {
@@ -59,8 +62,10 @@ interface ShowInfo extends event.Event<Kind> {
 type Msg = Remove | ShowError | ShowInfo;
 
 const filter = (id: ID): void => {
-  notifications = notifications.filter(n => n.id !== id);
-  store.set(notifications);
+  const notifications = get(notificationsStore).filter(
+    (n: Notification) => n.id !== id
+  );
+  notificationsStore.set(notifications);
 };
 
 const show = (
@@ -86,8 +91,9 @@ const show = (
     },
   };
 
-  notifications = [notification, ...notifications];
-  store.set(notifications);
+  const notifications = get(notificationsStore);
+  notifications.unshift(notification);
+  notificationsStore.set(notifications);
 
   setTimeout(() => {
     filter(id);