Git-Commit Weather Station: Node.js + RedisTimeSeries + IPFS + WebRTC (Docker)¶

// server.js
import Fastify from "fastify";
import { spawn } from "node:child_process";
import Redis from "ioredis";
import { fetch } from "undici";
import { v4 as uuidv4 } from "uuid";
import WebSocket from "ws";

const app = Fastify({ logger: true });
const redis = new Redis(process.env.REDIS_URL);

// WebRTC signaling endpoints
const peers = new Map(); // id -> { offer/answer, iceCandidates, send(json) }

// --- 1) Git -> "Weather" metrics (commit velocity, merge rate, rebase storms)
function gitStats() {
  return new Promise((resolve, reject) => {
    const since = "--since=5 minutes ago";
    const log = spawn("git", ["-C", "/work/repo", "log", since, "--pretty=%H %P %s"]);
    let out = "";
    let err = "";

    log.stdout.on("data", d => out += d.toString());
    log.stderr.on("data", d => err += d.toString());

    log.on("close", (code) => {
      if (code !== 0) {
        console.warn("Git command failed:", err);
        resolve({ commits: 0, merges: 0, rebases: 0, ts: Date.now() });
        return;
      }

      const lines = out.trim().split("\n").filter(Boolean);
      const commits = lines.length;
      const merges = lines.filter(l => l.split(" ").slice(1, -1).length >= 2).length;
      const rebases = lines.filter(l => /rebase/i.test(l)).length;

      resolve({ commits, merges, rebases, ts: Date.now() });
    });
  });
}

// --- 2) Ingest to RedisTimeSeries
async function recordStats(s) {
  try {
    // Create time series if they don't exist
    await redis.call("TS.CREATE", "git:commits", "RETENTION", "86400000", "LABELS", "metric", "commits").catch(() => {});
    await redis.call("TS.CREATE", "git:merges", "RETENTION", "86400000", "LABELS", "metric", "merges").catch(() => {});
    await redis.call("TS.CREATE", "git:rebases", "RETENTION", "86400000", "LABELS", "metric", "rebases").catch(() => {});

    // Add data points
    await redis.call("TS.ADD", "git:commits", s.ts, s.commits);
    await redis.call("TS.ADD", "git:merges", s.ts, s.merges);
    await redis.call("TS.ADD", "git:rebases", s.ts, s.rebases);

    app.log.info({ commits: s.commits, merges: s.merges, rebases: s.rebases }, "Recorded weather stats");
  } catch (error) {
    app.log.error({ error }, "Failed to record stats");
  }
}

// --- 3) Snapshot to IPFS (periodic content-addressed history)
async function snapshotToIPFS(seriesJson) {
  try {
    const boundary = uuidv4();
    const body = `--${boundary}\r\nContent-Disposition: form-data; name="file"; filename="snapshot.json"\r\nContent-Type: application/json\r\n\r\n${JSON.stringify(seriesJson)}\r\n--${boundary}--\r\n`;

    const response = await fetch(`${process.env.IPFS_API}/add`, {
      method: "POST",
      headers: { "Content-Type": `multipart/form-data; boundary=${boundary}` },
      body
    });

    const text = await response.text();
    const cid = text.match(/"Hash":"([^"]+)"/)?.[1] || "unknown";
    return cid;
  } catch (error) {
    app.log.error({ error }, "Failed to snapshot to IPFS");
    return null;
  }
}

// --- 4) WebRTC signaling endpoints
app.post("/signal/offer", async (req, reply) => {
  const { id, sdp } = req.body;
  peers.set(id, { sdpOffer: sdp, queue: [] });
  app.log.info({ id }, "Received WebRTC offer");
  return { ok: true };
});

app.post("/signal/answer", async (req, reply) => {
  const { id, sdp } = req.body;
  const p = peers.get(id);
  if (!p) return reply.code(404).send({ error: "Peer not found" });

  p.sdpAnswer = sdp;
  app.log.info({ id }, "Received WebRTC answer");
  return { ok: true };
});

app.get("/signal/status/:id", async (req, reply) => {
  const p = peers.get(req.params.id);
  return p ? { haveAnswer: !!p.sdpAnswer } : reply.code(404).send({ error: "Peer not found" });
});

// --- 5) REST API for recent data (fallback)
app.get("/api/recent", async (req, reply) => {
  try {
    const now = Date.now();
    const from = now - 5 * 60 * 1000; // Last 5 minutes

    const [commits, merges, rebases] = await Promise.all([
      redis.call("TS.RANGE", "git:commits", from, now),
      redis.call("TS.RANGE", "git:merges", from, now),
      redis.call("TS.RANGE", "git:rebases", from, now),
    ]);

    return { commits, merges, rebases, timestamp: now };
  } catch (error) {
    app.log.error({ error }, "Failed to get recent data");
    return reply.code(500).send({ error: "Failed to get recent data" });
  }
});

// --- 6) WebSocket for real-time updates
const wss = new WebSocket.Server({ port: 3001 });
const clients = new Set();

wss.on("connection", (ws) => {
  clients.add(ws);
  app.log.info("WebSocket client connected");

  ws.on("close", () => {
    clients.delete(ws);
    app.log.info("WebSocket client disconnected");
  });
});

// Broadcast function
function broadcast(data) {
  const message = JSON.stringify(data);
  clients.forEach(client => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(message);
    }
  });
}

// Start server
app.listen({ port: 3000, host: "0.0.0.0" }, (err, address) => {
  if (err) throw err;
  app.log.info(`Weather station listening at ${address}`);
});

// --- 7) Main loop: poll git, write RedisTS, periodic IPFS snapshot, broadcast
setInterval(async () => {
  try {
    const stats = await gitStats();
    await recordStats(stats);

    // Periodic IPFS snapshot (20% chance every 5 minutes)
    if (Math.random() < 0.2) {
      const now = Date.now();
      const from = now - 15 * 60 * 1000; // 15 minutes window

      const window = {
        commits: await redis.call("TS.RANGE", "git:commits", from, now),
        merges: await redis.call("TS.RANGE", "git:merges", from, now),
        rebases: await redis.call("TS.RANGE", "git:rebases", from, now),
        timestamp: now
      };

      const cid = await snapshotToIPFS(window);
      if (cid) {
        app.log.info({ cid }, "Pinned snapshot to IPFS");
      }
    }

    // Broadcast live update
    broadcast({ type: "tick", payload: stats });

  } catch (error) {
    app.log.error({ error }, "Error in main loop");
  }
}, 5000); // Poll every 5 seconds

// Essential Git weather operations
// 1. Poll Git for stats
const stats = await gitStats();

// 2. Store in RedisTimeSeries
await redis.call("TS.ADD", "git:commits", stats.ts, stats.commits);

// 3. Snapshot to IPFS
const cid = await snapshotToIPFS(window);

// 4. Broadcast to clients
broadcast({ type: "tick", payload: stats });

// 5. Visualize in browser
particleSystem.storm(stats);