Replace the three upload Workflows with inline retries and throttling

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",

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
+);

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": {}
+  }
+}

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
     }
   ]
 }

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";

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<UploadTarget>().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];