feat: better action chaining · exosphere.site/airglow@50665b7

+225 -28

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 type { Action, FetchStep } from "../../lib/db/schema.js"; 3 + import { isRecordProducingAction, type Action, type FetchStep } from "../../lib/db/schema.js"; 4 4 import RecordFormBuilder from "./RecordFormBuilder.js"; 5 5 import * as s from "./AutomationForm.css.ts"; 6 6 ··· 33 33 labels: string[]; 34 34 comment: string; 35 35 }; 36 - type ActionDraft = WebhookDraft | RecordDraft | BskyPostDraft; 36 + type PatchRecordDraft = { 37 + type: "patch-record"; 38 + targetCollection: string; 39 + baseRecordUri: string; 40 + recordTemplate: string; 41 + comment: string; 42 + }; 43 + type ActionDraft = WebhookDraft | RecordDraft | BskyPostDraft | PatchRecordDraft; 37 44 38 45 export type AutomationInitial = { 39 46 rkey?: string; ··· 96 103 } 97 104 98 105 // --------------------------------------------------------------------------- 99 - // Record action editor 106 + // Shared NSID schema hook for record action editors 100 107 // --------------------------------------------------------------------------- 101 108 102 - function RecordActionEditor({ 103 - action, 104 - onChange, 105 - placeholders, 106 - }: { 107 - action: RecordDraft; 108 - onChange: (a: RecordDraft) => void; 109 - placeholders: string[]; 110 - }) { 109 + function useNsidSchema(initialCollection?: string) { 111 110 const [targetSchema, setTargetSchema] = useState<RecordSchema | null>(null); 112 111 const [targetSchemaLoading, setTargetSchemaLoading] = useState(false); 113 112 const [targetSchemaError, setTargetSchemaError] = useState(""); ··· 115 114 const debounceRef = useRef<ReturnType<typeof setTimeout> | null>(null); 116 115 const suggestDebounceRef = useRef<ReturnType<typeof setTimeout> | null>(null); 117 116 const lastSuggestPrefix = useRef(""); 118 - const [datalistId] = useState(() => `nsid-target-${Math.random().toString(36).slice(2, 8)}`); 117 + const [datalistId] = useState(() => `nsid-${Math.random().toString(36).slice(2, 8)}`); 119 118 const initialFetched = useRef(false); 120 119 121 120 const fetchTargetSchema = useCallback((nsid: string) => { 122 121 if (debounceRef.current) clearTimeout(debounceRef.current); 123 - if (!nsid) { 124 - setTargetSchema(null); 125 - setTargetSchemaError(""); 126 - return; 127 - } 128 - if (!NSID_RE.test(nsid)) { 122 + if (!nsid || !NSID_RE.test(nsid)) { 129 123 setTargetSchema(null); 130 124 setTargetSchemaError(""); 131 125 return; ··· 141 135 setTargetSchema(null); 142 136 } else { 143 137 setTargetSchema(data.record ?? null); 144 - if (!data.record) { 145 - setTargetSchemaError("No record schema found for this collection"); 146 - } 138 + if (!data.record) setTargetSchemaError("No record schema found for this collection"); 147 139 } 148 140 } catch { 149 141 setTargetSchemaError("Failed to fetch target collection schema"); ··· 174 166 }, 300); 175 167 }, []); 176 168 177 - // Fetch schema for pre-filled targetCollection 178 - if (!initialFetched.current && action.targetCollection) { 169 + if (!initialFetched.current && initialCollection) { 179 170 initialFetched.current = true; 180 - fetchTargetSchema(action.targetCollection); 171 + fetchTargetSchema(initialCollection); 181 172 } 182 173 174 + return { 175 + targetSchema, 176 + targetSchemaLoading, 177 + targetSchemaError, 178 + nsidSuggestions, 179 + datalistId, 180 + fetchTargetSchema, 181 + fetchSuggestions, 182 + }; 183 + } 184 + 185 + // --------------------------------------------------------------------------- 186 + // Record action editor 187 + // --------------------------------------------------------------------------- 188 + 189 + function RecordActionEditor({ 190 + action, 191 + onChange, 192 + placeholders, 193 + }: { 194 + action: RecordDraft; 195 + onChange: (a: RecordDraft) => void; 196 + placeholders: string[]; 197 + }) { 198 + const { 199 + targetSchema, 200 + targetSchemaLoading, 201 + targetSchemaError, 202 + nsidSuggestions, 203 + datalistId, 204 + fetchTargetSchema, 205 + fetchSuggestions, 206 + } = useNsidSchema(action.targetCollection); 207 + 183 208 return ( 184 209 <> 185 210 <div class={s.fieldGroup}> ··· 330 355 } 331 356 332 357 // --------------------------------------------------------------------------- 358 + // Patch Record action editor 359 + // --------------------------------------------------------------------------- 360 + 361 + function PatchRecordActionEditor({ 362 + action, 363 + onChange, 364 + placeholders, 365 + }: { 366 + action: PatchRecordDraft; 367 + onChange: (a: PatchRecordDraft) => void; 368 + placeholders: string[]; 369 + }) { 370 + const { 371 + targetSchema, 372 + targetSchemaLoading, 373 + targetSchemaError, 374 + nsidSuggestions, 375 + datalistId, 376 + fetchTargetSchema, 377 + fetchSuggestions, 378 + } = useNsidSchema(action.targetCollection); 379 + 380 + return ( 381 + <> 382 + <div class={s.fieldGroup}> 383 + <label class={s.label}>Target Collection</label> 384 + <input 385 + class={s.input} 386 + type="text" 387 + list={datalistId} 388 + placeholder="e.g. site.standard.document" 389 + value={action.targetCollection} 390 + onInput={(e: Event) => { 391 + const val = (e.target as HTMLInputElement).value; 392 + onChange({ ...action, targetCollection: val }); 393 + fetchTargetSchema(val); 394 + fetchSuggestions(val); 395 + }} 396 + required 397 + /> 398 + <span class={s.hint}>NSID of the collection containing the record to update</span> 399 + <datalist id={datalistId}> 400 + {nsidSuggestions.map((nsid) => ( 401 + <option key={nsid} value={nsid} /> 402 + ))} 403 + </datalist> 404 + </div> 405 + 406 + <div class={s.fieldGroup}> 407 + <label class={s.label}>Base Record URI</label> 408 + <input 409 + class={s.input} 410 + type="text" 411 + placeholder="at://{{event.did}}/{{event.commit.collection}}/{{event.commit.rkey}}" 412 + value={action.baseRecordUri} 413 + onInput={(e: Event) => 414 + onChange({ ...action, baseRecordUri: (e.target as HTMLInputElement).value }) 415 + } 416 + required 417 + /> 418 + <span class={s.hint}>AT URI of the record to update. Supports {"{{placeholders}}"}.</span> 419 + </div> 420 + 421 + {(targetSchema || targetSchemaLoading) && ( 422 + <RecordFormBuilder 423 + schema={targetSchema} 424 + loading={targetSchemaLoading} 425 + error={targetSchemaError} 426 + placeholders={placeholders} 427 + initialTemplate={action.recordTemplate || undefined} 428 + onChange={(tpl) => onChange({ ...action, recordTemplate: tpl })} 429 + /> 430 + )} 431 + 432 + {!targetSchema && !targetSchemaLoading && action.targetCollection && ( 433 + <div class={s.fieldGroup}> 434 + <label class={s.label}>Patch Template</label> 435 + {targetSchemaError && <span class={s.errorText}>{targetSchemaError}</span>} 436 + <textarea 437 + class={s.textarea} 438 + placeholder={'{\n "bskyPostRef": "{{action0.uri}}",\n "updatedAt": "{{now}}"\n}'} 439 + value={action.recordTemplate} 440 + onInput={(e: Event) => 441 + onChange({ 442 + ...action, 443 + recordTemplate: (e.target as HTMLTextAreaElement).value, 444 + }) 445 + } 446 + required 447 + /> 448 + <span class={s.hint}> 449 + Only include the fields you want to change. They will be merged on top of the existing 450 + record. 451 + </span> 452 + </div> 453 + )} 454 + </> 455 + ); 456 + } 457 + 458 + // --------------------------------------------------------------------------- 333 459 // Copy-to-clipboard placeholder 334 460 // --------------------------------------------------------------------------- 335 461 ··· 375 501 textTemplate: a.textTemplate, 376 502 langs: a.langs ?? [], 377 503 labels: a.labels ?? [], 504 + comment: a.comment ?? "", 505 + }; 506 + } 507 + if (a.$type === "patch-record") { 508 + return { 509 + type: "patch-record", 510 + targetCollection: a.targetCollection, 511 + baseRecordUri: a.baseRecordUri, 512 + recordTemplate: a.recordTemplate, 378 513 comment: a.comment ?? "", 379 514 }; 380 515 } ··· 562 697 setFetches((prev) => prev.map((f, i) => (i === index ? { ...f, [key]: val } : f))); 563 698 }, []); 564 699 565 - const addAction = useCallback((type: "webhook" | "record" | "bsky-post") => { 700 + const addAction = useCallback((type: "webhook" | "record" | "bsky-post" | "patch-record") => { 566 701 if (type === "webhook") { 567 702 setActions((prev) => [...prev, { type: "webhook", callbackUrl: "", comment: "" }]); 568 703 } else if (type === "bsky-post") { ··· 570 705 ...prev, 571 706 { type: "bsky-post", textTemplate: "", langs: [], labels: [], comment: "" }, 572 707 ]); 708 + } else if (type === "patch-record") { 709 + setActions((prev) => [ 710 + ...prev, 711 + { 712 + type: "patch-record", 713 + targetCollection: "", 714 + baseRecordUri: "", 715 + recordTemplate: "", 716 + comment: "", 717 + }, 718 + ]); 573 719 } else { 574 720 setActions((prev) => [ 575 721 ...prev, ··· 618 764 ...comment, 619 765 }; 620 766 } 767 + if (a.type === "patch-record") { 768 + return { 769 + type: "patch-record", 770 + targetCollection: a.targetCollection, 771 + baseRecordUri: a.baseRecordUri, 772 + recordTemplate: a.recordTemplate, 773 + ...comment, 774 + }; 775 + } 621 776 return { 622 777 type: "record", 623 778 targetCollection: a.targetCollection, ··· 657 812 [previewPayload, isEdit], 658 813 ); 659 814 815 + const actionResultPlaceholders = actions.flatMap((a, i) => 816 + isRecordProducingAction(a.type) ? [`action${i}.uri`, `action${i}.cid`, `action${i}.rkey`] : [], 817 + ); 818 + 660 819 const allPlaceholders = [ 661 820 ...BUILTIN_PLACEHOLDERS, 662 821 ...fields.map((f) => `event.commit.record.${f.path}`), 663 822 ...fetches 664 823 .filter((f) => f.name) 665 824 .flatMap((f) => [`${f.name}.uri`, `${f.name}.cid`, `${f.name}.record.*`]), 825 + ...actionResultPlaceholders, 666 826 ]; 667 827 668 828 const conditionFields = [...BUILTIN_CONDITION_FIELDS, ...fields]; ··· 817 977 </div> 818 978 )} 819 979 980 + {actionResultPlaceholders.length > 0 && ( 981 + <div class={s.placeholderGroup}> 982 + <div class={s.placeholderGroupTitle}>Action results</div> 983 + <span class={s.hint}> 984 + Results from preceding actions (available in subsequent actions). 985 + </span> 986 + {actions.map((a, i) => 987 + isRecordProducingAction(a.type) ? ( 988 + <div key={i}> 989 + <CopyPlaceholder value={`action${i}.uri`}> 990 + <span class={s.placeholderDesc}>AT URI from action {i + 1}</span> 991 + </CopyPlaceholder> 992 + <CopyPlaceholder value={`action${i}.cid`}> 993 + <span class={s.placeholderDesc}>Content hash from action {i + 1}</span> 994 + </CopyPlaceholder> 995 + <CopyPlaceholder value={`action${i}.rkey`}> 996 + <span class={s.placeholderDesc}>Record key from action {i + 1}</span> 997 + </CopyPlaceholder> 998 + </div> 999 + ) : null, 1000 + )} 1001 + </div> 1002 + )} 1003 + 820 1004 <div class={s.placeholderGroup}> 821 1005 <div class={s.placeholderGroupTitle}>Helpers</div> 822 1006 <CopyPlaceholder value="now"> ··· 1008 1192 <div class={s.actionsSection}> 1009 1193 <div> 1010 1194 <h3>Actions</h3> 1011 - <p class={s.hint}>Add one or more actions to execute when a matching event occurs.</p> 1195 + <p class={s.hint}> 1196 + Actions run sequentially. Each action can reference the results of preceding ones. 1197 + </p> 1012 1198 </div> 1013 1199 1014 1200 {actions.map((action, i) => ( ··· 1019 1205 ? "Webhook" 1020 1206 : action.type === "bsky-post" 1021 1207 ? "Bluesky Post" 1022 - : "Record"}{" "} 1208 + : action.type === "patch-record" 1209 + ? "Patch Record" 1210 + : "Record"}{" "} 1023 1211 {actions.filter((a, j) => a.type === action.type && j <= i).length} 1024 1212 </span> 1025 1213 <button type="button" class={s.removeBtn} onClick={() => removeAction(i)}> ··· 1030 1218 <WebhookActionEditor action={action} onChange={(a) => updateAction(i, a)} /> 1031 1219 ) : action.type === "bsky-post" ? ( 1032 1220 <BskyPostActionEditor action={action} onChange={(a) => updateAction(i, a)} /> 1221 + ) : action.type === "patch-record" ? ( 1222 + <PatchRecordActionEditor 1223 + action={action} 1224 + onChange={(a) => updateAction(i, a)} 1225 + placeholders={allPlaceholders} 1226 + /> 1033 1227 ) : ( 1034 1228 <RecordActionEditor 1035 1229 action={action} ··· 1061 1255 </button> 1062 1256 <button type="button" class={s.addBtn} onClick={() => addAction("bsky-post")}> 1063 1257 + Add Bluesky post 1258 + </button> 1259 + <button type="button" class={s.addBtn} onClick={() => addAction("patch-record")}> 1260 + + Patch record 1064 1261 </button> 1065 1262 </div> 1066 1263 </div>

