add framework for post update pushes to pubsub

app/models/post.rb

@@ -1383,15 +1383,9 @@ class Post < ActiveRecord::Base
     end
 
     def notify_pubsub
-      return
       return unless Danbooru.config.google_api_project
 
-      pubsub = Google::Apis::PubsubV1::PubsubService.new
-      pubsub.authorization = Google::Auth.get_application_default([Google::Apis::PubsubV1::AUTH_PUBSUB])
-      topic = "projects/#{Danbooru.config.google_api_project}/topics/post_updates"
-      request = Google::Apis::PubsubV1::PublishRequest.new(messages: [])
-      request.messages << Google::Apis::PubsubV1::Message.new(data: id.to_s)
-      pubsub.publish_topic(topic, request)
+      PostUpdate.insert(id)
     end
   end
 

app/models/post_update.rb

@@ -0,0 +1,23 @@
+class PostUpdate
+  def self.insert(post_id)
+    ActiveRecord::Base.execute_sql("insert into post_updates (post_id) values (?) on conflict do nothing", post_id)
+  end
+
+  def self.get
+    ActiveRecord::Base.select_values_sql("delete from post_updates returning post_id")
+  end
+
+  def self.push
+    return unless Danbooru.config.google_api_project
+
+    pubsub = Google::Apis::PubsubV1::PubsubService.new
+    pubsub.authorization = Google::Auth.get_application_default([Google::Apis::PubsubV1::AUTH_PUBSUB])
+    topic = "projects/#{Danbooru.config.google_api_project}/topics/post_updates"
+    post_ids = get()
+
+    post_ids.in_groups_of(1_000, false).each do |group|
+      request = Google::Apis::PubsubV1::PublishRequest.new(messages: group.map {|x| Google::Apis::PubsubV1::Message.new(data: x.to_s)})
+      pubsub.publish_topic(topic, request)
+    end
+  end
+end

config/schedule.rb

@@ -24,6 +24,10 @@ every 1.month, :at => "2:00 am" do
 end
 
 if environment == "production"
+  every 1.hour do
+    runner "PostUpdate.push"
+  end
+
   every 1.hour do
     runner "AmazonBackup.execute"
   end

db/migrate/20160820003534_create_post_updates.rb

@@ -0,0 +1,9 @@
+class CreatePostUpdates < ActiveRecord::Migration
+  def up
+    execute "create unlogged table post_updates ( post_id integer, constraint unique_post_id unique(post_id) )"
+  end
+
+  def down
+    execute "drop table post_updates"
+  end
+end

db/structure.sql

@@ -2654,6 +2654,15 @@ CREATE SEQUENCE post_flags_id_seq
 ALTER SEQUENCE post_flags_id_seq OWNED BY post_flags.id;
 
 
+--
+-- Name: post_updates; Type: TABLE; Schema: public; Owner: -
+--
+
+CREATE UNLOGGED TABLE post_updates (
+    post_id integer
+);
+
+
 --
 -- Name: post_versions; Type: TABLE; Schema: public; Owner: -
 --
@@ -4750,6 +4759,14 @@ ALTER TABLE ONLY transaction_log_items
     ADD CONSTRAINT transaction_log_items_pkey PRIMARY KEY (id);
 
 
+--
+-- Name: unique_post_id; Type: CONSTRAINT; Schema: public; Owner: -
+--
+
+ALTER TABLE ONLY post_updates
+    ADD CONSTRAINT unique_post_id UNIQUE (post_id);
+
+
 --
 -- Name: uploads_pkey; Type: CONSTRAINT; Schema: public; Owner: -
 --
@@ -7391,3 +7408,5 @@ INSERT INTO schema_migrations (version) VALUES ('20160222211328');
 
 INSERT INTO schema_migrations (version) VALUES ('20160526174848');
 
+INSERT INTO schema_migrations (version) VALUES ('20160820003534');
+