From 95053cdedec227cc12a9520f44602d1eea597661 Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Wed, 2 Sep 2026 13:19:43 -0300 Subject: [PATCH 1/7] feat(config): stream member last-activity as a usage event feed Adds an opt-in usage event feed that streams each org's audit-log activity (web/API actions, excluding raw git operations and bot actors) as incremental usage events, letting the platform derive last-activity for members without a per-user API field to sync directly. The config field is intentionally hidden from this connector's CLI/GUI since it only applies to GitHub Enterprise audit-log access; the enterprise connector variant sets it directly on the shared config struct. Co-Authored-By: Claude Sonnet 5 --- pkg/config/conf.gen.go | 1 + pkg/config/config.go | 13 ++ pkg/connector/connector.go | 14 +- pkg/connector/usage_event_feed.go | 250 +++++++++++++++++++++++++ pkg/connector/usage_event_feed_test.go | 213 +++++++++++++++++++++ 5 files changed, 490 insertions(+), 1 deletion(-) create mode 100644 pkg/connector/usage_event_feed.go create mode 100644 pkg/connector/usage_event_feed_test.go diff --git a/pkg/config/conf.gen.go b/pkg/config/conf.gen.go index 710e713a..6541f0f2 100644 --- a/pkg/config/conf.gen.go +++ b/pkg/config/conf.gen.go @@ -15,6 +15,7 @@ type Github struct { SyncSecrets bool `mapstructure:"sync-secrets"` OmitArchivedRepositories bool `mapstructure:"omit-archived-repositories"` DirectCollaboratorsOnly bool `mapstructure:"direct-collaborators-only"` + SyncLastActivity bool `mapstructure:"sync-last-activity"` } func (c *Github) findFieldByTag(tagValue string) (any, bool) { diff --git a/pkg/config/config.go b/pkg/config/config.go index 5c676fac..5a142d66 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -91,6 +91,18 @@ var ( field.WithDescription("Organization of your github app"), field.WithRequired(true), ) + + // syncLastActivity is hidden from this connector's GUI config and --help + // since it only applies to GitHub Enterprise audit-log access. + // baton-github-enterprise sets it directly on the shared Github struct, + // bypassing this CLI layer, so hiding it here doesn't affect that connector. + syncLastActivity = field.BoolField( + "sync-last-activity", + field.WithDisplayName("Sync users last activity"), + field.WithDescription("See when members were last active in your organizations."), + field.WithHidden(true), + field.WithExportTarget(field.ExportTargetCLIOnly), + ) ) //go:generate go run ./gen @@ -107,6 +119,7 @@ var Config = field.NewConfiguration( syncSecrets, omitArchivedRepositories, directCollaboratorsOnly, + syncLastActivity, }, field.WithConnectorDisplayName("GitHub v2"), field.WithHelpUrl("/docs/baton/github-v2"), diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index b38ed4e8..20a5f780 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -127,6 +127,7 @@ type GitHub struct { omitArchivedRepositories bool directCollaboratorsOnly bool enterprises []string + syncLastActivity bool } func (gh *GitHub) ResourceSyncers(ctx context.Context) []connectorbuilder.ResourceSyncerV2 { @@ -157,8 +158,18 @@ func (gh *GitHub) ResourceSyncers(ctx context.Context) []connectorbuilder.Resour return resourceSyncers } +func (gh *GitHub) EventFeeds(_ context.Context) []connectorbuilder.EventFeed { + if !gh.syncLastActivity { + return nil + } + + return []connectorbuilder.EventFeed{ + newUsageEventFeed(gh.client, gh.orgs), + } +} + // Metadata returns metadata about the connector. -func (gh *GitHub) Metadata(ctx context.Context) (*v2.ConnectorMetadata, error) { +func (gh *GitHub) Metadata(_ context.Context) (*v2.ConnectorMetadata, error) { return &v2.ConnectorMetadata{ DisplayName: "GitHub", AccountCreationSchema: &v2.ConnectorAccountCreationSchema{ @@ -346,6 +357,7 @@ func newWithGithubPAT(ctx context.Context, ghc *cfg.Github) (*GitHub, error) { syncSecrets: ghc.SyncSecrets, omitArchivedRepositories: ghc.OmitArchivedRepositories, directCollaboratorsOnly: ghc.DirectCollaboratorsOnly, + syncLastActivity: ghc.SyncLastActivity, }, nil } diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go new file mode 100644 index 00000000..23684f9d --- /dev/null +++ b/pkg/connector/usage_event_feed.go @@ -0,0 +1,250 @@ +package connector + +import ( + "context" + "encoding/base64" + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/annotations" + "github.com/conductorone/baton-sdk/pkg/pagination" + "github.com/google/go-github/v69/github" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" + "google.golang.org/protobuf/types/known/timestamppb" +) + +// defaultActivityLookback bounds the very first poll when no earliest-event +// boundary is given yet; later polls advance via the feed's own cursor. +const defaultActivityLookback = 1 * time.Hour + +// maxAuditLogPagesPerCall caps pages walked per org per call so one very +// active org can't stall the feed; remaining pages resume via the cursor. +const maxAuditLogPagesPerCall = 20 + +// usageEventFeed streams member activity from each org's audit log as usage +// events, since GitHub has no per-user "last activity" field to sync directly. +type usageEventFeed struct { + client *github.Client + orgs []string +} + +func newUsageEventFeed(client *github.Client, orgs []string) *usageEventFeed { + return &usageEventFeed{client: client, orgs: orgs} +} + +func (f *usageEventFeed) EventFeedMetadata(_ context.Context) *v2.EventFeedMetadata { + return &v2.EventFeedMetadata{ + Id: "github_usage_event_feed", + SupportedEventTypes: []v2.EventType{v2.EventType_EVENT_TYPE_USAGE}, + } +} + +// usageEventPageToken tracks progress through one pass over every configured +// org's audit log, walked newest-first until an entry at or before Since is +// reached (already seen in a previous pass). +type usageEventPageToken struct { + Orgs []string `json:"orgs,omitempty"` + OrgIndex int `json:"org_index"` + AuditLogCursor string `json:"audit_log_cursor,omitempty"` + Since string `json:"since,omitempty"` +} + +func unmarshalUsageEventPageToken(pToken *pagination.StreamToken) (*usageEventPageToken, error) { + pt := &usageEventPageToken{} + if pToken == nil || pToken.Cursor == "" { + return pt, nil + } + data, err := base64.StdEncoding.DecodeString(pToken.Cursor) + if err != nil { + return nil, fmt.Errorf("baton-github: failed to decode usage event feed cursor: %w", err) + } + if err := json.Unmarshal(data, pt); err != nil { + return nil, fmt.Errorf("baton-github: failed to unmarshal usage event feed cursor: %w", err) + } + return pt, nil +} + +func (pt *usageEventPageToken) marshal() (string, error) { + data, err := json.Marshal(pt) + if err != nil { + return "", fmt.Errorf("baton-github: failed to marshal usage event feed cursor: %w", err) + } + return base64.StdEncoding.EncodeToString(data), nil +} + +func (f *usageEventFeed) ListEvents( + ctx context.Context, + earliestEvent *timestamppb.Timestamp, + pToken *pagination.StreamToken, +) ([]*v2.Event, *pagination.StreamState, annotations.Annotations, error) { + l := ctxzap.Extract(ctx) + + if f.client == nil { + return nil, &pagination.StreamState{HasMore: false}, nil, nil + } + + cursor, err := unmarshalUsageEventPageToken(pToken) + if err != nil { + return nil, nil, nil, err + } + + if len(cursor.Orgs) == 0 { + // Snapshot the org list and "since" boundary once per pass, so + // mid-pass config changes don't shift what gets walked. + orgs, err := getOrgs(ctx, f.client, f.orgs) + if err != nil { + return nil, nil, nil, fmt.Errorf("baton-github: failed to list orgs for usage event feed: %w", err) + } + if len(orgs) == 0 { + return nil, &pagination.StreamState{HasMore: false}, nil, nil + } + + since := time.Now().Add(-defaultActivityLookback) + // Guard against a zero/degenerate earliestEvent producing a + // nonsensical "since year 1" query that GitHub's search parser rejects. + if earliestEvent != nil { + if t := earliestEvent.AsTime(); !t.IsZero() && t.After(time.Unix(0, 0)) { + since = t + } + } + + cursor = &usageEventPageToken{ + Orgs: orgs, + Since: since.Format(time.RFC3339), + } + } + + since, err := time.Parse(time.RFC3339, cursor.Since) + if err != nil { + return nil, nil, nil, fmt.Errorf("baton-github: invalid usage event feed cursor timestamp: %w", err) + } + // created:>= is sent server-side so GitHub excludes already-seen + // entries; the check below stays as a safety net in case it's ignored. + sincePhrase := "created:>=" + since.UTC().Format("2006-01-02T15:04:05-07:00") + + var events []*v2.Event + + //TODO(jdc): Probably change this for loop for a series of requests that uses a more complex pagination cursor. + for page := 0; page < maxAuditLogPagesPerCall; page++ { + orgName := cursor.Orgs[cursor.OrgIndex] + + opts := &github.GetAuditLogOptions{ + Order: github.Ptr("desc"), + // "web" excludes raw git-protocol events (push/fetch/clone), + // which dominate audit-log volume without losing members who are + // otherwise covered by their web/API activity. + Include: github.Ptr("web"), + Phrase: github.Ptr(sincePhrase), + ListCursorOptions: github.ListCursorOptions{ + PerPage: maxPageSize, + Page: cursor.AuditLogCursor, + }, + } + + entries, resp, err := f.client.Organizations.GetAuditLog(ctx, orgName, opts) + if err != nil { + l.Debug("failed to fetch audit log for org, skipping it for this pass", + zap.String("org", orgName), zap.Error(err)) + entries = nil + resp = nil + } + + exhausted := true + for _, entry := range entries { + evt, ts, ok := usageEventFromAuditEntry(orgName, entry) + if !ok { + continue + } + + if !ts.After(since) { + // Descending order: everything after this entry is even + // older, so this org is done for this pass. + // This is an extra safeguard since the server should have filter these already. + break + } + events = append(events, evt) + exhausted = false + } + + if resp != nil && resp.NextPageToken != "" && !exhausted { + cursor.AuditLogCursor = resp.NextPageToken + continue + } + + // Done with this org for this pass - advance to the next one. + cursor.OrgIndex++ + cursor.AuditLogCursor = "" + if cursor.OrgIndex >= len(cursor.Orgs) { + // Pass complete - the next call gets a fresh earliestEvent, so + // nothing needs to survive in the cursor. + tokenStr, err := (&usageEventPageToken{}).marshal() + if err != nil { + return nil, nil, nil, err + } + return events, &pagination.StreamState{Cursor: tokenStr, HasMore: false}, nil, nil + } + } + + tokenStr, err := cursor.marshal() + if err != nil { + return nil, nil, nil, err + } + return events, &pagination.StreamState{Cursor: tokenStr, HasMore: true}, nil, nil +} + +// usageEventFromAuditEntry converts one audit-log entry into a usage event +// tying the actor to the org they acted in. Returns ok=false when the entry +// can't be attributed to a synced user resource. +func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Event, time.Time, bool) { + actor := entry.GetActor() + actorID := entry.GetActorID() + ts := entry.GetTimestamp().Time + if actorID == 0 || ts.IsZero() { + return nil, time.Time{}, false + } + + // actor_is_bot is real but undocumented, so it only surfaces via + // AdditionalFields; trust it when present, else fall back to the + // "[bot]" login suffix. Either way, bot/App actors aren't synced as user + // resources, so an event attributed to one wouldn't correlate to anything. + if isBot, ok := entry.AdditionalFields["actor_is_bot"].(bool); ok { + if isBot { + return nil, time.Time{}, false + } + } else if strings.HasSuffix(actor, "[bot]") { + return nil, time.Time{}, false + } + + orgID := entry.GetOrgID() + if orgID == 0 { + return nil, time.Time{}, false + } + + return &v2.Event{ + Id: entry.GetDocumentID(), + OccurredAt: timestamppb.New(ts), + Event: &v2.Event_UsageEvent{ + UsageEvent: &v2.UsageEvent{ + TargetResource: &v2.Resource{ + Id: &v2.ResourceId{ + ResourceType: resourceTypeOrg.Id, + Resource: strconv.FormatInt(orgID, 10), + }, + DisplayName: orgName, + }, + ActorResource: &v2.Resource{ + Id: &v2.ResourceId{ + ResourceType: resourceTypeUser.Id, + Resource: strconv.FormatInt(actorID, 10), + }, + DisplayName: actor, + }, + }, + }, + }, ts, true +} diff --git a/pkg/connector/usage_event_feed_test.go b/pkg/connector/usage_event_feed_test.go new file mode 100644 index 00000000..c49cd0f3 --- /dev/null +++ b/pkg/connector/usage_event_feed_test.go @@ -0,0 +1,213 @@ +package connector + +import ( + "context" + "net/http" + "testing" + "time" + + "github.com/conductorone/baton-sdk/pkg/pagination" + "github.com/google/go-github/v69/github" + "github.com/migueleliasweb/go-github-mock/src/mock" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestUsageEventFromAuditEntry(t *testing.T) { + ts := time.Date(2024, 6, 1, 0, 0, 0, 0, time.UTC) + + tests := []struct { + name string + entry *github.AuditEntry + ok bool + }{ + { + name: "valid entry", + entry: &github.AuditEntry{ + Actor: github.Ptr("octocat"), + ActorID: github.Ptr(int64(123)), + OrgID: github.Ptr(int64(456)), + Timestamp: &github.Timestamp{Time: ts}, + }, + ok: true, + }, + { + name: "missing actor id", + entry: &github.AuditEntry{ + Actor: github.Ptr("octocat"), + OrgID: github.Ptr(int64(456)), + Timestamp: &github.Timestamp{Time: ts}, + }, + ok: false, + }, + { + name: "missing org id", + entry: &github.AuditEntry{ + Actor: github.Ptr("octocat"), + ActorID: github.Ptr(int64(123)), + Timestamp: &github.Timestamp{Time: ts}, + }, + ok: false, + }, + { + name: "missing timestamp", + entry: &github.AuditEntry{ + Actor: github.Ptr("octocat"), + ActorID: github.Ptr(int64(123)), + OrgID: github.Ptr(int64(456)), + }, + ok: false, + }, + { + name: "bot actor by login suffix", + entry: &github.AuditEntry{ + Actor: github.Ptr("dependabot[bot]"), + ActorID: github.Ptr(int64(123)), + OrgID: github.Ptr(int64(456)), + Timestamp: &github.Timestamp{Time: ts}, + }, + ok: false, + }, + { + name: "bot actor by actor_is_bot field", + entry: &github.AuditEntry{ + Actor: github.Ptr("some-app-installation"), + ActorID: github.Ptr(int64(123)), + OrgID: github.Ptr(int64(456)), + Timestamp: &github.Timestamp{Time: ts}, + AdditionalFields: map[string]interface{}{"actor_is_bot": true}, + }, + ok: false, + }, + { + name: "actor_is_bot false overrides a non-matching suffix check", + entry: &github.AuditEntry{ + Actor: github.Ptr("octocat"), + ActorID: github.Ptr(int64(123)), + OrgID: github.Ptr(int64(456)), + Timestamp: &github.Timestamp{Time: ts}, + AdditionalFields: map[string]interface{}{"actor_is_bot": false}, + }, + ok: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + evt, evtTs, ok := usageEventFromAuditEntry("octo-org", tt.entry) + require.Equal(t, tt.ok, ok) + if !tt.ok { + return + } + require.Equal(t, ts, evtTs) + require.Equal(t, "123", evt.GetUsageEvent().GetActorResource().GetId().GetResource()) + require.Equal(t, "456", evt.GetUsageEvent().GetTargetResource().GetId().GetResource()) + require.Equal(t, resourceTypeUser.Id, evt.GetUsageEvent().GetActorResource().GetId().GetResourceType()) + require.Equal(t, resourceTypeOrg.Id, evt.GetUsageEvent().GetTargetResource().GetId().GetResourceType()) + }) + } +} + +func TestUsageEventFeed_ListEvents_GracefulDegradation(t *testing.T) { + ctx := context.Background() + + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusForbidden) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, nil, nil) + require.NoError(t, err) + require.Empty(t, events) + require.False(t, state.HasMore) +} + +func TestUsageEventFeed_ListEvents_FiltersToSinceBoundary(t *testing.T) { + ctx := context.Background() + + since := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + newer1 := since.Add(2 * time.Hour) + newer2 := since.Add(1 * time.Hour) + older := since.Add(-1 * time.Hour) + + // Entries in descending order, as requested (Order: "desc"). + entries := []*github.AuditEntry{ + {Actor: github.Ptr("alice"), ActorID: github.Ptr(int64(1)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer1}}, + {Actor: github.Ptr("bob"), ActorID: github.Ptr(int64(2)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer2}}, + {Actor: github.Ptr("carol"), ActorID: github.Ptr(int64(3)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: older}}, + } + + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatch(mock.GetOrgsAuditLogByOrg, entries), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, timestamppb.New(since), nil) + require.NoError(t, err) + require.Len(t, events, 2) + require.False(t, state.HasMore) +} + +func TestUsageEventFeed_ListEvents_ZeroEarliestEventFallsBackToDefaultLookback(t *testing.T) { + ctx := context.Background() + + var gotPhrase string + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPhrase = r.URL.Query().Get("phrase") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte("[]")) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + // A zero timestamppb.Timestamp mirrors what an unset/degenerate + // caller-supplied start-at looks like after round-tripping through + // timestamppb - it must not be trusted as a real boundary. + events, state, _, err := f.ListEvents(ctx, ×tamppb.Timestamp{}, nil) + require.NoError(t, err) + require.Empty(t, events) + require.False(t, state.HasMore) + require.NotContains(t, gotPhrase, "0001-01-01") + require.Contains(t, gotPhrase, "created:>=") +} + +func TestUsageEventFeed_ListEvents_NoOrgs(t *testing.T) { + ctx := context.Background() + + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{}), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, nil, nil) + require.NoError(t, err) + require.Empty(t, events) + require.False(t, state.HasMore) +} + +func TestUsageEventFeed_ListEvents_NilClient(t *testing.T) { + ctx := context.Background() + + f := newUsageEventFeed(nil, nil) + + events, state, _, err := f.ListEvents(ctx, nil, &pagination.StreamToken{}) + require.NoError(t, err) + require.Empty(t, events) + require.False(t, state.HasMore) +} From fc29653e08d22cef09bd869307cd81f9bb2b0f77 Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Thu, 3 Sep 2026 00:06:57 -0300 Subject: [PATCH 2/7] fix(connector): wire syncLastActivity through the GitHub App auth path newWithGithubApp built its GitHub struct without copying SyncLastActivity, so EventFeeds() always returned nil for App-authenticated connectors regardless of the config value, with no error or log. Co-Authored-By: Claude Sonnet 5 --- pkg/connector/connector.go | 1 + pkg/connector/usage_event_feed.go | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 20a5f780..72f0bf07 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -464,6 +464,7 @@ func newWithGithubApp(ctx context.Context, ghc *cfg.Github) (*GitHub, error) { syncSecrets: ghc.SyncSecrets, omitArchivedRepositories: ghc.OmitArchivedRepositories, directCollaboratorsOnly: ghc.DirectCollaboratorsOnly, + syncLastActivity: ghc.SyncLastActivity, } return gh, nil } diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 23684f9d..688c74b7 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -129,7 +129,7 @@ func (f *usageEventFeed) ListEvents( var events []*v2.Event - //TODO(jdc): Probably change this for loop for a series of requests that uses a more complex pagination cursor. + // TODO(jdc): Probably change this for loop for a series of requests that uses a more complex pagination cursor. for page := 0; page < maxAuditLogPagesPerCall; page++ { orgName := cursor.Orgs[cursor.OrgIndex] From a16e4cb1a93c68a3324d80a2a57e5d68575ab1a3 Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Thu, 3 Sep 2026 01:05:29 -0300 Subject: [PATCH 3/7] fix(connector): fix pagination, surface rate limits, harden event ids - Track the since-boundary explicitly instead of inferring it from whether an event was emitted, so a page of only filtered-out entries (bots, missing IDs) no longer ends an org's walk early and silently drops later pages of real activity. - Surface the tightest rate limit seen across a call's audit-log requests as an annotation, including on error responses, so the SDK can pace polling instead of hitting 429s. - Fall back to a synthesized event id (org/actor/timestamp/action) when GitHub omits _document_id, avoiding ambiguous dedup on empty ids. Co-Authored-By: Claude Sonnet 5 --- pkg/connector/usage_event_feed.go | 50 ++++-- pkg/connector/usage_event_feed_test.go | 224 ++++++++++++++++++++++++- 2 files changed, 258 insertions(+), 16 deletions(-) diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 688c74b7..1472f1bf 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -128,6 +128,8 @@ func (f *usageEventFeed) ListEvents( sincePhrase := "created:>=" + since.UTC().Format("2006-01-02T15:04:05-07:00") var events []*v2.Event + // Tightest (lowest Remaining) rate limit seen across this call's requests. + var tightestRateLimit *v2.RateLimitDescription // TODO(jdc): Probably change this for loop for a series of requests that uses a more complex pagination cursor. for page := 0; page < maxAuditLogPagesPerCall; page++ { @@ -147,6 +149,15 @@ func (f *usageEventFeed) ListEvents( } entries, resp, err := f.client.Organizations.GetAuditLog(ctx, orgName, opts) + // Read rate-limit headers before the error branch nils resp, since a + // 429 still carries them. + if resp != nil { + if rl, rlErr := extractRateLimitData(resp); rlErr == nil { + if tightestRateLimit == nil || rl.GetRemaining() < tightestRateLimit.GetRemaining() { + tightestRateLimit = rl + } + } + } if err != nil { l.Debug("failed to fetch audit log for org, skipping it for this pass", zap.String("org", orgName), zap.Error(err)) @@ -154,7 +165,7 @@ func (f *usageEventFeed) ListEvents( resp = nil } - exhausted := true + reachedBoundary := false for _, entry := range entries { evt, ts, ok := usageEventFromAuditEntry(orgName, entry) if !ok { @@ -162,16 +173,15 @@ func (f *usageEventFeed) ListEvents( } if !ts.After(since) { - // Descending order: everything after this entry is even - // older, so this org is done for this pass. - // This is an extra safeguard since the server should have filter these already. + // Descending order, so everything after this is even older; + // a safety net in case the server-side phrase filter missed it. + reachedBoundary = true break } events = append(events, evt) - exhausted = false } - if resp != nil && resp.NextPageToken != "" && !exhausted { + if resp != nil && resp.NextPageToken != "" && !reachedBoundary { cursor.AuditLogCursor = resp.NextPageToken continue } @@ -186,7 +196,11 @@ func (f *usageEventFeed) ListEvents( if err != nil { return nil, nil, nil, err } - return events, &pagination.StreamState{Cursor: tokenStr, HasMore: false}, nil, nil + var annos annotations.Annotations + if tightestRateLimit != nil { + annos.WithRateLimiting(tightestRateLimit) + } + return events, &pagination.StreamState{Cursor: tokenStr, HasMore: false}, annos, nil } } @@ -194,7 +208,11 @@ func (f *usageEventFeed) ListEvents( if err != nil { return nil, nil, nil, err } - return events, &pagination.StreamState{Cursor: tokenStr, HasMore: true}, nil, nil + var annos annotations.Annotations + if tightestRateLimit != nil { + annos.WithRateLimiting(tightestRateLimit) + } + return events, &pagination.StreamState{Cursor: tokenStr, HasMore: true}, annos, nil } // usageEventFromAuditEntry converts one audit-log entry into a usage event @@ -208,10 +226,9 @@ func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Eve return nil, time.Time{}, false } - // actor_is_bot is real but undocumented, so it only surfaces via - // AdditionalFields; trust it when present, else fall back to the - // "[bot]" login suffix. Either way, bot/App actors aren't synced as user - // resources, so an event attributed to one wouldn't correlate to anything. + // actor_is_bot is real but undocumented (only in AdditionalFields); trust + // it when present, else fall back to the "[bot]" login suffix. Bots + // aren't synced as users, so their events wouldn't correlate to anything. if isBot, ok := entry.AdditionalFields["actor_is_bot"].(bool); ok { if isBot { return nil, time.Time{}, false @@ -225,8 +242,15 @@ func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Eve return nil, time.Time{}, false } + id := entry.GetDocumentID() + if id == "" { + // No stable ID from GitHub - synthesize one so dedup doesn't collapse + // every entry missing _document_id into one event. + id = fmt.Sprintf("%d:%d:%d:%s", orgID, actorID, ts.Unix(), entry.GetAction()) + } + return &v2.Event{ - Id: entry.GetDocumentID(), + Id: id, OccurredAt: timestamppb.New(ts), Event: &v2.Event_UsageEvent{ UsageEvent: &v2.UsageEvent{ diff --git a/pkg/connector/usage_event_feed_test.go b/pkg/connector/usage_event_feed_test.go index c49cd0f3..3c5cd6fb 100644 --- a/pkg/connector/usage_event_feed_test.go +++ b/pkg/connector/usage_event_feed_test.go @@ -2,10 +2,13 @@ package connector import ( "context" + "fmt" "net/http" + "strings" "testing" "time" + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/pagination" "github.com/google/go-github/v69/github" "github.com/migueleliasweb/go-github-mock/src/mock" @@ -108,6 +111,38 @@ func TestUsageEventFromAuditEntry(t *testing.T) { } } +func TestUsageEventFromAuditEntry_IdFallback(t *testing.T) { + ts := time.Date(2024, 6, 1, 12, 0, 0, 0, time.UTC) + + t.Run("uses the real document id when present", func(t *testing.T) { + entry := &github.AuditEntry{ + Actor: github.Ptr("octocat"), + ActorID: github.Ptr(int64(123)), + OrgID: github.Ptr(int64(456)), + Timestamp: &github.Timestamp{Time: ts}, + DocumentID: github.Ptr("real-doc-id"), + Action: github.Ptr("repo.create"), + } + evt, _, ok := usageEventFromAuditEntry("octo-org", entry) + require.True(t, ok) + require.Equal(t, "real-doc-id", evt.GetId()) + }) + + t.Run("synthesizes a stable id when the document id is missing", func(t *testing.T) { + entry := &github.AuditEntry{ + Actor: github.Ptr("octocat"), + ActorID: github.Ptr(int64(123)), + OrgID: github.Ptr(int64(456)), + Timestamp: &github.Timestamp{Time: ts}, + Action: github.Ptr("repo.create"), + } + evt, _, ok := usageEventFromAuditEntry("octo-org", entry) + require.True(t, ok) + require.Equal(t, fmt.Sprintf("456:123:%d:repo.create", ts.Unix()), evt.GetId()) + require.NotEmpty(t, evt.GetId()) + }) +} + func TestUsageEventFeed_ListEvents_GracefulDegradation(t *testing.T) { ctx := context.Background() @@ -175,9 +210,8 @@ func TestUsageEventFeed_ListEvents_ZeroEarliestEventFallsBackToDefaultLookback(t f := newUsageEventFeed(github.NewClient(httpClient), nil) - // A zero timestamppb.Timestamp mirrors what an unset/degenerate - // caller-supplied start-at looks like after round-tripping through - // timestamppb - it must not be trusted as a real boundary. + // A zero timestamppb.Timestamp mirrors a degenerate caller-supplied + // start-at, which must not be trusted as a real boundary. events, state, _, err := f.ListEvents(ctx, ×tamppb.Timestamp{}, nil) require.NoError(t, err) require.Empty(t, events) @@ -186,6 +220,190 @@ func TestUsageEventFeed_ListEvents_ZeroEarliestEventFallsBackToDefaultLookback(t require.Contains(t, gotPhrase, "created:>=") } +func TestUsageEventFeed_ListEvents_ContinuesPastAnAllFilteredPage(t *testing.T) { + ctx := context.Background() + + since := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + newer := since.Add(1 * time.Hour) + + // Page 1 is entirely bot activity - every entry gets filtered out by + // usageEventFromAuditEntry, so no event is ever appended on this page. + page1 := []*github.AuditEntry{ + {Actor: github.Ptr("dependabot[bot]"), ActorID: github.Ptr(int64(1)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer}}, + } + page2 := []*github.AuditEntry{ + {Actor: github.Ptr("octocat"), ActorID: github.Ptr(int64(2)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer}}, + } + + calls := 0 + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.Header().Set("Content-Type", "application/json") + if calls == 1 { + w.Header().Set("Link", `; rel="next"`) + _, _ = w.Write(mock.MustMarshal(page1)) + return + } + _, _ = w.Write(mock.MustMarshal(page2)) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, timestamppb.New(since), nil) + require.NoError(t, err) + require.Equal(t, 2, calls, "an all-filtered page must not be mistaken for the since boundary") + require.Len(t, events, 1) + require.Equal(t, "2", events[0].GetUsageEvent().GetActorResource().GetId().GetResource()) + require.False(t, state.HasMore) +} + +func TestUsageEventFeed_ListEvents_ReturnsTightestRateLimit(t *testing.T) { + ctx := context.Background() + + since := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + newer := since.Add(1 * time.Hour) + + page1 := []*github.AuditEntry{ + {Actor: github.Ptr("octocat"), ActorID: github.Ptr(int64(1)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer}}, + } + page2 := []*github.AuditEntry{ + {Actor: github.Ptr("alice"), ActorID: github.Ptr(int64(2)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer}}, + } + + calls := 0 + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.Header().Set("Content-Type", "application/json") + w.Header().Set("X-Ratelimit-Limit", "1750") + if calls == 1 { + // First call reports plenty of budget left. + w.Header().Set("X-Ratelimit-Remaining", "500") + w.Header().Set("Link", `; rel="next"`) + _, _ = w.Write(mock.MustMarshal(page1)) + return + } + // Second call is the tighter one - this is the value that + // should win. + w.Header().Set("X-Ratelimit-Remaining", "10") + _, _ = w.Write(mock.MustMarshal(page2)) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, _, annos, err := f.ListEvents(ctx, timestamppb.New(since), nil) + require.NoError(t, err) + require.Equal(t, 2, calls) + require.Len(t, events, 2) + + var rl v2.RateLimitDescription + found, err := annos.Pick(&rl) + require.NoError(t, err) + require.True(t, found, "expected a rate-limit annotation to be returned") + require.Equal(t, int64(10), rl.GetRemaining()) + require.Equal(t, int64(1750), rl.GetLimit()) +} + +func TestUsageEventFeed_ListEvents_AdvancesAcrossMultipleOrgs(t *testing.T) { + ctx := context.Background() + + since := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + newer := since.Add(1 * time.Hour) + + entriesByOrg := map[string][]*github.AuditEntry{ + "octo-org-a": { + {Actor: github.Ptr("alice"), ActorID: github.Ptr(int64(1)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer}}, + }, + "octo-org-b": { + {Actor: github.Ptr("bob"), ActorID: github.Ptr(int64(2)), OrgID: github.Ptr(int64(8)), Timestamp: &github.Timestamp{Time: newer}}, + }, + } + + var seenOrgs []string + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{ + {Login: github.Ptr("octo-org-a")}, + {Login: github.Ptr("octo-org-b")}, + }), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + org := strings.Split(r.URL.Path, "/")[2] + seenOrgs = append(seenOrgs, org) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(mock.MustMarshal(entriesByOrg[org])) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, timestamppb.New(since), nil) + require.NoError(t, err) + require.False(t, state.HasMore) + require.Equal(t, []string{"octo-org-a", "octo-org-b"}, seenOrgs, "should walk both orgs, in order, within one call") + require.Len(t, events, 2) + + actorIDs := []string{ + events[0].GetUsageEvent().GetActorResource().GetId().GetResource(), + events[1].GetUsageEvent().GetActorResource().GetId().GetResource(), + } + require.ElementsMatch(t, []string{"1", "2"}, actorIDs) +} + +func TestUsageEventFeed_ListEvents_ResumesFromPersistedCursor(t *testing.T) { + ctx := context.Background() + + since := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + newer := since.Add(1 * time.Hour) + + // Simulate a previous call that already finished "octo-org-a" and was + // mid-page through "octo-org-b" with its own audit-log cursor. + resumeToken := &usageEventPageToken{ + Orgs: []string{"octo-org-a", "octo-org-b"}, + OrgIndex: 1, + AuditLogCursor: "existing-cursor", + Since: since.Format(time.RFC3339), + } + cursorStr, err := resumeToken.marshal() + require.NoError(t, err) + + var gotOrg, gotPage string + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotOrg = strings.Split(r.URL.Path, "/")[2] + gotPage = r.URL.Query().Get("page") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(mock.MustMarshal([]*github.AuditEntry{ + {Actor: github.Ptr("bob"), ActorID: github.Ptr(int64(2)), OrgID: github.Ptr(int64(8)), Timestamp: &github.Timestamp{Time: newer}}, + })) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, nil, &pagination.StreamToken{Cursor: cursorStr}) + require.NoError(t, err) + require.Equal(t, "octo-org-b", gotOrg, "should resume at the persisted org, not restart from octo-org-a") + require.Equal(t, "existing-cursor", gotPage, "should resume with the persisted audit-log cursor") + require.Len(t, events, 1) + require.False(t, state.HasMore) +} + func TestUsageEventFeed_ListEvents_NoOrgs(t *testing.T) { ctx := context.Background() From efd6e70149600ae41e472cb52be6e19520ffe8bc Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Thu, 3 Sep 2026 02:00:30 -0300 Subject: [PATCH 4/7] fix(connector): classify audit-log errors instead of swallowing all of them Skip-and-continue is now restricted to permanent per-org conditions (403/404), logged at Warn per this repo's log-level convention. Rate limits, 5xx, and any other error now abort the call instead of silently completing the pass and permanently losing the unfetched activity window on the next poll. Co-Authored-By: Claude Sonnet 5 --- pkg/connector/usage_event_feed.go | 26 +++++++++-- pkg/connector/usage_event_feed_test.go | 64 ++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 4 deletions(-) diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 1472f1bf..d4bc1186 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -4,6 +4,7 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "fmt" "strconv" "strings" @@ -159,10 +160,27 @@ func (f *usageEventFeed) ListEvents( } } if err != nil { - l.Debug("failed to fetch audit log for org, skipping it for this pass", - zap.String("org", orgName), zap.Error(err)) - entries = nil - resp = nil + // Skip-and-continue only for permanent per-org conditions (no + // audit-log access); anything else aborts instead of wasting the + // rest of the page budget. Rate-limit checks come first since + // GitHub can signal rate limiting via a 403. + var rateLimitErr *github.RateLimitError + var abuseRateLimitErr *github.AbuseRateLimitError + retryable := errors.As(err, &rateLimitErr) || errors.As(err, &abuseRateLimitErr) || + isRatelimited(resp) || isTemporarilyUnavailable(resp) + + switch { + case retryable: + return nil, nil, nil, wrapGitHubError(err, resp, + fmt.Sprintf("baton-github: failed to fetch audit log for org %s", orgName)) + case isNotFoundError(resp) || isPermissionError(resp): + l.Warn("org lacks audit-log access, skipping it for this pass", + zap.String("org", orgName), zap.Error(err)) + entries, resp = nil, nil + default: + return nil, nil, nil, wrapGitHubError(err, resp, + fmt.Sprintf("baton-github: failed to fetch audit log for org %s", orgName)) + } } reachedBoundary := false diff --git a/pkg/connector/usage_event_feed_test.go b/pkg/connector/usage_event_feed_test.go index 3c5cd6fb..a15b0245 100644 --- a/pkg/connector/usage_event_feed_test.go +++ b/pkg/connector/usage_event_feed_test.go @@ -164,6 +164,70 @@ func TestUsageEventFeed_ListEvents_GracefulDegradation(t *testing.T) { require.False(t, state.HasMore) } +func TestUsageEventFeed_ListEvents_SkipsOrgOn404(t *testing.T) { + ctx := context.Background() + + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, nil, nil) + require.NoError(t, err) + require.Empty(t, events) + require.False(t, state.HasMore) +} + +func TestUsageEventFeed_ListEvents_AbortsOnServerError(t *testing.T) { + ctx := context.Background() + + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, nil, nil) + require.Error(t, err, "a 5xx should abort the call, not be swallowed as a successful empty pass") + require.Nil(t, events) + require.Nil(t, state) +} + +func TestUsageEventFeed_ListEvents_AbortsOnRateLimit(t *testing.T) { + ctx := context.Background() + + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatch(mock.GetUserOrgs, []*github.Organization{{Login: github.Ptr("octo-org")}}), + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Ratelimit-Remaining", "0") + w.WriteHeader(http.StatusForbidden) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + events, state, _, err := f.ListEvents(ctx, nil, nil) + require.Error(t, err, "a rate-limited 403 (Remaining: 0) should abort, not be treated as a permission error") + require.Nil(t, events) + require.Nil(t, state) +} + func TestUsageEventFeed_ListEvents_FiltersToSinceBoundary(t *testing.T) { ctx := context.Background() From e4611ac81ffa8e3d5927c6987e41ee9b4259b70a Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Thu, 3 Sep 2026 02:44:22 -0300 Subject: [PATCH 5/7] fix(connector): fix usage event id collisions and boundary detection Use nanosecond precision for the synthesized event id and the persisted "since" cursor to avoid same-second id collisions and duplicate re-emitted events. Also check every audit entry's raw timestamp against the boundary, not just ones that pass the bot filter, so an all-filtered page doesn't stall pagination past "since". Co-Authored-By: Claude Sonnet 5 --- pkg/connector/usage_event_feed.go | 21 ++++++++++----------- pkg/connector/usage_event_feed_test.go | 2 +- 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index d4bc1186..4c73cfdd 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -116,11 +116,11 @@ func (f *usageEventFeed) ListEvents( cursor = &usageEventPageToken{ Orgs: orgs, - Since: since.Format(time.RFC3339), + Since: since.Format(time.RFC3339Nano), } } - since, err := time.Parse(time.RFC3339, cursor.Since) + since, err := time.Parse(time.RFC3339Nano, cursor.Since) if err != nil { return nil, nil, nil, fmt.Errorf("baton-github: invalid usage event feed cursor timestamp: %w", err) } @@ -185,17 +185,16 @@ func (f *usageEventFeed) ListEvents( reachedBoundary := false for _, entry := range entries { - evt, ts, ok := usageEventFromAuditEntry(orgName, entry) - if !ok { - continue - } - - if !ts.After(since) { - // Descending order, so everything after this is even older; - // a safety net in case the server-side phrase filter missed it. + // Check every entry's timestamp, even filtered ones, so an all-bot page still stops pagination. + if ts := entry.GetTimestamp().Time; !ts.IsZero() && !ts.After(since) { reachedBoundary = true break } + + evt, _, ok := usageEventFromAuditEntry(orgName, entry) + if !ok { + continue + } events = append(events, evt) } @@ -264,7 +263,7 @@ func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Eve if id == "" { // No stable ID from GitHub - synthesize one so dedup doesn't collapse // every entry missing _document_id into one event. - id = fmt.Sprintf("%d:%d:%d:%s", orgID, actorID, ts.Unix(), entry.GetAction()) + id = fmt.Sprintf("%d:%d:%d:%s", orgID, actorID, ts.UnixNano(), entry.GetAction()) } return &v2.Event{ diff --git a/pkg/connector/usage_event_feed_test.go b/pkg/connector/usage_event_feed_test.go index a15b0245..fdf80520 100644 --- a/pkg/connector/usage_event_feed_test.go +++ b/pkg/connector/usage_event_feed_test.go @@ -138,7 +138,7 @@ func TestUsageEventFromAuditEntry_IdFallback(t *testing.T) { } evt, _, ok := usageEventFromAuditEntry("octo-org", entry) require.True(t, ok) - require.Equal(t, fmt.Sprintf("456:123:%d:repo.create", ts.Unix()), evt.GetId()) + require.Equal(t, fmt.Sprintf("456:123:%d:repo.create", ts.UnixNano()), evt.GetId()) require.NotEmpty(t, evt.GetId()) }) } From e0fcb758c18a00ed36d988427279173b2b9b483c Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Thu, 3 Sep 2026 02:53:43 -0300 Subject: [PATCH 6/7] docs(connector): correct maxAuditLogPagesPerCall comment scope The page budget is a single counter shared across every org processed in a call, not a per-org cap as the comment claimed. Co-Authored-By: Claude Sonnet 5 --- pkg/connector/usage_event_feed.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 4c73cfdd..013be931 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -23,8 +23,8 @@ import ( // boundary is given yet; later polls advance via the feed's own cursor. const defaultActivityLookback = 1 * time.Hour -// maxAuditLogPagesPerCall caps pages walked per org per call so one very -// active org can't stall the feed; remaining pages resume via the cursor. +// maxAuditLogPagesPerCall caps total pages walked across all orgs per call +// so one very active org can't stall the feed; remaining pages resume via the cursor. const maxAuditLogPagesPerCall = 20 // usageEventFeed streams member activity from each org's audit log as usage From 348ca528184fbf179e172e7f0bccc70e3cfd533c Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Thu, 3 Sep 2026 03:13:11 -0300 Subject: [PATCH 7/7] chore: address PR comments suggestions --- pkg/connector/usage_event_feed.go | 19 ++++++---- pkg/connector/usage_event_feed_test.go | 49 +++++++++++++++++++++++--- 2 files changed, 57 insertions(+), 11 deletions(-) diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 013be931..efd24be5 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -120,6 +120,11 @@ func (f *usageEventFeed) ListEvents( } } + if cursor.OrgIndex < 0 || cursor.OrgIndex >= len(cursor.Orgs) { + cursor.OrgIndex = 0 + cursor.AuditLogCursor = "" + } + since, err := time.Parse(time.RFC3339Nano, cursor.Since) if err != nil { return nil, nil, nil, fmt.Errorf("baton-github: invalid usage event feed cursor timestamp: %w", err) @@ -191,7 +196,7 @@ func (f *usageEventFeed) ListEvents( break } - evt, _, ok := usageEventFromAuditEntry(orgName, entry) + evt, ok := usageEventFromAuditEntry(orgName, entry) if !ok { continue } @@ -235,12 +240,12 @@ func (f *usageEventFeed) ListEvents( // usageEventFromAuditEntry converts one audit-log entry into a usage event // tying the actor to the org they acted in. Returns ok=false when the entry // can't be attributed to a synced user resource. -func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Event, time.Time, bool) { +func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Event, bool) { actor := entry.GetActor() actorID := entry.GetActorID() ts := entry.GetTimestamp().Time if actorID == 0 || ts.IsZero() { - return nil, time.Time{}, false + return nil, false } // actor_is_bot is real but undocumented (only in AdditionalFields); trust @@ -248,15 +253,15 @@ func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Eve // aren't synced as users, so their events wouldn't correlate to anything. if isBot, ok := entry.AdditionalFields["actor_is_bot"].(bool); ok { if isBot { - return nil, time.Time{}, false + return nil, false } } else if strings.HasSuffix(actor, "[bot]") { - return nil, time.Time{}, false + return nil, false } orgID := entry.GetOrgID() if orgID == 0 { - return nil, time.Time{}, false + return nil, false } id := entry.GetDocumentID() @@ -287,5 +292,5 @@ func usageEventFromAuditEntry(orgName string, entry *github.AuditEntry) (*v2.Eve }, }, }, - }, ts, true + }, true } diff --git a/pkg/connector/usage_event_feed_test.go b/pkg/connector/usage_event_feed_test.go index fdf80520..723c62c0 100644 --- a/pkg/connector/usage_event_feed_test.go +++ b/pkg/connector/usage_event_feed_test.go @@ -97,12 +97,12 @@ func TestUsageEventFromAuditEntry(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - evt, evtTs, ok := usageEventFromAuditEntry("octo-org", tt.entry) + evt, ok := usageEventFromAuditEntry("octo-org", tt.entry) require.Equal(t, tt.ok, ok) if !tt.ok { return } - require.Equal(t, ts, evtTs) + require.Equal(t, ts, evt.GetOccurredAt().AsTime()) require.Equal(t, "123", evt.GetUsageEvent().GetActorResource().GetId().GetResource()) require.Equal(t, "456", evt.GetUsageEvent().GetTargetResource().GetId().GetResource()) require.Equal(t, resourceTypeUser.Id, evt.GetUsageEvent().GetActorResource().GetId().GetResourceType()) @@ -123,7 +123,7 @@ func TestUsageEventFromAuditEntry_IdFallback(t *testing.T) { DocumentID: github.Ptr("real-doc-id"), Action: github.Ptr("repo.create"), } - evt, _, ok := usageEventFromAuditEntry("octo-org", entry) + evt, ok := usageEventFromAuditEntry("octo-org", entry) require.True(t, ok) require.Equal(t, "real-doc-id", evt.GetId()) }) @@ -136,7 +136,7 @@ func TestUsageEventFromAuditEntry_IdFallback(t *testing.T) { Timestamp: &github.Timestamp{Time: ts}, Action: github.Ptr("repo.create"), } - evt, _, ok := usageEventFromAuditEntry("octo-org", entry) + evt, ok := usageEventFromAuditEntry("octo-org", entry) require.True(t, ok) require.Equal(t, fmt.Sprintf("456:123:%d:repo.create", ts.UnixNano()), evt.GetId()) require.NotEmpty(t, evt.GetId()) @@ -468,6 +468,47 @@ func TestUsageEventFeed_ListEvents_ResumesFromPersistedCursor(t *testing.T) { require.False(t, state.HasMore) } +func TestUsageEventFeed_ListEvents_RecoversFromOutOfBoundsCursor(t *testing.T) { + ctx := context.Background() + + since := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + newer := since.Add(1 * time.Hour) + + // A corrupted/stale cursor: OrgIndex points past the end of Orgs. + badToken := &usageEventPageToken{ + Orgs: []string{"octo-org"}, + OrgIndex: 5, + AuditLogCursor: "stale-cursor", + Since: since.Format(time.RFC3339Nano), + } + cursorStr, err := badToken.marshal() + require.NoError(t, err) + + var gotPage string + httpClient := mock.NewMockedHTTPClient( + mock.WithRequestMatchHandler( + mock.GetOrgsAuditLogByOrg, + http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPage = r.URL.Query().Get("page") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(mock.MustMarshal([]*github.AuditEntry{ + {Actor: github.Ptr("octocat"), ActorID: github.Ptr(int64(1)), OrgID: github.Ptr(int64(9)), Timestamp: &github.Timestamp{Time: newer}}, + })) + }), + ), + ) + + f := newUsageEventFeed(github.NewClient(httpClient), nil) + + require.NotPanics(t, func() { + events, state, _, err := f.ListEvents(ctx, nil, &pagination.StreamToken{Cursor: cursorStr}) + require.NoError(t, err) + require.Len(t, events, 1) + require.False(t, state.HasMore) + }) + require.Empty(t, gotPage, "should discard the stale per-org cursor when OrgIndex is reset") +} + func TestUsageEventFeed_ListEvents_NoOrgs(t *testing.T) { ctx := context.Background()