OStatus partial support for group subscriptions:

* detection of group feeds is currently a nasty hack based on presence of '/groups/' in URL -- should use some property on the feed?
* listing for the remote group is kinda cruddy; needs to be named more cleanly
* still need to establish per-author profiles (easier once we have the updated Atom code in)
* group delivery probably not right yet
* saving of group messages still triggering some weird behavior

Added support for since_id and max_id on group timeline feeds as a free extra. Enjoy!

actions/apitimelinegroup.php

@@ -130,7 +130,7 @@ class ApiTimelineGroupAction extends ApiPrivateAuthAction
         case 'atom':
             $selfuri = common_root_url() .
                 'api/statusnet/groups/timeline/' .
-                    $this->group->nickname . '.atom';
+                    $this->group->id . '.atom';
             $this->showAtomTimeline(
                 $this->notices,
                 $title,

actions/showgroup.php

@@ -330,13 +330,13 @@ class ShowgroupAction extends GroupDesignAction
                      new Feed(Feed::RSS2,
                               common_local_url('ApiTimelineGroup',
                                                array('format' => 'rss',
-                                                     'id' => $this->group->nickname)),
+                                                     'id' => $this->group->id)),
                               sprintf(_('Notice feed for %s group (RSS 2.0)'),
                                       $this->group->nickname)),
                      new Feed(Feed::ATOM,
                               common_local_url('ApiTimelineGroup',
                                                array('format' => 'atom',
-                                                     'id' => $this->group->nickname)),
+                                                     'id' => $this->group->id)),
                               sprintf(_('Notice feed for %s group (Atom)'),
                                       $this->group->nickname)),
                      new Feed(Feed::FOAF,

classes/Notice.php

@@ -783,7 +783,7 @@ class Notice extends Memcached_DataObject
 
             $result = $gi->insert();
 
-            if (!result) {
+            if (!$result) {
                 common_log_db_error($gi, 'INSERT', __FILE__);
                 throw new ServerException(_('Problem saving group inbox.'));
             }
@@ -917,7 +917,7 @@ class Notice extends Memcached_DataObject
     /**
      * Same calculation as saveGroups but without the saving
      * @fixme merge the functions
-     * @return array of Group objects
+     * @return array of Group_inbox objects
      */
     function getGroups()
     {

classes/User_group.php

@@ -49,12 +49,12 @@ class User_group extends Memcached_DataObject
                                 array('id' => $this->id));
     }
 
-    function getNotices($offset, $limit)
+    function getNotices($offset, $limit, $since_id=null, $max_id=null)
     {
         $ids = Notice::stream(array($this, '_streamDirect'),
                               array(),
                               'user_group:notice_ids:' . $this->id,
-                              $offset, $limit);
+                              $offset, $limit, $since_id, $max_id);
 
         return Notice::getStreamByIds($ids);
     }

lib/util.php

