zzstoatzz.io
This repository has no description
1#!/usr/bin/env -S uv run --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# dependencies = ["httpx", "websockets"]
5# ///
6"""proof of concept: orchestration records on a stock bluesky PDS.
7
8writes io.prefect.* records to an ordinary bsky.social account and shows
9them arriving on the public jetstream — i.e. identity-authenticated writes
10and a free event stream, no infrastructure of ours.
11
12usage:
13 ATPROTO_DEVLOG_HANDLE=zzstoatzzdevlog.bsky.social \
14 ATPROTO_DEVLOG_PASSWORD=... ./scripts/devlog-poc
15"""
16import asyncio
17import json
18import os
19import sys
20from datetime import datetime, timezone
21
22import httpx
23import websockets
24
25PDS = "https://bsky.social"
26JETSTREAM = "wss://jetstream2.us-east.bsky.network/subscribe"
27HANDLE = os.environ.get("ATPROTO_DEVLOG_HANDLE", "zzstoatzzdevlog.bsky.social")
28PASSWORD = os.environ["ATPROTO_DEVLOG_PASSWORD"]
29
30COLLECTION = "io.prefect.v0.event"
31
32
33def now() -> str:
34 return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
35
36
37async def main() -> None:
38 async with httpx.AsyncClient() as http:
39 session = (
40 await http.post(
41 f"{PDS}/xrpc/com.atproto.server.createSession",
42 json={"identifier": HANDLE, "password": PASSWORD},
43 )
44 ).raise_for_status().json()
45 did = session["did"]
46 headers = {"Authorization": f"Bearer {session['accessJwt']}"}
47 print(f"authenticated as {HANDLE} ({did})")
48
49 record = {
50 "$type": COLLECTION,
51 "occurred": now(),
52 "event": "prefect.flow-run.Completed",
53 "resource": {
54 "prefect.resource.id": "prefect.flow-run.demo-run-0001",
55 "prefect.resource.name": "demo",
56 },
57 "payload": {
58 "validated_state": {"type": "COMPLETED", "name": "Completed", "timestamp": now()},
59 },
60 }
61
62 async with websockets.connect(
63 f"{JETSTREAM}?wantedCollections={COLLECTION}&dids={did}"
64 ) as ws:
65 print(f"listening on jetstream for {COLLECTION} from {did} ...")
66
67 created = (
68 await http.post(
69 f"{PDS}/xrpc/com.atproto.repo.createRecord",
70 headers=headers,
71 json={"repo": did, "collection": COLLECTION, "record": record},
72 )
73 ).raise_for_status().json()
74 print(f"wrote record: {created['uri']} (cid {created['cid'][:16]}...)")
75
76 event = json.loads(await asyncio.wait_for(ws.recv(), timeout=30))
77 commit = event.get("commit", {})
78 print("jetstream event received:")
79 print(f" did: {event.get('did')}")
80 print(f" collection: {commit.get('collection')}")
81 print(f" rkey: {commit.get('rkey')}")
82 print(f" event: {commit.get('record', {}).get('event')}")
83
84 listed = (
85 await http.get(
86 f"{PDS}/xrpc/com.atproto.repo.listRecords",
87 params={"repo": did, "collection": COLLECTION},
88 )
89 ).raise_for_status().json()
90 print(f"records now in {COLLECTION}: {len(listed['records'])}")
91 print("=== a stock bluesky pds just served as orchestration storage + event stream ===")
92
93
94if __name__ == "__main__":
95 try:
96 asyncio.run(main())
97 except KeyError:
98 sys.exit("set ATPROTO_DEVLOG_PASSWORD")