OStatus refactoring to clean up profile vs feed and fix up subscription issues.

PuSH subscription maintenance broken back out to FeedSub, letting Ostatus_profile deal with the profile level (user or group, with unique id URI)

8 files changed, 741 insertions, 852 deletions

diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php
index e548a151c..4ebe4551e 100644
--- a/plugins/OStatus/OStatusPlugin.php
+++ b/plugins/OStatus/OStatusPlugin.php
@@ -1,17 +1,7 @@
 <?php
 /*
-StatusNet Plugin: 0.9
-Plugin Name: FeedSub
-Plugin URI: http://status.net/wiki/Feed_subscription
-Description: FeedSub allows subscribing to real-time updates from external feeds supporting PubHubSubbub protocol.
-Version: 0.1
-Author: Brion Vibber <brion@status.net>
-Author URI: http://status.net/
-*/
-
-/*
  * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2009, StatusNet, Inc.
+ * Copyright (C) 2009-2010, StatusNet, Inc.
  *
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Affero General Public License as published by
@@ -28,17 +18,12 @@ Author URI: http://status.net/
  */
 
 /**
- * @package FeedSubPlugin
+ * @package OStatusPlugin
  * @maintainer Brion Vibber <brion@status.net>
  */
 
 if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
 
-define('FEEDSUB_SERVICE', 100); // fixme -- avoid hardcoding these?
-
-// We bundle the XML_Parse_Feed library...
-set_include_path(get_include_path() . PATH_SEPARATOR . dirname(__FILE__) . '/extlib');
-
 class FeedSubException extends Exception
 {
 }
@@ -259,24 +244,6 @@ class OStatusPlugin extends Plugin
     }
 
     /**
-     * Notify remote server when one of our users subscribes.
-     * @fixme Check and restart the PuSH subscription if needed
-     *
-     * @param User $user
-     * @param Profile $other
-     * @return hook return value
-     */
-    function onEndSubscribe($user, $other)
-    {
-        $oprofile = Ostatus_profile::staticGet('profile_id', $other->id);
-        if ($oprofile) {
-            // Notify the remote server of the unsub, if supported.
-            $oprofile->notify($user->getProfile(), ActivityVerb::FOLLOW, $oprofile);
-        }
-        return true;
-    }
-
-    /**
      * Notify remote server and garbage collect unused feeds on unsubscribe.
      * @fixme send these operations to background queues
      *
@@ -309,6 +276,7 @@ class OStatusPlugin extends Plugin
     function onCheckSchema() {
         $schema = Schema::get();
         $schema->ensureTable('ostatus_profile', Ostatus_profile::schemaDef());
+        $schema->ensureTable('feedsub', FeedSub::schemaDef());
         $schema->ensureTable('hubsub', HubSub::schemaDef());
         return true;
     }
@@ -345,4 +313,19 @@ class OStatusPlugin extends Plugin
             return false;
         }
     }
+
+    /**
+     * Send incoming PuSH feeds for OStatus endpoints in for processing.
+     *
+     * @param FeedSub $feedsub
+     * @param DOMDocument $feed
+     * @return mixed hook return code
+     */
+    function onStartFeedSubReceive($feedsub, $feed)
+    {
+        $oprofile = Ostatus_profile::staticGet('feeduri', $feedsub->uri);
+        if ($oprofile) {
+            $oprofile->processFeed($feed);
+        }
+    }
 }
