Migrate payload storage to SQLite

compose.yaml

@@ -18,6 +18,7 @@ services:
       API_BEARER_TOKEN: ${API_BEARER_TOKEN}
       PUBLIC_DOMAIN: ${PUBLIC_DOMAIN:-http://localhost}
       HOST_PORT: ${HOST_PORT:-8152}
+      DB_FILE: /data/payloads.sqlite
       DATA_FILE: /data/payloads.json
     volumes:
       - server-data:/data

server/README.md

@@ -23,10 +23,17 @@ Læsning er offentlig:
 ```text
 GET /api/health
 GET /api/activity
-GET /api/payloads
+GET /api/payloads?limit=100&offset=0
+GET /api/payloads/count
 GET /api/payloads/:id
 ```
 
+Payloads gemmes i SQLite med WAL-mode. Liste-endpointet returnerer de nyeste
+100 records som standard, understøtter `limit` (maks. 1000) og `offset`, og
+sender det samlede antal i `X-Total-Count`-headeren. En eksisterende
+`/data/payloads.json` migreres automatisk, når SQLite-databasen oprettes, og
+fjernes efter en vellykket migrering.
+
 Skrivning kræver `Authorization: Bearer <API_BEARER_TOKEN>`:
 
 ```text

server/public/app.js

@@ -126,18 +126,18 @@ async function refresh() {
     const dashboardRequest = { headers: { "X-Monitor-Request": "dashboard" } };
     const [healthResponse, payloadResponse, activityResponse] = await Promise.all([
       fetch("/api/health", dashboardRequest),
-      fetch("/api/payloads", dashboardRequest),
+      fetch("/api/payloads/count", dashboardRequest),
       fetch("/api/activity", dashboardRequest),
     ]);
     if (![healthResponse, payloadResponse, activityResponse].every((response) => response.ok)) {
       throw new Error("Serveren returnerede en fejl");
     }
-    const [health, payloads, activity] = await Promise.all([
+    const [health, payloadSummary, activity] = await Promise.all([
       healthResponse.json(), payloadResponse.json(), activityResponse.json(),
     ]);
     elements.health.textContent = `API online · ${new Date(health.timestamp).toLocaleTimeString("en-GB")}`;
     elements.health.className = "status online";
-    elements.payloadCount.textContent = payloads.length;
+    elements.payloadCount.textContent = payloadSummary.count;
     elements.apiCount.textContent = activity.filter((event) => event.source === "api").length;
     elements.serverCount.textContent = activity.filter((event) => event.source === "server").length;
     requests = activity.filter((event) => event.source === "api");

server/src/server.js

@@ -1,14 +1,48 @@
 import crypto from "node:crypto";
 import fs from "node:fs/promises";
 import path from "node:path";
+import { DatabaseSync } from "node:sqlite";
 import express from "express";
 
 const port = 3000;
-const dataFile = process.env.DATA_FILE ?? "/data/payloads.json";
+const databaseFile = process.env.DB_FILE ?? "/data/payloads.sqlite";
+const legacyDataFile = process.env.DATA_FILE ?? "/data/payloads.json";
 const bearerToken = process.env.API_BEARER_TOKEN ?? "";
 const publicDir = path.resolve("public");
 const activity = [];
-let payloads = [];
+
+await fs.mkdir(path.dirname(databaseFile), { recursive: true });
+const databaseAlreadyExists = await fs.access(databaseFile).then(() => true, () => false);
+const database = new DatabaseSync(databaseFile);
+database.exec(`
+  PRAGMA journal_mode = WAL;
+  PRAGMA synchronous = NORMAL;
+  CREATE TABLE IF NOT EXISTS payloads (
+    id TEXT PRIMARY KEY,
+    topic TEXT NOT NULL,
+    payload_json TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL
+  );
+  CREATE INDEX IF NOT EXISTS payloads_created_at_idx ON payloads(created_at DESC);
+  CREATE INDEX IF NOT EXISTS payloads_topic_idx ON payloads(topic);
+`);
+
+const statements = {
+  count: database.prepare("SELECT COUNT(*) AS count FROM payloads"),
+  get: database.prepare("SELECT * FROM payloads WHERE id = ?"),
+  list: database.prepare("SELECT * FROM payloads ORDER BY created_at DESC LIMIT ? OFFSET ?"),
+  insert: database.prepare(`
+    INSERT INTO payloads (id, topic, payload_json, created_at, updated_at)
+    VALUES (?, ?, ?, ?, ?)
+  `),
+  update: database.prepare(`
+    UPDATE payloads
+    SET topic = ?, payload_json = ?, updated_at = ?
+    WHERE id = ?
+  `),
+  delete: database.prepare("DELETE FROM payloads WHERE id = ?"),
+};
 
 function record(source, action, detail = {}) {
   activity.unshift({
@@ -52,20 +86,49 @@ function isDashboardRequest(req) {
   }
 }
 
-async function loadPayloads() {
+function deserializePayload(row) {
+  if (!row) return null;
+  return {
+    id: row.id,
+    topic: row.topic,
+    payload: JSON.parse(row.payload_json),
+    createdAt: row.created_at,
+    updatedAt: row.updated_at,
+  };
+}
+
+async function migrateLegacyPayloads() {
+  if (databaseAlreadyExists) return;
   try {
-    payloads = JSON.parse(await fs.readFile(dataFile, "utf8"));
+    const payloads = JSON.parse(await fs.readFile(legacyDataFile, "utf8"));
+    if (Array.isArray(payloads) && payloads.length > 0) {
+      database.exec("BEGIN");
+      try {
+        for (const payload of payloads) {
+          statements.insert.run(
+            payload.id,
+            payload.topic,
+            JSON.stringify(payload.payload),
+            payload.createdAt,
+            payload.updatedAt,
+          );
+        }
+        database.exec("COMMIT");
+      } catch (error) {
+        database.exec("ROLLBACK");
+        throw error;
+      }
+    }
+    await fs.unlink(legacyDataFile);
   } catch (error) {
     if (error.code !== "ENOENT") throw error;
-    await persistPayloads();
   }
 }
 
-async function persistPayloads() {
+function parsePageInteger(value, fallback, maximum) {
-  await fs.mkdir(path.dirname(dataFile), { recursive: true });
+  const parsed = Number.parseInt(value, 10);
-  const temporaryFile = `${dataFile}.tmp`;
+  if (!Number.isFinite(parsed) || parsed < 0) return fallback;
-  await fs.writeFile(temporaryFile, JSON.stringify(payloads, null, 2));
+  return Math.min(parsed, maximum);
-  await fs.rename(temporaryFile, dataFile);
 }
 
 function requireBearer(req, res, next) {
@@ -108,15 +171,26 @@ app.get("/api/health", (_req, res) => {
   res.json({ status: "ok", timestamp: new Date().toISOString() });
 });
 
-app.get("/api/payloads", (_req, res) => res.json(payloads));
+app.get("/api/payloads", (req, res) => {
+  const limit = parsePageInteger(req.query.limit, 100, 1000);
+  const offset = parsePageInteger(req.query.offset, 0, Number.MAX_SAFE_INTEGER);
+  const total = statements.count.get().count;
+  const payloads = statements.list.all(limit, offset).map(deserializePayload);
+  res.set("X-Total-Count", String(total));
+  res.json(payloads);
+});
+
+app.get("/api/payloads/count", (_req, res) => {
+  res.json({ count: statements.count.get().count });
+});
 
 app.get("/api/payloads/:id", (req, res) => {
-  const payload = payloads.find((item) => item.id === req.params.id);
+  const payload = deserializePayload(statements.get.get(req.params.id));
   if (!payload) return res.status(404).json({ error: "Payload not found" });
   res.json(payload);
 });
 
-app.post("/api/payloads", requireBearer, async (req, res, next) => {
+app.post("/api/payloads", requireBearer, (req, res, next) => {
   try {
     const now = new Date().toISOString();
     const payload = {
@@ -126,8 +200,7 @@ app.post("/api/payloads", requireBearer, async (req, res, next) => {
       createdAt: now,
       updatedAt: now,
     };
-    payloads.push(payload);
+    statements.insert.run(payload.id, payload.topic, JSON.stringify(payload.payload), now, now);
-    await persistPayloads();
     res.locals.activityDetail = { payload };
     res.status(201).json(payload);
   } catch (error) {
@@ -135,30 +208,29 @@ app.post("/api/payloads", requireBearer, async (req, res, next) => {
   }
 });
 
-app.patch("/api/payloads/:id", requireBearer, async (req, res, next) => {
+app.patch("/api/payloads/:id", requireBearer, (req, res, next) => {
   try {
-    const index = payloads.findIndex((item) => item.id === req.params.id);
+    const existing = deserializePayload(statements.get.get(req.params.id));
-    if (index === -1) return res.status(404).json({ error: "Payload not found" });
+    if (!existing) return res.status(404).json({ error: "Payload not found" });
-    payloads[index] = {
+    const payload = {
-      ...payloads[index],
+      ...existing,
       ...(typeof req.body.topic === "string" ? { topic: req.body.topic } : {}),
       ...(Object.hasOwn(req.body, "payload") ? { payload: req.body.payload } : {}),
       updatedAt: new Date().toISOString(),
     };
-    await persistPayloads();
+    statements.update.run(payload.topic, JSON.stringify(payload.payload), payload.updatedAt, payload.id);
-    res.locals.activityDetail = { payload: payloads[index] };
+    res.locals.activityDetail = { payload };
-    res.json(payloads[index]);
+    res.json(payload);
   } catch (error) {
     next(error);
   }
 });
 
-app.delete("/api/payloads/:id", requireBearer, async (req, res, next) => {
+app.delete("/api/payloads/:id", requireBearer, (req, res, next) => {
   try {
-    const index = payloads.findIndex((item) => item.id === req.params.id);
+    const removed = deserializePayload(statements.get.get(req.params.id));
-    if (index === -1) return res.status(404).json({ error: "Payload not found" });
+    if (!removed) return res.status(404).json({ error: "Payload not found" });
-    const [removed] = payloads.splice(index, 1);
+    statements.delete.run(req.params.id);
-    await persistPayloads();
     res.locals.activityDetail = { payload: removed };
     res.status(204).end();
   } catch (error) {
@@ -177,5 +249,5 @@ app.use((error, req, res, _next) => {
   res.status(status).json({ error: error.message });
 });
 
-await loadPayloads();
+await migrateLegacyPayloads();
 app.listen(port, () => record("server", "started", { port }));