fix(fcm): Optimize SendEachInBatch with worker pool (#695)

* Refactor: Optimize SendEachInBatch with worker pool

The `sendEachInBatch` function, used for sending FCM messages individually within a batch, previously created a new goroutine for each message. This could lead to high CPU usage for large batches.

This commit refactors `sendEachInBatch` to use a fixed-size pool of concurrent operations (10 operations) to manage message sending. This limits the number of active goroutines, reducing CPU overhead and improving resource utilization.

Key changes:
- Implemented early validation: All messages are now validated upfront. If any message is invalid, the function returns an error immediately without attempting to send any messages.
- Introduced a mechanism for managing concurrent operations: Manages a fixed number of goroutines to process message sending tasks. Messages are distributed via channels.
- Ensured result ordering: The order of responses in `BatchResponse.Responses` correctly matches the order of the input messages, even with concurrent processing.
- Updated unit tests in `messaging_batch_test.go` to comprehensively cover the new implementation, including scenarios for varying batch sizes, partial failures, early validation, and response ordering.
- Confirmed that the existing HTTP client continues to leverage HTTP/2.

* Refactor: Increase SendEachInBatch worker pool size to 50

This commit increases the number of goroutines in the
`sendEachInBatch` function from 10 to 50.

This change is based on your feedback to potentially improve
throughput for large batches of FCM messages. The number 50 was
chosen as a significant increase over the previous conservative
value of 10, aiming to provide better concurrency for I/O-bound
operations without being excessively high.

The goal is to allow for more parallel processing of messages up
to the maximum batch limit of 500. Performance in specific
environments should be monitored to ensure this change has the
desired effect without causing undue resource strain.

Unit tests in `messaging_batch_test.go` have been reviewed and
adjusted to ensure they remain meaningful with the new pool size,
particularly scenarios testing behavior when the number of messages
is less than, equal to, or greater than the number of concurrent
processes.

* Fix: Correct lint and build errors in messaging batch code

This commit addresses two issues introduced in previous changes:

1.  A syntax error in `messaging/messaging_batch_test.go` caused by a duplicated `t.Run` call and associated redundant variable declarations.
2.  An "imported and not used" error for the "sync" package in `messaging/messaging_batch.go`, which occurred after `sync.WaitGroup` was removed and the import statement was not.

These fixes ensure the code builds cleanly and passes lint checks.

* Fix: Add missing sync import to messaging_batch_test.go

This commit re-adds the `import "sync"` statement to
`messaging/messaging_batch_test.go`.

The `sync` package, specifically `sync.Mutex`, is used within
the test suite (e.g., in `TestSendEachWorkerPoolScenarios` and
`TestSendEachResponseOrderWithConcurrency`) to protect shared
variables like hit counters and logs from race conditions when
accessed by concurrently running mock HTTP server handlers.

A previous change inadvertently removed this import while cleaning
up unused imports in other files, leading to "undefined: sync"
build errors in the test file. This commit corrects that oversight.

* Fix: Correct multiple failures in messaging test suite

This commit addresses several issues in the messaging test suite
(`messaging_batch_test.go`) that were causing test failures,
including data races and incorrect test logic:

1.  **Data Races:**
    *   In `TestSendEachPartialFailure` and `TestSendEachTotalFailure`,
        the `serverHitCount` variable in mock server handlers was accessed
        concurrently without synchronization. This has been fixed by
        introducing a `sync.Mutex` to protect access to `serverHitCount`
        within each test case's server setup.

2.  **Early Validation Test (`TestSendEachEarlyValidationSkipsSend`):**
    *   The `invalid_last_message` test was failing because the
        type of invalidity used (large payload) was not caught by the
        early `validateMessage` scan in `sendEachInBatch`.
    *   Fixed by changing the invalid message to one with both Topic and
        Token defined, which `validateMessage` correctly flags, ensuring
        the test now accurately verifies early exit behavior and that no
        server calls are made.

3.  **Partial Failure Scenario (`TestSendEachWorkerPoolScenarios`):**
    *   The test for partial failures
        (`NumMessages_75_AllSuccess_false_...`) was failing due to a
        mismatch between the mock server's failure simulation (based on
        request arrival order) and the test's assertions (based on
        original message index).
    *   Fixed by modifying the mock server to parse the original message
        index from the message content (topic) and use this for
        failure simulation, ensuring deterministic alignment with assertions.

