From d27907d7e53f37ced8c567a3144758c5b0e5e34a Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Thu, 17 Sep 2026 14:09:39 -0400 Subject: [PATCH 1/3] feat(fcm): Migrate topic management to FCM v1 API Migrated SubscribeToTopic and UnsubscribeFromTopic to use the FCM v1 API with concurrent worker pool execution. Deprecated legacy IID methods and updated unit tests. --- messaging/topic_mgt.go | 209 +++++++++++++++++++++++- messaging/topic_mgt_test.go | 311 ++++++++++++++++++++++++++---------- 2 files changed, 434 insertions(+), 86 deletions(-) diff --git a/messaging/topic_mgt.go b/messaging/topic_mgt.go index 0e7e9d0c..94b44e31 100644 --- a/messaging/topic_mgt.go +++ b/messaging/topic_mgt.go @@ -18,7 +18,9 @@ import ( "context" "encoding/json" "fmt" + "io" "net/http" + "net/url" "strings" "firebase.google.com/go/v4/internal" @@ -76,10 +78,10 @@ func newIIDClient(hc *http.Client, conf *internal.MessagingConfig) *iidClient { } } -// SubscribeToTopic subscribes a list of registration tokens to a topic. +// SubscribeToTopicLegacy subscribes a list of registration tokens to a topic using the legacy Instance ID API. // -// The tokens list must not be empty, and have at most 1000 tokens. -func (c *iidClient) SubscribeToTopic(ctx context.Context, tokens []string, topic string) (*TopicManagementResponse, error) { +// Deprecated: Use SubscribeToTopic instead. +func (c *iidClient) SubscribeToTopicLegacy(ctx context.Context, tokens []string, topic string) (*TopicManagementResponse, error) { req := &iidRequest{ Topic: topic, Tokens: tokens, @@ -88,10 +90,10 @@ func (c *iidClient) SubscribeToTopic(ctx context.Context, tokens []string, topic return c.makeTopicManagementRequest(ctx, req) } -// UnsubscribeFromTopic unsubscribes a list of registration tokens from a topic. +// UnsubscribeFromTopicLegacy unsubscribes a list of registration tokens from a topic using the legacy Instance ID API. // -// The tokens list must not be empty, and have at most 1000 tokens. -func (c *iidClient) UnsubscribeFromTopic(ctx context.Context, tokens []string, topic string) (*TopicManagementResponse, error) { +// Deprecated: Use UnsubscribeFromTopic instead. +func (c *iidClient) UnsubscribeFromTopicLegacy(ctx context.Context, tokens []string, topic string) (*TopicManagementResponse, error) { req := &iidRequest{ Topic: topic, Tokens: tokens, @@ -161,3 +163,198 @@ func handleIIDError(resp *internal.Response) error { return base } + +func validateTopicManagementArgs(tokens []string, topic string) (string, error) { + if len(tokens) == 0 { + return "", fmt.Errorf("no tokens specified") + } + if len(tokens) > 1000 { + return "", fmt.Errorf("tokens list must not contain more than 1000 items") + } + for _, token := range tokens { + if token == "" { + return "", fmt.Errorf("tokens list must not contain empty strings") + } + } + + if topic == "" { + return "", fmt.Errorf("topic name not specified") + } + if !topicNamePattern.MatchString(topic) { + return "", fmt.Errorf("invalid topic name: %q", topic) + } + + topicName := strings.TrimPrefix(topic, "/topics/") + return topicName, nil +} + +type topicJob struct { + token string + index int +} + +type topicResult struct { + index int + success bool + reason string +} + +// SubscribeToTopic subscribes a list of registration tokens to a topic via the FCM v1 API. +// +// The tokens list must not be empty, and have at most 1000 tokens. +func (c *fcmClient) SubscribeToTopic(ctx context.Context, tokens []string, topic string) (*TopicManagementResponse, error) { + return c.makeTopicManagementRequestV1(ctx, tokens, topic, true) +} + +// UnsubscribeFromTopic unsubscribes a list of registration tokens from a topic via the FCM v1 API. +// +// The tokens list must not be empty, and have at most 1000 tokens. +func (c *fcmClient) UnsubscribeFromTopic(ctx context.Context, tokens []string, topic string) (*TopicManagementResponse, error) { + return c.makeTopicManagementRequestV1(ctx, tokens, topic, false) +} + +func (c *fcmClient) makeTopicManagementRequestV1(ctx context.Context, tokens []string, topic string, isSubscribe bool) (*TopicManagementResponse, error) { + topicName, err := validateTopicManagementArgs(tokens, topic) + if err != nil { + return nil, err + } + + numWorkers := len(tokens) + if numWorkers > 100 { + numWorkers = 100 + } + + jobs := make(chan topicJob, len(tokens)) + results := make(chan topicResult, len(tokens)) + + for w := 0; w < numWorkers; w++ { + go func() { + for j := range jobs { + success, reason := c.makeTopicManagementSingleRequest(ctx, j.token, topicName, isSubscribe) + results <- topicResult{ + index: j.index, + success: success, + reason: reason, + } + } + }() + } + + for idx, token := range tokens { + jobs <- topicJob{token: token, index: idx} + } + close(jobs) + + resps := make([]topicResult, len(tokens)) + for i := 0; i < len(tokens); i++ { + res := <-results + resps[res.index] = res + } + + tmr := &TopicManagementResponse{} + for _, res := range resps { + if res.success { + tmr.SuccessCount++ + } else { + tmr.FailureCount++ + tmr.Errors = append(tmr.Errors, &ErrorInfo{ + Index: res.index, + Reason: res.reason, + }) + } + } + + return tmr, nil +} + +func (c *fcmClient) makeTopicManagementSingleRequest(ctx context.Context, token, topicName string, isSubscribe bool) (bool, string) { + encodedToken := url.PathEscape(token) + var request *internal.Request + + if isSubscribe { + request = &internal.Request{ + Method: http.MethodPost, + URL: fmt.Sprintf("%s/projects/%s/registrations/%s/topicSubscriptions?topic_name=%s", c.fcmEndpoint, c.project, encodedToken, url.QueryEscape(topicName)), + Body: internal.NewJSONEntity(map[string]interface{}{}), + SuccessFn: func(resp *internal.Response) bool { + return resp.Status == http.StatusOK || resp.Status == http.StatusConflict + }, + } + } else { + request = &internal.Request{ + Method: http.MethodDelete, + URL: fmt.Sprintf("%s/projects/%s/registrations/%s/topicSubscriptions/%s?allow_missing=true", c.fcmEndpoint, c.project, encodedToken, url.PathEscape(topicName)), + } + } + + resp, err := c.httpClient.Do(ctx, request) + if err == nil { + if resp != nil && isSubscribe && resp.Status == http.StatusConflict { + return true, "" + } + return true, "" + } + + var respBody []byte + var status int + if fe, ok := err.(*internal.FirebaseError); ok && fe.Response != nil { + status = fe.Response.StatusCode + if fe.Response.Body != nil { + respBody, _ = io.ReadAll(fe.Response.Body) + } + } + + if isSubscribe && status == http.StatusConflict { + return true, "" + } + + var parsed struct { + Error struct { + Status string `json:"status"` + Message string `json:"message"` + Details []struct { + Type string `json:"@type"` + ErrorCode string `json:"errorCode"` + } `json:"details"` + } `json:"error"` + } + + if len(respBody) > 0 { + _ = json.Unmarshal(respBody, &parsed) + } + + if isSubscribe && parsed.Error.Status == "ALREADY_EXISTS" { + return true, "" + } + + for _, d := range parsed.Error.Details { + if d.Type == "type.googleapis.com/google.firebase.fcm.v1.FcmError" && d.ErrorCode != "" { + return false, d.ErrorCode + } + } + + if parsed.Error.Status != "" { + return false, parsed.Error.Status + } + + if parsed.Error.Message != "" { + return false, parsed.Error.Message + } + + switch status { + case http.StatusBadRequest: + return false, "INVALID_ARGUMENT" + case http.StatusUnauthorized, http.StatusForbidden: + return false, "PERMISSION_DENIED" + case http.StatusNotFound: + return false, "NOT_FOUND" + case http.StatusTooManyRequests: + return false, "RESOURCE_EXHAUSTED" + case http.StatusInternalServerError: + return false, "INTERNAL" + case http.StatusServiceUnavailable: + return false, "DEADLINE_EXCEEDED" + default: + return false, "UNKNOWN_ERROR" + } +} diff --git a/messaging/topic_mgt_test.go b/messaging/topic_mgt_test.go index c0703022..9d508a7d 100644 --- a/messaging/topic_mgt_test.go +++ b/messaging/topic_mgt_test.go @@ -22,20 +22,28 @@ import ( "net/http/httptest" "reflect" "strings" + "sync" "testing" - "firebase.google.com/go/v4/errorutils" "firebase.google.com/go/v4/internal" ) func TestSubscribe(t *testing.T) { - var tr *http.Request - var b []byte + var mu sync.Mutex + var requests []*http.Request ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - tr = r - b, _ = io.ReadAll(r.Body) + mu.Lock() + requests = append(requests, r) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") - w.Write([]byte("{\"results\": [{}, {\"error\": \"error_reason\"}]}")) + if strings.Contains(r.URL.Path, "id2") { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(`{"error": {"status": "INVALID_ARGUMENT", "message": "error_reason"}}`)) + } else { + w.WriteHeader(http.StatusOK) + w.Write([]byte("{}")) + } })) defer ts.Close() @@ -44,34 +52,196 @@ func TestSubscribe(t *testing.T) { if err != nil { t.Fatal(err) } - client.iidEndpoint = ts.URL + "/v1" + client.fcmEndpoint = ts.URL resp, err := client.SubscribeToTopic(ctx, []string{"id1", "id2"}, "test-topic") if err != nil { t.Fatal(err) } - checkIIDRequest(t, b, tr, iidSubscribe) - checkTopicMgtResponse(t, resp) + checkTopicMgtResponse(t, resp, "INVALID_ARGUMENT") + if len(requests) != 2 { + t.Errorf("got %d requests, want 2", len(requests)) + } } -func TestInvalidSubscribe(t *testing.T) { +func TestSubscribeAlreadyExists409(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusConflict) + w.Write([]byte(`{"error": {"status": "ALREADY_EXISTS", "message": "Already exists"}}`)) + })) + defer ts.Close() + ctx := context.Background() client, err := NewClient(ctx, testMessagingConfig) if err != nil { t.Fatal(err) } - for _, tc := range invalidTopicMgtArgs { - t.Run(tc.name, func(t *testing.T) { - resp, err := client.SubscribeToTopic(ctx, tc.tokens, tc.topic) - if err == nil || err.Error() != tc.want { - t.Errorf( - "SubscribeToTopic(%s) = (%#v, %v); want = (nil, %q)", tc.name, resp, err, tc.want) - } - }) + client.fcmEndpoint = ts.URL + + resp, err := client.SubscribeToTopic(ctx, []string{"id1"}, "test-topic") + if err != nil { + t.Fatal(err) + } + if resp.SuccessCount != 1 || resp.FailureCount != 0 { + t.Errorf("resp = (%d, %d), want (1, 0)", resp.SuccessCount, resp.FailureCount) } } func TestUnsubscribe(t *testing.T) { + var mu sync.Mutex + var requests []*http.Request + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + requests = append(requests, r) + mu.Unlock() + + w.Header().Set("Content-Type", "application/json") + if strings.Contains(r.URL.Path, "id2") { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(`{"error": {"status": "INVALID_ARGUMENT", "message": "error_reason"}}`)) + } else { + w.WriteHeader(http.StatusOK) + w.Write([]byte("{}")) + } + })) + defer ts.Close() + + ctx := context.Background() + client, err := NewClient(ctx, testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.fcmEndpoint = ts.URL + + resp, err := client.UnsubscribeFromTopic(ctx, []string{"id1", "id2"}, "test-topic") + if err != nil { + t.Fatal(err) + } + checkTopicMgtResponse(t, resp, "INVALID_ARGUMENT") + if len(requests) != 2 { + t.Errorf("got %d requests, want 2", len(requests)) + } +} + +func TestUnsubscribeNotFound404(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error": {"status": "NOT_FOUND", "message": "Not found"}}`)) + })) + defer ts.Close() + + ctx := context.Background() + client, err := NewClient(ctx, testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.fcmEndpoint = ts.URL + + resp, err := client.UnsubscribeFromTopic(ctx, []string{"id1"}, "test-topic") + if err != nil { + t.Fatal(err) + } + if resp.SuccessCount != 0 || resp.FailureCount != 1 { + t.Errorf("resp = (%d, %d), want (0, 1)", resp.SuccessCount, resp.FailureCount) + } + if len(resp.Errors) != 1 || resp.Errors[0].Reason != "NOT_FOUND" { + t.Errorf("Errors[0].Reason = %q, want NOT_FOUND", resp.Errors[0].Reason) + } +} + +func TestTopicManagementFcmErrorDetails(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{ + "error": { + "status": "NOT_FOUND", + "details": [ + { + "@type": "type.googleapis.com/google.firebase.fcm.v1.FcmError", + "errorCode": "UNREGISTERED" + } + ] + } + }`)) + })) + defer ts.Close() + + ctx := context.Background() + client, err := NewClient(ctx, testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.fcmEndpoint = ts.URL + + resp, err := client.SubscribeToTopic(ctx, []string{"id1"}, "test-topic") + if err != nil { + t.Fatal(err) + } + if resp.SuccessCount != 0 || resp.FailureCount != 1 { + t.Errorf("resp = (%d, %d), want (0, 1)", resp.SuccessCount, resp.FailureCount) + } + if len(resp.Errors) != 1 || resp.Errors[0].Reason != "UNREGISTERED" { + t.Errorf("Errors[0].Reason = %q, want UNREGISTERED", resp.Errors[0].Reason) + } +} + +func TestTopicManagement500Error(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(`{"error": null}`)) + })) + defer ts.Close() + + ctx := context.Background() + client, err := NewClient(ctx, testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.fcmEndpoint = ts.URL + + resp, err := client.SubscribeToTopic(ctx, []string{"id1"}, "test-topic") + if err != nil { + t.Fatal(err) + } + if resp.SuccessCount != 0 || resp.FailureCount != 1 { + t.Errorf("resp = (%d, %d), want (0, 1)", resp.SuccessCount, resp.FailureCount) + } + if len(resp.Errors) != 1 || resp.Errors[0].Reason != "INTERNAL" { + t.Errorf("Errors[0].Reason = %q, want INTERNAL", resp.Errors[0].Reason) + } +} + +func TestTopicManagementNonJsonError(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte("not json")) + })) + defer ts.Close() + + ctx := context.Background() + client, err := NewClient(ctx, testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.fcmEndpoint = ts.URL + + resp, err := client.SubscribeToTopic(ctx, []string{"id1"}, "test-topic") + if err != nil { + t.Fatal(err) + } + if resp.SuccessCount != 0 || resp.FailureCount != 1 { + t.Errorf("resp = (%d, %d), want (0, 1)", resp.SuccessCount, resp.FailureCount) + } + if len(resp.Errors) != 1 || resp.Errors[0].Reason != "INVALID_ARGUMENT" { + t.Errorf("Errors[0].Reason = %q, want INVALID_ARGUMENT", resp.Errors[0].Reason) + } +} + +func TestSubscribeLegacy(t *testing.T) { var tr *http.Request var b []byte ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -89,15 +259,41 @@ func TestUnsubscribe(t *testing.T) { } client.iidEndpoint = ts.URL + "/v1" - resp, err := client.UnsubscribeFromTopic(ctx, []string{"id1", "id2"}, "test-topic") + resp, err := client.SubscribeToTopicLegacy(ctx, []string{"id1", "id2"}, "test-topic") + if err != nil { + t.Fatal(err) + } + checkIIDRequest(t, b, tr, iidSubscribe) + checkTopicMgtResponse(t, resp, "error_reason") +} + +func TestUnsubscribeLegacy(t *testing.T) { + var tr *http.Request + var b []byte + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tr = r + b, _ = io.ReadAll(r.Body) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte("{\"results\": [{}, {\"error\": \"error_reason\"}]}")) + })) + defer ts.Close() + + ctx := context.Background() + client, err := NewClient(ctx, testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.iidEndpoint = ts.URL + "/v1" + + resp, err := client.UnsubscribeFromTopicLegacy(ctx, []string{"id1", "id2"}, "test-topic") if err != nil { t.Fatal(err) } checkIIDRequest(t, b, tr, iidUnsubscribe) - checkTopicMgtResponse(t, resp) + checkTopicMgtResponse(t, resp, "error_reason") } -func TestInvalidUnsubscribe(t *testing.T) { +func TestInvalidSubscribe(t *testing.T) { ctx := context.Background() client, err := NewClient(ctx, testMessagingConfig) if err != nil { @@ -105,74 +301,29 @@ func TestInvalidUnsubscribe(t *testing.T) { } for _, tc := range invalidTopicMgtArgs { t.Run(tc.name, func(t *testing.T) { - resp, err := client.UnsubscribeFromTopic(ctx, tc.tokens, tc.topic) + resp, err := client.SubscribeToTopic(ctx, tc.tokens, tc.topic) if err == nil || err.Error() != tc.want { t.Errorf( - "UnsubscribeFromTopic(%s) = (%#v, %v); want = (nil, %q)", tc.name, resp, err, tc.want) + "SubscribeToTopic(%s) = (%#v, %v); want = (nil, %q)", tc.name, resp, err, tc.want) } }) } } -func TestTopicManagementError(t *testing.T) { - var resp string - var status int - ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(status) - w.Header().Set("Content-Type", "application/json") - w.Write([]byte(resp)) - })) - defer ts.Close() - +func TestInvalidUnsubscribe(t *testing.T) { ctx := context.Background() client, err := NewClient(ctx, testMessagingConfig) if err != nil { t.Fatal(err) } - client.iidEndpoint = ts.URL + "/v1" - client.iidClient.httpClient.RetryConfig = nil - - cases := []struct { - name, resp, want string - status int - check func(err error) bool - }{ - { - name: "EmptyResponse", - resp: "{}", - want: "unexpected http response with status: 500\n{}", - status: http.StatusInternalServerError, - check: errorutils.IsInternal, - }, - { - name: "ErrorCode", - resp: "{\"error\": \"INVALID_ARGUMENT\"}", - want: "error while calling the iid service: INVALID_ARGUMENT", - status: http.StatusBadRequest, - check: errorutils.IsInvalidArgument, - }, - { - name: "NotJson", - resp: "not json", - want: "unexpected http response with status: 500\nnot json", - status: http.StatusInternalServerError, - check: errorutils.IsInternal, - }, - } - - for _, tc := range cases { - resp = tc.resp - status = tc.status - - tmr, err := client.SubscribeToTopic(ctx, []string{"id1"}, "topic") - if err == nil || err.Error() != tc.want || !tc.check(err) { - t.Errorf("SubscribeToTopic(%s) = (%#v, %v); want = (nil, %q)", tc.name, tmr, err, tc.want) - } - - tmr, err = client.UnsubscribeFromTopic(ctx, []string{"id1"}, "topic") - if err == nil || err.Error() != tc.want || !tc.check(err) { - t.Errorf("UnsubscribeFromTopic(%s) = (%#v, %v); want = (nil, %q)", tc.name, tmr, err, tc.want) - } + for _, tc := range invalidTopicMgtArgs { + t.Run(tc.name, func(t *testing.T) { + resp, err := client.UnsubscribeFromTopic(ctx, tc.tokens, tc.topic) + if err == nil || err.Error() != tc.want { + t.Errorf( + "UnsubscribeFromTopic(%s) = (%#v, %v); want = (nil, %q)", tc.name, resp, err, tc.want) + } + }) } } @@ -205,7 +356,7 @@ func checkIIDRequest(t *testing.T, b []byte, tr *http.Request, op string) { } } -func checkTopicMgtResponse(t *testing.T, resp *TopicManagementResponse) { +func checkTopicMgtResponse(t *testing.T, resp *TopicManagementResponse, wantReason string) { if resp.SuccessCount != 1 { t.Errorf("SuccessCount = %d; want = %d", resp.SuccessCount, 1) } @@ -219,8 +370,8 @@ func checkTopicMgtResponse(t *testing.T, resp *TopicManagementResponse) { if e.Index != 1 { t.Errorf("ErrorInfo.Index = %d; want = %d", e.Index, 1) } - if e.Reason != "error_reason" { - t.Errorf("ErrorInfo.Reason = %s; want = %s", e.Reason, "error_reason") + if e.Reason != wantReason { + t.Errorf("ErrorInfo.Reason = %s; want = %s", e.Reason, wantReason) } } From c3fafd128f6182f0a8657d40fb60ecaa8067650c Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Mon, 21 Sep 2026 15:21:48 -0400 Subject: [PATCH 2/3] fix(fcm): Check context cancellation and avoid storing request pointers in tests Address review feedback on PR #785: - Return ctx.Err() directly if context is cancelled during topic management requests. - Use an integer request counter in test handlers instead of storing http.Request pointers. - Add TestTopicManagementContextCancelled unit test. --- messaging/topic_mgt.go | 4 ++++ messaging/topic_mgt_test.go | 43 ++++++++++++++++++++++++++++++------- 2 files changed, 39 insertions(+), 8 deletions(-) diff --git a/messaging/topic_mgt.go b/messaging/topic_mgt.go index 94b44e31..2228f478 100644 --- a/messaging/topic_mgt.go +++ b/messaging/topic_mgt.go @@ -264,6 +264,10 @@ func (c *fcmClient) makeTopicManagementRequestV1(ctx context.Context, tokens []s } } + if err := ctx.Err(); err != nil { + return nil, err + } + return tmr, nil } diff --git a/messaging/topic_mgt_test.go b/messaging/topic_mgt_test.go index 9d508a7d..68f507a5 100644 --- a/messaging/topic_mgt_test.go +++ b/messaging/topic_mgt_test.go @@ -30,10 +30,10 @@ import ( func TestSubscribe(t *testing.T) { var mu sync.Mutex - var requests []*http.Request + var requestCount int ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { mu.Lock() - requests = append(requests, r) + requestCount++ mu.Unlock() w.Header().Set("Content-Type", "application/json") @@ -59,8 +59,8 @@ func TestSubscribe(t *testing.T) { t.Fatal(err) } checkTopicMgtResponse(t, resp, "INVALID_ARGUMENT") - if len(requests) != 2 { - t.Errorf("got %d requests, want 2", len(requests)) + if requestCount != 2 { + t.Errorf("got %d requests, want 2", requestCount) } } @@ -90,10 +90,10 @@ func TestSubscribeAlreadyExists409(t *testing.T) { func TestUnsubscribe(t *testing.T) { var mu sync.Mutex - var requests []*http.Request + var requestCount int ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { mu.Lock() - requests = append(requests, r) + requestCount++ mu.Unlock() w.Header().Set("Content-Type", "application/json") @@ -119,8 +119,8 @@ func TestUnsubscribe(t *testing.T) { t.Fatal(err) } checkTopicMgtResponse(t, resp, "INVALID_ARGUMENT") - if len(requests) != 2 { - t.Errorf("got %d requests, want 2", len(requests)) + if requestCount != 2 { + t.Errorf("got %d requests, want 2", requestCount) } } @@ -241,6 +241,33 @@ func TestTopicManagementNonJsonError(t *testing.T) { } } +func TestTopicManagementContextCancelled(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte("{}")) + })) + defer ts.Close() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + client, err := NewClient(context.Background(), testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.fcmEndpoint = ts.URL + + resp, err := client.SubscribeToTopic(ctx, []string{"id1"}, "test-topic") + if err != context.Canceled { + t.Errorf("SubscribeToTopic() = (%#v, %v); want = (nil, %v)", resp, err, context.Canceled) + } + + resp, err = client.UnsubscribeFromTopic(ctx, []string{"id1"}, "test-topic") + if err != context.Canceled { + t.Errorf("UnsubscribeFromTopic() = (%#v, %v); want = (nil, %v)", resp, err, context.Canceled) + } +} + func TestSubscribeLegacy(t *testing.T) { var tr *http.Request var b []byte From 4ce2a0ffe4d03385bb34ac7a06d0009c42ac49fa Mon Sep 17 00:00:00 2001 From: Lahiru Maramba Date: Tue, 22 Sep 2026 17:04:20 -0400 Subject: [PATCH 3/3] fix(fcm): Address PR review comments on FCM v1 topic management Address review feedback from yvonnep165 and jonathanedey on PR #785: - Extract maxTopicManagementWorkers as a named constant. - Use internal.HasSuccessStatus in topic subscription SuccessFn. - Simplify error handling in makeTopicManagementSingleRequest by leveraging parsed FirebaseError. - Remove redundant 409 conflict checks and unused io import. - Add unit test for non-200 2xx success response. --- messaging/topic_mgt.go | 83 +++++++------------------------------ messaging/topic_mgt_test.go | 22 ++++++++++ 2 files changed, 36 insertions(+), 69 deletions(-) diff --git a/messaging/topic_mgt.go b/messaging/topic_mgt.go index 2228f478..81d0bd1f 100644 --- a/messaging/topic_mgt.go +++ b/messaging/topic_mgt.go @@ -18,7 +18,6 @@ import ( "context" "encoding/json" "fmt" - "io" "net/http" "net/url" "strings" @@ -27,9 +26,10 @@ import ( ) const ( - iidEndpoint = "https://iid.googleapis.com/iid/v1" - iidSubscribe = "batchAdd" - iidUnsubscribe = "batchRemove" + iidEndpoint = "https://iid.googleapis.com/iid/v1" + iidSubscribe = "batchAdd" + iidUnsubscribe = "batchRemove" + maxTopicManagementWorkers = 100 ) // TopicManagementResponse is the result produced by topic management operations. @@ -220,8 +220,8 @@ func (c *fcmClient) makeTopicManagementRequestV1(ctx context.Context, tokens []s } numWorkers := len(tokens) - if numWorkers > 100 { - numWorkers = 100 + if numWorkers > maxTopicManagementWorkers { + numWorkers = maxTopicManagementWorkers } jobs := make(chan topicJob, len(tokens)) @@ -281,7 +281,7 @@ func (c *fcmClient) makeTopicManagementSingleRequest(ctx context.Context, token, URL: fmt.Sprintf("%s/projects/%s/registrations/%s/topicSubscriptions?topic_name=%s", c.fcmEndpoint, c.project, encodedToken, url.QueryEscape(topicName)), Body: internal.NewJSONEntity(map[string]interface{}{}), SuccessFn: func(resp *internal.Response) bool { - return resp.Status == http.StatusOK || resp.Status == http.StatusConflict + return internal.HasSuccessStatus(resp) || resp.Status == http.StatusConflict }, } } else { @@ -291,74 +291,19 @@ func (c *fcmClient) makeTopicManagementSingleRequest(ctx context.Context, token, } } - resp, err := c.httpClient.Do(ctx, request) + _, err := c.httpClient.Do(ctx, request) if err == nil { - if resp != nil && isSubscribe && resp.Status == http.StatusConflict { - return true, "" - } return true, "" } - var respBody []byte - var status int - if fe, ok := err.(*internal.FirebaseError); ok && fe.Response != nil { - status = fe.Response.StatusCode - if fe.Response.Body != nil { - respBody, _ = io.ReadAll(fe.Response.Body) + if fe, ok := err.(*internal.FirebaseError); ok { + if code, ok := fe.Ext["messagingErrorCode"].(string); ok && code != "" { + return false, code } - } - - if isSubscribe && status == http.StatusConflict { - return true, "" - } - - var parsed struct { - Error struct { - Status string `json:"status"` - Message string `json:"message"` - Details []struct { - Type string `json:"@type"` - ErrorCode string `json:"errorCode"` - } `json:"details"` - } `json:"error"` - } - - if len(respBody) > 0 { - _ = json.Unmarshal(respBody, &parsed) - } - - if isSubscribe && parsed.Error.Status == "ALREADY_EXISTS" { - return true, "" - } - - for _, d := range parsed.Error.Details { - if d.Type == "type.googleapis.com/google.firebase.fcm.v1.FcmError" && d.ErrorCode != "" { - return false, d.ErrorCode + if fe.ErrorCode != "" && fe.ErrorCode != internal.Unknown { + return false, string(fe.ErrorCode) } } - if parsed.Error.Status != "" { - return false, parsed.Error.Status - } - - if parsed.Error.Message != "" { - return false, parsed.Error.Message - } - - switch status { - case http.StatusBadRequest: - return false, "INVALID_ARGUMENT" - case http.StatusUnauthorized, http.StatusForbidden: - return false, "PERMISSION_DENIED" - case http.StatusNotFound: - return false, "NOT_FOUND" - case http.StatusTooManyRequests: - return false, "RESOURCE_EXHAUSTED" - case http.StatusInternalServerError: - return false, "INTERNAL" - case http.StatusServiceUnavailable: - return false, "DEADLINE_EXCEEDED" - default: - return false, "UNKNOWN_ERROR" - } + return false, "UNKNOWN_ERROR" } diff --git a/messaging/topic_mgt_test.go b/messaging/topic_mgt_test.go index 68f507a5..13bafeb9 100644 --- a/messaging/topic_mgt_test.go +++ b/messaging/topic_mgt_test.go @@ -88,6 +88,28 @@ func TestSubscribeAlreadyExists409(t *testing.T) { } } +func TestSubscribe204Success(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNoContent) + })) + defer ts.Close() + + ctx := context.Background() + client, err := NewClient(ctx, testMessagingConfig) + if err != nil { + t.Fatal(err) + } + client.fcmEndpoint = ts.URL + + resp, err := client.SubscribeToTopic(ctx, []string{"id1"}, "test-topic") + if err != nil { + t.Fatal(err) + } + if resp.SuccessCount != 1 || resp.FailureCount != 0 { + t.Errorf("resp = (%d, %d), want (1, 0)", resp.SuccessCount, resp.FailureCount) + } +} + func TestUnsubscribe(t *testing.T) { var mu sync.Mutex var requestCount int