+82 -4

app/routes/api/automations/[rkey].ts

··· 9 9 type WebhookAction, 10 10 type RecordAction, 11 11 type BskyPostAction, 12 + type PatchRecordAction, 12 13 type FetchStep, 13 14 } from "@/db/schema.js"; 14 15 import { isValidNsid } from "@/lexicons/resolver.js"; ··· 21 22 } from "@/automations/pds.js"; 22 23 import { verifyCallback } from "@/automations/verify.js"; 23 24 import { assertPublicUrl, UrlGuardError } from "@/url-guard.js"; 24 - import { validateTemplate, validateTextTemplate, validateFetchStep } from "@/actions/template.js"; 25 + import { 26 + validateTemplate, 27 + validateTextTemplate, 28 + validateBaseRecordUri, 29 + validateFetchStep, 30 + } from "@/actions/template.js"; 25 31 import { notifyAutomationChange } from "@/jetstream/consumer.js"; 26 32 27 33 type ActionInput = ··· 32 38 textTemplate: string; 33 39 langs?: string[]; 34 40 labels?: string[]; 41 + comment?: string; 42 + } 43 + | { 44 + type: "patch-record"; 45 + targetCollection: string; 46 + baseRecordUri: string; 47 + recordTemplate: string; 35 48 comment?: string; 36 49 }; 37 50 ··· 215 228 216 229 const newLocalActions: Action[] = []; 217 230 const newPdsActions: PdsAction[] = []; 231 + const actionResultNames: string[] = []; 218 232 219 - for (const input of body.actions) { 233 + for (const [actionIndex, input] of body.actions.entries()) { 220 234 if (input.type === "webhook") { 221 235 if (!input.callbackUrl) { 222 236 return c.json({ error: "callbackUrl is required for webhook actions" }, 400); ··· 259 273 if (!input.recordTemplate) { 260 274 return c.json({ error: "recordTemplate is required for record actions" }, 400); 261 275 } 262 - const templateValidation = validateTemplate(input.recordTemplate, fetchNames); 276 + const templateValidation = validateTemplate( 277 + input.recordTemplate, 278 + fetchNames, 279 + actionResultNames, 280 + ); 263 281 if (!templateValidation.valid) { 264 282 return c.json({ error: templateValidation.error }, 400); 265 283 } ··· 276 294 recordTemplate: input.recordTemplate, 277 295 ...(input.comment ? { comment: input.comment } : {}), 278 296 }); 297 + actionResultNames.push(`action${actionIndex}`); 279 298 } else if (input.type === "bsky-post") { 280 299 if (!input.textTemplate || !input.textTemplate.trim()) { 281 300 return c.json({ error: "textTemplate is required for bsky-post actions" }, 400); 282 301 } 283 - const textValidation = validateTextTemplate(input.textTemplate, fetchNames); 302 + const textValidation = validateTextTemplate( 303 + input.textTemplate, 304 + fetchNames, 305 + actionResultNames, 306 + ); 284 307 if (!textValidation.valid) { 285 308 return c.json({ error: textValidation.error }, 400); 286 309 } ··· 317 340 ...(labels && labels.length > 0 ? { labels } : {}), 318 341 ...(input.comment ? { comment: input.comment } : {}), 319 342 }); 343 + actionResultNames.push(`action${actionIndex}`); 344 + } else if (input.type === "patch-record") { 345 + if (!input.targetCollection) { 346 + return c.json({ error: "targetCollection is required for patch-record actions" }, 400); 347 + } 348 + if (!isValidNsid(input.targetCollection)) { 349 + return c.json({ error: "Invalid target collection NSID" }, 400); 350 + } 351 + if (!input.baseRecordUri) { 352 + return c.json({ error: "baseRecordUri is required for patch-record actions" }, 400); 353 + } 354 + const uriValidation = validateBaseRecordUri( 355 + input.baseRecordUri, 356 + fetchNames, 357 + actionResultNames, 358 + ); 359 + if (!uriValidation.valid) { 360 + return c.json({ error: uriValidation.error }, 400); 361 + } 362 + if (!input.recordTemplate) { 363 + return c.json({ error: "recordTemplate is required for patch-record actions" }, 400); 364 + } 365 + const templateValidation = validateTemplate( 366 + input.recordTemplate, 367 + fetchNames, 368 + actionResultNames, 369 + ); 370 + if (!templateValidation.valid) { 371 + return c.json({ error: templateValidation.error }, 400); 372 + } 373 + 374 + newLocalActions.push({ 375 + $type: "patch-record", 376 + targetCollection: input.targetCollection, 377 + baseRecordUri: input.baseRecordUri, 378 + recordTemplate: input.recordTemplate, 379 + ...(input.comment ? { comment: input.comment } : {}), 380 + } satisfies PatchRecordAction); 381 + newPdsActions.push({ 382 + $type: "run.airglow.automation#patchRecordAction", 383 + targetCollection: input.targetCollection, 384 + baseRecordUri: input.baseRecordUri, 385 + recordTemplate: input.recordTemplate, 386 + ...(input.comment ? { comment: input.comment } : {}), 387 + }); 388 + actionResultNames.push(`action${actionIndex}`); 320 389 } else { 321 390 return c.json({ error: "Invalid action type" }, 400); 322 391 } ··· 362 431 textTemplate: a.textTemplate, 363 432 ...(a.langs && a.langs.length > 0 ? { langs: a.langs } : {}), 364 433 ...(a.labels && a.labels.length > 0 ? { labels: a.labels } : {}), 434 + ...(a.comment ? { comment: a.comment } : {}), 435 + }; 436 + } 437 + if (a.$type === "patch-record") { 438 + return { 439 + $type: "run.airglow.automation#patchRecordAction", 440 + targetCollection: a.targetCollection, 441 + baseRecordUri: a.baseRecordUri, 442 + recordTemplate: a.recordTemplate, 365 443 ...(a.comment ? { comment: a.comment } : {}), 366 444 }; 367 445 }

