From 5e0b0051e5c58ce0c4fc5186d96f8fcb070eb73b Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Wed, 3 Jun 2026 03:07:18 -0300 Subject: [PATCH 1/5] feat: filter admin event feed server-side by event name The Google Reports API accepts a single eventName query parameter per request. Previously the admin feed fetched all admin events with no filter and discarded anything that wasn't GROUP_SETTINGS or USER_SETTINGS on the client side. Now ListEvents issues one targeted ListActivities call per relevant event name (15 total: 7 group + 8 user). Per-event-name page tokens are stored in the cursor so continuation calls only re-request names that still have pages remaining. Co-Authored-By: Claude Sonnet 4.6 --- pkg/connector/admin_event_feed.go | 129 ++++++++++++++++--------- pkg/connector/admin_event_feed_test.go | 74 ++++++++------ pkg/connector/usage_event_feed.go | 11 ++- 3 files changed, 136 insertions(+), 78 deletions(-) diff --git a/pkg/connector/admin_event_feed.go b/pkg/connector/admin_event_feed.go index f8490bba..f76d686f 100644 --- a/pkg/connector/admin_event_feed.go +++ b/pkg/connector/admin_event_feed.go @@ -23,6 +23,40 @@ import ( gwclient "github.com/conductorone/baton-google-workspace/pkg/client" ) +// adminEventNames lists the admin-application event names the feed subscribes to. +// The Google Reports API accepts one eventName per request, so ListEvents issues +// one ListActivities call per name and merges the results. +var adminEventNames = []string{ + // GROUP_SETTINGS + "CREATE_GROUP", + "CHANGE_GROUP_DESCRIPTION", + "CHANGE_GROUP_NAME", + "CHANGE_GROUP_EMAIL", + "ADD_GROUP_MEMBER", + "UPDATE_GROUP_MEMBER", + "DELETE_GROUP", + // USER_SETTINGS + "ACCEPT_USER_INVITATION", + "CHANGE_USER_ORGANIZATION", + "ADD_DISPLAY_NAME", + "CHANGE_DISPLAY_NAME", + "CHANGE_FIRST_NAME", + "CHANGE_LAST_NAME", + "CREATE_USER", + "RENAME_USER", +} + +// adminGroupEventNames is the subset of adminEventNames that belong to GROUP_SETTINGS handling. +var adminGroupEventNames = map[string]bool{ + "CREATE_GROUP": true, + "CHANGE_GROUP_DESCRIPTION": true, + "CHANGE_GROUP_NAME": true, + "CHANGE_GROUP_EMAIL": true, + "ADD_GROUP_MEMBER": true, + "UPDATE_GROUP_MEMBER": true, + "DELETE_GROUP": true, +} + type cacheEntry struct { Id string DisplayName string @@ -43,68 +77,74 @@ type adminEventFeed struct { func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Timestamp, pToken *pagination.StreamToken) ([]*v2.Event, *pagination.StreamState, annotations.Annotations, error) { l := ctxzap.Extract(ctx) - var streamState *pagination.StreamState - cursor, err := unmarshalPageToken(pToken, startAt) if err != nil { return nil, nil, nil, fmt.Errorf("failed to unmarshal page token: %w", err) } - r, err := f.client.ListActivities(ctx, "all", "admin", "", cursor.StartAt, cursor.NextPageToken, int64(pToken.Size)) - if err != nil { - return nil, nil, nil, fmt.Errorf("google-workspace: failed to list admin activities: %w", err) - } - latestEvent, err := time.Parse(time.RFC3339, cursor.LatestEventSeen) if err != nil { return nil, nil, nil, fmt.Errorf("failed to parse latest event time in admin event feed: %w", err) } - events := make([]*v2.Event, 0) - for _, activity := range r.Items { - occurredAt := convertIdTimeToTimestamp(activity.Id.Time) - if occurredAt == nil { - // Set occurred at to epoch so that it should never be after the latest event - // Unless latest event is before epoch for some reason - occurredAt = timestamppb.New(time.Unix(0, 0)) + // On the first call EventPageTokens is nil — fetch every event name from StartAt. + // On continuation calls fetch only names that still have a next page token. + fetchNames := adminEventNames + if len(cursor.EventPageTokens) > 0 { + fetchNames = make([]string, 0, len(cursor.EventPageTokens)) + for name := range cursor.EventPageTokens { + fetchNames = append(fetchNames, name) } - if occurredAt.AsTime().After(latestEvent) { - cursor.LatestEventSeen = occurredAt.AsTime().Format(time.RFC3339) - latestEvent = occurredAt.AsTime() + } + + events := make([]*v2.Event, 0) + nextTokens := make(map[string]string) + + for _, eventName := range fetchNames { + r, err := f.client.ListActivities(ctx, "all", "admin", eventName, cursor.StartAt, cursor.EventPageTokens[eventName], int64(pToken.Size)) + if err != nil { + return nil, nil, nil, fmt.Errorf("google-workspace: failed to list admin activities for %s: %w", eventName, err) } - // There can be multiple events, have not found an example of this yet - for _, evt := range activity.Events { - switch evt.Type { - case "GROUP_SETTINGS": - changeEvents, err := f.handleGroupEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt) - if err != nil { - l.Error("failed to handle group event", zap.Error(err)) - continue + + for _, activity := range r.Items { + occurredAt := convertIdTimeToTimestamp(activity.Id.Time) + if occurredAt == nil { + occurredAt = timestamppb.New(time.Unix(0, 0)) + } + if occurredAt.AsTime().After(latestEvent) { + cursor.LatestEventSeen = occurredAt.AsTime().Format(time.RFC3339) + latestEvent = occurredAt.AsTime() + } + for _, evt := range activity.Events { + var changeEvents []*v2.Event + var evtErr error + if adminGroupEventNames[evt.Name] { + changeEvents, evtErr = f.handleGroupEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt) + } else { + changeEvents, evtErr = f.handleUserEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt) } - events = append(events, changeEvents...) - case "USER_SETTINGS": - changeEvents, err := f.handleUserEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt) - if err != nil { - l.Error("failed to handle user event", zap.Error(err)) + if evtErr != nil { + l.Error("failed to handle admin event", zap.String("event_name", evt.Name), zap.Error(evtErr)) continue } events = append(events, changeEvents...) - default: - l.Debug("google-workspace-event-feed: skipping event", zap.String("event", evt.Name), zap.String("type", evt.Type)) - continue } } + + if r.NextPageToken != "" { + nextTokens[eventName] = r.NextPageToken + } } - l.Debug("google-workspace-event-feed: listed events", - zap.Int("count", len(r.Items)), - zap.String("next_page_token", r.NextPageToken), - zap.Any("start_at", startAt), - zap.Any("latest_event", cursor.LatestEventSeen), + l.Debug("google-workspace-event-feed: listed admin events", + zap.Int("event_names_fetched", len(fetchNames)), + zap.Int("events_produced", len(events)), + zap.String("latest_event", cursor.LatestEventSeen), ) - cursor.NextPageToken = r.NextPageToken - if r.NextPageToken == "" { + hasMore := len(nextTokens) > 0 + cursor.EventPageTokens = nextTokens + if !hasMore { cursor.StartAt = cursor.LatestEventSeen cursor.LatestEventSeen = "" } @@ -113,12 +153,11 @@ func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti if err != nil { return nil, nil, nil, fmt.Errorf("failed to marshal cursor token in admin event feed: %w", err) } - streamState = &pagination.StreamState{ - Cursor: cursorToken, - HasMore: r.NextPageToken != "", - } - return events, streamState, nil, nil + return events, &pagination.StreamState{ + Cursor: cursorToken, + HasMore: hasMore, + }, nil, nil } func (f *adminEventFeed) handleGroupEvent(ctx context.Context, uniqueQualifier int64, occurredAt *timestamppb.Timestamp, activityEvt *reports.ActivityEvents) ([]*v2.Event, error) { diff --git a/pkg/connector/admin_event_feed_test.go b/pkg/connector/admin_event_feed_test.go index 6707186e..89d3ee35 100644 --- a/pkg/connector/admin_event_feed_test.go +++ b/pkg/connector/admin_event_feed_test.go @@ -26,12 +26,18 @@ type safeUserResponse struct { } // Minimal fake for Reports Activities.List + Directory lookups used by admin_event_feed. -func newAdminFeedTestServer(users map[string]*directoryAdmin.User, groups map[string]*directoryAdmin.Group, activities *reportsAdmin.Activities) *httptest.Server { +// activitiesByEvent maps eventName query parameter values to the Activities response +// to return. An unrecognised eventName returns an empty Activities response. +func newAdminFeedTestServer(users map[string]*directoryAdmin.User, groups map[string]*directoryAdmin.Group, activitiesByEvent map[string]*reportsAdmin.Activities) *httptest.Server { mux := http.NewServeMux() mux.HandleFunc("/admin/reports/v1/activity/users/all/applications/admin", func(w http.ResponseWriter, r *http.Request) { - // ignore query parsing beyond pageToken/startTime for now - _ = json.NewEncoder(w).Encode(activities) + eventName := r.URL.Query().Get("eventName") + resp, ok := activitiesByEvent[eventName] + if !ok { + resp = &reportsAdmin.Activities{} + } + _ = json.NewEncoder(w).Encode(resp) }) mux.HandleFunc("/admin/directory/v1/users/", func(w http.ResponseWriter, r *http.Request) { @@ -75,40 +81,50 @@ func TestAdminEventFeed_GroupAndUserEvents(t *testing.T) { "group@example.com": {Id: "group-1", Name: "Group One", Email: "group@example.com"}, } - // Build Activities with admin events we handle + // Build per-event-name Activities responses. The feed now issues one + // ListActivities call per event name, so the mock routes by eventName. now := time.Now().UTC().Format(time.RFC3339) - acts := &reportsAdmin.Activities{ - Items: []*reportsAdmin.Activity{ - { + actsByEvent := map[string]*reportsAdmin.Activities{ + "CHANGE_GROUP_NAME": { + Items: []*reportsAdmin.Activity{{ Id: &reportsAdmin.ActivityId{Time: now, UniqueQualifier: 123}, - Events: []*reportsAdmin.ActivityEvents{ - { - Type: "GROUP_SETTINGS", - Name: "CHANGE_GROUP_NAME", - Parameters: []*reportsAdmin.ActivityEventsParameters{ - {Name: "GROUP_EMAIL", Value: "group@example.com"}, - }, + Events: []*reportsAdmin.ActivityEvents{{ + Type: "GROUP_SETTINGS", + Name: "CHANGE_GROUP_NAME", + Parameters: []*reportsAdmin.ActivityEventsParameters{ + {Name: "GROUP_EMAIL", Value: "group@example.com"}, }, - { - Type: "GROUP_SETTINGS", Name: "ADD_GROUP_MEMBER", - Parameters: []*reportsAdmin.ActivityEventsParameters{ - {Name: "GROUP_EMAIL", Value: "group@example.com"}, - {Name: "USER_EMAIL", Value: "user@example.com"}, - }, + }}, + }}, + }, + "ADD_GROUP_MEMBER": { + Items: []*reportsAdmin.Activity{{ + Id: &reportsAdmin.ActivityId{Time: now, UniqueQualifier: 124}, + Events: []*reportsAdmin.ActivityEvents{{ + Type: "GROUP_SETTINGS", + Name: "ADD_GROUP_MEMBER", + Parameters: []*reportsAdmin.ActivityEventsParameters{ + {Name: "GROUP_EMAIL", Value: "group@example.com"}, + {Name: "USER_EMAIL", Value: "user@example.com"}, }, - }, - }, - { + }}, + }}, + }, + "CHANGE_FIRST_NAME": { + Items: []*reportsAdmin.Activity{{ Id: &reportsAdmin.ActivityId{Time: now, UniqueQualifier: 456}, - Events: []*reportsAdmin.ActivityEvents{ - {Type: "USER_SETTINGS", Name: "CHANGE_FIRST_NAME", Parameters: []*reportsAdmin.ActivityEventsParameters{{Name: "USER_EMAIL", Value: "user@example.com"}}}, - }, - }, + Events: []*reportsAdmin.ActivityEvents{{ + Type: "USER_SETTINGS", + Name: "CHANGE_FIRST_NAME", + Parameters: []*reportsAdmin.ActivityEventsParameters{ + {Name: "USER_EMAIL", Value: "user@example.com"}, + }, + }}, + }}, }, - NextPageToken: "", } - server := newAdminFeedTestServer(users, groups, acts) + server := newAdminFeedTestServer(users, groups, actsByEvent) defer server.Close() dir := newTestDirectoryService(t, server.URL, server.Client()) diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 994c40a6..3c8fafc4 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -81,10 +81,13 @@ func hasParameter(name string, parameters []*reportsAdmin.ActivityEventsParamete } type pageToken struct { - LatestEventSeen string `json:"latest_event_seen,omitempty"` - NextPageToken string `json:"next_page_token,omitempty"` - StartAt string `json:"start_at,omitempty"` - PageSize int `json:"page_size,omitempty"` + LatestEventSeen string `json:"latest_event_seen,omitempty"` + NextPageToken string `json:"next_page_token,omitempty"` + StartAt string `json:"start_at,omitempty"` + PageSize int `json:"page_size,omitempty"` + // EventPageTokens holds per-event-name pagination cursors for feeds that + // issue one ListActivities request per event name (e.g. adminEventFeed). + EventPageTokens map[string]string `json:"event_page_tokens,omitempty"` } func unmarshalPageToken(token *pagination.StreamToken, defaultStart *timestamppb.Timestamp) (*pageToken, error) { From eec7d0f9bef297344e1294fd122b8ef900d3f7d6 Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Wed, 3 Jun 2026 04:51:31 -0300 Subject: [PATCH 2/5] feat: request up to 1k entries per request on admin event feed --- pkg/connector/admin_event_feed.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pkg/connector/admin_event_feed.go b/pkg/connector/admin_event_feed.go index f76d686f..1a4a5793 100644 --- a/pkg/connector/admin_event_feed.go +++ b/pkg/connector/admin_event_feed.go @@ -64,6 +64,10 @@ type cacheEntry struct { type cacheMap map[string]cacheEntry +// adminActivitiesPageSize is the number of activity items requested per ListActivities +// call. The Google Reports API maximum is 1000. +const adminActivitiesPageSize = 1000 + type adminEventFeed struct { client *gwclient.GoogleWorkspaceClient @@ -101,7 +105,7 @@ func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti nextTokens := make(map[string]string) for _, eventName := range fetchNames { - r, err := f.client.ListActivities(ctx, "all", "admin", eventName, cursor.StartAt, cursor.EventPageTokens[eventName], int64(pToken.Size)) + r, err := f.client.ListActivities(ctx, "all", "admin", eventName, cursor.StartAt, cursor.EventPageTokens[eventName], adminActivitiesPageSize) if err != nil { return nil, nil, nil, fmt.Errorf("google-workspace: failed to list admin activities for %s: %w", eventName, err) } From 9d11cc13f6fa6d256a94829e10a10309ad3d382b Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Thu, 4 Jun 2026 10:47:47 -0300 Subject: [PATCH 3/5] fix: increase page size for usage_event_feed requests --- pkg/connector/usage_event_feed.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pkg/connector/usage_event_feed.go b/pkg/connector/usage_event_feed.go index 3c8fafc4..27f9452a 100644 --- a/pkg/connector/usage_event_feed.go +++ b/pkg/connector/usage_event_feed.go @@ -23,6 +23,10 @@ import ( var privateAppIDRegex = regexp.MustCompile("[0-9]{21}") +// usageActivitiesPageSize is the number of activity items requested per ListActivities +// call. The Google Reports API maximum is 1000. +const usageActivitiesPageSize = 1000 + // maxEventFeedLookback caps how far back event feeds query the Google Reports API. // Google page tokens expire after ~24h, so a cursor left mid-pagination (e.g. after // a connector restart or a transient timeout) would otherwise keep requesting the @@ -154,7 +158,7 @@ func (f *usageEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti return nil, nil, nil, fmt.Errorf("failed to unmarshal page token in usage event feed: %w", err) } - r, err := f.c.ListActivities(ctx, "all", "token", "authorize", cursor.StartAt, cursor.NextPageToken, int64(pToken.Size)) + r, err := f.c.ListActivities(ctx, "all", "token", "authorize", cursor.StartAt, cursor.NextPageToken, usageActivitiesPageSize) if err != nil { return nil, nil, nil, fmt.Errorf("google-workspace: failed to list token activities: %w", err) } From 98dd90ffcc8fbc4656a2a503ee1d5be6b3364099 Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Tue, 9 Jun 2026 21:25:27 -0300 Subject: [PATCH 4/5] fix: skip sibling events in admin event feed to prevent duplicate emissions When the Reports API returns a multi-event activity, the inner loop now skips events whose name doesn't match the current per-name query, ensuring each event is processed exactly once. Co-Authored-By: Claude Sonnet 4.6 --- pkg/connector/admin_event_feed.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pkg/connector/admin_event_feed.go b/pkg/connector/admin_event_feed.go index 1a4a5793..8bda7257 100644 --- a/pkg/connector/admin_event_feed.go +++ b/pkg/connector/admin_event_feed.go @@ -120,6 +120,9 @@ func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti latestEvent = occurredAt.AsTime() } for _, evt := range activity.Events { + if evt.Name != eventName { + continue + } var changeEvents []*v2.Event var evtErr error if adminGroupEventNames[evt.Name] { From 868d049f52d54e1d6e174bb9253c5feb684bd9cf Mon Sep 17 00:00:00 2001 From: Javier David Carnelli Date: Wed, 10 Jun 2026 13:45:38 -0300 Subject: [PATCH 5/5] refactor: hoist adminGroupEventNames lookup out of inner event loop Co-Authored-By: Claude Sonnet 4.6 --- pkg/connector/admin_event_feed.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/connector/admin_event_feed.go b/pkg/connector/admin_event_feed.go index 8bda7257..28ac5d91 100644 --- a/pkg/connector/admin_event_feed.go +++ b/pkg/connector/admin_event_feed.go @@ -105,6 +105,7 @@ func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti nextTokens := make(map[string]string) for _, eventName := range fetchNames { + isGroupEvent := adminGroupEventNames[eventName] r, err := f.client.ListActivities(ctx, "all", "admin", eventName, cursor.StartAt, cursor.EventPageTokens[eventName], adminActivitiesPageSize) if err != nil { return nil, nil, nil, fmt.Errorf("google-workspace: failed to list admin activities for %s: %w", eventName, err) @@ -125,7 +126,7 @@ func (f *adminEventFeed) ListEvents(ctx context.Context, startAt *timestamppb.Ti } var changeEvents []*v2.Event var evtErr error - if adminGroupEventNames[evt.Name] { + if isGroupEvent { changeEvents, evtErr = f.handleGroupEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt) } else { changeEvents, evtErr = f.handleUserEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt)