Something went wrong. Try again.
🌉 A bridge between decentralized social networks fed.brid.gy
Something went wrong. Try again.
3.8 kB · 132 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133"""Single-instance hub for ATProto subscription (firehose) server."""from functools import lru_cachefrom ipaddress import ip_address, ip_networkimport loggingimport osfrom pathlib import Pathimport socketfrom threading import Timer
from arroba import firehoseimport arroba.serverfrom flask import Flask, render_templateimport lexrpc.clientimport lexrpc.flask_serverfrom oauth_dropins.webutil.appengine_info import DEBUG, LOCAL_SERVERfrom oauth_dropins.webutil import appengine_config, flask_utilimport pytz
# all protocolsimport activitypub, atproto, nostr, webimport commonimport models
# as of 2024-07-10BSKY_TEAM_CIDRS = ( # https://discord.com/channels/1097580399187738645/1115973909624397855/1260356452162469969 ip_network('209.249.133.120/29'), ip_network('108.179.139.0/24'), # https://github.com/bluesky-social/atproto/discussions/3036#discussioncomment-11431550 ip_network('67.213.161.32/29'), # https://github.com/bluesky-social/atproto/discussions/3036#discussioncomment-11892019 ip_network('38.120.64.66/32'), # bsky message from bnewbold ip_network('38.142.8.130/32'), ip_network('38.143.58.47/32'),)BSKY_TEAM_HOSTS = ( 'zip.zayo.com', # maybe? https://github.com/bluesky-social/atproto/discussions/3036#discussioncomment-11399854)
logger = logging.getLogger(__name__)
models.reset_protocol_properties()
# start firehose server thread## ...*before* initializing Flask app and request handlers, including health check,# so that we don't go into service and start serving subscribers until the preload# window is loadedif LOCAL_SERVER or not DEBUG: firehose.start()
# Flask appapp = Flask(__name__)app.json.compact = Falseapp_dir = Path(__file__).parentapp.config.from_pyfile(app_dir / 'config.py')
import faulthandlerfaulthandler.enable()import signalfaulthandler.register(signal.SIGTERM)# faulthandler.register(signal.SIGKILL)
app.wsgi_app = flask_util.ndb_context_middleware( app.wsgi_app, client=appengine_config.ndb_client, **common.NDB_CONTEXT_KWARGS)
@app.get('/liveness_check')@app.get('/readiness_check')def health_check(): """App Engine Flex health checks.
https://cloud.google.com/appengine/docs/flexible/reference/app-yaml?tab=python#updated_health_checks """ return 'OK'
# ATProto XRPC serverlexrpc.flask_server.init_flask(arroba.server.server, app)
@app.post('/queue/atproto-commit')@flask_util.cloud_tasks_only(log=False)def atproto_commit(): """Handler for atproto-commit tasks.
Triggers `subscribeRepos` to check for new commits. """ firehose.send_events() return 'OK'
@lru_cachedef gethostbyaddr(addr): """Wrapper for :func:`socket.gethostbyaddr` that caches the result.""" for subnet in BSKY_TEAM_CIDRS: if ip_address(addr) in subnet: return 'bsky'
try: return socket.gethostbyaddr(addr)[0] except socket.herror: return None
@app.get('/admin/atproto')def atproto_admin(): return render_template( 'atproto.html', firehose=firehose, gethostbyaddr=gethostbyaddr, len=len, pytz=pytz, subscribers=lexrpc.flask_server.subscribers, )
# send requestCrawl to relay every 5m.# delay 15s at startup because we're not up and serving XRPCs at this point yet.# not sure why not.if 'GAE_INSTANCE' in os.environ: # prod def request_crawl(): bgs = lexrpc.client.Client(f'https://{os.environ["BGS_HOST"]}', headers={'User-Agent': common.USER_AGENT}) resp = bgs.com.atproto.sync.requestCrawl({'hostname': os.environ['PDS_HOST']}) logger.info(resp) Timer(5 * 60, request_crawl).start()
Timer(15, request_crawl).start() logger.info('Will send relay requestCrawl in 15s')