6 files changed, 67 insertions, 6 deletions
diff --git a/app/services/backup_service.rb b/app/services/backup_service.rb
index 4cfa22ab8..5fcc98057 100644
--- a/app/services/backup_service.rb
+++ b/app/services/backup_service.rb
@@ -45,6 +45,7 @@ class BackupService < BaseService
           dump_media_attachments!(tar)
           dump_outbox!(tar)
           dump_likes!(tar)
+          dump_bookmarks!(tar)
           dump_actor!(tar)
         end
       end
@@ -85,6 +86,7 @@ class BackupService < BaseService
     actor[:image][:url] = 'header' + File.extname(actor[:image][:url]) if actor[:image]
     actor[:outbox]      = 'outbox.json'
     actor[:likes]       = 'likes.json'
+    actor[:bookmarks]   = 'bookmarks.json'
 
     download_to_tar(tar, account.avatar, 'avatar' + File.extname(account.avatar.path)) if account.avatar.exists?
     download_to_tar(tar, account.header, 'header' + File.extname(account.header.path)) if account.header.exists?
@@ -115,6 +117,25 @@ class BackupService < BaseService
     end
   end
 
+  def dump_bookmarks!(tar)
+    collection = serialize(ActivityPub::CollectionPresenter.new(id: 'bookmarks.json', type: :ordered, size: 0, items: []), ActivityPub::CollectionSerializer)
+
+    Status.reorder(nil).joins(:bookmarks).includes(:account).merge(account.bookmarks).find_in_batches do |statuses|
+      statuses.each do |status|
+        collection[:totalItems] += 1
+        collection[:orderedItems] << ActivityPub::TagManager.instance.uri_for(status)
+      end
+
+      GC.start
+    end
+
+    json = Oj.dump(collection)
+
+    tar.add_file_simple('bookmarks.json', 0o444, json.bytesize) do |io|
+      io.write(json)
+    end
+  end
+
   def collection_presenter
     ActivityPub::CollectionPresenter.new(
       id: 'outbox.json',
diff --git a/app/services/batched_remove_status_service.rb b/app/services/batched_remove_status_service.rb
index cd3b14e08..d78f506c6 100644
--- a/app/services/batched_remove_status_service.rb
+++ b/app/services/batched_remove_status_service.rb
@@ -44,6 +44,7 @@ class BatchedRemoveStatusService < BaseService
     # Cannot be batched
     statuses.each do |status|
       unpush_from_public_timelines(status)
+      unpush_from_direct_timelines(status) if status.direct_visibility?
       batch_salmon_slaps(status) if status.local?
     end
 
@@ -100,6 +101,16 @@ class BatchedRemoveStatusService < BaseService
     end
   end
 
+  def unpush_from_direct_timelines(status)
+    payload = @json_payloads[status.id]
+    redis.pipelined do
+      @mentions[status.id].each do |mention|
+        redis.publish("timeline:direct:#{mention.account.id}", payload) if mention.account.local?
+      end
+      redis.publish("timeline:direct:#{status.account.id}", payload) if status.account.local?
+    end
+  end
+
   def batch_salmon_slaps(status)
     return if @mentions[status.id].empty?
 
diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb
index f3e9c855d..de7c031d8 100644
--- a/app/services/fan_out_on_write_service.rb
+++ b/app/services/fan_out_on_write_service.rb
@@ -6,14 +6,17 @@ class FanOutOnWriteService < BaseService
   def call(status)
     raise Mastodon::RaceConditionError if status.visibility.nil?
 
+    deliver_to_self(status) if status.account.local?
+
     render_anonymous_payload(status)
 
     if status.direct_visibility?
+      deliver_to_mentioned_followers(status)
+      deliver_to_direct_timelines(status)
       deliver_to_own_conversation(status)
     elsif status.limited_visibility?
       deliver_to_mentioned_followers(status)
     else
-      deliver_to_self(status) if status.account.local?
       deliver_to_followers(status)
       deliver_to_lists(status)
     end
@@ -91,6 +94,16 @@ class FanOutOnWriteService < BaseService
     Redis.current.publish('timeline:public:local:media', @payload) if status.local?
   end
 
+  def deliver_to_direct_timelines(status)
+    Rails.logger.debug "Delivering status #{status.id} to direct timelines"
+
+    status.mentions.includes(:account).each do |mention|
+      Redis.current.publish("timeline:direct:#{mention.account.id}", @payload) if mention.account.local?
+    end
+
+    Redis.current.publish("timeline:direct:#{status.account.id}", @payload) if status.account.local?
+  end
+
   def deliver_to_own_conversation(status)
     AccountConversation.add_status(status.account, status)
   end
diff --git a/app/services/post_status_service.rb b/app/services/post_status_service.rb
index 686b10c58..cfb266fbb 100644
--- a/app/services/post_status_service.rb
+++ b/app/services/post_status_service.rb
@@ -46,7 +46,10 @@ class PostStatusService < BaseService
   private
 
   def preprocess_attributes!
-    @text         = @options.delete(:spoiler_text) if @text.blank? && @options[:spoiler_text].present?
+    if @text.blank? && @options[:spoiler_text].present?
+     @text = '.'
+     @text = @media.find(&:video?) ? '📹' : '🖼' if @media.size > 0
+    end
     @visibility   = @options[:visibility] || @account.user&.setting_default_privacy
     @visibility   = :unlisted if @visibility == :public && @account.silenced
     @scheduled_at = @options[:scheduled_at]&.to_datetime
@@ -86,8 +89,10 @@ class PostStatusService < BaseService
   def postprocess_status!
     LinkCrawlWorker.perform_async(@status.id) unless @status.spoiler_text?
     DistributionWorker.perform_async(@status.id)
-    Pubsubhubbub::DistributionWorker.perform_async(@status.stream_entry.id)
-    ActivityPub::DistributionWorker.perform_async(@status.id)
+    unless @status.local_only?
+      Pubsubhubbub::DistributionWorker.perform_async(@status.stream_entry.id)
+      ActivityPub::DistributionWorker.perform_async(@status.id)
+    end
   end
 
   def validate_media!
diff --git a/app/services/reblog_service.rb b/app/services/reblog_service.rb
index 33ddef8b8..03db27406 100644
--- a/app/services/reblog_service.rb
+++ b/app/services/reblog_service.rb
@@ -20,8 +20,11 @@ class ReblogService < BaseService
     reblog = account.statuses.create!(reblog: reblogged_status, text: '')
 
     DistributionWorker.perform_async(reblog.id)
-    Pubsubhubbub::DistributionWorker.perform_async(reblog.stream_entry.id)
-    ActivityPub::DistributionWorker.perform_async(reblog.id)
+
+    unless reblogged_status.local_only?
+      Pubsubhubbub::DistributionWorker.perform_async(reblog.stream_entry.id)
+      ActivityPub::DistributionWorker.perform_async(reblog.id)
+    end
 
     create_notification(reblog)
     bump_potential_friendship(account, reblog)
diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb
index 2012f15fc..99c8e6cbb 100644
--- a/app/services/remove_status_service.rb
+++ b/app/services/remove_status_service.rb
@@ -22,6 +22,7 @@ class RemoveStatusService < BaseService
     remove_from_hashtags
     remove_from_public
     remove_from_media if status.media_attachments.any?
+    remove_from_direct if status.direct_visibility?
 
     @status.destroy!
 
@@ -156,4 +157,11 @@ class RemoveStatusService < BaseService
     redis.publish('timeline:public:media', @payload)
     redis.publish('timeline:public:local:media', @payload) if @status.local?
   end
+
+  def remove_from_direct
+    @mentions.each do |mention|
+      Redis.current.publish("timeline:direct:#{mention.account.id}", @payload) if mention.account.local?
+    end
+    Redis.current.publish("timeline:direct:#{@account.id}", @payload) if @account.local?
+  end
 end