Merge branch 'redis-stream-cache'

Gemfile

@@ -126,6 +126,7 @@ group :test do
   gem "selenium-webdriver", "~> 2.7.0"
   gem 'webmock', :require => false
   gem 'sqlite3'
+  gem 'mock_redis'
 end
 
 group :development do

Gemfile.lock

@@ -256,6 +256,7 @@ GEM
     mobile-fu (0.2.1)
       rack-mobile-detect
       rails
+    mock_redis (0.2.0)
     moneta (0.6.0)
     mongrel (1.1.5)
       cgi_multipart_eof_fix (>= 2.4)
@@ -506,6 +507,7 @@ DEPENDENCIES
   linecache (= 0.43)
   mini_magick (= 3.2)
   mobile-fu
+  mock_redis
   mongrel
   mysql2 (= 0.2.13)
   newrelic_rpm

app/models/post.rb

@@ -138,6 +138,10 @@ class Post < ActiveRecord::Base
     false
   end
 
+  def triggers_caching?
+    true
+  end
+
   def comment_email_subject
     I18n.t('notifier.a_post_you_shared')
   end

lib/diaspora/redis_cache.rb

@@ -0,0 +1,82 @@
+#   Copyright (c) 2010-2011, Diaspora Inc.  This file is
+#   licensed under the Affero General Public License version 3 or later.  See
+#   the COPYRIGHT file.
+#
+
+class RedisCache
+
+  SUPPORTED_CACHES = [:created_at] #['updated_at', 
+  CACHE_LIMIT = 100
+
+  def initialize(user, order_field)
+    @user = user
+    @order_field = order_field.to_s
+  end
+
+  # @return [Boolean]
+  def cache_exists?
+    self.size != 0
+  end
+
+  # @return [Integer] the cardinality of the redis set
+  def size
+    redis.zcard(set_key)
+  end
+
+  def post_ids(time=Time.now, limit=15)
+    post_ids = redis.zrevrangebyscore(set_key, time.to_i, "-inf")
+    post_ids[0...limit]
+  end
+
+  def ensure_populated!
+    self.repopulate! unless cache_exists?
+  end
+
+  def repopulate!
+    self.populate! && self.trim!
+  end
+
+  def populate!
+    # user executes query and gets back hashes
+    sql = @user.visible_posts_sql(:limit => CACHE_LIMIT, :order => self.order)
+    hashes = Post.connection.select_all(sql)
+
+    # hashes are inserted into set in a single transaction
+    redis.multi do
+      hashes.each do |h|
+        self.redis.zadd(set_key, h[@order_field].to_i, h["id"])
+      end
+    end
+  end
+
+  def trim!
+    self.redis.zremrangebyrank(set_key, 0, -(CACHE_LIMIT+1))
+  end
+
+  # @param order [Symbol, String]
+  # @return [Boolean]
+  def self.supported_order?(order)
+    SUPPORTED_CACHES.include?(order.to_sym)
+  end
+
+  def order
+    "#{@order_field} DESC"
+  end
+
+  def add(score, id)
+    return unless self.cache_exists?
+    self.redis.zadd(set_key, score.to_i, id)
+    self.trim!
+  end
+
+  protected
+  # @return [Redis]
+  def redis
+    @redis ||= Redis.new
+  end
+
+  # @return [String]
+  def set_key
+    @set_key ||= "cache_stream_#{@user.id}_#{@order_field}"
+  end
+end

lib/diaspora/user/querying.rb

@@ -2,6 +2,8 @@
 #   licensed under the Affero General Public License version 3 or later.  See
 #   the COPYRIGHT file.
 
+require 'lib/diaspora/redis_cache'
+
 module Diaspora
   module UserModules
     module Querying
@@ -13,21 +15,37 @@ module Diaspora
         post ||= Post.where(key => id, :public => true).where(opts).first
       end
 
-      def visible_posts(opts = {})
+      def visible_posts(opts={})
-        defaults = {
+        opts = prep_opts(opts)
-          :type => ['StatusMessage', 'Photo'],
+        post_ids = visible_post_ids(opts)
-          :order => 'updated_at DESC',
+        Post.where(:id => post_ids).select('DISTINCT posts.*').limit(opts[:limit]).order(opts[:order_with_table])
-          :limit => 15,
+      end
-          :hidden => false
-        }
-        opts = defaults.merge(opts)
 
-        order_field = opts[:order].split.first.to_sym
+      def visible_post_ids(opts={})
-        order_with_table = 'posts.' + opts[:order]
+        opts = prep_opts(opts)
 
-        opts[:max_time] = Time.at(opts[:max_time]) if opts[:max_time].is_a?(Integer)
+        if AppConfig[:redis_cache] && RedisCache.supported_order?(opts[:order_field]) && opts[:all_aspects?].present?
-        opts[:max_time] ||= Time.now + 1
+          cache = RedisCache.new(self, opts[:order_field])
 