+73 -4

app/routes/api/automations/index.ts

··· 8 8 type WebhookAction, 9 9 type RecordAction, 10 10 type BskyPostAction, 11 + type PatchRecordAction, 11 12 type FetchStep, 12 13 } from "@/db/schema.js"; 13 14 import { config } from "@/config.js"; ··· 20 21 type PdsAction, 21 22 type PdsFetchStep, 22 23 } from "@/automations/pds.js"; 23 - import { validateTemplate, validateTextTemplate, validateFetchStep } from "@/actions/template.js"; 24 + import { 25 + validateTemplate, 26 + validateTextTemplate, 27 + validateBaseRecordUri, 28 + validateFetchStep, 29 + } from "@/actions/template.js"; 24 30 import { notifyAutomationChange } from "@/jetstream/consumer.js"; 25 31 26 32 type ActionInput = ··· 31 37 textTemplate: string; 32 38 langs?: string[]; 33 39 labels?: string[]; 40 + comment?: string; 41 + } 42 + | { 43 + type: "patch-record"; 44 + targetCollection: string; 45 + baseRecordUri: string; 46 + recordTemplate: string; 34 47 comment?: string; 35 48 }; 36 49 ··· 176 189 // Validate each action and build local + PDS action arrays 177 190 const localActions: Action[] = []; 178 191 const pdsActions: PdsAction[] = []; 192 + const actionResultNames: string[] = []; 179 193 180 - for (const input of body.actions) { 194 + for (const [actionIndex, input] of body.actions.entries()) { 181 195 if (input.type === "webhook") { 182 196 if (!input.callbackUrl) { 183 197 return c.json({ error: "callbackUrl is required for webhook actions" }, 400); ··· 214 228 if (!input.recordTemplate) { 215 229 return c.json({ error: "recordTemplate is required for record actions" }, 400); 216 230 } 217 - const templateValidation = validateTemplate(input.recordTemplate, fetchNames); 231 + const templateValidation = validateTemplate( 232 + input.recordTemplate, 233 + fetchNames, 234 + actionResultNames, 235 + ); 218 236 if (!templateValidation.valid) { 219 237 return c.json({ error: templateValidation.error }, 400); 220 238 } ··· 231 249 recordTemplate: input.recordTemplate, 232 250 ...(input.comment ? { comment: input.comment } : {}), 233 251 }); 252 + actionResultNames.push(`action${actionIndex}`); 234 253 } else if (input.type === "bsky-post") { 235 254 if (!input.textTemplate || !input.textTemplate.trim()) { 236 255 return c.json({ error: "textTemplate is required for bsky-post actions" }, 400); 237 256 } 238 - const textValidation = validateTextTemplate(input.textTemplate, fetchNames); 257 + const textValidation = validateTextTemplate( 258 + input.textTemplate, 259 + fetchNames, 260 + actionResultNames, 261 + ); 239 262 if (!textValidation.valid) { 240 263 return c.json({ error: textValidation.error }, 400); 241 264 } ··· 272 295 ...(labels && labels.length > 0 ? { labels } : {}), 273 296 ...(input.comment ? { comment: input.comment } : {}), 274 297 }); 298 + actionResultNames.push(`action${actionIndex}`); 299 + } else if (input.type === "patch-record") { 300 + if (!input.targetCollection) { 301 + return c.json({ error: "targetCollection is required for patch-record actions" }, 400); 302 + } 303 + if (!isValidNsid(input.targetCollection)) { 304 + return c.json({ error: "Invalid target collection NSID" }, 400); 305 + } 306 + if (!input.baseRecordUri) { 307 + return c.json({ error: "baseRecordUri is required for patch-record actions" }, 400); 308 + } 309 + const uriValidation = validateBaseRecordUri( 310 + input.baseRecordUri, 311 + fetchNames, 312 + actionResultNames, 313 + ); 314 + if (!uriValidation.valid) { 315 + return c.json({ error: uriValidation.error }, 400); 316 + } 317 + if (!input.recordTemplate) { 318 + return c.json({ error: "recordTemplate is required for patch-record actions" }, 400); 319 + } 320 + const templateValidation = validateTemplate( 321 + input.recordTemplate, 322 + fetchNames, 323 + actionResultNames, 324 + ); 325 + if (!templateValidation.valid) { 326 + return c.json({ error: templateValidation.error }, 400); 327 + } 328 + 329 + localActions.push({ 330 + $type: "patch-record", 331 + targetCollection: input.targetCollection, 332 + baseRecordUri: input.baseRecordUri, 333 + recordTemplate: input.recordTemplate, 334 + ...(input.comment ? { comment: input.comment } : {}), 335 + } satisfies PatchRecordAction); 336 + pdsActions.push({ 337 + $type: "run.airglow.automation#patchRecordAction", 338 + targetCollection: input.targetCollection, 339 + baseRecordUri: input.baseRecordUri, 340 + recordTemplate: input.recordTemplate, 341 + ...(input.comment ? { comment: input.comment } : {}), 342 + }); 343 + actionResultNames.push(`action${actionIndex}`); 275 344 } else { 276 345 return c.json({ error: "Invalid action type" }, 400); 277 346 }

