fix cache expiry, write db migration, integrate did resolver into service proxying

··· 5 from aiohttp import web 6 7 from . import database 8 9 MILLIPDS_DB = web.AppKey("MILLIPDS_DB", database.Database) 10 MILLIPDS_AIOHTTP_CLIENT = web.AppKey( ··· 16 MILLIPDS_FIREHOSE_QUEUES_LOCK = web.AppKey( 17 "MILLIPDS_FIREHOSE_QUEUES_LOCK", asyncio.Lock 18 ) 19 20 21 # these helpers are useful for conciseness and type hinting ··· 35 return req.app[MILLIPDS_FIREHOSE_QUEUES_LOCK] 36 37 38 __all__ = [ 39 "MILLIPDS_DB", 40 "MILLIPDS_AIOHTTP_CLIENT", 41 "MILLIPDS_FIREHOSE_QUEUES", 42 "MILLIPDS_FIREHOSE_QUEUES_LOCK", 43 "get_db", 44 "get_client", 45 "get_firehose_queues", 46 "get_firehose_queues_lock", 47 ]

··· 5 from aiohttp import web 6 7 from . import database 8 + from .did import DIDResolver 9 10 MILLIPDS_DB = web.AppKey("MILLIPDS_DB", database.Database) 11 MILLIPDS_AIOHTTP_CLIENT = web.AppKey( ··· 17 MILLIPDS_FIREHOSE_QUEUES_LOCK = web.AppKey( 18 "MILLIPDS_FIREHOSE_QUEUES_LOCK", asyncio.Lock 19 ) 20 + MILLIPDS_DID_RESOLVER = web.AppKey("MILLIPDS_DID_RESOLVER", DIDResolver) 21 22 23 # these helpers are useful for conciseness and type hinting ··· 37 return req.app[MILLIPDS_FIREHOSE_QUEUES_LOCK] 38 39 40 + def get_did_resolver(req: web.Request): 41 + return req.app[MILLIPDS_DID_RESOLVER] 42 + 43 + 44 __all__ = [ 45 "MILLIPDS_DB", 46 "MILLIPDS_AIOHTTP_CLIENT", 47 "MILLIPDS_FIREHOSE_QUEUES", 48 "MILLIPDS_FIREHOSE_QUEUES_LOCK", 49 + "MILLIPDS_DID_RESOLVER", 50 "get_db", 51 "get_client", 52 "get_firehose_queues", 53 "get_firehose_queues_lock", 54 + "get_did_resolver", 55 ]

+15 -12

src/millipds/appview_proxy.py

··· 12 logger = logging.getLogger(__name__) 13 14 15 - # TODO: this should be done via actual DID resolution, not hardcoded! 16 - SERVICE_ROUTES = { 17 - "did:web:api.bsky.chat#bsky_chat": "https://api.bsky.chat", 18 - "did:web:discover.bsky.app#bsky_fg": "https://discover.bsky.app", 19 - "did:plc:ar7c4by46qjdydhdevvrndac#atproto_labeler": "https://mod.bsky.app", 20 - } 21 - 22 - 23 @authenticated 24 async def service_proxy(request: web.Request, service: Optional[str] = None): 25 """ ··· 30 logger.info(f"proxying lxm {lxm}") 31 db = get_db(request) 32 if service: 33 - service_did = service.partition("#")[0] 34 - service_route = SERVICE_ROUTES.get(service) 35 - if service_route is None: 36 return web.HTTPBadRequest(f"unable to resolve service {service!r}") 37 - else: 38 service_did = db.config["bsky_appview_did"] 39 service_route = db.config["bsky_appview_pfx"] 40

··· 12 logger = logging.getLogger(__name__) 13 14 15 @authenticated 16 async def service_proxy(request: web.Request, service: Optional[str] = None): 17 """ ··· 22 logger.info(f"proxying lxm {lxm}") 23 db = get_db(request) 24 if service: 25 + service_did, _, fragment = service.partition("#") 26 + fragment = "#" + fragment 27 + did_doc = await get_did_resolver(request).resolve_with_db_cache( 28 + db, service_did 29 + ) 30 + if did_doc is None: 31 + return web.HTTPInternalServerError( 32 + f"unable to resolve service {service!r}" 33 + ) 34 + for service in did_doc.get("service", []): 35 + if service.get("id") == fragment: 36 + service_route = service["serviceEndpoint"] 37 + break 38 + else: 39 return web.HTTPBadRequest(f"unable to resolve service {service!r}") 40 + else: # fall thru to assuming bsky appview 41 service_did = db.config["bsky_appview_did"] 42 service_route = db.config["bsky_appview_pfx"] 43

+3 -2

src/millipds/did.py

··· 55 # try the db first 56 now = int(time.time()) 57 row = db.con.execute( 58 - "SELECT doc FROM did_cache WHERE did=? AND expires_at<?", (did, now) 59 ).fetchone() 60 61 # cache hit ··· 71 ) 72 try: 73 doc = await self.resolve_uncached(did) 74 except Exception as e: 75 - logger.exception(f"Error resolving DID {did}: {e}") 76 doc = None 77 78 # update "now" because resolution might've taken a while

··· 55 # try the db first 56 now = int(time.time()) 57 row = db.con.execute( 58 + "SELECT doc FROM did_cache WHERE did=? AND ?<expires_at", (did, now) 59 ).fetchone() 60 61 # cache hit ··· 71 ) 72 try: 73 doc = await self.resolve_uncached(did) 74 + logger.info(f"Successfully resolved {did}") 75 except Exception as e: 76 + logger.exception(f"Error resolving {did}: {e}") 77 doc = None 78 79 # update "now" because resolution might've taken a while

src/millipds/service.py

··· 26 from .appview_proxy import service_proxy 27 from .auth_bearer import authenticated 28 from .app_util import * 29 30 logger = logging.getLogger(__name__) 31 ··· 402 {"User-Agent": importlib.metadata.version("millipds")} 403 ) 404 405 app = web.Application(middlewares=[cors, atproto_service_proxy_middleware]) 406 app[MILLIPDS_DB] = db 407 app[MILLIPDS_AIOHTTP_CLIENT] = client 408 app[MILLIPDS_FIREHOSE_QUEUES] = set() 409 app[MILLIPDS_FIREHOSE_QUEUES_LOCK] = asyncio.Lock() 410 411 app.add_routes(routes) 412 app.add_routes(auth_oauth.routes)

··· 26 from .appview_proxy import service_proxy 27 from .auth_bearer import authenticated 28 from .app_util import * 29 + from .did import DIDResolver 30 31 logger = logging.getLogger(__name__) 32 ··· 403 {"User-Agent": importlib.metadata.version("millipds")} 404 ) 405 406 + did_resolver = DIDResolver(client, static_config.PLC_DIRECTORY_HOST) 407 + 408 app = web.Application(middlewares=[cors, atproto_service_proxy_middleware]) 409 app[MILLIPDS_DB] = db 410 app[MILLIPDS_AIOHTTP_CLIENT] = client 411 app[MILLIPDS_FIREHOSE_QUEUES] = set() 412 app[MILLIPDS_FIREHOSE_QUEUES_LOCK] = asyncio.Lock() 413 + app[MILLIPDS_DID_RESOLVER] = did_resolver 414 415 app.add_routes(routes) 416 app.add_routes(auth_oauth.routes)

+5 -3

src/millipds/static_config.py

··· 10 11 GROUPNAME = "millipds-sock" 12 13 - MILLIPDS_DB_VERSION = ( 14 - 1 # this gets bumped if we make breaking changes to the db schema 15 - ) 16 ATPROTO_REPO_VERSION_3 = 3 # might get bumped if the atproto spec changes 17 CAR_VERSION_1 = 1 18 ··· 27 28 DID_CACHE_TTL = 60 * 60 # 1 hour 29 DID_CACHE_ERROR_TTL = 60 * 5 # 5 mins

··· 10 11 GROUPNAME = "millipds-sock" 12 13 + # this gets bumped if we make breaking changes to the db schema 14 + MILLIPDS_DB_VERSION = 2 15 + 16 ATPROTO_REPO_VERSION_3 = 3 # might get bumped if the atproto spec changes 17 CAR_VERSION_1 = 1 18 ··· 27 28 DID_CACHE_TTL = 60 * 60 # 1 hour 29 DID_CACHE_ERROR_TTL = 60 * 5 # 5 mins 30 + 31 + PLC_DIRECTORY_HOST = "https://plc.directory"