+          cache.ensure_populated!
+          post_ids = cache.post_ids(opts[:max_time], opts[:limit])
+        end
+
+        if post_ids.blank? || post_ids.length < opts[:limit]
+          visible_ids_from_sql(opts)
+        else
+          post_ids
+        end
+      end
+
+      # @return [Array<Integer>]
+      def visible_ids_from_sql(opts={})
+        opts = prep_opts(opts)
+        Post.connection.select_values(visible_posts_sql(opts))
+      end
+
+      def visible_posts_sql(opts={})
+        opts = prep_opts(opts)
         select_clause ='DISTINCT posts.id, posts.updated_at AS updated_at, posts.created_at AS created_at'
 
         posts_from_others = Post.joins(:contacts).where( :pending => false, :type => opts[:type], :post_visibilities => {:hidden => opts[:hidden]}, :contacts => {:user_id => self.id})
@@ -39,20 +57,10 @@ module Diaspora
           posts_from_self = posts_from_self.joins(:aspect_visibilities).where(:aspect_visibilities => {:aspect_id => opts[:by_members_of]})
         end
 
-        unless sqlite? 
+        posts_from_others = posts_from_others.select(select_clause).order(opts[:order_with_table]).where(Post.arel_table[opts[:order_field]].lt(opts[:max_time]))
-          posts_from_others = posts_from_others.select(select_clause).order(order_with_table).where(Post.arel_table[order_field].lt(opts[:max_time]))
+        posts_from_self = posts_from_self.select(select_clause).order(opts[:order_with_table]).where(Post.arel_table[opts[:order_field]].lt(opts[:max_time]))
-          posts_from_self = posts_from_self.select(select_clause).order(order_with_table).where(Post.arel_table[order_field].lt(opts[:max_time]))
 
-          all_posts = "(#{posts_from_others.to_sql} LIMIT #{opts[:limit]}) UNION ALL (#{posts_from_self.to_sql} LIMIT #{opts[:limit]}) ORDER BY #{opts[:order]} LIMIT #{opts[:limit]}"
+        "(#{posts_from_others.to_sql} LIMIT #{opts[:limit]}) UNION ALL (#{posts_from_self.to_sql} LIMIT #{opts[:limit]}) ORDER BY #{opts[:order]} LIMIT #{opts[:limit]}"
-        else
-          posts_from_others = posts_from_others.select(select_clause)
-          posts_from_self = posts_from_self.select(select_clause)
-          all_posts = "#{posts_from_others.to_sql} UNION ALL #{posts_from_self.to_sql} ORDER BY #{opts[:order]} LIMIT #{opts[:limit]}"
-        end
-
-        post_ids = Post.connection.select_values(all_posts)
-
-        Post.where(:id => post_ids).select('DISTINCT posts.*').limit(opts[:limit]).order(order_with_table)
       end
 
       def contact_for(person)
@@ -112,6 +120,26 @@ module Diaspora
 
         Post.where(:id => post_ids, :pending => false).select('DISTINCT posts.*').order("posts.created_at DESC")
       end
+
+      protected
+
+      # @return [Hash]
+      def prep_opts(opts)
+        defaults = {
+          :type => ['StatusMessage', 'Photo'],
+          :order => 'updated_at DESC',
+          :limit => 15,
+          :hidden => false
+        }
+        opts = defaults.merge(opts)
+
+        opts[:order_field] = opts[:order].split.first.to_sym
+        opts[:order_with_table] = 'posts.' + opts[:order]
+
+        opts[:max_time] = Time.at(opts[:max_time]) if opts[:max_time].is_a?(Integer)
+        opts[:max_time] ||= Time.now + 1
+        opts
+      end
     end
   end
 end

lib/postzord/receiver.rb

@@ -0,0 +1,21 @@
+#   Copyright (c) 2010-2011, Diaspora Inc.  This file is
+#   licensed under the Affero General Public License version 3 or later.  See
+#   the COPYRIGHT file.
+
+
+class Postzord::Receiver
+  require File.join(Rails.root, 'lib/postzord/receiver/private')
+  require File.join(Rails.root, 'lib/postzord/receiver/public')
+
+  def perform!
+    receive!
+    update_cache! if cache?
+  end
+
+  # @return [Boolean]
+  def cache?
+    self.respond_to?(:update_cache!) && AppConfig[:redis_cache] &&
+    @object.respond_to?(:triggers_caching?) && @object.triggers_caching?
+  end
+end
+

lib/postzord/receiver/local_batch.rb

