Don't forget to lycansubscribe
mackuba.eu
1require 'didkit'
2require 'minisky'
3
4require_relative 'at_uri'
5require_relative 'models/post'
6require_relative 'models/user'
7
8class PostDownloader
9 attr_accessor :report, :stop_when_empty
10
11 def initialize
12 @sky = Minisky.new(ENV['APPVIEW_HOST'] || 'public.api.bsky.app', nil)
13
14 @total_count = 0
15 @oldest_imported = Time.now
16 @account_status_cache = {}
17 end
18
19 def import_from_queue(queue)
20 loop do
21 items = queue.pop_batch
22
23 if items.empty?
24 if @stop_when_empty
25 return
26 else
27 sleep 1
28 next
29 end
30 end
31
32 @report&.update(queue: { length: queue.length })
33
34 process_items(items)
35 end
36 end
37
38 def process_items(items)
39 existing_posts = Post.where(rkey: items.map { |x| AT_URI(x.post_uri).rkey }).to_a
40
41 items.dup.each do |item|
42 if post = existing_posts.detect { |post| post.at_uri == item.post_uri }
43 update_item(item, post)
44 items.delete(item)
45 end
46 end
47
48 return if items.empty?
49
50 begin
51 response = @sky.get_request('app.bsky.feed.getPosts', { uris: items.map(&:post_uri).uniq })
52
53 response['posts'].each do |data|
54 current_items = items.select { |x| x.post_uri == data['uri'] }
55 items -= current_items
56
57 begin
58 post = save_post(data['uri'], data['record'])
59
60 if post.valid?
61 current_items.each { |i| update_item(i, post) }
62 else
63 puts "Invalid post #{data['uri']}: #{post.errors.full_messages.join("; ")}"
64 current_items.each { |i| invalidate_item(i) }
65 end
66 rescue InvalidRecordError => e
67 puts "Error in PostDownloader: #{data['uri']}: #{e.class}: #{e}"
68 current_items.each { |i| i.update!(queue: nil) }
69 end
70 end
71
72 check_missing_items(items)
73 rescue StandardError => e
74 puts "Error in PostDownloader: #{e.class}: #{e}"
75 end
76 end
77
78 def save_post(post_uri, record)
79 did, _, rkey = AT_URI(post_uri)
80
81 author = User.find_or_create_by!(did: did)
82
83 if post = Post.find_by(user: author, rkey: rkey)
84 return post
85 else
86 post = build_post(author, rkey, record)
87 post.save
88 post
89 end
90 end
91
92 def build_post(author, rkey, record)
93 text = record.delete('text')
94 created = record.delete('createdAt')
95
96 record.delete('$type')
97
98 Post.new(
99 user: author,
100 rkey: rkey,
101 time: Time.parse(created),
102 text: text,
103 data: JSON.generate(record)
104 )
105 rescue StandardError
106 raise InvalidRecordError
107 end
108
109 def update_item(item, post)
110 item.update!(post: post, post_uri: nil, queue: nil)
111
112 @total_count += 1
113 @oldest_imported = [@oldest_imported, item.time].min
114
115 @report&.update(downloader: { downloaded_posts: @total_count, oldest_date: @oldest_imported })
116 end
117
118 def invalidate_item(item)
119 @total_count += 1
120 @oldest_imported = [@oldest_imported, item.time].min
121
122 @report&.update(downloader: { downloaded_posts: @total_count, oldest_date: @oldest_imported })
123
124 item.destroy
125 end
126
127 def check_missing_items(items)
128 return if items.empty?
129
130 dids = items.map { |x| AT_URI(x.post_uri).repo }.uniq
131 response = @sky.get_request('app.bsky.actor.getProfiles', { actors: dids })
132 active_dids = response['profiles'].map { |x| x['did'] }
133
134 items.each do |item|
135 did = AT_URI(item.post_uri).repo
136 did_obj = DID.new(did)
137
138 if active_dids.include?(did)
139 # account exists but post doesn't, delete the post reference
140 item.destroy
141 else
142 begin
143 status = if @account_status_cache.has_key?(did) # don't retry if status was nil
144 @account_status_cache[did]
145 else
146 @account_status_cache[did] ||= did_obj.account_status
147 end
148
149 case status
150 when :active
151 # account is active but wasn't returned in getProfiles, probably was suspended on the AppView
152 puts "#{item.post_uri}: account #{did} exists on the PDS, account must have been taken down"
153 item.destroy
154 when nil
155 # account was deleted, so all posts were deleted too
156 puts "#{item.post_uri}: account #{did} doesn't exist on the PDS, post must have been deleted"
157 item.destroy
158 else
159 # account is inactive/suspended, but could come back, so leave it for now
160 puts "#{item.post_uri}: account #{did} is inactive: #{status}"
161 end
162 rescue StandardError => e
163 hostname = did_obj.document.pds_host rescue "???"
164 puts "#{item.post_uri}: couldn't check account status for #{did} on #{hostname}: #{e.class}: #{e}"
165
166 # delete reference if the account's PDS is the old bsky.social (so it must have been deleted pre Nov 2023)
167 item.destroy if hostname == 'bsky.social'
168 end
169 end
170
171 if !item.destroyed?
172 item.update!(queue: nil)
173 end
174 end
175 end
176end