Merge pull request #104 from abayomicornelius/feat/airdrop-expiry-reconciliation-job

feat: automatic airdrop expiry reconciliation job

Author: prodbycorne

.env.example

@@ -46,6 +46,15 @@ PRICE_STALE_THRESHOLD_MINUTES=5
 # PRICE_ANOMALY_THRESHOLD_PCT: number. Default: 20.
 PRICE_ANOMALY_THRESHOLD_PCT=20
 
+# AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS: number. Default: 60.
+AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS=60
+
+# AIRDROP_LEDGER_CACHE_TTL_MS: number. Default: 5000.
+AIRDROP_LEDGER_CACHE_TTL_MS=5000
+
+# AIRDROP_EXPIRY_SCAN_BATCH_SIZE: number. Default: 100.
+AIRDROP_EXPIRY_SCAN_BATCH_SIZE=100
+
 # API key auth
 ADMIN_API_KEY=
 

README.md

@@ -39,7 +39,7 @@ Registers subscriber endpoints for SmartDrop lifecycle events and delivers signe
 - `airdrop.created`
 - `airdrop.executing`
 - `airdrop.completed`
-- `airdrop.failed`
+- `airdrop.failed` — fired automatically when an airdrop expires (see below), in addition to any other failure path
 - `recipient.claimed`
 
 **Features:**
@@ -49,6 +49,28 @@ Registers subscriber endpoints for SmartDrop lifecycle events and delivers signe
 - Delivery logs with response code, error, duration, and attempt count
 - Dead-letter storage after retry exhaustion
 
+### Airdrop Expiry Reconciliation
+
+Airdrops carry an `expiry_ledger`, validated as being in the future only at
+creation/update time. A background job (`src/jobs/airdropExpiry.js`, same
+`start()`/`stop()` pattern as the price-refresh and webhook-retry jobs)
+periodically re-checks that condition against the live network:
+
+- Every `AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS` (default 60s), fetches the
+  current Horizon ledger sequence and scans every airdrop still in a
+  non-terminal status (`draft`, `executing`).
+- Any airdrop whose `expiry_ledger` has passed is atomically transitioned to
+  `expired` and fires an `airdrop.failed` webhook event (`data.reason:
+  "expired"`) to every subscriber registered for it — no client action
+  required.
+- The transition is idempotent: re-running the check against an
+  already-expired airdrop is a guaranteed no-op, so the webhook fires
+  exactly once per airdrop even if the job runs again before anything else
+  changes its status.
+- If Horizon is temporarily unreachable, the job logs a warning and skips
+  that cycle rather than crashing — airdrops are simply re-checked on the
+  next tick.
+
 ---
 
 ## 🚀 Quick Start (Docker Development)

src/config.js

