hub: implement atproto-commit task handler, other misc setup

app.py

@@ -6,7 +6,7 @@ registered.
 from flask_app import app
 
 # import all modules to register their Flask handlers
-import activitypub, convert, follow, pages, redirect, superfeedr, ui, webfinger, web
+import activitypub, atproto, convert, follow, pages, redirect, superfeedr, ui, webfinger, web
 
 import models
 models.reset_protocol_properties()

config.py

@@ -1,7 +1,9 @@
-"""Flask config.
+"""Flask config and env vars.
 
 https://flask.palletsprojects.com/en/latest/config/
 """
+import os
+
 from oauth_dropins.webutil import appengine_info, util
 
 # This is primarily for flashed messages, since we don't use session data
@@ -21,3 +23,7 @@ else:
     ENV = 'production'
     CACHE_TYPE = 'SimpleCache'
     SECRET_KEY = util.read('flask_secret_key')
+
+os.environ.setdefault('APPVIEW_HOST', 'api.bsky-sandbox.dev')
+os.environ.setdefault('BGS_HOST', 'bgs.bsky-sandbox.dev')
+os.environ.setdefault('PLC_HOST', 'plc.bsky-sandbox.dev')

hub.py

@@ -6,9 +6,10 @@ import os
 from pathlib import Path
 
 import arroba.server
+from arroba.datastore_storage import DatastoreStorage
 from arroba.util import service_jwt
 from arroba import xrpc_sync
-from flask import Flask
+from flask import Flask, request
 import google.cloud.logging
 import lexrpc.server
 import lexrpc.flask_server
@@ -27,23 +28,6 @@ util.set_user_agent(USER_AGENT)
 logger = logging.getLogger(__name__)
 
 
-# #
-# # App Engine config
-# #
-# if appengine_info.LOCAL:
-#     logger.info('Running locally')
-#     # creds = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS')
-#     # assert not creds or creds.endswith('fake_user_account.json')
-#     # os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = \
-#     #     os.path.join(os.path.dirname(__file__), 'fake_user_account.json')
-#     # os.environ.setdefault('CLOUDSDK_CORE_PROJECT', 'app')
-#     # os.environ.setdefault('DATASTORE_DATASET', 'app')
-#     # os.environ.setdefault('GOOGLE_CLOUD_PROJECT', 'app')
-#     # os.environ.setdefault('DATASTORE_EMULATOR_HOST', 'localhost:8089')
-# else:
-#     logger.info('Running against production GAE')
-
-
 #
 # Flask app
 #
@@ -64,11 +48,11 @@ app.wsgi_app = flask_util.ndb_context_middleware(
 @app.get('/liveness_check')
 @app.get('/readiness_check')
 def health_check():
-  """App Engine Flex health checks.
+    """App Engine Flex health checks.
 
-  https://cloud.google.com/appengine/docs/flexible/reference/app-yaml?tab=python#updated_health_checks
+    https://cloud.google.com/appengine/docs/flexible/reference/app-yaml?tab=python#updated_health_checks
-  """
+    """
-  return 'OK'
+    return 'OK'
 
 
 #
@@ -77,17 +61,19 @@ def health_check():
 lexrpc.flask_server.init_flask(arroba.server.server, app)
 
 
-# # requestCrawl in prod
+@app.post('/_ah/queue/atproto-commit')
-# # not working because the BGS immediately tries to connect and errors if it can't,
+def atproto_commit():
-# # and evidently we're not quite serving subscribeRepos here yet. not sure why not.
+    """Handler for atproto-commit tasks. Enqueues the commit for subscribeRepos.
-# if is_prod:
+    """
-#     url = f'https://{os.environ["BGS_HOST"]}/xrpc/com.atproto.sync.requestCrawl'
+    logger.info(f'Params: {request.values}')
-#     logger.info(f'Fetching {url}')
+    seq = request.form['seq']
-#     jwt = service_jwt(os.environ['BGS_HOST'], TODO repo did, TODO repo key)
+    logger.info(f'Got atproto-commit task for seq {seq}')
-#     resp = requests.get(url, params={'hostname': os.environ['PDS_HOST']},
+
-#                         headers={'User-Agent': USER_AGENT,
+    if not util.is_int(seq):
-#                                  'Authorization': f'Bearer {jwt}',
+        flask_util.error(f'seq {seq} is not int')
-#                                 })
+
-#     logger.info(resp.content)
+    for commit_data in DatastoreStorage().read_commits_by_seq(start=int(seq)):
-#     resp.raise_for_status()
+        logger.info(f'Enqueueing commit {commit_data.commit.cid}')
-#     logger.info('OK')
+        xrpc_sync.enqueue_commit(commit_data)
+
+    return 'OK'