Webhooks for the AT Protocol
airglow.run
atproto
atprotocol
automation
webhook
feat: did filters in jetstream url for data heavy collections
+1368
-218
+2
-1
.claude/launch.json
+2
-1
.claude/launch.json
+4
.env.example
+4
.env.example
···
7
7
SECRETS_KEY= # optional, enables encrypted user secrets. openssl rand -base64 32
8
8
NSID_ALLOWLIST=
9
9
NSID_BLOCKLIST=
10
+
# Glob patterns (e.g. app.bsky.*) for NSIDs that require automations to
11
+
# declare a non-empty wantedDids list. Use for high-volume collections
12
+
# where a firehose-wide subscription would be too noisy.
13
+
NSID_REQUIRES_DIDS=
10
14
# Hand-picked automation AT URIs, comma-separated. Shown on the homepage
11
15
# and pinned at the top of the gallery tagged "Featured".
12
16
# Example: at://did:plc:abc/run.airglow.automation/xyz,at://did:plc:def/run.airglow.automation/abc
+113
-3
app/islands/AutomationForm.tsx
+113
-3
app/islands/AutomationForm.tsx
···
1
1
import { useState, useCallback, useRef, useMemo, useEffect } from "hono/jsx";
2
2
import type { RecordSchema } from "../../lib/lexicons/schema-types.js";
3
+
import { nsidRequiresWantedDids } from "../../lib/lexicons/match.js";
3
4
import { isRecordProducingAction, type Action, type FetchStep } from "../../lib/db/schema.js";
4
5
import { ACTION_CATALOGUE, type AddableActionId } from "../../lib/automations/action-catalogue.js";
5
6
import { SCOPE_INSUFFICIENT, redirectToScopeUpgrade } from "../../lib/auth/scope-errors.js";
···
69
70
actions: Action[];
70
71
fetches: FetchStep[];
71
72
conditions: Array<{ field: string; operator: string; value: string; comment?: string }>;
73
+
wantedDids: string[];
72
74
active: boolean;
75
+
};
76
+
77
+
export type AutomationFormConfig = {
78
+
/** NSID glob patterns whose automations must declare wantedDids. */
79
+
nsidRequireDids: string[];
73
80
};
74
81
75
82
const NSID_RE = /^[a-z][a-z0-9-]*(\.[a-zA-Z][a-zA-Z0-9-]*){2,}$/;
···
687
694
return conditions.map((c) => ({ ...c, comment: c.comment ?? "" }));
688
695
}
689
696
690
-
export default function AutomationForm({ initial }: { initial?: AutomationInitial }) {
697
+
export default function AutomationForm({
698
+
initial,
699
+
formConfig,
700
+
}: {
701
+
initial?: AutomationInitial;
702
+
formConfig?: AutomationFormConfig;
703
+
}) {
704
+
const nsidRequireDids = formConfig?.nsidRequireDids ?? [];
691
705
const isEdit = !!initial?.rkey;
692
706
const initialLexicon = initial?.lexicon ?? getInitialParam("lexicon");
693
707
const [name, setName] = useState(initial?.name ?? "");
···
701
715
const [conditions, setConditions] = useState<Condition[]>(
702
716
initial ? toConditionDrafts(initial.conditions) : [],
703
717
);
718
+
const [wantedDids, setWantedDids] = useState<string[]>(initial?.wantedDids ?? []);
704
719
const [fetches, setFetches] = useState<FetchDraft[]>(
705
720
initial ? toFetchDrafts(initial.fetches) : [],
706
721
);
···
724
739
return true;
725
740
if (JSON.stringify(conditions) !== JSON.stringify(toConditionDrafts(initial.conditions)))
726
741
return true;
742
+
if (JSON.stringify(wantedDids) !== JSON.stringify(initial.wantedDids ?? [])) return true;
727
743
if (JSON.stringify(fetches) !== JSON.stringify(toFetchDrafts(initial.fetches))) return true;
728
744
if (JSON.stringify(actions) !== JSON.stringify(toActionDrafts(initial.actions))) return true;
729
745
return false;
···
733
749
description ||
734
750
lexicon ||
735
751
conditions.length ||
752
+
wantedDids.length ||
736
753
fetches.length ||
737
754
actions.length
738
755
);
739
-
}, [name, description, lexicon, operations, conditions, fetches, actions, isEdit]);
756
+
}, [name, description, lexicon, operations, conditions, wantedDids, fetches, actions, isEdit]);
740
757
741
758
useEffect(() => {
742
759
if (!isDirty) return;
···
840
857
[],
841
858
);
842
859
860
+
const addWantedDid = useCallback((value: string) => {
861
+
setWantedDids((prev) => (prev.includes(value) ? prev : [...prev, value]));
862
+
}, []);
863
+
const removeWantedDid = useCallback((index: number) => {
864
+
setWantedDids((prev) => prev.filter((_, i) => i !== index));
865
+
}, []);
866
+
const updateWantedDid = useCallback((index: number, value: string) => {
867
+
setWantedDids((prev) => prev.map((d, i) => (i === index ? value : d)));
868
+
}, []);
869
+
const clearWantedDids = useCallback(() => setWantedDids([]), []);
870
+
871
+
const wantedDidsRequired = useMemo(
872
+
() => nsidRequiresWantedDids(lexicon, nsidRequireDids),
873
+
[lexicon, nsidRequireDids],
874
+
);
875
+
843
876
const addFetch = useCallback(() => {
844
877
setFetches((prev) => [...prev, { name: "", uri: "", comment: "" }]);
845
878
}, []);
···
922
955
...(c.comment ? { comment: c.comment } : {}),
923
956
}));
924
957
}
958
+
const trimmedWantedDids = wantedDids.map((d) => d.trim()).filter(Boolean);
959
+
if (trimmedWantedDids.length > 0 || isEdit) {
960
+
payload.wantedDids = trimmedWantedDids;
961
+
}
925
962
payload.actions = actions.map((a) => {
926
963
const comment = a.comment ? { comment: a.comment } : {};
927
964
if (a.type === "webhook") {
···
984
1021
};
985
1022
});
986
1023
return JSON.stringify(payload, null, 2);
987
-
}, [name, description, lexicon, operations, fetches, conditions, actions]);
1024
+
}, [name, description, lexicon, operations, fetches, conditions, wantedDids, actions]);
988
1025
989
1026
const handleSubmit = useCallback(
990
1027
async (e: Event) => {
···
1117
1154
</div>
1118
1155
)}
1119
1156
</div>
1157
+
1158
+
{NSID_RE.test(lexicon) && (wantedDidsRequired || wantedDids.length > 0) && (
1159
+
<div class={s.fieldGroup}>
1160
+
<span class={s.label}>
1161
+
Watched repos {wantedDidsRequired && <span class={s.hint}>(required)</span>}
1162
+
</span>
1163
+
{wantedDidsRequired ? (
1164
+
<>
1165
+
<span class={s.hint}>
1166
+
DIDs to subscribe to at the Jetstream level — events from other repos are not
1167
+
received. Use <code>{"{{self}}"}</code> for your own DID.
1168
+
</span>
1169
+
<span class={s.hint}>
1170
+
This NSID ({lexicon}) requires at least one watched repo on this instance.
1171
+
</span>
1172
+
{wantedDids.map((did, i) => (
1173
+
<div key={i} class={s.conditionRow}>
1174
+
<div class={s.conditionValue}>
1175
+
<input
1176
+
class={s.input}
1177
+
type="text"
1178
+
placeholder="did:plc:… or {{self}}"
1179
+
value={did}
1180
+
onInput={(e: Event) =>
1181
+
updateWantedDid(i, (e.target as HTMLInputElement).value)
1182
+
}
1183
+
/>
1184
+
</div>
1185
+
<button type="button" class={s.removeBtn} onClick={() => removeWantedDid(i)}>
1186
+
Remove
1187
+
</button>
1188
+
</div>
1189
+
))}
1190
+
{wantedDids.length < 10 && (
1191
+
<div class={s.conditionRow}>
1192
+
{!wantedDids.includes("{{self}}") && (
1193
+
<button
1194
+
type="button"
1195
+
class={s.addBtn}
1196
+
onClick={() => addWantedDid("{{self}}")}
1197
+
>
1198
+
+ Watch my DID ({"{{self}}"})
1199
+
</button>
1200
+
)}
1201
+
<button type="button" class={s.addBtn} onClick={() => addWantedDid("")}>
1202
+
+ Add DID
1203
+
</button>
1204
+
</div>
1205
+
)}
1206
+
</>
1207
+
) : (
1208
+
<>
1209
+
<span class={s.hint}>
1210
+
This NSID no longer requires watched repos. Use a condition on{" "}
1211
+
<code>event.did</code> instead, or clear these to let events through on the
1212
+
shared stream.
1213
+
</span>
1214
+
{wantedDids.map((did, i) => (
1215
+
<div key={i} class={s.conditionRow}>
1216
+
<div class={s.conditionValue}>
1217
+
<input class={s.input} type="text" value={did} readOnly />
1218
+
</div>
1219
+
</div>
1220
+
))}
1221
+
<div class={s.conditionRow}>
1222
+
<button type="button" class={s.removeBtn} onClick={clearWantedDids}>
1223
+
Clear watched repos
1224
+
</button>
1225
+
</div>
1226
+
</>
1227
+
)}
1228
+
</div>
1229
+
)}
1120
1230
1121
1231
<div class={s.fieldGroup}>
1122
1232
<span class={s.label}>Trigger events</span>
+1
app/routes/api/automations/[rkey].test.ts
+1
app/routes/api/automations/[rkey].test.ts
+16
app/routes/api/automations/[rkey].ts
+16
app/routes/api/automations/[rkey].ts
···
13
13
type BookmarkAction,
14
14
type FetchStep,
15
15
} from "@/db/schema.js";
16
+
import { config } from "@/config.js";
16
17
import { isValidNsid } from "@/lexicons/resolver.js";
17
18
import {
18
19
getRecord,
···
37
38
BCP47_RE,
38
39
validateWebhookHeaders,
39
40
validateBookmarkInput,
41
+
resolveWantedDids,
40
42
} from "@/actions/validation.js";
41
43
import { AUTOMATION_LIMITS } from "@/automations/limits.js";
42
44
import { notifyAutomationChange } from "@/jetstream/consumer.js";
···
84
86
),
85
87
fetches: auto.fetches,
86
88
conditions: auto.conditions,
89
+
wantedDids: auto.wantedDids,
87
90
active: auto.active,
88
91
dryRun: auto.dryRun,
89
92
indexedAt: auto.indexedAt.getTime(),
···
116
119
actions?: ActionInput[];
117
120
fetches?: Array<{ name: string; uri: string; comment?: string }>;
118
121
conditions?: Array<{ field: string; operator?: string; value: string; comment?: string }>;
122
+
wantedDids?: string[];
119
123
active?: boolean;
120
124
dryRun?: boolean;
121
125
}>();
···
171
175
return c.json({ error: `Invalid condition operator: ${cond.operator}` }, 400);
172
176
}
173
177
}
178
+
179
+
const wd = resolveWantedDids(
180
+
body.wantedDids,
181
+
auto.lexicon,
182
+
config.nsidRequireDids,
183
+
auto.wantedDids,
184
+
);
185
+
if (!wd.valid) return c.json({ error: wd.error }, 400);
186
+
const wantedDids = wd.value;
187
+
174
188
const active = body.active ?? auto.active;
175
189
const dryRun = body.dryRun ?? auto.dryRun;
176
190
···
519
533
...(f.comment ? { comment: f.comment } : {}),
520
534
})),
521
535
conditions,
536
+
...(wantedDids.length > 0 ? { wantedDids } : {}),
522
537
active,
523
538
dryRun,
524
539
createdAt,
···
539
554
actions: localActions,
540
555
fetches: localFetches,
541
556
conditions,
557
+
wantedDids,
542
558
active,
543
559
dryRun,
544
560
indexedAt: now,
+1
app/routes/api/automations/[rkey]/logs.test.ts
+1
app/routes/api/automations/[rkey]/logs.test.ts
+1
app/routes/api/automations/index.test.ts
+1
app/routes/api/automations/index.test.ts
+9
app/routes/api/automations/index.ts
+9
app/routes/api/automations/index.ts
···
36
36
BCP47_RE,
37
37
validateWebhookHeaders,
38
38
validateBookmarkInput,
39
+
resolveWantedDids,
39
40
} from "@/actions/validation.js";
40
41
import { AUTOMATION_LIMITS } from "@/automations/limits.js";
41
42
import { computeRequiredScope, scopeCoversActions } from "@/auth/client.js";
···
68
69
),
69
70
fetches: r.fetches,
70
71
conditions: r.conditions,
72
+
wantedDids: r.wantedDids,
71
73
active: r.active,
72
74
indexedAt: r.indexedAt.getTime(),
73
75
})),
···
84
86
actions: ActionInput[];
85
87
fetches?: Array<{ name: string; uri: string; comment?: string }>;
86
88
conditions?: Array<{ field: string; operator?: string; value: string; comment?: string }>;
89
+
wantedDids?: string[];
87
90
active?: boolean;
88
91
dryRun?: boolean;
89
92
}>();
···
153
156
return c.json({ error: `Invalid condition operator: ${cond.operator}` }, 400);
154
157
}
155
158
}
159
+
160
+
const wd = resolveWantedDids(body.wantedDids, body.lexicon, config.nsidRequireDids);
161
+
if (!wd.valid) return c.json({ error: wd.error }, 400);
162
+
const wantedDids = wd.value;
156
163
157
164
// Validate and normalize fetch steps
158
165
const localFetches: FetchStep[] = [];
···
402
409
actions: pdsActions,
403
410
fetches: pdsFetches.length > 0 ? pdsFetches : undefined,
404
411
conditions,
412
+
...(wantedDids.length > 0 ? { wantedDids } : {}),
405
413
active,
406
414
dryRun,
407
415
createdAt: now.toISOString(),
···
426
434
actions: localActions,
427
435
fetches: localFetches,
428
436
conditions,
437
+
wantedDids,
429
438
active,
430
439
dryRun,
431
440
indexedAt: now,
+17
app/routes/dashboard/automations/[rkey].tsx
+17
app/routes/dashboard/automations/[rkey].tsx
···
139
139
</details>
140
140
</Card>
141
141
142
+
{auto.wantedDids.length > 0 && (
143
+
<Card variant="flat">
144
+
<Stack gap={3}>
145
+
<h3 class={inlineCluster}>
146
+
<Filter size={18} /> Watched repos
147
+
</h3>
148
+
<ul class={plainList}>
149
+
{auto.wantedDids.map((did, i) => (
150
+
<li key={i}>
151
+
<InlineCode>{did}</InlineCode>
152
+
</li>
153
+
))}
154
+
</ul>
155
+
</Stack>
156
+
</Card>
157
+
)}
158
+
142
159
{auto.conditions.length > 0 && (
143
160
<Card variant="flat">
144
161
<Stack gap={3}>
+3
app/routes/dashboard/automations/[rkey]/duplicate.tsx
+3
app/routes/dashboard/automations/[rkey]/duplicate.tsx
···
3
3
import { ArrowLeft } from "../../../../icons.js";
4
4
import { db } from "@/db/index.js";
5
5
import { automations } from "@/db/schema.js";
6
+
import { config } from "@/config.js";
6
7
import { AppShell } from "../../../../components/Layout/AppShell/index.js";
7
8
import { Header } from "../../../../components/Layout/Header/index.js";
8
9
import { Container } from "../../../../components/Layout/Container/index.js";
···
61
62
actions: auto.actions,
62
63
fetches: auto.fetches,
63
64
conditions: auto.conditions,
65
+
wantedDids: auto.wantedDids,
64
66
active: false,
65
67
}}
68
+
formConfig={{ nsidRequireDids: config.nsidRequireDids }}
66
69
/>
67
70
</Card>
68
71
</Container>
+3
app/routes/dashboard/automations/[rkey]/edit.tsx
+3
app/routes/dashboard/automations/[rkey]/edit.tsx
···
3
3
import { ArrowLeft } from "../../../../icons.js";
4
4
import { db } from "@/db/index.js";
5
5
import { automations } from "@/db/schema.js";
6
+
import { config } from "@/config.js";
6
7
import { AppShell } from "../../../../components/Layout/AppShell/index.js";
7
8
import { Header } from "../../../../components/Layout/Header/index.js";
8
9
import { Container } from "../../../../components/Layout/Container/index.js";
···
62
63
actions: auto.actions,
63
64
fetches: auto.fetches,
64
65
conditions: auto.conditions,
66
+
wantedDids: auto.wantedDids,
65
67
active: auto.active,
66
68
}}
69
+
formConfig={{ nsidRequireDids: config.nsidRequireDids }}
67
70
/>
68
71
</Card>
69
72
</Container>
+6
-1
app/routes/dashboard/automations/new.tsx
+6
-1
app/routes/dashboard/automations/new.tsx
···
3
3
import { ArrowLeft } from "../../../icons.js";
4
4
import { db } from "@/db/index.js";
5
5
import { automations } from "@/db/schema.js";
6
+
import { config } from "@/config.js";
6
7
import { AppShell } from "../../../components/Layout/AppShell/index.js";
7
8
import { Header } from "../../../components/Layout/Header/index.js";
8
9
import { Container } from "../../../components/Layout/Container/index.js";
···
39
40
}),
40
41
fetches: source.fetches,
41
42
conditions: source.conditions,
43
+
wantedDids: source.wantedDids,
42
44
active: false,
43
45
};
44
46
}
···
58
60
}
59
61
/>
60
62
<Card variant="flat">
61
-
<AutomationForm initial={initial} />
63
+
<AutomationForm
64
+
initial={initial}
65
+
formConfig={{ nsidRequireDids: config.nsidRequireDids }}
66
+
/>
62
67
</Card>
63
68
</Container>
64
69
</AppShell>,
+9
lexicons/run/airglow/automation.json
+9
lexicons/run/airglow/automation.json
···
60
60
"ref": "#condition"
61
61
}
62
62
},
63
+
"wantedDids": {
64
+
"type": "array",
65
+
"description": "Repo DIDs to subscribe to at the Jetstream level. Required for NSIDs that the instance restricts to known-DID subscriptions. Supports the {{self}} placeholder for the owner's DID.",
66
+
"maxLength": 10,
67
+
"items": {
68
+
"type": "string",
69
+
"maxLength": 256
70
+
}
71
+
},
63
72
"fetches": {
64
73
"type": "array",
65
74
"description": "Records to fetch from PDS before executing actions. Fetched data is available as named variables in action templates.",
+58
lib/actions/validation.test.ts
+58
lib/actions/validation.test.ts
···
1
+
import { describe, it, expect } from "vitest";
2
+
import { validateWantedDids } from "./validation.js";
3
+
4
+
describe("validateWantedDids", () => {
5
+
it("accepts undefined/null as empty", () => {
6
+
expect(validateWantedDids(undefined)).toEqual({ valid: true, value: [] });
7
+
expect(validateWantedDids(null)).toEqual({ valid: true, value: [] });
8
+
});
9
+
10
+
it("rejects non-array input", () => {
11
+
const res = validateWantedDids("did:plc:abc");
12
+
expect(res.valid).toBe(false);
13
+
});
14
+
15
+
it("accepts valid did:plc DIDs", () => {
16
+
const res = validateWantedDids(["did:plc:abcdefghijklmnopqrstuvwx"]);
17
+
expect(res).toEqual({ valid: true, value: ["did:plc:abcdefghijklmnopqrstuvwx"] });
18
+
});
19
+
20
+
it("accepts the {{self}} placeholder", () => {
21
+
const res = validateWantedDids(["{{self}}"]);
22
+
expect(res).toEqual({ valid: true, value: ["{{self}}"] });
23
+
});
24
+
25
+
it("accepts did:web DIDs", () => {
26
+
const res = validateWantedDids(["did:web:example.com"]);
27
+
expect(res).toEqual({ valid: true, value: ["did:web:example.com"] });
28
+
});
29
+
30
+
it("rejects malformed DIDs", () => {
31
+
const res = validateWantedDids(["not-a-did"]);
32
+
expect(res.valid).toBe(false);
33
+
});
34
+
35
+
it("dedupes and trims entries", () => {
36
+
const did = "did:plc:abcdefghijklmnopqrstuvwx";
37
+
const res = validateWantedDids([did, ` ${did} `, "{{self}}"]);
38
+
expect(res).toEqual({ valid: true, value: [did, "{{self}}"] });
39
+
});
40
+
41
+
it("drops empty strings", () => {
42
+
const res = validateWantedDids(["", " ", "{{self}}"]);
43
+
expect(res).toEqual({ valid: true, value: ["{{self}}"] });
44
+
});
45
+
46
+
it("rejects non-string entries", () => {
47
+
const res = validateWantedDids([123]);
48
+
expect(res.valid).toBe(false);
49
+
});
50
+
51
+
it("enforces the max length", () => {
52
+
const base = "abcdefghijklmnopqrstuvw";
53
+
const suffixes = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"];
54
+
const dids = suffixes.map((s) => `did:plc:${base}${s}`);
55
+
const res = validateWantedDids(dids);
56
+
expect(res.valid).toBe(false);
57
+
});
58
+
});
+67
lib/actions/validation.ts
+67
lib/actions/validation.ts
···
1
1
import { SECRET_NAME_RE, SECRET_REF_RE } from "../secrets/store.js";
2
2
import { AUTOMATION_LIMITS, BOOKMARK_LIMITS } from "../automations/limits.js";
3
+
import { nsidRequiresWantedDids } from "../lexicons/match.js";
3
4
import { validateTextTemplate } from "./template.js";
4
5
5
6
export type ActionInput =
···
37
38
export const VALID_OPERATIONS = new Set(["create", "update", "delete"]);
38
39
export const VALID_BSKY_LABELS = new Set(["sexual", "nudity", "porn", "graphic-media"]);
39
40
export const BCP47_RE = /^[a-z]{2,3}(-[A-Za-z0-9]{1,8})*$/;
41
+
42
+
// did:plc:<24 base32 chars> or did:web:<domain>
43
+
const DID_RE = /^did:(plc:[a-z2-7]{24}|web:[a-z0-9.-]+(?::\d+)?(?:\/[^\s]*)?)$/;
44
+
45
+
/**
46
+
* Normalize and validate a wantedDids list. Accepts the literal `{{self}}`
47
+
* placeholder alongside concrete DIDs. Returns a deduped, trimmed list.
48
+
*/
49
+
export function validateWantedDids(
50
+
input: unknown,
51
+
): { valid: true; value: string[] } | { valid: false; error: string } {
52
+
if (input === undefined || input === null) return { valid: true, value: [] };
53
+
if (!Array.isArray(input)) {
54
+
return { valid: false, error: "wantedDids must be an array" };
55
+
}
56
+
if (input.length > AUTOMATION_LIMITS.wantedDids) {
57
+
return {
58
+
valid: false,
59
+
error: `Maximum ${AUTOMATION_LIMITS.wantedDids} wantedDids allowed`,
60
+
};
61
+
}
62
+
const seen = new Set<string>();
63
+
const out: string[] = [];
64
+
for (const raw of input) {
65
+
if (typeof raw !== "string") {
66
+
return { valid: false, error: "wantedDids entries must be strings" };
67
+
}
68
+
const v = raw.trim();
69
+
if (!v) continue;
70
+
if (v !== "{{self}}" && !DID_RE.test(v)) {
71
+
return { valid: false, error: `Invalid DID: "${v}"` };
72
+
}
73
+
if (seen.has(v)) continue;
74
+
seen.add(v);
75
+
out.push(v);
76
+
}
77
+
return { valid: true, value: out };
78
+
}
79
+
80
+
/**
81
+
* Resolve wantedDids for a create/update call: validate the input (falling
82
+
* back to `existing` when undefined) and enforce the `nsidRequireDids`
83
+
* policy gate. Returns the final list or a caller-ready error string.
84
+
*/
85
+
export function resolveWantedDids(
86
+
input: unknown,
87
+
lexicon: string,
88
+
nsidRequireDids: string[],
89
+
existing?: string[],
90
+
): { valid: true; value: string[] } | { valid: false; error: string } {
91
+
let value: string[];
92
+
if (input === undefined && existing !== undefined) {
93
+
value = existing;
94
+
} else {
95
+
const r = validateWantedDids(input);
96
+
if (!r.valid) return r;
97
+
value = r.value;
98
+
}
99
+
if (nsidRequiresWantedDids(lexicon, nsidRequireDids) && value.length === 0) {
100
+
return {
101
+
valid: false,
102
+
error: `Automations on ${lexicon} must specify at least one wantedDid (use {{self}} for the owner's DID).`,
103
+
};
104
+
}
105
+
return { valid: true, value };
106
+
}
40
107
41
108
// System headers that cannot be overridden by custom webhook headers
42
109
const SYSTEM_HEADERS = new Set([
+1
lib/automations/limits.ts
+1
lib/automations/limits.ts
+1
lib/automations/pds.ts
+1
lib/automations/pds.ts
+4
lib/config.ts
+4
lib/config.ts
···
44
44
secretsKey,
45
45
nsidAllowlist: env("NSID_ALLOWLIST", "").split(",").filter(Boolean),
46
46
nsidBlocklist: env("NSID_BLOCKLIST", "").split(",").filter(Boolean),
47
+
// NSIDs listed here are only allowed when the automation declares a non-empty
48
+
// wantedDids. Used to gate high-volume collections (e.g. app.bsky.*) on
49
+
// Jetstream-level DID filtering instead of a blanket firehose subscription.
50
+
nsidRequireDids: env("NSID_REQUIRES_DIDS", "").split(",").filter(Boolean),
47
51
// Hand-picked AT URIs (at://did/run.airglow.automation/rkey), comma-separated.
48
52
// These appear on the homepage and at the top of the gallery tagged "Featured".
49
53
featuredAutomations: env("FEATURED_AUTOMATIONS", "")
+1
lib/db/migrations/0007_magenta_moira_mactaggert.sql
+1
lib/db/migrations/0007_magenta_moira_mactaggert.sql
···
1
+
ALTER TABLE `automations` ADD `wanted_dids` text DEFAULT '[]' NOT NULL;
+566
lib/db/migrations/meta/0007_snapshot.json
+566
lib/db/migrations/meta/0007_snapshot.json
···
1
+
{
2
+
"version": "6",
3
+
"dialect": "sqlite",
4
+
"id": "be7ee96c-dd9b-4de3-b74e-0b9139455f3f",
5
+
"prevId": "b0b35a17-c3c4-48be-8978-59eea412ec36",
6
+
"tables": {
7
+
"automations": {
8
+
"name": "automations",
9
+
"columns": {
10
+
"uri": {
11
+
"name": "uri",
12
+
"type": "text",
13
+
"primaryKey": true,
14
+
"notNull": true,
15
+
"autoincrement": false
16
+
},
17
+
"did": {
18
+
"name": "did",
19
+
"type": "text",
20
+
"primaryKey": false,
21
+
"notNull": true,
22
+
"autoincrement": false
23
+
},
24
+
"rkey": {
25
+
"name": "rkey",
26
+
"type": "text",
27
+
"primaryKey": false,
28
+
"notNull": true,
29
+
"autoincrement": false
30
+
},
31
+
"name": {
32
+
"name": "name",
33
+
"type": "text",
34
+
"primaryKey": false,
35
+
"notNull": true,
36
+
"autoincrement": false
37
+
},
38
+
"description": {
39
+
"name": "description",
40
+
"type": "text",
41
+
"primaryKey": false,
42
+
"notNull": false,
43
+
"autoincrement": false
44
+
},
45
+
"lexicon": {
46
+
"name": "lexicon",
47
+
"type": "text",
48
+
"primaryKey": false,
49
+
"notNull": true,
50
+
"autoincrement": false
51
+
},
52
+
"operation": {
53
+
"name": "operation",
54
+
"type": "text",
55
+
"primaryKey": false,
56
+
"notNull": true,
57
+
"autoincrement": false,
58
+
"default": "'[\"create\"]'"
59
+
},
60
+
"actions": {
61
+
"name": "actions",
62
+
"type": "text",
63
+
"primaryKey": false,
64
+
"notNull": true,
65
+
"autoincrement": false,
66
+
"default": "'[]'"
67
+
},
68
+
"fetches": {
69
+
"name": "fetches",
70
+
"type": "text",
71
+
"primaryKey": false,
72
+
"notNull": true,
73
+
"autoincrement": false,
74
+
"default": "'[]'"
75
+
},
76
+
"conditions": {
77
+
"name": "conditions",
78
+
"type": "text",
79
+
"primaryKey": false,
80
+
"notNull": true,
81
+
"autoincrement": false,
82
+
"default": "'[]'"
83
+
},
84
+
"wanted_dids": {
85
+
"name": "wanted_dids",
86
+
"type": "text",
87
+
"primaryKey": false,
88
+
"notNull": true,
89
+
"autoincrement": false,
90
+
"default": "'[]'"
91
+
},
92
+
"active": {
93
+
"name": "active",
94
+
"type": "integer",
95
+
"primaryKey": false,
96
+
"notNull": true,
97
+
"autoincrement": false,
98
+
"default": false
99
+
},
100
+
"dry_run": {
101
+
"name": "dry_run",
102
+
"type": "integer",
103
+
"primaryKey": false,
104
+
"notNull": true,
105
+
"autoincrement": false,
106
+
"default": false
107
+
},
108
+
"indexed_at": {
109
+
"name": "indexed_at",
110
+
"type": "integer",
111
+
"primaryKey": false,
112
+
"notNull": true,
113
+
"autoincrement": false
114
+
}
115
+
},
116
+
"indexes": {
117
+
"automations_did_idx": {
118
+
"name": "automations_did_idx",
119
+
"columns": [
120
+
"did"
121
+
],
122
+
"isUnique": false
123
+
},
124
+
"automations_active_indexed_at_idx": {
125
+
"name": "automations_active_indexed_at_idx",
126
+
"columns": [
127
+
"active",
128
+
"indexed_at"
129
+
],
130
+
"isUnique": false
131
+
}
132
+
},
133
+
"foreignKeys": {},
134
+
"compositePrimaryKeys": {},
135
+
"uniqueConstraints": {},
136
+
"checkConstraints": {}
137
+
},
138
+
"delivery_logs": {
139
+
"name": "delivery_logs",
140
+
"columns": {
141
+
"id": {
142
+
"name": "id",
143
+
"type": "integer",
144
+
"primaryKey": true,
145
+
"notNull": true,
146
+
"autoincrement": true
147
+
},
148
+
"automation_uri": {
149
+
"name": "automation_uri",
150
+
"type": "text",
151
+
"primaryKey": false,
152
+
"notNull": true,
153
+
"autoincrement": false
154
+
},
155
+
"action_index": {
156
+
"name": "action_index",
157
+
"type": "integer",
158
+
"primaryKey": false,
159
+
"notNull": true,
160
+
"autoincrement": false,
161
+
"default": 0
162
+
},
163
+
"event_time_us": {
164
+
"name": "event_time_us",
165
+
"type": "integer",
166
+
"primaryKey": false,
167
+
"notNull": true,
168
+
"autoincrement": false
169
+
},
170
+
"payload": {
171
+
"name": "payload",
172
+
"type": "text",
173
+
"primaryKey": false,
174
+
"notNull": false,
175
+
"autoincrement": false
176
+
},
177
+
"status_code": {
178
+
"name": "status_code",
179
+
"type": "integer",
180
+
"primaryKey": false,
181
+
"notNull": false,
182
+
"autoincrement": false
183
+
},
184
+
"message": {
185
+
"name": "message",
186
+
"type": "text",
187
+
"primaryKey": false,
188
+
"notNull": false,
189
+
"autoincrement": false
190
+
},
191
+
"error": {
192
+
"name": "error",
193
+
"type": "text",
194
+
"primaryKey": false,
195
+
"notNull": false,
196
+
"autoincrement": false
197
+
},
198
+
"dry_run": {
199
+
"name": "dry_run",
200
+
"type": "integer",
201
+
"primaryKey": false,
202
+
"notNull": true,
203
+
"autoincrement": false,
204
+
"default": false
205
+
},
206
+
"attempt": {
207
+
"name": "attempt",
208
+
"type": "integer",
209
+
"primaryKey": false,
210
+
"notNull": true,
211
+
"autoincrement": false,
212
+
"default": 1
213
+
},
214
+
"created_at": {
215
+
"name": "created_at",
216
+
"type": "integer",
217
+
"primaryKey": false,
218
+
"notNull": true,
219
+
"autoincrement": false
220
+
}
221
+
},
222
+
"indexes": {
223
+
"delivery_logs_automation_uri_id_idx": {
224
+
"name": "delivery_logs_automation_uri_id_idx",
225
+
"columns": [
226
+
"automation_uri",
227
+
"id"
228
+
],
229
+
"isUnique": false
230
+
}
231
+
},
232
+
"foreignKeys": {
233
+
"delivery_logs_automation_uri_automations_uri_fk": {
234
+
"name": "delivery_logs_automation_uri_automations_uri_fk",
235
+
"tableFrom": "delivery_logs",
236
+
"tableTo": "automations",
237
+
"columnsFrom": [
238
+
"automation_uri"
239
+
],
240
+
"columnsTo": [
241
+
"uri"
242
+
],
243
+
"onDelete": "cascade",
244
+
"onUpdate": "no action"
245
+
}
246
+
},
247
+
"compositePrimaryKeys": {},
248
+
"uniqueConstraints": {},
249
+
"checkConstraints": {}
250
+
},
251
+
"favicon_cache": {
252
+
"name": "favicon_cache",
253
+
"columns": {
254
+
"domain": {
255
+
"name": "domain",
256
+
"type": "text",
257
+
"primaryKey": true,
258
+
"notNull": true,
259
+
"autoincrement": false
260
+
},
261
+
"data": {
262
+
"name": "data",
263
+
"type": "text",
264
+
"primaryKey": false,
265
+
"notNull": true,
266
+
"autoincrement": false
267
+
},
268
+
"content_type": {
269
+
"name": "content_type",
270
+
"type": "text",
271
+
"primaryKey": false,
272
+
"notNull": true,
273
+
"autoincrement": false
274
+
},
275
+
"fetched_at": {
276
+
"name": "fetched_at",
277
+
"type": "integer",
278
+
"primaryKey": false,
279
+
"notNull": true,
280
+
"autoincrement": false
281
+
}
282
+
},
283
+
"indexes": {},
284
+
"foreignKeys": {},
285
+
"compositePrimaryKeys": {},
286
+
"uniqueConstraints": {},
287
+
"checkConstraints": {}
288
+
},
289
+
"lexicon_cache": {
290
+
"name": "lexicon_cache",
291
+
"columns": {
292
+
"nsid": {
293
+
"name": "nsid",
294
+
"type": "text",
295
+
"primaryKey": true,
296
+
"notNull": true,
297
+
"autoincrement": false
298
+
},
299
+
"schema": {
300
+
"name": "schema",
301
+
"type": "text",
302
+
"primaryKey": false,
303
+
"notNull": true,
304
+
"autoincrement": false
305
+
},
306
+
"fetched_at": {
307
+
"name": "fetched_at",
308
+
"type": "integer",
309
+
"primaryKey": false,
310
+
"notNull": true,
311
+
"autoincrement": false
312
+
}
313
+
},
314
+
"indexes": {},
315
+
"foreignKeys": {},
316
+
"compositePrimaryKeys": {},
317
+
"uniqueConstraints": {},
318
+
"checkConstraints": {}
319
+
},
320
+
"oauth_sessions": {
321
+
"name": "oauth_sessions",
322
+
"columns": {
323
+
"key": {
324
+
"name": "key",
325
+
"type": "text",
326
+
"primaryKey": true,
327
+
"notNull": true,
328
+
"autoincrement": false
329
+
},
330
+
"value": {
331
+
"name": "value",
332
+
"type": "text",
333
+
"primaryKey": false,
334
+
"notNull": true,
335
+
"autoincrement": false
336
+
},
337
+
"expires_at": {
338
+
"name": "expires_at",
339
+
"type": "integer",
340
+
"primaryKey": false,
341
+
"notNull": false,
342
+
"autoincrement": false
343
+
}
344
+
},
345
+
"indexes": {},
346
+
"foreignKeys": {},
347
+
"compositePrimaryKeys": {},
348
+
"uniqueConstraints": {},
349
+
"checkConstraints": {}
350
+
},
351
+
"oauth_states": {
352
+
"name": "oauth_states",
353
+
"columns": {
354
+
"key": {
355
+
"name": "key",
356
+
"type": "text",
357
+
"primaryKey": true,
358
+
"notNull": true,
359
+
"autoincrement": false
360
+
},
361
+
"value": {
362
+
"name": "value",
363
+
"type": "text",
364
+
"primaryKey": false,
365
+
"notNull": true,
366
+
"autoincrement": false
367
+
},
368
+
"expires_at": {
369
+
"name": "expires_at",
370
+
"type": "integer",
371
+
"primaryKey": false,
372
+
"notNull": false,
373
+
"autoincrement": false
374
+
}
375
+
},
376
+
"indexes": {},
377
+
"foreignKeys": {},
378
+
"compositePrimaryKeys": {},
379
+
"uniqueConstraints": {},
380
+
"checkConstraints": {}
381
+
},
382
+
"secret_events": {
383
+
"name": "secret_events",
384
+
"columns": {
385
+
"id": {
386
+
"name": "id",
387
+
"type": "integer",
388
+
"primaryKey": true,
389
+
"notNull": true,
390
+
"autoincrement": true
391
+
},
392
+
"did": {
393
+
"name": "did",
394
+
"type": "text",
395
+
"primaryKey": false,
396
+
"notNull": true,
397
+
"autoincrement": false
398
+
},
399
+
"name": {
400
+
"name": "name",
401
+
"type": "text",
402
+
"primaryKey": false,
403
+
"notNull": true,
404
+
"autoincrement": false
405
+
},
406
+
"action": {
407
+
"name": "action",
408
+
"type": "text",
409
+
"primaryKey": false,
410
+
"notNull": true,
411
+
"autoincrement": false
412
+
},
413
+
"created_at": {
414
+
"name": "created_at",
415
+
"type": "integer",
416
+
"primaryKey": false,
417
+
"notNull": true,
418
+
"autoincrement": false
419
+
}
420
+
},
421
+
"indexes": {},
422
+
"foreignKeys": {},
423
+
"compositePrimaryKeys": {},
424
+
"uniqueConstraints": {},
425
+
"checkConstraints": {}
426
+
},
427
+
"user_secrets": {
428
+
"name": "user_secrets",
429
+
"columns": {
430
+
"id": {
431
+
"name": "id",
432
+
"type": "integer",
433
+
"primaryKey": true,
434
+
"notNull": true,
435
+
"autoincrement": true
436
+
},
437
+
"did": {
438
+
"name": "did",
439
+
"type": "text",
440
+
"primaryKey": false,
441
+
"notNull": true,
442
+
"autoincrement": false
443
+
},
444
+
"name": {
445
+
"name": "name",
446
+
"type": "text",
447
+
"primaryKey": false,
448
+
"notNull": true,
449
+
"autoincrement": false
450
+
},
451
+
"encrypted_value": {
452
+
"name": "encrypted_value",
453
+
"type": "blob",
454
+
"primaryKey": false,
455
+
"notNull": true,
456
+
"autoincrement": false
457
+
},
458
+
"created_at": {
459
+
"name": "created_at",
460
+
"type": "integer",
461
+
"primaryKey": false,
462
+
"notNull": true,
463
+
"autoincrement": false
464
+
},
465
+
"updated_at": {
466
+
"name": "updated_at",
467
+
"type": "integer",
468
+
"primaryKey": false,
469
+
"notNull": true,
470
+
"autoincrement": false
471
+
}
472
+
},
473
+
"indexes": {
474
+
"user_secrets_did_name_unique": {
475
+
"name": "user_secrets_did_name_unique",
476
+
"columns": [
477
+
"did",
478
+
"name"
479
+
],
480
+
"isUnique": true
481
+
}
482
+
},
483
+
"foreignKeys": {
484
+
"user_secrets_did_users_did_fk": {
485
+
"name": "user_secrets_did_users_did_fk",
486
+
"tableFrom": "user_secrets",
487
+
"tableTo": "users",
488
+
"columnsFrom": [
489
+
"did"
490
+
],
491
+
"columnsTo": [
492
+
"did"
493
+
],
494
+
"onDelete": "cascade",
495
+
"onUpdate": "no action"
496
+
}
497
+
},
498
+
"compositePrimaryKeys": {},
499
+
"uniqueConstraints": {},
500
+
"checkConstraints": {}
501
+
},
502
+
"users": {
503
+
"name": "users",
504
+
"columns": {
505
+
"id": {
506
+
"name": "id",
507
+
"type": "integer",
508
+
"primaryKey": true,
509
+
"notNull": true,
510
+
"autoincrement": true
511
+
},
512
+
"did": {
513
+
"name": "did",
514
+
"type": "text",
515
+
"primaryKey": false,
516
+
"notNull": true,
517
+
"autoincrement": false
518
+
},
519
+
"handle": {
520
+
"name": "handle",
521
+
"type": "text",
522
+
"primaryKey": false,
523
+
"notNull": true,
524
+
"autoincrement": false
525
+
},
526
+
"scope": {
527
+
"name": "scope",
528
+
"type": "text",
529
+
"primaryKey": false,
530
+
"notNull": false,
531
+
"autoincrement": false
532
+
},
533
+
"created_at": {
534
+
"name": "created_at",
535
+
"type": "integer",
536
+
"primaryKey": false,
537
+
"notNull": true,
538
+
"autoincrement": false
539
+
}
540
+
},
541
+
"indexes": {
542
+
"users_did_unique": {
543
+
"name": "users_did_unique",
544
+
"columns": [
545
+
"did"
546
+
],
547
+
"isUnique": true
548
+
}
549
+
},
550
+
"foreignKeys": {},
551
+
"compositePrimaryKeys": {},
552
+
"uniqueConstraints": {},
553
+
"checkConstraints": {}
554
+
}
555
+
},
556
+
"views": {},
557
+
"enums": {},
558
+
"_meta": {
559
+
"schemas": {},
560
+
"tables": {},
561
+
"columns": {}
562
+
},
563
+
"internal": {
564
+
"indexes": {}
565
+
}
566
+
}
+7
lib/db/migrations/meta/_journal.json
+7
lib/db/migrations/meta/_journal.json
+1
lib/db/schema.ts
+1
lib/db/schema.ts
···
87
87
.notNull()
88
88
.$type<Array<{ field: string; operator: string; value: string; comment?: string }>>()
89
89
.default([]),
90
+
wantedDids: text("wanted_dids", { mode: "json" }).notNull().$type<string[]>().default([]),
90
91
active: integer("active", { mode: "boolean" }).notNull().default(false),
91
92
dryRun: integer("dry_run", { mode: "boolean" }).notNull().default(false),
92
93
indexedAt: integer("indexed_at", { mode: "timestamp_ms" }).notNull(),
+240
-118
lib/jetstream/consumer.test.ts
+240
-118
lib/jetstream/consumer.test.ts
···
6
6
jetstreamUrl: "wss://jetstream.test/subscribe",
7
7
nsidAllowlist: [],
8
8
nsidBlocklist: [],
9
+
nsidRequireDids: [],
9
10
},
10
11
}));
11
12
···
14
15
return { db: createTestDb() };
15
16
});
16
17
17
-
import { JetstreamConsumer, type MatchedEvent } from "./consumer.js";
18
+
import { JetstreamManager, type MatchedEvent } from "./consumer.js";
18
19
import { db } from "../db/index.js";
19
20
import { automations } from "../db/schema.js";
20
21
import { makeAutomation, makeEvent } from "../test/fixtures.js";
···
28
29
}
29
30
vi.stubGlobal("WebSocket", MockWebSocket);
30
31
31
-
describe("JetstreamConsumer", () => {
32
+
describe("JetstreamManager", () => {
32
33
let handler: ReturnType<typeof vi.fn<(match: MatchedEvent) => void>>;
33
-
let consumer: JetstreamConsumer;
34
+
let manager: JetstreamManager;
34
35
35
36
beforeEach(async () => {
36
37
vi.useFakeTimers();
37
38
handler = vi.fn();
38
-
consumer = new JetstreamConsumer(handler);
39
+
manager = new JetstreamManager(handler);
39
40
40
41
await db.delete(automations);
41
42
});
42
43
43
44
afterEach(() => {
44
-
consumer.stop();
45
+
manager.stop();
45
46
vi.useRealTimers();
46
47
});
47
48
···
68
69
}),
69
70
]);
70
71
71
-
await consumer.refreshAutomations();
72
+
await manager.refreshAutomations();
72
73
73
-
// The consumer is internal, but we can test it indirectly via processEvent
74
-
const event = makeEvent({
75
-
commit: {
76
-
rev: "r",
77
-
operation: "create",
78
-
collection: "app.bsky.feed.like",
79
-
rkey: "rk",
80
-
record: {},
81
-
},
82
-
});
83
-
// Access processEvent via prototype
84
-
(consumer as any).processEvent(event);
74
+
manager.dispatch(
75
+
makeEvent({
76
+
commit: {
77
+
rev: "r",
78
+
operation: "create",
79
+
collection: "app.bsky.feed.like",
80
+
rkey: "rk",
81
+
record: {},
82
+
},
83
+
}),
84
+
);
85
85
86
-
// Both like automations should match (empty conditions = match all)
87
86
expect(handler).toHaveBeenCalledTimes(2);
88
87
});
89
88
···
106
105
}),
107
106
]);
108
107
109
-
await consumer.refreshAutomations();
108
+
await manager.refreshAutomations();
110
109
111
-
// Blocked lexicon should not trigger handler
112
-
(consumer as any).processEvent(
110
+
manager.dispatch(
113
111
makeEvent({
114
112
commit: {
115
113
rev: "r",
···
122
120
);
123
121
expect(handler).not.toHaveBeenCalled();
124
122
125
-
// Allowed lexicon should still work
126
-
(consumer as any).processEvent(
123
+
manager.dispatch(
127
124
makeEvent({
128
125
commit: {
129
126
rev: "r",
···
136
133
);
137
134
expect(handler).toHaveBeenCalledOnce();
138
135
139
-
// Reset
140
136
(config as any).nsidBlocklist = [];
141
137
});
142
138
···
145
141
.insert(automations)
146
142
.values(makeAutomation({ uri: "at://u/s/1", rkey: "1", active: false }));
147
143
148
-
await consumer.refreshAutomations();
144
+
await manager.refreshAutomations();
149
145
150
-
const event = makeEvent();
151
-
(consumer as any).processEvent(event);
146
+
manager.dispatch(makeEvent());
152
147
153
148
expect(handler).not.toHaveBeenCalled();
154
149
});
150
+
151
+
it("skips automations on NSIDs that require wantedDids but have none", async () => {
152
+
const { config } = await import("../config.js");
153
+
(config as any).nsidRequireDids = ["app.bsky.*"];
154
+
const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
155
+
156
+
await db.insert(automations).values([
157
+
makeAutomation({
158
+
uri: "at://u/s/1",
159
+
rkey: "1",
160
+
lexicon: "app.bsky.feed.like",
161
+
wantedDids: [],
162
+
active: true,
163
+
}),
164
+
makeAutomation({
165
+
uri: "at://u/s/2",
166
+
rkey: "2",
167
+
lexicon: "app.bsky.feed.like",
168
+
wantedDids: ["{{self}}"],
169
+
active: true,
170
+
}),
171
+
]);
172
+
173
+
await manager.refreshAutomations();
174
+
175
+
expect(manager.subscriptionCount).toBe(1);
176
+
expect(warn).toHaveBeenCalled();
177
+
178
+
(config as any).nsidRequireDids = [];
179
+
warn.mockRestore();
180
+
});
155
181
});
156
182
157
-
describe("processEvent", () => {
183
+
describe("partitioning", () => {
184
+
it("puts automations with no wantedDids into the global subscription", async () => {
185
+
await db
186
+
.insert(automations)
187
+
.values(makeAutomation({ uri: "at://u/s/1", rkey: "1", wantedDids: [], active: true }));
188
+
189
+
await manager.refreshAutomations();
190
+
191
+
expect(manager.subscriptionKeys).toEqual([""]);
192
+
});
193
+
194
+
it("puts automations with disjoint DID-sets into separate subscriptions", async () => {
195
+
await db.insert(automations).values([
196
+
makeAutomation({
197
+
uri: "at://u/s/1",
198
+
rkey: "1",
199
+
did: "did:plc:a",
200
+
wantedDids: ["{{self}}"],
201
+
active: true,
202
+
}),
203
+
makeAutomation({
204
+
uri: "at://u/s/2",
205
+
rkey: "2",
206
+
did: "did:plc:b",
207
+
wantedDids: ["{{self}}"],
208
+
active: true,
209
+
}),
210
+
]);
211
+
212
+
await manager.refreshAutomations();
213
+
214
+
expect(manager.subscriptionCount).toBe(2);
215
+
expect(new Set(manager.subscriptionKeys)).toEqual(new Set(["did:plc:a", "did:plc:b"]));
216
+
});
217
+
218
+
it("groups automations sharing a DID-set into a single subscription", async () => {
219
+
await db.insert(automations).values([
220
+
makeAutomation({
221
+
uri: "at://u/s/1",
222
+
rkey: "1",
223
+
did: "did:plc:a",
224
+
lexicon: "app.bsky.feed.like",
225
+
wantedDids: ["did:plc:x", "did:plc:y"],
226
+
active: true,
227
+
}),
228
+
makeAutomation({
229
+
uri: "at://u/s/2",
230
+
rkey: "2",
231
+
did: "did:plc:a",
232
+
lexicon: "app.bsky.feed.post",
233
+
wantedDids: ["did:plc:y", "did:plc:x"], // reverse order → same canonical key
234
+
active: true,
235
+
}),
236
+
]);
237
+
238
+
await manager.refreshAutomations();
239
+
240
+
expect(manager.subscriptionKeys).toEqual(["did:plc:x,did:plc:y"]);
241
+
});
242
+
243
+
it("tears down a subscription when its last automation is deleted", async () => {
244
+
await db.insert(automations).values(
245
+
makeAutomation({
246
+
uri: "at://u/s/1",
247
+
rkey: "1",
248
+
did: "did:plc:a",
249
+
wantedDids: ["{{self}}"],
250
+
active: true,
251
+
}),
252
+
);
253
+
await manager.refreshAutomations();
254
+
expect(manager.subscriptionCount).toBe(1);
255
+
256
+
await db.delete(automations);
257
+
await manager.refreshAutomations();
258
+
expect(manager.subscriptionCount).toBe(0);
259
+
});
260
+
261
+
it("resolves {{self}} to the automation owner's DID", async () => {
262
+
await db.insert(automations).values(
263
+
makeAutomation({
264
+
uri: "at://u/s/1",
265
+
rkey: "1",
266
+
did: "did:plc:owner",
267
+
wantedDids: ["{{self}}", "did:plc:other"],
268
+
active: true,
269
+
}),
270
+
);
271
+
272
+
await manager.refreshAutomations();
273
+
274
+
expect(manager.subscriptionKeys).toEqual(["did:plc:other,did:plc:owner"]);
275
+
});
276
+
});
277
+
278
+
describe("dispatch", () => {
158
279
it("calls handler for matching automations", async () => {
159
280
await db.insert(automations).values(
160
281
makeAutomation({
···
164
285
conditions: [],
165
286
}),
166
287
);
167
-
await consumer.refreshAutomations();
288
+
await manager.refreshAutomations();
168
289
169
290
const event = makeEvent({
170
291
commit: {
···
175
296
record: {},
176
297
},
177
298
});
178
-
(consumer as any).processEvent(event);
299
+
manager.dispatch(event);
179
300
180
301
expect(handler).toHaveBeenCalledOnce();
181
302
const match = handler.mock.calls[0]![0];
···
184
305
});
185
306
186
307
it("does not call handler for non-matching collection", async () => {
187
-
await db.insert(automations).values(
188
-
makeAutomation({
189
-
uri: "at://u/s/1",
190
-
rkey: "1",
191
-
lexicon: "app.bsky.feed.post",
308
+
await db
309
+
.insert(automations)
310
+
.values(makeAutomation({ uri: "at://u/s/1", rkey: "1", lexicon: "app.bsky.feed.post" }));
311
+
await manager.refreshAutomations();
312
+
313
+
manager.dispatch(
314
+
makeEvent({
315
+
commit: {
316
+
rev: "r",
317
+
operation: "create",
318
+
collection: "app.bsky.feed.like",
319
+
rkey: "rk",
320
+
record: {},
321
+
},
192
322
}),
193
323
);
194
-
await consumer.refreshAutomations();
195
-
196
-
const event = makeEvent({
197
-
commit: {
198
-
rev: "r",
199
-
operation: "create",
200
-
collection: "app.bsky.feed.like",
201
-
rkey: "rk",
202
-
record: {},
203
-
},
204
-
});
205
-
(consumer as any).processEvent(event);
206
324
207
325
expect(handler).not.toHaveBeenCalled();
208
326
});
···
216
334
operations: ["create"],
217
335
}),
218
336
);
219
-
await consumer.refreshAutomations();
337
+
await manager.refreshAutomations();
220
338
221
-
const event = makeEvent({
222
-
commit: {
223
-
rev: "r",
224
-
operation: "delete",
225
-
collection: "app.bsky.feed.like",
226
-
rkey: "rk",
227
-
record: {},
228
-
},
229
-
});
230
-
(consumer as any).processEvent(event);
339
+
manager.dispatch(
340
+
makeEvent({
341
+
commit: {
342
+
rev: "r",
343
+
operation: "delete",
344
+
collection: "app.bsky.feed.like",
345
+
rkey: "rk",
346
+
record: {},
347
+
},
348
+
}),
349
+
);
231
350
232
351
expect(handler).not.toHaveBeenCalled();
233
352
});
···
241
360
operations: ["create", "delete"],
242
361
}),
243
362
);
244
-
await consumer.refreshAutomations();
363
+
await manager.refreshAutomations();
245
364
246
-
const event = makeEvent({
247
-
commit: {
248
-
rev: "r",
249
-
operation: "delete",
250
-
collection: "app.bsky.feed.like",
251
-
rkey: "rk",
252
-
record: {},
253
-
},
254
-
});
255
-
(consumer as any).processEvent(event);
365
+
manager.dispatch(
366
+
makeEvent({
367
+
commit: {
368
+
rev: "r",
369
+
operation: "delete",
370
+
collection: "app.bsky.feed.like",
371
+
rkey: "rk",
372
+
record: {},
373
+
},
374
+
}),
375
+
);
256
376
257
377
expect(handler).toHaveBeenCalledOnce();
258
378
});
···
266
386
conditions: [{ field: "event.did", operator: "eq", value: "did:plc:specific" }],
267
387
}),
268
388
);
269
-
await consumer.refreshAutomations();
389
+
await manager.refreshAutomations();
270
390
271
-
const event = makeEvent({
272
-
did: "did:plc:other",
273
-
commit: {
274
-
rev: "r",
275
-
operation: "create",
276
-
collection: "app.bsky.feed.like",
277
-
rkey: "rk",
278
-
record: {},
279
-
},
280
-
});
281
-
(consumer as any).processEvent(event);
391
+
manager.dispatch(
392
+
makeEvent({
393
+
did: "did:plc:other",
394
+
commit: {
395
+
rev: "r",
396
+
operation: "create",
397
+
collection: "app.bsky.feed.like",
398
+
rkey: "rk",
399
+
record: {},
400
+
},
401
+
}),
402
+
);
282
403
283
404
expect(handler).not.toHaveBeenCalled();
284
405
});
···
290
411
makeAutomation({ uri: "at://u/s/1", rkey: "1", lexicon: "app.bsky.feed.like" }),
291
412
makeAutomation({ uri: "at://u/s/2", rkey: "2", lexicon: "app.bsky.feed.like" }),
292
413
]);
293
-
await consumer.refreshAutomations();
414
+
await manager.refreshAutomations();
294
415
295
-
const event = makeEvent({
296
-
commit: {
297
-
rev: "r",
298
-
operation: "create",
299
-
collection: "app.bsky.feed.like",
300
-
rkey: "rk",
301
-
record: {},
302
-
},
303
-
});
304
-
(consumer as any).processEvent(event);
416
+
manager.dispatch(
417
+
makeEvent({
418
+
commit: {
419
+
rev: "r",
420
+
operation: "create",
421
+
collection: "app.bsky.feed.like",
422
+
rkey: "rk",
423
+
record: {},
424
+
},
425
+
}),
426
+
);
305
427
306
428
expect(handler).toHaveBeenCalledTimes(2);
307
429
});
···
316
438
conditions: [{ field: "event.did", operator: "eq", value: "{{self}}" }],
317
439
}),
318
440
);
319
-
await consumer.refreshAutomations();
441
+
await manager.refreshAutomations();
320
442
321
-
// Event from the owner
322
-
const ownerEvent = makeEvent({
323
-
did: "did:plc:owner",
324
-
commit: {
325
-
rev: "r",
326
-
operation: "create",
327
-
collection: "app.bsky.feed.like",
328
-
rkey: "rk",
329
-
record: {},
330
-
},
331
-
});
332
-
(consumer as any).processEvent(ownerEvent);
443
+
manager.dispatch(
444
+
makeEvent({
445
+
did: "did:plc:owner",
446
+
commit: {
447
+
rev: "r",
448
+
operation: "create",
449
+
collection: "app.bsky.feed.like",
450
+
rkey: "rk",
451
+
record: {},
452
+
},
453
+
}),
454
+
);
333
455
expect(handler).toHaveBeenCalledOnce();
334
456
335
457
handler.mockClear();
336
458
337
-
// Event from someone else
338
-
const otherEvent = makeEvent({
339
-
did: "did:plc:someone-else",
340
-
commit: {
341
-
rev: "r",
342
-
operation: "create",
343
-
collection: "app.bsky.feed.like",
344
-
rkey: "rk2",
345
-
record: {},
346
-
},
347
-
});
348
-
(consumer as any).processEvent(otherEvent);
459
+
manager.dispatch(
460
+
makeEvent({
461
+
did: "did:plc:someone-else",
462
+
commit: {
463
+
rev: "r",
464
+
operation: "create",
465
+
collection: "app.bsky.feed.like",
466
+
rkey: "rk2",
467
+
record: {},
468
+
},
469
+
}),
470
+
);
349
471
expect(handler).not.toHaveBeenCalled();
350
472
});
351
473
});
+183
-81
lib/jetstream/consumer.ts
+183
-81
lib/jetstream/consumer.ts
···
2
2
import { db } from "../db/index.js";
3
3
import { automations } from "../db/schema.js";
4
4
import { config } from "../config.js";
5
-
import { isNsidAllowed } from "../lexicons/resolver.js";
6
-
import { matchConditions, type JetstreamEvent } from "./matcher.js";
5
+
import { isNsidAllowed, nsidRequiresWantedDids } from "../lexicons/resolver.js";
6
+
import { matchConditions, resolveConditionValue, type JetstreamEvent } from "./matcher.js";
7
7
8
8
type Automation = typeof automations.$inferSelect;
9
9
···
14
14
15
15
type EventHandler = (match: MatchedEvent) => void | Promise<void>;
16
16
17
-
export class JetstreamConsumer {
17
+
/** Resolve `{{self}}` placeholders and dedupe/sort into a canonical list. */
18
+
function canonicalDids(wantedDids: string[], ownerDid: string): string[] {
19
+
const resolved = new Set<string>();
20
+
for (const entry of wantedDids) {
21
+
const did = resolveConditionValue(entry, ownerDid);
22
+
if (did) resolved.add(did);
23
+
}
24
+
return [...resolved].sort();
25
+
}
26
+
27
+
/** One Jetstream WebSocket, scoped to a fixed wantedDids set. */
28
+
export class JetstreamSubscription {
18
29
private ws: WebSocket | null = null;
19
30
private autosByCollectionOp = new Map<string, Automation[]>();
31
+
private collections: string[] = [];
20
32
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
21
33
private reconnectDelay = 1000;
22
34
private running = false;
23
-
private handler: EventHandler;
35
+
private readonly wantedDids: string[];
36
+
private readonly label: string;
37
+
private readonly handler: EventHandler;
24
38
25
-
constructor(handler: EventHandler) {
39
+
constructor(wantedDids: string[], handler: EventHandler) {
40
+
this.wantedDids = wantedDids;
26
41
this.handler = handler;
27
-
}
28
-
29
-
async start() {
30
-
this.running = true;
31
-
await this.refreshAutomations();
32
-
}
33
-
34
-
stop() {
35
-
this.running = false;
36
-
if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
37
-
this.ws?.close();
38
-
this.ws = null;
42
+
this.label = wantedDids.length === 0 ? "global" : `dids=${wantedDids.length}`;
39
43
}
40
44
41
-
async refreshAutomations() {
42
-
const rows = await db.query.automations.findMany({
43
-
where: eq(automations.active, true),
44
-
});
45
+
/** Replace the automation set. Reconnects only if wantedCollections changed. */
46
+
update(byCollectionOp: Map<string, Automation[]>) {
47
+
const newCollections = [...deriveCollections(byCollectionOp)].sort();
48
+
const changed =
49
+
newCollections.length !== this.collections.length ||
50
+
newCollections.some((c, i) => c !== this.collections[i]);
45
51
46
-
const byCollectionOp = new Map<string, Automation[]>();
47
-
for (const row of rows) {
48
-
if (!isNsidAllowed(row.lexicon, config.nsidAllowlist, config.nsidBlocklist)) continue;
49
-
for (const op of row.operations) {
50
-
const key = `${row.lexicon}\0${op}`;
51
-
const list = byCollectionOp.get(key) || [];
52
-
list.push(row);
53
-
byCollectionOp.set(key, list);
54
-
}
55
-
}
56
-
57
-
const deriveCollections = (map: Map<string, Automation[]>) => {
58
-
const cols = new Set<string>();
59
-
for (const key of map.keys()) cols.add(key.slice(0, key.indexOf("\0")));
60
-
return cols;
61
-
};
62
-
63
-
const oldCollections = deriveCollections(this.autosByCollectionOp);
64
-
const newCollections = deriveCollections(byCollectionOp);
65
52
this.autosByCollectionOp = byCollectionOp;
53
+
this.collections = newCollections;
66
54
67
-
const collectionsChanged =
68
-
oldCollections.size !== newCollections.size ||
69
-
[...newCollections].some((c) => !oldCollections.has(c)) ||
70
-
[...oldCollections].some((c) => !newCollections.has(c));
55
+
if (!this.running) return;
71
56
72
-
if (!collectionsChanged) return;
57
+
if (!changed) return;
73
58
74
-
const hadCollections = oldCollections.size > 0;
75
-
const hasCollections = newCollections.size > 0;
59
+
if (newCollections.length === 0) {
60
+
this.ws?.close();
61
+
return;
62
+
}
76
63
77
-
if (!hadCollections && hasCollections) {
64
+
if (!this.ws) {
78
65
this.connect();
79
-
} else if (hadCollections && !hasCollections) {
80
-
this.ws?.close();
81
-
} else if (this.ws && this.ws.readyState <= WebSocket.OPEN) {
82
-
this.ws.close(); // onclose reconnects with updated params
66
+
} else {
67
+
// onclose reconnects with the updated params
68
+
this.ws.close();
83
69
}
84
70
}
85
71
86
-
private connect() {
87
-
if (!this.running) return;
72
+
start() {
73
+
this.running = true;
74
+
if (this.collections.length > 0) this.connect();
75
+
}
76
+
77
+
stop() {
78
+
this.running = false;
88
79
if (this.reconnectTimer) {
89
80
clearTimeout(this.reconnectTimer);
90
81
this.reconnectTimer = null;
91
82
}
83
+
this.ws?.close();
84
+
this.ws = null;
85
+
}
92
86
93
-
const collections = [
94
-
...new Set([...this.autosByCollectionOp.keys()].map((k) => k.slice(0, k.indexOf("\0")))),
95
-
];
96
-
if (collections.length === 0) {
97
-
console.log("Jetstream: no active automations, waiting");
98
-
return;
87
+
private connect() {
88
+
if (!this.running) return;
89
+
if (this.reconnectTimer) {
90
+
clearTimeout(this.reconnectTimer);
91
+
this.reconnectTimer = null;
99
92
}
93
+
if (this.collections.length === 0) return;
100
94
101
95
const url = new URL(config.jetstreamUrl);
102
-
for (const col of collections) {
103
-
url.searchParams.append("wantedCollections", col);
104
-
}
96
+
for (const col of this.collections) url.searchParams.append("wantedCollections", col);
97
+
for (const did of this.wantedDids) url.searchParams.append("wantedDids", did);
98
+
105
99
console.log(
106
-
`Jetstream: connecting (${collections.length} collection${collections.length === 1 ? "" : "s"})`,
100
+
`Jetstream[${this.label}]: connecting (${this.collections.length} collection${this.collections.length === 1 ? "" : "s"}${this.wantedDids.length > 0 ? `, ${this.wantedDids.length} did${this.wantedDids.length === 1 ? "" : "s"}` : ""})`,
107
101
);
108
102
this.ws = new WebSocket(url.toString());
109
103
110
104
this.ws.addEventListener("open", () => {
111
-
console.log("Jetstream: connected");
105
+
console.log(`Jetstream[${this.label}]: connected`);
112
106
this.reconnectDelay = 1000;
113
107
});
114
108
···
116
110
try {
117
111
const event = JSON.parse(msg.data as string) as JetstreamEvent;
118
112
if (event.kind === "commit" && event.commit) {
119
-
this.processEvent(event);
113
+
this.dispatch(event);
120
114
}
121
115
} catch (err) {
122
-
console.error("Jetstream: failed to process message:", err);
116
+
console.error(`Jetstream[${this.label}]: failed to process message:`, err);
123
117
}
124
118
});
125
119
126
120
this.ws.addEventListener("close", () => {
127
121
this.ws = null;
128
-
if (this.running && this.autosByCollectionOp.size > 0) {
129
-
console.log(`Jetstream: reconnecting in ${this.reconnectDelay}ms`);
122
+
if (this.running && this.collections.length > 0) {
123
+
console.log(`Jetstream[${this.label}]: reconnecting in ${this.reconnectDelay}ms`);
130
124
this.reconnectTimer = setTimeout(() => {
131
125
this.reconnectTimer = null;
132
126
this.connect();
···
136
130
});
137
131
138
132
this.ws.addEventListener("error", (err) => {
139
-
console.error("Jetstream: WebSocket error:", err);
133
+
console.error(`Jetstream[${this.label}]: WebSocket error:`, err);
140
134
});
141
135
}
142
136
143
-
private processEvent(event: JetstreamEvent) {
137
+
dispatch(event: JetstreamEvent) {
144
138
const collection = event.commit!.collection;
145
139
const operation = event.commit!.operation;
146
140
const key = `${collection}\0${operation}`;
···
150
144
for (const auto of autos) {
151
145
if (matchConditions(event, auto.conditions, auto.did)) {
152
146
void Promise.resolve(this.handler({ automation: auto, event })).catch((err) => {
153
-
console.error("Jetstream: handler error:", err);
147
+
console.error(`Jetstream[${this.label}]: handler error:`, err);
154
148
});
155
149
}
156
150
}
157
151
}
158
152
}
159
153
154
+
function deriveCollections(map: Map<string, Automation[]>): Set<string> {
155
+
const cols = new Set<string>();
156
+
for (const key of map.keys()) cols.add(key.slice(0, key.indexOf("\0")));
157
+
return cols;
158
+
}
159
+
160
+
/** Manages one subscription per distinct wantedDids set (canonical key). */
161
+
export class JetstreamManager {
162
+
private subscriptions = new Map<string, JetstreamSubscription>();
163
+
private running = false;
164
+
private readonly handler: EventHandler;
165
+
166
+
constructor(handler: EventHandler) {
167
+
this.handler = handler;
168
+
}
169
+
170
+
async start() {
171
+
this.running = true;
172
+
await this.refreshAutomations();
173
+
}
174
+
175
+
stop() {
176
+
this.running = false;
177
+
for (const sub of this.subscriptions.values()) sub.stop();
178
+
this.subscriptions.clear();
179
+
}
180
+
181
+
async refreshAutomations() {
182
+
const rows = await db.query.automations.findMany({
183
+
where: eq(automations.active, true),
184
+
});
185
+
186
+
// Partition by canonical DID-set key. Empty key = global (no DID filter).
187
+
const partitions = new Map<
188
+
string,
189
+
{ wantedDids: string[]; byCollectionOp: Map<string, Automation[]> }
190
+
>();
191
+
192
+
for (const row of rows) {
193
+
if (!isNsidAllowed(row.lexicon, config.nsidAllowlist, config.nsidBlocklist)) continue;
194
+
195
+
const resolvedDids = canonicalDids(row.wantedDids, row.did);
196
+
if (
197
+
nsidRequiresWantedDids(row.lexicon, config.nsidRequireDids) &&
198
+
resolvedDids.length === 0
199
+
) {
200
+
console.warn(
201
+
`Jetstream: skipping ${row.uri} — ${row.lexicon} requires wantedDids but none are set`,
202
+
);
203
+
continue;
204
+
}
205
+
206
+
const key = resolvedDids.join(",");
207
+
let bucket = partitions.get(key);
208
+
if (!bucket) {
209
+
bucket = { wantedDids: resolvedDids, byCollectionOp: new Map() };
210
+
partitions.set(key, bucket);
211
+
}
212
+
for (const op of row.operations) {
213
+
const cellKey = `${row.lexicon}\0${op}`;
214
+
const list = bucket.byCollectionOp.get(cellKey) || [];
215
+
list.push(row);
216
+
bucket.byCollectionOp.set(cellKey, list);
217
+
}
218
+
}
219
+
220
+
// Tear down subscriptions with no partition.
221
+
for (const [key, sub] of this.subscriptions) {
222
+
if (!partitions.has(key)) {
223
+
sub.stop();
224
+
this.subscriptions.delete(key);
225
+
}
226
+
}
227
+
228
+
// Create or update subscriptions for each partition.
229
+
for (const [key, { wantedDids, byCollectionOp }] of partitions) {
230
+
let sub = this.subscriptions.get(key);
231
+
if (!sub) {
232
+
sub = new JetstreamSubscription(wantedDids, this.handler);
233
+
this.subscriptions.set(key, sub);
234
+
sub.update(byCollectionOp);
235
+
if (this.running) sub.start();
236
+
} else {
237
+
sub.update(byCollectionOp);
238
+
}
239
+
}
240
+
}
241
+
242
+
/** Number of live subscriptions (mainly for tests/introspection). */
243
+
get subscriptionCount() {
244
+
return this.subscriptions.size;
245
+
}
246
+
247
+
/** Canonical DID-set keys of live subscriptions (mainly for tests). */
248
+
get subscriptionKeys() {
249
+
return [...this.subscriptions.keys()];
250
+
}
251
+
252
+
/**
253
+
* Dispatch an event to every subscription. In production the WebSocket
254
+
* message handler per subscription does this directly; this helper makes
255
+
* the fan-out testable without standing up real sockets.
256
+
*/
257
+
dispatch(event: JetstreamEvent) {
258
+
for (const sub of this.subscriptions.values()) sub.dispatch(event);
259
+
}
260
+
}
261
+
160
262
// — Singleton management —
161
263
162
-
let consumer: JetstreamConsumer | null = null;
264
+
let manager: JetstreamManager | null = null;
163
265
164
-
export function startJetstream(handler: EventHandler): JetstreamConsumer {
165
-
if (consumer) return consumer;
166
-
consumer = new JetstreamConsumer(handler);
167
-
consumer.start().catch((err) => {
266
+
export function startJetstream(handler: EventHandler): JetstreamManager {
267
+
if (manager) return manager;
268
+
manager = new JetstreamManager(handler);
269
+
manager.start().catch((err) => {
168
270
console.error("Jetstream: failed to start:", err);
169
271
});
170
-
return consumer;
272
+
return manager;
171
273
}
172
274
173
-
export function getJetstream(): JetstreamConsumer | null {
174
-
return consumer;
275
+
export function getJetstream(): JetstreamManager | null {
276
+
return manager;
175
277
}
176
278
177
279
/** Call after creating/updating/deleting automations. */
178
280
export function notifyAutomationChange() {
179
-
consumer?.refreshAutomations().catch((err) => {
281
+
manager?.refreshAutomations().catch((err) => {
180
282
console.error("Jetstream: failed to refresh automations:", err);
181
283
});
182
284
}
+1
-1
lib/jetstream/matcher.ts
+1
-1
lib/jetstream/matcher.ts
···
45
45
return typeof value === "string" ? value : JSON.stringify(value);
46
46
}
47
47
48
-
function resolveConditionValue(value: string, ownerDid: string): string {
48
+
export function resolveConditionValue(value: string, ownerDid: string): string {
49
49
return value === "{{self}}" ? ownerDid : value;
50
50
}
51
51
+25
lib/lexicons/match.ts
+25
lib/lexicons/match.ts
···
1
+
// Browser-safe NSID glob-matching helpers. No node: imports — safe to use from
2
+
// both server code and client islands.
3
+
4
+
export function nsidMatches(nsid: string, pattern: string): boolean {
5
+
return pattern.endsWith(".*") ? nsid.startsWith(pattern.slice(0, -1)) : nsid === pattern;
6
+
}
7
+
8
+
/**
9
+
* Check whether an NSID is allowed by the instance's allowlist/blocklist.
10
+
* Blocklist takes precedence.
11
+
*/
12
+
export function isNsidAllowed(nsid: string, allowlist: string[], blocklist: string[]): boolean {
13
+
if (blocklist.some((p) => nsidMatches(nsid, p))) return false;
14
+
if (allowlist.length > 0) return allowlist.some((p) => nsidMatches(nsid, p));
15
+
return true;
16
+
}
17
+
18
+
/**
19
+
* Whether an NSID may only be subscribed to via an explicit wantedDids list.
20
+
* Used for high-volume collections (e.g. app.bsky.*) where a firehose-wide
21
+
* subscription would overwhelm the worker.
22
+
*/
23
+
export function nsidRequiresWantedDids(nsid: string, patterns: string[]): boolean {
24
+
return patterns.some((p) => nsidMatches(nsid, p));
25
+
}
+25
-1
lib/lexicons/resolver.test.ts
+25
-1
lib/lexicons/resolver.test.ts
···
1
1
import { describe, it, expect } from "vitest";
2
-
import { isValidNsid, nsidToAuthority, isNsidAllowed, parseLexicon } from "./resolver.js";
2
+
import {
3
+
isValidNsid,
4
+
nsidToAuthority,
5
+
isNsidAllowed,
6
+
nsidRequiresWantedDids,
7
+
parseLexicon,
8
+
} from "./resolver.js";
3
9
4
10
describe("isValidNsid", () => {
5
11
it("accepts valid 3-segment NSID", () => {
···
70
76
expect(
71
77
isNsidAllowed("app.bsky.feed.like", ["app.bsky.feed.like"], ["app.bsky.feed.like"]),
72
78
).toBe(false);
79
+
});
80
+
});
81
+
82
+
describe("nsidRequiresWantedDids", () => {
83
+
it("returns false for empty patterns", () => {
84
+
expect(nsidRequiresWantedDids("app.bsky.feed.like", [])).toBe(false);
85
+
});
86
+
87
+
it("matches exact pattern", () => {
88
+
expect(nsidRequiresWantedDids("app.bsky.feed.like", ["app.bsky.feed.like"])).toBe(true);
89
+
});
90
+
91
+
it("matches glob pattern", () => {
92
+
expect(nsidRequiresWantedDids("app.bsky.feed.like", ["app.bsky.*"])).toBe(true);
93
+
});
94
+
95
+
it("does not match unrelated NSID", () => {
96
+
expect(nsidRequiresWantedDids("sh.tangled.feed.star", ["app.bsky.*"])).toBe(false);
73
97
});
74
98
});
75
99
+1
-12
lib/lexicons/resolver.ts
+1
-12
lib/lexicons/resolver.ts
···
42
42
return parts.slice(0, 2).reverse().join(".");
43
43
}
44
44
45
-
/**
46
-
* Check whether an NSID is allowed by the instance's allowlist/blocklist.
47
-
* Blocklist takes precedence. Supports glob patterns like "app.bsky.*".
48
-
*/
49
-
export function isNsidAllowed(nsid: string, allowlist: string[], blocklist: string[]): boolean {
50
-
const matches = (pattern: string) =>
51
-
pattern.endsWith(".*") ? nsid.startsWith(pattern.slice(0, -1)) : nsid === pattern;
52
-
53
-
if (blocklist.some(matches)) return false;
54
-
if (allowlist.length > 0) return allowlist.some(matches);
55
-
return true;
56
-
}
45
+
export { isNsidAllowed, nsidRequiresWantedDids } from "./match.js";
57
46
58
47
/**
59
48
* Extract filterable fields from a lexicon record's properties.
+2
lib/test/fixtures.ts
+2
lib/test/fixtures.ts
···
21
21
actions: Action[];
22
22
fetches: FetchStep[];
23
23
conditions: Array<{ field: string; operator: string; value: string; comment?: string }>;
24
+
wantedDids: string[];
24
25
active: boolean;
25
26
indexedAt: Date;
26
27
};
···
112
113
actions: [makeWebhookAction()],
113
114
fetches: [],
114
115
conditions: [],
116
+
wantedDids: [],
115
117
active: true,
116
118
indexedAt: new Date("2024-01-01T00:00:00.000Z"),
117
119
...overrides,