feat: redundant storage

README.md

@@ -15,6 +15,25 @@ There are a few steps to the setup but it should hopefully be pretty straightfor
   4b. (optional) If you want publishing, you'll also need to add a Discord bot token with `wrangler secret put DISCORD_TOKEN`
 5. Run `npm run publish` :)
 
+## Data redundancy
+
+This worker is using KV as its main store, but if that was to suffer a problem, you can additionally
+save data to R2 or D1.
+
+```toml
+[[d1_databases]]
+binding = "D1"
+database_name = ""
+database_id = ""
+
+[[r2_buckets]]
+binding = "R2"
+bucket_name = ""
+```
+
+To setup D1, you need to run the following SQL statement:
+`CREATE TABLE IF NOT EXISTS KV (key TEXT UNIQUE, value TEXT)`
+
 ## Example
 ### New Incident
 ![New Incident](https://user-images.githubusercontent.com/8492901/131903623-352dd6ec-bd7f-470f-9468-4a271c4ddc69.png)

src/index.ts

@@ -1,6 +1,7 @@
 import { publishMessage, sendToDiscord } from './discord';
 import { attachDebug } from './utils';
 import Config from './config';
+import { retrieveFromStorage, saveToStorage } from './storage';
 
 export default {
   async fetch(req: Request, env: Env) {
@@ -35,19 +36,19 @@ export default {
     const json = await res.json<IncidentResponse>();
 
     await Promise.all(json.incidents.map(async incident => {
-      const kv = await env.KV.get<Incident>(incident.id, 'json');
+      const stored = await retrieveFromStorage<Incident>(env, incident.id);
 
-      console.log('-----\nIncident ' + incident.id + ' in KV: ' + (kv !== null) + '\n-----');
+      console.log('-----\nIncident ' + incident.id + ' in storage: ' + (stored !== null) + '\n-----');
 
       if (globalThis.DEBUG?.updateIncident === incident.id) {
         // Set update to now so we force an update
         incident.updated_at = new Date().toISOString();
       }
 
-      if (kv === null) {
+      if (stored === null) {
         await this.postNew(incident, env);
       } else {
-        await this.postUpdate(incident, kv, env);
+        await this.postUpdate(incident, stored, env);
       }
     }));
 
@@ -62,8 +63,8 @@ export default {
     if (messageId !== null) {
       incident.messageId = messageId;
     }
-    // Update KV
-    await env.KV.put(incident.id, JSON.stringify(incident));
+    // Update storage
+    await saveToStorage(env, incident.id, incident);
 
     // Check if we can publish
     if (messageId !== null && Config.PUBLISH_CHANNEL_ID !== '') {
@@ -89,8 +90,8 @@ export default {
     if (incident.updated_at !== cachedIncident.updated_at) {
       console.log('Updating incident:', incident.id);
 
-      // Update KV
-      await env.KV.put(incident.id, JSON.stringify(incident));
+      // Update storage 
+      await saveToStorage(env, incident.id, incident);
       // Update Discord
       await sendToDiscord(incident, env);
     }

src/storage.ts

@@ -0,0 +1,50 @@
+export async function saveToStorage(env: Env, key: string, value: any) {
+  const now = new Date().getTime();
+  const data = JSON.stringify({ ts: now, value });
+
+  await Promise.allSettled([
+    env.KV && env.KV.put(key, data),
+    env.D1 &&
+      env.D1.prepare(
+        `INSERT INTO KV(key, value) VALUES(?1, ?2) ON CONFLICT(key) DO UPDATE SET value = ?2`,
+      )
+        .bind(key, data)
+        .run(),
+    env.R2 && env.R2.put(key, data),
+  ]);
+}
+
+export async function retrieveFromStorage<T>(
+  env: Env,
+  key: string,
+): Promise<T | null> {
+  const rawValues = await Promise.allSettled([
+    env.KV && env.KV.get(key),
+    env.D1 &&
+      env.D1.prepare(`SELECT value FROM KV WHERE key = ?`).bind(key).first("value"),
+    env.R2 && env.R2.get(key).then((object) => object && object.text()),
+  ]);
+  const values = rawValues.filter(
+    (x) => x.status === "fulfilled" && x.value,
+  ) as Array<PromiseFulfilledResult<string>>;
+  // none of our storage providers had the value in stock, so either all of them are
+  // broken, or its just a new key
+  if (!values.length) return null;
+
+  const parsed: Array<{ ts: number; value: T }> = [];
+  for (const value of values) {
+    // lets make sure we handle data corruption cases
+    try {
+      parsed.push(JSON.parse(value.value));
+    } catch {}
+  }
+  if (!parsed.length) {
+    console.error("all data stores hold corrupted data, somehow...", values);
+    return null;
+  }
+
+  // to account for cases where a write failed, we have the timestamp!
+  // so we use whatever has the newest timestamp
+  const freshness = parsed.sort((a, b) => b.ts - a.ts);
+  return freshness[0].value;
+}

src/types.d.ts

@@ -6,6 +6,8 @@ declare module globalThis {
 
 interface Env {
   KV: KVNamespace;
+  D1?: D1Database;
+  R2?: R2Bucket;
   DISCORD_WEBHOOK: string;
   DISCORD_TOKEN: string;
 }
@@ -87,4 +89,4 @@ interface DiscordResponse {
 interface Debug {
   // Allows me to trigger an update for a specific incident
   updateIncident?: string;
-}
+}