+2 -6

app/routes/index.tsx

··· 48 48 lexicon, deliver webhooks, create records on your PDS, and track every run. 49 49 </p> 50 50 {user ? ( 51 - <Button href="/dashboard"> 52 - Go to Dashboard 53 - </Button> 51 + <Button href="/dashboard">Go to Dashboard</Button> 54 52 ) : ( 55 - <Button href="/auth/login"> 56 - Get Started 57 - </Button> 53 + <Button href="/auth/login">Get Started</Button> 58 54 )} 59 55 </section> 60 56

+8 -4

app/routes/og-image.tsx

··· 77 77 <body> 78 78 <div class="og"> 79 79 <div class="logo"> 80 - <svg viewBox="0 0 64 64" fill="none" xmlns="http://www.w3.org/2000/svg" width="80" height="80"> 80 + <svg 81 + viewBox="0 0 64 64" 82 + fill="none" 83 + xmlns="http://www.w3.org/2000/svg" 84 + width="80" 85 + height="80" 86 + > 81 87 <circle cx="32" cy="32" r="10" fill="#E8923A" /> 82 88 <circle cx="32" cy="32" r="20" stroke="#E8923A" stroke-width="3.5" opacity="0.45" /> 83 89 <circle cx="32" cy="32" r="29" stroke="#E8923A" stroke-width="2.5" opacity="0.2" /> 84 90 </svg> 85 91 </div> 86 92 <h1 class="title">Airglow</h1> 87 - <p class="tagline"> 88 - Webhooks & Automations for the AT Protocol 89 - </p> 93 + <p class="tagline">Webhooks & Automations for the AT Protocol</p> 90 94 <span class="url">airglow.run</span> 91 95 </div> 92 96 </body>

+64 -2

lexicons/run/airglow/automation.json

··· 37 37 }, 38 38 "actions": { 39 39 "type": "array", 40 - "description": "Actions to execute when a matching event occurs.", 40 + "description": "Actions to execute sequentially when a matching event occurs. Each action can reference the results of preceding actions.", 41 41 "minLength": 1, 42 42 "maxLength": 10, 43 43 "items": { 44 44 "type": "union", 45 - "refs": ["#webhookAction", "#recordAction"] 45 + "refs": ["#webhookAction", "#recordAction", "#bskyPostAction", "#patchRecordAction"] 46 46 } 47 47 }, 48 48 "conditions": { ··· 138 138 "comment": { 139 139 "type": "string", 140 140 "description": "Optional user note about this fetch step.", 141 + "maxLength": 512 142 + } 143 + } 144 + }, 145 + "bskyPostAction": { 146 + "type": "object", 147 + "description": "Create a Bluesky post when a matching event occurs.", 148 + "required": ["textTemplate"], 149 + "properties": { 150 + "textTemplate": { 151 + "type": "string", 152 + "description": "Plain text template with {{placeholder}} expressions resolved from event data.", 153 + "maxLength": 10240 154 + }, 155 + "langs": { 156 + "type": "array", 157 + "description": "BCP-47 language tags for the post.", 158 + "maxLength": 3, 159 + "items": { 160 + "type": "string", 161 + "maxLength": 16 162 + } 163 + }, 164 + "labels": { 165 + "type": "array", 166 + "description": "Self-label values for content warnings.", 167 + "maxLength": 4, 168 + "items": { 169 + "type": "string", 170 + "knownValues": ["sexual", "nudity", "porn", "graphic-media"] 171 + } 172 + }, 173 + "comment": { 174 + "type": "string", 175 + "description": "Optional user note about this action.", 176 + "maxLength": 512 177 + } 178 + } 179 + }, 180 + "patchRecordAction": { 181 + "type": "object", 182 + "description": "Update an existing record on the user's PDS by merging template fields on top of the current record.", 183 + "required": ["targetCollection", "baseRecordUri", "recordTemplate"], 184 + "properties": { 185 + "targetCollection": { 186 + "type": "string", 187 + "description": "NSID of the collection containing the record to update.", 188 + "maxLength": 256 189 + }, 190 + "baseRecordUri": { 191 + "type": "string", 192 + "description": "AT URI template resolving to the record to update, e.g. 'at://{{event.did}}/{{event.commit.collection}}/{{event.commit.rkey}}'.", 193 + "maxLength": 2048 194 + }, 195 + "recordTemplate": { 196 + "type": "string", 197 + "description": "JSON template with {{placeholder}} expressions. Fields are shallow-merged on top of the fetched base record.", 198 + "maxLength": 10240 199 + }, 200 + "comment": { 201 + "type": "string", 202 + "description": "Optional user note about this action.", 141 203 "maxLength": 512 142 204 } 143 205 }

+7 -4

lib/actions/bsky-post.ts

··· 3 3 import { createArbitraryRecord } from "../automations/pds.js"; 4 4 import { renderTextTemplate, type FetchContext } from "./template.js"; 5 5 import { detectFacets } from "./richtext.js"; 6 + import type { ActionResult } from "./executor.js"; 6 7 import type { MatchedEvent } from "../jetstream/consumer.js"; 7 8 8 9 const TARGET_COLLECTION = "app.bsky.feed.post"; ··· 12 13 match: MatchedEvent, 13 14 action: BskyPostAction, 14 15 fetchContext?: FetchContext, 15 - ): Promise<{ statusCode: number; error?: string }> { 16 + ): Promise<ActionResult> { 16 17 const { automation, event } = match; 17 18 18 19 let text: string; ··· 54 55 } 55 56 56 57 try { 57 - await createArbitraryRecord(automation.did, TARGET_COLLECTION, record); 58 - return { statusCode: 200 }; 58 + const created = await createArbitraryRecord(automation.did, TARGET_COLLECTION, record); 59 + return { statusCode: 200, uri: created.uri, cid: created.cid }; 59 60 } catch (err) { 60 61 const message = err instanceof Error ? err.message : String(err); 61 62 const statusMatch = message.match(/$(\d{3})$/); ··· 131 132 match: MatchedEvent, 132 133 actionIndex: number, 133 134 fetchContext?: FetchContext, 134 - ) { 135 + ): Promise<ActionResult> { 135 136 const action = match.automation.actions[actionIndex] as BskyPostAction; 136 137 const result = await execute(match, action, fetchContext); 137 138 ··· 150 151 if (!isSuccess(result.statusCode) && isRetryable(result.statusCode)) { 151 152 scheduleRetry(match, actionIndex, 0, fetchContext); 152 153 } 154 + 155 + return result; 153 156 }