@@ -1,6 +1,9 @@
-module Postzord
+#   Copyright (c) 2010-2011, Diaspora Inc.  This file is
-  module Receiver
+#   licensed under the Affero General Public License version 3 or later.  See
-    class LocalBatch
+#   the COPYRIGHT file.
+
+class Postzord::Receiver::LocalBatch < Postzord::Receiver
+
   attr_reader :object, :recipient_user_ids, :users
 
   def initialize(object, recipient_user_ids)
@@ -9,7 +12,7 @@ module Postzord
     @users = User.where(:id => @recipient_user_ids)
   end
 
-      def perform!
+  def receive!
     if @object.respond_to?(:relayable?)
       receive_relayable
     else
@@ -22,6 +25,13 @@ module Postzord
     notify_users
   end
 
+  def update_cache!
+    @users.each do |user|
+      cache = RedisCache.new(user, "created_at")
+      cache.add(@object.created_at.to_i, @object.id)
+    end
+  end
+
   # NOTE(copied over from receiver public)
   # @return [Object]
   def receive_relayable
@@ -66,6 +76,4 @@ module Postzord
       Notification.notify(user, @object, @object.author)
     end
   end
-    end
-  end
 end

lib/postzord/receiver/private.rb

@@ -1,13 +1,12 @@
 #   Copyright (c) 2010-2011, Diaspora Inc.  This file is
 #   licensed under the Affero General Public License version 3 or later.  See
 #   the COPYRIGHT file.
-#
+
 require File.join(Rails.root, 'lib/webfinger')
 require File.join(Rails.root, 'lib/diaspora/parser')
 
-module Postzord
+class Postzord::Receiver::Private < Postzord::Receiver
-  module Receiver
+
-    class Private
   def initialize(user, opts={})
     @user = user
     @user_person = @user.person
@@ -19,7 +18,7 @@ module Postzord
     @object = opts[:object]
   end
 
-      def perform
+  def receive!
     if @sender && self.salmon.verified_for_key?(@sender.public_key)
       parse_and_receive(salmon.parsed_data)
     else
@@ -49,6 +48,11 @@ module Postzord
     obj
   end
 
+  def update_cache!
+    cache = RedisCache.new(@user, "created_at")
+    cache.add(@object.created_at.to_i, @object.id)
+  end
+
   protected
   def salmon
     @salmon ||= Salmon::EncryptedSlap.from_xml(@salmon_xml, @user)
@@ -112,6 +116,4 @@ module Postzord
       @object.sender_handle = @sender.diaspora_handle
     end
   end
-    end
-  end
 end

lib/postzord/receiver/public.rb

@@ -2,9 +2,8 @@
 #   licensed under the Affero General Public License version 3 or later.  See
 #   the COPYRIGHT file.
 
-module Postzord
+class Postzord::Receiver::Public < Postzord::Receiver
-  module Receiver
+
-    class Public
   attr_accessor :salmon, :author
 
   def initialize(xml)
@@ -18,7 +17,7 @@ module Postzord
   end
 
   # @return [void]
-      def perform!
+  def receive!
     return false unless verified_signature?
     return unless save_object
 
@@ -59,6 +58,4 @@ module Postzord
   def object_can_be_public_and_it_is_not?
     @object.respond_to?(:public) && !@object.public?
   end
-    end
-  end
 end

lib/stream/aspect_stream.rb

@@ -39,7 +39,8 @@ class AspectStream < BaseStream
   # @return [ActiveRecord::Association<Post>] AR association of posts
   def posts
     # NOTE(this should be something like Post.all_for_stream(@user, aspect_ids, {}) that calls visible_posts
