diff --git a/pkg/connector/admin_event_feed.go b/pkg/connector/admin_event_feed.go index f8490bba..28ac5d91 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 @@ -30,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 @@ -43,68 +81,78 @@ 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 { + 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) } - // 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)) + + 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 { + if evt.Name != eventName { continue } - 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)) + var changeEvents []*v2.Event + var evtErr error + if isGroupEvent { + changeEvents, evtErr = f.handleGroupEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt) + } else { + changeEvents, evtErr = f.handleUserEvent(ctx, activity.Id.UniqueQualifier, occurredAt, evt) + } + 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 +161,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..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 @@ -81,10 +85,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) { @@ -151,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) }