+16 -4

lib/actions/executor.ts

··· 4 4 import { renderTemplate, type FetchContext } from "./template.js"; 5 5 import type { MatchedEvent } from "../jetstream/consumer.js"; 6 6 7 + /** Result returned by every action executor for chaining. */ 8 + export type ActionResult = { 9 + statusCode: number; 10 + error?: string; 11 + /** AT URI of the created/updated record (record-producing actions only). */ 12 + uri?: string; 13 + /** CID of the created/updated record (record-producing actions only). */ 14 + cid?: string; 15 + }; 16 + 7 17 const RETRY_DELAYS = [5_000, 30_000]; 8 18 9 19 async function execute( 10 20 match: MatchedEvent, 11 21 action: RecordAction, 12 22 fetchContext?: FetchContext, 13 - ): Promise<{ statusCode: number; error?: string }> { 23 + ): Promise<ActionResult> { 14 24 const { automation, event } = match; 15 25 16 26 let record: Record<string, unknown>; ··· 24 34 } 25 35 26 36 try { 27 - await createArbitraryRecord(automation.did, action.targetCollection, record); 28 - return { statusCode: 200 }; 37 + const created = await createArbitraryRecord(automation.did, action.targetCollection, record); 38 + return { statusCode: 200, uri: created.uri, cid: created.cid }; 29 39 } catch (err) { 30 40 const message = err instanceof Error ? err.message : String(err); 31 41 const statusMatch = message.match(/$(\d{3})$/); ··· 104 114 match: MatchedEvent, 105 115 actionIndex: number, 106 116 fetchContext?: FetchContext, 107 - ) { 117 + ): Promise<ActionResult> { 108 118 const action = match.automation.actions[actionIndex] as RecordAction; 109 119 const result = await execute(match, action, fetchContext); 110 120 ··· 126 136 if (!isSuccess(result.statusCode) && isRetryable(result.statusCode)) { 127 137 scheduleRetry(match, actionIndex, 0, fetchContext); 128 138 } 139 + 140 + return result; 129 141 }

+2 -12

lib/actions/fetcher.ts

··· 1 1 import { fetchRecord } from "../pds/resolver.js"; 2 2 import type { FetchStep } from "../db/schema.js"; 3 3 import type { JetstreamEvent } from "../jetstream/matcher.js"; 4 - import { resolveEventPlaceholder, PLACEHOLDER_RE, type FetchContext } from "./template.js"; 4 + import { resolveUriTemplate, type FetchContext } from "./template.js"; 5 5 6 6 // AT URI format: at://did/collection/rkey 7 7 const AT_URI_RE = /^at:\/\/[^/]+\/[^/]+\/[^/]+$/; 8 8 9 - /** Resolve a fetch URI template against event data. Returns empty string on non-string values. */ 10 - function resolveUri(uriTemplate: string, event: JetstreamEvent, ownerDid: string): string { 11 - return uriTemplate.replace(PLACEHOLDER_RE, (_, path: string) => { 12 - const value = resolveEventPlaceholder(path.trim(), event, ownerDid); 13 - if (typeof value === "string") return value; 14 - // Non-string values (objects, arrays, undefined) can't form valid AT URIs 15 - return ""; 16 - }); 17 - } 18 - 19 9 /** Resolve all fetch steps, returning available context and any errors. */ 20 10 export async function resolveFetches( 21 11 steps: FetchStep[], ··· 28 18 await Promise.all( 29 19 steps.map(async (step) => { 30 20 try { 31 - const resolvedUri = resolveUri(step.uri, event, ownerDid); 21 + const resolvedUri = resolveUriTemplate(step.uri, event, undefined, ownerDid); 32 22 if (!resolvedUri || !AT_URI_RE.test(resolvedUri)) { 33 23 errors.push({ 34 24 name: step.name,

+166

lib/actions/patch-record.ts

