zzstoatzz.io
draw things
doodl.waow.tech
draw
atproto
1// doodl bot — features new drawings on @doodl.waow.tech, in real time.
2//
3// a single Durable Object holds an OPEN websocket to jetstream.waow.tech
4// (filtered to tech.waow.doodl.drawing) and reacts the instant a drawing is
5// created. it keeps itself warm + reconnects via a self-scheduled DO alarm
6// (NOT a cron — there is no polling; events are processed live as they arrive),
7// and persists a cursor so a dropped socket replays without missing anything.
8// outbound WS from a Worker/DO uses the https scheme + Upgrade header (NOT wss)
9const JETSTREAM = "https://jetstream.waow.tech/subscribe";
10const SLINGSHOT = "https://slingshot.microcosm.blue";
11const DRAWING = "tech.waow.doodl.drawing";
12const CONSENT = "tech.waow.doodl.botConsent";
13const APP = "https://doodl.waow.tech";
14const BOT_HANDLE = "doodl.waow.tech";
15const HEARTBEAT_MS = 10_000; // keep the DO warm + reconnect if the socket dropped
16
17export class Sub2 {
18 constructor(state, env) {
19 this.state = state;
20 this.env = env;
21 this.ws = null;
22 this.seen = new Set(); // sync in-memory dedup claim (per instance)
23 this.stats = {
24 connects: 0,
25 lastConnect: null,
26 msgs: 0,
27 drawings: 0,
28 posts: 0,
29 errors: [],
30 };
31 }
32
33 // kick + status (and ?kill to stop an orphaned instance: clears its alarm)
34 async fetch(req) {
35 if (new URL(req.url).searchParams.has("kill")) {
36 await this.state.storage.deleteAlarm();
37 try {
38 this.ws && this.ws.close();
39 } catch {}
40 this.ws = null;
41 return new Response("stopped");
42 }
43 const a = await this.state.storage.getAlarm();
44 if (a === null) await this.state.storage.setAlarm(Date.now() + HEARTBEAT_MS);
45 await this.ensureConnected();
46 return Response.json({
47 wsReadyState: this.ws ? this.ws.readyState : null,
48 cursor: await this.state.storage.get("cursor"),
49 ...this.stats,
50 });
51 }
52
53 async alarm() {
54 // reschedule first so the loop can never break, then heal the connection
55 await this.state.storage.setAlarm(Date.now() + HEARTBEAT_MS);
56 await this.ensureConnected();
57 }
58
59 async ensureConnected() {
60 if (this.ws && this.ws.readyState === 1) return; // 1 = OPEN
61 await this.connect();
62 }
63
64 async connect() {
65 // resume from the last seen event; live (no cursor) on a cold start
66 const cursor = await this.state.storage.get("cursor");
67 const url =
68 `${JETSTREAM}?wantedCollections=${DRAWING}` +
69 (cursor ? `&cursor=${cursor}` : "");
70 this.stats.connects++;
71 try {
72 const resp = await fetch(url, { headers: { Upgrade: "websocket" } });
73 const ws = resp.webSocket;
74 this.stats.lastConnect = `status=${resp.status} ws=${ws ? "ok" : "null"}`;
75 if (!ws) return;
76 ws.accept();
77 this.ws = ws;
78 ws.addEventListener("message", (e) => this.onMessage(e.data));
79 ws.addEventListener("close", () => {
80 if (this.ws === ws) this.ws = null;
81 });
82 ws.addEventListener("error", () => {
83 if (this.ws === ws) this.ws = null;
84 });
85 } catch (e) {
86 this.stats.lastConnect = `error: ${e.message}`;
87 this.ws = null;
88 }
89 }
90
91 async onMessage(data) {
92 let m;
93 try {
94 m = JSON.parse(data);
95 } catch {
96 return;
97 }
98 this.stats.msgs++;
99 if (m.time_us) await this.state.storage.put("cursor", String(m.time_us));
100 const c = m.commit;
101 if (m.kind !== "commit" || c?.operation !== "create" || c.collection !== DRAWING)
102 return;
103 const did = m.did;
104 const rkey = c.rkey;
105 const cid = c.record?.image?.ref?.["$link"];
106 this.stats.drawings++;
107 if (!cid) return;
108
109 // dedup: claim the rkey synchronously (before any await) so concurrent
110 // replays of the same create can't all slip past the check, then back it
111 // with a persistent key that survives reconnects/restarts.
112 const key = `posted:${did}/${rkey}`;
113 if (this.seen.has(key)) return;
114 this.seen.add(key);
115 try {
116 if (await this.state.storage.get(key)) return;
117 if (!this.env.BOT_APP_PASSWORD) {
118 this.stats.errors.push("no BOT_APP_PASSWORD");
119 return;
120 }
121 if (!(await hasConsent(did))) return; // not opted in — silent skip
122 await postDrawing(this.env, did, rkey, cid);
123 await this.state.storage.put(key, 1);
124 this.stats.posts++;
125 } catch (e) {
126 this.seen.delete(key); // allow a retry on transient failure
127 this.stats.errors.push(`post ${rkey}: ${e.message}`.slice(0, 200));
128 }
129 }
130}
131
132export default {
133 // hitting the worker bootstraps / re-kicks the singleton subscriber.
134 // ?do=<name> targets a specific instance (e.g. to ?kill an orphan).
135 // ?purge deletes the bot's own feed posts (admin cleanup).
136 async fetch(req, env) {
137 const u = new URL(req.url);
138 if (u.searchParams.has("purge")) return purgePosts(env);
139 const name = u.searchParams.get("do") || "v1";
140 return env.SUB.get(env.SUB.idFromName(name)).fetch(req);
141 },
142};
143
144async function purgePosts(env) {
145 const bot = await miniDoc(BOT_HANDLE);
146 const sess = await jpost(bot.pds, "com.atproto.server.createSession", {
147 identifier: BOT_HANDLE,
148 password: env.BOT_APP_PASSWORD,
149 });
150 const auth = sess.accessJwt;
151 const list = await fetch(
152 `${bot.pds}/xrpc/com.atproto.repo.listRecords?repo=${bot.did}&collection=app.bsky.feed.post&limit=100`,
153 ).then((r) => r.json());
154 const rkeys = (list.records || []).map((r) => r.uri.split("/").pop());
155 for (const rkey of rkeys) {
156 await jpost(
157 bot.pds,
158 "com.atproto.repo.deleteRecord",
159 { repo: bot.did, collection: "app.bsky.feed.post", rkey },
160 auth,
161 );
162 }
163 return Response.json({ deleted: rkeys });
164}
165
166async function hasConsent(did) {
167 try {
168 const r = await fetch(
169 `${SLINGSHOT}/xrpc/com.atproto.repo.getRecord?repo=${did}&collection=${CONSENT}&rkey=self`,
170 );
171 return r.ok;
172 } catch {
173 return false;
174 }
175}
176
177async function miniDoc(idOrDid) {
178 const r = await fetch(
179 `${SLINGSHOT}/xrpc/blue.microcosm.identity.resolveMiniDoc?identifier=${idOrDid}`,
180 );
181 if (!r.ok) throw new Error(`resolveMiniDoc ${r.status}`);
182 return r.json();
183}
184
185async function postDrawing(env, did, rkey, cid) {
186 const [author, bot] = await Promise.all([miniDoc(did), miniDoc(BOT_HANDLE)]);
187
188 const imgRes = await fetch(
189 `${author.pds}/xrpc/com.atproto.sync.getBlob?did=${did}&cid=${cid}`,
190 );
191 if (!imgRes.ok) throw new Error(`getBlob ${imgRes.status}`);
192 const bytes = new Uint8Array(await imgRes.arrayBuffer());
193
194 const sess = await jpost(bot.pds, "com.atproto.server.createSession", {
195 identifier: BOT_HANDLE,
196 password: env.BOT_APP_PASSWORD,
197 });
198 const auth = sess.accessJwt;
199
200 const up = await rawPost(
201 bot.pds,
202 "com.atproto.repo.uploadBlob",
203 bytes,
204 "image/png",
205 auth,
206 );
207 const detail = `${APP}/l/${did}/${rkey}`;
208 const text = `a doodl by @${author.handle}\n${detail}`;
209 await jpost(
210 bot.pds,
211 "com.atproto.repo.createRecord",
212 {
213 repo: bot.did,
214 collection: "app.bsky.feed.post",
215 record: {
216 $type: "app.bsky.feed.post",
217 text,
218 createdAt: new Date().toISOString(),
219 facets: linkFacet(text, detail),
220 embed: {
221 $type: "app.bsky.embed.images",
222 images: [{ image: up.blob, alt: `a drawing by @${author.handle}` }],
223 },
224 },
225 },
226 auth,
227 );
228}
229
230async function jpost(pds, nsid, body, auth) {
231 const r = await fetch(`${pds}/xrpc/${nsid}`, {
232 method: "POST",
233 headers: {
234 "content-type": "application/json",
235 ...(auth ? { authorization: `Bearer ${auth}` } : {}),
236 },
237 body: JSON.stringify(body),
238 });
239 if (!r.ok) throw new Error(`${nsid} ${r.status} ${await r.text()}`);
240 return r.json();
241}
242
243async function rawPost(pds, nsid, bytes, mime, auth) {
244 const r = await fetch(`${pds}/xrpc/${nsid}`, {
245 method: "POST",
246 headers: { "content-type": mime, authorization: `Bearer ${auth}` },
247 body: bytes,
248 });
249 if (!r.ok) throw new Error(`${nsid} ${r.status}`);
250 return r.json();
251}
252
253function linkFacet(text, url) {
254 const enc = new TextEncoder();
255 const i = text.indexOf(url);
256 if (i < 0) return undefined;
257 const byteStart = enc.encode(text.slice(0, i)).length;
258 const byteEnd = byteStart + enc.encode(url).length;
259 return [
260 {
261 index: { byteStart, byteEnd },
262 features: [{ $type: "app.bsky.richtext.facet#link", uri: url }],
263 },
264 ];
265}