[8.16](backport #42682) x-pack/filebeat/input/entityanalytics/provider/activedirectory: fix use before init bug

CHANGELOG.next.asciidoc

@@ -107,6 +107,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Fix request trace filename handling in http_endpoint input. {pull}39410[39410]
 - Upgrade github.com/hashicorp/go-retryablehttp to mitigate CVE-2024-6104 {pull}40036[40036]
 - [Journald] Fixes handling of `journalctl` restart. A known symptom was broken multiline messages when there was a restart of journalctl while aggregating the lines. {issue}41331[41331] {pull}42595[42595]
+- Fix entityanalytics activedirectory provider full sync use before initialization bug. {pull}42682[42682]
 
 *Heartbeat*
 

x-pack/filebeat/input/entityanalytics/provider/activedirectory/activedirectory.go

@@ -215,13 +215,63 @@
 		return err
 	}
 
+<<<<<<< HEAD
 	if len(state.users) != 0 {
 		tracker := kvstore.NewTxTracker(ctx)
 
 		start := time.Now()
 		p.publishMarker(start, start, inputCtx.ID, true, client, tracker)
 		for _, u := range state.users {
 			p.publishUser(u, state, inputCtx.ID, client, tracker)
+=======
+	if len(users) != 0 || state.len() != 0 {
+		// Active Directory does not have a notion of deleted users
+		// beyond absence from the directory, so compare found users
+		// with users already known by the state store and if any
+		// are in the store but not returned in the previous fetch,
+		// mark them as deleted and publish the deletion. We do not
+		// have the time of the deletion, so use now.
+		if state.len() != 0 {
+			found := make(map[string]bool)
+			for _, u := range users {
+				found[u.ID] = true
+			}
+			deleted := make(map[string]*User)
+			now := time.Now()
+			state.forEach(func(u *User) {
+				if u.State == Deleted {
+					// We have already seen that this is deleted
+					// so we do not need to publish again. The
+					// user will be deleted from the store when
+					// the state is closed.
+					return
+				}
+				if found[u.ID] {
+					// We have the user, so we do not need to
+					// mark it as deleted.
+					return
+				}
+				// This modifies the state store's copy since u
+				// is a pointer held by the state store map.
+				u.State = Deleted
+				u.WhenChanged = now
+				deleted[u.ID] = u
+			})
+			for _, u := range deleted {
+				users = append(users, u)
+			}
+		}
+		if len(users) != 0 {
+			start := time.Now()
+			tracker := kvstore.NewTxTracker(ctx)
+			p.publishMarker(start, start, inputCtx.ID, true, client, tracker)
+			for _, u := range users {
+				p.publishUser(u, state, inputCtx.ID, client, tracker)
+			}
+			end := time.Now()
+			p.publishMarker(end, end, inputCtx.ID, false, client, tracker)
+			tracker.Wait()
+>>>>>>> 1473ae916 (x-pack/filebeat/input/entityanalytics/provider/activedirectory: fix use before init bug (#42682))
 		}
 
 		end := time.Now()
@@ -246,21 +296,21 @@
 // runIncrementalUpdate will run an incremental update. The process is similar
 // to full synchronization, except only users which have changed (newly
 // discovered, modified, or deleted) will be published.
 func (p *adInput) runIncrementalUpdate(inputCtx v2.Context, store *kvstore.Store, client beat.Client) error {
 	p.logger.Debugf("Running incremental update...")
 
 	state, err := newStateStore(store)
 	if err != nil {
 		return fmt.Errorf("unable to begin transaction: %w", err)
 	}
 	defer func() { // If commit is successful, call to this close will be no-op.
 		closeErr := state.close(false)
 		if closeErr != nil {
 			p.logger.Errorw("Error rolling back incremental update transaction", "error", closeErr)
 		}
 	}()
 
 	ctx := ctxtool.FromCanceller(inputCtx.Cancelation)
 	updatedUsers, err := p.doFetchUsers(ctx, state, false)
 	if err != nil {
 		return err