1import * as GCP from "alchemy/GCP";
2import * as Effect from "effect/Effect";
3import { HttpServerRequest } from "effect/http/HttpServerRequest";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5import { Heartbeats, type HeartbeatRow } from "./resources.ts";
6
7/** Most rows `GET /heartbeats` returns. */
8const MAX_LIMIT = 100;
9
10/**
11 * A Cloud Run service with two cron handlers and one read route.
12 *
13 * Each `consumeSchedule` creates a Cloud Scheduler job that `POST`s to a
14 * private path on this service with an OIDC token for the service's own
15 * runtime account. The runtime verifies that token before the handler
16 * runs, so the schedule routes stay closed to the public even though
17 * `GET /heartbeats` is open.
18 *
19 * - every minute — record a heartbeat and how late it was delivered.
20 * - daily at midnight UTC — count the last day's heartbeats and record it.
21 * - `GET /heartbeats?kind=&limit=` — the most recent rows, newest first.
22 */
23export default class Monitor extends GCP.Function<Monitor>()(
24 "Monitor",
25 {
26 main: import.meta.url,
27 // Opens `GET /heartbeats`. The schedule routes still reject any
28 // request without a valid token from Cloud Scheduler.
29 invokerIamDisabled: true,
30 },
31 Effect.gen(function* () {
32 const table = yield* Heartbeats;
33 const writer = yield* GCP.BigQuery.WriteTable(table);
34 const reader = yield* GCP.BigQuery.ReadTable(table);
35 // The table id is bound at deploy time and read inside handlers.
36 const tableId = yield* table.tableId;
37
38 const record = (row: HeartbeatRow) =>
39 writer.insert([row]).pipe(Effect.orDie);
40
41 yield* GCP.CloudScheduler.consumeSchedule(
42 "Heartbeat",
43 { schedule: "* * * * *", timeZone: "Etc/UTC" },
44 (event) =>
45 Effect.gen(function* () {
46 const receivedAt = new Date();
47 const scheduleTime = event.scheduleTime
48 ? new Date(event.scheduleTime)
49 : receivedAt;
50 yield* record({
51 kind: "heartbeat",
52 jobName: event.jobName,
53 scheduleTime,
54 receivedAt,
55 value: receivedAt.getTime() - scheduleTime.getTime(),
56 });
57 }),
58 );
59
60 yield* GCP.CloudScheduler.consumeSchedule(
61 "DailyRollup",
62 { schedule: "0 0 * * *", timeZone: "Etc/UTC" },
63 (event) =>
64 Effect.gen(function* () {
65 const heartbeats = yield* tableId;
66 const [counted] = yield* reader
67 .query(
68 `SELECT COUNT(*) AS n FROM \`${heartbeats}\`
69 WHERE kind = 'heartbeat'
70 AND receivedAt >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)`,
71 )
72 .pipe(Effect.orDie);
73 const receivedAt = new Date();
74 yield* record({
75 kind: "daily",
76 jobName: event.jobName,
77 scheduleTime: event.scheduleTime
78 ? new Date(event.scheduleTime)
79 : receivedAt,
80 receivedAt,
81 value: Number(counted?.n ?? 0),
82 });
83 }),
84 );
85
86 return {
87 fetch: Effect.gen(function* () {
88 const request = yield* HttpServerRequest;
89 const url = new URL(request.originalUrl);
90
91 if (request.method === "GET" && url.pathname === "/") {
92 return HttpServerResponse.text("ok");
93 }
94
95 if (request.method === "GET" && url.pathname === "/heartbeats") {
96 const kind = url.searchParams.get("kind");
97 const limit = Math.min(
98 Math.max(Number(url.searchParams.get("limit") ?? 20) || 20, 1),
99 MAX_LIMIT,
100 );
101 const heartbeats = yield* tableId;
102 const rows = yield* reader
103 .query(
104 `SELECT kind, jobName, scheduleTime, receivedAt, value
105 FROM \`${heartbeats}\`
106 ${kind ? "WHERE kind = @kind" : ""}
107 ORDER BY receivedAt DESC
108 LIMIT @limit`,
109 kind ? { kind, limit } : { limit },
110 )
111 .pipe(Effect.orDie);
112 return yield* HttpServerResponse.json({ heartbeats: rows });
113 }
114
115 return yield* HttpServerResponse.json(
116 { error: "not found" },
117 { status: 404 },
118 );
119 }),
120 };
121 }).pipe(
122 Effect.provide([
123 GCP.Run.ScheduleEventSource,
124 GCP.BigQuery.WriteTableHttp,
125 GCP.BigQuery.ReadTableHttp,
126 ]),
127 ),
128) {}