activity: feed endpoints

desk/app/activity.hoon

@@ -309,9 +309,34 @@
       [%x %all ~]
     ``activity-stream+!>(stream:base)
   ::
-      [%x %all start=@ count=@ ~]
+      [%x %all count=@ start=?(~ [u=@ ~])]
+    =/  start
+      ?~  start.pole  now.bowl
+      ?^  tim=(slaw %ud u.start.pole)  u.tim
+      (slav %da u.start.pole)
+    =/  count  (slav %ud count.pole)
     =-  ``activity-stream+!>((gas:on-event:a *stream:a -))
-    (tab:on-event:a stream:base `(slav %da start.pole) (slav %ud count.pole))
+    (bat:ex-event:a stream:base `start count)
+  ::
+      [%x %feed %init count=@ ~]
+    =-  ``activity-feed-init+!>(-)
+    =/  start  now.bowl
+    =/  count  (slav %ud count.pole)
+    :*  (feed %all start count)
+        (feed %mentions start count)
+        (feed %replies start count)
+    ==
+  ::
+      [%x %feed type=@ count=@ start=?(~ [u=@ ~])]
+    =/  start
+      ?~  start.pole  now.bowl
+      ?^  tim=(slaw %ud u.start.pole)  u.tim
+      (slav %da u.start.pole)
+    =/  count  (slav %ud count.pole)
+    =-  ``activity-feed+!>(-)
+    ~|  bad-feed-type+type.pole
+    ?>  ?=(?(%all %mentions %replies) type.pole)
+    (feed type.pole start count)
   ::
   ::  /each: unified feed (equality of outcome)
   ::TODO  want to be able to filter for specific events kind too, but that will
@@ -374,6 +399,96 @@
     ``activity-settings+!>(volume-settings)
   ==
 ::