-    @posts ||= user.visible_posts(:by_members_of => aspect_ids,
+    @posts ||= user.visible_posts(:all_aspects? => for_all_aspects?,
+                                   :by_members_of => aspect_ids,
                                    :type => TYPES_OF_POST_IN_STREAM,
                                    :order => "#{order} DESC",
                                    :max_time => max_time
@@ -51,6 +52,7 @@ class AspectStream < BaseStream
     @people ||= Person.all_from_aspects(aspect_ids, user).includes(:profile)
   end
 
+  # @return [String] URL
   def link(opts={})
     Rails.application.routes.url_helpers.aspects_path(opts.merge(:a_ids => aspect_ids))
   end
@@ -64,10 +66,20 @@ class AspectStream < BaseStream
     end
   end
 
+  # Only ajax in the stream if all aspects are present.
+  # In this case, we know we're on the first page of the stream,
+  # as the default view for aspects/index is showing posts from
+  # all a user's aspects.
+  #
+  # @return [Boolean] see #for_all_aspects?
   def ajax_stream?
     for_all_aspects?
   end
 
+  # The title that will display at the top of the stream's
+  # publisher box.
+  #
+  # @return [String]
   def title
     if self.for_all_aspects?
       I18n.t('aspects.aspect_stream.stream')
@@ -84,6 +96,9 @@ class AspectStream < BaseStream
     @all_aspects ||= aspect_ids.length == user.aspects.size
   end
 
+  # Provides a translated title for contacts box on the right pane.
+  #
+  # @return [String]
   def contacts_title
     if self.for_all_aspects? || self.aspect_ids.size > 1
       I18n.t('_contacts')
@@ -92,6 +107,10 @@ class AspectStream < BaseStream
     end
   end
 
+  # Provides a link to the user to the contacts page that corresponds with
+  # the stream's active aspects.
+  #
+  # @return [String] Link to contacts
   def contacts_link
     if for_all_aspects? || aspect_ids.size > 1
       Rails.application.routes.url_helpers.contacts_path

spec/integration/attack_vectors_spec.rb

@@ -22,7 +22,7 @@ describe "attack vectors" do
 
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
       expect {
-        zord.perform
+        zord.perform!
       }.should raise_error /not a valid object/
 
       bob.visible_posts.include?(post_from_non_contact).should be_false
@@ -39,7 +39,7 @@ describe "attack vectors" do
     salmon_xml = bob.salmon(original_message).xml_for(alice.person)
     zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
     expect {
-      zord.perform
+      zord.perform!
     }.should raise_error /not a valid object/
 
     alice.reload.visible_posts.should_not include(StatusMessage.find(original_message.id))
@@ -53,12 +53,12 @@ describe "attack vectors" do
         salmon_xml = eve.salmon(original_message).xml_for(bob.person)
 
         zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-        zord.perform
+        zord.perform!
 
         malicious_message = Factory.build(:status_message, :id => original_message.id, :text => 'BAD!!!', :author => alice.person)
         salmon_xml = alice.salmon(malicious_message).xml_for(bob.person)
         zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-        zord.perform
+        zord.perform!
 
         original_message.reload.text.should == "store this!"
       end
@@ -68,14 +68,14 @@ describe "attack vectors" do
 
         salmon_xml =  eve.salmon(original_message).xml_for(bob.person)
         zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-        zord.perform
+        zord.perform!
 
         lambda {
           malicious_message = Factory.build( :status_message, :id => original_message.id, :text => 'BAD!!!', :author => eve.person)
 
           salmon_xml2 = alice.salmon(malicious_message).xml_for(bob.person)
           zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-          zord.perform
+          zord.perform!
 
         }.should_not change{
           bob.reload.visible_posts.count
@@ -97,7 +97,7 @@ describe "attack vectors" do
 
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
       expect {
-        zord.perform
+        zord.perform!
       }.should raise_error /not a valid object/
 
       eve.reload.profile.first_name.should == first_name
@@ -109,7 +109,7 @@ describe "attack vectors" do
 
       salmon_xml = eve.salmon(original_message).xml_for(bob.person)
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-      zord.perform
+      zord.perform!
 
       bob.visible_posts.count.should == 1
       StatusMessage.count.should == 1
@@ -121,7 +121,7 @@ describe "attack vectors" do
 
       salmon_xml = alice.salmon(ret).xml_for(bob.person)
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-      zord.perform
+      zord.perform!
 
       StatusMessage.count.should == 1
       bob.visible_posts.count.should == 1
@@ -143,7 +143,7 @@ describe "attack vectors" do
       proc {
         salmon_xml = alice.salmon(ret).xml_for(bob.person)
         zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-        zord.perform
+        zord.perform!
       }.should_not raise_error
     end
 
@@ -152,7 +152,7 @@ describe "attack vectors" do
 
       salmon_xml = eve.salmon(original_message).xml_for(bob.person)
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-      zord.perform
+      zord.perform!
 
       bob.visible_posts.count.should == 1
 
@@ -164,7 +164,7 @@ describe "attack vectors" do
       salmon_xml = alice.salmon(ret).xml_for(bob.person)
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
       expect {
-        zord.perform
+        zord.perform!
       }.should raise_error /not a valid object/
 
       bob.reload.visible_posts.count.should == 1
@@ -180,7 +180,7 @@ describe "attack vectors" do
         salmon_xml = alice.salmon(ret).xml_for(bob.person)
 
         zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-        zord.perform
+        zord.perform!
 
       }.should_not change{bob.reload.contacts.count}
     end
@@ -196,7 +196,7 @@ describe "attack vectors" do
       salmon_xml = alice.salmon(ret).xml_for(bob.person)
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
       expect {
-        zord.perform
+        zord.perform!
       }.should raise_error /not a valid object/
 
       bob.reload.contacts.count.should == 2
@@ -207,7 +207,7 @@ describe "attack vectors" do
 
       salmon_xml = eve.salmon(original_message).xml_for(bob.person)
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-      zord.perform
+      zord.perform!
 
       original_message.diaspora_handle = alice.diaspora_handle
       original_message.text= "bad bad bad"
@@ -215,7 +215,7 @@ describe "attack vectors" do
       salmon_xml = alice.salmon(original_message).xml_for(bob.person)
 
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-      zord.perform
+      zord.perform!
 
       original_message.reload.text.should == "store this!"
     end

spec/integration/receiving_spec.rb

@@ -334,7 +334,7 @@ describe 'a user receives a post' do
       salmon_xml = salmon.xml_for(bob.person)
 
       zord = Postzord::Receiver::Private.new(bob, :salmon_xml => salmon_xml)
-      zord.perform
+      zord.perform!
 
       bob.visible_posts.include?(post).should be_true
     end

spec/lib/diaspora/redis_cache_spec.rb

@@ -0,0 +1,165 @@
+#   Copyright (c) 2010-2011, Diaspora Inc.  This file is
+#   licensed under the Affero General Public License version 3 or later.  See
+#   the COPYRIGHT file.
+
+require 'spec_helper'
+
+describe RedisCache do
+  before do
+    @redis = MockRedis.new
+    #@redis = Redis.new
+    #@redis.keys.each{|p| @redis.del(p)}
+
+    @cache = RedisCache.new(bob, :created_at)
+    @cache.stub(:redis).and_return(@redis)
+  end
+
+  it 'gets initialized with user and an created_at order' do
+    cache = RedisCache.new(bob, :created_at)
+    [:@user, :@order_field].each do |var|
+      cache.instance_variable_get(var).should_not be_blank
+    end
+  end
+
+  describe "#cache_exists?" do
+    it 'returns true if the sorted set exists' do
+      timestamp = Time.now.to_i
+      @redis.zadd("cache_stream_#{bob.id}_created_at", timestamp, "post_1")
+
+      @cache.cache_exists?.should be_true
+    end
+
+    it 'returns false if there is nothing in the set' do
+      @cache.cache_exists?.should be_false
+    end
+  end
+
+  describe "#post_ids" do
+    before do
+      @timestamps = []
+      @timestamp = Time.now.to_i
+      30.times do |n|
+        created_time = @timestamp - n*1000
+        @redis.zadd("cache_stream_#{bob.id}_created_at", created_time, n.to_s)
+        @timestamps << created_time
+      end
+    end
+
+    it 'returns the most recent post ids (default created at, limit 15)' do
+      @cache.post_ids.should =~ 15.times.map {|n| n.to_s}
+    end
+
+    it 'returns posts ids after the specified time' do
+      @cache.post_ids(@timestamps[15]).should =~ (15...30).map {|n| n.to_s}
+    end
+
+    it 'returns post ids with a non-default limit' do
+      @cache.post_ids(@timestamp, 20).should =~ 20.times.map {|n| n.to_s}
+    end
+  end
+
+  describe "#ensure_populated!" do
+    it 'does nothing if the cache is populated' do
+      @cache.stub(:cache_exists?).and_return(true)
+      @cache.should_not_receive(:repopulate!)
+
+      @cache.ensure_populated!
+    end
+
+    it 'clears and poplulates if the cache is not populated' do
+      @cache.stub(:cache_exists?).and_return(false)
+      @cache.should_receive(:repopulate!)
+
+      @cache.ensure_populated!
+    end
+  end
+
+  describe "#repopulate!" do
+    it 'populates' do
+      @cache.stub(:trim!).and_return(true)
+      @cache.should_receive(:populate!).and_return(true)
+      @cache.repopulate!
+    end
+
+    it 'trims' do
+      @cache.stub(:populate!).and_return(true)
+      @cache.should_receive(:trim!)
+      @cache.repopulate!
+    end
+  end
+
+  describe "#populate!" do
+    it 'queries the db with the visible post sql string' do
+      sql = "long_sql"
+      order = "created_at DESC"
+      @cache.should_receive(:order).and_return(order)
+      bob.should_receive(:visible_posts_sql).with(hash_including(
+                                                    :limit => 100,
+                                                    :order => order)).
+                                             and_return(sql)
+
+      Post.connection.should_receive(:select_all).with(sql).and_return([])
+
+      @cache.populate!
+    end
+
+    it 'adds the post from the hash to the cache'
+  end
+
+  describe "#trim!" do
+    it 'does nothing if the set is smaller than the cache limit' do
+      @timestamps = []
+      @timestamp = Time.now.to_i
+      30.times do |n|
+        created_time = @timestamp - n*1000
+        @redis.zadd("cache_stream_#{bob.id}_created_at", created_time, n.to_s)
+        @timestamps << created_time
+      end
+
+      post_ids = @cache.post_ids(Time.now.to_i, @cache.size)
+      @cache.trim!
+      @cache.post_ids(Time.now.to_i, @cache.size).should == post_ids
+    end
+
+    it 'trims the set to the cache limit' do
+      @timestamps = []
+      @timestamp = Time.now.to_i
+      120.times do |n|
+        created_time = @timestamp - n*1000
+        @redis.zadd("cache_stream_#{bob.id}_created_at", created_time, n.to_s)
+        @timestamps << created_time
+      end
+
+      post_ids = 100.times.map{|n| n.to_s}
+      @cache.trim!
+      @cache.post_ids(Time.now.to_i, @cache.size).should == post_ids[0...100]
+    end
+  end
+
+  describe "#add" do
+    before do
+      @cache.stub(:cache_exists?).and_return(true)
+      @id = 1
+      @score = 123
+    end
+
+    it "adds an id with a given score" do
+      @redis.should_receive(:zadd).with(@cache.send(:set_key), @score, @id)
+      @cache.add(@score, @id)
+    end
+
+    it 'trims' do
+      @cache.should_receive(:trim!)
+      @cache.add(@score, @id)
+    end
+
+    it "doesn't add if the cache does not exist" do
+      @cache.stub(:cache_exists?).and_return(false)
+
+      @redis.should_not_receive(:zadd)
+      @cache.add(@score, @id).should be_false
+    end
+  end
+
+  describe "#remove"
+end

spec/lib/postzord/receiver/local_batch_spec.rb

@@ -17,26 +17,26 @@ describe Postzord::Receiver::LocalBatch do
     end
   end
 
-  describe '#perform!' do
+  describe '#receive!' do
     it 'calls .create_post_visibilities' do
       receiver.should_receive(:create_post_visibilities)
-      receiver.perform!
+      receiver.receive!
     end
 
     it 'sockets to users' do
       pending 'not currently socketing'
       receiver.should_receive(:socket_to_users)
-      receiver.perform!
+      receiver.receive!
     end
 
     it 'notifies mentioned users' do
       receiver.should_receive(:notify_mentioned_users)
-      receiver.perform!
+      receiver.receive!
     end
 
     it 'notifies users' do
       receiver.should_receive(:notify_users)
-      receiver.perform!
+      receiver.receive!
     end
   end
 
@@ -111,4 +111,20 @@ describe Postzord::Receiver::LocalBatch do
       receiver.perform!
     end
   end
+
+  describe '#update_cache!' do
+    it 'adds to a redis cache for receiving_users' do
+      users = [alice, eve]
+      @zord = Postzord::Receiver::LocalBatch.new(@object, users.map{|u| u.id})
+
+      sort_order = "created_at"
+
+      cache = mock
+      RedisCache.should_receive(:new).exactly(users.length).times.with(instance_of(User), sort_order).and_return(cache)
+
+      cache.should_receive(:add).exactly(users.length).times.with(@object.created_at.to_i, @object.id)
+
+      @zord.update_cache!
+    end
+  end
 end

spec/lib/postzord/receiver/private_spec.rb

@@ -47,7 +47,7 @@ describe Postzord::Receiver::Private do
     end
   end
 
-  describe '#perform' do
+  describe '#receive!' do
     before do
       @zord = Postzord::Receiver::Private.new(@user, :salmon_xml => @salmon_xml)
       @salmon = @zord.instance_variable_get(:@salmon)
@@ -56,25 +56,25 @@ describe Postzord::Receiver::Private do
     context 'returns nil' do
       it 'if the salmon author does not exist' do
         @zord.instance_variable_set(:@sender, nil)
-        @zord.perform.should be_nil
+        @zord.perform!.should be_nil
       end
 
       it 'if the author does not match the signature' do
         @zord.instance_variable_set(:@sender, Factory(:person))
-        @zord.perform.should be_nil
+        @zord.perform!.should be_nil
       end
     end
 
     context 'returns the sent object' do
       it 'returns the received object on success' do
-        object = @zord.perform
+        @zord.perform!
-        object.should respond_to(:to_diaspora_xml)
+        @zord.instance_variable_get(:@object).should respond_to(:to_diaspora_xml)
       end
     end
 
     it 'parses the salmon object' do
       Diaspora::Parser.should_receive(:from_xml).with(@salmon.parsed_data).and_return(@original_post)
-      @zord.perform
+      @zord.perform!
     end
   end
 
@@ -86,7 +86,7 @@ describe Postzord::Receiver::Private do
 
     it 'calls Notification.notify if object responds to notification_type' do
       cm = Comment.new
-      cm.stub!(:receive).and_return(cm)
+      cm.stub(:receive).and_return(cm)
 
       Notification.should_receive(:notify).with(@user, cm, @person2)
       zord = Postzord::Receiver::Private.new(@user, :person => @person2, :object => cm)
@@ -103,4 +103,18 @@ describe Postzord::Receiver::Private do
       @zord.receive_object
     end
   end
+
+  describe '#update_cache!' do
+    it 'adds to redis cache for the given user' do
+      @original_post.save!
+
+      @zord = Postzord::Receiver::Private.new(@user, :person => @person2, :object => @original_post)
+
+      sort_order = "created_at"
+      cache = RedisCache.new(@user, sort_order)
+      RedisCache.should_receive(:new).with(@user, sort_order).and_return(cache)
+      cache.should_receive(:add).with(@original_post.created_at.to_i, @original_post.id)
+      @zord.update_cache!
+    end
+  end
 end

spec/lib/postzord/receiver_spec.rb

@@ -0,0 +1,71 @@
+#   Copyright (c) 2011, Diaspora Inc.  This file is
+#   licensed under the Affero General Public License version 3 or later.  See
+#   the COPYRIGHT file.
+
+require 'spec_helper'
+
+require File.join(Rails.root, 'lib/postzord/receiver')
+
+describe Postzord::Receiver do
+  before do
+    @receiver = Postzord::Receiver.new
+  end
+
+  describe "#perform!" do
+    before do
+      @receiver.stub(:receive!)
+    end
+
+    it 'calls receive!' do
+      @receiver.should_receive(:receive!)
+      @receiver.perform!
+    end
+
+    context 'update_cache!' do
+      it "gets called if cache?" do
+        @receiver.stub(:cache?).and_return(true)
+        @receiver.should_receive(:update_cache!)
+        @receiver.perform!
+      end
+
+      it "doesn't get called if !cache?" do
+        @receiver.stub(:cache?).and_return(false)
+        @receiver.should_not_receive(:update_cache!)
+        @receiver.perform!
+      end
+    end
+  end
+
+  describe "#cache?" do
+    before do
+      @receiver.stub(:respond_to?).with(:update_cache!).and_return(true)
+      AppConfig[:redis_cache] = true
+      @receiver.instance_variable_set(:@object, mock(:triggers_caching? => true))
+    end
+
+    it 'returns true if the receiver responds to update_cache and the application has caching enabled' do
+      @receiver.cache?.should be_true
+    end
+
+    it 'returns false if the receiver does not respond to update_cache' do
+      @receiver.stub(:respond_to?).with(:update_cache!).and_return(false)
+      @receiver.cache?.should be_false
+    end
+
+    it 'returns false if the application does not have caching set' do
+      AppConfig[:redis_cache] = false
+      @receiver.cache?.should be_false
+    end
+
+    it 'returns false if the object is does not respond to triggers_caching' do
+      @receiver.instance_variable_set(:@object, mock)
+      @receiver.cache?.should be_false
+    end 
+
+    it 'returns false if the object is not cacheable' do
+      @receiver.instance_variable_set(:@object, mock(:triggers_caching? => false))
+      @receiver.cache?.should be_false
+    end 
+  end
+end
+

spec/lib/stream/aspect_stream_spec.rb

@@ -73,6 +73,14 @@ describe AspectStream do
       @alice.should_receive(:visible_posts).with(hash_including(:max_time => instance_of(Time))).and_return(stub.as_null_object)
       stream.posts
     end
+
+    it 'passes for_all_aspects to visible posts' do
+      stream = AspectStream.new(@alice, [1,2], :max_time => 123)
+      all_aspects = mock
+      stream.stub(:for_all_aspects?).and_return(all_aspects)
+      @alice.should_receive(:visible_posts).with(hash_including(:all_aspects? => all_aspects)).and_return(stub.as_null_object)
+      stream.posts
+    end
   end
 
   describe '#people' do

spec/models/post_spec.rb

@@ -131,4 +131,10 @@ describe Post do
       @post.reload.updated_at.to_i.should == old_time.to_i
     end
   end
+
+  describe "triggers_caching?" do
+    it 'returns true' do
+      Post.new.triggers_caching?.should be_true
+    end
+  end
 end

spec/models/user/querying_spec.rb

@@ -12,49 +12,49 @@ describe User do
     @bobs_aspect = bob.aspects.where(:name => "generic").first
   end
 
-  describe "#visible_posts" do
+  describe "#visible_post_ids" do
     it "contains your public posts" do
       public_post = alice.post(:status_message, :text => "hi", :to => @alices_aspect.id, :public => true)
-      alice.visible_posts.should include(public_post)
+      alice.visible_post_ids.should include(public_post.id)
     end
 
     it "contains your non-public posts" do
       private_post = alice.post(:status_message, :text => "hi", :to => @alices_aspect.id, :public => false)
-      alice.visible_posts.should include(private_post)
+      alice.visible_post_ids.should include(private_post.id)
     end
 
     it "contains public posts from people you're following" do
       dogs = bob.aspects.create(:name => "dogs")
       bobs_public_post = bob.post(:status_message, :text => "hello", :public => true, :to => dogs.id)
-      alice.visible_posts.should include(bobs_public_post)
+      alice.visible_post_ids.should include(bobs_public_post.id)
     end
 
     it "contains non-public posts from people who are following you" do
       bobs_post = bob.post(:status_message, :text => "hello", :to => @bobs_aspect.id)
-      alice.visible_posts.should include(bobs_post)
+      alice.visible_post_ids.should include(bobs_post.id)
     end
 
     it "does not contain non-public posts from aspects you're not in" do
       dogs = bob.aspects.create(:name => "dogs")
       invisible_post = bob.post(:status_message, :text => "foobar", :to => dogs.id)
-      alice.visible_posts.should_not include(invisible_post)
+      alice.visible_post_ids.should_not include(invisible_post.id)
     end
 
     it "does not contain pending posts" do
       pending_post = bob.post(:status_message, :text => "hey", :public => true, :to => @bobs_aspect.id, :pending => true)
       pending_post.should be_pending
-      alice.visible_posts.should_not include pending_post
+      alice.visible_post_ids.should_not include pending_post.id
     end
 
     it "does not contain pending photos" do
       pending_photo = bob.post(:photo, :pending => true, :user_file=> File.open(photo_fixture_name), :to => @bobs_aspect)
-      alice.visible_posts.should_not include pending_photo
+      alice.visible_post_ids.should_not include pending_photo.id
     end
 
     it "respects the :type option" do
       photo = bob.post(:photo, :pending => false, :user_file=> File.open(photo_fixture_name), :to => @bobs_aspect)
-      alice.visible_posts.should include(photo)
+      alice.visible_post_ids.should include(photo.id)
-      alice.visible_posts(:type => 'StatusMessage').should_not include(photo)
+      alice.visible_post_ids(:type => 'StatusMessage').should_not include(photo.id)
     end
 
     it "does not contain duplicate posts" do
@@ -64,8 +64,8 @@ describe User do
 
       bobs_post = bob.post(:status_message, :text => "hai to all my people", :to => [@bobs_aspect.id, bobs_other_aspect.id])
 
-      alice.visible_posts.length.should == 1
+      alice.visible_post_ids.length.should == 1
-      alice.visible_posts.should include(bobs_post)
+      alice.visible_post_ids.should include(bobs_post.id)
     end
 
     describe 'hidden posts' do
@@ -76,15 +76,70 @@ describe User do
       end
 
       it "pulls back non hidden posts" do
-        alice.visible_posts.include?(@status).should be_true
+        alice.visible_post_ids.include?(@status.id).should be_true
       end
 
       it "does not pull back hidden posts" do
         @vis.update_attributes(:hidden => true)
-        alice.visible_posts.include?(@status).should be_false
+        alice.visible_post_ids.include?(@status.id).should be_false
       end
     end
 
+    context "RedisCache" do
+      before do
+        AppConfig[:redis_cache] = true
+        @opts = {:order => "created_at DESC", :all_aspects? => true}
+      end
+
+      after do
+        AppConfig[:redis_cache] = nil
+      end
+
+      it "gets populated with latest 100 posts" do
+        cache = mock(:cache_exists? => true, :ensure_populated! => mock, :post_ids => [])
+        RedisCache.stub(:new).and_return(cache)
+        cache.should_receive(:ensure_populated!)
+
+        alice.visible_post_ids(@opts)
+      end
+
+      it 'does not get used if if all_aspects? option is not present' do
+        RedisCache.should_not_receive(:new)
+
+        alice.visible_post_ids(@opts.merge({:all_aspects? => false}))
+      end
+
+      describe "#ensure_populated_cache" do
+        it 'does nothing if the cache is already populated'
+        it 're-populates the cache with the latest posts (in hashes)'
+      end
+
+      context 'populated cache' do
+        before do
+          @cache = mock(:cache_exists? => true, :ensure_populated! => mock)
+          RedisCache.stub(:new).and_return(@cache)
+        end
+
+        it "reads from the cache" do
+          @cache.should_receive(:post_ids).and_return([1,2,3])
+
+          alice.visible_post_ids(@opts.merge({:limit => 3})).should == [1,2,3]
+        end
+
+        it "queries if maxtime is later than the last cached post" do
+          @cache.stub(:post_ids).and_return([])
+          alice.should_receive(:visible_ids_from_sql)
+
+          alice.visible_post_ids(@opts)
+        end
+
+        it "does not get repopulated" do
+        end
+      end
+    end
+  end
+  
+  describe "#visible_posts" do
     context 'with many posts' do
       before do
         bob.move_contact(eve.person, @bobs_aspect, bob.aspects.create(:name => 'new aspect'))
@@ -101,6 +156,7 @@ describe User do
           end
         end
       end
+
       it 'works' do # The set up takes a looong time, so to save time we do several tests in one
         bob.visible_posts.length.should == 15 #it returns 15 by default
         bob.visible_posts.should == bob.visible_posts(:by_members_of => bob.aspects.map { |a| a.id }) # it is the same when joining through aspects