mackuba.eu
Don't forget to lycansubscribe
import likes & other things in the firehose client
+53
-1
app/firehose_client.rb
+53
-1
app/firehose_client.rb
···
1
1
require 'skyfall'
2
2
3
3
require_relative 'init'
4
+
require_relative 'models/post'
4
5
require_relative 'models/subscription'
6
+
require_relative 'models/user'
5
7
6
8
class FirehoseClient
7
9
attr_accessor :start_cursor, :service
···
15
17
16
18
def start
17
19
return if @sky
20
+
21
+
@active_users = load_users
18
22
19
23
log "Starting firehose process (YJIT = #{RubyVM::YJIT.enabled? ? 'on' : 'off'})"
20
24
···
94
98
Subscription.where(service: @service).update_all(cursor: cursor)
95
99
end
96
100
101
+
def load_users
102
+
User.active.map { |u| [u.did, u] }.then { |list| Hash[list] }
103
+
end
104
+
97
105
def process_message(msg)
98
106
save_cursor(msg.seq) if msg.seq % 1000 == 0
99
107
···
108
116
@replaying = false
109
117
end
110
118
119
+
@current_user = @active_users[msg.repo]
120
+
111
121
msg.operations.each do |op|
112
122
case op.type
123
+
when :bsky_like
124
+
process_like(msg, op)
125
+
when :bsky_repost
126
+
process_repost(msg, op)
113
127
when :bsky_post
114
-
# ...
128
+
process_post(msg, op)
115
129
end
116
130
end
117
131
end
···
120
134
def process_account_event(msg)
121
135
if msg.status == :deleted
122
136
# ...
137
+
end
138
+
end
139
+
140
+
def process_like(msg, op)
141
+
return unless @current_user
142
+
143
+
if op.action == :create
144
+
@current_user.likes.import_from_record(op.uri, op.raw_record)
145
+
elsif op.action == :delete
146
+
@current_user.likes.where(rkey: op.rkey).delete_all
147
+
end
148
+
end
149
+
150
+
def process_repost(msg, op)
151
+
return unless @current_user
152
+
153
+
if op.action == :create
154
+
@current_user.reposts.import_from_record(op.uri, op.raw_record)
155
+
elsif op.action == :delete
156
+
@current_user.reposts.where(rkey: op.rkey).delete_all
157
+
end
158
+
end
159
+
160
+
def process_post(msg, op)
161
+
if op.action == :create
162
+
if @current_user
163
+
@current_user.quotes.import_from_record(op.uri, op.raw_record)
164
+
@current_user.pins.import_from_record(op.uri, op.raw_record)
165
+
end
166
+
elsif op.action == :delete
167
+
if @current_user
168
+
@current_user.quotes.where(rkey: op.rkey).delete_all
169
+
@current_user.pins.where(rkey: op.rkey).delete_all
170
+
end
171
+
172
+
if post = Post.find_by_at_uri(op.uri)
173
+
post.destroy
174
+
end
123
175
end
124
176
end
125
177