GithubBackfill.java

// Class GithubBackfill (Worker) performs the initial 90-day historical backfill
// for newly onboarded GitHub orgs. It drains the vendor's events API page
// by page and emits each event to the connector DataSink, which guarantees
// at-least-once delivery to the downstream daily-activity processor.
//
// Cursor checkpointing ensures we can resume after restart without losing
// or duplicating any event. Each event carries a stable event_id; downstream
// dedups on it, so any spurious replays are safe.
//
// TODO(rohan): migrate to RateLimitV1.waitN once the new limiter ships.

package com.workloom.connector.githubbackfill;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

import com.fasterxml.jackson.databind.ObjectMapper;

import com.workloom.connector.datasink.DataSink;
import com.workloom.connector.vendorgateway.VendorGatewayClient;
import com.workloom.connector.vendorgateway.Response;

// Worker drains GitHub events for a single org from the backfill window
// up to "now", persisting a cursor after each successful page so the
// worker is safe to restart at any time.
public class GithubBackfill {

    // backfillWindow is the rolling history we ingest on first onboard.
    private static final long BACKFILL_WINDOW_MS = 90L * 24 * 60 * 60 * 1000;
    private static final int PAGE_SIZE = 100;
    private static final int MAX_RETRIES = 5;

    private static final ObjectMapper MAPPER = new ObjectMapper();

    // CursorStore persists the vendor's opaque pagination cursor per org.
    public interface CursorStore {
        String load(String orgID) throws Exception;
        void save(String orgID, String cursor) throws Exception;
    }

    // RateLimiter wraps the per-app GitHub credential bucket. All workers
    // across all orgs share a single instance.
    public static class RateLimiter {
        private int tokens;
        public RateLimiter(int tokens) { this.tokens = tokens; }
    }

    // Page is one decoded slice of the vendor events feed.
    public static class Page {
        public List<String> events;
        public String next_cursor;
    }

    private final String orgID;
    private final VendorGatewayClient gateway;
    private final DataSink sink;
    private final CursorStore cursorDB;
    private final RateLimiter rateLimit;     // shared across all orgs
    private final ExecutorService emitPool;  // unbounded, shared across all orgs

    public GithubBackfill(String orgID, VendorGatewayClient gateway, DataSink sink,
                          CursorStore cursorDB, RateLimiter rateLimit, ExecutorService emitPool) {
        this.orgID = orgID;
        this.gateway = gateway;
        this.sink = sink;
        this.cursorDB = cursorDB;
        this.rateLimit = rateLimit;
        this.emitPool = emitPool;
    }

    // run executes the backfill for this.orgID. It returns when the cursor
    // reaches the present (empty next_cursor) or the signal is aborted.
    public void run(AtomicBoolean signal) throws Exception {
        String cursor = cursorDB.load(orgID);

        List<Future<?>> pending = new ArrayList<>();
        for (;;) {
            if (signal.get()) {
                throw new RuntimeException("aborted");
            }

            Page page = fetchPage(cursor, signal);

            // Persist cursor first so a crash between fetch and emit does not
            // cause us to re-fetch the same page on restart.
            cursorDB.save(orgID, page.next_cursor);

            // Fan out emits per event for throughput. DataSink.emit is safe to
            // call concurrently and retries internally on transient failures.
            for (final String ev : page.events) {
                pending.add(emitPool.submit(() -> {
                    try {
                        sink.emit(ev, signal);
                    } catch (Exception err) {
                        // Best-effort; sink will retry.
                        return;
                    }
                }));
            }

            if (page.next_cursor.equals("")) {
                break;
            }
            cursor = page.next_cursor;
        }
    }

    private Page fetchPage(String cursor, AtomicBoolean signal) throws Exception {
        for (int attempt = 0; attempt < MAX_RETRIES; attempt++) {
            Response resp = gateway.get(
                "/orgs/" + orgID + "/events?cursor=" + cursor + "&limit=" + PAGE_SIZE,
                signal);

            if (resp.status() == 429) {
                // Vendor returns Retry-After in seconds per RFC 7231.
                int retryAfter = 0;
                try {
                    retryAfter = Integer.parseInt(resp.header("Retry-After"));
                } catch (NumberFormatException ignored) {
                }
                Thread.sleep(retryAfter);
                continue;
            }

            if (resp.status() >= 500) {
                Thread.sleep(attempt * 1000);
                continue;
            }

            InputStream body = resp.body();
            return MAPPER.readValue(body, Page.class);
        }
        throw new RuntimeException("exceeded max retries");
    }
}