6 files changed, 154 insertions, 4 deletions
diff --git a/plugins/OStatus/lib/hubprepqueuehandler.php b/plugins/OStatus/lib/hubprepqueuehandler.php
new file mode 100644
index 000000000..0d585938f
--- /dev/null
+++ b/plugins/OStatus/lib/hubprepqueuehandler.php
@@ -0,0 +1,87 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 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/>.
+ */
+
+/**
+ * When we have a large batch of PuSH consumers, we break the data set
+ * into smaller chunks. Enqueue final destinations...
+ *
+ * @package Hub
+ * @author Brion Vibber <brion@status.net>
+ */
+class HubPrepQueueHandler extends QueueHandler
+{
+    // Enqueue this many low-level distributions before re-queueing the rest
+    // of the batch to be processed later. Helps to keep latency down for other
+    // things happening during a particularly long OStatus delivery session.
+    //
+    // [Could probably ditch this if we had working message delivery priorities
+    // for queueing, but this isn't supported in ActiveMQ 5.3.]
+    const ROLLING_BATCH = 20;
+
+    function transport()
+    {
+        return 'hubprep';
+    }
+
+    function handle($data)
+    {
+        $topic = $data['topic'];
+        $atom = $data['atom'];
+        $pushCallbacks = $data['pushCallbacks'];
+
+        assert(is_string($atom));
+        assert(is_string($topic));
+        assert(is_array($pushCallbacks));
+
+        // Set up distribution for the first n subscribing sites...
+        // If we encounter an uncatchable error, queue handling should
+        // automatically re-run the batch, which could lead to some dupe
+        // distributions.
+        //
+        // Worst case is if one of these hubprep entries dies too many
+        // times and gets dropped; the rest of the batch won't get processed.
+        try {
+            $n = 0;
+            while (count($pushCallbacks) && $n < self::ROLLING_BATCH) {
+                $n++;
+                $callback = array_shift($pushCallbacks);
+                $sub = HubSub::staticGet($topic, $callback);
+                if (!$sub) {
+                    common_log(LOG_ERR, "Skipping PuSH delivery for deleted(?) consumer $callback on $topic");
+                    continue;
+                }
+
+                $sub->distribute($atom);
+            }
+        } catch (Exception $e) {
+            common_log(LOG_ERR, "Exception during PuSH batch out: " .
+                                $e->getMessage() .
+                                " prepping $topic to $callback");
+        }
+
+        // And re-queue the rest of the batch!
+        if (count($pushCallbacks) > 0) {
+            $sub = new HubSub();
+            $sub->topic = $topic;
+            $sub->bulkDistribute($atom, $pushCallbacks);
+        }
+
+        return true;
+    }
+}
diff --git a/plugins/Sitemap/SitemapPlugin.php b/plugins/Sitemap/SitemapPlugin.php
index d4d295237..b6d3b1ad3 100644
--- a/plugins/Sitemap/SitemapPlugin.php
+++ b/plugins/Sitemap/SitemapPlugin.php
@@ -202,6 +202,12 @@ class SitemapPlugin extends Plugin
                                                  null, false),
                                    new ColumnDef('modified', 'timestamp')));
 
+        $userCreated = $schema->getColumnDef('user', 'created');
+
+        if (empty($userCreated) || $userCreated->key != 'MUL') {
+            $schema->createIndex('user', 'created');
+        }
+
         return true;
     }
 
diff --git a/plugins/Sitemap/Sitemap_notice_count.php b/plugins/Sitemap/Sitemap_notice_count.php
index 2a375b3e4..6e0061e97 100644
--- a/plugins/Sitemap/Sitemap_notice_count.php
+++ b/plugins/Sitemap/Sitemap_notice_count.php
@@ -153,7 +153,9 @@ class Sitemap_notice_count extends Memcached_DataObject
                 $noticeCounts[$snc->notice_date] = $snc->notice_count;
             }
 
-            self::cacheSet('sitemap:notice:counts', $noticeCounts);
+            // Cache notice counts for 4 hours.
+
+            self::cacheSet('sitemap:notice:counts', $noticeCounts, null, time() + 4 * 60 * 60);
         }
 
         return $noticeCounts;
diff --git a/plugins/Sitemap/Sitemap_user_count.php b/plugins/Sitemap/Sitemap_user_count.php
index 64b4c3442..98dd05bfe 100644
--- a/plugins/Sitemap/Sitemap_user_count.php
+++ b/plugins/Sitemap/Sitemap_user_count.php
@@ -154,7 +154,9 @@ class Sitemap_user_count extends Memcached_DataObject
                 $userCounts[$suc->registration_date] = $suc->user_count;
             }
 
-            self::cacheSet('sitemap:user:counts', $userCounts);
+            // Cache user counts for 4 hours.
+
+            self::cacheSet('sitemap:user:counts', $userCounts, null, time() + 4 * 60 * 60);
         }
 
         return $userCounts;
diff --git a/plugins/Sitemap/scripts/updatecounts.php b/plugins/Sitemap/scripts/updatecounts.php
new file mode 100644
index 000000000..91bc0ac4e
--- /dev/null
+++ b/plugins/Sitemap/scripts/updatecounts.php
@@ -0,0 +1,36 @@
+#!/usr/bin/env php
+<?php
+/*
+ * StatusNet - a distributed open-source microblogging tool
+ * Copyright (C) 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/>.
+ */
+
+define('INSTALLDIR', realpath(dirname(__FILE__) . '/../../..'));
+
+$helptext = <<<END_OF_UPDATECOUNTS_HELP
+updatecounts.php [options]
+Update the notice and user counts cached in the database.
+
+END_OF_UPDATECOUNTS_HELP;
+
+require_once INSTALLDIR.'/scripts/commandline.inc';
+
+// Will fill the cache
+
+$userCounts = Sitemap_user_count::getAll();
+$noticeCounts = Sitemap_notice_count::getAll();
+
+echo "Done.\n";
diff --git a/plugins/TwitterBridge/twitter.php b/plugins/TwitterBridge/twitter.php
index 896eee2da..306ba2442 100644
--- a/plugins/TwitterBridge/twitter.php
+++ b/plugins/TwitterBridge/twitter.php
@@ -75,8 +75,6 @@ function save_twitter_user($twitter_id, $screen_name)
 
     if (!empty($fuser)) {
 
-        $result = true;
-
         // Delete old record if Twitter user changed screen name
 
         if ($fuser->nickname != $screen_name) {
@@ -88,6 +86,25 @@ function save_twitter_user($twitter_id, $screen_name)
                                          $screen_name,
                                          $oldname));
         }
+
+    } else {
+
+        // Kill any old, invalid records for this screen name
+
+        $fuser = Foreign_user::getByNickname($screen_name, TWITTER_SERVICE);
+
+        if (!empty($fuser)) {
+            $fuser->delete();
+            common_log(
+                LOG_INFO,
+                sprintf(
+                    'Twitter bridge - deteted old record for Twitter ' .
+                    'screen name "%s" belonging to Twitter ID %d.',
+                    $screen_name,
+                    $fuser->id
+                )
+            );
+        }
     }
 
     return add_twitter_user($twitter_id, $screen_name);