airdrops.js

const crypto = require('crypto');
const cache = require('./cache');
const logger = require('../logger');
const { Horizon } = require('@stellar/stellar-sdk');
const config = require('../config');

const IDS_KEY = 'airdrops:ids';

function airdropKey(id) {
  return `airdrop:${id}`;
}

function recipientsKey(airdropId) {
  return `airdrop:${airdropId}:recipients`;
}

// Tracks the set of addresses already stored for an airdrop for O(1) cross-request
// duplicate detection. Kept in sync with the recipients list by create/addRecipients/remove.
function recipientAddressSetKey(airdropId) {
  return `airdrop:${airdropId}:addresses`;
}

function generateId() {
  return `drop_${crypto.randomUUID().replace(/-/g, '').slice(0, 16)}`;
}

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();
  cachedLedger = ledger.records[0].sequence;
  cachedLedgerAt = now;
  return cachedLedger;
}

async function create(data) {
  const { name, description, asset, asset_issuer, total_amount, expiry_ledger, contract_airdrop_id, recipients = [] } = data;
  const id = generateId();

  const airdrop = {
    id,
    name,
    description,
    asset,
    asset_issuer,
    total_amount,
    expiry_ledger,
    // Linking field: once the on-chain airdrop ID is known (e.g. after the
    // Soroban contract is invoked externally), populate this so the REST
    // record can be correlated with indexer-observed on-chain state (#122).
    contract_airdrop_id: contract_airdrop_id || null,
    status: 'draft',
    created_at: new Date().toISOString(),
    updated_at: new Date().toISOString(),
  };

  const redis = cache.getClient();
  await cache.set(airdropKey(id), airdrop);
  await redis.zadd(IDS_KEY, Date.now(), id);

  if (recipients.length > 0) {
    await redis.rpush(recipientsKey(id), ...recipients.map((r) => JSON.stringify(r)));
    await redis.sadd(recipientAddressSetKey(id), ...recipients.map((r) => r.address));
  }

  return airdrop;
}

/**
 * Pages through the full airdrop ID sorted set via ZSCAN instead of ZREVRANGE. Used
 * by the expiry reconciliation job (#88), which needs to visit every
 * airdrop every cycle: ZREVRANGE with 0 -1 returns the whole set in one call
 * and would need it all held in memory at once, which doesn't scale as the
 * set grows. ZSCAN 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, batchWithScores] = await redis.zscan(IDS_KEY, cursor, 'COUNT', batchSize);
    cursor = nextCursor;
    const batch = batchWithScores.filter((_, index) => index % 2 === 0);
    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);
}

// Returns { airdrops, total } rather than a full pagination envelope — the
// route layer wraps this in the canonical envelope via
// utils/paginate.js's paginateResponse, the same split routes/alerts.js
// already uses for its own list endpoint (#131).
async function list(page = 1, limit = 20) {
  const redis = cache.getClient();
  const total = await redis.zcard(IDS_KEY);
  const start = (page - 1) * limit;
  const end = start + limit - 1;
  const paginatedIds = await redis.zrevrange(IDS_KEY, start, end);
  const airdrops = await Promise.all(paginatedIds.map((id) => cache.get(airdropKey(id))));

  return { airdrops: airdrops.filter(Boolean), total };
}

async function get(id) {
  return await cache.get(airdropKey(id));
}

async function update(id, data) {
  const airdrop = await get(id);
  if (!airdrop) return null;

  const { name, description, expiry_ledger, contract_airdrop_id } = data;
  const updated = {
    ...airdrop,
    name: name !== undefined ? name : airdrop.name,
    description: description !== undefined ? description : airdrop.description,
    expiry_ledger: expiry_ledger !== undefined ? expiry_ledger : airdrop.expiry_ledger,
    contract_airdrop_id: contract_airdrop_id !== undefined ? contract_airdrop_id : airdrop.contract_airdrop_id,
    updated_at: new Date().toISOString(),
  };

  await cache.set(airdropKey(id), updated);
  return updated;
}

async function remove(id) {
  const redis = cache.getClient();
  const existing = await get(id);
  if (!existing) return null;

  await cache.del(airdropKey(id));
  await cache.del(recipientsKey(id));
  await cache.del(recipientAddressSetKey(id));
  await redis.zrem(IDS_KEY, id);
  return existing;
}

async function cancel(id) {
  const airdrop = await get(id);
  if (!airdrop) return null;

  if (airdrop.status === 'cancelled') {
    return airdrop;
  }

  const updated = {
    ...airdrop,
    status: 'cancelled',
    updated_at: new Date().toISOString(),
  };

  await cache.set(airdropKey(id), updated);
  return updated;
}

// Returns an array of addresses that were already present in a prior call.
// An empty array means all recipients were accepted and stored.
async function addRecipients(airdropId, recipients) {
  const redis = cache.getClient();
  const addresses = recipients.map((r) => r.address);

  // SADD returns 1 for each newly added member, 0 for duplicates.
  // By comparing the added count against the total we identify which
  // addresses were already stored from the initial POST /airdrops body
  // or a prior POST /airdrops/:id/recipients call.
  const addedCounts = await Promise.all(
    addresses.map((addr) => redis.sadd(recipientAddressSetKey(airdropId), addr)),
  );

  const duplicates = addresses.filter((_, i) => addedCounts[i] === 0);
  if (duplicates.length > 0) {
    // Roll back the addresses we just added so the set stays consistent.
    await redis.srem(recipientAddressSetKey(airdropId), ...addresses.filter((_, i) => addedCounts[i] === 1));
    return duplicates;
  }

  await redis.rpush(recipientsKey(airdropId), ...recipients.map((r) => JSON.stringify(r)));
  return [];
}

// Returns { recipients, total } — see list()'s comment above.
async function listRecipients(airdropId, page = 1, limit = 20) {
  const redis = cache.getClient();
  const total = await redis.llen(recipientsKey(airdropId));
  const start = (page - 1) * limit;
  const end = start + limit - 1;
  const serializedRecipients = await redis.lrange(recipientsKey(airdropId), start, end);
  const recipients = serializedRecipients.map((r) => JSON.parse(r));

  return { recipients, total };
}

module.exports = {
  create,
  list,
  get,
  update,
  remove,
  cancel,
  addRecipients,
  listRecipients,
  getCurrentLedger,
  scanIds,
  markExpired,
  TERMINAL_STATUSES,
};