These changes should ensure the stability and correctness of the
messaging batch processing tests, especially those verifying
concurrent operations and error handling.

* Style: Ensure gofmt formatting for messaging batch code

I applied `gofmt -s .` to ensure consistent code formatting,
particularly for `messaging/messaging_batch.go` and
`messaging/messaging_batch_test.go`.

This commit reflects any adjustments made by gofmt to align
with standard Go formatting practices, including simplification
of struct initializations and consistent spacing.

* Style: Apply gofmt fixes including trailing whitespace

This commit applies gofmt formatting, with a specific focus
on removing trailing whitespace from messaging/messaging_batch_test.go
as identified by previous `gofmt -d` outputs.

This is part of a series of fixes to ensure the codebase
conforms to standard Go formatting practices and to resolve
various test failures and linting issues.

* fix typo in variable name

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: lahirumaramba

messaging/messaging_batch.go

@@ -27,7 +27,6 @@ import (
 	"mime/multipart"
 	"net/http"
 	"net/textproto"
-	"sync"
 
 	"firebase.google.com/go/v4/internal"
 )
@@ -165,53 +164,85 @@ func (c *fcmClient) sendEachInBatch(ctx context.Context, messages []*Message, dr
 		return nil, fmt.Errorf("messages must not contain more than %d elements", maxMessages)
 	}
 
-	var responses []*SendResponse = make([]*SendResponse, len(messages))
-	var wg sync.WaitGroup
-
 	for idx, m := range messages {
 		if err := validateMessage(m); err != nil {
 			return nil, fmt.Errorf("invalid message at index %d: %v", idx, err)
 		}
-		wg.Add(1)
-		go func(idx int, m *Message, dryRun bool, responses []*SendResponse) {
-			defer wg.Done()
-			var resp string
-			var err error
-			if dryRun {
-				resp, err = c.SendDryRun(ctx, m)
-			} else {
-				resp, err = c.Send(ctx, m)
-			}
-			if err == nil {
-				responses[idx] = &SendResponse{
-					Success:   true,
-					MessageID: resp,
-				}
-			} else {
-				responses[idx] = &SendResponse{
-					Success: false,
-					Error:   err,
-				}
-			}
-		}(idx, m, dryRun, responses)
 	}
-	// Wait for all SendDryRun/Send calls to finish
-	wg.Wait()
+
+	const numWorkers = 50
+	jobs := make(chan job, len(messages))
+	results := make(chan result, len(messages))
+
+	responses := make([]*SendResponse, len(messages))
+
+	for w := 0; w < numWorkers; w++ {
+		go worker(ctx, c, dryRun, jobs, results)
+	}
+
+	for idx, m := range messages {
+		jobs <- job{message: m, index: idx}
+	}
+	close(jobs)
+
+	for i := 0; i < len(messages); i++ {
+		res := <-results
+		responses[res.index] = res.response
+	}
 
 	successCount := 0
+	failureCount := 0
 	for _, r := range responses {
 		if r.Success {
 			successCount++
+		} else {
+			failureCount++
 		}
 	}
 
 	return &BatchResponse{
 		Responses:    responses,
 		SuccessCount: successCount,
-		FailureCount: len(responses) - successCount,
+		FailureCount: failureCount,
 	}, nil
 }
 
+type job struct {
+	message *Message
+	index   int
+}
+
+type result struct {
+	response *SendResponse
+	index    int
+}
+
+func worker(ctx context.Context, c *fcmClient, dryRun bool, jobs <-chan job, results chan<- result) {
+	for j := range jobs {
+		var respMsg string
+		var err error
+		if dryRun {
+			respMsg, err = c.SendDryRun(ctx, j.message)
+		} else {
+			respMsg, err = c.Send(ctx, j.message)
+		}
+
+		var sr *SendResponse
+		if err == nil {
+			sr = &SendResponse{
+				Success:   true,
+				MessageID: respMsg,
+			}
+		} else {
+			sr = &SendResponse{
+				Success: false,
+				Error:   err,
+			}
+		}
+		results <- result{response: sr, index: j.index}
+	}
+}
+
 // SendAll sends the messages in the given array via Firebase Cloud Messaging.
 //
 // The messages array may contain up to 500 messages. SendAll employs batching to send the entire