diff --git a/messaging/topic_mgt.go b/messaging/topic_mgt.go index 0e7e9d0c..81d0bd1f 100644 --- a/messaging/topic_mgt.go +++ b/messaging/topic_mgt.go @@ -19,15 +19,17 @@ import ( "encoding/json" "fmt" "net/http" + "net/url" "strings" "firebase.google.com/go/v4/internal" ) 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. @@ -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,147 @@ 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 > maxTopicManagementWorkers { + numWorkers = maxTopicManagementWorkers + } + + 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, + }) + } + } + + if err := ctx.Err(); err != nil { + return nil, err + } + + 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 internal.HasSuccessStatus(resp) || 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)), + } + } + + _, err := c.httpClient.Do(ctx, request) + if err == nil { + return true, "" + } + + if fe, ok := err.(*internal.FirebaseError); ok { + if code, ok := fe.Ext["messagingErrorCode"].(string); ok && code != "" { + return false, code + } + if fe.ErrorCode != "" && fe.ErrorCode != internal.Unknown { + return false, string(fe.ErrorCode) + } + } + + return false, "UNKNOWN_ERROR" +} diff --git a/messaging/topic_mgt_test.go b/messaging/topic_mgt_test.go index c0703022..13bafeb9 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 requestCount int ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - tr = r - b, _ = io.ReadAll(r.Body) + mu.Lock() + requestCount++ + 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,245 @@ 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 requestCount != 2 { + t.Errorf("got %d requests, want 2", requestCount) + } } -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 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 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + requestCount++ + 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 requestCount != 2 { + t.Errorf("got %d requests, want 2", requestCount) + } +} + +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 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 ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -89,15 +308,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 +350,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 +405,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 +419,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) } }