Something went wrong. Try again.
Documentation website about Ruby gems for ATProto ruby.sdk.blue
atprotocol bluesky ruby atproto
Something went wrong. Try again.
6.5 kB · 274 lines
Ruby
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275#!/usr/bin/env ruby
# Example: send push notifications to a client app about interactions with a given account.## We will monitor the firehose for any events that are relevant to a given user (in practice,# you would track some number of accounts that have push notification tokens listed in your# database) and then notify the configured mobile app push notification service (here we just# print it to stdout).
require 'bundler/setup'require 'didkit'require 'minisky'require 'skyfall'
$handle = ARGV[0]
if $handle.empty? puts "Usage: #{$PROGRAM_NAME} <handle | did>" exit 1end
$monitored_did = DID.resolve_handle($handle).to_s
if $monitored_did.empty? puts "Couldn't resolve handle: #{$handle}" exit 1end
class InvalidURIException < StandardError def initialize(uri) super("Invalid AT URI: #{uri}") endend
class AtURI attr_reader :did, :collection, :rkey
def initialize(uri) if uri =~ /\Aat:\/\/(did:[\w]+:[\w\.\-]+)\/([\w\.]+)\/([\w\-]+)\z/ @did = $1 @collection = $2 @rkey = $3 else raise InvalidURIException, uri end endend
class NotificationEngine def initialize(user_did) @user_did = user_did @appview = Minisky.new('api.bsky.app', nil) end
def connect @sky = Skyfall::Firehose.new('bsky.network', :subscribe_repos)
@sky.on_connect { puts "Connected, monitoring #{@user_did}" } @sky.on_disconnect { puts "Disconnected" } @sky.on_reconnect { puts "Reconnecting..." } @sky.on_error { |e| puts "ERROR: #{e}" }
@sky.on_message do |msg| process_message(msg) end
@sky.connect end
def disconnect @sky.disconnect end
def process_message(msg) # we're only interested in repo commit messages return if msg.type != :commit
# ignore user's own actions return if msg.repo == @user_did
msg.operations.each do |op| next if op.action != :create
begin case op.type when :bsky_post process_post(msg, op) when :bsky_like process_like(msg, op) when :bsky_repost process_repost(msg, op) when :bsky_follow process_follow(msg, op) end rescue StandardError => e puts "Error: #{e}" end end end
# posts
def process_post(msg, op) data = op.raw_record
if reply = data['reply'] # check for replies (direct only) if uri = reply.dig('parent', 'uri') parent_uri = AtURI.new(uri)
if parent_uri.did == @user_did send_reply_notification(msg, op) end end end
if embed = data['embed'] # check for quotes if uri = embed.dig('record', 'uri') quoted_uri = AtURI.new(uri)
if quoted_uri.did == @user_did send_quote_notification(msg, op) end end
# second type of quote (recordWithMedia) if uri = embed.dig('record', 'record', 'uri') quoted_uri = AtURI.new(uri)
if quoted_uri.did == @user_did send_quote_notification(msg, op) end end end
if facets = data['facets'] # check for mentions if facets.any? { |f| f['features']&.any? { |x| x['did'] == @user_did }} send_mention_notification(msg, op) end end end
def send_reply_notification(msg, op) handle = get_user_handle(msg.repo)
send_push("@#{handle} replied:", op.raw_record) end
def send_quote_notification(msg, op) handle = get_user_handle(msg.repo)
send_push("@#{handle} quoted you:", op.raw_record) end
def send_mention_notification(msg, op) handle = get_user_handle(msg.repo)
send_push("@#{handle} mentioned you:", op.raw_record) end
# likes
def process_like(msg, op) data = op.raw_record
if uri = data.dig('subject', 'uri') liked_uri = AtURI.new(uri)
if liked_uri.did == @user_did case liked_uri.collection when 'app.bsky.feed.post' send_post_like_notification(msg, liked_uri) when 'app.bsky.feed.generator' send_feed_like_notification(msg, liked_uri) end end end end
def send_post_like_notification(msg, uri) handle = get_user_handle(msg.repo) post = get_record(uri)
send_push("@#{handle} liked your post", post) end
def send_feed_like_notification(msg, uri) handle = get_user_handle(msg.repo) feed = get_record(uri)
send_push("@#{handle} liked your feed", feed) end
# reposts
def process_repost(msg, op) data = op.raw_record
if uri = data.dig('subject', 'uri') reposted_uri = AtURI.new(uri)
if reposted_uri.did == @user_did && reposted_uri.collection == 'app.bsky.feed.post' send_repost_notification(msg, reposted_uri) end end end
def send_repost_notification(msg, uri) handle = get_user_handle(msg.repo) post = get_record(uri)
send_push("@#{handle} reposted your post", post) end
# follows
def process_follow(msg, op) if op.raw_record['subject'] == @user_did send_follow_notification(msg) end end
def send_follow_notification(msg) handle = get_user_handle(msg.repo)
send_push("@#{handle} followed you", msg.repo) end
# Note: in this example, we're calling the Bluesky AppView to get details about the # person interacting with the user and the post/feed that was liked/reposted etc. # In a real app, you might run into rate limits if you do that, because these requests # will all be sent from the server's IP. # # So you might need to take a different route and send just the info that you have # here in the push notification data (the AT URI / DID) and fetch the details on the # client side, e.g. in a Notification Service Extension on iOS.
def get_user_handle(did) DID.new(did).get_verified_handle
# or, faster but less correct way (doesn't verify the handle): # DID.new(did).document.handles.first end
def get_record(uri) json = @appview.get_request('com.atproto.repo.getRecord', { repo: uri.did, collection: uri.collection, rkey: uri.rkey })
json['value'] end
def send_push(message, data = nil) # send the message to APNS/FCM here puts puts "[#{Time.now}] #{message} #{data&.inspect}" endend
engine = NotificationEngine.new($monitored_did)
# close the connection cleanly on Ctrl+Ctrap("SIGINT") { engine.disconnect }
engine.connect