+++  feed
+  |=  [type=?(%all %mentions %replies) start=time-id:a count=@ud]
+  |^
+  ^-  (list activity-bundle:a)
+  =-  happenings.-.-
+  ::  if start is now, need to increment to make sure we include latest
+  ::  event if that event somehow has now as its time
+  =/  real-start  ?:(=(start now.bowl) +(start) start)
+  %^  (dop:ex-event:a out)
+    (lot:on-event:a stream:base `*@da `real-start)   [~ count ~ ~]
+  |=  [acc=out =time =event:a]
+  ^-  [(unit event:a) ? out]
+  ?:  =(limit.acc 0)  [~ & acc]
+  :-  ~   :-  |
+  =/  =source:a  (determine-source -.event)
+  =/  latest
+    ?^  stored=(~(get by times.acc) source)  u.stored
+    ?~  new=(ram:on-event:a stream:(get-index source))
+      ::  should never happen but -\_(ツ)_/-
+      (sub start 1)
+    -.u.new
+  =/  new-times  (~(put by times.acc) source latest)
+  ::  we only care about posts/replies events that are notified, and we
+  ::  don't want to include events from sources whose latest event is
+  ::  after the start so we always get "new" sources when paging
+  ?.  ?&  notified.event
+          (lth latest start)
+          ?=(?(%post %reply %dm-post %dm-reply) -<.event)
+      ==
+    acc(times new-times)
+  :-  new-times
+  =/  mention=(unit activity-bundle:a)
+    ?.  |(?=(%all type) ?=(%mentions type))  ~
+    =/  is-mention
+      ?-  -<.event
+        %post  mention.event
+        %reply  mention.event
+        %dm-post  mention.event
+        %dm-reply  mention.event
+      ==
+    ?.  is-mention  ~
+    `[source time ~[[time event]]]
+  ?^  mention
+    [(sub limit.acc 1) (snoc happenings.acc u.mention) collapsed.acc]
+  =/  care
+    ?|  ?=(%all type)
+        &(?=(%replies type) ?=(?(%reply %dm-reply) -<.event))
+    ==
+  ::  make sure we care and haven't collapsed this event already
+  ?.  ?&  care
+          !(~(has in collapsed.acc) time)
+      ==
+    [limit happenings collapsed]:acc
+  =/  top  (top-messages source stream:(get-index source))
+  =/  collapsed
+    (~(gas in collapsed.acc) (turn top |=([=time-id:a *] time-id)))
+  [(sub limit.acc 1) (snoc happenings.acc [source time top]) collapsed]
+  +$  out
+    $:  times=(map source:a time-id:a)
+        limit=@ud
+        happenings=(list activity-bundle:a)
+        collapsed=(set time-id:a)
+    ==
+  --
+::
+++  top-messages
+  |=  [=source:a =stream:a]
+  |^
+  ^-  (list time-event:a)
+  =-  msgs.-.-
+  %^  (dop:ex-event:a out)  stream  [6 ~]
+  |=  [acc=out [=time =event:a]]
+  ?:  =(limit.acc 0)  [~ & acc]
+  ?:  child.event  [~ | acc]
+  ?.  ?=(?(%post %reply %dm-post %dm-reply) -<.event)  [~ | acc]
+  =/  is-mention
+    ?-  -<.event
+      %post  mention.event
+      %reply  mention.event
+      %dm-post  mention.event
+      %dm-reply  mention.event
+    ==
+  ?:  is-mention  [~ | acc]
+  [~ | [(sub limit.acc 1) (snoc msgs.acc [time event])]]
+  +$  out
+    $:  limit=@ud
+        msgs=(list time-event:a)
+    ==
+  --
+::
 ++  base
   ^-  index:a
   (~(got by indices) [%base ~])
@@ -388,14 +503,14 @@
     $(t +(t))
   =/  notify  notify:(get-volume inc)
   =/  =event:a  [inc notify |]
+  =/  =source:a  (determine-source inc)
   =?  cor  !importing
-    (give %fact ~[/] activity-update+!>([%add time-id event]))
+    (give %fact ~[/] activity-update+!>([%add source time-id event]))
   =?  cor  &(!importing notify)
     (give %fact ~[/notifications] activity-event+!>([time-id event]))
   =.  indices
     =/  =stream:a  (put:on-event:a stream:base time-id event)
     (~(put by indices) [%base ~] [stream reads:base])
-  =/  =source:a  (determine-source inc)
   ?+  -<.event  (add-to-index source time-id event)
       %chan-init
     =/  group-src  [%group group.event]

desk/lib/activity-json.hoon

@@ -144,6 +144,14 @@
         children+?~(children.sum ~ (activity u.children.sum))
     ==
   ::
+  ++  activity-bundle
+    |=  ab=activity-bundle:a
+    %-  pairs
+    :~  source+(source source.ab)
+        source-key+s+(string-source source.ab)
+        latest+s+(scot %ud latest.ab)
+        events+a+(turn events.ab time-event)
+    ==
   ++  event
     |=  e=event:a
     %-  pairs
@@ -226,7 +234,7 @@
   ++  time-event
     |=  te=time-event:a
     %-  pairs
-    :~  time+(time time.te)
+    :~  time+s+(scot %ud time.te)
         event+(event event.te)
     ==
   +|  %collections
@@ -276,6 +284,10 @@
     %+  turn  ~(tap by vm)
     |=  [e=event-type:a v=volume:a]
     [e (volume v)]
+  ++  feed
+    |=  f=feed:a
+    a+(turn f activity-bundle)
+  ::
   +|  %updates
   ++  update
     |=  u=update:a
@@ -288,10 +300,12 @@
     ==
   ::
   ++  added
-    |=  ad=time-event:a
+    |=  [src=source:a te=time-event:a]
     %-  pairs
-    :~  time+(time time.ad)
-        event+(event event.ad)
+    :~  source+(source src)
+        source-key+s+(string-source src)
+        time+(time time.te)
+        event+(event event.te)
     ==
   ::
   ++  read

desk/lib/mark-warmer.hoon

@@ -35,4 +35,7 @@
 /$  act-sum     %activity-summary     %json
 /$  act-vol     %activity-settings    %json
 /$  act-evt     %activity-event       %json
+/$  act-stream  %activity-stream      %json
+/$  act-feed    %activity-feed        %json
+/$  act-fe-it   %activity-feed-init   %json
 ~

desk/mar/activity/feed-init.hoon

@@ -0,0 +1,20 @@
+/-  a=activity
+/+  aj=activity-json
+|_  [all=feed:a mentions=feed:a replies=feed:a]
+++  grad  %noun
+++  grow
+  |%
+  ++  noun  [all mentions replies]
+  ++  json
+    =,  enjs:format
+    %-  pairs
+    :~  all+(feed:enjs:aj all)
+        mentions+(feed:enjs:aj mentions)
+        replies+(feed:enjs:aj replies)
+    ==
+  --
+++  grab
+  |%
+  ++  noun  [all=feed:a mentions=feed:a replies=feed:a]
+  --
+--

desk/mar/activity/feed.hoon

@@ -0,0 +1,14 @@
+/-  a=activity
+/+  aj=activity-json
+|_  =feed:a
+++  grad  %noun
+++  grow
+  |%
+  ++  noun  feed
+  ++  json  (feed:enjs:aj feed)
+  --
+++  grab
+  |%
+  ++  noun  feed:a
+  --
+--

desk/sur/activity.hoon

@@ -16,6 +16,8 @@
 +$  volume-map
   $~  default-volumes
   (map event-type volume)
+::  $feed: a set of grouped events
++$  feed  (list activity-bundle)
 +|  %actions
 ::  $action: how to interact with our activity stream
 ::
@@ -55,7 +57,7 @@
 ::    $adjust: the volume of a source was adjusted
 ::
 +$  update
-  $%  [%add time-event]
+  $%  [%add =source time-event]
       [%del =source]
       [%read =source =activity-summary]
       [%adjust =source volume-map=(unit volume-map)]
@@ -163,6 +165,12 @@
   ==
 +$  unread-point  [message-key count=@ud notify=_|]
 +$  volume  [unreads=? notify=?]
++$  activity-bundle
+  $:  =source
+      latest=time
+      events=(list time-event)
+  ==
+::
 +|  %primitives
 +$  whom
   $%  [%ship p=ship]

packages/shared/src/urbit/activity.ts

@@ -173,6 +173,13 @@ export interface ActivitySummary {
   children: Activity | null;
 }
 
+export interface ActivityBundle {
+  source: Source;
+  latest: string;
+  events: ActivityEvent[];
+  'source-key': string;
+}
+
 export type Activity = Record<string, ActivitySummary>;
 
 export type Indices = Record<string, IndexData>;
@@ -181,6 +188,8 @@ export type Stream = Record<string, ActivityEvent>;
 
 export type VolumeMap = Partial<Record<ExtendedEventType, Volume>>;
 
+export type ActivityFeed = ActivityBundle[];
+
 export type ReadAction =
   | { event: ActivityIncomingEvent }
   | { item: string }
@@ -218,8 +227,10 @@ export interface ActivityVolumeUpdate {
 
 export interface ActivityAddUpdate {
   add: {
+    source: Source;
     time: string;
     event: ActivityEvent;
+    'source-key': string;
   };
 }