src/webhook-redelivery.lambda.ts

Rama main · réplica de lectura del repositorio original.

import type { Octokit } from '@octokit/rest' with { 'resolution-mode': 'import' };
import { getAppOctokit, redeliver } from './lambda-github';

/**
 * Get webhook delivery failures since the last processed delivery ID.
 *
 * @internal
 */
async function newDeliveryFailures(octokit: Octokit, sinceId: bigint) {
  const deliveries: Map<string, { id: bigint; deliveredAt: Date; redelivery: boolean }> = new Map();
  const successfulDeliveries: Set<string> = new Set();
  const timeLimitMs = 1000 * 60 * 30; // don't look at deliveries over 30 minutes old
  let lastId = 0n;
  let processedCount = 0;

  for await (const response of octokit.paginate.iterator('GET /app/hook/deliveries')) {
    if (response.status !== 200) {
      throw new Error('Failed to fetch webhook deliveries');
    }

    for (const delivery of response.data) {
      const deliveryId = BigInt(delivery.id);
      const deliveredAt = new Date(delivery.delivered_at);
      const success = delivery.status === 'OK';

      if (deliveryId <= sinceId) {
        // stop processing if we reach the last processed delivery ID
        console.info({
          notice: 'Reached last processed delivery ID',
          sinceId: String(sinceId),
          deliveryId: String(deliveryId),
          guid: delivery.guid,
          processedCount,
        });
        return { deliveries, lastId };
      }

      lastId = deliveryId > lastId ? deliveryId : lastId;

      if (deliveredAt.getTime() < Date.now() - timeLimitMs) {
        // stop processing if the delivery is too old (for first iteration and performance of further iterations)
        console.info({
          notice: 'Stopping at old delivery',
          deliveryId: String(deliveryId),
          guid: delivery.guid,
          deliveredAt: deliveredAt,
          processedCount,
        });
        return { deliveries, lastId };
      }

      console.debug({
        notice: 'Processing webhook delivery',
        deliveryId: String(deliveryId),
        guid: delivery.guid,
        status: delivery.status,
        deliveredAt: delivery.delivered_at,
        redelivery: delivery.redelivery,
      });
      processedCount++;

      if (success) {
        successfulDeliveries.add(delivery.guid);
        continue;
      }

      if (successfulDeliveries.has(delivery.guid)) {
        // do not redeliver deliveries that were already successful
        continue;
      }

      deliveries.set(delivery.guid, { id: deliveryId, deliveredAt, redelivery: delivery.redelivery });
    }
  }

  console.info({
    notice: 'No more webhook deliveries to process',
    deliveryId: 'DONE',
    guid: 'DONE',
    deliveredAt: 'DONE',
    processedCount,
  });

  return { deliveries, lastId };
}

let lastDeliveryIdProcessed = 0n;
const failures: Map<string, { id: bigint; firstDeliveredAt: Date }> = new Map();

/**
 * Clear the cache of webhook delivery failures.
 *
 * For unit testing purposes only.
 *
 * @internal
 */
export function clearFailuresCache() {
  lastDeliveryIdProcessed = 0n;
  failures.clear();
}

export async function handler() {
  const octokit = await getAppOctokit();
  if (!octokit) {
    console.info({
      notice: 'Skipping webhook redelivery',
      reason: 'App installation might not be configured or the app is not installed.',
    });
    return;
  }

  // fetch deliveries since the last processed delivery ID
  // for any failures:
  //  1. if this is not a redelivery, save the delivery ID and time, and finally retry
  //  2. if this is a redelivery, check if the original delivery is still within the time limit and retry if it is
  const { deliveries, lastId } = await newDeliveryFailures(octokit, lastDeliveryIdProcessed);
  lastDeliveryIdProcessed = lastId > lastDeliveryIdProcessed ? lastId : lastDeliveryIdProcessed;
  const timeLimitMs = 1000 * 60 * 60 * 3; // retry for up to 3 hours
  for (const [guid, details] of deliveries) {
    if (!details.redelivery) {
      failures.set(guid, { id: details.id, firstDeliveredAt: details.deliveredAt });
      console.log({
        notice: 'Redelivering failed delivery',
        deliveryId: String(details.id),
        guid: guid,
        firstDeliveredAt: details.deliveredAt,
      });
      await redeliver(octokit, details.id);
    } else {
      // if this is a redelivery, check if the original delivery is still within the time limit
      const originalFailure = failures.get(guid);
      if (originalFailure) {
        if (new Date().getTime() - originalFailure.firstDeliveredAt.getTime() < timeLimitMs) {
          console.log({
            notice: 'Redelivering failed delivery',
            deliveryId: String(details.id),
            guid: guid,
            firstDeliveredAt: originalFailure.firstDeliveredAt,
          });
          await redeliver(octokit, details.id);
        } else {
          failures.delete(guid); // no need to keep track of this anymore
          console.log({
            notice: 'Skipping redelivery of old failed delivery',
            deliveryId: String(details.id),
            guid: guid,
            firstDeliveredAt: originalFailure?.firstDeliveredAt,
          });
        }
      } else {
        console.log({
          notice: 'Skipping redelivery of old failed delivery',
          deliveryId: String(details.id),
          guid: guid,
        });
      }
    }
  }
}