diff --git a/plugins/OStatus/actions/feedsubsettings.php b/plugins/OStatus/actions/feedsubsettings.php
index 6933c9bf2..3e1d0aa82 100644
--- a/plugins/OStatus/actions/feedsubsettings.php
+++ b/plugins/OStatus/actions/feedsubsettings.php
@@ -26,7 +26,7 @@ if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
 
 class FeedSubSettingsAction extends ConnectSettingsAction
 {
-    protected $feedurl;
+    protected $profile_uri;
     protected $preview;
     protected $munger;
 
@@ -88,7 +88,10 @@ class FeedSubSettingsAction extends ConnectSettingsAction
 
         $this->elementStart('ul', 'form_data');
         $this->elementStart('li', array('id' => 'settings_twitter_login_button'));
-        $this->input('feedurl', _('Feed URL'), $this->feedurl, _('Enter the URL of a PubSubHubbub-enabled feed'));
+        $this->input('profile_uri',
+                     _m('Feed URL'),
+                     $this->profile_uri,
+                     _m('Enter the profile URL of a PubSubHubbub-enabled feed'));
         $this->elementEnd('li');
         $this->elementEnd('ul');
 
@@ -145,79 +148,55 @@ class FeedSubSettingsAction extends ConnectSettingsAction
      */
     function validateFeed()
     {
-        $feedurl = trim($this->arg('feedurl'));
+        $profile_uri = trim($this->arg('profile_uri'));
         
-        if ($feedurl == '') {
-            $this->showForm(_m('Empty feed URL!'));
+        if ($profile_uri == '') {
+            $this->showForm(_m('Empty remote profile URL!'));
             return;
         }
-        $this->feedurl = $feedurl;
+        $this->profile_uri = $profile_uri;
         
-        // Get the canonical feed URI and check it
+        // @fixme validate, normalize bla bla
         try {
-            $discover = new FeedDiscovery();
-            $uri = $discover->discoverFromURL($feedurl);
+            $oprofile = Ostatus_profile::ensureProfile($this->profile_uri);
+            $this->oprofile = $oprofile;
+            return true;
         } catch (FeedSubBadURLException $e) {
-            $this->showForm(_m('Invalid URL or could not reach server.'));
-            return false;
+            $err = _m('Invalid URL or could not reach server.');
         } catch (FeedSubBadResponseException $e) {
-            $this->showForm(_m('Cannot read feed; server returned error.'));
-            return false;
+            $err = _m('Cannot read feed; server returned error.');
         } catch (FeedSubEmptyException $e) {
-            $this->showForm(_m('Cannot read feed; server returned an empty page.'));
-            return false;
+            $err = _m('Cannot read feed; server returned an empty page.');
         } catch (FeedSubBadHTMLException $e) {
-            $this->showForm(_m('Bad HTML, could not find feed link.'));
-            return false;
+            $err = _m('Bad HTML, could not find feed link.');
         } catch (FeedSubNoFeedException $e) {
-            $this->showForm(_m('Could not find a feed linked from this URL.'));
-            return false;
+            $err = _m('Could not find a feed linked from this URL.');
         } catch (FeedSubUnrecognizedTypeException $e) {
-            $this->showForm(_m('Not a recognized feed type.'));
-            return false;
+            $err = _m('Not a recognized feed type.');
         } catch (FeedSubException $e) {
             // Any new ones we forgot about
-            $this->showForm(_m('Bad feed URL.'));
-            return false;
+            $err = sprintf(_m('Bad feed URL: %s %s'), get_class($e), $e->getMessage());
         }
-        
-        $this->munger = $discover->feedMunger();
-        $this->profile = $this->munger->ostatusProfile();
 
-        if ($this->profile->huburi == '' && !common_config('feedsub', 'nohub')) {
-            $this->showForm(_m('Feed is not PuSH-enabled; cannot subscribe.'));
-            return false;
-        }
-        
-        return true;
+        $this->showForm($err);
+        return false;
     }
 
     function saveFeed()
     {
         if ($this->validateFeed()) {
             $this->preview = true;
-            $this->profile = Ostatus_profile::ensureProfile($this->munger);
-            if (!$this->profile) {
-                throw new ServerException("Feed profile was not saved properly.");
-            }
-
-            // If not already in use, subscribe to updates via the hub
-            if ($this->profile->sub_start) {
-                common_log(LOG_INFO, __METHOD__ . ": double the fun! new sub for {$this->profile->feeduri} last subbed {$this->profile->sub_start}");
-            } else {
-                $ok = $this->profile->subscribe();
-                common_log(LOG_INFO, __METHOD__ . ": sub was $ok");
-                if (!$ok) {
-                    $this->showForm(_m('Feed subscription failed! Bad response from hub.'));
-                    return;
-                }
-            }
 
             // And subscribe the current user to the local profile
             $user = common_current_user();
 
-            if ($this->profile->isGroup()) {
-                $group = $this->profile->localGroup();
+            if (!$this->oprofile->subscribe()) {
+                $this->showForm(_m("Failed to set up server-to-server subscription."));
+                return;
+            }
+
+            if ($this->oprofile->isGroup()) {
+                $group = $this->oprofile->localGroup();
                 if ($user->isMember($group)) {
                     $this->showForm(_m('Already a member!'));
                 } elseif (Group_member::join($this->profile->group_id, $user->id)) {
@@ -226,13 +205,13 @@ class FeedSubSettingsAction extends ConnectSettingsAction
                     $this->showForm(_m('Remote group join failed!'));
                 }
             } else {
-                $local = $this->profile->localProfile();
+                $local = $this->oprofile->localProfile();
                 if ($user->isSubscribed($local)) {
                     $this->showForm(_m('Already subscribed!'));
-                } elseif ($user->subscribeTo($local)) {
-                    $this->showForm(_m('Feed subscribed!'));
+                } elseif ($this->oprofile->subscribeLocalToRemote($user)) {
+                    $this->showForm(_m('Remote user subscribed!'));
                 } else {
-                    $this->showForm(_m('Feed subscription failed!'));
+                    $this->showForm(_m('Remote subscription failed!'));
                 }
             }
         }
@@ -248,17 +227,7 @@ class FeedSubSettingsAction extends ConnectSettingsAction
 
     function previewFeed()
     {
-        $profile = $this->munger->ostatusProfile();
-        $notice = $this->munger->notice(0, true); // preview
-
-        if ($notice) {
-            $this->element('b', null, 'Preview of latest post from this feed:');
-
-            $item = new NoticeList($notice, $this);
-            $item->show();
-        } else {
-            $this->element('b', null, 'No posts in this feed yet.');
-        }
+        $this->text('Profile preview should go here');
     }
 
     function showScripts()
diff --git a/plugins/OStatus/actions/pushcallback.php b/plugins/OStatus/actions/pushcallback.php
index ed859a32f..7e1227a66 100644
--- a/plugins/OStatus/actions/pushcallback.php
+++ b/plugins/OStatus/actions/pushcallback.php
@@ -48,9 +48,9 @@ class PushCallbackAction extends Action
             throw new ServerException('Empty or invalid feed id', 400);
         }
 
-        $profile = Ostatus_profile::staticGet('id', $feedid);
-        if (!$profile) {
-            throw new ServerException('Unknown OStatus/PuSH feed id ' . $feedid, 400);
+        $feedsub = FeedSub::staticGet('id', $feedid);
+        if (!$feedsub) {
+            throw new ServerException('Unknown PuSH feed id ' . $feedid, 400);
         }
 
         $hmac = '';
@@ -62,7 +62,7 @@ class PushCallbackAction extends Action
 
         // @fixme Queue this to a background process; we should return
         // as quickly as possible from a distribution POST.
-        $profile->postUpdates($post, $hmac);
+        $feedsub->receive($post, $hmac);
     }
     
     /**
@@ -81,29 +81,29 @@ class PushCallbackAction extends Action
             throw new ServerException("Bogus hub callback: bad mode", 404);
         }
         
-        $profile = Ostatus_profile::staticGet('feeduri', $topic);
-        if (!$profile) {
+        $feedsub = FeedSub::staticGet('uri', $topic);
+        if (!$feedsub) {
             common_log(LOG_WARNING, __METHOD__ . ": bogus hub callback for unknown feed $topic");
             throw new ServerException("Bogus hub callback: unknown feed", 404);
         }
 
-        if ($profile->verify_token !== $verify_token) {
+        if ($feedsub->verify_token !== $verify_token) {
             common_log(LOG_WARNING, __METHOD__ . ": bogus hub callback with bad token \"$verify_token\" for feed $topic");
             throw new ServerException("Bogus hub callback: bad token", 404);
         }
 
-        if ($mode != $profile->sub_state) {
-            common_log(LOG_WARNING, __METHOD__ . ": bogus hub callback with bad mode \"$mode\" for feed $topic in state \"{$profile->sub_state}\"");
+        if ($mode != $feedsub->sub_state) {
+            common_log(LOG_WARNING, __METHOD__ . ": bogus hub callback with bad mode \"$mode\" for feed $topic in state \"{$feedsub->sub_state}\"");
             throw new ServerException("Bogus hub callback: mode doesn't match subscription state.", 404);
         }
 
         // OK!
         if ($mode == 'subscribe') {
             common_log(LOG_INFO, __METHOD__ . ': sub confirmed');
-            $profile->confirmSubscribe($lease_seconds);
+            $feedsub->confirmSubscribe($lease_seconds);
         } else {
             common_log(LOG_INFO, __METHOD__ . ": unsub confirmed; deleting sub record for $topic");
-            $profile->confirmUnsubscribe();
+            $feedsub->confirmUnsubscribe();
         }
         print $challenge;
     }
diff --git a/plugins/OStatus/actions/salmon.php b/plugins/OStatus/actions/salmon.php
index 224134cd7..ea5b8e4ea 100644
--- a/plugins/OStatus/actions/salmon.php
+++ b/plugins/OStatus/actions/salmon.php
@@ -68,6 +68,9 @@ class SalmonAction extends Action
         return true;
     }
 
+    /**
+     * @fixme probably call Ostatus_profile::processFeed
+     */
     function handle($args)
     {
         common_log(LOG_INFO, 'Salmon: incoming post for user '. $this->user->id);
@@ -95,6 +98,9 @@ class SalmonAction extends Action
         }
     }
 
+    /**
+     * @fixme probably call Ostatus_profile::processFeed
+     */
     function handlePost()
     {
         switch ($this->act->object->type) {
@@ -111,14 +117,23 @@ class SalmonAction extends Action
         $profile = $this->ensureProfile();
     }
 
+    /**
+     * @fixme probably call Ostatus_profile::processFeed
+     */
     function handleFollow()
     {
     }
 
+    /**
+     * @fixme probably call Ostatus_profile::processFeed
+     */
     function handleFavorite()
     {
     }
 
+    /**
+     * @fixme probably call Ostatus_profile::processFeed
+     */
     function handleShare()
     {
     }
@@ -131,17 +146,13 @@ class SalmonAction extends Action
             throw new Exception("Received a salmon slap from unidentified actor.");
         }
 
-        $ostatusProfile = Ostatus_profile::staticGet('homeuri', $actor->id);
-
-        if (empty($ostatusProfile)) {
-            return $this->createProfile();
-        } else {
-            // XXX: can we receive a salmon slap from a group...?
-            assert(!empty($ostatusProfile->profile_id));
-            return Profile::staticGet($ostatusProfile->profile_id);
-        }
+        $ostatusProfile = Ostatus_profile::ensureActorProfile($this->act);
+        return $oprofile->localProfile();
     }
 
+    /**
+     * @fixme anything new in here probably should be merged into Ostatus_profile::ensureActorProfile and friends
+     */
     function createProfile()
     {
         $actor = $this->act->actor;
@@ -186,6 +197,9 @@ class SalmonAction extends Action
         return $profile;
     }
 
+    /**
+     * @fixme should be merged into Ostatus_profile
+     */
     function nicknameFromURI($uri)
     {
         preg_match('/(\w+):/', $uri, $matches);
diff --git a/plugins/OStatus/classes/FeedSub.php b/plugins/OStatus/classes/FeedSub.php
new file mode 100644
index 000000000..dc2c0b710
--- /dev/null
+++ b/plugins/OStatus/classes/FeedSub.php
@@ -0,0 +1,443 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 2009-2010, StatusNet, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+/**
+ * @package OStatusPlugin
+ * @maintainer Brion Vibber <brion@status.net>
+ */
+
+/*
+PuSH subscription flow:
+
+    $profile->subscribe()
+        generate random verification token
+            save to verify_token
+        sends a sub request to the hub...
+
+    main/push/callback
+        hub sends confirmation back to us via GET
+        We verify the request, then echo back the challenge.
+        On our end, we save the time we subscribed and the lease expiration
+
+    main/push/callback
+        hub sends us updates via POST
+
+*/
+
+class FeedDBException extends FeedSubException
+{
+    public $obj;
+
+    function __construct($obj)
+    {
+        parent::__construct('Database insert failure');
+        $this->obj = $obj;
+    }
+}
+
+/**
+ * FeedSub handles low-level PubHubSubbub (PuSH) subscriptions.
+ * Higher-level behavior building OStatus stuff on top is handled
+ * under Ostatus_profile.
+ */
+class FeedSub extends Memcached_DataObject
+{
+    public $__table = 'feedsub';
+
+    public $id;
+    public $feeduri;
+
+    // PuSH subscription data
+    public $huburi;
+    public $secret;
+    public $verify_token;
+    public $sub_state; // subscribe, active, unsubscribe, inactive
+    public $sub_start;
+    public $sub_end;
+    public $last_update;
+
+    public $created;
+    public $modified;
+
+    public /*static*/ function staticGet($k, $v=null)
+    {
+        return parent::staticGet(__CLASS__, $k, $v);
+    }
+
+    /**
+     * return table definition for DB_DataObject
+     *
+     * DB_DataObject needs to know something about the table to manipulate
+     * instances. This method provides all the DB_DataObject needs to know.
+     *
+     * @return array array of column definitions
+     */
+
+    function table()
+    {
+        return array('id' => DB_DATAOBJECT_INT + DB_DATAOBJECT_NOTNULL,
+                     'uri' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
+                     'huburi' =>  DB_DATAOBJECT_STR,
+                     'secret' => DB_DATAOBJECT_STR,
+                     'verify_token' => DB_DATAOBJECT_STR,
+                     'sub_state' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
+                     'sub_start' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
+                     'sub_end' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
+                     'last_update' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL,
+                     'created' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL,
+                     'modified' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL);
+    }
+
+    static function schemaDef()
+    {
+        return array(new ColumnDef('id', 'integer',
+                                   /*size*/ null,
+                                   /*nullable*/ false,
+                                   /*key*/ 'PRI',
+                                   /*default*/ '0',
+                                   /*extra*/ null,
+                                   /*auto_increment*/ true),
+                     new ColumnDef('uri', 'varchar',
+                                   255, false, 'UNI'),
+                     new ColumnDef('huburi', 'text',
+                                   null, true),
+                     new ColumnDef('verify_token', 'text',
+                                   null, true),
+                     new ColumnDef('secret', 'text',
+                                   null, true),
+                     new ColumnDef('sub_state', "enum('subscribe','active','unsubscribe','inactive')",
+                                   null, false),
+                     new ColumnDef('sub_start', 'datetime',
+                                   null, true),
+                     new ColumnDef('sub_end', 'datetime',
+                                   null, true),
+                     new ColumnDef('last_update', 'datetime',
+                                   null, false),
+                     new ColumnDef('created', 'datetime',
+                                   null, false),
+                     new ColumnDef('modified', 'datetime',
+                                   null, false));
+    }
+
+    /**
+     * return key definitions for DB_DataObject
+     *
+     * DB_DataObject needs to know about keys that the table has; this function
+     * defines them.
+     *
+     * @return array key definitions
+     */
+
+    function keys()
+    {
+        return array_keys($this->keyTypes());
+    }
+
+    /**
+     * return key definitions for Memcached_DataObject
+     *
+     * Our caching system uses the same key definitions, but uses a different
+     * method to get them.
+     *
+     * @return array key definitions
+     */
+
+    function keyTypes()
+    {
+        return array('id' => 'K', 'uri' => 'U');
+    }
+
+    function sequenceKey()
+    {
+        return array('id', true, false);
+    }
+
+    /**
+     * Fetch the StatusNet-side profile for this feed
+     * @return Profile
+     */
+    public function localProfile()
+    {
+        if ($this->profile_id) {
+            return Profile::staticGet('id', $this->profile_id);
+        }
+        return null;
+    }
+
+    /**
+     * Fetch the StatusNet-side profile for this feed
+     * @return Profile
+     */
+    public function localGroup()
+    {
+        if ($this->group_id) {
+            return User_group::staticGet('id', $this->group_id);
+        }
+        return null;
+    }
+
+    /**
+     * @param string $feeduri
+     * @return FeedSub
+     * @throws FeedSubException if feed is invalid or lacks PuSH setup
+     */
+    public static function ensureFeed($feeduri)
+    {
+        $current = self::staticGet('uri', $feeduri);
+        if ($current) {
+            return $current;
+        }
+
+        $discover = new FeedDiscovery();
+        $discover->discoverFromFeedURL($feeduri);
+
+        $huburi = $discover->getAtomLink('hub');
+        if (!$huburi) {
+            throw new FeedSubNoHubException();
+        }
+
+        $feedsub = new FeedSub();
+        $feedsub->uri = $feeduri;
+        $feedsub->huburi = $huburi;
+        $feedsub->sub_state = 'inactive';
+
+        $feedsub->created = common_sql_now();
+        $feedsub->modified = common_sql_now();
+
+        $result = $feedsub->insert();
+        if (empty($result)) {
+            throw new FeedDBException($feedsub);
+        }
+
+        return $feedsub;
+    }
+
+    /**
+     * Send a subscription request to the hub for this feed.
+     * The hub will later send us a confirmation POST to /main/push/callback.
+     *
+     * @return bool true on success, false on failure
+     * @throws ServerException if feed state is not valid
+     */
+    public function subscribe($mode='subscribe')
+    {
+        if ($this->sub_state && $this->sub_state != 'inactive') {
+            throw new ServerException("Attempting to start PuSH subscription to feed in state $this->sub_state");
+        }
+        if (empty($this->huburi)) {
+            if (common_config('feedsub', 'nohub')) {
+                // Fake it! We're just testing remote feeds w/o hubs.
+                return true;
+            } else {
+                throw new ServerException("Attempting to start PuSH subscription for feed with no hub");
+            }
+        }
+
+        return $this->doSubscribe('subscribe');
+    }
+
+    /**
+     * Send a PuSH unsubscription request to the hub for this feed.
+     * The hub will later send us a confirmation POST to /main/push/callback.
+     *
+     * @return bool true on success, false on failure
+     * @throws ServerException if feed state is not valid
+     */
+    public function unsubscribe() {
+        if ($this->sub_state != 'active') {
+            throw new ServerException("Attempting to end PuSH subscription to feed in state $this->sub_state");
+        }
+        if (empty($this->huburi)) {
+            if (common_config('feedsub', 'nohub')) {
+                // Fake it! We're just testing remote feeds w/o hubs.
+                return true;
+            } else {
+                throw new ServerException("Attempting to end PuSH subscription for feed with no hub");
+            }
+        }
+
+        return $this->doSubscribe('unsubscribe');
+    }
+
+    protected function doSubscribe($mode)
+    {
+        $orig = clone($this);
+        $this->verify_token = common_good_rand(16);
+        if ($mode == 'subscribe') {
+            $this->secret = common_good_rand(32);
+        }
+        $this->sub_state = $mode;
+        $this->update($orig);
+        unset($orig);
+
+        try {
+            $callback = common_local_url('pushcallback', array('feed' => $this->id));
+            $headers = array('Content-Type: application/x-www-form-urlencoded');
+            $post = array('hub.mode' => $mode,
+                          'hub.callback' => $callback,
+                          'hub.verify' => 'async',
+                          'hub.verify_token' => $this->verify_token,
+                          'hub.secret' => $this->secret,
+                          //'hub.lease_seconds' => 0,
+                          'hub.topic' => $this->uri);
+            $client = new HTTPClient();
+            $response = $client->post($this->huburi, $headers, $post);
+            $status = $response->getStatus();
+            if ($status == 202) {
+                common_log(LOG_INFO, __METHOD__ . ': sub req ok, awaiting verification callback');
+                return true;
+            } else if ($status == 204) {
+                common_log(LOG_INFO, __METHOD__ . ': sub req ok and verified');
+                return true;
+            } else if ($status >= 200 && $status < 300) {
+                common_log(LOG_ERR, __METHOD__ . ": sub req returned unexpected HTTP $status: " . $response->getBody());
+                return false;
+            } else {
+                common_log(LOG_ERR, __METHOD__ . ": sub req failed with HTTP $status: " . $response->getBody());
+                return false;
+            }
+        } catch (Exception $e) {
+            // wtf!
+            common_log(LOG_ERR, __METHOD__ . ": error \"{$e->getMessage()}\" hitting hub $this->huburi subscribing to $this->uri");
+
+            $orig = clone($this);
+            $this->verify_token = null;
+            $this->sub_state = null;
+            $this->update($orig);
+            unset($orig);
+
+            return false;
+        }
+    }
+
+    /**
+     * Save PuSH subscription confirmation.
+     * Sets approximate lease start and end times and finalizes state.
+     *
+     * @param int $lease_seconds provided hub.lease_seconds parameter, if given
+     */
+    public function confirmSubscribe($lease_seconds=0)
+    {
+        $original = clone($this);
+
+        $this->sub_state = 'active';
+        $this->sub_start = common_sql_date(time());
+        if ($lease_seconds > 0) {
+            $this->sub_end = common_sql_date(time() + $lease_seconds);
+        } else {
+            $this->sub_end = null;
+        }
+        $this->lastupdate = common_sql_now();
+
+        return $this->update($original);
+    }
+
+    /**
+     * Save PuSH unsubscription confirmation.
+     * Wipes active PuSH sub info and resets state.
+     */
+    public function confirmUnsubscribe()
+    {
+        $original = clone($this);
+
+        // @fixme these should all be null, but DB_DataObject doesn't save null values...?????
+        $this->verify_token = '';
+        $this->secret = '';
+        $this->sub_state = '';
+        $this->sub_start = '';
+        $this->sub_end = '';
+        $this->lastupdate = common_sql_now();
+
+        return $this->update($original);
+    }
+
+    /**
+     * Accept updates from a PuSH feed. If validated, this object and the
+     * feed (as a DOMDocument) will be passed to the StartFeedSubHandleFeed
+     * and EndFeedSubHandleFeed events for processing.
+     *
+     * @param string $post source of Atom or RSS feed
+     * @param string $hmac X-Hub-Signature header, if present
+     */
+    public function receive($post, $hmac)
+    {
+        common_log(LOG_INFO, __METHOD__ . ": packet for \"$this->uri\"! $hmac $post");
+
+        if ($this->sub_state != 'active') {
+            common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH for inactive feed $this->uri (in state '$this->sub_state')");
+            return;
+        }
+
+        if ($post === '') {
+            common_log(LOG_ERR, __METHOD__ . ": ignoring empty post");
+            return;
+        }
+
+        if (!$this->validatePushSig($post, $hmac)) {
+            // Per spec we silently drop input with a bad sig,
+            // while reporting receipt to the server.
+            return;
+        }
+
+        $feed = new DOMDocument();
+        if (!$feed->loadXML($post)) {
+            // @fixme might help to include the err message
+            common_log(LOG_ERR, __METHOD__ . ": ignoring invalid XML");
+            return;
+        }
+
+        Event::handle('StartFeedSubReceive', array($this, $feed));
+        Event::handle('EndFeedSubReceive', array($this, $feed));
+    }
+
+    /**
+     * Validate the given Atom chunk and HMAC signature against our
+     * shared secret that was set up at subscription time.
+     *
+     * If we don't have a shared secret, there should be no signature.
+     * If we we do, our the calculated HMAC should match theirs.
+     *
+     * @param string $post raw XML source as POSTed to us
+     * @param string $hmac X-Hub-Signature HTTP header value, or empty
+     * @return boolean true for a match
+     */
+    protected function validatePushSig($post, $hmac)
+    {
+        if ($this->secret) {
+            if (preg_match('/^sha1=([0-9a-fA-F]{40})$/', $hmac, $matches)) {
+                $their_hmac = strtolower($matches[1]);
+                $our_hmac = hash_hmac('sha1', $post, $this->secret);
+                if ($their_hmac === $our_hmac) {
+                    return true;
+                }
+                common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bad SHA-1 HMAC: got $their_hmac, expected $our_hmac");
+            } else {
+                common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bogus HMAC '$hmac'");
+            }
+        } else {
+            if (empty($hmac)) {
+                return true;
+            } else {
+                common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with unexpected HMAC '$hmac'");
+            }
+        }
+        return false;
+    }
+
+}
diff --git a/plugins/OStatus/classes/Ostatus_profile.php b/plugins/OStatus/classes/Ostatus_profile.php
index 486417617..1ce8ac491 100644
--- a/plugins/OStatus/classes/Ostatus_profile.php
+++ b/plugins/OStatus/classes/Ostatus_profile.php
@@ -18,62 +18,24 @@
  */
 
 /**
- * @package FeedSubPlugin
+ * @package OStatusPlugin
  * @maintainer Brion Vibber <brion@status.net>
  */
 
-/*
-PuSH subscription flow:
-
-    $profile->subscribe()
-        generate random verification token
-            save to verify_token
-        sends a sub request to the hub...
-
-    main/push/callback
-        hub sends confirmation back to us via GET
-        We verify the request, then echo back the challenge.
-        On our end, we save the time we subscribed and the lease expiration
-
-    main/push/callback
-        hub sends us updates via POST
-
-*/
-
-class FeedDBException extends FeedSubException
-{
-    public $obj;
-
-    function __construct($obj)
-    {
-        parent::__construct('Database insert failure');
-        $this->obj = $obj;
-    }
-}
-
 class Ostatus_profile extends Memcached_DataObject
 {
     public $__table = 'ostatus_profile';
 
-    public $id;
+    public $uri;
+
     public $profile_id;
     public $group_id;
 
     public $feeduri;
-    public $homeuri;
-
-    // PuSH subscription data
-    public $huburi;
-    public $secret;
-    public $verify_token;
-    public $sub_state; // subscribe, active, unsubscribe
-    public $sub_start;
-    public $sub_end;
-
     public $salmonuri;
 
     public $created;
-    public $lastupdate;
+    public $modified;
 
     public /*static*/ function staticGet($k, $v=null)
     {
@@ -91,56 +53,30 @@ class Ostatus_profile extends Memcached_DataObject
 
     function table()
     {
-        return array('id' => DB_DATAOBJECT_INT + DB_DATAOBJECT_NOTNULL,
+        return array('uri' => DB_DATAOBJECT_STR + 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,
-                     'secret' => DB_DATAOBJECT_STR,
-                     'verify_token' => DB_DATAOBJECT_STR,
-                     'sub_state' => DB_DATAOBJECT_STR,
-                     'sub_start' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
-                     'sub_end' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
                      'salmonuri' =>  DB_DATAOBJECT_STR,
                      'created' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL,
-                     'lastupdate' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL);
+                     'modified' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL);
     }
 
     static function schemaDef()
     {
-        return array(new ColumnDef('id', 'integer',
-                                   /*size*/ null,
-                                   /*nullable*/ false,
-                                   /*key*/ 'PRI',
-                                   /*default*/ '0',
-                                   /*extra*/ null,
-                                   /*auto_increment*/ true),
+        return array(new ColumnDef('uri', 'varchar',
+                                   255, false, 'PRI'),
                      new ColumnDef('profile_id', 'integer',
                                    null, true, 'UNI'),
                      new ColumnDef('group_id', 'integer',
                                    null, true, 'UNI'),
                      new ColumnDef('feeduri', 'varchar',
                                    255, false, 'UNI'),
-                     new ColumnDef('homeuri', 'varchar',
-                                   255, false),
-                     new ColumnDef('huburi', 'text',
-                                   null, true),
-                     new ColumnDef('verify_token', 'varchar',
-                                   32, true),
-                     new ColumnDef('secret', 'varchar',
-                                   64, true),
-                     new ColumnDef('sub_state', "enum('subscribe','active','unsubscribe')",
-                                   null, true),
-                     new ColumnDef('sub_start', 'datetime',
-                                   null, true),
-                     new ColumnDef('sub_end', 'datetime',
-                                   null, true),
                      new ColumnDef('salmonuri', 'text',
                                    null, true),
                      new ColumnDef('created', 'datetime',
                                    null, false),
-                     new ColumnDef('lastupdate', 'datetime',
+                     new ColumnDef('modified', 'datetime',
                                    null, false));
     }
 
@@ -169,12 +105,12 @@ class Ostatus_profile extends Memcached_DataObject
 
     function keyTypes()
     {
-        return array('id' => 'K', 'profile_id' => 'U', 'group_id' => 'U', 'feeduri' => 'U');
+        return array('uri' => 'K', 'profile_id' => 'U', 'group_id' => 'U', 'feeduri' => 'U');
     }
 
     function sequenceKey()
     {
-        return array('id', true, false);
+        return array(false, false, false);
     }
 
     /**
@@ -202,101 +138,6 @@ class Ostatus_profile extends Memcached_DataObject
     }
 
     /**
-     * @param FeedMunger $munger
-     * @param boolean $isGroup is this a group record?
-     * @return Ostatus_profile
-     */
-    public static function ensureProfile($munger)
-    {
-        $profile = $munger->ostatusProfile();
-
-        $current = self::staticGet('feeduri', $profile->feeduri);
-        if ($current) {
-            // @fixme we should probably update info as necessary
-            return $current;
-        }
-
-        $profile->query('BEGIN');
-
-        try {
-            $local = $munger->profile();
-
-            if ($profile->isGroup()) {
-                $group = new User_group();
-                $group->nickname = $local->nickname . '@remote'; // @fixme
-                $group->fullname = $local->fullname;
-                $group->homepage = $local->homepage;
-                $group->location = $local->location;
-                $group->created = $local->created;
-                $group->insert();
-                if (empty($result)) {
-                    throw new FeedDBException($group);
-                }
-                $profile->group_id = $group->id;
-            } else {
-                $result = $local->insert();
-                if (empty($result)) {
-                    throw new FeedDBException($local);
-                }
-                $profile->profile_id = $local->id;
-            }
-
-            $profile->created = common_sql_now();
-            $profile->lastupdate = common_sql_now();
-            $result = $profile->insert();
-            if (empty($result)) {
-                throw new FeedDBException($profile);
-            }
-
-            $profile->query('COMMIT');
-        } catch (FeedDBException $e) {
-            common_log_db_error($e->obj, 'INSERT', __FILE__);
-            $profile->query('ROLLBACK');
-            return false;
-        }
-
-        $avatar = $munger->getAvatar();
-        if ($avatar) {
-            try {
-                $profile->updateAvatar($avatar);
-            } catch (Exception $e) {
-                common_log(LOG_ERR, "Exception setting OStatus avatar: " .
-                                    $e->getMessage());
-            }
-        }
-
-        return $profile;
-    }
-
-    /**
-     * Download and update given avatar image
-     * @param string $url
-     * @throws Exception in various failure cases
-     */
-    public function updateAvatar($url)
-    {
-        // @fixme this should be better encapsulated
-        // ripped from oauthstore.php (for old OMB client)
-        $temp_filename = tempnam(sys_get_temp_dir(), 'listener_avatar');
-        copy($url, $temp_filename);
-        
-        // @fixme should we be using different ids?
-        $imagefile = new ImageFile($this->id, $temp_filename);
-        $filename = Avatar::filename($this->id,
-                                     image_type_to_extension($imagefile->type),
-                                     null,
-                                     common_timestamp());
-        rename($temp_filename, Avatar::path($filename));
-        if ($this->isGroup()) {
-            $group = $this->localGroup();
-            $group->setOriginal($filename);
-        } else {
-            $profile = $this->localProfile();
-            $profile->setOriginal($filename);
-        }
-    }
-
-    /**
      * Returns an XML string fragment with profile information as an
      * Activity Streams noun object with the given element type.
      *
@@ -345,7 +186,7 @@ class Ostatus_profile extends Memcached_DataObject
         $xs->element(
             'id',
             null,
-            $this->homeuri); // ?
+            $this->uri); // ?
         $xs->element('title', null, $self->getBestName());
 
         $xs->element(
@@ -370,142 +211,95 @@ class Ostatus_profile extends Memcached_DataObject
     }
 
     /**
-     * Send a subscription request to the hub for this feed.
-     * The hub will later send us a confirmation POST to /main/push/callback.
+     * Subscribe a local user to this remote user.
+     * PuSH subscription will be started if necessary, and we'll
+     * send a Salmon notification to the remote server if available
+     * notifying them of the sub.
      *
-     * @return bool true on success, false on failure
-     * @throws ServerException if feed state is not valid
+     * @param User $user
+     * @return boolean success
+     * @throws FeedException
      */
-    public function subscribe($mode='subscribe')
+    public function subscribeLocalToRemote(User $user)
     {
-        if ($this->sub_state != '') {
-            throw new ServerException("Attempting to start PuSH subscription to feed in state $this->sub_state");
+        if ($this->isGroup()) {
+            throw new ServerException("Can't subscribe to a remote group");
         }
-        if (empty($this->huburi)) {
-            if (common_config('feedsub', 'nohub')) {
-                // Fake it! We're just testing remote feeds w/o hubs.
+
+        if ($this->subscribe()) {
+            if ($user->subscribeTo($this->localProfile())) {
+                $this->notify($user->getProfile(), ActivityVerb::FOLLOW, $this);
                 return true;
-            } else {
-                throw new ServerException("Attempting to start PuSH subscription for feed with no hub");
             }
         }
-
-        return $this->doSubscribe('subscribe');
+        return false;
     }
 
     /**
-     * Send a PuSH unsubscription request to the hub for this feed.
-     * The hub will later send us a confirmation POST to /main/push/callback.
+     * Mark this remote profile as subscribing to the given local user,
+     * and send appropriate notifications to the user.
      *
-     * @return bool true on success, false on failure
-     * @throws ServerException if feed state is not valid
+     * This will generally be in response to a subscription notification
+     * from a foreign site to our local Salmon response channel.
+     *
+     * @param User $user
+     * @return boolean success
      */
-    public function unsubscribe() {
-        if ($this->sub_state != 'active') {
-            throw new ServerException("Attempting to end PuSH subscription to feed in state $this->sub_state");
-        }
-        if (empty($this->huburi)) {
-            if (common_config('feedsub', 'nohub')) {
-                // Fake it! We're just testing remote feeds w/o hubs.
-                return true;
-            } else {
-                throw new ServerException("Attempting to end PuSH subscription for feed with no hub");
-            }
-        }
-
-        return $this->doSubscribe('unsubscribe');
-    }
-
-    protected function doSubscribe($mode)
+    public function subscribeRemoteToLocal(User $user)
     {
-        $orig = clone($this);
-        $this->verify_token = common_good_rand(16);
-        if ($mode == 'subscribe') {
-            $this->secret = common_good_rand(32);
-        }
-        $this->sub_state = $mode;
-        $this->update($orig);
-        unset($orig);
-
-        try {
-            $callback = common_local_url('pushcallback', array('feed' => $this->id));
-            $headers = array('Content-Type: application/x-www-form-urlencoded');
-            $post = array('hub.mode' => $mode,
-                          'hub.callback' => $callback,
-                          'hub.verify' => 'async',
-                          'hub.verify_token' => $this->verify_token,
-                          'hub.secret' => $this->secret,
-                          //'hub.lease_seconds' => 0,
-                          'hub.topic' => $this->feeduri);
-            $client = new HTTPClient();
-            $response = $client->post($this->huburi, $headers, $post);
-            $status = $response->getStatus();
-            if ($status == 202) {
-                common_log(LOG_INFO, __METHOD__ . ': sub req ok, awaiting verification callback');
-                return true;
-            } else if ($status == 204) {
-                common_log(LOG_INFO, __METHOD__ . ': sub req ok and verified');
-                return true;
-            } else if ($status >= 200 && $status < 300) {
-                common_log(LOG_ERR, __METHOD__ . ": sub req returned unexpected HTTP $status: " . $response->getBody());
-                return false;
-            } else {
-                common_log(LOG_ERR, __METHOD__ . ": sub req failed with HTTP $status: " . $response->getBody());
-                return false;
-            }
-        } catch (Exception $e) {
-            // wtf!
-            common_log(LOG_ERR, __METHOD__ . ": error \"{$e->getMessage()}\" hitting hub $this->huburi subscribing to $this->feeduri");
+        if ($this->isGroup()) {
+            throw new ServerException("Remote groups can't subscribe to local users");
+        }
 
-            $orig = clone($this);
-            $this->verify_token = null;
-            $this->sub_state = null;
-            $this->update($orig);
-            unset($orig);
+        // @fixme use regular channels for subbing, once they accept remote profiles
+        $sub = new Subscription();
+        $sub->subscriber = $this->profile_id;
+        $sub->subscribed = $user->id;
+        $sub->created = common_sql_now(); // current time
 
-            return false;
+        if ($sub->insert()) {
+            // @fixme use subs_notify() if refactored to take profiles?
+            mail_subscribe_notify_profile($user, $this->localProfile());
+            return true;
         }
+        return false;
     }
 
     /**
-     * Save PuSH subscription confirmation.
-     * Sets approximate lease start and end times and finalizes state.
+     * Send a subscription request to the hub for this feed.
+     * The hub will later send us a confirmation POST to /main/push/callback.
      *
-     * @param int $lease_seconds provided hub.lease_seconds parameter, if given
+     * @return bool true on success, false on failure
+     * @throws ServerException if feed state is not valid
      */
-    public function confirmSubscribe($lease_seconds=0)
+    public function subscribe($mode='subscribe')
     {
-        $original = clone($this);
-
-        $this->sub_state = 'active';
-        $this->sub_start = common_sql_date(time());
-        if ($lease_seconds > 0) {
-            $this->sub_end = common_sql_date(time() + $lease_seconds);
-        } else {
-            $this->sub_end = null;
+        $feedsub = FeedSub::ensureFeed($this->feeduri);
+        if ($feedsub->sub_state == 'active' || $feedsub->sub_state == 'subscribe') {
+            return true;
+        } else if ($feedsub->sub_state == '' || $feedsub->sub_state == 'inactive') {
+            return $feedsub->subscribe();
+        } else if ('unsubscribe') {
+            throw new FeedSubException("Unsub is pending, can't subscribe...");
         }
-        $this->lastupdate = common_sql_now();
-
-        return $this->update($original);
     }
 
     /**
-     * Save PuSH unsubscription confirmation.
-     * Wipes active PuSH sub info and resets state.
+     * Send a PuSH unsubscription request to the hub for this feed.
+     * The hub will later send us a confirmation POST to /main/push/callback.
+     *
+     * @return bool true on success, false on failure
+     * @throws ServerException if feed state is not valid
      */
-    public function confirmUnsubscribe()
-    {
-        $original = clone($this);
-
-        // @fixme these should all be null, but DB_DataObject doesn't save null values...?????
-        $this->verify_token = '';
-        $this->secret = '';
-        $this->sub_state = '';
-        $this->sub_start = '';
-        $this->sub_end = '';
-        $this->lastupdate = common_sql_now();
-
-        return $this->update($original);
+    public function unsubscribe() {
+        $feedsub = FeedSub::staticGet('uri', $this->feeduri);
+        if ($feedsub->sub_state == 'active') {
+            return $feedsub->unsubscribe();
+        } else if ($feedsub->sub_state == '' || $feedsub->sub_state == 'inactive' || $feedsub->sub_state == 'unsubscribe') {
+            return true;
+        } else if ($feedsub->sub_state == 'subscribe') {
+            throw new FeedSubException("Feed is awaiting subscription, can't unsub...");
+        }
     }
 
     /**
@@ -543,10 +337,7 @@ class Ostatus_profile extends Memcached_DataObject
             $entry->elementEnd('entry');
 
             $feed = $this->atomFeed($actor);
-            #$feed->initFeed();
             $feed->addEntry($entry);
-            #$feed->renderEntries();
-            #$feed->endFeed();
 
             $xml = $feed->getString();
             common_log(LOG_INFO, "Posting to Salmon endpoint $salmon: $xml");
@@ -600,36 +391,10 @@ class Ostatus_profile extends Memcached_DataObject
      * Currently assumes that all items in the feed are new,
      * coming from a PuSH hub.
      *
-     * @param string $post source of Atom or RSS feed
-     * @param string $hmac X-Hub-Signature header, if present
+     * @param DOMDocument $feed
      */
-    public function postUpdates($post, $hmac)
+    public function processFeed($feed)
     {
-        common_log(LOG_INFO, __METHOD__ . ": packet for \"$this->feeduri\"! $hmac $post");
-
-        if ($this->sub_state != 'active') {
-            common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH for inactive feed $this->feeduri (in state '$this->sub_state')");
-            return;
-        }
-
-        if ($post === '') {
-            common_log(LOG_ERR, __METHOD__ . ": ignoring empty post");
-            return;
-        }
-
-        if (!$this->validatePushSig($post, $hmac)) {
-            // Per spec we silently drop input with a bad sig,
-            // while reporting receipt to the server.
-            return;
-        }
-
-        $feed = new DOMDocument();
-        if (!$feed->loadXML($post)) {
-            // @fixme might help to include the err message
-            common_log(LOG_ERR, __METHOD__ . ": ignoring invalid XML");
-            return;
-        }
-
         $entries = $feed->getElementsByTagNameNS(Activity::ATOM, 'entry');
         if ($entries->length == 0) {
             common_log(LOG_ERR, __METHOD__ . ": no entries in feed update, ignoring");
@@ -643,40 +408,6 @@ class Ostatus_profile extends Memcached_DataObject
     }
 
     /**
-     * Validate the given Atom chunk and HMAC signature against our
-     * shared secret that was set up at subscription time.
-     *
-     * If we don't have a shared secret, there should be no signature.
-     * If we we do, our the calculated HMAC should match theirs.
-     *
-     * @param string $post raw XML source as POSTed to us
-     * @param string $hmac X-Hub-Signature HTTP header value, or empty
-     * @return boolean true for a match
-     */
-    protected function validatePushSig($post, $hmac)
-    {
-        if ($this->secret) {
-            if (preg_match('/^sha1=([0-9a-fA-F]{40})$/', $hmac, $matches)) {
-                $their_hmac = strtolower($matches[1]);
-                $our_hmac = hash_hmac('sha1', $post, $this->secret);
-                if ($their_hmac === $our_hmac) {
-                    return true;
-                }
-                common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bad SHA-1 HMAC: got $their_hmac, expected $our_hmac");
-            } else {
-                common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bogus HMAC '$hmac'");
-            }
-        } else {
-            if (empty($hmac)) {
-                return true;
-            } else {
-                common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with unexpected HMAC '$hmac'");
-            }
-        }
-        return false;
-    }
-
-    /**
      * Process a posted entry from this feed source.
      *
      * @param DOMElement $entry
@@ -704,14 +435,14 @@ class Ostatus_profile extends Memcached_DataObject
     {
         if ($this->isGroup()) {
             // @fixme validate these profiles in some way!
-            $oprofile = $this->ensureActorProfile($activity);
+            $oprofile = self::ensureActorProfile($activity);
         } else {
-            $actorUri = $this->getActorProfileURI($activity);
-            if ($actorUri == $this->homeuri) {
+            $actorUri = self::getActorProfileURI($activity);
+            if ($actorUri == $this->uri) {
                 // @fixme check if profile info has changed and update it
             } else {
                 // @fixme drop or reject the messages once we've got the canonical profile URI recorded sanely
-                common_log(LOG_INFO, "OStatus: Warning: non-group post with unexpected author: $actorUri expected $this->homeuri");
+                common_log(LOG_INFO, "OStatus: Warning: non-group post with unexpected author: $actorUri expected $this->uri");
                 //return;
             }
             $oprofile = $this;
@@ -788,13 +519,72 @@ class Ostatus_profile extends Memcached_DataObject
     }
 
     /**
+     * @param string $profile_url
+     * @return Ostatus_profile
+     * @throws FeedSubException
+     */
+    public static function ensureProfile($profile_uri)
+    {
+        // Get the canonical feed URI and check it
+        $discover = new FeedDiscovery();
+        $feeduri = $discover->discoverFromURL($profile_uri);
+
+        $feedsub = FeedSub::ensureFeed($feeduri, $discover->feed);
+        $huburi = $discover->getAtomLink('hub');
+        $salmonuri = $discover->getAtomLink('salmon');
+
+        if (!$huburi) {
+            // We can only deal with folks with a PuSH hub
+            throw new FeedSubNoHubException();
+        }
+
+        // Ok this is going to be a terrible hack!
+        // Won't be suitable for groups, empty feeds, or getting
+        // info that's only available on the profile page.
+        $entries = $discover->feed->getElementsByTagNameNS(Activity::ATOM, 'entry');
+        if (!$entries || $entries->length == 0) {
+            throw new FeedSubException('empty feed');
+        }
+        $first = new Activity($entries->item(0), $discover->feed);
+        return self::ensureActorProfile($first, $feeduri);
+    }
+
+    /**
+     * Download and update given avatar image
+     * @param string $url
+     * @throws Exception in various failure cases
+     */
+    protected function updateAvatar($url)
+    {
+        // @fixme this should be better encapsulated
+        // ripped from oauthstore.php (for old OMB client)
+        $temp_filename = tempnam(sys_get_temp_dir(), 'listener_avatar');
+        copy($url, $temp_filename);
+        
+        // @fixme should we be using different ids?
+        $imagefile = new ImageFile($this->id, $temp_filename);
+        $filename = Avatar::filename($this->id,
+                                     image_type_to_extension($imagefile->type),
+                                     null,
+                                     common_timestamp());
+        rename($temp_filename, Avatar::path($filename));
+        if ($this->isGroup()) {
+            $group = $this->localGroup();
+            $group->setOriginal($filename);
+        } else {
+            $profile = $this->localProfile();
+            $profile->setOriginal($filename);
+        }
+    }
+
+    /**
      * Get an appropriate avatar image source URL, if available.
      *
      * @param ActivityObject $actor
      * @param DOMElement $feed
      * @return string
      */
-    function getAvatar($actor, $feed)
+    protected static function getAvatar($actor, $feed)
     {
         $url = '';
         $icon = '';
@@ -833,13 +623,18 @@ class Ostatus_profile extends Memcached_DataObject
     }
 
     /**
-     * @fixme move off of ostatus_profile or static?
+     * Fetch, or build if necessary, an Ostatus_profile for the actor
+     * in a given Activity Streams activity.
+     *
+     * @param Activity $activity
+     * @param string $feeduri if we already know the canonical feed URI!
+     * @return Ostatus_profile
      */
-    function ensureActorProfile($activity)
+    public static function ensureActorProfile($activity, $feeduri=null)
     {
-        $profile = $this->getActorProfile($activity);
+        $profile = self::getActorProfile($activity);
         if (!$profile) {
-            $profile = $this->createActorProfile($activity);
+            $profile = self::createActorProfile($activity, $feeduri);
         }
         return $profile;
     }
@@ -848,10 +643,10 @@ class Ostatus_profile extends Memcached_DataObject
      * @param Activity $activity
      * @return mixed matching Ostatus_profile or false if none known
      */
-    function getActorProfile($activity)
+    protected static function getActorProfile($activity)
     {
-        $homeuri = $this->getActorProfileURI($activity);
-        return Ostatus_profile::staticGet('homeuri', $homeuri);
+        $homeuri = self::getActorProfileURI($activity);
+        return self::staticGet('uri', $homeuri);
     }
 
     /**
@@ -859,7 +654,7 @@ class Ostatus_profile extends Memcached_DataObject
      * @return string
      * @throws ServerException
      */
-    function getActorProfileURI($activity)
+    protected static function getActorProfileURI($activity)
     {
         $opts = array('allowed_schemes' => array('http', 'https'));
         $actor = $activity->actor;
@@ -873,14 +668,19 @@ class Ostatus_profile extends Memcached_DataObject
     }
 
     /**
-     *
+     * @fixme validate stuff somewhere
      */
-    function createActorProfile($activity)
+    protected static function createActorProfile($activity, $feeduri=null)
     {
-        $actor = $activity->actor();
-        $homeuri = $this->getActivityProfileURI($activity);
-        $nickname = $this->getAuthorNick($activity);
-        $avatar = $this->getAvatar($actor, $feed);
+        $actor = $activity->actor;
+        $homeuri = self::getActorProfileURI($activity);
+        $nickname = self::getAuthorNick($activity);
+        $avatar = self::getAvatar($actor, $feed);
+
+        if (!$homeuri) {
+            common_log(LOG_DEBUG, __METHOD__ . " empty actor profile URI: " . var_export($activity, true));
+            throw new ServerException("No profile URI");
+        }
 
         $profile = new Profile();
         $profile->nickname   = $nickname;
@@ -894,9 +694,7 @@ class Ostatus_profile extends Memcached_DataObject
         // @todo lat/lon/location?
 
         $ok = $profile->insert();
-        if ($ok) {
-            $this->updateAvatar($profile, $avatar);
-        } else {
+        if (!$ok) {
             throw new ServerException("Can't save local profile");
         }
 
@@ -904,11 +702,15 @@ class Ostatus_profile extends Memcached_DataObject
         // or need to split out some of the feed stuff
         // so we can leave it empty until later.
         $oprofile = new Ostatus_profile();
-        $oprofile->homeuri = $homeuri;
+        $oprofile->uri = $homeuri;
+        if ($feeduri) {
+            $oprofile->feeduri = $feeduri;
+        }
         $oprofile->profile_id = $profile->id;
 
         $ok = $oprofile->insert();
         if ($ok) {
+            $oprofile->updateAvatar($avatar);
             return $oprofile;
         } else {
             throw new ServerException("Can't save OStatus profile");
@@ -920,13 +722,13 @@ class Ostatus_profile extends Memcached_DataObject
      * @param Activity $activity
      * @return string
      */
-    function getAuthorNick($activity)
+    protected static function getAuthorNick($activity)
     {
         // @fixme not technically part of the actor?
         foreach (array($activity->entry, $activity->feed) as $source) {
-            $author = ActivityUtil::child($source, 'author', Activity::ATOM);
+            $author = ActivityUtils::child($source, 'author', Activity::ATOM);
             if ($author) {
-                $name = ActivityUtil::child($author, 'name', Activity::ATOM);
+                $name = ActivityUtils::child($author, 'name', Activity::ATOM);
                 if ($name) {
                     return trim($name->textContent);
                 }
diff --git a/plugins/OStatus/lib/feeddiscovery.php b/plugins/OStatus/lib/feeddiscovery.php
index 39985fc90..7afb71bdc 100644
--- a/plugins/OStatus/lib/feeddiscovery.php
+++ b/plugins/OStatus/lib/feeddiscovery.php
@@ -48,6 +48,14 @@ class FeedSubNoFeedException extends FeedSubException
 {
 }
 
+class FeedSubBadXmlException extends FeedSubException
+{
+}
+
+class FeedSubNoHubException extends FeedSubException
+{
+}
+
 /**
  * Given a web page or feed URL, discover the final location of the feed
  * and return its current contents.
@@ -57,21 +65,25 @@ class FeedSubNoFeedException extends FeedSubException
  *   if ($feed->discoverFromURL($url)) {
  *     print $feed->uri;
  *     print $feed->type;
- *     processFeed($feed->body);
+ *     processFeed($feed->feed); // DOMDocument
  *   }
  */
 class FeedDiscovery
 {
     public $uri;
     public $type;
-    public $body;
+    public $feed;
 
+    /** Post-initialize query helper... */
+    public function getLink($rel, $type=null)
+    {
+        // @fixme check for non-Atom links in RSS2 feeds as well
+        return self::getAtomLink($rel, $type);
+    }
 
-    public function feedMunger()
+    public function getAtomLink($rel, $type=null)
     {
-        require_once 'XML/Feed/Parser.php';
-        $feed = new XML_Feed_Parser($this->body, false, false, true); // @fixme
-        return new FeedMunger($feed, $this->uri);
+        return ActivityUtils::getLink($this->feed->documentElement, $rel, $type);
     }
 
     /**
@@ -90,6 +102,7 @@ class FeedDiscovery
             $client = new HTTPClient();
             $response = $client->get($url);
         } catch (HTTP_Request2_Exception $e) {
+            common_log(LOG_ERR, __METHOD__ . " Failure for $url - " . $e->getMessage());
             throw new FeedSubBadURLException($e);
         }
 
@@ -107,7 +120,12 @@ class FeedDiscovery
         
         return $this->initFromResponse($response);
     }
-    
+
+    function discoverFromFeedURL($url)
+    {
+        return $this->discoverFromURL($url, false);
+    }
+
     function initFromResponse($response)
     {
         if (!$response->isOk()) {
@@ -122,16 +140,26 @@ class FeedDiscovery
 
         $type = $response->getHeader('Content-Type');
         if (preg_match('!^(text/xml|application/xml|application/(rss|atom)\+xml)!i', $type)) {
-            $this->uri = $sourceurl;
-            $this->type = $type;
-            $this->body = $body;
-            return true;
+            return $this->init($sourceurl, $type, $body);
         } else {
             common_log(LOG_WARNING, "Unrecognized feed type $type for $sourceurl");
             throw new FeedSubUnrecognizedTypeException($type);
         }
     }
 
+    function init($sourceurl, $type, $body)
+    {
+        $feed = new DOMDocument();
+        if ($feed->loadXML($body)) {
+            $this->uri = $sourceurl;
+            $this->type = $type;
+            $this->feed = $feed;
+            return $this->uri;
+        } else {
+            throw new FeedSubBadXmlException($url);
+        }
+    }
+
     /**
      * @param string $url source URL, used to resolve relative links
      * @param string $body HTML body text
diff --git a/plugins/OStatus/lib/feedmunger.php b/plugins/OStatus/lib/feedmunger.php
deleted file mode 100644
index e8c46de90..000000000
--- a/plugins/OStatus/lib/feedmunger.php
+++ /dev/null
@@ -1,350 +0,0 @@
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-/**
- * @package FeedSubPlugin
- * @maintainer Brion Vibber <brion@status.net>
- */
-
-if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
-
-class FeedSubPreviewNotice extends Notice
-{
-    protected $fetched = true;
-
-    function __construct($profile)
-    {
-        $this->profile = $profile;
-        $this->profile_id = 0;
-    }
-    
-    function getProfile()
-    {
-        return $this->profile;
-    }
-    
-    function find()
-    {
-        return true;
-    }
-    
-    function fetch()
-    {
-        $got = $this->fetched;
-        $this->fetched = false;
-        return $got;
-    }
-}
-
-class FeedSubPreviewProfile extends Profile
-{
-    function getAvatar($width, $height=null)
-    {
-        return new FeedSubPreviewAvatar($width, $height, $this->avatar);
-    }
-}
-
-class FeedSubPreviewAvatar extends Avatar
-{
-    function __construct($width, $height, $remote)
-    {
-        $this->remoteImage = $remote;
-    }
-
-    function displayUrl() {
-        return $this->remoteImage;
-    }
-}
-
-class FeedMunger
-{
-    /**
-     * @param XML_Feed_Parser $feed
-     */
-    function __construct($feed, $url=null)
-    {
-        $this->feed = $feed;
-        $this->url = $url;
-    }
-    
-    function ostatusProfile()
-    {
-        $profile = new Ostatus_profile();
-        $profile->feeduri = $this->url;
-        $profile->homeuri = $this->feed->link;
-        $profile->huburi = $this->getHubLink();
-        $salmon = $this->getSalmonLink();
-        if ($salmon) {
-            $profile->salmonuri = $salmon;
-        }
-        return $profile;
-    }
-
-    function getAtomLink($item, $attribs=array())
-    {
-        // XML_Feed_Parser gets confused by multiple <link> elements.
-        $dom = $item->model;
-
-        // Note that RSS feeds would embed an <atom:link> so this should work for both.
-        /// http://code.google.com/p/pubsubhubbub/wiki/RssFeeds
-        // <link rel='hub' href='http://pubsubhubbub.appspot.com/'/>
-        $links = $dom->getElementsByTagNameNS('http://www.w3.org/2005/Atom', 'link');
-        for ($i = 0; $i < $links->length; $i++) {
-            $node = $links->item($i);
-            if ($node->hasAttributes()) {
-                $href = $node->attributes->getNamedItem('href');
-                if ($href) {
-                    $matches = 0;
-                    foreach ($attribs as $name => $val) {
-                        $attrib = $node->attributes->getNamedItem($name);
-                        if ($attrib && $attrib->value == $val) {
-                            $matches++;
-                        }
-                    }
-                    if ($matches == count($attribs)) {
-                        return $href->value;
-                    }
-                }
-            }
-        }
-        return false;
-    }
-
-    function getRssLink($item)
-    {
-        // XML_Feed_Parser gets confused by multiple <link> elements.
-        $dom = $item->model;
-
-        // Note that RSS feeds would embed an <atom:link> so this should work for both.
-        /// http://code.google.com/p/pubsubhubbub/wiki/RssFeeds
-        // <link rel='hub' href='http://pubsubhubbub.appspot.com/'/>
-        $links = $dom->getElementsByTagName('link');
-        for ($i = 0; $i < $links->length; $i++) {
-            $node = $links->item($i);
-            if (!$node->hasAttributes()) {
-                return $node->textContent;
-            }
-        }
-        return false;
-    }
-
-    function getAltLink($item)
-    {
-        // Check for an atom link...
-        $link = $this->getAtomLink($item, array('rel' => 'alternate', 'type' => 'text/html'));
-        if (!$link) {
-            $link = $this->getRssLink($item);
-        }
-        return $link;
-    }
-
-    function getHubLink()
-    {
-        return $this->getAtomLink($this->feed, array('rel' => 'hub'));
-    }
-
-    function getSalmonLink()
-    {
-        return $this->getAtomLink($this->feed, array('rel' => 'salmon'));
-    }
-
-    function getSelfLink()
-    {
-        return $this->getAtomLink($this->feed, array('rel' => 'self'));
-    }
-
-    /**
-     * Get an appropriate avatar image source URL, if available.
-     * @return mixed string or false
-     */
-    function getAvatar()
-    {
-        $logo = $this->feed->logo;
-        if ($logo) {
-            return $logo;
-        }
-        $icon = $this->feed->icon;
-        if ($icon) {
-            return $icon;
-        }
-        return common_path('plugins/OStatus/images/48px-Feed-icon.svg.png');
-    }
-
-    function profile($preview=false)
-    {
-        if ($preview) {
-            $profile = new FeedSubPreviewProfile();
-        } else {
-            $profile = new Profile();
-        }
-        
-        // @todo validate/normalize nick?
-        $profile->nickname   = $this->feed->title;
-        $profile->fullname   = $this->feed->title;
-        $profile->homepage   = $this->getAltLink($this->feed);
-        $profile->bio        = $this->feed->description;
-        $profile->profileurl = $this->getAltLink($this->feed);
-
-        if ($preview) {
-            $profile->avatar = $this->getAvatar();
-        }
-        
-        // @todo tags from categories
-        // @todo lat/lon/location?
-
-        return $profile;
-    }
-
-    function notice($index=1, $preview=false)
-    {
-        $entry = $this->feed->getEntryByOffset($index);
-        if (!$entry) {
-            return null;
-        }
-
-        if ($preview) {
-            $notice = new FeedSubPreviewNotice($this->profile(true));
-            $notice->id = -1;
-        } else {
-            $notice = new Notice();
-            $notice->profile_id = $this->profileIdForEntry($index);
-        }
-
-        $link = $this->getAltLink($entry);
-        if (empty($link)) {
-            if (preg_match('!^https?://!', $entry->id)) {
-                $link = $entry->id;
-                common_log(LOG_DEBUG, "No link on entry, using URL from id: $link");
-            }
-        }
-        $notice->uri = $link;
-        $notice->url = $link;
-        $notice->content = $this->noticeFromEntry($entry);
-        $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';
-
-        $location = $this->getLocation($entry);
-        if ($location) {
-            if ($location->location_id) {
-                $notice->location_ns = $location->location_ns;
-                $notice->location_id = $location->location_id;
-            }
-            $notice->lat = $location->lat;
-            $notice->lon = $location->lon;
-        }
-
-        return $notice;
-    }
-
-    function profileIdForEntry($index=1)
-    {
-        // hack hack hack
-        // should get profile for this entry's author...
-        $feeduri = $this->getSelfLink();
-        $remote = Ostatus_profile::staticGet('feeduri', $feeduri);
-        if ($remote) {
-            return $remote->profile_id;
-        } else {
-            throw new Exception("Can't find feed profile for $feeduri");
-        }
-    }
-
-    /**
-     * Parse location given as a GeoRSS-simple point, if provided.
-     * http://www.georss.org/simple
-     *
-     * @param feed item $entry
-     * @return mixed Location or false
-     */
-    function getLocation($entry)
-    {
-        $dom = $entry->model;
-        $points = $dom->getElementsByTagNameNS('http://www.georss.org/georss', 'point');
-        
-        for ($i = 0; $i < $points->length; $i++) {
-            $point = $points->item(0)->textContent;
-            $point = str_replace(',', ' ', $point); // per spec "treat commas as whitespace"
-            $point = preg_replace('/\s+/', ' ', $point);
-            $point = trim($point);
-            $coords = explode(' ', $point);
-            if (count($coords) == 2) {
-                list($lat, $lon) = $coords;
-                if (is_numeric($lat) && is_numeric($lon)) {
-                    common_log(LOG_INFO, "Looking up location for $lat $lon from georss");
-                    return Location::fromLatLon($lat, $lon);
-                }
-            }
-            common_log(LOG_ERR, "Ignoring bogus georss:point value $point");
-        }
-
-        return false;
-    }
-
-    /**
-     * @param XML_Feed_Type $entry
-     * @return string notice text, within post size limit
-     */
-    function noticeFromEntry($entry)
-    {
-        $max = Notice::maxContent();
-        $ellipsis = "\xe2\x80\xa6"; // U+2026 HORIZONTAL ELLIPSIS
-        $title = $entry->title;
-        $link = $entry->link;
-
-        // @todo We can get <category> entries like this:
-        // $cats = $entry->getCategory('category', array(0, true));
-        // but it feels like an awful hack. If it's accessible cleanly,
-        // try adding #hashtags from the categories/tags on a post.
-
-        $title = $entry->title;
-        $link = $this->getAltLink($entry);
-        if ($link) {
-            // Blog post or such...
-            // @todo Should we force a language here?
-            $format = _m('New post: "%1$s" %2$s');
-            $out = sprintf($format, $title, $link);
-
-            // Trim link if needed...
-            if (mb_strlen($out) > $max) {
-                $link = common_shorten_url($link);
-                $out = sprintf($format, $title, $link);
-            }
-
-            // Trim title if needed...
-            if (mb_strlen($out) > $max) {
-                $used = mb_strlen($out) - mb_strlen($title);
-                $available = $max - $used - mb_strlen($ellipsis);
-                $title = mb_substr($title, 0, $available) . $ellipsis;
-                $out = sprintf($format, $title, $link);
-            }
-        } else {
-            // No link? Consider a bare status update.
-            if (mb_strlen($title) > $max) {
-                $available = $max - mb_strlen($ellipsis);
-                $out = mb_substr($title, 0, $available) . $ellipsis;
-            } else {
-                $out = $title;
-            }
-        }
-        
-        return $out;
-    }
-}