··· 1 + import { db } from "../db/index.js"; 2 + import { deliveryLogs, type PatchRecordAction } from "../db/schema.js"; 3 + import { patchArbitraryRecord } from "../automations/pds.js"; 4 + import { fetchRecord, parseAtUri } from "../pds/resolver.js"; 5 + import { renderTemplate, resolveUriTemplate, type FetchContext } from "./template.js"; 6 + import type { ActionResult } from "./executor.js"; 7 + import type { MatchedEvent } from "../jetstream/consumer.js"; 8 + 9 + const RETRY_DELAYS = [5_000, 30_000]; 10 + 11 + async function execute( 12 + match: MatchedEvent, 13 + action: PatchRecordAction, 14 + fetchContext?: FetchContext, 15 + ): Promise<ActionResult> { 16 + const { automation, event } = match; 17 + 18 + const resolvedUri = resolveUriTemplate(action.baseRecordUri, event, fetchContext, automation.did); 19 + let parsed: { did: string; collection: string; rkey: string }; 20 + try { 21 + parsed = parseAtUri(resolvedUri); 22 + } catch { 23 + return { statusCode: 0, error: `Invalid base record URI: ${resolvedUri}` }; 24 + } 25 + 26 + // Fetch base record and render patch template in parallel (they're independent) 27 + const [fetchResult, templateResult] = await Promise.allSettled([ 28 + fetchRecord(resolvedUri), 29 + renderTemplate(action.recordTemplate, event, fetchContext, automation.did), 30 + ]); 31 + 32 + if (fetchResult.status === "rejected") { 33 + const err = fetchResult.reason; 34 + return { 35 + statusCode: 0, 36 + error: `Failed to fetch base record: ${err instanceof Error ? err.message : String(err)}`, 37 + }; 38 + } 39 + if (templateResult.status === "rejected") { 40 + const err = templateResult.reason; 41 + return { 42 + statusCode: 0, 43 + error: `Template error: ${err instanceof Error ? err.message : String(err)}`, 44 + }; 45 + } 46 + 47 + const baseRecord = fetchResult.value.record; 48 + const patch = templateResult.value; 49 + 50 + // Shallow merge: strip $type from base (putArbitraryRecord re-adds it) 51 + const { $type: _, ...baseFields } = baseRecord; 52 + const merged = { ...baseFields, ...patch }; 53 + 54 + try { 55 + const result = await patchArbitraryRecord( 56 + automation.did, 57 + parsed.collection, 58 + parsed.rkey, 59 + merged, 60 + ); 61 + return { statusCode: 200, uri: result.uri, cid: result.cid }; 62 + } catch (err) { 63 + const message = err instanceof Error ? err.message : String(err); 64 + const statusMatch = message.match(/$(\d{3})$/); 65 + const statusCode = statusMatch ? Number(statusMatch[1]) : 0; 66 + return { statusCode, error: message }; 67 + } 68 + } 69 + 70 + async function logDelivery( 71 + automationUri: string, 72 + actionIndex: number, 73 + eventTimeUs: number, 74 + payload: string | null, 75 + statusCode: number, 76 + error: string | null, 77 + attempt: number, 78 + ) { 79 + await db.insert(deliveryLogs).values({ 80 + automationUri, 81 + actionIndex, 82 + eventTimeUs, 83 + payload, 84 + statusCode, 85 + error, 86 + attempt, 87 + createdAt: new Date(), 88 + }); 89 + } 90 + 91 + function isSuccess(code: number): boolean { 92 + return code === 200; 93 + } 94 + 95 + function isRetryable(code: number): boolean { 96 + return code >= 500 || code === 0; 97 + } 98 + 99 + function scheduleRetry( 100 + match: MatchedEvent, 101 + actionIndex: number, 102 + retryIndex: number, 103 + fetchContext?: FetchContext, 104 + ) { 105 + if (retryIndex >= RETRY_DELAYS.length) return; 106 + 107 + setTimeout(async () => { 108 + try { 109 + const action = match.automation.actions[actionIndex] as PatchRecordAction; 110 + const result = await execute(match, action, fetchContext); 111 + const body = JSON.stringify({ 112 + targetCollection: action.targetCollection, 113 + baseRecordUri: action.baseRecordUri, 114 + recordTemplate: action.recordTemplate, 115 + }); 116 + 117 + await logDelivery( 118 + match.automation.uri, 119 + actionIndex, 120 + match.event.time_us, 121 + isSuccess(result.statusCode) ? null : body, 122 + result.statusCode, 123 + result.error ?? null, 124 + retryIndex + 2, 125 + ); 126 + 127 + if (!isSuccess(result.statusCode) && isRetryable(result.statusCode)) { 128 + scheduleRetry(match, actionIndex, retryIndex + 1, fetchContext); 129 + } 130 + } catch (err) { 131 + console.error("Patch-record retry error:", err); 132 + } 133 + }, RETRY_DELAYS[retryIndex]); 134 + } 135 + 136 + /** Execute a patch-record action for a matched event. */ 137 + export async function executePatchRecord( 138 + match: MatchedEvent, 139 + actionIndex: number, 140 + fetchContext?: FetchContext, 141 + ): Promise<ActionResult> { 142 + const action = match.automation.actions[actionIndex] as PatchRecordAction; 143 + const result = await execute(match, action, fetchContext); 144 + 145 + const body = JSON.stringify({ 146 + targetCollection: action.targetCollection, 147 + baseRecordUri: action.baseRecordUri, 148 + recordTemplate: action.recordTemplate, 149 + }); 150 + 151 + await logDelivery( 152 + match.automation.uri, 153 + actionIndex, 154 + match.event.time_us, 155 + isSuccess(result.statusCode) ? null : body, 156 + result.statusCode, 157 + result.error ?? null, 158 + 1, 159 + ); 160 + 161 + if (!isSuccess(result.statusCode) && isRetryable(result.statusCode)) { 162 + scheduleRetry(match, actionIndex, 0, fetchContext); 163 + } 164 + 165 + return result; 166 + }

+63 -2

lib/actions/template.ts

··· 7 7 8 8 export type FetchContext = Record< 9 9 string, 10 - { uri: string; cid: string; record: Record<string, unknown> } 10 + { uri: string; cid: string; rkey?: string; record: Record<string, unknown> } 11 11 >; 12 12 13 13 /** ··· 105 105 return resolvePlaceholder(path, event, undefined, ownerDid); 106 106 } 107 107 108 + /** Resolve a URI template against event data and full fetch/action context. */ 109 + export function resolveUriTemplate( 110 + uriTemplate: string, 111 + event: JetstreamEvent, 112 + fetchContext?: FetchContext, 113 + ownerDid?: string, 114 + ): string { 115 + return uriTemplate.replace(PLACEHOLDER_RE, (_, path: string) => { 116 + const value = resolvePlaceholder(path.trim(), event, fetchContext, ownerDid); 117 + if (typeof value === "string") return value; 118 + return ""; 119 + }); 120 + } 121 + 108 122 /** Validate template syntax at creation time. */ 109 123 export function validateTemplate( 110 124 template: string, 111 125 fetchNames?: string[], 126 + actionNames?: string[], 112 127 ): { valid: true; placeholders: string[] } | { valid: false; error: string } { 113 128 // Check that the template is valid JSON (ignoring placeholders). 114 129 // Collect placeholder paths, then replace all {{...}} with a plain string ··· 138 153 } 139 154 140 155 const fetchSet = new Set(fetchNames ?? []); 156 + const actionSet = new Set(actionNames ?? []); 141 157 for (const p of placeholders) { 142 158 const call = parseFunctionCall(p); 143 159 const toValidate = call ? call.arg : p; 144 160 if (toValidate === "now" || toValidate === "self" || toValidate.startsWith("event.")) continue; 145 161 const root = toValidate.split(".")[0]!; 146 162 if (fetchSet.has(root)) continue; 163 + if (actionSet.has(root)) continue; 147 164 return { valid: false, error: `Invalid placeholder: {{${p}}}` }; 148 165 } 149 166 ··· 151 168 } 152 169 153 170 const FETCH_NAME_RE = /^[a-zA-Z_][a-zA-Z0-9_]*$/; 171 + const ACTION_NAME_RE = /^action\d+$/; 154 172 const RESERVED_FETCH_NAMES = new Set(["event", "now", "self"]); 155 173 156 174 /** Validate a single fetch step name + URI. */ ··· 162 180 if (!name || !FETCH_NAME_RE.test(name)) { 163 181 return { valid: false, error: `Invalid fetch name: "${name}". Must be a valid identifier.` }; 164 182 } 165 - if (RESERVED_FETCH_NAMES.has(name)) { 183 + if (RESERVED_FETCH_NAMES.has(name) || ACTION_NAME_RE.test(name)) { 166 184 return { valid: false, error: `Fetch name "${name}" is reserved` }; 167 185 } 168 186 if (seenNames.has(name)) { ··· 207 225 return { valid: true, placeholders }; 208 226 } 209 227 228 + /** Validate a base record URI template (for patch-record actions). 229 + * Unlike fetch URIs, these also accept fetch context and action result placeholders. */ 230 + export function validateBaseRecordUri( 231 + uri: string, 232 + fetchNames?: string[], 233 + actionNames?: string[], 234 + ): { valid: true; placeholders: string[] } | { valid: false; error: string } { 235 + if (!uri.trim()) { 236 + return { valid: false, error: "Base record URI is required" }; 237 + } 238 + 239 + const placeholders: string[] = []; 240 + uri.replace(PLACEHOLDER_RE, (_, path: string) => { 241 + placeholders.push(path.trim()); 242 + return ""; 243 + }); 244 + 245 + if (placeholders.length === 0 && !uri.startsWith("at://")) { 246 + return { 247 + valid: false, 248 + error: "Base record URI must contain {{placeholders}} or be a literal AT URI", 249 + }; 250 + } 251 + 252 + const fetchSet = new Set(fetchNames ?? []); 253 + const actionSet = new Set(actionNames ?? []); 254 + for (const p of placeholders) { 255 + if (parseFunctionCall(p)) { 256 + return { valid: false, error: `Function calls not allowed in base record URI: {{${p}}}` }; 257 + } 258 + if (p === "now" || p === "self" || p.startsWith("event.")) continue; 259 + const root = p.split(".")[0]!; 260 + if (fetchSet.has(root)) continue; 261 + if (actionSet.has(root)) continue; 262 + return { valid: false, error: `Invalid placeholder in base record URI: {{${p}}}` }; 263 + } 264 + 265 + return { valid: true, placeholders }; 266 + } 267 + 210 268 /** Validate a plain-text template (for bsky-post actions). */ 211 269 export function validateTextTemplate( 212 270 template: string, 213 271 fetchNames?: string[], 272 + actionNames?: string[], 214 273 ): { valid: true; placeholders: string[] } | { valid: false; error: string } { 215 274 if (!template.trim()) { 216 275 return { valid: false, error: "Text template is required" }; ··· 223 282 }); 224 283 225 284 const fetchSet = new Set(fetchNames ?? []); 285 + const actionSet = new Set(actionNames ?? []); 226 286 for (const p of placeholders) { 227 287 const call = parseFunctionCall(p); 228 288 const toValidate = call ? call.arg : p; 229 289 if (toValidate === "now" || toValidate === "self" || toValidate.startsWith("event.")) continue; 230 290 const root = toValidate.split(".")[0]!; 231 291 if (fetchSet.has(root)) continue; 292 + if (actionSet.has(root)) continue; 232 293 return { valid: false, error: `Invalid placeholder: {{${p}}}` }; 233 294 } 234 295

+29 -1

lib/automations/pds.ts

··· 52 52 comment?: string; 53 53 }; 54 54 55 - export type PdsAction = PdsWebhookAction | PdsRecordAction | PdsBskyPostAction; 55 + type PdsPatchRecordAction = { 56 + $type: "run.airglow.automation#patchRecordAction"; 57 + targetCollection: string; 58 + baseRecordUri: string; 59 + recordTemplate: string; 60 + comment?: string; 61 + }; 62 + 63 + export type PdsAction = 64 + | PdsWebhookAction 65 + | PdsRecordAction 66 + | PdsBskyPostAction 67 + | PdsPatchRecordAction; 56 68 57 69 export type PdsFetchStep = { 58 70 $type: "run.airglow.automation#fetchStep"; ··· 176 188 }); 177 189 return { uri: data.uri as string, cid: data.cid as string }; 178 190 } 191 + 192 + /** Update a record on the user's PDS via putRecord (for patch-record actions). */ 193 + export async function patchArbitraryRecord( 194 + did: string, 195 + collection: string, 196 + rkey: string, 197 + record: Record<string, unknown>, 198 + ): Promise<{ uri: string; cid: string }> { 199 + const data = await pdsCall(did, "com.atproto.repo.putRecord", { 200 + repo: did, 201 + collection, 202 + rkey, 203 + record: { $type: collection, ...record }, 204 + }); 205 + return { uri: data.uri as string, cid: data.cid as string }; 206 + }

