Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
3.2 kB · 86 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687import { z } from "zod";import type { JsonObject } from "../core/json.js";import type { XWebhookSourceManifest } from "../runtime/manifest.js";import type { JazzThoughtStore } from "../jazz/store.js";
export const sourceControlProjectionVersion = 1;
const runtimeStateSchema = z.object({ state: z.literal("ready"), readyAt: z.iso.datetime(), revision: z.string().min(1).max(200),}).strict();
const upstreamStateSchema = z.object({ registered: z.boolean(), valid: z.boolean(), webhookId: z.string().regex(/^[0-9]{1,19}$/).optional(), desiredSubscriptionCount: z.number().int().min(0).max(1_500), liveSubscriptionCount: z.number().int().min(0).max(1_500), subscriptionsConverged: z.boolean(), checkedAt: z.iso.datetime(),}).strict();
export const sourceControlPayloadSchema = z.object({ version: z.literal(sourceControlProjectionVersion), source: z.string().min(1).max(200), kind: z.literal("x-webhook"), enabled: z.boolean(), lane: z.string().min(1).max(100), expectedSubscriptionCount: z.number().int().min(1).max(1_500), eventTypes: z.array(z.string().min(1).max(100)).min(1).max(100), configuredAt: z.iso.datetime(), updatedAt: z.iso.datetime(), runtime: runtimeStateSchema.optional(), upstream: upstreamStateSchema.optional(),}).strict();
export type SourceControlPayload = z.infer<typeof sourceControlPayloadSchema>;export type SourceControlRuntimeState = z.infer<typeof runtimeStateSchema>;export type SourceControlUpstreamState = z.infer<typeof upstreamStateSchema>;
export async function updateXSourceControl( store: JazzThoughtStore, source: XWebhookSourceManifest, update: { runtime?: SourceControlRuntimeState; upstream?: SourceControlUpstreamState; updatedAt?: string; } = {},): Promise<SourceControlPayload> { const updatedAt = update.updatedAt ?? new Date().toISOString(); const currentSource = (await store.listSources()).find((candidate) => candidate.id === source.id); const parsedCurrent = currentSource?.config.control ? sourceControlPayloadSchema.safeParse(currentSource.config.control) : undefined; const current = parsedCurrent?.success ? parsedCurrent.data : undefined; const eventTypes = [...new Set(source.expectedSubscriptions.map((item) => item.eventType))].sort(); const runtime = update.runtime ?? current?.runtime; const upstream = update.upstream ?? current?.upstream; const payload = sourceControlPayloadSchema.parse({ version: sourceControlProjectionVersion, source: source.id, kind: "x-webhook", enabled: source.enabled, lane: source.lane, expectedSubscriptionCount: source.expectedSubscriptions.length, eventTypes, configuredAt: current?.configuredAt ?? updatedAt, updatedAt, ...(runtime ? { runtime } : {}), ...(upstream ? { upstream } : {}), }); await store.registerSource({ id: source.id, kind: "x-webhook", enabled: source.enabled, config: { lane: source.lane, expectedSubscriptionCount: source.expectedSubscriptions.length, eventTypes, control: payload as unknown as JsonObject, }, updatedAt, }); return payload;}