[{"_1":2,"_175":-5,"_176":-5},"loaderData",{"_3":4,"_92":93},"root",{"_5":6,"_35":36,"_65":66},"ctx",{"_7":8,"_23":-5,"_24":25,"_26":27,"_28":20,"_29":20,"_30":20,"_31":32,"_33":20,"_34":20},"auth",{"_9":-5,"_10":-5,"_11":-5,"_12":-5,"_13":-5,"_14":-5,"_15":-5,"_16":-5,"_17":-5,"_18":-5,"_19":20,"_21":22},"sessionClaims","sessionId","sessionStatus","userId","actor","orgId","orgRole","orgSlug","orgPermissions","factorVerificationAge","isAuthenticated",false,"tokenType","session_token","user","preferencesKeybindings","default","preferencesTheme","light","preferencesFixers","preferencesCopilot","preferencesSelectionAi","notificationCount",0,"flagDenols","isTeamOrgMember","ENV",{"_37":38,"_39":40,"_41":42,"_43":44,"_45":46,"_47":48,"_49":50,"_51":52,"_53":54,"_55":56,"_57":58,"_59":60,"_61":62,"_63":64},"REACT_APP_SENTRY_DSN","https://586020d86f05489db63a554690acf1e9@o1398359.ingest.sentry.io/6724499","VAL_RUNTIME_NAME","prod","RENDER_GIT_COMMIT","4e736cb42e83c17c10d9f33412034a74b06f3941","RENDER_GIT_BRANCH","main","CLERK_PUBLISHABLE_KEY","pk_live_Y2xlcmsudmFsLnRvd24k","POSTHOG_API_TOKEN","phc_nmoyuAkStEZuUp69rYlg1DZgrIt19Gx0Y4Hwf8tGgJP","POSTHOG_API_HOST","https://wart.val.town","WEB_API_TEMPLATE","https://{val}.web.val.run","APP_URL","https://www.val.town","MODULE_URL","https://esm.town","VANITY_WEB_API_TEMPLATE","https://{label}.val.run/","DENOLS_URL","https://lsp.val.town","NODE_ENV","production","API_URL","https://api.val.town","clerkState",{"_67":68},"__internal_clerk_state",{"_69":70,"_71":46,"_72":73,"_74":73,"_75":20,"_76":73,"_77":73,"_78":73,"_79":73,"_80":81,"_90":20,"_91":20},"__clerk_ssr_state",{"_21":22,"_9":-5,"_10":-5,"_11":-5,"_12":-5,"_13":-5,"_14":-5,"_15":-5,"_16":-5,"_17":-5,"_18":-5,"_19":20},"__publishableKey","__proxyUrl","","__domain","__isSatellite","__signInUrl","__signUpUrl","__afterSignInUrl","__afterSignUpUrl","__clerk_debug",{"_82":20,"_19":20,"_83":73,"_84":85,"_86":73,"_87":46,"_88":20,"_89":73},"isSignedIn","proxyUrl","reason","session-token-and-uat-missing","message","publishableKey","isSatellite","domain","__telemetryDisabled","__telemetryDebug","routes/embed.x.$handle.$name.$",{"_94":95,"_96":97,"_98":99,"_100":101,"_150":151,"_172":173,"_174":-5},"handle","mhalle","name","bluesky-data","path","backend/jetstreamAppview.cron.ts","project",{"_102":103,"_96":97,"_104":105,"_106":-5,"_107":20,"_108":109,"_110":73,"_111":-5,"_12":112,"_113":114,"_115":116,"_117":116,"_118":-5,"_119":32,"_14":120,"_121":122,"_123":122,"_124":125,"_126":125,"_127":125,"_128":122,"_129":122,"_23":130,"_133":134,"_136":137,"_149":20},"id","d2f81940-a381-11f0-b4b2-0224a6c84d84","createdAt",["D",1759843843402],"deletedAt","legacyEnvVars","privacy","public","description","imageUrl","0c3483fd-9552-4b4f-a670-7e10b8e6366a","pinnedFileId","b3f8f532-a3f4-11f0-b878-0224a6c84d84","pinPreview",true,"pinMigrated","screenshotVersion","likeCount","fd108e66-3b67-470c-ac8a-5708885d9422","userPerm","r","valPerm","blobPerm","rw","emailPerm","sqlitePerm","projectPerm","telemetryPerm",{"_94":95,"_131":132},"profileImageUrl","https://img.clerk.com/eyJ0eXBlIjoicHJveHkiLCJzcmMiOiJodHRwczovL2ltYWdlcy5jbGVyay5kZXYvb2F1dGhfZ2l0aHViL2ltZ18yaVJXQjhPYjlqMlhiekhuZ2xpTVBlQVp6WXgifQ","org",{"_94":95,"_135":132,"_12":112},"avatar","branch",{"_138":139,"_96":44,"_102":140,"_141":103,"_142":-5,"_143":-5,"_144":-5,"_145":-5,"_146":-5,"_147":97,"_148":95},"version",40,"d2f8893e-a381-11f0-b4b2-0224a6c84d84","projectId","forkedBranchId","forkedBranchVersion","forkedProjectPrivacy","forkedProjectId","forkedProjectOrgId","projectName","userHandle","isOwned","file",{"_102":152,"_96":153,"_154":155,"_156":157,"_158":140,"_159":160,"_161":116,"_104":162,"_138":163,"_164":165,"_166":167,"_168":20,"_169":-5,"_98":170},"7813bf98-a971-11f0-8e3a-0224a6c84d84","jetstreamAppview.cron.ts","fileId","b3d031f6-a3f4-11f0-8e27-0224a6c84d84","parentId","b39211d2-a3f4-11f0-99aa-0224a6c84d84","branchId","branchVersion",39,"latest",["D",1760496525929],7,"fileType","script","content","import { BskyAgent } from \"https://esm.sh/@atproto/api@0.13.18?target=deno\";\nimport {\n TESTDATA_COLLECTION,\n TestDataRecord,\n} from \"../shared/atprotoTestData.ts\";\nimport {\n ensureTestDataTable,\n getLatestCursor,\n insertTestDataRow,\n TESTDATA_TABLE,\n} from \"./database/queries.ts\";\n\n// The global Jetstream proxy only supports `commit` for now; additional wanted types\n// or other filters would require a custom Jetstream deployment.\nconst DEFAULT_WANTED = \"commit\";\n\nconst JETSTREAM_URL = Deno.env.get(\"JETSTREAM_URL\") ??\n \"https://jetstream.atproto.tools/subscribe\";\nconst APPVIEW_SERVICE_URL = Deno.env.get(\"BLUESKY_SERVICE_URL\") ??\n \"https://bsky.social\";\nconst IDENTIFIER = Deno.env.get(\"BLUESKY_HANDLE\");\nconst APP_PASSWORD = Deno.env.get(\"BLUESKY_APP_PASSWORD\");\nconst STREAM_TIMEOUT_MS = Number(Deno.env.get(\"JETSTREAM_TIMEOUT_MS\") ?? 15000);\nconst MAX_STREAM_EVENTS = Number(Deno.env.get(\"JETSTREAM_MAX_EVENTS\") ?? 64);\nconst REPLAY_MARGIN = Number(Deno.env.get(\"JETSTREAM_REPLAY_MARGIN\") ?? 5);\n\nfunction ensureCredentials(): { identifier: string; appPassword: string } {\n if (!IDENTIFIER) {\n throw new Error(\"BLUESKY_HANDLE environment variable is required\");\n }\n if (!APP_PASSWORD) {\n throw new Error(\"BLUESKY_APP_PASSWORD environment variable is required\");\n }\n return { identifier: IDENTIFIER, appPassword: APP_PASSWORD };\n}\n\ninterface JetstreamMessage {\n kind: string;\n seq: number;\n time: string;\n repo?: string;\n commit?: {\n rev: string;\n time: string;\n };\n ops?: Array<{\n action: \"create\" | \"update\" | \"delete\";\n path: string;\n cid?: string | { $link: string };\n }>;\n operations?: Array<{\n action: \"create\" | \"update\" | \"delete\";\n path: string;\n cid?: string | { $link: string };\n }>;\n}\n\ninterface StreamState {\n processedOps: number;\n storedEvents: number;\n highestCursor?: number;\n}\n\nexport default async function jetstreamCron(): Promise {\n await ensureTestDataTable();\n const latestCursor = await getLatestCursor();\n const startCursor = typeof latestCursor === \"number\"\n ? Math.max(latestCursor - REPLAY_MARGIN, 0)\n : undefined;\n\n const agent = new BskyAgent({ service: APPVIEW_SERVICE_URL });\n const { identifier, appPassword } = ensureCredentials();\n await agent.login({ identifier, password: appPassword });\n const state: StreamState = {\n processedOps: 0,\n storedEvents: 0,\n highestCursor: startCursor,\n };\n\n await processJetstream({ agent, startCursor, state });\n\n console.log(\n JSON.stringify(\n {\n event: \"jetstream.sync\",\n table: TESTDATA_TABLE,\n startCursor,\n lastCursor: state.highestCursor ?? startCursor ?? null,\n processedOps: state.processedOps,\n storedEvents: state.storedEvents,\n },\n null,\n 2,\n ),\n );\n}\n\nasync function processJetstream({\n agent,\n startCursor,\n state,\n}: {\n agent: BskyAgent;\n startCursor?: number;\n state: StreamState;\n}): Promise {\n const url = new URL(JETSTREAM_URL);\n if (url.hostname === \"jetstream.atproto.tools\") {\n // The public Jetstream proxy requires `wanted` to be provided explicitly.\n url.searchParams.set(\"wanted\", DEFAULT_WANTED);\n } else if (!url.searchParams.has(\"wanted\")) {\n url.searchParams.set(\"wanted\", DEFAULT_WANTED);\n }\n if (\n typeof startCursor === \"number\" && startCursor > 0 &&\n !url.searchParams.has(\"cursor\")\n ) {\n url.searchParams.set(\"cursor\", startCursor.toString());\n }\n\n const wsUrl = toWebSocketUrl(url);\n const ws = new WebSocket(wsUrl);\n ws.binaryType = \"arraybuffer\";\n\n const tasks: Promise[] = [];\n\n await new Promise((resolve, reject) => {\n const timeout = setTimeout(() => {\n ws.close(1000, \"stream timeout\");\n }, STREAM_TIMEOUT_MS);\n\n ws.addEventListener(\"open\", () => {\n // Connection established; nothing else to send.\n });\n\n ws.addEventListener(\"message\", (event) => {\n if (state.processedOps >= MAX_STREAM_EVENTS) {\n ws.close(1000, \"reached max events\");\n return;\n }\n const task = normalizeEventData(event.data)\n .then((payload) => handleJetstreamMessage(payload, agent, state))\n .catch((error) => {\n console.error(\"Jetstream message handling error\", error);\n });\n tasks.push(task);\n task.finally(() => {\n if (\n state.processedOps >= MAX_STREAM_EVENTS &&\n ws.readyState === WebSocket.OPEN\n ) {\n ws.close(1000, \"reached max events\");\n }\n });\n });\n\n ws.addEventListener(\"error\", (event) => {\n clearTimeout(timeout);\n reject(new Error(\"Jetstream WebSocket connection error\"));\n });\n\n ws.addEventListener(\"close\", () => {\n clearTimeout(timeout);\n resolve();\n });\n });\n\n if (tasks.length > 0) {\n await Promise.all(tasks);\n }\n}\nasync function handleJetstreamMessage(\n raw: string,\n agent: BskyAgent,\n state: StreamState,\n): Promise {\n if (!raw) {\n return;\n }\n let message: JetstreamMessage;\n try {\n message = JSON.parse(raw) as JetstreamMessage;\n } catch (error) {\n console.error(\"Failed to parse Jetstream payload\", error, raw);\n return;\n }\n const operations = message.ops ?? message.operations;\n if (message.kind !== \"commit\" || !message.repo || !operations) {\n return;\n }\n\n const relevantOps = operations.filter((op) =>\n op.path.startsWith(`${TESTDATA_COLLECTION}/`)\n );\n if (!relevantOps.length) {\n return;\n }\n\n const now = new Date().toISOString();\n for (const op of relevantOps) {\n const [, rkey] = op.path.split(\"/\");\n if (!rkey) {\n continue;\n }\n state.processedOps += 1;\n state.highestCursor = Math.max(state.highestCursor ?? 0, message.seq);\n if (op.action === \"delete\") {\n continue;\n }\n try {\n const recordResponse = await agent.com.atproto.repo.getRecord({\n repo: message.repo,\n collection: TESTDATA_COLLECTION,\n rkey,\n });\n const record = recordResponse.data.value as TestDataRecord;\n if (record.$type !== TESTDATA_COLLECTION) {\n continue;\n }\n await insertTestDataRow({\n cursor: message.seq,\n repo: message.repo,\n collection: TESTDATA_COLLECTION,\n rkey,\n record,\n eventTime: message.time,\n receivedAt: now,\n });\n state.storedEvents += 1;\n } catch (error) {\n console.error(\"Failed to fetch record for op\", { op, error });\n }\n }\n}\n\nfunction toWebSocketUrl(url: URL): string {\n const next = new URL(url.toString());\n if (next.protocol === \"https:\") {\n next.protocol = \"wss:\";\n } else if (next.protocol === \"http:\") {\n next.protocol = \"ws:\";\n }\n return next.toString();\n}\n\nconst binaryDecoder = new TextDecoder();\n\nfunction normalizeEventData(data: unknown): Promise {\n if (typeof data === \"string\") {\n return Promise.resolve(data);\n }\n if (data instanceof ArrayBuffer) {\n return Promise.resolve(binaryDecoder.decode(new Uint8Array(data)));\n }\n if (data instanceof Uint8Array) {\n return Promise.resolve(binaryDecoder.decode(data));\n }\n if (typeof Blob !== \"undefined\" && data instanceof Blob) {\n return data.text();\n }\n if (\n data &&\n typeof (data as { text?: () => Promise }).text === \"function\"\n ) {\n return (data as { text: () => Promise }).text();\n }\n return Promise.resolve(\"\");\n}\n","deleted","label",[171,153],"backend","permalink","https://www.val.town/x/mhalle/bluesky-data/code/backend/jetstreamAppview.cron.ts","extension","actionData","errors"]