@@ -40,6 +40,9 @@ const env = cleanEnv(rawEnv, {
   PRICE_ANOMALY_THRESHOLD_PCT: num({ default: 20 }),
   PRICE_SOURCE_CIRCUIT_COOLDOWN_MS: num({ default: 15 * 60 * 1000 }),
   PRICE_SOURCE_CIRCUIT_REMINDER_MS: num({ default: 5 * 60 * 1000 }),
+  AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS: num({ default: 60 }),
+  AIRDROP_LEDGER_CACHE_TTL_MS: num({ default: 5000 }),
+  AIRDROP_EXPIRY_SCAN_BATCH_SIZE: num({ default: 100 }),
   LOG_LEVEL: str({
     default: 'info',
     choices: ['debug', 'info', 'warn', 'error'],
@@ -86,6 +89,19 @@ module.exports = {
     // line per fetch cycle for the entire cooldown window.
     circuitReminderIntervalMs: env.PRICE_SOURCE_CIRCUIT_REMINDER_MS,
   },
+  airdrops: {
+    // How often the expiry reconciliation job scans non-terminal airdrops
+    // against the live Horizon ledger sequence.
+    expiryCheckIntervalSeconds: env.AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS,
+    // getCurrentLedger() is a live Horizon call with no caching; a job that
+    // polls frequently should reuse the same ledger sequence for this long
+    // rather than hitting Horizon once per airdrop per cycle.
+    ledgerCacheTtlMs: env.AIRDROP_LEDGER_CACHE_TTL_MS,
+    // SSCAN batch size used when scanning the full airdrop ID set — keeps
+    // each Redis round-trip small instead of loading the whole set (SMEMBERS)
+    // into memory at once.
+    expiryScanBatchSize: env.AIRDROP_EXPIRY_SCAN_BATCH_SIZE,
+  },
   auth: {
     adminApiKey: env.ADMIN_API_KEY,
   },

src/index.js

@@ -8,6 +8,7 @@ const cache = require('./services/cache');
 const priceOracle = require('./services/priceOracle');
 const priceRefreshJob = require('./jobs/priceRefresh');
 const webhookRetryWorker = require('./jobs/webhookRetryWorker');
+const airdropExpiryJob = require('./jobs/airdropExpiry');
 const buildCorsMiddleware = require('./middleware/cors');
 const { requestIdMiddleware } = require('./middleware/requestId');
 const { requireApiKey } = require('./middleware/auth');
@@ -56,6 +57,7 @@ function shutdown(signal) {
     logger.info(`${signal} received, shutting down`);
     priceRefreshJob.stop();
     webhookRetryWorker.stop();
+    airdropExpiryJob.stop();
     require('./ws/PriceSubscriptionManager').stopHeartbeat();
     if (server) server.close();
     await cache.disconnect();
@@ -69,6 +71,7 @@ if (require.main === module) {
     priceWebSocket.attach(server);
     priceRefreshJob.start();
     webhookRetryWorker.start();
+    airdropExpiryJob.start();
   });
 
   process.on('SIGTERM', shutdown('SIGTERM'));

src/jobs/airdropExpiry.js

@@ -0,0 +1,130 @@
+'use strict';
+
+const cron = require('node-cron');
+const airdropsService = require('../services/airdrops');
+const webhookDispatcher = require('../services/webhookDispatcher');
+const config = require('../config');
+const logger = require('../logger');
+
+let scheduledTask = null;
+
+/**
+ * One reconciliation pass: scans every non-terminal airdrop and expires any
+ * whose expiry_ledger has passed the current Horizon ledger. Exported
+ * separately from start() so tests can drive a single tick deterministically
+ * instead of waiting on cron.
+ */
+async function tick() {
+  let currentLedger;
+  try {
+    currentLedger = await airdropsService.getCurrentLedger();
+  } catch (err) {
+    // Matches priceOracle.js's graceful-degradation style: Horizon being
+    // temporarily unreachable is expected and recoverable — log and skip
+    // this cycle rather than crashing the job or throwing out of the cron
+    // callback.
+    logger.warn('Airdrop expiry check skipped, Horizon unreachable', { error: err.message });
+    return;
+  }
+
+  let expiredCount = 0;
+  let scannedCount = 0;
+
+  for await (const batch of airdropsService.scanIds()) {
+    for (const id of batch) {
+      scannedCount += 1;
+
+      let airdrop;
+      try {
+        airdrop = await airdropsService.get(id);
+      } catch (err) {
+        logger.error('Airdrop expiry check failed to read airdrop, skipping', {
+          airdrop_id: id,
+          error: err.message,
+        });
+        continue;
+      }
+
+      if (!airdrop || airdropsService.TERMINAL_STATUSES.has(airdrop.status)) continue;
+      if (!airdrop.expiry_ledger || airdrop.expiry_ledger > currentLedger) continue;
+
+      // Cheap pre-filter above avoids an unnecessary Lua round trip for the
+      // (typically vast majority of) airdrops nowhere near expiry.
+      // markExpired re-checks status and expiry_ledger atomically — if this
+      // pre-filter read was stale, or another cycle/process already
+      // transitioned it, markExpired safely no-ops instead of double-firing.
+      let updated;
+      try {
+        updated = await airdropsService.markExpired(id, currentLedger);
+      } catch (err) {
+        logger.error('Airdrop expiry transition failed, skipping', {
+          airdrop_id: id,
+          error: err.message,
+        });
+        continue;
+      }
+      if (!updated) continue;
+
+      expiredCount += 1;
+      try {
+        await webhookDispatcher.dispatch({
+          event_type: 'airdrop.failed',
+          event_id: `evt_airdrop_expired_${id}_${currentLedger}`,
+          data: {
+            airdrop_id: id,
+            reason: 'expired',
+            expiry_ledger: updated.expiry_ledger,
+            current_ledger: currentLedger,
+          },
+        });
+      } catch (err) {
+        // The transition already committed — the airdrop is correctly
+        // expired regardless of whether the webhook delivery attempt
+        // itself failed to enqueue. Losing this specific delivery on a
+        // dispatch-time error (as opposed to an individual subscriber's
+        // endpoint failing, which webhookDispatcher already retries) is an
+        // accepted gap here — see #84 for the broader non-atomic-writes
+        // theme this falls under.
+        logger.error('Airdrop expiry webhook dispatch failed', {
+          airdrop_id: id,
+          error: err.message,
+        });
+      }
+    }
+  }
+
+  logger.info('Airdrop expiry check completed', {
+    currentLedger,
+    scanned: scannedCount,
+    expired: expiredCount,
+  });
+}
+
+function start() {
+  if (scheduledTask) return;
+
+  const intervalSeconds = config.airdrops.expiryCheckIntervalSeconds;
+  const cronExpression = `*/${intervalSeconds} * * * * *`;
+
+  scheduledTask = cron.schedule(
+    cronExpression,
+    () => {
+      tick().catch((err) => {
+        logger.error('Airdrop expiry check failed', { error: err.message });
+      });
+    },
+    { scheduled: true },
+  );
+
+  logger.info('Airdrop expiry job started', { intervalSeconds });
+}
+
+function stop() {
+  if (scheduledTask) {
+    scheduledTask.stop();
+    scheduledTask = null;
+    logger.info('Airdrop expiry job stopped');
+  }
+}
+
+module.exports = { start, stop, tick };

src/services/airdrops.js

@@ -20,9 +20,25 @@ function generateId() {
 
 const horizon = new Horizon.Server(config.stellar.horizonUrl);
 
+// getCurrentLedger() is a live Horizon call. Callers that need to check many
+// airdrops in quick succession (the expiry reconciliation job, in
+// particular — see #88) would otherwise issue one Horizon request per
+// airdrop per cycle; cache the result briefly so bursts of calls within the
+// same window reuse one ledger read instead of hammering Horizon, the same
+// rate-limit concern already applied to CoinGecko/CoinMarketCap elsewhere.
+let cachedLedger = null;
+let cachedLedgerAt = 0;
+
 async function getCurrentLedger() {
+  const now = Date.now();
+  if (cachedLedger !== null && now - cachedLedgerAt < config.airdrops.ledgerCacheTtlMs) {
+    return cachedLedger;
+  }
+
   const ledger = await horizon.ledgers().order('desc').limit(1).call();
-  return ledger.records[0].sequence;
+  cachedLedger = ledger.records[0].sequence;
+  cachedLedgerAt = now;
+  return cachedLedger;
 }
 
 async function create(data) {
@@ -53,6 +69,69 @@ async function create(data) {
   return airdrop;
 }
 
+/**
+ * Pages through the full airdrop ID set via SSCAN instead of SMEMBERS. Used
+ * by the expiry reconciliation job (#88), which needs to visit every
+ * airdrop every cycle: SMEMBERS returns the whole set in one blocking call
+ * and would need it all held in memory at once, which doesn't scale as the
+ * set grows. SSCAN pages incrementally with a small, bounded cursor cost per
+ * call. `list()` above is unchanged — this is a separate, job-internal
+ * scanning path, not a replacement for the paginated HTTP listing endpoint.
+ */
+async function* scanIds(batchSize = config.airdrops.expiryScanBatchSize) {
+  const redis = cache.getClient();
+  let cursor = '0';
+  do {
+    const [nextCursor, batch] = await redis.sscan(IDS_KEY, cursor, 'COUNT', batchSize);
+    cursor = nextCursor;
+    if (batch.length > 0) {
+      yield batch;
+    }
+  } while (cursor !== '0');
+}
+
+// Statuses an airdrop cannot leave once reached.
+const TERMINAL_STATUSES = new Set(['completed', 'failed', 'cancelled', 'expired']);
+
+/**
+ * Atomically transitions an airdrop to 'expired' if — and only if — it's
+ * still in a non-terminal status *and* its expiry_ledger has actually
+ * passed, checked and written in a single Lua script so two processes (or
+ * two overlapping job cycles) racing on the same airdrop can't both "win"
+ * and each fire a duplicate webhook. Returns the updated airdrop on a
+ * successful transition, or null if nothing changed (already terminal, not
+ * yet expired, or the airdrop doesn't exist) — callers use that to decide
+ * whether to dispatch a webhook.
+ */
+const MARK_EXPIRED_SCRIPT = `
+local raw = redis.call('GET', KEYS[1])
+if not raw then return false end
+local airdrop = cjson.decode(raw)
+local terminal = { completed = true, failed = true, cancelled = true, expired = true }
+if terminal[airdrop.status] then return false end
+if not airdrop.expiry_ledger or tonumber(airdrop.expiry_ledger) > tonumber(ARGV[1]) then
+  return false
+end
+airdrop.status = 'expired'
+airdrop.updated_at = ARGV[2]
+local updated = cjson.encode(airdrop)
+redis.call('SET', KEYS[1], updated)
+return updated
+`;
+
+async function markExpired(id, currentLedger) {
+  const redis = cache.getClient();
+  const result = await redis.eval(
+    MARK_EXPIRED_SCRIPT,
+    1,
+    airdropKey(id),
+    currentLedger,
+    new Date().toISOString(),
+  );
+  if (!result) return null;
+  return JSON.parse(result);
+}
+
 async function list(page = 1, limit = 20) {
   const redis = cache.getClient();
   const ids = await redis.smembers(IDS_KEY);
@@ -156,4 +235,7 @@ module.exports = {
   addRecipients,
   listRecipients,
   getCurrentLedger,
+  scanIds,
+  markExpired,
+  TERMINAL_STATUSES,
 };

src/services/webhookEvents.js

@@ -10,7 +10,16 @@ const POOL_EVENTS = Object.freeze([
 
 const PRICE_EVENTS = Object.freeze(['price.alert']);
 
-const ALL_EVENTS = Object.freeze([...POOL_EVENTS, ...PRICE_EVENTS]);
+// Only 'airdrop.failed' is registered here — it's the one event this
+// codebase actually dispatches today (the expiry reconciliation job, #88).
+// The README also documents airdrop.created/executing/completed, but
+// nothing in the codebase dispatches those yet; registering unused event
+// names here would let a client subscribe to something that can never
+// fire, so they're left out until whatever feature actually dispatches
+// them lands.
+const AIRDROP_EVENTS = Object.freeze(['airdrop.failed']);
+
+const ALL_EVENTS = Object.freeze([...POOL_EVENTS, ...PRICE_EVENTS, ...AIRDROP_EVENTS]);
 const EVENT_SET = new Set(ALL_EVENTS);
 
 const WILDCARD = '*';
@@ -33,6 +42,7 @@ function matchesSubscription(subscribedEvents, eventType) {
 module.exports = {
   POOL_EVENTS,
   PRICE_EVENTS,
+  AIRDROP_EVENTS,
   ALL_EVENTS,
   WILDCARD,
   isKnownEvent,