@@ -697,7 +697,7 @@ function common_group_link($sender_id, $nickname)
 {
     $sender = Profile::staticGet($sender_id);
     $group = User_group::getForNickname($nickname);
-    if ($group && $sender->isMember($group)) {
+    if ($sender && $group && $sender->isMember($group)) {
         $attrs = array('href' => $group->permalink(),
                        'class' => 'url');
         if (!empty($group->fullname)) {

plugins/OStatus/OStatusPlugin.php

@@ -107,11 +107,11 @@ class OStatusPlugin extends Plugin
 
     /**
      * Set up a PuSH hub link to our internal link for canonical timeline
-     * Atom feeds for users.
+     * Atom feeds for users and groups.
      */
     function onStartApiAtom(Action $action)
     {
-        if ($action instanceof ApiTimelineUserAction) {
+        if ($action instanceof ApiTimelineUserAction || $action instanceof ApiTimelineGroupAction) {
             $id = $action->arg('id');
             if (strval(intval($id)) === strval($id)) {
                 // Canonical form of id in URL?

plugins/OStatus/actions/feedsubsettings.php

@@ -217,6 +217,15 @@ class FeedSubSettingsAction extends ConnectSettingsAction
                 throw new ServerException("Feed profile was not saved properly.");
             }
 
+            if ($this->feedinfo->isGroup()) {
+                if ($user->isMember($profile)) {
+                    $this->showForm(_m('Already a member!'));
+                } elseif (Group_member::join($this->feedinfo->group_id, $user->id)) {
+                    $this->showForm(_m('Joined remote group!'));
+                } else {
+                    $this->showForm(_m('Remote group join failed!'));
+                }
+            } else {
                 if ($user->isSubscribed($profile)) {
                     $this->showForm(_m('Already subscribed!'));
                 } elseif ($user->subscribeTo($profile)) {
@@ -226,6 +235,7 @@ class FeedSubSettingsAction extends ConnectSettingsAction
                 }
             }
         }
+    }
 
     function validateAndPreview()
     {

plugins/OStatus/classes/Feedinfo.php

@@ -89,7 +89,8 @@ class Feedinfo extends Memcached_DataObject
     function table()
     {
         return array('id' => DB_DATAOBJECT_INT + DB_DATAOBJECT_NOTNULL,
-                     'profile_id' => DB_DATAOBJECT_INT + DB_DATAOBJECT_NOTNULL,
+                     'profile_id' => DB_DATAOBJECT_INT,
+                     'group_id' => DB_DATAOBJECT_INT,
                      'feeduri' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
                      'homeuri' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
                      'huburi' =>  DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
@@ -111,7 +112,9 @@ class Feedinfo extends Memcached_DataObject
                                    /*extra*/ null,
                                    /*auto_increment*/ true),
                      new ColumnDef('profile_id', 'integer',
-                                   null, false),
+                                   null, true),
+                     new ColumnDef('group_id', 'integer',
+                                   null, true),
                      new ColumnDef('feeduri', 'varchar',
                                    255, false, 'UNI'),
                      new ColumnDef('homeuri', 'varchar',
@@ -176,6 +179,7 @@ class Feedinfo extends Memcached_DataObject
 
     /**
      * @param FeedMunger $munger
+     * @param boolean $isGroup is this a group record?
      * @return Feedinfo
      */
     public static function ensureProfile($munger)
@@ -217,6 +221,22 @@ class Feedinfo extends Memcached_DataObject
             }
 
             $feedinfo->profile_id = $profile->id;
+            if ($feedinfo->isGroup()) {
+                $group = new User_group();
+                $group->nickname = $profile->nickname . '@remote'; // @fixme
+                $group->fullname = $profile->fullname;
+                $group->homepage = $profile->homepage;
+                $group->location = $profile->location;
+                $group->created = $profile->created;
+                $group->insert();
+
+                if ($avatar) {
+                    $group->setOriginal($filename);
+                }
+
+                $feedinfo->group_id = $group->id;
+            }
+
             $result = $feedinfo->insert();
             if (empty($result)) {
                 throw new FeedDBException($feedinfo);
@@ -231,6 +251,14 @@ class Feedinfo extends Memcached_DataObject
         return $feedinfo;
     }
 
+    /**
+     * Damn dirty hack!
+     */
+    function isGroup()
+    {
+        return (strpos($this->feeduri, '/groups/') !== false);
+    }
+
     /**
      * Send a subscription request to the hub for this feed.
      * The hub will later send us a confirmation POST to /feedsub/callback.
@@ -325,6 +353,7 @@ class Feedinfo extends Memcached_DataObject
             $dupe = new Notice();
             $dupe->uri = $notice->uri;
             if ($dupe->find(true)) {
+                // @fixme we might have to do individual and group delivery separately!
                 common_log(LOG_WARNING, __METHOD__ . ": tried to save dupe notice for entry {$notice->uri} of feed {$this->feeduri}");
                 continue;
             }
@@ -333,9 +362,25 @@ class Feedinfo extends Memcached_DataObject
                 $id = $notice->insert();
                 Event::handle('EndNoticeSave', array($notice));
             }
-            $notice->addToInboxes();
-
             common_log(LOG_INFO, __METHOD__ . ": saved notice {$notice->id} for entry $index of update to \"{$this->feeduri}\"");
+
+            common_log(LOG_DEBUG, "going to check group delivery...");
+            if ($this->group_id) {
+                $group = User_group::staticGet($this->group_id);
+                if ($group) {
+                    common_log(LOG_INFO, __METHOD__ . ": saving to local shadow group $group->id $group->nickname");
+                    $groups = array($group);
+                } else {
+                    common_log(LOG_INFO, __METHOD__ . ": lost the local shadow group?");
+                }
+            } else {
+                common_log(LOG_INFO, __METHOD__ . ": no local shadow groups");
+                $groups = array();
+            }
+            common_log(LOG_DEBUG, "going to add to inboxes...");
+            $notice->addToInboxes($groups, array());
+            common_log(LOG_DEBUG, "added to inboxes.");
+
             $hits++;
         }
         if ($hits == 0) {

plugins/OStatus/lib/feedmunger.php

@@ -221,7 +221,7 @@ class FeedMunger
         $notice->uri = $link;
         $notice->url = $link;
         $notice->content = $this->noticeFromEntry($entry);
-        $notice->rendered = common_render_content($notice->content, $notice);
+        $notice->rendered = common_render_content($notice->content, $notice); // @fixme this is failing on group posts
         $notice->created = common_sql_date($entry->updated); // @fixme
         $notice->is_local = Notice::GATEWAY;
         $notice->source = 'feed';

plugins/OStatus/lib/hubdistribqueuehandler.php

@@ -34,6 +34,14 @@ class HubDistribQueueHandler extends QueueHandler
     {
         assert($notice instanceof Notice);
 
+        $this->pushUser($notice);
+        foreach ($notice->getGroups() as $group) {
+            $this->pushGroup($notice, $group->group_id);
+        }
+    }
+    
+    function pushUser($notice)
+    {
         // See if there's any PuSH subscriptions, including OStatus clients.
         // @fixme handle group subscriptions as well
         // http://identi.ca/api/statuses/user_timeline/1.atom
@@ -43,18 +51,40 @@ class HubDistribQueueHandler extends QueueHandler
         $sub = new HubSub();
         $sub->topic = $feed;
         if ($sub->find()) {
-            common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $feed");
-            $qm = QueueManager::get();
             $atom = $this->userFeedForNotice($notice);
+            $this->pushFeeds($atom, $sub);
+        } else {
+            common_log(LOG_INFO, "No PuSH subscribers for $feed");
+        }
+    }
+
+    function pushGroup($notice, $group_id)
+    {
+        $feed = common_local_url('ApiTimelineGroup',
+                                 array('id' => $group_id,
+                                       'format' => 'atom'));
+        $sub = new HubSub();
+        $sub->topic = $feed;
+        if ($sub->find()) {
+            common_log(LOG_INFO, "Building PuSH feed for $feed");
+            $atom = $this->groupFeedForNotice($group_id, $notice);
+            $this->pushFeeds($atom, $sub);
+        } else {
+            common_log(LOG_INFO, "No PuSH subscribers for $feed");
+        }
+    }
+
+    
+    function pushFeeds($atom, $sub)
+    {
+        common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic");
+        $qm = QueueManager::get();
         while ($sub->fetch()) {
-                common_log(LOG_INFO, "Prepping PuSH distribution to $sub->callback for $feed");
+            common_log(LOG_INFO, "Prepping PuSH distribution to $sub->callback for $sub->topic");
             $data = array('sub' => clone($sub),
                           'atom' => $atom);
             $qm->enqueue($data, 'hubout');
         }
-        } else {
-            common_log(LOG_INFO, "No PuSH subscribers for $feed");
-        }
     }
 
     /**
@@ -83,5 +113,29 @@ class HubDistribQueueHandler extends QueueHandler
         common_log(LOG_DEBUG, $feed);
         return $feed;
     }
+
+    function groupFeedForNotice($group_id, $notice)
+    {
+        // @fixme this feels VERY hacky...
+        // should probably be a cleaner way to do it
+
+        ob_start();
+        $api = new ApiTimelineGroupAction();
+        $args = array('id' => $group_id,
+                      'format' => 'atom',
+                      'max_id' => $notice->id,
+                      'since_id' => $notice->id - 1);
+        $api->prepare($args);
+        $api->handle($args);
+        $feed = ob_get_clean();
+        
+        // ...and override the content-type back to something normal... eww!
+        // hope there's no other headers that got set while we weren't looking.
+        header('Content-Type: text/html; charset=utf-8');
+
+        common_log(LOG_DEBUG, $feed);
+        return $feed;
+    }
+
 }