diff --git a/website/app/routes/uploadFromWeatherStation.ts b/website/app/routes/uploadFromWeatherStation.ts index e4a21c0..a0fd489 100644 --- a/website/app/routes/uploadFromWeatherStation.ts +++ b/website/app/routes/uploadFromWeatherStation.ts @@ -4,6 +4,7 @@ import { observationFromWeatherStation, observationInsertSchema, } from "../../database/schema.d"; +import { dispatchObservation } from "../../workers/uploads"; import type { Route } from "./+types/uploadFromWeatherStation"; export async function action({ request, context, params }: Route.ActionArgs) { @@ -86,19 +87,11 @@ export async function action({ request, context, params }: Route.ActionArgs) { context.db.insert(schema.Observations).values(fullValidation.data) ); context.cloudflare.ctx.waitUntil( - context.cloudflare.env.WORKFLOW_UPLOAD_TO_METOFFICE.create({ - params: fullValidation.data, - }) - ); - context.cloudflare.ctx.waitUntil( - context.cloudflare.env.WORKFLOW_UPLOAD_TO_WINDGURU.create({ - params: fullValidation.data, - }) - ); - context.cloudflare.ctx.waitUntil( - context.cloudflare.env.WORKFLOW_UPLOAD_TO_WINDY.create({ - params: fullValidation.data, - }) + dispatchObservation( + context.cloudflare.env, + context.db, + fullValidation.data + ) ); return data({ message: "Observation received", diff --git a/website/database/migrations/0004_wealthy_wildside.sql b/website/database/migrations/0004_wealthy_wildside.sql new file mode 100644 index 0000000..91989bf --- /dev/null +++ b/website/database/migrations/0004_wealthy_wildside.sql @@ -0,0 +1,6 @@ +CREATE TABLE `upload_state` ( + `target` text PRIMARY KEY NOT NULL, + `last_attempt_at` integer, + `last_success_at` integer, + `last_error` text +); diff --git a/website/database/migrations/meta/0004_snapshot.json b/website/database/migrations/meta/0004_snapshot.json new file mode 100644 index 0000000..370be98 --- /dev/null +++ b/website/database/migrations/meta/0004_snapshot.json @@ -0,0 +1,179 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "5ec74d8e-20f9-4102-a14d-1bc93b75e84d", + "prevId": "2f57c7cd-0a6e-4818-b5e9-f7e278994bff", + "tables": { + "disregarded_observations": { + "name": "disregarded_observations", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "timestamp": { + "name": "timestamp", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "disregardReasonFriendly": { + "name": "disregardReasonFriendly", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "disregardReasonDetailed": { + "name": "disregardReasonDetailed", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "heartbeats": { + "name": "heartbeats", + "columns": { + "hour_start_timestamp": { + "name": "hour_start_timestamp", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "ping_count": { + "name": "ping_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "observations": { + "name": "observations", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "timestamp": { + "name": "timestamp", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "exported_to_r2": { + "name": "exported_to_r2", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "upload_state": { + "name": "upload_state", + "columns": { + "target": { + "name": "target", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "last_attempt_at": { + "name": "last_attempt_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_success_at": { + "name": "last_success_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_error": { + "name": "last_error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/website/database/migrations/meta/_journal.json b/website/database/migrations/meta/_journal.json index 76855c6..5f5156e 100644 --- a/website/database/migrations/meta/_journal.json +++ b/website/database/migrations/meta/_journal.json @@ -29,6 +29,13 @@ "when": 1754915033969, "tag": "0003_powerful_dagger", "breakpoints": true + }, + { + "idx": 4, + "version": "6", + "when": 1785085307986, + "tag": "0004_wealthy_wildside", + "breakpoints": true } ] } \ No newline at end of file diff --git a/website/database/schema.d.ts b/website/database/schema.d.ts index 78f0822..4880a79 100644 --- a/website/database/schema.d.ts +++ b/website/database/schema.d.ts @@ -1,3 +1,4 @@ export * from "./schema/DisregardedObservations"; export * from "./schema/Heartbeats"; export * from "./schema/Observations"; +export * from "./schema/UploadState"; diff --git a/website/database/schema/UploadState.ts b/website/database/schema/UploadState.ts new file mode 100644 index 0000000..5169a16 --- /dev/null +++ b/website/database/schema/UploadState.ts @@ -0,0 +1,22 @@ +import { integer, sqliteTable, text } from "drizzle-orm/sqlite-core"; + +/** + * One row per external weather service we upload observations to. + * + * `lastAttemptAt` drives the per-service throttle - observations arrive from the + * weather station roughly every 30 seconds, which is far more often than the + * services will accept, so we only dispatch to a service once its minimum + * interval has elapsed. See workers/uploads/index.ts for the intervals. + * + * `lastError` exists so upload failures are still visible now that these + * uploads no longer run as Workflows (which had their own dashboard). + */ +export const UploadState = sqliteTable("upload_state", { + target: text("target").$type().primaryKey(), // The service, e.g. "windguru" + lastAttemptAt: integer("last_attempt_at", { mode: "timestamp" }), // Last time we tried, successful or not + lastSuccessAt: integer("last_success_at", { mode: "timestamp" }), // Last time an upload succeeded + lastError: text("last_error"), // Why the last attempt failed, or null if it succeeded +}); + +export const uploadTargets = ["windguru", "metoffice", "windy"] as const; +export type UploadTarget = (typeof uploadTargets)[number]; diff --git a/website/worker-configuration.d.ts b/website/worker-configuration.d.ts index 5a22737..0f3e27b 100644 --- a/website/worker-configuration.d.ts +++ b/website/worker-configuration.d.ts @@ -1,6 +1,6 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: db3bd3af6c26d8b3568abd26bb3ee392) -// Runtime types generated with workerd@1.20250617.0 2025-04-05 nodejs_compat +// Generated by Wrangler by running `wrangler types` (hash: 6b253f0de7e6b493a682c1416e2aca7e) +// Runtime types generated with workerd@1.20250803.0 2025-04-05 nodejs_compat declare namespace Cloudflare { interface Env { KV: KVNamespace; @@ -8,9 +8,6 @@ declare namespace Cloudflare { DB: D1Database; CF_VERSION_METADATA: WorkerVersionMetadata; WORKFLOW_OVERNIGHT_SAVE_TO_R2: Workflow; - WORKFLOW_UPLOAD_TO_WINDGURU: Workflow; - WORKFLOW_UPLOAD_TO_METOFFICE: Workflow; - WORKFLOW_UPLOAD_TO_WINDY: Workflow; } } interface Env extends Cloudflare.Env {} @@ -353,7 +350,7 @@ interface ExecutionContext { type ExportedHandlerFetchHandler = (request: Request>, env: Env, ctx: ExecutionContext) => Response | Promise; type ExportedHandlerTailHandler = (events: TraceItem[], env: Env, ctx: ExecutionContext) => void | Promise; type ExportedHandlerTraceHandler = (traces: TraceItem[], env: Env, ctx: ExecutionContext) => void | Promise; -type ExportedHandlerTailStreamHandler = (event: TailStream.TailEvent, env: Env, ctx: ExecutionContext) => TailStream.TailEventHandlerType | Promise; +type ExportedHandlerTailStreamHandler = (event: TailStream.TailEvent, env: Env, ctx: ExecutionContext) => TailStream.TailEventHandlerType | Promise; type ExportedHandlerScheduledHandler = (controller: ScheduledController, env: Env, ctx: ExecutionContext) => void | Promise; type ExportedHandlerQueueHandler = (batch: MessageBatch, env: Env, ctx: ExecutionContext) => void | Promise; type ExportedHandlerTestHandler = (controller: TestController, env: Env, ctx: ExecutionContext) => void | Promise; @@ -1106,6 +1103,47 @@ interface ErrorEventErrorEventInit { colno?: number; error?: any; } +/** + * A message received by a target object. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent) + */ +declare class MessageEvent extends Event { + constructor(type: string, initializer: MessageEventInit); + /** + * Returns the data of the message. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/data) + */ + readonly data: any; + /** + * Returns the origin of the message, for server-sent events and cross-document messaging. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/origin) + */ + readonly origin: string | null; + /** + * Returns the last event ID string, for server-sent events. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/lastEventId) + */ + readonly lastEventId: string; + /** + * Returns the WindowProxy of the source window, for cross-document messaging, and the MessagePort being attached, in the connect event fired at SharedWorkerGlobalScope objects. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/source) + */ + readonly source: MessagePort | null; + /** + * Returns the MessagePort array sent with the message, for cross-document messaging and channel messaging. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/ports) + */ + readonly ports: MessagePort[]; +} +interface MessageEventInit { + data: ArrayBuffer | string; +} /** * Provides a way to easily construct a set of key/value pairs representing form fields and their values, which can then be easily sent using the XMLHttpRequest.send() method. It uses the same format a form would use if the encoding type were set to "multipart/form-data". * @@ -1414,7 +1452,7 @@ interface RequestInit { signal?: (AbortSignal | null); encodeResponseBody?: "automatic" | "manual"; } -type Service = Fetcher; +type Service Rpc.WorkerEntrypointBranded) | Rpc.WorkerEntrypointBranded | ExportedHandler | undefined = undefined> = T extends new (...args: any[]) => Rpc.WorkerEntrypointBranded ? Fetcher> : T extends Rpc.WorkerEntrypointBranded ? Fetcher : T extends Exclude ? never : Fetcher; type Fetcher = (T extends Rpc.EntrypointBranded ? Rpc.Provider : unknown) & { fetch(input: RequestInfo | URL, init?: RequestInit): Promise; connect(address: SocketAddress | string, options?: SocketOptions): Socket; @@ -2287,23 +2325,6 @@ interface CloseEventInit { reason?: string; wasClean?: boolean; } -/** - * A message received by a target object. - * - * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent) - */ -declare class MessageEvent extends Event { - constructor(type: string, initializer: MessageEventInit); - /** - * Returns the data of the message. - * - * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessageEvent/data) - */ - readonly data: ArrayBuffer | string; -} -interface MessageEventInit { - data: ArrayBuffer | string; -} type WebSocketEventMap = { close: CloseEvent; message: MessageEvent; @@ -2491,6 +2512,38 @@ interface ContainerStartupOptions { enableInternet: boolean; env?: Record; } +/** + * This Channel Messaging API interface represents one of the two ports of a MessageChannel, allowing messages to be sent from one port and listening out for them arriving at the other. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort) + */ +interface MessagePort extends EventTarget { + /** + * Posts a message through the channel. Objects listed in transfer are transferred, not just cloned, meaning that they are no longer usable on the sending side. + * + * Throws a "DataCloneError" DOMException if transfer contains duplicate objects or port, or if message could not be cloned. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort/postMessage) + */ + postMessage(data?: any, options?: (any[] | MessagePortPostMessageOptions)): void; + /** + * Disconnects the port, so that it is no longer active. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort/close) + */ + close(): void; + /** + * Begins dispatching messages received on the port. + * + * [MDN Reference](https://developer.mozilla.org/docs/Web/API/MessagePort/start) + */ + start(): void; + get onmessage(): any | null; + set onmessage(value: any | null); +} +interface MessagePortPostMessageOptions { + transfer?: any[]; +} type AiImageClassificationInput = { image: number[]; }; @@ -5266,7 +5319,7 @@ type AiModelListType = Record; declare abstract class Ai { aiGatewayLogId: string | null; gateway(gatewayId: string): AiGateway; - autorag(autoragId: string): AutoRAG; + autorag(autoragId?: string): AutoRAG; run(model: Name, inputs: InputOptions, options?: Options): Promise { * * @example 395747 */ - asn: number; + asn?: number; /** * The organization which owns the ASN of the incoming request. * * @example "Google Cloud" */ - asOrganization: string; + asOrganization?: string; /** * The original value of the `Accept-Encoding` header if Cloudflare modified it. * @@ -5910,7 +5963,7 @@ interface IncomingRequestCfPropertiesCloudflareForSaaSEnterprise { * This field is only present if you have Cloudflare for SaaS enabled on your account * and you have followed the [required steps to enable it]((https://developers.cloudflare.com/cloudflare-for-platforms/cloudflare-for-saas/domain-support/custom-metadata/)). */ - hostMetadata: HostMetadata; + hostMetadata?: HostMetadata; } interface IncomingRequestCfPropertiesCloudflareAccessOrApiShield { /** @@ -6336,6 +6389,22 @@ declare module "cloudflare:email" { }; export { _EmailMessage as EmailMessage }; } +/** + * Hello World binding to serve as an explanatory example. DO NOT USE + */ +interface HelloWorldBinding { + /** + * Retrieve the current stored value + */ + get(): Promise<{ + value: string; + ms?: number; + }>; + /** + * Set a new stored value + */ + set(value: string): Promise; +} interface Hyperdrive { /** * Connect directly to Hyperdrive as if it's your database, returning a TCP socket. @@ -6421,7 +6490,7 @@ type ImageTransform = { rotate?: 0 | 90 | 180 | 270; saturation?: number; sharpen?: number; - trim?: "border" | { + trim?: 'border' | { top?: number; bottom?: number; left?: number; @@ -6443,6 +6512,9 @@ type ImageDrawOptions = { bottom?: number; right?: number; }; +type ImageInputOptions = { + encoding?: 'base64'; +}; type ImageOutputOptions = { format: 'image/jpeg' | 'image/png' | 'image/gif' | 'image/webp' | 'image/avif' | 'rgb' | 'rgba'; quality?: number; @@ -6454,13 +6526,13 @@ interface ImagesBinding { * @throws {@link ImagesError} with code 9412 if input is not an image * @param stream The image bytes */ - info(stream: ReadableStream): Promise; + info(stream: ReadableStream, options?: ImageInputOptions): Promise; /** * Begin applying a series of transformations to an image * @param stream The image bytes * @returns A transform handle */ - input(stream: ReadableStream): ImageTransformer; + input(stream: ReadableStream, options?: ImageInputOptions): ImageTransformer; } interface ImageTransformer { /** @@ -6483,6 +6555,9 @@ interface ImageTransformer { */ output(options: ImageOutputOptions): Promise; } +type ImageTransformationOutputOptions = { + encoding?: 'base64'; +}; interface ImageTransformationResult { /** * The image as a response, ready to store in cache or return to users @@ -6495,7 +6570,7 @@ interface ImageTransformationResult { /** * The bytes of the response */ - image(): ReadableStream; + image(options?: ImageTransformationOutputOptions): ReadableStream; } interface ImagesError extends Error { readonly code: number; @@ -6720,6 +6795,19 @@ declare namespace Cloudflare { interface Env { } } +declare module 'cloudflare:node' { + export interface DefaultHandler { + fetch?(request: Request): Response | Promise; + tail?(events: TraceItem[]): void | Promise; + trace?(traces: TraceItem[]): void | Promise; + scheduled?(controller: ScheduledController): void | Promise; + queue?(batch: MessageBatch): void | Promise; + test?(controller: TestController): void | Promise; + } + export function httpServerHandler(options: { + port: number; + }, handlers?: Omit): DefaultHandler; +} declare module 'cloudflare:workers' { export type RpcStub = Rpc.Stub; export const RpcStub: { @@ -6793,6 +6881,7 @@ declare module 'cloudflare:workers' { constructor(ctx: ExecutionContext, env: Env); run(event: Readonly>, step: WorkflowStep): Promise; } + export function waitUntil(promise: Promise): void; export const env: Cloudflare.Env; } interface SecretsStoreSecret { @@ -6815,7 +6904,7 @@ declare namespace TailStream { readonly type: "fetch"; readonly method: string; readonly url: string; - readonly cfJson: string; + readonly cfJson?: object; readonly headers: Header[]; } interface JsRpcEventInfo { @@ -6926,7 +7015,7 @@ declare namespace TailStream { interface Log { readonly type: "log"; readonly level: "debug" | "error" | "info" | "log" | "warn"; - readonly message: string; + readonly message: object; } interface Return { readonly type: "return"; @@ -6947,17 +7036,27 @@ declare namespace TailStream { readonly type: "attributes"; readonly info: Attribute[]; } - interface TailEvent { - readonly traceId: string; + type EventType = Onset | Outcome | Hibernate | SpanOpen | SpanClose | DiagnosticChannelEvent | Exception | Log | Return | Link | Attributes; + interface TailEvent { readonly invocationId: string; readonly spanId: string; readonly timestamp: Date; readonly sequence: number; - readonly event: Onset | Outcome | Hibernate | SpanOpen | SpanClose | DiagnosticChannelEvent | Exception | Log | Return | Link | Attributes; + readonly event: Event; } - type TailEventHandler = (event: TailEvent) => void | Promise; - type TailEventHandlerName = "outcome" | "hibernate" | "spanOpen" | "spanClose" | "diagnosticChannel" | "exception" | "log" | "return" | "link" | "attributes"; - type TailEventHandlerObject = Record; + type TailEventHandler = (event: TailEvent) => void | Promise; + type TailEventHandlerObject = { + outcome?: TailEventHandler; + hibernate?: TailEventHandler; + spanOpen?: TailEventHandler; + spanClose?: TailEventHandler; + diagnosticChannel?: TailEventHandler; + exception?: TailEventHandler; + log?: TailEventHandler; + return?: TailEventHandler; + link?: TailEventHandler; + attributes?: TailEventHandler; + }; type TailEventHandlerType = TailEventHandler | TailEventHandlerObject; } // Copyright (c) 2022-2023 Cloudflare, Inc. diff --git a/website/workers/app.ts b/website/workers/app.ts index d69ac72..69fe387 100644 --- a/website/workers/app.ts +++ b/website/workers/app.ts @@ -1,7 +1,7 @@ import { WorkflowEntrypoint, type WorkflowEvent, - WorkflowStep, + type WorkflowStep, } from "cloudflare:workers"; import { NonRetryableError } from "cloudflare:workflows"; import { and, asc, eq, gte, lt, sql } from "drizzle-orm"; @@ -99,218 +99,6 @@ export default { }, } satisfies ExportedHandler; -export class UploadToWindGuru extends WorkflowEntrypoint< - Env, - schema.ObservationInsert -> { - async run( - event: WorkflowEvent, - step: WorkflowStep, - ) { - await step.do( - "Upload to WindGuru", - { - retries: { - limit: 2, - delay: 60000, - backoff: "exponential", - }, - timeout: "5 seconds", - }, - async () => { - const WINDGURU_UID = await this.env.KV.get("WINDGURU_UID"); - const WINDGURU_PASSWORD = await this.env.KV.get("WINDGURU_PASSWORD"); - if (!WINDGURU_UID || !WINDGURU_PASSWORD) - throw new NonRetryableError("Missing Windguru credentials"); - const payloadData = await schema.observationInsertSchema.safeParseAsync( - event.payload, - ); - if (!payloadData.success) - throw new NonRetryableError("Issue with incoming data"); - const data = payloadData.data; - - const salt = Date.now(); - const hash = await crypto.subtle.digest( - { - name: "MD5", - }, - new TextEncoder().encode( - `${salt}${WINDGURU_UID}${WINDGURU_PASSWORD}`, - ), - ); - const params = new URLSearchParams({ - uid: WINDGURU_UID, - interval: "120", - wind_avg: (data.data.wind2MinAverage / 1.151).toString(), - wind_direction: data.data.windDirection.toString(), - temperature: data.data.temperatureC.toString(), - datetime: data.timestamp.toISOString(), - salt: salt.toString(), - hash: Array.from(new Uint8Array(hash)) - .map((b) => b.toString(16).padStart(2, "0")) - .join(""), - }); - const response = await fetch( - `https://www.windguru.cz/upload/api.php?${params.toString()}`, - { - method: "GET", - }, - ).then((response) => { - if (!response.ok) - throw new Error("Windguru failed, status: " + response.status); - return response.text(); - }); - if (response !== "OK") { - throw new Error(`Windguru upload failed: ${response}`); - } - }, - ); - } -} - -export class UploadToMetOffice extends WorkflowEntrypoint< - Env, - schema.ObservationInsert -> { - async run( - event: WorkflowEvent, - step: WorkflowStep, - ) { - await step.do( - "Upload to Met Office", - { - retries: { - limit: 2, - delay: 60000, - backoff: "exponential", - }, - timeout: "5 seconds", - }, - async () => { - const METOFFICE_SITE_ID = await this.env.KV.get("METOFFICE_SITE_ID"); - const METOFFICE_AUTH_KEY = await this.env.KV.get("METOFFICE_AUTH_KEY"); - if (!METOFFICE_SITE_ID || !METOFFICE_AUTH_KEY) - throw new NonRetryableError("Missing Met Office credentials"); - const payloadData = await schema.observationInsertSchema.safeParseAsync( - event.payload, - ); - if (!payloadData.success) - throw new NonRetryableError("Issue with incoming data"); - const data = payloadData.data; - - const params = new URLSearchParams({ - siteid: METOFFICE_SITE_ID, - siteAuthenticationKey: METOFFICE_AUTH_KEY, - softwaretype: "PSC Weather Station", - dateutc: data.timestamp - .toISOString() - .replace("T", "+") - .replace(/:/g, "%3A") - .split(".")[0], - dailyrainin: (data.data.lastHourRain * 0.0393701).toString(), // real number [in]; rain inches over the past hour (alternative to precip) - dewptf: data.data.dewPoint.toString(), - tempf: data.data.temperatureF.toString(), // real number [°F]; air temperature - windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) - winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction - //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) - //windgustdir: data.data.windGustDirection.toString(), // integer number [deg]; instantaneous wind direction - }); - const response = await fetch( - `http://wow.metoffice.gov.uk/automaticreading?${params.toString()}`, - { - method: "GET", - }, - ).then((response) => { - if (response.status === 429) - throw new NonRetryableError( - "Met office asked for a backoff, skip this observation", - ); - if (!response.ok) - throw new Error( - "Met Office upload failed, status: " + response.status, - ); - return response.json(); - }); - if ( - !response || - typeof response !== "object" || - Object.keys(response).length !== 0 - ) - throw new Error( - `Met Office upload failed: ${JSON.stringify(response)}`, - ); - }, - ); - } -} - -export class UploadToWindy extends WorkflowEntrypoint< - Env, - schema.ObservationInsert -> { - async run( - event: WorkflowEvent, - step: WorkflowStep, - ) { - await step.do( - "Upload to Windy", - { - retries: { - limit: 2, - delay: 60000, - backoff: "exponential", - }, - timeout: "5 seconds", - }, - async () => { - const WINDY_API_KEY = await this.env.KV.get("WINDY_API_KEY"); - const WINDY_STATION_ID = await this.env.KV.get("WINDY_STATION_ID"); - if (!WINDY_API_KEY || !WINDY_STATION_ID) - throw new NonRetryableError("Missing Windy credentials"); - const payloadData = await schema.observationInsertSchema.safeParseAsync( - event.payload, - ); - if (!payloadData.success) - throw new NonRetryableError("Issue with incoming data"); - const data = payloadData.data; - - const params = new URLSearchParams({ - station: WINDY_STATION_ID, // 32 bit integer; required for multiple stations; default value 0; alternative names: si, stationId - ts: data.timestamp.getTime().toString(), - temp: data.data.temperatureC.toString(), // real number [°C]; air temperature - windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) - winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction - //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) - dewpoint: data.data.dewPoint.toString(), // real number [°C]; - precip: data.data.lastHourRain.toString(), // real number [mm]; precipitation over the past hour - }); - const response = await fetch( - `https://stations.windy.com/pws/update/${WINDY_API_KEY}?${params.toString()}`, - { - method: "GET", - }, - ); - const responseText = await response.text(); - if ( - responseText.length > 0 && - responseText.includes( - "Measurement sent too soon, update interval is 5 minutes", - ) - ) { - throw new NonRetryableError( - "Windy asked for a backoff, skip this observation", - ); - } else if (!response.ok) - throw new Error("Windy upload failed, status: " + response.status); - const responseJson = JSON.parse(responseText); - if (!responseJson || typeof responseJson !== "object") - throw new Error(`Windy upload failed: ${JSON.stringify(response)}`); - return responseText; - }, - ); - } -} - interface OvernightSaveToR2Params { dayToProcess: Date; } diff --git a/website/workers/uploads/index.ts b/website/workers/uploads/index.ts new file mode 100644 index 0000000..36a9b77 --- /dev/null +++ b/website/workers/uploads/index.ts @@ -0,0 +1,130 @@ +import { eq } from "drizzle-orm"; +import type { DrizzleD1Database } from "drizzle-orm/d1"; +import * as schema from "../../database/schema.d"; +import type { ObservationInsert, UploadTarget } from "../../database/schema.d"; +import { uploadToMetOffice } from "./metoffice"; +import { withRetry } from "./retry"; +import { uploadToWindGuru } from "./windguru"; +import { uploadToWindy } from "./windy"; + +interface UploadTargetConfig { + /** Human readable name, used in logs */ + name: string; + /** + * Shortest gap we will leave between dispatches to this service. The weather + * station reports roughly every 30 seconds, which is far more often than any + * of these services want, so most observations are skipped for most targets. + */ + minimumIntervalMs: number; + upload: (env: Env, data: ObservationInsert) => Promise; +} + +const TARGETS: Record = { + windguru: { + name: "Windguru", + minimumIntervalMs: 2 * 60 * 1000, // Matches the interval=120 we send in the payload + upload: uploadToWindGuru, + }, + metoffice: { + name: "Met Office", + minimumIntervalMs: 5 * 60 * 1000, // WOW guidance for automatic readings, and stops the 429s + upload: uploadToMetOffice, + }, + windy: { + name: "Windy", + minimumIntervalMs: 5 * 60 * 1000, // Windy's hard minimum - it rejects anything sooner + upload: uploadToWindy, + }, +}; + +/** Keep `last_error` from growing unbounded */ +const MAX_ERROR_LENGTH = 500; + +const describeError = (error: unknown) => + (error instanceof Error ? error.message : String(error)).slice( + 0, + MAX_ERROR_LENGTH, + ); + +/** + * Uploads an observation to every external service that is due one. + * + * Intended to be called from `ctx.waitUntil` so the weather station gets its + * response immediately - it can spend up to ~35 seconds retrying. The + * observation is already saved to D1 by the caller before this runs, so a + * failure here only ever costs us a data point on a third party site, and the + * next observation arrives ~30 seconds later regardless. + */ +export async function dispatchObservation( + env: Env, + db: DrizzleD1Database, + observation: ObservationInsert, +) { + try { + const now = new Date(); + const state = await db.select().from(schema.UploadState); + const lastAttemptByTarget = new Map( + state.map((row) => [row.target, row.lastAttemptAt]), + ); + + const dueTargets = schema.uploadTargets.filter((target) => { + const lastAttemptAt = lastAttemptByTarget.get(target); + if (!lastAttemptAt) return true; + return ( + now.getTime() - lastAttemptAt.getTime() >= + TARGETS[target].minimumIntervalMs + ); + }); + if (dueTargets.length === 0) { + console.debug("No upload targets are due, skipping this observation"); + return; + } + + // Claim the slot before uploading, so an observation that arrives while a + // slow retry is still running doesn't dispatch to the same service again + const [firstClaim, ...remainingClaims] = dueTargets.map((target) => + db + .insert(schema.UploadState) + .values({ target, lastAttemptAt: now }) + .onConflictDoUpdate({ + target: schema.UploadState.target, + set: { lastAttemptAt: now }, + }), + ); + await db.batch([firstClaim, ...remainingClaims]); + + const results = await Promise.allSettled( + dueTargets.map((target) => + withRetry(TARGETS[target].name, () => + TARGETS[target].upload(env, observation), + ), + ), + ); + + const [firstUpdate, ...remainingUpdates] = dueTargets.map( + (target, index) => { + const result = results[index]; + if (result.status === "fulfilled") { + console.log(`Uploaded observation to ${TARGETS[target].name}`); + return db + .update(schema.UploadState) + .set({ lastSuccessAt: now, lastError: null }) + .where(eq(schema.UploadState.target, target)); + } + console.error( + `Failed to upload observation to ${TARGETS[target].name}`, + result.reason, + ); + return db + .update(schema.UploadState) + .set({ lastError: describeError(result.reason) }) + .where(eq(schema.UploadState.target, target)); + }, + ); + await db.batch([firstUpdate, ...remainingUpdates]); + } catch (error) { + // Never let this reject - it runs detached in waitUntil and the observation + // itself has already been saved + console.error("Failed to dispatch observation to upload targets", error); + } +} diff --git a/website/workers/uploads/metoffice.ts b/website/workers/uploads/metoffice.ts new file mode 100644 index 0000000..8115f8d --- /dev/null +++ b/website/workers/uploads/metoffice.ts @@ -0,0 +1,46 @@ +import type { ObservationInsert } from "../../database/schema.d"; +import { NonRetryableUploadError, UPLOAD_TIMEOUT_MS } from "./retry"; + +export async function uploadToMetOffice(env: Env, data: ObservationInsert) { + const METOFFICE_SITE_ID = await env.KV.get("METOFFICE_SITE_ID"); + const METOFFICE_AUTH_KEY = await env.KV.get("METOFFICE_AUTH_KEY"); + if (!METOFFICE_SITE_ID || !METOFFICE_AUTH_KEY) + throw new NonRetryableUploadError("Missing Met Office credentials"); + + const params = new URLSearchParams({ + siteid: METOFFICE_SITE_ID, + siteAuthenticationKey: METOFFICE_AUTH_KEY, + softwaretype: "PSC Weather Station", + // WOW wants "YYYY-mm-DD HH:mm:ss" - URLSearchParams does the escaping for + // us, so this must be the plain unescaped value + dateutc: data.timestamp.toISOString().replace("T", " ").split(".")[0], + dailyrainin: (data.data.lastHourRain * 0.0393701).toString(), // real number [in]; rain inches over the past hour (alternative to precip) + dewptf: data.data.dewPoint.toString(), + tempf: data.data.temperatureF.toString(), // real number [°F]; air temperature + windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) + winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction + //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) + //windgustdir: data.data.windGustDirection.toString(), // integer number [deg]; instantaneous wind direction + }); + const response = await fetch( + `http://wow.metoffice.gov.uk/automaticreading?${params.toString()}`, + { + method: "GET", + signal: AbortSignal.timeout(UPLOAD_TIMEOUT_MS), + }, + ).then((response) => { + if (response.status === 429) + throw new NonRetryableUploadError( + "Met office asked for a backoff, skip this observation", + ); + if (!response.ok) + throw new Error("Met Office upload failed, status: " + response.status); + return response.json(); + }); + if ( + !response || + typeof response !== "object" || + Object.keys(response).length !== 0 + ) + throw new Error(`Met Office upload failed: ${JSON.stringify(response)}`); +} diff --git a/website/workers/uploads/retry.ts b/website/workers/uploads/retry.ts new file mode 100644 index 0000000..b99e0cf --- /dev/null +++ b/website/workers/uploads/retry.ts @@ -0,0 +1,58 @@ +/** + * Retries used to come from Cloudflare Workflows, which billed per step and so + * became expensive at one observation every ~30 seconds across three services. + * These uploads now run inline under `ctx.waitUntil`, so retries have to finish + * within the invocation that received the observation. + * + * Three attempts spread over ~18 seconds (plus up to 5 seconds per attempt + * waiting on the network) keeps the worst case around 33 seconds. That fits + * inside the shortest throttle window (2 minutes, see ./index.ts), so a slow + * retry can never overlap the next dispatch to the same service. Sleeping costs + * no CPU time, which matters on the free plan's 10ms per-invocation limit. + */ +const RETRY_DELAYS_MS = [3000, 15000]; + +/** How long a single request to an external service may take. */ +export const UPLOAD_TIMEOUT_MS = 5000; + +/** + * Thrown when retrying would be pointless or unwelcome - missing credentials, + * data the service will never accept, or the service explicitly asking us to + * back off. Replaces `NonRetryableError` from `cloudflare:workflows`. + */ +export class NonRetryableUploadError extends Error { + constructor(message: string, options?: ErrorOptions) { + super(message, options); + this.name = "NonRetryableUploadError"; + } +} + +const sleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * Runs `upload`, retrying transient failures. Gives up immediately on a + * NonRetryableUploadError, and rethrows the final error if every attempt fails. + */ +export async function withRetry( + name: string, + upload: () => Promise, +): Promise { + let lastError: unknown; + for (let attempt = 0; attempt <= RETRY_DELAYS_MS.length; attempt++) { + if (attempt > 0) await sleep(RETRY_DELAYS_MS[attempt - 1]); + try { + return await upload(); + } catch (error) { + lastError = error; + if (error instanceof NonRetryableUploadError) throw error; + console.warn( + `${name} upload attempt ${attempt + 1} of ${ + RETRY_DELAYS_MS.length + 1 + } failed`, + error, + ); + } + } + throw lastError; +} diff --git a/website/workers/uploads/windguru.ts b/website/workers/uploads/windguru.ts new file mode 100644 index 0000000..0a9c967 --- /dev/null +++ b/website/workers/uploads/windguru.ts @@ -0,0 +1,43 @@ +import type { ObservationInsert } from "../../database/schema.d"; +import { NonRetryableUploadError, UPLOAD_TIMEOUT_MS } from "./retry"; + +export async function uploadToWindGuru(env: Env, data: ObservationInsert) { + const WINDGURU_UID = await env.KV.get("WINDGURU_UID"); + const WINDGURU_PASSWORD = await env.KV.get("WINDGURU_PASSWORD"); + if (!WINDGURU_UID || !WINDGURU_PASSWORD) + throw new NonRetryableUploadError("Missing Windguru credentials"); + + const salt = Date.now(); + const hash = await crypto.subtle.digest( + { + name: "MD5", + }, + new TextEncoder().encode(`${salt}${WINDGURU_UID}${WINDGURU_PASSWORD}`), + ); + const params = new URLSearchParams({ + uid: WINDGURU_UID, + interval: "120", + wind_avg: (data.data.wind2MinAverage / 1.151).toString(), + wind_direction: data.data.windDirection.toString(), + temperature: data.data.temperatureC.toString(), + datetime: data.timestamp.toISOString(), + salt: salt.toString(), + hash: Array.from(new Uint8Array(hash)) + .map((b) => b.toString(16).padStart(2, "0")) + .join(""), + }); + const response = await fetch( + `https://www.windguru.cz/upload/api.php?${params.toString()}`, + { + method: "GET", + signal: AbortSignal.timeout(UPLOAD_TIMEOUT_MS), + }, + ).then((response) => { + if (!response.ok) + throw new Error("Windguru failed, status: " + response.status); + return response.text(); + }); + if (response !== "OK") { + throw new Error(`Windguru upload failed: ${response}`); + } +} diff --git a/website/workers/uploads/windy.ts b/website/workers/uploads/windy.ts new file mode 100644 index 0000000..8d68e93 --- /dev/null +++ b/website/workers/uploads/windy.ts @@ -0,0 +1,42 @@ +import type { ObservationInsert } from "../../database/schema.d"; +import { NonRetryableUploadError, UPLOAD_TIMEOUT_MS } from "./retry"; + +export async function uploadToWindy(env: Env, data: ObservationInsert) { + const WINDY_API_KEY = await env.KV.get("WINDY_API_KEY"); + const WINDY_STATION_ID = await env.KV.get("WINDY_STATION_ID"); + if (!WINDY_API_KEY || !WINDY_STATION_ID) + throw new NonRetryableUploadError("Missing Windy credentials"); + + const params = new URLSearchParams({ + station: WINDY_STATION_ID, // 32 bit integer; required for multiple stations; default value 0; alternative names: si, stationId + ts: data.timestamp.getTime().toString(), + temp: data.data.temperatureC.toString(), // real number [°C]; air temperature + windspeedmph: data.data.windSpeed.toString(), // real number [mph]; wind speed (alternative to wind) + winddir: data.data.windDirection.toString(), // integer number [deg]; instantaneous wind direction + //windgustmph: data.data.windGust.toString(), // real number [mph]; current wind gust (alternative to gust) + dewpoint: data.data.dewPoint.toString(), // real number [°C]; + precip: data.data.lastHourRain.toString(), // real number [mm]; precipitation over the past hour + }); + const response = await fetch( + `https://stations.windy.com/pws/update/${WINDY_API_KEY}?${params.toString()}`, + { + method: "GET", + signal: AbortSignal.timeout(UPLOAD_TIMEOUT_MS), + }, + ); + const responseText = await response.text(); + if ( + responseText.length > 0 && + responseText.includes( + "Measurement sent too soon, update interval is 5 minutes", + ) + ) { + throw new NonRetryableUploadError( + "Windy asked for a backoff, skip this observation", + ); + } else if (!response.ok) + throw new Error("Windy upload failed, status: " + response.status); + const responseJson = JSON.parse(responseText); + if (!responseJson || typeof responseJson !== "object") + throw new Error(`Windy upload failed: ${responseText}`); +} diff --git a/website/wrangler.jsonc b/website/wrangler.jsonc index 4930bfc..27ac56d 100644 --- a/website/wrangler.jsonc +++ b/website/wrangler.jsonc @@ -22,21 +22,6 @@ "binding": "WORKFLOW_OVERNIGHT_SAVE_TO_R2", "class_name": "OvernightSaveToR2", }, - { - "name": "PSCWeather-UploadReceivedObservationToWindGuru", - "binding": "WORKFLOW_UPLOAD_TO_WINDGURU", - "class_name": "UploadToWindGuru", - }, - { - "name": "PSCWeather-UploadReceivedObservationToMetOffice", - "binding": "WORKFLOW_UPLOAD_TO_METOFFICE", - "class_name": "UploadToMetOffice", - }, - { - "name": "PSCWeather-UploadReceivedObservationToWindy", - "binding": "WORKFLOW_UPLOAD_TO_WINDY", - "class_name": "UploadToWindy", - }, ], "r2_buckets": [ {