+7

lib/automations/sanitize.ts

··· 9 9 langs?: string[]; 10 10 labels?: string[]; 11 11 comment?: string; 12 + } 13 + | { 14 + $type: "patch-record"; 15 + targetCollection: string; 16 + baseRecordUri: string; 17 + recordTemplate: string; 18 + comment?: string; 12 19 }; 13 20 14 21 /** Strip instance-local secrets and truncate webhook URLs to domain-only. */

+15 -1

lib/db/schema.ts

··· 31 31 comment?: string; 32 32 }; 33 33 34 - export type Action = WebhookAction | RecordAction | BskyPostAction; 34 + export type PatchRecordAction = { 35 + $type: "patch-record"; 36 + targetCollection: string; 37 + baseRecordUri: string; // AT URI template for the record to update 38 + recordTemplate: string; // JSON template — fields merged on top of base 39 + comment?: string; 40 + }; 41 + 42 + export type Action = WebhookAction | RecordAction | BskyPostAction | PatchRecordAction; 43 + 44 + /** Action types that produce a record result (uri, cid, rkey) for chaining. */ 45 + const RECORD_PRODUCING_TYPES = new Set(["record", "bsky-post", "patch-record"]); 46 + export function isRecordProducingAction(type: string): boolean { 47 + return RECORD_PRODUCING_TYPES.has(type); 48 + } 35 49 36 50 export type FetchStep = { 37 51 name: string;

+100 -10

lib/jetstream/handler.test.ts

··· 8 8 executeAction: vi.fn(), 9 9 })); 10 10 11 + vi.mock("@/actions/bsky-post.js", () => ({ 12 + executeBskyPost: vi.fn(), 13 + })); 14 + 15 + vi.mock("@/actions/patch-record.js", () => ({ 16 + executePatchRecord: vi.fn(), 17 + })); 18 + 11 19 vi.mock("@/actions/fetcher.js", () => ({ 12 20 resolveFetches: vi.fn(), 13 21 })); ··· 15 23 import { handleMatchedEvent } from "./handler.js"; 16 24 import { dispatch } from "../webhooks/dispatcher.js"; 17 25 import { executeAction } from "../actions/executor.js"; 26 + import { executeBskyPost } from "../actions/bsky-post.js"; 27 + import { executePatchRecord } from "../actions/patch-record.js"; 18 28 import { resolveFetches } from "../actions/fetcher.js"; 19 - import { makeMatch, makeWebhookAction, makeRecordAction, makeFetchStep } from "../test/fixtures.js"; 29 + import { 30 + makeMatch, 31 + makeWebhookAction, 32 + makeRecordAction, 33 + makeBskyPostAction, 34 + makeFetchStep, 35 + } from "../test/fixtures.js"; 36 + import type { ActionResult } from "../actions/executor.js"; 37 + 38 + const ok: ActionResult = { statusCode: 200 }; 39 + const okWithUri: ActionResult = { 40 + statusCode: 200, 41 + uri: "at://did:plc:test/app.bsky.feed.post/abc123", 42 + cid: "bafytest", 43 + }; 20 44 21 45 const mockDispatch = vi.mocked(dispatch); 22 46 const mockExecuteAction = vi.mocked(executeAction); 47 + const mockExecuteBskyPost = vi.mocked(executeBskyPost); 48 + const mockExecutePatchRecord = vi.mocked(executePatchRecord); 23 49 const mockResolveFetches = vi.mocked(resolveFetches); 24 50 25 51 describe("handleMatchedEvent", () => { 26 52 beforeEach(() => { 27 - mockDispatch.mockReset().mockResolvedValue(undefined); 28 - mockExecuteAction.mockReset().mockResolvedValue(undefined); 53 + mockDispatch.mockReset().mockResolvedValue(ok); 54 + mockExecuteAction.mockReset().mockResolvedValue(okWithUri); 55 + mockExecuteBskyPost.mockReset().mockResolvedValue(okWithUri); 56 + mockExecutePatchRecord.mockReset().mockResolvedValue(okWithUri); 29 57 mockResolveFetches.mockReset(); 30 58 }); 31 59 ··· 48 76 49 77 await handleMatchedEvent(match); 50 78 51 - expect(mockExecuteAction).toHaveBeenCalledWith(match, 0, {}); 79 + expect(mockExecuteAction).toHaveBeenCalledOnce(); 80 + expect(mockExecuteAction).toHaveBeenCalledWith(match, 0, expect.any(Object)); 52 81 expect(mockDispatch).not.toHaveBeenCalled(); 53 82 }); 54 83 ··· 67 96 await handleMatchedEvent(match); 68 97 69 98 expect(mockDispatch).toHaveBeenCalledTimes(2); 70 - expect(mockDispatch).toHaveBeenCalledWith(match, 0, {}); 71 - expect(mockDispatch).toHaveBeenCalledWith(match, 2, {}); 72 99 expect(mockExecuteAction).toHaveBeenCalledOnce(); 73 - expect(mockExecuteAction).toHaveBeenCalledWith(match, 1, {}); 100 + // Verify correct action indices were called 101 + expect(mockDispatch.mock.calls[0]![1]).toBe(0); 102 + expect(mockExecuteAction.mock.calls[0]![1]).toBe(1); 103 + expect(mockDispatch.mock.calls[1]![1]).toBe(2); 74 104 }); 75 105 76 106 it("resolves fetches and passes context to all actions", async () => { ··· 129 159 expect(mockDispatch).toHaveBeenCalledWith(match, 0, partialContext); 130 160 }); 131 161 132 - it("one action failing does not block others", async () => { 133 - mockDispatch.mockRejectedValueOnce(new Error("webhook failed")); 162 + it("stops chain when an action fails", async () => { 163 + mockDispatch.mockResolvedValueOnce({ statusCode: 500, error: "webhook failed" }); 134 164 135 165 const match = makeMatch({ 136 166 automation: { ··· 141 171 142 172 await handleMatchedEvent(match); 143 173 144 - // Both actions were called despite the first one rejecting 174 + // First action failed, second should not run 175 + expect(mockDispatch).toHaveBeenCalledOnce(); 176 + expect(mockExecuteAction).not.toHaveBeenCalled(); 177 + }); 178 + 179 + it("stops chain when an action throws", async () => { 180 + mockDispatch.mockRejectedValueOnce(new Error("unexpected")); 181 + 182 + const match = makeMatch({ 183 + automation: { 184 + actions: [makeWebhookAction(), makeRecordAction()], 185 + fetches: [], 186 + }, 187 + }); 188 + 189 + await handleMatchedEvent(match); 190 + 191 + expect(mockDispatch).toHaveBeenCalledOnce(); 192 + expect(mockExecuteAction).not.toHaveBeenCalled(); 193 + }); 194 + 195 + it("accumulates action results into fetchContext", async () => { 196 + const match = makeMatch({ 197 + automation: { 198 + actions: [makeRecordAction(), makeBskyPostAction()], 199 + fetches: [], 200 + }, 201 + }); 202 + 203 + await handleMatchedEvent(match); 204 + 205 + // Second action should receive fetchContext with action0 result 206 + expect(mockExecuteBskyPost).toHaveBeenCalledWith( 207 + match, 208 + 1, 209 + expect.objectContaining({ 210 + action0: { 211 + uri: okWithUri.uri, 212 + cid: okWithUri.cid, 213 + rkey: "abc123", 214 + record: {}, 215 + }, 216 + }), 217 + ); 218 + }); 219 + 220 + it("webhooks do not produce action results", async () => { 221 + const match = makeMatch({ 222 + automation: { 223 + actions: [makeWebhookAction(), makeRecordAction()], 224 + fetches: [], 225 + }, 226 + }); 227 + 228 + await handleMatchedEvent(match); 229 + 145 230 expect(mockDispatch).toHaveBeenCalledOnce(); 146 231 expect(mockExecuteAction).toHaveBeenCalledOnce(); 232 + // fetchContext should NOT contain action0 (webhook has no uri/cid) 233 + // but it does contain action1 from the record action (mutated after call) 234 + // So we verify the record action result is present but webhook result is not 235 + const finalCtx = mockExecuteAction.mock.calls[0]![2]!; 236 + expect(finalCtx).not.toHaveProperty("action0"); 147 237 }); 148 238 });

