feat(backend): readiness probes, IPFS pinning, GitHub task verification and live leaderboard

CHANGELOG.md

@@ -9,6 +9,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- Backend: `/readyz` now verifies the database, Redis and Soroban RPC (RPC is `degraded` unless
+  `READINESS_REQUIRE_RPC=true`), and `/healthz` reports uptime and RPC pool state (#1251).
+- Backend: IPFS pin service for campaign images, rules, badge and metadata via Pinata, with
+  `POST /campaigns/:id/ipfs-pins` and background pinning on campaign creation (#1255).
+- Backend: GitHub webhook listener (`POST /webhooks/github`, HMAC-verified) that records merged PRs
+  and completed issues for developer-bounty task verification (#1256).
+- Backend: real-time campaign leaderboard over WebSocket (`leaderboard` channel) with rank-change
+  diffs (#1257). Fixes the WebSocket server constructor for `ws` 8.
 - Backend: asynchronous campaign export worker (`POST /campaigns/:id/exports`) on the durable job
   queue (#1260), Horizon deposit watcher for operator reward-token deposits (#1261), IP-cluster
   fraud detection with an admin review API (#1258), and a generated OpenAPI reference with Swagger

backend/.env.example

@@ -152,6 +152,25 @@ VAPID_SUBJECT=mailto:notifications@trivela.app
 # POST /api/v1/social-tasks/twitter/verify responds 503 TWITTER_NOT_CONFIGURED.
 # TWITTER_BEARER_TOKEN=
 
+# ── Readiness probes (#1251) ──────────────────────────────────────────────
+# /readyz checks the database, Redis (when REDIS_URL is set) and the Soroban RPC.
+# The RPC only fails readiness when this is true; otherwise it reports "degraded".
+# READINESS_REQUIRE_RPC=false
+# Per-check timeout in milliseconds.
+# READINESS_TIMEOUT_MS=2000
+
+# ── IPFS pinning (#1255) ──────────────────────────────────────────────────
+# Pins campaign images, rules, badge and metadata JSON through Pinata.
+# PINATA_JWT=
+# PINATA_API_URL=https://api.pinata.cloud
+
+# ── GitHub webhook (#1256) ────────────────────────────────────────────────
+# Shared secret configured on the GitHub webhook (POST /api/v1/webhooks/github).
+# GITHUB_WEBHOOK_SECRET=
+
+# ── Live leaderboard (#1257) ──────────────────────────────────────────────
+# Coalesce bursts of awarded points into one leaderboard push (milliseconds).
+# LEADERBOARD_DEBOUNCE_MS=200
 # ── Fraud detection (#1258) ───────────────────────────────────────────────
 # Distinct accounts registering from one IP subnet (/24 IPv4, /64 IPv6) within the
 # window before a fraud flag is raised for review.

backend/src/db/migrations/044_github_task_events.js

@@ -0,0 +1,26 @@
+export const version = 44;
+export const description = 'Add github_task_events for verified developer-bounty tasks';
+
+/**
+ * Merged pull requests and closed issues reported by the GitHub webhook (#1256).
+ * `delivery_id` is unique so GitHub redeliveries are idempotent.
+ */
+export function up(db) {
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS github_task_events (
+      id            INTEGER PRIMARY KEY AUTOINCREMENT,
+      delivery_id   TEXT NOT NULL UNIQUE,
+      kind          TEXT NOT NULL CHECK (kind IN ('pr_merged', 'issue_closed')),
+      repo          TEXT NOT NULL,
+      number        INTEGER NOT NULL,
+      github_login  TEXT NOT NULL,
+      url           TEXT,
+      title         TEXT,
+      occurred_at   TEXT,
+      recorded_at   TEXT NOT NULL,
+      UNIQUE (kind, repo, number, github_login)
+    );
+    CREATE INDEX IF NOT EXISTS idx_github_task_events_login
+      ON github_task_events(github_login COLLATE NOCASE, id DESC);
+  `);
+}

backend/src/db/migrations/045_ipfs_pins.js

@@ -0,0 +1,21 @@
+export const version = 45;
+export const description = 'Add ipfs_pins for campaign asset and metadata pinning';
+
+/** Pins created by the IPFS pin service (#1255). `content_hash` de-duplicates identical content. */
+export function up(db) {
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS ipfs_pins (
+      id            INTEGER PRIMARY KEY AUTOINCREMENT,
+      campaign_id   TEXT NOT NULL,
+      kind          TEXT NOT NULL CHECK (kind IN ('metadata', 'image', 'rules', 'badge')),
+      content_hash  TEXT NOT NULL,
+      cid           TEXT NOT NULL,
+      name          TEXT,
+      size_bytes    INTEGER,
+      provider      TEXT NOT NULL,
+      created_at    TEXT NOT NULL,
+      UNIQUE (campaign_id, kind, content_hash)
+    );
+    CREATE INDEX IF NOT EXISTS idx_ipfs_pins_campaign ON ipfs_pins(campaign_id, id DESC);
+  `);
+}

backend/src/index.js

@@ -112,6 +112,11 @@ import {
   createNotificationPreferencesRoutes,
 } from './routes/notifications.js';
 import { createOperatorBalanceJob } from './jobs/operatorBalanceJob.js';
+import { createGithubTaskStore } from './services/githubWebhook.js';
+import { createGithubTaskRoutes, createGithubWebhookRoutes } from './routes/githubWebhook.js';
+import { createIpfsPinService, createPinataProvider } from './services/ipfsPinService.js';
+import { IPFS_PIN_JOB_TYPE, createIpfsPinRoutes } from './routes/ipfsPins.js';
+import { createLeaderboardGateway, leaderboardRoom } from './websocket/leaderboardGateway.js';
 import { createSqliteExportJobRepository } from './dal/sqliteExportJobRepository.js';
 import {
   CAMPAIGN_EXPORT_JOB_TYPE,
@@ -377,6 +382,25 @@ export async function createApp(options = {}) {
   const auditLogRepository = dal.auditLogs;
   const webhookRepository = dal.webhooks;
   const referralRepository = dal.referrals;
+
+  // #1257 — live leaderboard pushes over the WebSocket gateway
+  const leaderboardGateway = createLeaderboardGateway({
+    getLeaderboard: (campaignId) =>
+      referralRepository
+        .getLeaderboard(campaignId, { limit: 10 })
+        .rows.map((/** @type {any} */ r) => ({
+          wallet: r.referrerAddress,
+          rank: r.rank,
+          points: r.referralCount,
+        })),
+    publish: (campaignId, message) =>
+      getWebSocketServer()?.broadcast(leaderboardRoom(campaignId), message) ?? 0,
+    debounceMs: normalizePositiveInteger(
+      /** @type {any} */ (options.leaderboardDebounceMs) ?? process.env.LEADERBOARD_DEBOUNCE_MS,
+      200,
+    ),
+    logger: log,
+  });
   const variantRepository = dal.variants;
   const cohortRepository = dal.cohorts;
   const pushSubscriptionRepository = dal.pushSubscriptions;
@@ -704,6 +728,19 @@ export async function createApp(options = {}) {
   app.use(createDeprecationMiddleware({ log }));
   app.use(traceparentMiddleware());
   app.use(requestLogger);
+  // #1256 — GitHub webhook. Mounted before express.json(): the HMAC signature is
+  // computed over the raw request bytes.
+  const githubTaskStore = createGithubTaskStore({ db: dal.db, logger: log });
+  app.use(
+    API_V1_PREFIX,
+    createGithubWebhookRoutes({
+      store: githubTaskStore,
+      getSecret: () =>
+        String(options.githubWebhookSecret ?? process.env.GITHUB_WEBHOOK_SECRET ?? ''),
+      logger: log,
+    }),
+  );
+
   app.use(express.json({ limit: jsonBodyLimit }));
 
   const uploadDir = process.env.UPLOAD_DIR ?? './uploads';
@@ -856,6 +893,21 @@ export async function createApp(options = {}) {
     setInterval(doExport, 24 * 60 * 60 * 1_000).unref?.();
   }
 
+  // #1255 — IPFS pinning of campaign assets (enabled when PINATA_JWT is set)
+  const ipfsPinService =
+    /** @type {any} */ (options.ipfsPinService) ??
+    (process.env.PINATA_JWT
+      ? createIpfsPinService({
+          db: dal.db,
+          provider: createPinataProvider({
+            jwt: process.env.PINATA_JWT,
+            baseUrl: process.env.PINATA_API_URL,
+            fetchImpl,
+          }),
+          fetchImpl,
+          logger: log,
+        })
+      : null);
   // #1260 — asynchronous campaign exports, processed off the request path
   const campaignExportJobRepository = createSqliteExportJobRepository({ db: dal.db });
   const campaignExportWorker = createCampaignExportWorker({
@@ -870,6 +922,11 @@ export async function createApp(options = {}) {
   const durableJobQueue = createDurableJobQueue({
     store: jobQueueStore,
     handlers: {
+      // #1255 — background pinning, enqueued when a campaign is created
+      [IPFS_PIN_JOB_TYPE]: async (/** @type {any} */ payload) => {
+        const target = campaignRepository.getById(String(payload?.campaignId));
+        if (target && ipfsPinService) await ipfsPinService.pinCampaign(target);
+      },
       [CAMPAIGN_EXPORT_JOB_TYPE]: campaignExportWorker.handle,
       // #922 — end-of-campaign claimable balance creation, enqueued from
       // POST /campaigns/:id/claimable-balances instead of running inline.
@@ -1028,12 +1085,53 @@ export async function createApp(options = {}) {
     res.json(payload);
   });
 
-  const probeHandlers = createProbeHandlers({ getIsShuttingDown: () => isShuttingDown });
+  // #1251 — readiness verifies the database, Redis (when configured) and the
+  // Soroban RPC. The RPC is reported but only fails readiness when
+  // READINESS_REQUIRE_RPC=true, so an RPC blip does not evict every pod.
+  const readinessRequireRpc = ['1', 'true'].includes(
+    String(options.readinessRequireRpc ?? process.env.READINESS_REQUIRE_RPC ?? '').toLowerCase(),
+  );
+  const probeHandlers = createProbeHandlers({
+    getIsShuttingDown: () => isShuttingDown,
+    timeoutMs: normalizePositiveInteger(
+      /** @type {any} */ (options.readinessTimeoutMs) ?? process.env.READINESS_TIMEOUT_MS,
+      2000,
+    ),
+    checks: {
+      database: {
+        run: () => {
+          dal.db.prepare('SELECT 1 AS ok').get();
+        },
+      },
+      redis: usageRedisClient
+        ? {
+            run: async () => {
+              const pong = await usageRedisClient.ping();
+              if (pong !== 'PONG') throw new Error(`unexpected PING reply: ${pong}`);
+            },
+          }
+        : null,
+      rpc: {
+        required: readinessRequireRpc,
+        run: async () => {
+          const result = await checkSorobanRpcHealth({
+            rpcUrl: rpcPool.getHealthyRpcUrl(),
+            fetchImpl,
+          });
+          if (/** @type {any} */ (result).status !== 'ok') {
+            throw new Error(`Soroban RPC is ${/** @type {any} */ (result).status}`);
+          }
+          return { pool: rpcPool.getStatus() };
+        },
+      },
+    },
+    healthDetails: () => ({ rpc: rpcPool.getStatus() }),
+  });
   app.get('/health/live', probeHandlers.livenessHandler);
   app.get('/health/ready', probeHandlers.readinessHandler);
   app.get('/livez', probeHandlers.livenessHandler);
   app.get('/readyz', probeHandlers.readinessHandler);
-  app.get('/healthz', probeHandlers.livenessHandler);
+  app.get('/healthz', probeHandlers.healthHandler);
 
   app.get('/ready', (_req, res) => {
     if (isShuttingDown) {
@@ -1547,6 +1645,10 @@ export async function createApp(options = {}) {
         });
       }
 
+      if (ipfsPinService && !options.disableJobs) {
+        durableJobQueue.enqueue(IPFS_PIN_JOB_TYPE, { campaignId: campaign.id }, { maxAttempts: 3 });
+      }
+
       shortCache.clear();
       return res.status(201).json(serializeCampaign(campaign));
     } catch (error) {
@@ -2956,6 +3058,8 @@ export async function createApp(options = {}) {
         });
       }
 
+      // Points were awarded to the referrer: push new standings to WebSocket subscribers (#1257).
+      leaderboardGateway.trigger(campaign.id);
       // #1258 — flag clusters of signups from one IP subnet / proxy exit node.
       // Detection is advisory: a failure here must never fail the signup.
       try {
@@ -3205,6 +3309,19 @@ export async function createApp(options = {}) {
     });
     app.use(prefix, rateLimiter, auditRouter);
 
+    // #1256 — verified GitHub tasks (master key)
+    app.use(
+      `${prefix}/admin/github-tasks`,
+      rateLimiter,
+      requireMasterKey,
+      createGithubTaskRoutes({ store: githubTaskStore }),
+    );
+
+    // #1255 — IPFS pinning of campaign assets (API key; the router applies the guard itself)
+    app.use(
+      prefix,
+      rateLimiter,
+      createIpfsPinRoutes({ getService: () => ipfsPinService, campaignRepository, guard }),
     // #1258 — fraud flag review (master key)
     app.use(
       `${prefix}/admin/fraud`,
@@ -3405,8 +3522,11 @@ export async function createApp(options = {}) {
   // Central error handler — must be registered after all routes
   app.use(errorHandler);
 
+  app._leaderboardGateway = leaderboardGateway;
+
   app._close = () => {
     isShuttingDown = true;
+    leaderboardGateway.stop();
     depositWatcher.stop();
     try {
       dal.db.close();
@@ -3444,6 +3564,8 @@ export async function startServer(options = {}) {
     try {
       initializeWebSocket(server, {
         path: process.env.WEBSOCKET_PATH || '/ws',
+        leaderboardSnapshot: (/** @type {string} */ campaignId) =>
+          app._leaderboardGateway.snapshot(campaignId),
       });
       log.info('WebSocket server initialized on /ws');
     } catch (error) {