X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=classes%2FNotice.php;h=4f23e35007f4cac4f099832fa6df805185082d2f;hb=61fde6c8fff5201393db4496918fe193fe960c01;hp=6610721564ff45ee19dca9714035fb3de0e82664;hpb=496547699847761b0dfcc0d51182e28298428db4;p=quix0rs-gnu-social.git diff --git a/classes/Notice.php b/classes/Notice.php index 6610721564..4f23e35007 100644 --- a/classes/Notice.php +++ b/classes/Notice.php @@ -29,6 +29,7 @@ * @author Robin Millette * @author Sarven Capadisli * @author Tom Adams + * @copyright 2009 Free Software Foundation, Inc http://www.fsf.org * @license GNU Affero General Public License http://www.gnu.org/licenses/ */ @@ -41,10 +42,10 @@ if (!defined('STATUSNET') && !defined('LACONICA')) { */ require_once INSTALLDIR.'/classes/Memcached_DataObject.php'; -/* We keep the first three 20-notice pages, plus one for pagination check, +/* We keep 200 notices, the max number of notices available per API request, * in the memcached cache. */ -define('NOTICE_CACHE_WINDOW', 61); +define('NOTICE_CACHE_WINDOW', 200); define('MAX_BOXCARS', 128); @@ -55,24 +56,26 @@ class Notice extends Memcached_DataObject public $__table = 'notice'; // table name public $id; // int(4) primary_key not_null - public $profile_id; // int(4) not_null + public $profile_id; // int(4) multiple_key not_null public $uri; // varchar(255) unique_key - public $content; // text() - public $rendered; // text() + public $content; // text + public $rendered; // text public $url; // varchar(255) - public $created; // datetime() not_null - public $modified; // timestamp() not_null default_CURRENT_TIMESTAMP + public $created; // datetime multiple_key not_null default_0000-00-00%2000%3A00%3A00 + public $modified; // timestamp not_null default_CURRENT_TIMESTAMP public $reply_to; // int(4) - public $is_local; // tinyint(1) + public $is_local; // int(4) public $source; // varchar(32) public $conversation; // int(4) public $lat; // decimal(10,7) public $lon; // decimal(10,7) public $location_id; // int(4) public $location_ns; // int(4) + public $repeat_of; // int(4) /* Static get */ - function staticGet($k,$v=NULL) { + function staticGet($k,$v=NULL) + { return Memcached_DataObject::staticGet('Notice',$k,$v); } @@ -87,48 +90,60 @@ class Notice extends Memcached_DataObject function getProfile() { - return Profile::staticGet('id', $this->profile_id); + $profile = Profile::staticGet('id', $this->profile_id); + + if (empty($profile)) { + // TRANS: Server exception thrown when a user profile for a notice cannot be found. + // TRANS: %1$d is a profile ID (number), %2$d is a notice ID (number). + throw new ServerException(sprintf(_('No such profile (%1$d) for notice (%2$d).'), $this->profile_id, $this->id)); + } + + return $profile; } function delete() { - $this->blowCaches(true); - $this->blowFavesCache(true); - $this->blowSubsCache(true); - // For auditing purposes, save a record that the notice // was deleted. - $deleted = new Deleted_notice(); + // @fixme we have some cases where things get re-run and so the + // insert fails. + $deleted = Deleted_notice::staticGet('id', $this->id); + if (!$deleted) { + $deleted = new Deleted_notice(); - $deleted->id = $this->id; - $deleted->profile_id = $this->profile_id; - $deleted->uri = $this->uri; - $deleted->created = $this->created; - $deleted->deleted = common_sql_now(); + $deleted->id = $this->id; + $deleted->profile_id = $this->profile_id; + $deleted->uri = $this->uri; + $deleted->created = $this->created; + $deleted->deleted = common_sql_now(); - $this->query('BEGIN'); + $deleted->insert(); + } - $deleted->insert(); + if (Event::handle('NoticeDeleteRelated', array($this))) { - //Null any notices that are replies to this notice - $this->query(sprintf("UPDATE notice set reply_to = null WHERE reply_to = %d", $this->id)); - $related = array('Reply', - 'Fave', - 'Notice_tag', - 'Group_inbox', - 'Queue_item', - 'Notice_inbox'); - - foreach ($related as $cls) { - $inst = new $cls(); - $inst->notice_id = $this->id; - $inst->delete(); + // Clear related records + + $this->clearReplies(); + $this->clearRepeats(); + $this->clearFaves(); + $this->clearTags(); + $this->clearGroupInboxes(); + + // NOTE: we don't clear inboxes + // NOTE: we don't clear queue items } + $result = parent::delete(); - $this->query('COMMIT'); + + $this->blowOnDelete(); + return $result; } + /** + * Extract #hashtags from this notice's content and save them to the database. + */ function saveTags() { /* extract all #hastags */ @@ -137,21 +152,34 @@ class Notice extends Memcached_DataObject return true; } + /* Add them to the database */ + return $this->saveKnownTags($match[1]); + } + + /** + * Record the given set of hash tags in the db for this notice. + * Given tag strings will be normalized and checked for dupes. + */ + function saveKnownTags($hashtags) + { //turn each into their canonical tag //this is needed to remove dupes before saving e.g. #hash.tag = #hashtag - $hashtags = array(); - for($i=0; $isaveTag($hashtag); + self::blow('profile:notice_ids_tagged:%d:%s', $this->profile_id, $hashtag); } return true; } + /** + * Record a single hash tag as associated with this notice. + * Tag format and uniqueness must be validated by caller. + */ function saveTag($hashtag) { $tag = new Notice_tag(); @@ -161,43 +189,106 @@ class Notice extends Memcached_DataObject $id = $tag->insert(); if (!$id) { - throw new ServerException(sprintf(_('DB error inserting hashtag: %s'), + // TRANS: Server exception. %s are the error details. + throw new ServerException(sprintf(_('Database error inserting hashtag: %s'), $last_error->message)); return; } + + // if it's saved, blow its cache + $tag->blowCache(false); } - static function saveNew($profile_id, $content, $source=null, - $is_local=Notice::LOCAL_PUBLIC, $reply_to=null, $uri=null, $created=null, - $lat=null, $lon=null, $location_id=null, $location_ns=null) { + /** + * Save a new notice and push it out to subscribers' inboxes. + * Poster's permissions are checked before sending. + * + * @param int $profile_id Profile ID of the poster + * @param string $content source message text; links may be shortened + * per current user's preference + * @param string $source source key ('web', 'api', etc) + * @param array $options Associative array of optional properties: + * string 'created' timestamp of notice; defaults to now + * int 'is_local' source/gateway ID, one of: + * Notice::LOCAL_PUBLIC - Local, ok to appear in public timeline + * Notice::REMOTE_OMB - Sent from a remote OMB service; + * hide from public timeline but show in + * local "and friends" timelines + * Notice::LOCAL_NONPUBLIC - Local, but hide from public timeline + * Notice::GATEWAY - From another non-OMB service; + * will not appear in public views + * float 'lat' decimal latitude for geolocation + * float 'lon' decimal longitude for geolocation + * int 'location_id' geoname identifier + * int 'location_ns' geoname namespace to interpret location_id + * int 'reply_to'; notice ID this is a reply to + * int 'repeat_of'; notice ID this is a repeat of + * string 'uri' unique ID for notice; defaults to local notice URL + * string 'url' permalink to notice; defaults to local notice URL + * string 'rendered' rendered HTML version of content + * array 'replies' list of profile URIs for reply delivery in + * place of extracting @-replies from content. + * array 'groups' list of group IDs to deliver to, in place of + * extracting ! tags from content + * array 'tags' list of hashtag strings to save with the notice + * in place of extracting # tags from content + * array 'urls' list of attached/referred URLs to save with the + * notice in place of extracting links from content + * @fixme tag override + * + * @return Notice + * @throws ClientException + */ + static function saveNew($profile_id, $content, $source, $options=null) { + $defaults = array('uri' => null, + 'url' => null, + 'reply_to' => null, + 'repeat_of' => null); + + if (!empty($options)) { + $options = $options + $defaults; + extract($options); + } else { + extract($defaults); + } + + if (!isset($is_local)) { + $is_local = Notice::LOCAL_PUBLIC; + } $profile = Profile::staticGet($profile_id); $final = common_shorten_links($content); if (Notice::contentTooLong($final)) { + // TRANS: Client exception thrown if a notice contains too many characters. throw new ClientException(_('Problem saving notice. Too long.')); } if (empty($profile)) { + // TRANS: Client exception thrown when trying to save a notice for an unknown user. throw new ClientException(_('Problem saving notice. Unknown user.')); } if (common_config('throttle', 'enabled') && !Notice::checkEditThrottle($profile_id)) { common_log(LOG_WARNING, 'Excessive posting by profile #' . $profile_id . '; throttled.'); + // TRANS: Client exception thrown when a user tries to post too many notices in a given time frame. throw new ClientException(_('Too many notices too fast; take a breather '. 'and post again in a few minutes.')); } if (common_config('site', 'dupelimit') > 0 && !Notice::checkDupes($profile_id, $final)) { common_log(LOG_WARNING, 'Dupe posting by profile #' . $profile_id . '; throttled.'); + // TRANS: Client exception thrown when a user tries to post too many duplicate notices in a given time frame. throw new ClientException(_('Too many duplicate messages too quickly;'. ' take a breather and post again in a few minutes.')); } if (!$profile->hasRight(Right::NEWNOTICE)) { common_log(LOG_WARNING, "Attempted post from user disallowed to post: " . $profile->nickname); - throw new ClientException(_('You are banned from posting notices on this site.')); + + // TRANS: Client exception thrown when a user tries to post while being banned. + throw new ClientException(_('You are banned from posting notices on this site.'), 403); } $notice = new Notice(); @@ -221,11 +312,18 @@ class Notice extends Memcached_DataObject } $notice->content = $final; - $notice->rendered = common_render_content($final, $notice); + $notice->source = $source; $notice->uri = $uri; + $notice->url = $url; - $notice->reply_to = self::getReplyTo($reply_to, $profile_id, $source, $final); + // Handle repeat case + + if (isset($repeat_of)) { + $notice->repeat_of = $repeat_of; + } else { + $notice->reply_to = self::getReplyTo($reply_to, $profile_id, $source, $final); + } if (!empty($notice->reply_to)) { $reply = Notice::staticGet('id', $notice->reply_to); @@ -235,33 +333,28 @@ class Notice extends Memcached_DataObject if (!empty($lat) && !empty($lon)) { $notice->lat = $lat; $notice->lon = $lon; + } + + if (!empty($location_ns) && !empty($location_id)) { $notice->location_id = $location_id; $notice->location_ns = $location_ns; - } else if (!empty($location_ns) && !empty($location_id)) { - $location = Location::fromId($location_id, $location_ns); - if (!empty($location)) { - $notice->lat = $location->lat; - $notice->lon = $location->lon; - $notice->location_id = $location_id; - $notice->location_ns = $location_ns; - } + } + + if (!empty($rendered)) { + $notice->rendered = $rendered; } else { - $notice->lat = $profile->lat; - $notice->lon = $profile->lon; - $notice->location_id = $profile->location_id; - $notice->location_ns = $profile->location_ns; + $notice->rendered = common_render_content($final, $notice); } if (Event::handle('StartNoticeSave', array(&$notice))) { // XXX: some of these functions write to the DB - $notice->query('BEGIN'); - $id = $notice->insert(); if (!$id) { common_log_db_error($notice, 'INSERT', __FILE__); + // TRANS: Server exception thrown when a notice cannot be saved. throw new ServerException(_('Problem saving notice.')); } @@ -280,36 +373,99 @@ class Notice extends Memcached_DataObject // the beginning of a new conversation. if (empty($notice->conversation)) { - $notice->conversation = $notice->id; + $conv = Conversation::create(); + $notice->conversation = $conv->id; $changed = true; } if ($changed) { if (!$notice->update($orig)) { common_log_db_error($notice, 'UPDATE', __FILE__); + // TRANS: Server exception thrown when a notice cannot be updated. throw new ServerException(_('Problem saving notice.')); } } - // XXX: do we need to change this for remote users? + } + + # Clear the cache for subscribed users, so they'll update at next request + # XXX: someone clever could prepend instead of clearing the cache + + $notice->blowOnInsert(); + + // Save per-notice metadata... + + if (isset($replies)) { + $notice->saveKnownReplies($replies); + } else { + $notice->saveReplies(); + } + if (isset($tags)) { + $notice->saveKnownTags($tags); + } else { $notice->saveTags(); + } - $notice->addToInboxes(); + // Note: groups may save tags, so must be run after tags are saved + // to avoid errors on duplicates. + if (isset($groups)) { + $notice->saveKnownGroups($groups); + } else { + $notice->saveGroups(); + } + if (isset($urls)) { + $notice->saveKnownUrls($urls); + } else { $notice->saveUrls(); + } + + // Prepare inbox delivery, may be queued to background. + $notice->distribute(); + + return $notice; + } + + function blowOnInsert($conversation = false) + { + self::blow('profile:notice_ids:%d', $this->profile_id); + self::blow('public'); - $notice->query('COMMIT'); + // XXX: Before we were blowing the casche only if the notice id + // was not the root of the conversation. What to do now? - Event::handle('EndNoticeSave', array($notice)); + self::blow('notice:conversation_ids:%d', $this->conversation); + + if (!empty($this->repeat_of)) { + self::blow('notice:repeats:%d', $this->repeat_of); } - # Clear the cache for subscribed users, so they'll update at next request - # XXX: someone clever could prepend instead of clearing the cache + $original = Notice::staticGet('id', $this->repeat_of); + + if (!empty($original)) { + $originalUser = User::staticGet('id', $original->profile_id); + if (!empty($originalUser)) { + self::blow('user:repeats_of_me:%d', $originalUser->id); + } + } - $notice->blowCaches(); + $profile = Profile::staticGet($this->profile_id); + if (!empty($profile)) { + $profile->blowNoticeCount(); + } + } - return $notice; + /** + * Clear cache entries related to this notice at delete time. + * Necessary to avoid breaking paging on public, profile timelines. + */ + function blowOnDelete() + { + $this->blowOnInsert(); + + self::blow('profile:notice_ids:%d;last', $this->profile_id); + self::blow('public;last'); } /** save all urls in the notice to the db @@ -323,6 +479,25 @@ class Notice extends Memcached_DataObject common_replace_urls_callback($this->content, array($this, 'saveUrl'), $this->id); } + /** + * Save the given URLs as related links/attachments to the db + * + * follow redirects and save all available file information + * (mimetype, date, size, oembed, etc.) + * + * @return void + */ + function saveKnownUrls($urls) + { + // @fixme validation? + foreach (array_unique($urls) as $url) { + File::processNew($url, $this->id); + } + } + + /** + * @private callback + */ function saveUrl($data) { list($url, $notice_id) = $data; File::processNew($url, $notice_id); @@ -408,867 +583,1061 @@ class Notice extends Memcached_DataObject if ($f2p->find()) { while ($f2p->fetch()) { $f = File::staticGet($f2p->file_id); - $att[] = clone($f); + if ($f) { + $att[] = clone($f); + } } } return $att; } - function blowCaches($blowLast=false) - { - $this->blowSubsCache($blowLast); - $this->blowNoticeCache($blowLast); - $this->blowRepliesCache($blowLast); - $this->blowPublicCache($blowLast); - $this->blowTagCache($blowLast); - $this->blowGroupCache($blowLast); - $this->blowConversationCache($blowLast); - $profile = Profile::staticGet($this->profile_id); - $profile->blowNoticeCount(); - } - - function blowConversationCache($blowLast=false) + function getStreamByIds($ids) { $cache = common_memcache(); - if ($cache) { - $ck = common_cache_key('notice:conversation_ids:'.$this->conversation); - $cache->delete($ck); - if ($blowLast) { - $cache->delete($ck.';last'); - } - } - } - function blowGroupCache($blowLast=false) - { - $cache = common_memcache(); - if ($cache) { - $group_inbox = new Group_inbox(); - $group_inbox->notice_id = $this->id; - if ($group_inbox->find()) { - while ($group_inbox->fetch()) { - $cache->delete(common_cache_key('user_group:notice_ids:' . $group_inbox->group_id)); - if ($blowLast) { - $cache->delete(common_cache_key('user_group:notice_ids:' . $group_inbox->group_id.';last')); - } - $member = new Group_member(); - $member->group_id = $group_inbox->group_id; - if ($member->find()) { - while ($member->fetch()) { - $cache->delete(common_cache_key('notice_inbox:by_user:' . $member->profile_id)); - if ($blowLast) { - $cache->delete(common_cache_key('notice_inbox:by_user:' . $member->profile_id . ';last')); - } - } - } + if (!empty($cache)) { + $notices = array(); + foreach ($ids as $id) { + $n = Notice::staticGet('id', $id); + if (!empty($n)) { + $notices[] = $n; } } - $group_inbox->free(); - unset($group_inbox); - } - } - - function blowTagCache($blowLast=false) - { - $cache = common_memcache(); - if ($cache) { - $tag = new Notice_tag(); - $tag->notice_id = $this->id; - if ($tag->find()) { - while ($tag->fetch()) { - $tag->blowCache($blowLast); - $ck = 'profile:notice_ids_tagged:' . $this->profile_id . ':' . $tag->tag; - - $cache->delete($ck); - if ($blowLast) { - $cache->delete($ck . ';last'); - } - } + return new ArrayWrapper($notices); + } else { + $notice = new Notice(); + if (empty($ids)) { + //if no IDs requested, just return the notice object + return $notice; } - $tag->free(); - unset($tag); - } - } - - function blowSubsCache($blowLast=false) - { - $cache = common_memcache(); - if ($cache) { - $user = new User(); + $notice->whereAdd('id in (' . implode(', ', $ids) . ')'); - $UT = common_config('db','type')=='pgsql'?'"user"':'user'; - $user->query('SELECT id ' . + $notice->find(); - "FROM $UT JOIN subscription ON $UT.id = subscription.subscriber " . - 'WHERE subscription.subscribed = ' . $this->profile_id); + $temp = array(); - while ($user->fetch()) { - $cache->delete(common_cache_key('notice_inbox:by_user:'.$user->id)); - $cache->delete(common_cache_key('notice_inbox:by_user_own:'.$user->id)); - if ($blowLast) { - $cache->delete(common_cache_key('notice_inbox:by_user:'.$user->id.';last')); - $cache->delete(common_cache_key('notice_inbox:by_user_own:'.$user->id.';last')); - } + while ($notice->fetch()) { + $temp[$notice->id] = clone($notice); } - $user->free(); - unset($user); - } - } - function blowNoticeCache($blowLast=false) - { - if ($this->is_local) { - $cache = common_memcache(); - if (!empty($cache)) { - $cache->delete(common_cache_key('profile:notice_ids:'.$this->profile_id)); - if ($blowLast) { - $cache->delete(common_cache_key('profile:notice_ids:'.$this->profile_id.';last')); - } - } - } - } + $wrapped = array(); - function blowRepliesCache($blowLast=false) - { - $cache = common_memcache(); - if ($cache) { - $reply = new Reply(); - $reply->notice_id = $this->id; - if ($reply->find()) { - while ($reply->fetch()) { - $cache->delete(common_cache_key('reply:stream:'.$reply->profile_id)); - if ($blowLast) { - $cache->delete(common_cache_key('reply:stream:'.$reply->profile_id.';last')); - } + foreach ($ids as $id) { + if (array_key_exists($id, $temp)) { + $wrapped[] = $temp[$id]; } } - $reply->free(); - unset($reply); + + return new ArrayWrapper($wrapped); } } - function blowPublicCache($blowLast=false) + function publicStream($offset=0, $limit=20, $since_id=0, $max_id=0) { - if ($this->is_local == Notice::LOCAL_PUBLIC) { - $cache = common_memcache(); - if ($cache) { - $cache->delete(common_cache_key('public')); - if ($blowLast) { - $cache->delete(common_cache_key('public').';last'); - } - } - } + $ids = Notice::stream(array('Notice', '_publicStreamDirect'), + array(), + 'public', + $offset, $limit, $since_id, $max_id); + return Notice::getStreamByIds($ids); } - function blowFavesCache($blowLast=false) + function _publicStreamDirect($offset=0, $limit=20, $since_id=0, $max_id=0) { - $cache = common_memcache(); - if ($cache) { - $fave = new Fave(); - $fave->notice_id = $this->id; - if ($fave->find()) { - while ($fave->fetch()) { - $cache->delete(common_cache_key('fave:ids_by_user:'.$fave->user_id)); - $cache->delete(common_cache_key('fave:by_user_own:'.$fave->user_id)); - if ($blowLast) { - $cache->delete(common_cache_key('fave:ids_by_user:'.$fave->user_id.';last')); - $cache->delete(common_cache_key('fave:by_user_own:'.$fave->user_id.';last')); - } - } - } - $fave->free(); - unset($fave); - } - } + $notice = new Notice(); - # XXX: too many args; we need to move to named params or even a separate - # class for notice streams + $notice->selectAdd(); // clears it + $notice->selectAdd('id'); - static function getStream($qry, $cachekey, $offset=0, $limit=20, $since_id=0, $max_id=0, $order=null, $since=null) { + $notice->orderBy('id DESC'); - if (common_config('memcached', 'enabled')) { + if (!is_null($offset)) { + $notice->limit($offset, $limit); + } - # Skip the cache if this is a since, since_id or max_id qry - if ($since_id > 0 || $max_id > 0 || $since) { - return Notice::getStreamDirect($qry, $offset, $limit, $since_id, $max_id, $order, $since); - } else { - return Notice::getCachedStream($qry, $cachekey, $offset, $limit, $order); - } + if (common_config('public', 'localonly')) { + $notice->whereAdd('is_local = ' . Notice::LOCAL_PUBLIC); + } else { + # -1 == blacklisted, -2 == gateway (i.e. Twitter) + $notice->whereAdd('is_local !='. Notice::LOCAL_NONPUBLIC); + $notice->whereAdd('is_local !='. Notice::GATEWAY); } - return Notice::getStreamDirect($qry, $offset, $limit, $since_id, $max_id, $order, $since); - } + if ($since_id != 0) { + $notice->whereAdd('id > ' . $since_id); + } - static function getStreamDirect($qry, $offset, $limit, $since_id, $max_id, $order, $since) { + if ($max_id != 0) { + $notice->whereAdd('id <= ' . $max_id); + } - $needAnd = false; - $needWhere = true; + $ids = array(); - if (preg_match('/\bWHERE\b/i', $qry)) { - $needWhere = false; - $needAnd = true; + if ($notice->find()) { + while ($notice->fetch()) { + $ids[] = $notice->id; + } } - if ($since_id > 0) { + $notice->free(); + $notice = NULL; - if ($needWhere) { - $qry .= ' WHERE '; - $needWhere = false; - } else { - $qry .= ' AND '; - } + return $ids; + } - $qry .= ' notice.id > ' . $since_id; - } + function conversationStream($id, $offset=0, $limit=20, $since_id=0, $max_id=0) + { + $ids = Notice::stream(array('Notice', '_conversationStreamDirect'), + array($id), + 'notice:conversation_ids:'.$id, + $offset, $limit, $since_id, $max_id); - if ($max_id > 0) { + return Notice::getStreamByIds($ids); + } - if ($needWhere) { - $qry .= ' WHERE '; - $needWhere = false; - } else { - $qry .= ' AND '; - } + function _conversationStreamDirect($id, $offset=0, $limit=20, $since_id=0, $max_id=0) + { + $notice = new Notice(); - $qry .= ' notice.id <= ' . $max_id; - } + $notice->selectAdd(); // clears it + $notice->selectAdd('id'); - if ($since) { + $notice->conversation = $id; - if ($needWhere) { - $qry .= ' WHERE '; - $needWhere = false; - } else { - $qry .= ' AND '; - } + $notice->orderBy('id DESC'); - $qry .= ' notice.created > \'' . date('Y-m-d H:i:s', $since) . '\''; + if (!is_null($offset)) { + $notice->limit($offset, $limit); } - # Allow ORDER override - - if ($order) { - $qry .= $order; - } else { - $qry .= ' ORDER BY notice.created DESC, notice.id DESC '; + if ($since_id != 0) { + $notice->whereAdd('id > ' . $since_id); } - if (common_config('db','type') == 'pgsql') { - $qry .= ' LIMIT ' . $limit . ' OFFSET ' . $offset; - } else { - $qry .= ' LIMIT ' . $offset . ', ' . $limit; + if ($max_id != 0) { + $notice->whereAdd('id <= ' . $max_id); } - $notice = new Notice(); + $ids = array(); - $notice->query($qry); + if ($notice->find()) { + while ($notice->fetch()) { + $ids[] = $notice->id; + } + } - return $notice; - } + $notice->free(); + $notice = NULL; - # XXX: this is pretty long and should probably be broken up into - # some helper functions + return $ids; + } - static function getCachedStream($qry, $cachekey, $offset, $limit, $order) { + /** + * Is this notice part of an active conversation? + * + * @return boolean true if other messages exist in the same + * conversation, false if this is the only one + */ + function hasConversation() + { + if (!empty($this->conversation)) { + $conversation = Notice::conversationStream( + $this->conversation, + 1, + 1 + ); + if ($conversation->N > 0) { + return true; + } + } + return false; + } - # If outside our cache window, just go to the DB + /** + * Pull up a full list of local recipients who will be getting + * this notice in their inbox. Results will be cached, so don't + * change the input data wily-nilly! + * + * @param array $groups optional list of Group objects; + * if left empty, will be loaded from group_inbox records + * @param array $recipient optional list of reply profile ids + * if left empty, will be loaded from reply records + * @return array associating recipient user IDs with an inbox source constant + */ + function whoGets($groups=null, $recipients=null) + { + $c = self::memcache(); - if ($offset + $limit > NOTICE_CACHE_WINDOW) { - return Notice::getStreamDirect($qry, $offset, $limit, null, null, $order, null); + if (!empty($c)) { + $ni = $c->get(common_cache_key('notice:who_gets:'.$this->id)); + if ($ni !== false) { + return $ni; + } } - # Get the cache; if we can't, just go to the DB - - $cache = common_memcache(); + if (is_null($groups)) { + $groups = $this->getGroups(); + } - if (empty($cache)) { - return Notice::getStreamDirect($qry, $offset, $limit, null, null, $order, null); + if (is_null($recipients)) { + $recipients = $this->getReplies(); } - # Get the notices out of the cache + $users = $this->getSubscribedUsers(); - $notices = $cache->get(common_cache_key($cachekey)); + // FIXME: kind of ignoring 'transitional'... + // we'll probably stop supporting inboxless mode + // in 0.9.x - # On a cache hit, return a DB-object-like wrapper + $ni = array(); - if ($notices !== false) { - $wrapper = new ArrayWrapper(array_slice($notices, $offset, $limit)); - return $wrapper; + foreach ($users as $id) { + $ni[$id] = NOTICE_INBOX_SOURCE_SUB; } - # If the cache was invalidated because of new data being - # added, we can try and just get the new stuff. We keep an additional - # copy of the data at the key + ';last' - - # No cache hit. Try to get the *last* cached version + foreach ($groups as $group) { + $users = $group->getUserMembers(); + foreach ($users as $id) { + if (!array_key_exists($id, $ni)) { + $ni[$id] = NOTICE_INBOX_SOURCE_GROUP; + } + } + } - $last_notices = $cache->get(common_cache_key($cachekey) . ';last'); + foreach ($recipients as $recipient) { + if (!array_key_exists($recipient, $ni)) { + $ni[$recipient] = NOTICE_INBOX_SOURCE_REPLY; + } + } - if ($last_notices) { + // Exclude any deleted, non-local, or blocking recipients. + $profile = $this->getProfile(); + foreach ($ni as $id => $source) { + $user = User::staticGet('id', $id); + if (empty($user) || $user->hasBlocked($profile)) { + unset($ni[$id]); + } + } - # Reverse-chron order, so last ID is last. + if (!empty($c)) { + // XXX: pack this data better + $c->set(common_cache_key('notice:who_gets:'.$this->id), $ni); + } - $last_id = $last_notices[0]->id; + return $ni; + } - # XXX: this assumes monotonically increasing IDs; a fair - # bet with our DB. + /** + * Adds this notice to the inboxes of each local user who should receive + * it, based on author subscriptions, group memberships, and @-replies. + * + * Warning: running a second time currently will make items appear + * multiple times in users' inboxes. + * + * @fixme make more robust against errors + * @fixme break up massive deliveries to smaller background tasks + * + * @param array $groups optional list of Group objects; + * if left empty, will be loaded from group_inbox records + * @param array $recipient optional list of reply profile ids + * if left empty, will be loaded from reply records + */ + function addToInboxes($groups=null, $recipients=null) + { + $ni = $this->whoGets($groups, $recipients); - $new_notice = Notice::getStreamDirect($qry, 0, NOTICE_CACHE_WINDOW, - $last_id, null, $order, null); + $ids = array_keys($ni); - if ($new_notice) { - $new_notices = array(); - while ($new_notice->fetch()) { - $new_notices[] = clone($new_notice); - } - $new_notice->free(); - $notices = array_slice(array_merge($new_notices, $last_notices), - 0, NOTICE_CACHE_WINDOW); + // We remove the author (if they're a local user), + // since we'll have already done this in distribute() - # Store the array in the cache for next time + $i = array_search($this->profile_id, $ids); - $result = $cache->set(common_cache_key($cachekey), $notices); - $result = $cache->set(common_cache_key($cachekey) . ';last', $notices); + if ($i !== false) { + unset($ids[$i]); + } - # return a wrapper of the array for use now + // Bulk insert - return new ArrayWrapper(array_slice($notices, $offset, $limit)); - } - } + Inbox::bulkInsert($this->id, $ids); - # Otherwise, get the full cache window out of the DB + return; + } - $notice = Notice::getStreamDirect($qry, 0, NOTICE_CACHE_WINDOW, null, null, $order, null); + function getSubscribedUsers() + { + $user = new User(); - # If there are no hits, just return the value + if(common_config('db','quote_identifiers')) + $user_table = '"user"'; + else $user_table = 'user'; - if (empty($notice)) { - return $notice; - } + $qry = + 'SELECT id ' . + 'FROM '. $user_table .' JOIN subscription '. + 'ON '. $user_table .'.id = subscription.subscriber ' . + 'WHERE subscription.subscribed = %d '; - # Pack results into an array + $user->query(sprintf($qry, $this->profile_id)); - $notices = array(); + $ids = array(); - while ($notice->fetch()) { - $notices[] = clone($notice); + while ($user->fetch()) { + $ids[] = $user->id; } - $notice->free(); + $user->free(); - # Store the array in the cache for next time + return $ids; + } - $result = $cache->set(common_cache_key($cachekey), $notices); - $result = $cache->set(common_cache_key($cachekey) . ';last', $notices); + /** + * Record this notice to the given group inboxes for delivery. + * Overrides the regular parsing of !group markup. + * + * @param string $group_ids + * @fixme might prefer URIs as identifiers, as for replies? + * best with generalizations on user_group to support + * remote groups better. + */ + function saveKnownGroups($group_ids) + { + if (!is_array($group_ids)) { + // TRANS: Server exception thrown when no array is provided to the method saveKnownGroups(). + throw new ServerException(_("Bad type provided to saveKnownGroups")); + } - # return a wrapper of the array for use now + $groups = array(); + foreach (array_unique($group_ids) as $id) { + $group = User_group::staticGet('id', $id); + if ($group) { + common_log(LOG_ERR, "Local delivery to group id $id, $group->nickname"); + $result = $this->addToGroupInbox($group); + if (!$result) { + common_log_db_error($gi, 'INSERT', __FILE__); + } - $wrapper = new ArrayWrapper(array_slice($notices, $offset, $limit)); + // @fixme should we save the tags here or not? + $groups[] = clone($group); + } else { + common_log(LOG_ERR, "Local delivery to group id $id skipped, doesn't exist"); + } + } - return $wrapper; + return $groups; } - function getStreamByIds($ids) + /** + * Parse !group delivery and record targets into group_inbox. + * @return array of Group objects + */ + function saveGroups() { - $cache = common_memcache(); + // Don't save groups for repeats - if (!empty($cache)) { - $notices = array(); - foreach ($ids as $id) { - $n = Notice::staticGet('id', $id); - if (!empty($n)) { - $notices[] = $n; - } + if (!empty($this->repeat_of)) { + return array(); + } + + $groups = array(); + + /* extract all !group */ + $count = preg_match_all('/(?:^|\s)!([A-Za-z0-9]{1,64})/', + strtolower($this->content), + $match); + if (!$count) { + return $groups; + } + + $profile = $this->getProfile(); + + /* Add them to the database */ + + foreach (array_unique($match[1]) as $nickname) { + /* XXX: remote groups. */ + $group = User_group::getForNickname($nickname, $profile); + + if (empty($group)) { + continue; } - return new ArrayWrapper($notices); - } else { - $notice = new Notice(); - if (empty($ids)) { - //if no IDs requested, just return the notice object - return $notice; + + // we automatically add a tag for every group name, too + + $tag = Notice_tag::pkeyGet(array('tag' => common_canonical_tag($nickname), + 'notice_id' => $this->id)); + + if (is_null($tag)) { + $this->saveTag($nickname); } - $notice->whereAdd('id in (' . implode(', ', $ids) . ')'); - $notice->orderBy('id DESC'); - $notice->find(); - return $notice; + if ($profile->isMember($group)) { + + $result = $this->addToGroupInbox($group); + + if (!$result) { + common_log_db_error($gi, 'INSERT', __FILE__); + } + + $groups[] = clone($group); + } } + + return $groups; } - function publicStream($offset=0, $limit=20, $since_id=0, $max_id=0, $since=null) + function addToGroupInbox($group) { - $ids = Notice::stream(array('Notice', '_publicStreamDirect'), - array(), - 'public', - $offset, $limit, $since_id, $max_id, $since); + $gi = Group_inbox::pkeyGet(array('group_id' => $group->id, + 'notice_id' => $this->id)); - return Notice::getStreamByIds($ids); - } + if (empty($gi)) { - function _publicStreamDirect($offset=0, $limit=20, $since_id=0, $max_id=0, $since=null) - { - $notice = new Notice(); + $gi = new Group_inbox(); - $notice->selectAdd(); // clears it - $notice->selectAdd('id'); + $gi->group_id = $group->id; + $gi->notice_id = $this->id; + $gi->created = $this->created; - $notice->orderBy('id DESC'); + $result = $gi->insert(); - if (!is_null($offset)) { - $notice->limit($offset, $limit); - } + if (!$result) { + common_log_db_error($gi, 'INSERT', __FILE__); + // TRANS: Server exception thrown when an update for a group inbox fails. + throw new ServerException(_('Problem saving group inbox.')); + } - if (common_config('public', 'localonly')) { - $notice->whereAdd('is_local = ' . Notice::LOCAL_PUBLIC); - } else { - # -1 == blacklisted, -2 == gateway (i.e. Twitter) - $notice->whereAdd('is_local !='. Notice::LOCAL_NONPUBLIC); - $notice->whereAdd('is_local !='. Notice::GATEWAY); + self::blow('user_group:notice_ids:%d', $gi->group_id); } - if ($since_id != 0) { - $notice->whereAdd('id > ' . $since_id); - } + return true; + } - if ($max_id != 0) { - $notice->whereAdd('id <= ' . $max_id); + /** + * Save reply records indicating that this notice needs to be + * delivered to the local users with the given URIs. + * + * Since this is expected to be used when saving foreign-sourced + * messages, we won't deliver to any remote targets as that's the + * source service's responsibility. + * + * Mail notifications etc will be handled later. + * + * @param array of unique identifier URIs for recipients + */ + function saveKnownReplies($uris) + { + if (empty($uris)) { + return; } - if (!is_null($since)) { - $notice->whereAdd('created > \'' . date('Y-m-d H:i:s', $since) . '\''); - } + $sender = Profile::staticGet($this->profile_id); - $ids = array(); + foreach (array_unique($uris) as $uri) { - if ($notice->find()) { - while ($notice->fetch()) { - $ids[] = $notice->id; + $profile = Profile::fromURI($uri); + + if (empty($profile)) { + common_log(LOG_WARNING, "Unable to determine profile for URI '$uri'"); + continue; } - } - $notice->free(); - $notice = NULL; + if ($profile->hasBlocked($sender)) { + common_log(LOG_INFO, "Not saving reply to profile {$profile->id} ($uri) from sender {$sender->id} because of a block."); + continue; + } - return $ids; - } + $reply = new Reply(); - function conversationStream($id, $offset=0, $limit=20, $since_id=0, $max_id=0, $since=null) - { - $ids = Notice::stream(array('Notice', '_conversationStreamDirect'), - array($id), - 'notice:conversation_ids:'.$id, - $offset, $limit, $since_id, $max_id, $since); + $reply->notice_id = $this->id; + $reply->profile_id = $profile->id; - return Notice::getStreamByIds($ids); + common_log(LOG_INFO, __METHOD__ . ": saving reply: notice $this->id to profile $profile->id"); + + $id = $reply->insert(); + } + + return; } - function _conversationStreamDirect($id, $offset=0, $limit=20, $since_id=0, $max_id=0, $since=null) + /** + * Pull @-replies from this message's content in StatusNet markup format + * and save reply records indicating that this message needs to be + * delivered to those users. + * + * Mail notifications to local profiles will be sent later. + * + * @return array of integer profile IDs + */ + + function saveReplies() { - $notice = new Notice(); + // Don't save reply data for repeats - $notice->selectAdd(); // clears it - $notice->selectAdd('id'); + if (!empty($this->repeat_of)) { + return array(); + } - $notice->conversation = $id; + $sender = Profile::staticGet($this->profile_id); - $notice->orderBy('id DESC'); + // @todo ideally this parser information would only + // be calculated once. - if (!is_null($offset)) { - $notice->limit($offset, $limit); - } + $mentions = common_find_mentions($this->content, $this); - if ($since_id != 0) { - $notice->whereAdd('id > ' . $since_id); - } + $replied = array(); - if ($max_id != 0) { - $notice->whereAdd('id <= ' . $max_id); - } + // store replied only for first @ (what user/notice what the reply directed, + // we assume first @ is it) + + foreach ($mentions as $mention) { + + foreach ($mention['mentioned'] as $mentioned) { + + // skip if they're already covered + + if (!empty($replied[$mentioned->id])) { + continue; + } + + // Don't save replies from blocked profile to local user - if (!is_null($since)) { - $notice->whereAdd('created > \'' . date('Y-m-d H:i:s', $since) . '\''); + $mentioned_user = User::staticGet('id', $mentioned->id); + if (!empty($mentioned_user) && $mentioned_user->hasBlocked($sender)) { + continue; + } + + $reply = new Reply(); + + $reply->notice_id = $this->id; + $reply->profile_id = $mentioned->id; + + $id = $reply->insert(); + + if (!$id) { + common_log_db_error($reply, 'INSERT', __FILE__); + // TRANS: Server exception thrown when a reply cannot be saved. + // TRANS: %1$d is a notice ID, %2$d is the ID of the mentioned user. + throw new ServerException(sprintf(_("Could not save reply for %1$d, %2$d."), $this->id, $mentioned->id)); + } else { + $replied[$mentioned->id] = 1; + self::blow('reply:stream:%d', $mentioned->id); + } + } } + $recipientIds = array_keys($replied); + + return $recipientIds; + } + + /** + * Pull the complete list of @-reply targets for this notice. + * + * @return array of integer profile ids + */ + function getReplies() + { + // XXX: cache me + $ids = array(); - if ($notice->find()) { - while ($notice->fetch()) { - $ids[] = $notice->id; + $reply = new Reply(); + $reply->selectAdd(); + $reply->selectAdd('profile_id'); + $reply->notice_id = $this->id; + + if ($reply->find()) { + while($reply->fetch()) { + $ids[] = $reply->profile_id; } } - $notice->free(); - $notice = NULL; + $reply->free(); return $ids; } - function addToInboxes() + /** + * Send e-mail notifications to local @-reply targets. + * + * Replies must already have been saved; this is expected to be run + * from the distrib queue handler. + */ + function sendReplyNotifications() { - // XXX: loads constants + // Don't send reply notifications for repeats - $inbox = new Notice_inbox(); + if (!empty($this->repeat_of)) { + return array(); + } - $users = $this->getSubscribedUsers(); + $recipientIds = $this->getReplies(); - // FIXME: kind of ignoring 'transitional'... - // we'll probably stop supporting inboxless mode - // in 0.9.x + foreach ($recipientIds as $recipientId) { + $user = User::staticGet('id', $recipientId); + if (!empty($user)) { + mail_notify_attn($user, $this); + } + } + } - $ni = array(); + /** + * Pull list of groups this notice needs to be delivered to, + * as previously recorded by saveGroups() or saveKnownGroups(). + * + * @return array of Group objects + */ + function getGroups() + { + // Don't save groups for repeats - foreach ($users as $id) { - $ni[$id] = NOTICE_INBOX_SOURCE_SUB; + if (!empty($this->repeat_of)) { + return array(); } - $groups = $this->saveGroups(); - $profile = $this->getProfile(); + // XXX: cache me - foreach ($groups as $group) { - $users = $group->getUserMembers(); - foreach ($users as $id) { - if (!array_key_exists($id, $ni)) { - $user = User::staticGet('id', $id); - if (!$user->hasBlocked($profile)) { - $ni[$id] = NOTICE_INBOX_SOURCE_GROUP; - } + $groups = array(); + + $gi = new Group_inbox(); + + $gi->selectAdd(); + $gi->selectAdd('group_id'); + + $gi->notice_id = $this->id; + + if ($gi->find()) { + while ($gi->fetch()) { + $group = User_group::staticGet('id', $gi->group_id); + if ($group) { + $groups[] = $group; } } } - $recipients = $this->saveReplies(); + $gi->free(); - foreach ($recipients as $recipient) { + return $groups; + } - if (!array_key_exists($recipient, $ni)) { - $recipientUser = User::staticGet('id', $recipient); - if (!empty($recipientUser)) { - $ni[$recipient] = NOTICE_INBOX_SOURCE_REPLY; - } + function asActivity() + { + $profile = $this->getProfile(); + + $act = new Activity(); + + $act->actor = ActivityObject::fromProfile($profile); + $act->verb = ActivityVerb::POST; + $act->objects[] = ActivityObject::fromNotice($this); + + $act->time = strtotime($this->created); + $act->link = $this->bestUrl(); + + $act->content = common_xml_safe_str($this->rendered); + $act->id = $this->uri; + $act->title = common_xml_safe_str($this->content); + + $ctx = new ActivityContext(); + + if (!empty($this->reply_to)) { + $reply = Notice::staticGet('id', $this->reply_to); + if (!empty($reply)) { + $ctx->replyToID = $reply->uri; + $ctx->replyToUrl = $reply->bestUrl(); } } - $cnt = 0; + $ctx->location = $this->getLocation(); - $qryhdr = 'INSERT INTO notice_inbox (user_id, notice_id, source, created) VALUES '; - $qry = $qryhdr; + $conv = null; - foreach ($ni as $id => $source) { - if ($cnt > 0) { - $qry .= ', '; - } - $qry .= '('.$id.', '.$this->id.', '.$source.", '".$this->created. "') "; - $cnt++; - if (rand() % NOTICE_INBOX_SOFT_LIMIT == 0) { - // FIXME: Causes lag in replicated servers - // Notice_inbox::gc($id); - } - if ($cnt >= MAX_BOXCARS) { - $inbox = new Notice_inbox(); - $result = $inbox->query($qry); - if (PEAR::isError($result)) { - common_log_db_error($inbox, $qry); - } - $qry = $qryhdr; - $cnt = 0; + if (!empty($this->conversation)) { + $conv = Conversation::staticGet('id', $this->conversation); + if (!empty($conv)) { + $ctx->conversation = $conv->uri; } } - if ($cnt > 0) { - $inbox = new Notice_inbox(); - $result = $inbox->query($qry); - if (PEAR::isError($result)) { - common_log_db_error($inbox, $qry); + $reply_ids = $this->getReplies(); + + foreach ($reply_ids as $id) { + $profile = Profile::staticGet('id', $id); + if (!empty($profile)) { + $ctx->attention[] = $profile->getUri(); } } - return; + $groups = $this->getGroups(); + + foreach ($groups as $group) { + $ctx->attention[] = $group->uri; + } + + $act->context = $ctx; + + return $act; } - function getSubscribedUsers() + // This has gotten way too long. Needs to be sliced up into functional bits + // or ideally exported to a utility class. + + function asAtomEntry($namespace=false, $source=false, $author=true, $cur=null) { - $user = new User(); + $profile = $this->getProfile(); - if(common_config('db','quote_identifiers')) - $user_table = '"user"'; - else $user_table = 'user'; + $xs = new XMLStringer(true); - $qry = - 'SELECT id ' . - 'FROM '. $user_table .' JOIN subscription '. - 'ON '. $user_table .'.id = subscription.subscriber ' . - 'WHERE subscription.subscribed = %d '; + if ($namespace) { + $attrs = array('xmlns' => 'http://www.w3.org/2005/Atom', + 'xmlns:thr' => 'http://purl.org/syndication/thread/1.0', + 'xmlns:georss' => 'http://www.georss.org/georss', + 'xmlns:activity' => 'http://activitystrea.ms/spec/1.0/', + 'xmlns:media' => 'http://purl.org/syndication/atommedia', + 'xmlns:poco' => 'http://portablecontacts.net/spec/1.0', + 'xmlns:ostatus' => 'http://ostatus.org/schema/1.0', + 'xmlns:statusnet' => 'http://status.net/schema/api/1/'); + } else { + $attrs = array(); + } - $user->query(sprintf($qry, $this->profile_id)); + if (Event::handle('StartActivityStart', array(&$this, &$xs, &$attrs))) { + $xs->elementStart('entry', $attrs); + Event::handle('EndActivityStart', array(&$this, &$xs, &$attrs)); + } - $ids = array(); + if (Event::handle('StartActivitySource', array(&$this, &$xs))) { - while ($user->fetch()) { - $ids[] = $user->id; - } + if ($source) { - $user->free(); + $atom_feed = $profile->getAtomFeed(); - return $ids; - } + if (!empty($atom_feed)) { - function saveGroups() - { - $groups = array(); + $xs->elementStart('source'); - /* extract all !group */ - $count = preg_match_all('/(?:^|\s)!([A-Za-z0-9]{1,64})/', - strtolower($this->content), - $match); - if (!$count) { - return $groups; + // XXX: we should store the actual feed ID + + $xs->element('id', null, $atom_feed); + + // XXX: we should store the actual feed title + + $xs->element('title', null, $profile->getBestName()); + + $xs->element('link', array('rel' => 'alternate', + 'type' => 'text/html', + 'href' => $profile->profileurl)); + + $xs->element('link', array('rel' => 'self', + 'type' => 'application/atom+xml', + 'href' => $atom_feed)); + + $xs->element('icon', null, $profile->avatarUrl(AVATAR_PROFILE_SIZE)); + + $notice = $profile->getCurrentNotice(); + + if (!empty($notice)) { + $xs->element('updated', null, self::utcDate($notice->created)); + } + + $user = User::staticGet('id', $profile->id); + + if (!empty($user)) { + $xs->element('link', array('rel' => 'license', + 'href' => common_config('license', 'url'))); + } + + $xs->elementEnd('source'); + } + } + Event::handle('EndActivitySource', array(&$this, &$xs)); } - $profile = $this->getProfile(); + $title = common_xml_safe_str($this->content); - /* Add them to the database */ + if (Event::handle('StartActivityTitle', array(&$this, &$xs, &$title))) { + $xs->element('title', null, $title); + Event::handle('EndActivityTitle', array($this, &$xs, $title)); + } - foreach (array_unique($match[1]) as $nickname) { - /* XXX: remote groups. */ - $group = User_group::getForNickname($nickname); + $atomAuthor = ''; - if (empty($group)) { - continue; + if ($author) { + $atomAuthor = $profile->asAtomAuthor($cur); + } + + if (Event::handle('StartActivityAuthor', array(&$this, &$xs, &$atomAuthor))) { + if (!empty($atomAuthor)) { + $xs->raw($atomAuthor); + Event::handle('EndActivityAuthor', array(&$this, &$xs, &$atomAuthor)); } + } - // we automatically add a tag for every group name, too + $actor = ''; - $tag = Notice_tag::pkeyGet(array('tag' => common_canonical_tag($nickname), - 'notice_id' => $this->id)); + if ($author) { + $actor = $profile->asActivityActor(); + } - if (is_null($tag)) { - $this->saveTag($nickname); + if (Event::handle('StartActivityActor', array(&$this, &$xs, &$actor))) { + if (!empty($actor)) { + $xs->raw($actor); + Event::handle('EndActivityActor', array(&$this, &$xs, &$actor)); } + } + + $url = $this->bestUrl(); + + if (Event::handle('StartActivityLink', array(&$this, &$xs, &$url))) { + $xs->element('link', array('rel' => 'alternate', + 'type' => 'text/html', + 'href' => $url)); + Event::handle('EndActivityLink', array(&$this, &$xs, $url)); + } + + $id = $this->uri; + + if (Event::handle('StartActivityId', array(&$this, &$xs, &$id))) { + $xs->element('id', null, $id); + Event::handle('EndActivityId', array(&$this, &$xs, $id)); + } + + $published = self::utcDate($this->created); + + if (Event::handle('StartActivityPublished', array(&$this, &$xs, &$published))) { + $xs->element('published', null, $published); + Event::handle('EndActivityPublished', array(&$this, &$xs, $published)); + } + + $updated = $published; // XXX: notices are usually immutable + + if (Event::handle('StartActivityUpdated', array(&$this, &$xs, &$updated))) { + $xs->element('updated', null, $updated); + Event::handle('EndActivityUpdated', array(&$this, &$xs, $updated)); + } - if ($profile->isMember($group)) { + $content = common_xml_safe_str($this->rendered); - $result = $this->addToGroupInbox($group); + if (Event::handle('StartActivityContent', array(&$this, &$xs, &$content))) { + $xs->element('content', array('type' => 'html'), $content); + Event::handle('EndActivityContent', array(&$this, &$xs, $content)); + } - if (!$result) { - common_log_db_error($gi, 'INSERT', __FILE__); - } + // Most of our notices represent POSTing a NOTE. This is the default verb + // for activity streams, so we normally just leave it out. - $groups[] = clone($group); - } + $verb = ActivityVerb::POST; + + if (Event::handle('StartActivityVerb', array(&$this, &$xs, &$verb))) { + $xs->element('activity:verb', null, $verb); + Event::handle('EndActivityVerb', array(&$this, &$xs, $verb)); } - return $groups; - } + // We use the default behavior for activity streams: if there's no activity:object, + // then treat the entry itself as the object. Here, you can set the type of that object, + // which is normally a NOTE. - function addToGroupInbox($group) - { - $gi = Group_inbox::pkeyGet(array('group_id' => $group->id, - 'notice_id' => $this->id)); + $type = ActivityObject::NOTE; - if (empty($gi)) { + if (Event::handle('StartActivityDefaultObjectType', array(&$this, &$xs, &$type))) { + $xs->element('activity:object-type', null, $type); + Event::handle('EndActivityDefaultObjectType', array(&$this, &$xs, $type)); + } - $gi = new Group_inbox(); + // Since we usually use the entry itself as an object, we don't have an explicit + // object. Some extensions may want to add them (for photo, event, music, etc.). - $gi->group_id = $group->id; - $gi->notice_id = $this->id; - $gi->created = $this->created; + $objects = array(); - return $gi->insert(); + if (Event::handle('StartActivityObjects', array(&$this, &$xs, &$objects))) { + foreach ($objects as $object) { + $xs->raw($object->asString()); + } + Event::handle('EndActivityObjects', array(&$this, &$xs, $objects)); } - return true; - } + $noticeInfoAttr = array('local_id' => $this->id); // local notice ID (useful to clients for ordering) - function saveReplies() - { - // Alternative reply format - $tname = false; - if (preg_match('/^T ([A-Z0-9]{1,64}) /', $this->content, $match)) { - $tname = $match[1]; + $ns = $this->getSource(); + + if (!empty($ns)) { + $noticeInfoAttr['source'] = $ns->code; + if (!empty($ns->url)) { + $noticeInfoAttr['source_link'] = $ns->url; + if (!empty($ns->name)) { + $noticeInfoAttr['source'] = '' + . htmlspecialchars($ns->name) + . ''; + } + } } - // extract all @messages - $cnt = preg_match_all('/(?:^|\s)@([a-z0-9]{1,64})/', $this->content, $match); - $names = array(); + if (!empty($cur)) { + $noticeInfoAttr['favorite'] = ($cur->hasFave($this)) ? "true" : "false"; + $profile = $cur->getProfile(); + $noticeInfoAttr['repeated'] = ($profile->hasRepeated($this->id)) ? "true" : "false"; + } - if ($cnt || $tname) { - // XXX: is there another way to make an array copy? - $names = ($tname) ? array_unique(array_merge(array(strtolower($tname)), $match[1])) : array_unique($match[1]); + if (!empty($this->repeat_of)) { + $noticeInfoAttr['repeat_of'] = $this->repeat_of; } - $sender = Profile::staticGet($this->profile_id); + if (Event::handle('StartActivityNoticeInfo', array(&$this, &$xs, &$noticeInfoAttr))) { + $xs->element('statusnet:notice_info', $noticeInfoAttr, null); + Event::handle('EndActivityNoticeInfo', array(&$this, &$xs, $noticeInfoAttr)); + } - $replied = array(); + $replyNotice = null; - // store replied only for first @ (what user/notice what the reply directed, - // we assume first @ is it) + if ($this->reply_to) { + $replyNotice = Notice::staticGet('id', $this->reply_to); + } - for ($i=0; $icreated); - if (empty($recipient)) { - continue; - } - // Don't save replies from blocked profile to local user - $recipient_user = User::staticGet('id', $recipient->id); - if (!empty($recipient_user) && $recipient_user->hasBlocked($sender)) { - continue; - } - $reply = new Reply(); - $reply->notice_id = $this->id; - $reply->profile_id = $recipient->id; - $id = $reply->insert(); - if (!$id) { - $last_error = &PEAR::getStaticProperty('DB_DataObject','lastError'); - common_log(LOG_ERR, 'DB error inserting reply: ' . $last_error->message); - common_server_error(sprintf(_('DB error inserting reply: %s'), $last_error->message)); - return array(); - } else { - $replied[$recipient->id] = 1; - } - } - - // Hash format replies, too - $cnt = preg_match_all('/(?:^|\s)@#([a-z0-9]{1,64})/', $this->content, $match); - if ($cnt) { - foreach ($match[1] as $tag) { - $tagged = Profile_tag::getTagged($sender->id, $tag); - foreach ($tagged as $t) { - if (!$replied[$t->id]) { - // Don't save replies from blocked profile to local user - $t_user = User::staticGet('id', $t->id); - if ($t_user && $t_user->hasBlocked($sender)) { - continue; - } - $reply = new Reply(); - $reply->notice_id = $this->id; - $reply->profile_id = $t->id; - $id = $reply->insert(); - if (!$id) { - common_log_db_error($reply, 'INSERT', __FILE__); - return array(); - } else { - $replied[$recipient->id] = 1; - } - } - } + if (Event::handle('StartActivityInReplyTo', array(&$this, &$xs, &$replyNotice))) { + if (!empty($replyNotice)) { + $xs->element('link', array('rel' => 'related', + 'href' => $replyNotice->bestUrl())); + $xs->element('thr:in-reply-to', + array('ref' => $replyNotice->uri, + 'href' => $replyNotice->bestUrl())); + Event::handle('EndActivityInReplyTo', array(&$this, &$xs, $replyNotice)); } } - $recipientIds = array_keys($replied); + $conv = null; - foreach ($recipientIds as $recipient) { - $user = User::staticGet('id', $recipient); - if ($user) { - mail_notify_attn($user, $this); + if (!empty($this->conversation)) { + $conv = Conversation::staticGet('id', $this->conversation); + } + + if (Event::handle('StartActivityConversation', array(&$this, &$xs, &$conv))) { + if (!empty($conv)) { + $xs->element('link', array('rel' => 'ostatus:conversation', + 'href' => $conv->uri)); } + Event::handle('EndActivityConversation', array(&$this, &$xs, $conv)); } - return $recipientIds; - } + $replyProfiles = array(); - function asAtomEntry($namespace=false, $source=false) - { - $profile = $this->getProfile(); + $reply_ids = $this->getReplies(); - $xs = new XMLStringer(true); + foreach ($reply_ids as $id) { + $profile = Profile::staticGet('id', $id); + if (!empty($profile)) { + $replyProfiles[] = $profile; + } + } - if ($namespace) { - $attrs = array('xmlns' => 'http://www.w3.org/2005/Atom', - 'xmlns:thr' => 'http://purl.org/syndication/thread/1.0'); - } else { - $attrs = array(); + if (Event::handle('StartActivityAttentionProfiles', array(&$this, &$xs, &$replyProfiles))) { + foreach ($replyProfiles as $profile) { + $xs->element('link', array('rel' => 'ostatus:attention', + 'href' => $profile->getUri())); + $xs->element('link', array('rel' => 'mentioned', + 'href' => $profile->getUri())); + } + Event::handle('EndActivityAttentionProfiles', array(&$this, &$xs, $replyProfiles)); } - $xs->elementStart('entry', $attrs); + $groups = $this->getGroups(); - if ($source) { - $xs->elementStart('source'); - $xs->element('title', null, $profile->nickname . " - " . common_config('site', 'name')); - $xs->element('link', array('href' => $profile->profileurl)); - $user = User::staticGet('id', $profile->id); - if (!empty($user)) { - $atom_feed = common_local_url('ApiTimelineUser', - array('format' => 'atom', - 'id' => $profile->nickname)); - $xs->element('link', array('rel' => 'self', - 'type' => 'application/atom+xml', - 'href' => $profile->profileurl)); - $xs->element('link', array('rel' => 'license', - 'href' => common_config('license', 'url'))); + if (Event::handle('StartActivityAttentionGroups', array(&$this, &$xs, &$groups))) { + foreach ($groups as $group) { + $xs->element('link', array('rel' => 'ostatus:attention', + 'href' => $group->permalink())); + $xs->element('link', array('rel' => 'mentioned', + 'href' => $group->permalink())); } - - $xs->element('icon', null, $profile->avatarUrl(AVATAR_PROFILE_SIZE)); + Event::handle('EndActivityAttentionGroups', array(&$this, &$xs, $groups)); } - $xs->elementStart('author'); - $xs->element('name', null, $profile->nickname); - $xs->element('uri', null, $profile->profileurl); - $xs->elementEnd('author'); + $repeat = null; - if ($source) { - $xs->elementEnd('source'); + if (!empty($this->repeat_of)) { + $repeat = Notice::staticGet('id', $this->repeat_of); } - $xs->element('title', null, $this->content); - $xs->element('summary', null, $this->content); - - $xs->element('link', array('rel' => 'alternate', - 'href' => $this->bestUrl())); + if (Event::handle('StartActivityForward', array(&$this, &$xs, &$repeat))) { + if (!empty($repeat)) { + $xs->element('ostatus:forward', + array('ref' => $repeat->uri, + 'href' => $repeat->bestUrl())); + } - $xs->element('id', null, $this->uri); + Event::handle('EndActivityForward', array(&$this, &$xs, $repeat)); + } - $xs->element('published', null, common_date_w3dtf($this->created)); - $xs->element('updated', null, common_date_w3dtf($this->modified)); + $tags = $this->getTags(); - if ($this->reply_to) { - $reply_notice = Notice::staticGet('id', $this->reply_to); - if (!empty($reply_notice)) { - $xs->element('link', array('rel' => 'related', - 'href' => $reply_notice->bestUrl())); - $xs->element('thr:in-reply-to', - array('ref' => $reply_notice->uri, - 'href' => $reply_notice->bestUrl())); + if (Event::handle('StartActivityCategories', array(&$this, &$xs, &$tags))) { + foreach ($tags as $tag) { + $xs->element('category', array('term' => $tag)); } + Event::handle('EndActivityCategories', array(&$this, &$xs, $tags)); } - $xs->element('content', array('type' => 'html'), $this->rendered); + // Enclosures - $tag = new Notice_tag(); - $tag->notice_id = $this->id; - if ($tag->find()) { - while ($tag->fetch()) { - $xs->element('category', array('term' => $tag->tag)); + $enclosures = array(); + + $attachments = $this->attachments(); + + foreach ($attachments as $attachment) { + $enclosure = $attachment->getEnclosure(); + if ($enclosure) { + $enclosures[] = $enclosure; } } - $tag->free(); - # Enclosures - $attachments = $this->attachments(); - if($attachments){ - foreach($attachments as $attachment){ - $enclosure=$attachment->getEnclosure(); - if ($enclosure) { - $attributes = array('rel'=>'enclosure','href'=>$enclosure->url,'type'=>$enclosure->mimetype,'length'=>$enclosure->size); - if($enclosure->title){ - $attributes['title']=$enclosure->title; - } - $xs->element('link', $attributes, null); + if (Event::handle('StartActivityEnclosures', array(&$this, &$xs, &$enclosures))) { + foreach ($enclosures as $enclosure) { + $attributes = array('rel' => 'enclosure', + 'href' => $enclosure->url, + 'type' => $enclosure->mimetype, + 'length' => $enclosure->size); + + if ($enclosure->title) { + $attributes['title'] = $enclosure->title; } + + $xs->element('link', $attributes, null); } + Event::handle('EndActivityEnclosures', array(&$this, &$xs, $enclosures)); } - if (!empty($this->lat) && !empty($this->lon)) { - $xs->elementStart('geo', array('xmlns:georss' => 'http://www.georss.org/georss')); - $xs->element('georss:point', null, $this->lat . ' ' . $this->lon); - $xs->elementEnd('geo'); + $lat = $this->lat; + $lon = $this->lon; + + if (Event::handle('StartActivityGeo', array(&$this, &$xs, &$lat, &$lon))) { + if (!empty($lat) && !empty($lon)) { + $xs->element('georss:point', null, $lat . ' ' . $lon); + } + Event::handle('EndActivityGeo', array(&$this, &$xs, $lat, $lon)); } - $xs->elementEnd('entry'); + if (Event::handle('StartActivityEnd', array(&$this, &$xs))) { + $xs->elementEnd('entry'); + Event::handle('EndActivityEnd', array(&$this, &$xs)); + } return $xs->getString(); } + /** + * Returns an XML string fragment with a reference to a notice as an + * Activity Streams noun object with the given element type. + * + * Assumes that 'activity' namespace has been previously defined. + * + * @param string $element one of 'subject', 'object', 'target' + * @return string + */ + function asActivityNoun($element) + { + $noun = ActivityObject::fromNotice($this); + return $noun->asString('activity:' . $element); + } + function bestUrl() { if (!empty($this->url)) { @@ -1281,23 +1650,23 @@ class Notice extends Memcached_DataObject } } - function stream($fn, $args, $cachekey, $offset=0, $limit=20, $since_id=0, $max_id=0, $since=null) + function stream($fn, $args, $cachekey, $offset=0, $limit=20, $since_id=0, $max_id=0) { $cache = common_memcache(); if (empty($cache) || - $since_id != 0 || $max_id != 0 || (!is_null($since) && $since > 0) || + $since_id != 0 || $max_id != 0 || is_null($limit) || ($offset + $limit) > NOTICE_CACHE_WINDOW) { return call_user_func_array($fn, array_merge($args, array($offset, $limit, $since_id, - $max_id, $since))); + $max_id))); } $idkey = common_cache_key($cachekey); $idstr = $cache->get($idkey); - if (!empty($idstr)) { + if ($idstr !== false) { // Cache hit! Woohoo! $window = explode(',', $idstr); $ids = array_slice($window, $offset, $limit); @@ -1306,7 +1675,7 @@ class Notice extends Memcached_DataObject $laststr = $cache->get($idkey.';last'); - if (!empty($laststr)) { + if ($laststr !== false) { $window = explode(',', $laststr); $last_id = $window[0]; $new_ids = call_user_func_array($fn, array_merge($args, array(0, NOTICE_CACHE_WINDOW, @@ -1394,6 +1763,10 @@ class Notice extends Memcached_DataObject // Figure out who that is. $sender = Profile::staticGet('id', $profile_id); + if (empty($sender)) { + return null; + } + $recipient = common_relative_profile($sender, $nickname, common_sql_now()); if (empty($recipient)) { @@ -1441,4 +1814,349 @@ class Notice extends Memcached_DataObject return $location; } + + function repeat($repeater_id, $source) + { + $author = Profile::staticGet('id', $this->profile_id); + + // TRANS: Message used to repeat a notice. RT is the abbreviation of 'retweet'. + // TRANS: %1$s is the repeated user's name, %2$s is the repeated notice. + $content = sprintf(_('RT @%1$s %2$s'), + $author->nickname, + $this->content); + + $maxlen = common_config('site', 'textlimit'); + if ($maxlen > 0 && mb_strlen($content) > $maxlen) { + // Web interface and current Twitter API clients will + // pull the original notice's text, but some older + // clients and RSS/Atom feeds will see this trimmed text. + // + // Unfortunately this is likely to lose tags or URLs + // at the end of long notices. + $content = mb_substr($content, 0, $maxlen - 4) . ' ...'; + } + + return self::saveNew($repeater_id, $content, $source, + array('repeat_of' => $this->id)); + } + + // These are supposed to be in chron order! + + function repeatStream($limit=100) + { + $cache = common_memcache(); + + if (empty($cache)) { + $ids = $this->_repeatStreamDirect($limit); + } else { + $idstr = $cache->get(common_cache_key('notice:repeats:'.$this->id)); + if ($idstr !== false) { + $ids = explode(',', $idstr); + } else { + $ids = $this->_repeatStreamDirect(100); + $cache->set(common_cache_key('notice:repeats:'.$this->id), implode(',', $ids)); + } + if ($limit < 100) { + // We do a max of 100, so slice down to limit + $ids = array_slice($ids, 0, $limit); + } + } + + return Notice::getStreamByIds($ids); + } + + function _repeatStreamDirect($limit) + { + $notice = new Notice(); + + $notice->selectAdd(); // clears it + $notice->selectAdd('id'); + + $notice->repeat_of = $this->id; + + $notice->orderBy('created'); // NB: asc! + + if (!is_null($offset)) { + $notice->limit($offset, $limit); + } + + $ids = array(); + + if ($notice->find()) { + while ($notice->fetch()) { + $ids[] = $notice->id; + } + } + + $notice->free(); + $notice = NULL; + + return $ids; + } + + function locationOptions($lat, $lon, $location_id, $location_ns, $profile = null) + { + $options = array(); + + if (!empty($location_id) && !empty($location_ns)) { + + $options['location_id'] = $location_id; + $options['location_ns'] = $location_ns; + + $location = Location::fromId($location_id, $location_ns); + + if (!empty($location)) { + $options['lat'] = $location->lat; + $options['lon'] = $location->lon; + } + + } else if (!empty($lat) && !empty($lon)) { + + $options['lat'] = $lat; + $options['lon'] = $lon; + + $location = Location::fromLatLon($lat, $lon); + + if (!empty($location)) { + $options['location_id'] = $location->location_id; + $options['location_ns'] = $location->location_ns; + } + } else if (!empty($profile)) { + + if (isset($profile->lat) && isset($profile->lon)) { + $options['lat'] = $profile->lat; + $options['lon'] = $profile->lon; + } + + if (isset($profile->location_id) && isset($profile->location_ns)) { + $options['location_id'] = $profile->location_id; + $options['location_ns'] = $profile->location_ns; + } + } + + return $options; + } + + function clearReplies() + { + $replyNotice = new Notice(); + $replyNotice->reply_to = $this->id; + + //Null any notices that are replies to this notice + + if ($replyNotice->find()) { + while ($replyNotice->fetch()) { + $orig = clone($replyNotice); + $replyNotice->reply_to = null; + $replyNotice->update($orig); + } + } + + // Reply records + + $reply = new Reply(); + $reply->notice_id = $this->id; + + if ($reply->find()) { + while($reply->fetch()) { + self::blow('reply:stream:%d', $reply->profile_id); + $reply->delete(); + } + } + + $reply->free(); + } + + function clearRepeats() + { + $repeatNotice = new Notice(); + $repeatNotice->repeat_of = $this->id; + + //Null any notices that are repeats of this notice + + if ($repeatNotice->find()) { + while ($repeatNotice->fetch()) { + $orig = clone($repeatNotice); + $repeatNotice->repeat_of = null; + $repeatNotice->update($orig); + } + } + } + + function clearFaves() + { + $fave = new Fave(); + $fave->notice_id = $this->id; + + if ($fave->find()) { + while ($fave->fetch()) { + self::blow('fave:ids_by_user_own:%d', $fave->user_id); + self::blow('fave:ids_by_user_own:%d;last', $fave->user_id); + self::blow('fave:ids_by_user:%d', $fave->user_id); + self::blow('fave:ids_by_user:%d;last', $fave->user_id); + $fave->delete(); + } + } + + $fave->free(); + } + + function clearTags() + { + $tag = new Notice_tag(); + $tag->notice_id = $this->id; + + if ($tag->find()) { + while ($tag->fetch()) { + self::blow('profile:notice_ids_tagged:%d:%s', $this->profile_id, common_keyize($tag->tag)); + self::blow('profile:notice_ids_tagged:%d:%s;last', $this->profile_id, common_keyize($tag->tag)); + self::blow('notice_tag:notice_ids:%s', common_keyize($tag->tag)); + self::blow('notice_tag:notice_ids:%s;last', common_keyize($tag->tag)); + $tag->delete(); + } + } + + $tag->free(); + } + + function clearGroupInboxes() + { + $gi = new Group_inbox(); + + $gi->notice_id = $this->id; + + if ($gi->find()) { + while ($gi->fetch()) { + self::blow('user_group:notice_ids:%d', $gi->group_id); + $gi->delete(); + } + } + + $gi->free(); + } + + function distribute() + { + // We always insert for the author so they don't + // have to wait + + $user = User::staticGet('id', $this->profile_id); + if (!empty($user)) { + Inbox::insertNotice($user->id, $this->id); + } + + if (common_config('queue', 'inboxes')) { + // If there's a failure, we want to _force_ + // distribution at this point. + try { + $qm = QueueManager::get(); + $qm->enqueue($this, 'distrib'); + } catch (Exception $e) { + // If the exception isn't transient, this + // may throw more exceptions as DQH does + // its own enqueueing. So, we ignore them! + try { + $handler = new DistribQueueHandler(); + $handler->handle($this); + } catch (Exception $e) { + common_log(LOG_ERR, "emergency redistribution resulted in " . $e->getMessage()); + } + // Re-throw so somebody smarter can handle it. + throw $e; + } + } else { + $handler = new DistribQueueHandler(); + $handler->handle($this); + } + } + + function insert() + { + $result = parent::insert(); + + if ($result) { + // Profile::hasRepeated() abuses pkeyGet(), so we + // have to clear manually + if (!empty($this->repeat_of)) { + $c = self::memcache(); + if (!empty($c)) { + $ck = self::multicacheKey('Notice', + array('profile_id' => $this->profile_id, + 'repeat_of' => $this->repeat_of)); + $c->delete($ck); + } + } + } + + return $result; + } + + /** + * Get the source of the notice + * + * @return Notice_source $ns A notice source object. 'code' is the only attribute + * guaranteed to be populated. + */ + function getSource() + { + $ns = new Notice_source(); + if (!empty($this->source)) { + switch ($this->source) { + case 'web': + case 'xmpp': + case 'mail': + case 'omb': + case 'system': + case 'api': + $ns->code = $this->source; + break; + default: + $ns = Notice_source::staticGet($this->source); + if (!$ns) { + $ns = new Notice_source(); + $ns->code = $this->source; + $app = Oauth_application::staticGet('name', $this->source); + if ($app) { + $ns->name = $app->name; + $ns->url = $app->source_url; + } + } + break; + } + } + return $ns; + } + + /** + * Determine whether the notice was locally created + * + * @return boolean locality + */ + + public function isLocal() + { + return ($this->is_local == Notice::LOCAL_PUBLIC || + $this->is_local == Notice::LOCAL_NONPUBLIC); + } + + public function getTags() + { + $tags = array(); + $tag = new Notice_tag(); + $tag->notice_id = $this->id; + if ($tag->find()) { + while ($tag->fetch()) { + $tags[] = $tag->tag; + } + } + $tag->free(); + return $tags; + } + + static private function utcDate($dt) + { + $dateStr = date('d F Y H:i:s', strtotime($dt)); + $d = new DateTime($dateStr, new DateTimeZone('UTC')); + return $d->format(DATE_W3C); + } }