Release to sus

Dockerfile

@@ -14,12 +14,13 @@ COPY stuff/Inconsolata-Bold.otf /usr/share/fonts/truetype/inconsolata
 
 RUN apt update && apt install -y file procps figlet fortune cowsay pslist inkscape imagemagick --no-install-recommends && rm -rf /var/lib/apt/lists/*
 
-COPY *.deno.ts ind*x.html tgbot.deno.ts ./
-RUN deno cache server.deno.ts
+COPY ind*x.html ./
+COPY src src
+RUN deno cache src/server.deno.ts
 COPY static static
 COPY --from=blog-builder /srv/jekyll/build/ ./static/blog
 
 ENV PATH "$PATH:/usr/games"
 COPY ./static/amogus.cow /usr/share/cowsay/cows
 
-CMD ["sh", "-c", "deno run --unstable-cron --allow-all server.deno.ts 2>&1 | sed -u -e \"s/$TG_BOT_TOKEN/<REDACTED>/g\" >> static/persistent/log.txt"]
+CMD ["sh", "-c", "deno run --unstable-cron --allow-all src/server.deno.ts 2>&1 | sed -u -e \"s/$TG_BOT_TOKEN/<REDACTED>/g\" >> static/persistent/log.txt"]

src/replication.deno.ts

@@ -0,0 +1,205 @@
+// deno-lint-ignore-file no-explicit-any
+// The authors disclaim copyright to this source code (they are ashamed to
+// admit they wrote it)
+
+import { handleTgUpdate } from "./tg/bot.deno.ts";
+import { DOMAIN, genRandomToken, tgCall } from "./tg/utils.deno.ts";
+import { RequestEvent } from "./utils.deno.ts";
+
+const currentPeers = new Map<string, { myToken: string; theirToken: string }>(
+  (Deno.env.get("REPLICATION_ENDPOINTS")?.split(",") || []).map((peer) => [
+    peer.trim(),
+    { myToken: genRandomToken(16), theirToken: genRandomToken(16) },
+  ]),
+);
+
+function listPeers() {
+  console.log("Current replication peers:");
+  for (const [endpoint, tokens] of currentPeers.entries()) {
+    console.log(
+      `- ${endpoint}: myToken=${
+        tokens.myToken.slice(
+          0,
+          4,
+        )
+      }..., theirToken=${tokens.theirToken.slice(0, 4)}...`,
+    );
+  }
+}
+
+export async function handleReplication(
+  e: RequestEvent,
+): Promise<Response | null> {
+  const url = new URL(e.request.url);
+
+  if (url.pathname === "/replication/event") {
+    if (e.request.method.toUpperCase() !== "POST") {
+      return new Response("Method Not Allowed", { status: 405 });
+    }
+    const body = await e.request.json();
+    const { data, depth, token, src } = body;
+    if (!token || currentPeers.get(src)?.theirToken !== token) {
+      console.log(
+        `Unauthorized replication attempt from ${src} (${
+          token.slice(0, 4)
+        }...)`,
+      );
+      return new Response("Unauthorized", { status: 401 });
+    }
+    await Promise.all([
+      e.respondWith(
+        new Response("processing", {
+          status: 200,
+          headers: {
+            "Content-Type": "text/plain",
+          },
+        }),
+      ),
+      processTgUpdate(data, depth),
+    ]);
+    return null;
+  }
+
+  if (url.pathname === "/replication/register") {
+    const data = await e.request.json();
+    const { myToken, yourToken, src, confirmOnly } = data;
+    console.log(
+      `Received replication registration${
+        confirmOnly ? " (confirm-only)" : ""
+      } request from ${data.src}`,
+    );
+    if (
+      typeof myToken !== "string" ||
+      typeof yourToken !== "string" ||
+      typeof src !== "string" ||
+      !myToken ||
+      !yourToken ||
+      !src
+    ) {
+      return new Response("Bad Request", { status: 400 });
+    }
+    const peerData = currentPeers.get(src);
+    if (!peerData) {
+      return new Response("You are not my friend", { status: 403 });
+    }
+    if (peerData.myToken === yourToken && peerData.theirToken === myToken) {
+      return new Response("ok", { status: 200 });
+    }
+
+    if (confirmOnly) {
+      return new Response("I did not ask for this", { status: 409 });
+    }
+
+    const resp = await fetch(`https://${src}/replication/register`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        myToken: yourToken,
+        yourToken: myToken,
+        src: DOMAIN,
+        confirmOnly: true,
+      }),
+    });
+    if (!resp.ok) {
+      console.log(
+        `Peer ${src} requested registration, but didn't confirm:`,
+        await resp.text(),
+      );
+      return new Response("You seem kinda sus", { status: 403 });
+    }
+
+    peerData.myToken = yourToken;
+    peerData.theirToken = myToken;
+    console.log(`Peer ${src} has updated their registration.`);
+    listPeers();
+    return new Response("oh hi", { status: 200 });
+  }
+
+  return new Response("Not Found", { status: 404 });
+}
+
+export async function replicateData(data: any, depth: number) {
+  for (const [endpoint, peerData] of currentPeers.entries()) {
+    if (!peerData.myToken) continue;
+    try {
+      const resp = await fetch(`https://${endpoint}/replication/event`, {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({
+          data,
+          depth,
+          token: peerData.myToken,
+          src: DOMAIN,
+        }),
+      });
+      if (resp.status === 401) {
+        peerData.myToken = genRandomToken(16);
+        peerData.theirToken = endpoint === DOMAIN
+          ? peerData.myToken
+          : genRandomToken(16);
+        console.log(
+          `Peer ${endpoint} rejected replication (401). Re-registering...`,
+        );
+        const regResp = await fetch(
+          `https://${endpoint}/replication/register`,
+          {
+            method: "POST",
+            headers: {
+              "Content-Type": "application/json",
+            },
+            body: JSON.stringify({
+              myToken: peerData.myToken,
+              yourToken: peerData.theirToken,
+              src: DOMAIN,
+            }),
+          },
+        );
+        if (!regResp.ok) {
+          throw new Error(
+            `Re-registration to ${endpoint} failed: ${await regResp.text()}`,
+          );
+        }
+        console.log(`Re-registration to ${endpoint} successful.`);
+        listPeers();
+        // Retry replication once after re-registering
+        const retryResp = await fetch(`https://${endpoint}/replication/event`, {
+          method: "POST",
+          headers: {
+            "Content-Type": "application/json",
+          },
+          body: JSON.stringify({
+            data,
+            depth,
+            token: peerData.myToken,
+            src: DOMAIN,
+          }),
+        });
+        if (!retryResp.ok) {
+          console.error(
+            `Replication to ${endpoint} after re-registration failed:`,
+            await retryResp.text(),
+          );
+        }
+      } else if (!resp.ok) {
+        console.error(`Replication to ${endpoint} failed:`, await resp.text());
+      }
+    } catch (e) {
+      console.error(`Replication to ${endpoint} failed:`, e);
+    }
+  }
+}
+
+export async function processTgUpdate(data: any, depth: number) {
+  for await (const dato of handleTgUpdate(data)) {
+    if (depth >= 5) {
+      tgCall({ text: "🔥" });
+      continue;
+    }
+
+    await replicateData(dato, depth + 1);
+  }
+}

server.deno.ts → src/server.deno.ts

@@ -3,18 +3,19 @@
 
 import { serveDir } from "https://deno.land/std@0.190.0/http/file_server.ts";
 import {
-  handleRequest as handleTgRequest,
-  handleTgWeb,
   get_video,
   getSticekr,
   getSticekrCount,
-  RequestEvent,
-  init as tgBotInit,
-  webhookPath as tgWebhookPath,
-} from "./tgbot.deno.ts";
+  handleTgWeb,
+} from "./tg/bot.deno.ts";
+import { init as tgBotInit } from "./tg/init.deno.ts";
+import { webhookPath as tgWebhookPath } from "./tg/utils.deno.ts";
+import { handleReplication, processTgUpdate } from "./replication.deno.ts";
+import { RequestEvent } from "./utils.deno.ts";
+import { webhookUrlToken } from "./tg/utils.deno.ts";
 
 const indexContent = new TextDecoder().decode(
-  await Deno.readFile("index.html")
+  await Deno.readFile("index.html"),
 );
 const indxContent = new TextDecoder().decode(await Deno.readFile("indx.html"));
 
@@ -32,9 +33,11 @@ async function handleHttp(request: Request): Promise<Response> {
       const resp = await r;
       const end = performance.now();
       console.log(
-        `${new Date().toISOString()} ${resp.status} ${request.method} ${
-          request.url
-        } ${(end - start).toFixed(1)}ms`
+        `${
+          new Date().toISOString()
+        } ${resp.status} ${request.method} ${request.url} ${
+          (end - start).toFixed(1)
+        }ms`,
       );
       resolve(resp);
     },
@@ -46,31 +49,65 @@ async function handleHttp(request: Request): Promise<Response> {
         await mockEvent.respondWith(response);
       }
     })
-    .catch((err) => console.error(err));
+    .catch((err) => {
+      console.error(err);
+      mockEvent.respondWith(
+        new Response("Internal Server Error", { status: 500 }),
+      );
+    });
 
   return await responsePromise;
 }
 
 async function handleEvent(e: RequestEvent): Promise<Response | null> {
   const url = new URL(e.request.url);
   if (url.pathname === tgWebhookPath) {
-    await handleTgRequest(e);
+    if (
+      e.request.method.toUpperCase() !== "POST" ||
+      e.request.headers.get("X-Telegram-Bot-Api-Secret-Token") !==
+        webhookUrlToken
+    ) {
+      return new Response("You shall not pass", {
+        status: 401,
+        headers: {
+          "Content-Type": "text/plain",
+        },
+      });
+    }
+
+    const data = await e.request.json();
+
+    await Promise.all([
+      e.respondWith(
+        new Response("processing", {
+          status: 200,
+          headers: {
+            "Content-Type": "text/plain",
+          },
+        }),
+      ),
+      processTgUpdate(data, 0),
+    ]);
     return null;
   }
 
   if (url.pathname === "/" || url.pathname === "/index.html") {
     return Math.random() < 0.01
       ? new Response(indxContent, {
-          headers: {
-            "content-type": "text/html; charset=utf-8",
-          },
-          status: 418,
-        })
+        headers: {
+          "content-type": "text/html; charset=utf-8",
+        },
+        status: 418,
+      })
       : new Response(indexContent, {
-          headers: {
-            "content-type": "text/html; charset=utf-8",
-          },
-        });
+        headers: {
+          "content-type": "text/html; charset=utf-8",
+        },
+      });
+  }
+
+  if (url.pathname.startsWith("/replication/")) {
+    return await handleReplication(e);
   }
 
   if (url.pathname === "/about") {
@@ -92,7 +129,7 @@ async function handleEvent(e: RequestEvent): Promise<Response | null> {
   }
 
   if (url.pathname.startsWith("/api/videos/")) {
-    const index = parseInt(url.pathname.split('/').pop()!);
+    const index = parseInt(url.pathname.split("/").pop()!);
     const videoBytes = get_video(index);
 
     if (videoBytes === null) {
@@ -105,7 +142,7 @@ async function handleEvent(e: RequestEvent): Promise<Response | null> {
     return new Response(videoBytes.buffer as ArrayBuffer, {
       headers: {
         "Content-Type": "video/mp4",
-        "Access-Control-Allow-Origin": "*"
+        "Access-Control-Allow-Origin": "*",
       },
     });
   }
@@ -115,9 +152,9 @@ async function handleEvent(e: RequestEvent): Promise<Response | null> {
       headers: { "Content-Type": "application/json" },
     });
   }
-  
+
   if (url.pathname.startsWith("/api/stickers/")) {
-    const index = parseInt(url.pathname.split('/').pop()!);
+    const index = parseInt(url.pathname.split("/").pop()!);
     const sticekrBytes = await getSticekr(index);
 
     if (sticekrBytes === null) {
@@ -129,7 +166,7 @@ async function handleEvent(e: RequestEvent): Promise<Response | null> {
 
     return new Response(sticekrBytes.buffer as ArrayBuffer, {
       headers: {
-        "Content-Type": "image/webp"
+        "Content-Type": "image/webp",
       },
     });
   }
@@ -158,4 +195,4 @@ async function handleEvent(e: RequestEvent): Promise<Response | null> {
 
 await tgBotInit();
 
-Deno.serve({ port: 8000 }, handleHttp);
+Deno.serve({ port: parseInt(Deno.env.get("PORT") || "8000") }, handleHttp);