+46 -12

lib/jetstream/handler.ts

··· 1 1 import { db } from "../db/index.js"; 2 - import { deliveryLogs, type Action } from "../db/schema.js"; 2 + import { deliveryLogs, type Action, isRecordProducingAction } from "../db/schema.js"; 3 3 import { dispatch, buildPayload } from "../webhooks/dispatcher.js"; 4 - import { executeAction } from "../actions/executor.js"; 4 + import { executeAction, type ActionResult } from "../actions/executor.js"; 5 5 import { executeBskyPost } from "../actions/bsky-post.js"; 6 + import { executePatchRecord } from "../actions/patch-record.js"; 6 7 import { resolveFetches } from "../actions/fetcher.js"; 7 8 import { renderTemplate, renderTextTemplate, type FetchContext } from "../actions/template.js"; 8 9 import type { MatchedEvent } from "./consumer.js"; 9 10 10 11 /** Handle a matched Jetstream event: resolve fetches, then dispatch all actions. */ 11 12 export async function handleMatchedEvent(match: MatchedEvent) { 12 - let fetchContext: Record<string, { uri: string; cid: string; record: Record<string, unknown> }> = 13 - {}; 13 + let fetchContext: FetchContext = {}; 14 14 if (match.automation.fetches.length > 0) { 15 15 const result = await resolveFetches( 16 16 match.automation.fetches, ··· 30 30 : []; 31 31 for (let i = 0; i < match.automation.actions.length; i++) { 32 32 const action = match.automation.actions[i]!; 33 - logDryRun(match, i, action, fetchContext, fetchErrors).catch((err) => { 34 - console.error(`Dry-run log error for action ${i}:`, err); 35 - }); 33 + await logDryRun(match, i, action, fetchContext, fetchErrors); 34 + // Inject synthetic action result so subsequent dry-run actions can reference {{actionN.*}} 35 + if (isRecordProducingAction(action.$type)) { 36 + fetchContext[`action${i}`] = { 37 + uri: `at://dry-run/${action.$type}/placeholder`, 38 + cid: "dry-run-cid", 39 + rkey: "placeholder", 40 + record: {}, 41 + }; 42 + } 36 43 } 37 44 return; 38 45 } ··· 44 51 ? executeBskyPost 45 52 : action.$type === "record" 46 53 ? executeAction 47 - : dispatch; 48 - handler(match, i, fetchContext).catch((err) => { 49 - console.error(`Action ${i} (${action.$type}) delivery error:`, err); 50 - }); 54 + : action.$type === "patch-record" 55 + ? executePatchRecord 56 + : dispatch; 57 + 58 + try { 59 + const result: ActionResult = await handler(match, i, fetchContext); 60 + 61 + // Accumulate result into fetchContext for downstream actions 62 + if (result.uri && result.cid) { 63 + const rkey = result.uri.split("/").pop() ?? ""; 64 + fetchContext[`action${i}`] = { uri: result.uri, cid: result.cid, rkey, record: {} }; 65 + } 66 + 67 + // Fail-fast: stop chain on error 68 + if (!isActionSuccess(result.statusCode)) { 69 + console.error( 70 + `Action ${i} (${action.$type}) failed (${result.statusCode}), stopping chain: ${result.error ?? ""}`, 71 + ); 72 + break; 73 + } 74 + } catch (err) { 75 + console.error(`Action ${i} (${action.$type}) threw, stopping chain:`, err); 76 + break; 77 + } 51 78 } 79 + } 80 + 81 + function isActionSuccess(code: number): boolean { 82 + return code >= 200 && code < 400; 52 83 } 53 84 54 85 async function logDryRun( ··· 88 119 fetchContext, 89 120 match.automation.did, 90 121 ); 91 - message = `Would create record in ${action.targetCollection}`; 122 + message = 123 + action.$type === "patch-record" 124 + ? `Would patch record in ${action.targetCollection} via ${action.baseRecordUri}` 125 + : `Would create record in ${action.targetCollection}`; 92 126 payload = JSON.stringify(rendered); 93 127 } catch (err) { 94 128 error = `Template error: ${err instanceof Error ? err.message : String(err)}`;

+1 -1

lib/pds/resolver.ts

··· 8 8 const AT_URI_RE = /^at:\/\/(did:[^/]+)\/([^/]+)\/([^/]+)$/; 9 9 10 10 /** Parse an AT URI into its components. */ 11 - function parseAtUri(uri: string): { did: string; collection: string; rkey: string } { 11 + export function parseAtUri(uri: string): { did: string; collection: string; rkey: string } { 12 12 const match = uri.match(AT_URI_RE); 13 13 if (!match) throw new Error(`Invalid AT URI: ${uri}`); 14 14 return { did: match[1]!, collection: match[2]!, rkey: match[3]! };

+11

lib/test/fixtures.ts

··· 4 4 WebhookAction, 5 5 RecordAction, 6 6 BskyPostAction, 7 + PatchRecordAction, 7 8 FetchStep, 8 9 } from "../db/schema.js"; 9 10 import type { MatchedEvent } from "../jetstream/consumer.js"; ··· 68 69 return { 69 70 $type: "bsky-post", 70 71 textTemplate: "Post by {{event.did}}", 72 + ...overrides, 73 + }; 74 + } 75 + 76 + export function makePatchRecordAction(overrides?: Partial<PatchRecordAction>): PatchRecordAction { 77 + return { 78 + $type: "patch-record", 79 + targetCollection: "site.standard.document", 80 + baseRecordUri: "at://{{event.did}}/{{event.commit.collection}}/{{event.commit.rkey}}", 81 + recordTemplate: '{"bskyPostRef":"{{action0.uri}}","updatedAt":"{{now}}"}', 71 82 ...overrides, 72 83 }; 73 84 }

+4 -1

lib/webhooks/dispatcher.ts

··· 2 2 import { deliveryLogs, type WebhookAction } from "../db/schema.js"; 3 3 import { sign } from "./signer.js"; 4 4 import { assertPublicUrl } from "../url-guard.js"; 5 + import type { ActionResult } from "../actions/executor.js"; 5 6 import type { MatchedEvent } from "../jetstream/consumer.js"; 6 7 import type { FetchContext } from "../actions/template.js"; 7 8 ··· 158 159 match: MatchedEvent, 159 160 actionIndex: number, 160 161 fetchContext?: FetchContext, 161 - ) { 162 + ): Promise<ActionResult> { 162 163 const { automation, event } = match; 163 164 const action = automation.actions[actionIndex] as WebhookAction; 164 165 const payload = buildPayload(match, fetchContext); ··· 188 189 0, 189 190 ); 190 191 } 192 + 193 + return { statusCode: result.statusCode, error: result.error }; 191 194 }

Configure Feed

Configure Feed