zzstoatzz.io
draw things
doodl.waow.tech
draw
atproto
bot: fix duplicate posts via single Sub2 DO instance
multiple orphaned Durable Object instances each held their own jetstream
socket and posted independently. delete the entire Subscriber class (wiping
every instance) and recreate as a single Sub2 singleton; back the in-memory
dedup claim with a persistent key.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Opus 4.8 (1M context)
date
(May 31, 2026, 12:16 AM -0500)
commit
2d77f15c
2d77f15c871c21b89266f3e1512b6becd71e4ea8
parent
b4aa1575
b4aa157596bd684fa9db77439248efecf37598f9
+84
-16
+77
-14
bot/bot.js
+77
-14
bot/bot.js
···
5
5
// created. it keeps itself warm + reconnects via a self-scheduled DO alarm
6
6
// (NOT a cron — there is no polling; events are processed live as they arrive),
7
7
// and persists a cursor so a dropped socket replays without missing anything.
8
-
const JETSTREAM = "wss://jetstream.waow.tech/subscribe";
8
+
// outbound WS from a Worker/DO uses the https scheme + Upgrade header (NOT wss)
9
+
const JETSTREAM = "https://jetstream.waow.tech/subscribe";
9
10
const SLINGSHOT = "https://slingshot.microcosm.blue";
10
11
const DRAWING = "tech.waow.doodl.drawing";
11
12
const CONSENT = "tech.waow.doodl.botConsent";
···
13
14
const BOT_HANDLE = "doodl.waow.tech";
14
15
const HEARTBEAT_MS = 10_000; // keep the DO warm + reconnect if the socket dropped
15
16
16
-
export class Subscriber {
17
+
export class Sub2 {
17
18
constructor(state, env) {
18
19
this.state = state;
19
20
this.env = env;
20
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
+
};
21
31
}
22
32
23
-
// kick endpoint — first hit bootstraps the alarm loop, then it self-sustains
24
-
async fetch() {
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
+
}
25
43
const a = await this.state.storage.getAlarm();
26
44
if (a === null) await this.state.storage.setAlarm(Date.now() + HEARTBEAT_MS);
27
45
await this.ensureConnected();
28
-
return new Response("subscriber alive");
46
+
return Response.json({
47
+
wsReadyState: this.ws ? this.ws.readyState : null,
48
+
cursor: await this.state.storage.get("cursor"),
49
+
...this.stats,
50
+
});
29
51
}
30
52
31
53
async alarm() {
···
40
62
}
41
63
42
64
async connect() {
65
+
// resume from the last seen event; live (no cursor) on a cold start
43
66
const cursor = await this.state.storage.get("cursor");
44
67
const url =
45
68
`${JETSTREAM}?wantedCollections=${DRAWING}` +
46
69
(cursor ? `&cursor=${cursor}` : "");
70
+
this.stats.connects++;
47
71
try {
48
72
const resp = await fetch(url, { headers: { Upgrade: "websocket" } });
49
73
const ws = resp.webSocket;
74
+
this.stats.lastConnect = `status=${resp.status} ws=${ws ? "ok" : "null"}`;
50
75
if (!ws) return;
51
76
ws.accept();
52
77
this.ws = ws;
···
57
82
ws.addEventListener("error", () => {
58
83
if (this.ws === ws) this.ws = null;
59
84
});
60
-
} catch {
85
+
} catch (e) {
86
+
this.stats.lastConnect = `error: ${e.message}`;
61
87
this.ws = null;
62
88
}
63
89
}
···
69
95
} catch {
70
96
return;
71
97
}
98
+
this.stats.msgs++;
72
99
if (m.time_us) await this.state.storage.put("cursor", String(m.time_us));
73
100
const c = m.commit;
74
101
if (m.kind !== "commit" || c?.operation !== "create" || c.collection !== DRAWING)
···
76
103
const did = m.did;
77
104
const rkey = c.rkey;
78
105
const cid = c.record?.image?.ref?.["$link"];
106
+
this.stats.drawings++;
79
107
if (!cid) return;
80
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.
81
112
const key = `posted:${did}/${rkey}`;
82
-
if (await this.state.storage.get(key)) return;
83
-
if (!this.env.BOT_APP_PASSWORD) return; // idle until the secret is set
84
-
if (!(await hasConsent(did))) return;
113
+
if (this.seen.has(key)) return;
114
+
this.seen.add(key);
85
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
86
122
await postDrawing(this.env, did, rkey, cid);
87
123
await this.state.storage.put(key, 1);
88
-
console.log("posted", did, rkey);
124
+
this.stats.posts++;
89
125
} catch (e) {
90
-
console.log("post failed", did, rkey, e.message);
126
+
this.seen.delete(key); // allow a retry on transient failure
127
+
this.stats.errors.push(`post ${rkey}: ${e.message}`.slice(0, 200));
91
128
}
92
129
}
93
130
}
94
131
95
132
export default {
96
-
// hitting the worker bootstraps / re-kicks the singleton subscriber
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).
97
136
async fetch(req, env) {
98
-
const stub = env.SUB.get(env.SUB.idFromName("singleton"));
99
-
return stub.fetch(req);
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);
100
141
},
101
142
};
143
+
144
+
async 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
+
}
102
165
103
166
async function hasConsent(did) {
104
167
try {
+7
-2
bot/wrangler.jsonc
+7
-2
bot/wrangler.jsonc
···
4
4
"compatibility_date": "2026-05-01",
5
5
"account_id": "3e9ba01cd687b3c4d29033908177072e",
6
6
"workers_dev": true,
7
+
"triggers": { "crons": [] },
7
8
"durable_objects": {
8
-
"bindings": [{ "name": "SUB", "class_name": "Subscriber" }]
9
+
"bindings": [{ "name": "SUB", "class_name": "Sub2" }]
9
10
},
10
-
"migrations": [{ "tag": "v1", "new_sqlite_classes": ["Subscriber"] }],
11
+
"migrations": [
12
+
{ "tag": "v1", "new_sqlite_classes": ["Subscriber"] },
13
+
{ "tag": "v2", "deleted_classes": ["Subscriber"] },
14
+
{ "tag": "v3", "new_sqlite_classes": ["Sub2"] }
15
+
],
11
16
"observability": { "enabled": true }
12
17
}