update service and scripts

metrics-collector/src/gh-contributions-to-csv.js

@@ -20,12 +20,11 @@ function extractData(data) {
 
     // Collect all months from all projects
     projects.forEach(project => {
-        data[project].forEach(entry => {
-            const { data } = entry;
-            Object.values(data).forEach(item => {
-                Object.keys(item).forEach(month => {
-                    allMonths.add(month);
-                });
+        const lastEntry = data[project][data[project].length - 1];
+        const { data: lastEntryData } = lastEntry;
+        Object.values(lastEntryData).forEach(item => {
+            Object.keys(item).forEach(month => {
+                allMonths.add(month);
             });
         });
     });

metrics-collector/src/gh-metrics.ts

@@ -7,15 +7,18 @@ import { readJsonFile, writeJsonFile } from "./utils";
 
 const orgName = "TBD54566975";
 const repos = [
+  "tbdex",
   "tbdex-js",
   "tbdex-kt",
   "tbdex-swift",
   "tbdex-rs",
+  "web5-spec",
   "web5-js",
   "web5-kt",
   "web5-swift",
   "web5-rs",
   "dwn-sdk-js",
+  "dwn-server",
 ];
 
 const KNOWN_PAST_MEMBERS = ["amika-sq"];

metrics-collector/src/index.ts

@@ -5,12 +5,14 @@ import yargs from "yargs";
 import { hideBin } from "yargs/helpers";
 
 import { collectGhMetrics } from "./gh-metrics";
-import { collectNpmMetrics, saveNpmMetrics } from "./npm-metrics";
+import { collectNpmMetrics } from "./npm-metrics";
 import {
   collectSonatypeMetrics,
   saveSonatypeMetrics,
 } from "./sonatype-metrics";
-import { getYesterdayDate } from "./utils";
+import { getYesterdayDate, readJsonFile } from "./utils";
+import { readFile, writeFile } from "fs/promises";
+import { existsSync } from "fs";
 
 const isLocalPersistence = process.env.PERSIST_LOCAL_FILES === "true";
 
@@ -71,7 +73,12 @@ async function main() {
   if (collectNpm) {
     console.info(`\n\n============\n\n>>> Collecting metrics for NPM...`);
     if (initialLoadFromDate) {
-      await initialLoad(initialLoadFromDate, metricDate, collectNpmMetrics);
+      await initialLoad(
+        "npm-metrics",
+        initialLoadFromDate,
+        metricDate,
+        collectNpmMetrics
+      );
     } else {
       await collectNpmMetrics(metricDateStr);
     }
@@ -84,6 +91,7 @@ async function main() {
     );
     if (initialLoadFromDate) {
       await initialLoad(
+        "sonatype-metrics",
         initialLoadFromDate,
         metricDate,
         collectSonatypeMetrics,
@@ -102,45 +110,69 @@ async function main() {
 
   const localCollection = !collectGh && !collectNpm && !collectSonatype;
   if (localCollection) {
-    console.info(
-      `\n\n============\n\n>>> Collecting local metrics...`
-    );
+    console.info(`\n\n============\n\n>>> Collecting local metrics...`);
     // await saveNpmMetrics();
     // await saveSonatypeMetrics();
     await collectGhMetrics(true);
+    // await saveNpmMetrics();
+    await saveSonatypeMetrics();
+    // await collectGhMetrics(true);
   }
 }
 
 async function initialLoad(
+  metricName: string,
   initialLoadFromDate: Date,
   initialLoadToDate: Date,
   collectMetrics: (metricDate: string) => Promise<void>,
-  monthlyInterval = false
+  monthlyInterval = false,
+  skipLastSavedState = false
 ) {
-  let date = initialLoadFromDate;
+  const lastSavedState =
+    !skipLastSavedState && (await getLastSavedState(metricName));
+  const date = lastSavedState || initialLoadFromDate;
+
   if (monthlyInterval) {
     // Change the date to the first day of the month
     date.setDate(0);
   }
 
   while (date <= initialLoadToDate) {
     const dateStr = date.toISOString().split("T")[0];
-    console.log(`\n\n>>> Collecting metrics for date: ${dateStr}`);
+    console.log(`\n\n>>> Collecting metric ${metricName} for date: ${dateStr}`);
     await collectMetrics(dateStr);
-
     if (monthlyInterval) {
       // Move to the next month (JS will handle year change automatically)
       date.setMonth(date.getMonth() + 1);
     } else {
       date.setDate(date.getDate() + 1);
     }
+    await saveLastSavedState(metricName, date);
   }
 }
 
-main().then(() => {
-  console.log("Data collection completed successfully");
-  process.exit(0);
-}).catch((error) => {
-  console.error("Data collection failed", error);
-  process.exit(1);
-});
+export const getLastSavedState = async (metricName: string) => {
+  const stateDir = process.env.LAST_SAVED_STATE_PATH || "./";
+  const filePath = `${stateDir}/last-saved-state-${metricName}`;
+  if (!existsSync(filePath)) {
+    return undefined;
+  }
+  const lastSavedState = await readFile(filePath);
+  return new Date(lastSavedState.toString("utf8"));
+};
+
+export const saveLastSavedState = (metricName: string, date: Date) => {
+  const stateDir = process.env.LAST_SAVED_STATE_PATH || "./";
+  const filePath = `${stateDir}/last-saved-state-${metricName}`;
+  return writeFile(filePath, date.toISOString());
+};
+
+main()
+  .then(() => {
+    console.log("Data collection completed successfully");
+    process.exit(0);
+  })
+  .catch((error) => {
+    console.error("Data collection failed", error);
+    process.exit(1);
+  });

metrics-collector/src/service-app.ts

@@ -35,12 +35,24 @@ const initDb = async () => {
         value DOUBLE PRECISION NOT NULL,
         labels JSONB,
         metric_timestamp TIMESTAMPTZ NOT NULL DEFAULT now(),
-        created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+        updated_at TIMESTAMPTZ
     );
 
     CREATE INDEX IF NOT EXISTS idx_metric_name ON metrics(metric_name);
     CREATE INDEX IF NOT EXISTS idx_metric_timestamp ON metrics(metric_timestamp);
     CREATE INDEX IF NOT EXISTS idx_metric_labels ON metrics USING GIN(labels);
+
+    DO $$
+    BEGIN
+        IF NOT EXISTS (
+            SELECT 1 FROM pg_constraint 
+            WHERE conname = 'metrics_unique_constraint'
+        ) THEN
+            ALTER TABLE metrics ADD CONSTRAINT metrics_unique_constraint 
+            UNIQUE (metric_name, labels, metric_timestamp);
+        END IF;
+    END $$;
     `;
   await pool.query(createTableQuery);
   console.log("Database initialized");
@@ -84,12 +96,20 @@ const seedDb = async () => {
     (a, b) => new Date(a.timestamp).getTime() - new Date(b.timestamp).getTime()
   );
 
-  const insertQuery = `
-    INSERT INTO metrics (metric_name, value, labels, metric_timestamp) VALUES ($1, $2, $3, $4)
+  const upsertQuery = `
+    INSERT INTO metrics (metric_name, value, labels, metric_timestamp) 
+      VALUES ($1, $2, $3, $4)
+      ON CONFLICT (metric_name, labels, metric_timestamp)
+      DO UPDATE SET 
+        value = EXCLUDED.value, 
+        updated_at = CASE 
+          WHEN metrics.value = EXCLUDED.value THEN metrics.updated_at 
+          ELSE now() 
+        END
     `;
 
   for (const val of values) {
-    await pool.query(insertQuery, [
+    await pool.query(upsertQuery, [
       val.metricName,
       val.value,
       val.labels,
@@ -106,29 +126,67 @@ if (!process.env.DB_SKIP_INIT) {
 
 // Collect Metrics Endpoint
 app.post("/api/v1/metrics", async (req: any, res: any) => {
-  const { metricName, value, labels, timestamp } = req.body;
+  const { metricName, value, labels, timestamp, operation = "set" } = req.body;
   if (!metricName || value == undefined) {
     return res
       .status(400)
       .send({ error: "Missing required fields: metricName and value" });
   }
 
   try {
-    await pool.query(
-      "INSERT INTO metrics (metric_name, value, labels, metric_timestamp) VALUES ($1, $2, $3, $4)",
-      [
+    let query;
+
+    switch (operation) {
+      case "inc":
+        query = `
+          INSERT INTO metrics (metric_name, value, labels, metric_timestamp)
+          VALUES ($1, $2, $3, $4)
+          ON CONFLICT (metric_name, labels, metric_timestamp)
+          DO UPDATE SET 
+            value = metrics.value + EXCLUDED.value,
+            updated_at = CASE 
+              WHEN metrics.value = metrics.value + EXCLUDED.value THEN metrics.updated_at 
+              ELSE now() 
+            END
+        `;
+        break;
+      case "set":
+        query = `
+          INSERT INTO metrics (metric_name, value, labels, metric_timestamp)
+          VALUES ($1, $2, $3, $4)
+          ON CONFLICT (metric_name, labels, metric_timestamp)
+          DO UPDATE SET 
+            value = EXCLUDED.value, 
+            updated_at = CASE 
+              WHEN metrics.value = EXCLUDED.value THEN metrics.updated_at 
+              ELSE now() 
+            END
+        `;
+        break;
+      default:
+        throw new Error(`Unsupported operation: ${operation}`);
+    }
+
+    const params = [
+      metricName,
+      value,
+      labels || undefined,
+      timestamp || new Date().toISOString(),
+    ];
+
+    await pool.query(query, params);
+
+    console.info(
+      `>>> Metric ${operation}: ${JSON.stringify({
         metricName,
         value,
-        labels || undefined,
-        timestamp || new Date().toISOString(),
-      ]
-    );
-    console.info(
-      ">>> Metric collected: ",
-      JSON.stringify({ metricName, value, labels, timestamp })
+        labels,
+        timestamp,
+      })}`
     );
-    res.status(201).send({ message: "Metric collected" });
+    res.status(201).send();
   } catch (err) {
+    console.error("Failed to collect metric", err);
     res.status(500).send({ error: "Failed to collect metric" });
   }
 });

metrics-collector/src/sonatype-metrics.ts

@@ -72,17 +72,32 @@ export async function saveSonatypeMetrics() {
   const metrics = [];
 
   for (const artifact of artifacts) {
-    const [rawDownloads, uniqueIPs] = await Promise.all([
-      getArtifactStats(projectId, groupId, artifact, "raw"),
-      getArtifactStats(projectId, groupId, artifact, "ip"),
-    ]);
+    if (!["tbdex", "web5"].find((a) => artifact.includes(a))) {
+      continue;
+    }
 
-    metrics.push({
+    const rawDownloads = await getArtifactStats(
+      projectId,
+      groupId,
+      artifact,
+      "raw"
+    );
+    const uniqueIPs = await getArtifactStats(
+      projectId,
+      groupId,
+      artifact,
+      "ip"
+    );
+
+    const artifactMetrics = {
       artifact,
       timestamp,
       rawDownloads: rawDownloads.total,
       uniqueIPs: uniqueIPs.total,
-    });
+    };
+    console.info({ [artifact]: artifactMetrics });
+    metrics.push(artifactMetrics);
+    await new Promise((resolve) => setTimeout(resolve, 5000)); // prevent rate limit
   }
 
   console.info("Sonatype metrics collected successfully", { metrics });
@@ -180,8 +195,19 @@ async function getArtifactStats(
       }
     );
 
-    const data = await response.json();
-    return data.data;
+    const responseText = await response.text();
+    try {
+      const data = JSON.parse(responseText);
+      return data.data;
+    } catch (error) {
+      console.error(
+        "Failed to parse response as JSON:",
+        response.status,
+        response.statusText,
+        responseText
+      );
+      throw new Error("Unable to parse response as JSON");
+    }
   } catch (error) {
     console.error(
       `Error fetching ${type} stats for artifact ${artifactId}:`,