From 03e80cbb17d8624cebd5efd630ec7d724abec9b5 Mon Sep 17 00:00:00 2001 From: Thibault Le Ouay Date: Thu, 22 Oct 2026 12:24:43 +0200 Subject: [PATCH] workflows: publish checker tasks to Pub/Sub behind CHECKER_TRANSPORT (#2964) CHECKER_TRANSPORT=pubsub publishes each check to checker- (or checker--30s), both 30s halves at once, with jobType, periodicity, monitorId and region attributes. The default stays cloudtasks. The payload builder moves to checker-payload.ts so both transports send the same body. checker-regions.json mirrors the terraform topic list, and a test fails when packages/regions gains an active region it lacks. @google-cloud/pubsub joins the side-car externals. A scoped override moves its semantic-conventions pin to the catalog's 1.43.0 so the side-car ships one copy. Co-authored-by: Claude Opus 5.5 --- apps/workflows/AGENTS.md | 2 +- apps/workflows/package.json | 1 + apps/workflows/scripts/sidecar-package.mjs | 8 +- .../src/cron/checker-payload.test.ts | 270 ++++++++++++++++++ apps/workflows/src/cron/checker-payload.ts | 199 +++++++++++++ apps/workflows/src/cron/checker-regions.json | 29 ++ apps/workflows/src/cron/checker.ts | 269 ++++++----------- apps/workflows/src/env.ts | 3 + apps/workflows/tsdown.config.ts | 12 +- pnpm-lock.yaml | 100 +++++++ pnpm-workspace.yaml | 6 + 11 files changed, 718 insertions(+), 181 deletions(-) create mode 100644 apps/workflows/src/cron/checker-payload.test.ts create mode 100644 apps/workflows/src/cron/checker-payload.ts create mode 100644 apps/workflows/src/cron/checker-regions.json diff --git a/apps/workflows/AGENTS.md b/apps/workflows/AGENTS.md index 5789fdc4d..cf168496d 100644 --- a/apps/workflows/AGENTS.md +++ b/apps/workflows/AGENTS.md @@ -6,7 +6,7 @@ this app has (Vitest strips types without checking them). Constraints that follow from the build: -- `@libsql/client`, `libsql` and `@google-cloud/tasks` stay external in +- `@libsql/client`, `libsql`, `@google-cloud/tasks` and `@google-cloud/pubsub` stay external in `tsdown.config.ts` (native addon, CJS `__dirname` reads) and ship as a side-car `node_modules` in the image. A new dependency with either trait joins that list and the `externals` in `apps/workflows/scripts/sidecar-package.mjs`, which diff --git a/apps/workflows/package.json b/apps/workflows/package.json index 4aebcccf3..e3af95b15 100644 --- a/apps/workflows/package.json +++ b/apps/workflows/package.json @@ -8,6 +8,7 @@ "build": "tsdown" }, "dependencies": { + "@google-cloud/pubsub": "catalog:", "@google-cloud/tasks": "catalog:", "@hono/node-server": "catalog:", "@libsql/client": "catalog:", diff --git a/apps/workflows/scripts/sidecar-package.mjs b/apps/workflows/scripts/sidecar-package.mjs index 343735501..bc8346f33 100644 --- a/apps/workflows/scripts/sidecar-package.mjs +++ b/apps/workflows/scripts/sidecar-package.mjs @@ -13,7 +13,12 @@ const workflows = new URL("../package.json", import.meta.url); // Resolve from real paths: pnpm only links a package's deps inside its store dir. const resolve = (name, from) => realpathSync(findPackageJSON(name, from)); -const externals = ["@libsql/client", "libsql", "@google-cloud/tasks"]; +const externals = [ + "@libsql/client", + "libsql", + "@google-cloud/tasks", + "@google-cloud/pubsub", +]; const versions = new Map(); const seen = new Set(); @@ -47,6 +52,7 @@ const libsqlClient = resolve("@libsql/client", workflows); walk(libsqlClient); walk(resolve("libsql", libsqlClient)); walk(resolve("@google-cloud/tasks", workflows)); +walk(resolve("@google-cloud/pubsub", workflows)); const dependencies = {}; const overrides = {}; diff --git a/apps/workflows/src/cron/checker-payload.test.ts b/apps/workflows/src/cron/checker-payload.test.ts new file mode 100644 index 000000000..15c897114 --- /dev/null +++ b/apps/workflows/src/cron/checker-payload.test.ts @@ -0,0 +1,270 @@ +import { readFileSync } from "node:fs"; + +import { selectMonitorSchema } from "@openstatus/db/src/schema"; +import { regionDict } from "@openstatus/regions"; +import { describe, expect, test } from "vitest"; +import { z } from "zod"; + +import { + type CheckerPublisher, + type TaskInput, + buildCheckerPayload, + checkerTopic, + publishCheckerTasks, +} from "./checker-payload"; + +const monitorRow = ( + overrides: Record, +) => + selectMonitorSchema.parse({ + id: 42, + jobType: "http", + periodicity: "1m", + status: "active", + active: true, + regions: "ams", + url: "https://example.com", + name: "example", + externalName: null, + description: "", + headers: '[{"key":"x-api-key","value":"secret"}]', + body: "", + method: "GET", + workspaceId: 7, + timeout: 45000, + degradedAfter: null, + assertions: null, + otelEndpoint: null, + otelHeaders: null, + public: false, + retry: 3, + followRedirects: true, + grpcService: null, + grpcTls: "tls", + createdAt: new Date(0), + updatedAt: new Date(0), + deletedAt: null, + ...overrides, + }); + +const input = ( + overrides: Partial & Pick, +): TaskInput => ({ + timestamp: 1_700_000_000_000, + status: "active", + region: "ams", + ...overrides, +}); + +type Published = { + topic: string; + payload: ReturnType; + attributes: Record; +}; + +const recordingPublisher = (failTopics: string[] = []) => { + const published: Published[] = []; + const publisher: CheckerPublisher = { + publish: async (topic, data, attributes) => { + if (failTopics.includes(topic)) { + throw new Error(`topic ${topic} not found`); + } + published.push({ + topic, + payload: JSON.parse(data.toString()), + attributes, + }); + return String(published.length); + }, + }; + return { published, publisher }; +}; + +describe("buildCheckerPayload", () => { + test("http keeps the request shape the checker binds", () => { + const payload = buildCheckerPayload( + input({ + row: monitorRow({ + assertions: + '[{"version":"v1","type":"status","compare":"eq","target":200}]', + }), + status: "error", + }), + ); + + expect(payload).toMatchObject({ + workspaceId: "7", + monitorId: "42", + url: "https://example.com", + method: "GET", + cronTimestamp: 1_700_000_000_000, + headers: [{ key: "x-api-key", value: "secret" }], + status: "error", + assertions: [ + { version: "v1", type: "status", compare: "eq", target: 200 }, + ], + trigger: "cron", + retry: 3, + followRedirects: true, + }); + }); + + test.each(["tcp", "dns", "icmp"] as const)( + "%s sends the url as uri", + (jobType) => { + const payload = buildCheckerPayload( + input({ row: monitorRow({ jobType, url: "example.com:443" }) }), + ); + + expect(payload).toMatchObject({ + monitorId: "42", + uri: "example.com:443", + trigger: "cron", + }); + expect(payload).not.toHaveProperty("url"); + }, + ); + + test("grpc sends headers as a map and defaults tls", () => { + const payload = buildCheckerPayload( + input({ + row: monitorRow({ + jobType: "grpc", + url: "grpc.example.com:443", + grpcService: "health", + grpcTls: null, + }), + }), + ); + + expect(payload).toMatchObject({ + uri: "grpc.example.com:443", + service: "health", + tls: "tls", + headers: { "x-api-key": "secret" }, + }); + }); +}); + +describe("checkerTopic", () => { + test("30s monitors get their own topic", () => { + expect(checkerTopic("ams", "30s")).toBe("checker-ams-30s"); + expect(checkerTopic("koyeb_fra", "1m")).toBe("checker-koyeb_fra"); + }); +}); + +describe("publishCheckerTasks", () => { + test("publishes to the region topic with routing attributes", async () => { + const { published, publisher } = recordingPublisher(); + const row = monitorRow({ jobType: "tcp" }); + + const result = await publishCheckerTasks( + [input({ row, region: "iad" })], + "5m", + publisher, + ); + + expect(result).toMatchObject({ success: 1, failed: 0 }); + expect(published).toEqual([ + { + topic: "checker-iad", + payload: buildCheckerPayload(input({ row, region: "iad" })), + attributes: { + jobType: "tcp", + periodicity: "5m", + monitorId: "42", + region: "iad", + }, + }, + ]); + }); + + test("publishes both 30s halves at once, 30s apart", async () => { + const { published, publisher } = recordingPublisher(); + const row = monitorRow({ periodicity: "30s" }); + const timestamp = 1_700_000_000_000; + + await publishCheckerTasks( + [ + input({ row, timestamp }), + input({ row, timestamp: timestamp + 30_000 }), + ], + "30s", + publisher, + ); + + expect(published.map((p) => p.topic)).toEqual([ + "checker-ams-30s", + "checker-ams-30s", + ]); + expect(published.map((p) => p.payload.cronTimestamp)).toEqual([ + timestamp, + timestamp + 30_000, + ]); + }); + + test("counts failures per region without dropping the rest", async () => { + const { published, publisher } = recordingPublisher(["checker-gru"]); + const row = monitorRow({}); + + const result = await publishCheckerTasks( + [ + input({ row, region: "gru" }), + input({ row, region: "ams" }), + input({ row, region: "gru" }), + ], + "1m", + publisher, + ); + + expect(result.success).toBe(1); + expect(result.failed).toBe(2); + expect(result.failedByRegion).toEqual({ gru: 2 }); + expect(result.errors[0]).toContain("checker-gru"); + expect(published).toHaveLength(1); + }); + + test("does not retry a rejected publish, which would duplicate the check", async () => { + let calls = 0; + const failing: CheckerPublisher = { + publish: () => { + calls++; + return Promise.reject(new Error("unavailable")); + }, + }; + + const result = await publishCheckerTasks( + [input({ row: monitorRow({}) })], + "1m", + failing, + ); + + expect(result).toMatchObject({ success: 0, failed: 1 }); + expect(calls).toBe(1); + }); +}); + +describe("checker-regions.json", () => { + test("lists every active checker region, so infra has a topic for each", () => { + const snapshot = z + .array(z.string()) + .parse( + JSON.parse( + readFileSync( + new URL("./checker-regions.json", import.meta.url), + "utf8", + ), + ), + ); + const active = Object.values(regionDict) + .filter((r) => !r.deprecated && r.provider !== "private") + .map((r) => r.code); + + const missing = active.filter((code) => !snapshot.includes(code)); + expect( + missing, + "add these regions to checker-regions.json and to checker_regions in openstatus-org/infra/gcp/pubsub.tf", + ).toEqual([]); + expect(snapshot.filter((code) => !active.includes(code))).toEqual([]); + }); +}); diff --git a/apps/workflows/src/cron/checker-payload.ts b/apps/workflows/src/cron/checker-payload.ts new file mode 100644 index 000000000..700f8968a --- /dev/null +++ b/apps/workflows/src/cron/checker-payload.ts @@ -0,0 +1,199 @@ +import type { MonitorStatus } from "@openstatus/db/src/schema"; +import type { selectMonitorSchema } from "@openstatus/db/src/schema"; +import type { + Region, + monitorPeriodicitySchema, +} from "@openstatus/db/src/schema/constants"; +import { + type DNSPayloadSchema, + type grpcPayloadSchema, + type httpPayloadSchema, + type icmpPayloadSchema, + type tpcPayloadSchema, + transformHeaders, +} from "@openstatus/utils"; +import { Effect, Result } from "effect"; +import type { z } from "zod"; + +export type Periodicity = z.infer; + +export type TaskInput = { + row: z.infer; + timestamp: number; + status: MonitorStatus; + region: Region; +}; + +export type CheckerPayload = + | z.infer + | z.infer + | z.infer + | z.infer + | z.infer; + +// timestamp needs to be in ms +export function buildCheckerPayload({ + row, + timestamp, + status, +}: Omit): CheckerPayload { + const otelConfig = row.otelEndpoint + ? { + endpoint: row.otelEndpoint, + headers: transformHeaders(row.otelHeaders), + } + : undefined; + + switch (row.jobType) { + case "http": + return { + workspaceId: String(row.workspaceId), + monitorId: String(row.id), + url: row.url, + method: row.method || "GET", + cronTimestamp: timestamp, + body: row.body, + headers: row.headers, + status: status, + assertions: row.assertions ? JSON.parse(row.assertions) : null, + degradedAfter: row.degradedAfter, + timeout: row.timeout, + trigger: "cron", + otelConfig, + retry: row.retry || 3, + followRedirects: + row.followRedirects === null ? true : row.followRedirects, + }; + case "tcp": + return { + workspaceId: String(row.workspaceId), + monitorId: String(row.id), + uri: row.url, + status: status, + assertions: row.assertions ? JSON.parse(row.assertions) : null, + cronTimestamp: timestamp, + degradedAfter: row.degradedAfter, + timeout: row.timeout, + trigger: "cron", + retry: row.retry || 3, + otelConfig, + }; + case "dns": + return { + workspaceId: String(row.workspaceId), + monitorId: String(row.id), + uri: row.url, + cronTimestamp: timestamp, + status: status, + assertions: row.assertions ? JSON.parse(row.assertions) : null, + degradedAfter: row.degradedAfter, + timeout: row.timeout, + trigger: "cron", + otelConfig, + retry: row.retry || 3, + }; + case "icmp": + return { + workspaceId: String(row.workspaceId), + monitorId: String(row.id), + uri: row.url, + cronTimestamp: timestamp, + status: status, + degradedAfter: row.degradedAfter, + timeout: row.timeout, + trigger: "cron", + otelConfig, + retry: row.retry || 3, + }; + case "grpc": + return { + workspaceId: String(row.workspaceId), + monitorId: String(row.id), + uri: row.url, + service: row.grpcService ?? undefined, + tls: row.grpcTls ?? "tls", + headers: transformHeaders(row.headers), + cronTimestamp: timestamp, + status: status, + degradedAfter: row.degradedAfter, + timeout: row.timeout, + trigger: "cron", + otelConfig, + retry: row.retry || 3, + }; + default: + throw new Error("Invalid jobType"); + } +} + +export function checkerTopic(region: Region, periodicity: Periodicity) { + return periodicity === "30s" ? `checker-${region}-30s` : `checker-${region}`; +} + +export type CheckerPublisher = { + publish: ( + topic: string, + data: Buffer, + attributes: Record, + ) => Promise; +}; + +export type PublishResult = { + success: number; + failed: number; + failedByRegion: Partial>; + errors: string[]; +}; + +export async function publishCheckerTasks( + inputs: TaskInput[], + periodicity: Periodicity, + publisher: CheckerPublisher, +): Promise { + const results = await Effect.runPromise( + Effect.forEach( + inputs, + (input) => + Effect.tryPromise({ + try: () => + publisher.publish( + checkerTopic(input.region, periodicity), + Buffer.from(JSON.stringify(buildCheckerPayload(input))), + { + jobType: input.row.jobType, + periodicity, + monitorId: String(input.row.id), + region: input.region, + }, + ), + catch: (err) => (err instanceof Error ? err : new Error(String(err))), + // No retry here, unlike the Cloud Tasks path: there a retried task + // dedupes on its name (ALREADY_EXISTS), but every publishMessage is + // a new message. The client already retries the batch RPC itself, so + // a rejection here is final and counts as a missed check. + }).pipe(Effect.result), + // Unbounded so every publish lands in the client's batch window at once; + // the Pub/Sub client does the batching, not us. + { concurrency: "unbounded" }, + ), + ); + + const result: PublishResult = { + success: 0, + failed: 0, + failedByRegion: {}, + errors: [], + }; + results.forEach((outcome, i) => { + if (Result.isSuccess(outcome)) { + result.success++; + return; + } + const region = inputs[i].region; + result.failed++; + result.failedByRegion[region] = (result.failedByRegion[region] ?? 0) + 1; + result.errors.push(outcome.failure.message); + }); + + return result; +} diff --git a/apps/workflows/src/cron/checker-regions.json b/apps/workflows/src/cron/checker-regions.json new file mode 100644 index 000000000..9f7ad8d48 --- /dev/null +++ b/apps/workflows/src/cron/checker-regions.json @@ -0,0 +1,29 @@ +[ + "ams", + "arn", + "cdg", + "dfw", + "ewr", + "fra", + "gru", + "iad", + "jnb", + "lax", + "lhr", + "nrt", + "ord", + "sjc", + "sin", + "syd", + "yyz", + "koyeb_fra", + "koyeb_par", + "koyeb_sfo", + "koyeb_sin", + "koyeb_tyo", + "koyeb_was", + "railway_us-west2", + "railway_us-east4-eqdc4a", + "railway_europe-west4-drams3a", + "railway_asia-southeast1-eqsg3a" +] diff --git a/apps/workflows/src/cron/checker.ts b/apps/workflows/src/cron/checker.ts index 837124e77..9ed195be1 100644 --- a/apps/workflows/src/cron/checker.ts +++ b/apps/workflows/src/cron/checker.ts @@ -1,3 +1,4 @@ +import { PubSub, type Topic } from "@google-cloud/pubsub"; import { CloudTasksClient } from "@google-cloud/tasks"; import type { google } from "@google-cloud/tasks/build/protos"; import { getLogger } from "@logtape/logtape"; @@ -11,7 +12,6 @@ import { notInArray, } from "@openstatus/db"; import { - type MonitorStatus, maintenance, monitor, monitorStatusTable, @@ -19,32 +19,23 @@ import { selectMonitorStatusSchema, } from "@openstatus/db/src/schema"; import type { Region } from "@openstatus/db/src/schema/constants"; -import type { monitorPeriodicitySchema } from "@openstatus/db/src/schema/constants"; import { maintenancesToPageComponents, pageComponent, } from "@openstatus/db/src/schema/page_components"; import { regionDict } from "@openstatus/regions"; -import { - type DNSPayloadSchema, - type httpPayloadSchema, - type grpcPayloadSchema, - type icmpPayloadSchema, - type tpcPayloadSchema, - transformHeaders, -} from "@openstatus/utils"; import { Effect, Result, Schedule } from "effect"; import { z } from "zod"; import { env } from "../env"; import { db } from "../lib/db"; - -type TaskInput = { - row: z.infer; - timestamp: number; - status: MonitorStatus; - region: Region; -}; +import { + type CheckerPublisher, + type Periodicity, + type TaskInput, + buildCheckerPayload, + publishCheckerTasks, +} from "./checker-payload"; export const isAuthorizedDomain = (url: string) => { return url.includes(env().SITE_URL); @@ -77,35 +68,47 @@ function isGcpConfigured(): boolean { // Only initialize CloudTasksClient if GCP credentials are properly configured // This allows self-hosted deployments to run without GCP Cloud Tasks let client: CloudTasksClient | null = null; +let publisher: CheckerPublisher | null = null; if (isGcpConfigured()) { - client = new CloudTasksClient({ - // fallback: true, + const options = { projectId: env().GCP_PROJECT_ID, credentials: { client_email: env().GCP_CLIENT_EMAIL, private_key: env().GCP_PRIVATE_KEY.replaceAll("\\n", "\n"), }, - }); + }; + client = new CloudTasksClient(options); + publisher = createPublisher(new PubSub(options)); +} + +function createPublisher(pubsub: PubSub): CheckerPublisher { + const topics = new Map(); + return { + publish: (name, data, attributes) => { + let topic = topics.get(name); + if (!topic) { + topic = pubsub.topic(name, { + batching: { maxMessages: 1000, maxMilliseconds: 10 }, + }); + topics.set(name, topic); + } + return topic.publishMessage({ data, attributes }); + }, + }; } export async function sendCheckerTasks( - periodicity: z.infer, + periodicity: Periodicity, ): Promise<{ success: number; failed: number }> { - // If GCP Cloud Tasks is not configured, skip task creation - if (!client) { + // If GCP is not configured, skip task creation + if (!client || !publisher) { logger.warn("GCP Cloud Tasks not configured - skipping checker tasks", { periodicity, }); return { success: 0, failed: 0 }; } - const parent = client.queuePath( - env().GCP_PROJECT_ID, - env().GCP_LOCATION, - periodicity, - ); - const timestamp = Date.now(); const currentMaintenance = db @@ -216,12 +219,65 @@ export async function sendCheckerTasks( } } + const transport = env().CHECKER_TRANSPORT; + const { success, failed } = + transport === "pubsub" + ? await publishToPubSub(taskInputs, periodicity, publisher) + : await createCloudTasks(taskInputs, periodicity, client); + + logger.info("Completed cron job", { + periodicity, + transport, + total_tasks: taskInputs.length, + success_count: success, + failed_count: failed, + duration_ms: Date.now() - timestamp, + }); + if (failed > 0) { + logger.error("Cron job had failures", { + periodicity, + transport, + failed_count: failed, + success_count: success, + }); + } + + return { success, failed }; +} + +async function publishToPubSub( + taskInputs: TaskInput[], + periodicity: Periodicity, + publisher: CheckerPublisher, +) { + const result = await publishCheckerTasks(taskInputs, periodicity, publisher); + if (result.failed > 0) { + logger.error("Pub/Sub publish failed", { + periodicity, + failed_by_region: result.failedByRegion, + error_message: result.errors[0], + }); + } + return result; +} + +async function createCloudTasks( + taskInputs: TaskInput[], + periodicity: Periodicity, + client: CloudTasksClient, +) { + const parent = client.queuePath( + env().GCP_PROJECT_ID, + env().GCP_LOCATION, + periodicity, + ); + const results = await Effect.runPromise( Effect.forEach( taskInputs, (input) => Effect.tryPromise({ - try: () => createCronTask(input, parent), + try: () => createCronTask(input, parent, client), catch: (err) => { if (err instanceof Error && "code" in err && err.code === 6) { return "ALREADY_EXISTS" as const; @@ -253,157 +309,18 @@ export async function sendCheckerTasks( } } - const success = results.filter(Result.isSuccess).length; - const failed = results.filter(Result.isFailure).length; - - logger.info("Completed cron job", { - periodicity, - total_tasks: taskInputs.length, - success_count: success, - failed_count: failed, - duration_ms: Date.now() - timestamp, - }); - if (failed > 0) { - logger.error("Cron job had failures", { - periodicity, - failed_count: failed, - success_count: success, - }); - } - - return { success, failed }; + return { + success: results.filter(Result.isSuccess).length, + failed: results.filter(Result.isFailure).length, + }; } -// timestamp needs to be in ms + const createCronTask = async ( { row, timestamp, status, region }: TaskInput, parent: string, + client: CloudTasksClient, ) => { - // This should never happen due to the guard in sendCheckerTasks, but TypeScript needs this - if (!client) { - throw new Error("GCP Cloud Tasks client is not initialized"); - } - - let payload: - | z.infer - | z.infer - | z.infer - | z.infer - | z.infer - | null = null; - - // - if (row.jobType === "http") { - payload = { - workspaceId: String(row.workspaceId), - monitorId: String(row.id), - url: row.url, - method: row.method || "GET", - cronTimestamp: timestamp, - body: row.body, - headers: row.headers, - status: status, - assertions: row.assertions ? JSON.parse(row.assertions) : null, - degradedAfter: row.degradedAfter, - timeout: row.timeout, - trigger: "cron", - otelConfig: row.otelEndpoint - ? { - endpoint: row.otelEndpoint, - headers: transformHeaders(row.otelHeaders), - } - : undefined, - retry: row.retry || 3, - followRedirects: - row.followRedirects === null ? true : row.followRedirects, - }; - } - if (row.jobType === "tcp") { - payload = { - workspaceId: String(row.workspaceId), - monitorId: String(row.id), - uri: row.url, - status: status, - assertions: row.assertions ? JSON.parse(row.assertions) : null, - cronTimestamp: timestamp, - degradedAfter: row.degradedAfter, - timeout: row.timeout, - trigger: "cron", - retry: row.retry || 3, - otelConfig: row.otelEndpoint - ? { - endpoint: row.otelEndpoint, - headers: transformHeaders(row.otelHeaders), - } - : undefined, - }; - } - if (row.jobType === "dns") { - payload = { - workspaceId: String(row.workspaceId), - monitorId: String(row.id), - uri: row.url, - cronTimestamp: timestamp, - status: status, - assertions: row.assertions ? JSON.parse(row.assertions) : null, - degradedAfter: row.degradedAfter, - timeout: row.timeout, - trigger: "cron", - otelConfig: row.otelEndpoint - ? { - endpoint: row.otelEndpoint, - headers: transformHeaders(row.otelHeaders), - } - : undefined, - retry: row.retry || 3, - }; - } - - if (row.jobType === "icmp") { - payload = { - workspaceId: String(row.workspaceId), - monitorId: String(row.id), - uri: row.url, - cronTimestamp: timestamp, - status: status, - degradedAfter: row.degradedAfter, - timeout: row.timeout, - trigger: "cron", - otelConfig: row.otelEndpoint - ? { - endpoint: row.otelEndpoint, - headers: transformHeaders(row.otelHeaders), - } - : undefined, - retry: row.retry || 3, - }; - } - - if (row.jobType === "grpc") { - payload = { - workspaceId: String(row.workspaceId), - monitorId: String(row.id), - uri: row.url, - service: row.grpcService ?? undefined, - tls: row.grpcTls ?? "tls", - headers: transformHeaders(row.headers), - cronTimestamp: timestamp, - status: status, - degradedAfter: row.degradedAfter, - timeout: row.timeout, - trigger: "cron", - otelConfig: row.otelEndpoint - ? { - endpoint: row.otelEndpoint, - headers: transformHeaders(row.otelHeaders), - } - : undefined, - retry: row.retry || 3, - }; - } - - if (!payload) { - throw new Error("Invalid jobType"); - } + const payload = buildCheckerPayload({ row, timestamp, status }); const regionInfo = regionDict[region]; let regionHeader = {}; if (regionInfo.provider === "fly") { diff --git a/apps/workflows/src/env.ts b/apps/workflows/src/env.ts index f245135d2..638169314 100644 --- a/apps/workflows/src/env.ts +++ b/apps/workflows/src/env.ts @@ -9,6 +9,9 @@ export const env = () => GCP_CLIENT_EMAIL: z.string().prefault(""), GCP_PRIVATE_KEY: z.string().prefault(""), GCP_LOCATION: z.string().prefault("europe-west1"), + CHECKER_TRANSPORT: z + .enum(["cloudtasks", "pubsub"]) + .prefault("cloudtasks"), CRON_SECRET: z.string().prefault(""), SITE_URL: z.string().prefault("http://localhost:3000"), DATABASE_URL: z.string().prefault("http://localhost:8080"), diff --git a/apps/workflows/tsdown.config.ts b/apps/workflows/tsdown.config.ts index c4aebd513..42f64dadc 100644 --- a/apps/workflows/tsdown.config.ts +++ b/apps/workflows/tsdown.config.ts @@ -1,9 +1,15 @@ import { defineConfig } from "tsdown"; // Inline everything except the packages that can't be bundled: libsql loads a -// native .node addon, and google-proto-files (under @google-cloud/tasks) reads -// files via CJS __dirname. Those three ship as a side-car node_modules. -const external = ["@libsql/client", "libsql", "@google-cloud/tasks"]; +// native .node addon, and google-gax / google-proto-files (under +// @google-cloud/tasks and @google-cloud/pubsub) read files via CJS __dirname. +// Those ship as a side-car node_modules. +const external = [ + "@libsql/client", + "libsql", + "@google-cloud/tasks", + "@google-cloud/pubsub", +]; export default defineConfig({ entry: ["src/serve.ts"], diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 13a24b963..9a39c4d23 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -167,6 +167,9 @@ catalogs: '@dnd-kit/utilities': specifier: 3.2.2 version: 3.2.2 + '@google-cloud/pubsub': + specifier: 6.2.0 + version: 6.2.0 '@google-cloud/tasks': specifier: 7.1.1 version: 7.1.1 @@ -669,6 +672,9 @@ catalogs: specifier: 4.6.5 version: 4.6.5 +overrides: + '@google-cloud/pubsub>@opentelemetry/semantic-conventions': 1.43.0 + importers: .: @@ -1821,6 +1827,9 @@ importers: apps/workflows: dependencies: + '@google-cloud/pubsub': + specifier: 'catalog:' + version: 6.2.0(supports-color@8.1.1) '@google-cloud/tasks': specifier: 'catalog:' version: 7.1.1(supports-color@8.1.1) @@ -4796,6 +4805,30 @@ packages: '@formatjs/intl-localematcher@0.8.14': resolution: {integrity: sha512-8O7V38QmbB11+ryJ3KM/znrzXtiqgpXzqNoqDsndHZZ5/zDKTTL+cf5F0mzXkQEwhx8IlWnqYvEovf2S5EwLTw==} + '@google-cloud/paginator@7.2.0': + resolution: {integrity: sha512-l+x6YObHGTuqBCMvC0SXo3Cnyqg4FkWWKYRptf46PACHJ5GAUuhsd2b3wphgGrg1ybs/LAb+q5MnUTuG/3+v7w==} + engines: {node: '>=22'} + + '@google-cloud/precise-date@6.2.0': + resolution: {integrity: sha512-Pvh11jzEdzCLeqXNv2H40JGUsrUjpzBJQ6Ai4aNxlLlrziyVKzz71RhMmsadAjIrnEBDiPW9Uku4OuR3f6kb2g==} + engines: {node: '>=22'} + + '@google-cloud/projectify@6.2.0': + resolution: {integrity: sha512-QXBPu/D+1zgJ9Ydw+GXYws9OipW62nxfwsGl7x3JUTDim9NCGb7R+h7Sj3HKo1WMYb/StvOYyUBZrqU2pkhuiw==} + engines: {node: '>=22'} + + '@google-cloud/promisify@6.2.0': + resolution: {integrity: sha512-0i3JkkwAdGGk6AlMnFm1Vf+XtpIL73ki/MZWvXDPoN37nABf/9LFHtYeo8tW77vHSQS00ykC2ygHnUL1gkYAXg==} + engines: {node: '>=22'} + + '@google-cloud/pubsub-api@0.3.0': + resolution: {integrity: sha512-x1L+lrIHgaMwN5fAhnNVS7rh0BZn9eKmqOBXnOUM3jkilF4GE4ooETS9SYnCtPCVHwQtByDTh5b7KIMUW7XAHQ==} + engines: {node: '>=22'} + + '@google-cloud/pubsub@6.2.0': + resolution: {integrity: sha512-E9mJwt+j5HuQTkK1aJXnNmuXgG5hPRbSqRHisOm+N55Fm2RAxDCpX4ii2Hws7te3FMENpcmDS0BqQv+GIjQcXw==} + engines: {node: '>=22'} + '@google-cloud/tasks@7.1.1': resolution: {integrity: sha512-Pt8bAeYwAjPXXCnNziPAykOruLVbeQcpB6qnWkLIKAjHkyudrwEXOhucuD0IcNSgSegXBzH4BQKVjjrlo5lMOQ==} engines: {node: '>=22'} @@ -8961,6 +8994,10 @@ packages: resolution: {integrity: sha512-ik3ZgC9dY/lYVVM++OISsaYDeg1tb0VtP5uL3ouh1koGOaUMDPpbFIei4JkFimWUFPn90sbMNMXQAIVOlnYKJA==} engines: {node: '>=10'} + arrify@2.0.1: + resolution: {integrity: sha512-3duEwti880xqi4eAMN8AyR4a0ByT90zoYdLlevfrvU43vb0YZwZVfxOgxWrLXXXpyugL0hNZc9G6BiB5B3nUug==} + engines: {node: '>=8'} + assertion-error@2.0.1: resolution: {integrity: sha512-Izi8RQcffqCeNVgFigKli1ssklIbpHnCYc6AknXGYoB6grJqyeby7jv12JUQgmTAnIDnbck1uxksT4dzN3PWBA==} engines: {node: '>=12'} @@ -10089,6 +10126,10 @@ packages: headers-polyfill@5.0.1: resolution: {integrity: sha512-1TJ6Fih/b8h5TIcv+1+Hw0PDQWJTKDKzFZzcKOiW1wJza3XoAQlkCuXLbymPYB8+ZQyw8mHvdw560e8zVFIWyA==} + heap-js@2.7.1: + resolution: {integrity: sha512-EQfezRg0NCZGNlhlDR3Evrw1FVL2G3LhU7EgPoxufQKruNBSYA8MiRPHeWbU+36o+Fhel0wMwM+sLEiBAlNLJA==} + engines: {node: '>=10.0.0'} + hoist-non-react-statics@3.3.2: resolution: {integrity: sha512-/gGivxi8JPKWNm/W0jSmzcMPpfpPLc3dY/6GxhX2hQ9iGj3aDfklV4ET7NjKpSinLpJ5vafa9iiGIEZg10SfBw==} @@ -10302,6 +10343,9 @@ packages: resolution: {integrity: sha512-rbku49cWloU5bSMI+zaRaXdQHXnthP6DZ/vLnfdSKyL4zUzuWnomtOEiZZOd+ioQ+avFo/qau3KPTc7Fjy1uPA==} engines: {node: '>=12'} + is-stream-ended@0.1.4: + resolution: {integrity: sha512-xj0XPvmr7bQFTvirqnFr50o0hQIh6ZItDqloxt5aJrR4NQsYeSsyFQERYGCAzfindAcnKjINnwEEgLx4IqVzQw==} + is-stream@2.0.1: resolution: {integrity: sha512-hFoiJiTl63nn+kstHGBtewWSKnQLpyb155KHheA1l39uvtO9nWIop1p3udqPcUd/xbF1VLMO4n7OI6p7RbngDg==} engines: {node: '>=8'} @@ -10638,6 +10682,9 @@ packages: lodash.camelcase@4.3.0: resolution: {integrity: sha512-TwuEnCnxbc3rAvhf/LbG7tJUDzhqXyFnv3dtzLOPgCG/hODL7WFnsbwktkD7yUV0RrreP/l1PALq/YSg6VvjlA==} + lodash.snakecase@4.1.1: + resolution: {integrity: sha512-QZ1d4xoBHYUeuouhEq3lk3Uq7ldgyFXGBhg04+oRLnIz8o9T65Eh+8YdroUwn846zchkA9yDsDl5CVVaV2nqYw==} + lodash@4.18.1: resolution: {integrity: sha512-dMInicTPVE8d1e5otfwmmjlxkZoUpiVLwyeTdUsi/Caj/gfzzblBcCE5sRHV/AsjuCmxWrte2TNGSYuCeCq+0Q==} @@ -11300,6 +11347,10 @@ packages: vite-plus: optional: true + p-defer@3.0.0: + resolution: {integrity: sha512-ugZxsxmtTln604yeYd29EGrNhazN2lywetzpKhfmQjW/VJmhpDmWbiX+h0zL8V91R0UXkhb3KtPmyq9PZw3aYw==} + engines: {node: '>=8'} + p-finally@1.0.0: resolution: {integrity: sha512-LICb2p9CB7FS+0eR1oqWnHhp0FljGLZCWBE9aix0Uye9W8LTQPwMTYVGWQWIw9RdQiDg4+epXQODwIYJtSJaow==} engines: {node: '>=4'} @@ -14097,6 +14148,45 @@ snapshots: dependencies: '@formatjs/fast-memoize': 3.1.7 + '@google-cloud/paginator@7.2.0': + dependencies: + extend: 3.0.2 + + '@google-cloud/precise-date@6.2.0': {} + + '@google-cloud/projectify@6.2.0': {} + + '@google-cloud/promisify@6.2.0': {} + + '@google-cloud/pubsub-api@0.3.0(supports-color@8.1.1)': + dependencies: + google-gax: 6.4.0(supports-color@8.1.1) + transitivePeerDependencies: + - supports-color + + '@google-cloud/pubsub@6.2.0(supports-color@8.1.1)': + dependencies: + '@google-cloud/paginator': 7.2.0 + '@google-cloud/precise-date': 6.2.0 + '@google-cloud/projectify': 6.2.0 + '@google-cloud/promisify': 6.2.0 + '@google-cloud/pubsub-api': 0.3.0(supports-color@8.1.1) + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': 1.43.0 + arrify: 2.0.1 + extend: 3.0.2 + google-auth-library: 11.1.0(supports-color@8.1.1) + google-gax: 6.4.0(supports-color@8.1.1) + google-logging-utils: 2.0.1 + heap-js: 2.7.1 + is-stream-ended: 0.1.4 + lodash.snakecase: 4.1.1 + long: 5.3.2 + p-defer: 3.0.0 + transitivePeerDependencies: + - supports-color + '@google-cloud/tasks@7.1.1(supports-color@8.1.1)': dependencies: google-gax: 6.4.0(supports-color@8.1.1) @@ -17640,6 +17730,8 @@ snapshots: dependencies: tslib: 2.8.1 + arrify@2.0.1: {} + assertion-error@2.0.1: {} ast-types@0.16.3: @@ -18839,6 +18931,8 @@ snapshots: '@types/set-cookie-parser': 2.4.10 set-cookie-parser: 3.1.2 + heap-js@2.7.1: {} + hoist-non-react-statics@3.3.2: dependencies: react-is: 16.13.1 @@ -19033,6 +19127,8 @@ snapshots: is-regexp@3.1.0: {} + is-stream-ended@0.1.4: {} + is-stream@2.0.1: {} is-stream@4.0.1: {} @@ -19324,6 +19420,8 @@ snapshots: lodash.camelcase@4.3.0: {} + lodash.snakecase@4.1.1: {} + lodash@4.18.1: {} log-symbols@6.0.0: @@ -20262,6 +20360,8 @@ snapshots: '@oxlint/binding-win32-x64-msvc': 1.85.0 oxlint-tsgolint: 7.0.2002 + p-defer@3.0.0: {} + p-finally@1.0.0: {} p-limit@2.3.0: diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 6075b575c..2dd776871 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -30,6 +30,11 @@ allowBuilds: msw: true protobufjs: true sharp: true +# @google-cloud/pubsub pins semantic-conventions ~1.39 while the @opentelemetry/core +# it shares with us resolves to the catalog's 1.43; the workflows side-car can only +# ship one copy. The package is constants only, and newer releases only add more. +overrides: + "@google-cloud/pubsub>@opentelemetry/semantic-conventions": 1.43.0 autoInstallPeers: true # `turbo` is deliberately pinned literally in package.json rather than listed # here: Vercel can't read a version through `catalog:`, and on failing to infer @@ -56,6 +61,7 @@ catalog: "@dnd-kit/modifiers": 9.0.0 "@dnd-kit/sortable": 10.0.0 "@dnd-kit/utilities": 3.2.2 + "@google-cloud/pubsub": 6.2.0 "@google-cloud/tasks": 7.1.1 "@hono/mcp": 0.2.5 "@hono/node-server": 2.1.4 -- 2.54.0