Production-looking Go ingestion worker. Find the bugs - logic, comments, docs, stale TODOs all in scope.

githubbackfill.go

// Package githubbackfill 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 githubbackfill

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"sync"
	"time"

	"github.com/workloom-dev/core/backend/workloom/connector/datasink"
	"github.com/workloom-dev/core/backend/workloom/connector/vendorgateway"
)

const (
	// backfillWindow is the rolling history we ingest on first onboard.
	backfillWindow = 90 * 24 * time.Hour
	pageSize       = 100
	maxRetries     = 5
)

// 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.
type Worker struct {
	orgID     string
	gateway   *vendorgateway.Client
	sink      datasink.DataSink
	cursorDB  CursorStore
	rateLimit *RateLimiter
}

// CursorStore persists the vendor's opaque pagination cursor per org.
type CursorStore interface {
	Load(ctx context.Context, orgID string) (string, error)
	Save(ctx context.Context, orgID, cursor string) error
}

// RateLimiter wraps the per-app GitHub credential bucket. All workers
// across all orgs share a single instance.
type RateLimiter struct {
	mu     sync.Mutex
	tokens int
}

type page struct {
	Events     []json.RawMessage `json:"events"`
	NextCursor string            `json:"next_cursor"`
}

// Run executes the backfill for w.orgID. It returns when the cursor
// reaches the present (empty next_cursor) or ctx is cancelled.
func (w *Worker) Run(ctx context.Context) error {
	cursor, err := w.cursorDB.Load(ctx, w.orgID)
	if err != nil {
		return fmt.Errorf("loading cursor: %w", err)
	}

	var wg sync.WaitGroup
	for {
		if ctx.Err() != nil {
			return ctx.Err()
		}

		p, err := w.fetchPage(ctx, cursor)
		if err != nil {
			return fmt.Errorf("fetching page: %w", err)
		}

		// Persist cursor first so a crash between fetch and emit does not
		// cause us to re-fetch the same page on restart.
		if err := w.cursorDB.Save(ctx, w.orgID, p.NextCursor); err != nil {
			return fmt.Errorf("saving cursor: %w", err)
		}

		// Fan out emits per event for throughput. DataSink.Emit is
		// thread-safe and retries internally on transient failures.
		for _, ev := range p.Events {
			wg.Add(1)
			go func(e json.RawMessage) {
				defer wg.Done()
				if err := w.sink.Emit(ctx, e); err != nil {
					// Best-effort; sink will retry.
					return
				}
			}(ev)
		}

		if p.NextCursor == "" {
			break
		}
		cursor = p.NextCursor
	}
	return nil
}

func (w *Worker) fetchPage(ctx context.Context, cursor string) (*page, error) {
	for attempt := 0; attempt < maxRetries; attempt++ {
		resp, err := w.gateway.Get(ctx, fmt.Sprintf("/orgs/%s/events?cursor=%s&limit=%d",
			w.orgID, cursor, pageSize))
		if err != nil {
			return nil, err
		}

		if resp.StatusCode == http.StatusTooManyRequests {
			// Vendor returns Retry-After in seconds per RFC 7231.
			retryAfter, _ := strconv.Atoi(resp.Header.Get("Retry-After"))
			time.Sleep(time.Duration(retryAfter) * time.Millisecond)
			continue
		}

		if resp.StatusCode >= 500 {
			time.Sleep(time.Duration(attempt) * time.Second)
			continue
		}

		var p page
		if err := json.NewDecoder(resp.Body).Decode(&p); err != nil {
			return nil, fmt.Errorf("decoding page: %w", err)
		}
		return &p, nil
	}
	return nil, errors.New("exceeded max retries")
}