fix(asana): page through stories, subtasks and tags and stop on null next_page

backend/plugins/asana/tasks/shared.go

@@ -0,0 +1,54 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+
+	"github.com/apache/devlake/core/errors"
+	"github.com/apache/devlake/helpers/pluginhelper/api"
+)
+
+// asanaPageSize is the largest page Asana list endpoints return.
+const asanaPageSize = 100
+
+// setAsanaPageQuery sets the page size and, after the first page, the offset
+// that the previous page returned in next_page.
+func setAsanaPageQuery(query url.Values, reqData *api.RequestData) {
+	query.Set("limit", fmt.Sprintf("%d", reqData.Pager.Size))
+	if offset, ok := reqData.CustomData.(string); ok && offset != "" {
+		query.Set("offset", offset)
+	}
+}
+
+// getNextAsanaPageOffset returns the offset of the next page. Asana sets
+// next_page to null when there are no more pages, so the collection ends
+// there: requesting a page without an offset would start again at page 1.
+func getNextAsanaPageOffset(_ *api.RequestData, prevPageResponse *http.Response) (interface{}, errors.Error) {
+	var resp asanaListResponse
+	err := api.UnmarshalResponse(prevPageResponse, &resp)
+	if err != nil {
+		return nil, err
+	}
+	if resp.NextPage == nil || resp.NextPage.Offset == "" {
+		return nil, api.ErrFinishCollect
+	}
+	return resp.NextPage.Offset, nil
+}

backend/plugins/asana/tasks/shared_test.go

@@ -0,0 +1,65 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package tasks
+
+import (
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/apache/devlake/helpers/pluginhelper/api"
+	"github.com/stretchr/testify/assert"
+)
+
+func asanaListPage(t *testing.T, body string) *http.Response {
+	req, err := http.NewRequest(http.MethodGet, "https://app.asana.com/api/1.0/tasks/1/stories", nil)
+	assert.Nil(t, err)
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       io.NopCloser(strings.NewReader(body)),
+		Request:    req,
+	}
+}
+
+func TestSetAsanaPageQuery(t *testing.T) {
+	first := url.Values{}
+	setAsanaPageQuery(first, &api.RequestData{Pager: &api.Pager{Page: 1, Size: asanaPageSize}})
+	assert.Equal(t, "100", first.Get("limit"))
+	assert.False(t, first.Has("offset"))
+
+	next := url.Values{}
+	setAsanaPageQuery(next, &api.RequestData{Pager: &api.Pager{Page: 2, Size: asanaPageSize}, CustomData: "eyJ0eXAiOJiKV1iQLCJhbGciOiJIUzI1NiJ9"})
+	assert.Equal(t, "100", next.Get("limit"))
+	assert.Equal(t, "eyJ0eXAiOJiKV1iQLCJhbGciOiJIUzI1NiJ9", next.Get("offset"))
+}
+
+func TestGetNextAsanaPageOffset(t *testing.T) {
+	offset, err := getNextAsanaPageOffset(nil, asanaListPage(t, `{
+		"data": [{"gid": "1"}],
+		"next_page": {"offset": "eyJ0eXAiOJiKV1iQLCJhbGciOiJIUzI1NiJ9", "path": "/tasks/1/stories?limit=100&offset=eyJ0eXAiOJiKV1iQLCJhbGciOiJIUzI1NiJ9", "uri": "https://app.asana.com/api/1.0/tasks/1/stories?limit=100&offset=eyJ0eXAiOJiKV1iQLCJhbGciOiJIUzI1NiJ9"}
+	}`))
+	assert.Nil(t, err)
+	assert.Equal(t, "eyJ0eXAiOJiKV1iQLCJhbGciOiJIUzI1NiJ9", offset)
+
+	// A full last page has "next_page": null. Going on without an offset
+	// would request page 1 again, so the collection must end here.
+	_, err = getNextAsanaPageOffset(nil, asanaListPage(t, `{"data": [{"gid": "1"}], "next_page": null}`))
+	assert.ErrorIs(t, err, api.ErrFinishCollect)
+}

backend/plugins/asana/tasks/story_collector.go

@@ -73,13 +73,15 @@ func CollectStory(taskCtx plugin.SubTaskContext) errors.Error {
 		},
 		ApiClient:   data.ApiClient,
 		Input:       iterator,
+		PageSize:    asanaPageSize,
 		UrlTemplate: "tasks/{{ .Input.Gid }}/stories",
 		Query: func(reqData *api.RequestData) (url.Values, errors.Error) {
 			query := url.Values{}
 			query.Set("opt_fields", "gid,resource_type,resource_subtype,text,html_text,is_pinned,is_edited,sticker_name,created_at,created_by,target")
-			query.Set("limit", "100")
+			setAsanaPageQuery(query, reqData)
 			return query, nil
 		},
+		GetNextPageCustomData: getNextAsanaPageOffset,
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var resp asanaListResponse
 			err := api.UnmarshalResponse(res, &resp)

backend/plugins/asana/tasks/subtask_collector.go

@@ -74,13 +74,15 @@ func CollectSubtask(taskCtx plugin.SubTaskContext) errors.Error {
 		},
 		ApiClient:   data.ApiClient,
 		Input:       iterator,
+		PageSize:    asanaPageSize,
 		UrlTemplate: "tasks/{{ .Input.Gid }}/subtasks",
 		Query: func(reqData *api.RequestData) (url.Values, errors.Error) {
 			query := url.Values{}
 			query.Set("opt_fields", "gid,name,notes,resource_type,resource_subtype,completed,completed_at,due_on,created_at,modified_at,permalink_url,assignee,created_by,parent,num_subtasks,memberships.section,memberships.project")
-			query.Set("limit", "100")
+			setAsanaPageQuery(query, reqData)
 			return query, nil
 		},
+		GetNextPageCustomData: getNextAsanaPageOffset,
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var resp asanaListResponse
 			err := api.UnmarshalResponse(res, &resp)

backend/plugins/asana/tasks/tag_collector.go

@@ -73,13 +73,15 @@ func CollectTag(taskCtx plugin.SubTaskContext) errors.Error {
 		},
 		ApiClient:   data.ApiClient,
 		Input:       iterator,
+		PageSize:    asanaPageSize,
 		UrlTemplate: "tasks/{{ .Input.Gid }}/tags",
 		Query: func(reqData *api.RequestData) (url.Values, errors.Error) {
 			query := url.Values{}
 			query.Set("opt_fields", "gid,name,resource_type,color,notes,permalink_url")
-			query.Set("limit", "100")
+			setAsanaPageQuery(query, reqData)
 			return query, nil
 		},
+		GetNextPageCustomData: getNextAsanaPageOffset,
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var resp asanaListResponse
 			err := api.UnmarshalResponse(res, &resp)

backend/plugins/asana/tasks/task_collector.go

@@ -61,31 +61,16 @@ func CollectTask(taskCtx plugin.SubTaskContext) errors.Error {
 			Table: rawTaskTable,
 		},
 		ApiClient:   data.ApiClient,
-		PageSize:    100,
+		PageSize:    asanaPageSize,
 		UrlTemplate: "projects/{{ .Params.ProjectId }}/tasks",
 		Query: func(reqData *api.RequestData) (url.Values, errors.Error) {
 			query := url.Values{}
-			query.Set("limit", "100")
 			// Request all fields needed for transformation including section name
 			query.Set("opt_fields", "gid,name,notes,resource_type,resource_subtype,completed,completed_at,due_on,created_at,modified_at,permalink_url,assignee,assignee.name,created_by,created_by.name,parent,num_subtasks,memberships.section,memberships.section.name,memberships.project")
-			if reqData.CustomData != nil {
-				if offset, ok := reqData.CustomData.(string); ok && offset != "" {
-					query.Set("offset", offset)
-				}
-			}
+			setAsanaPageQuery(query, reqData)
 			return query, nil
 		},
-		GetNextPageCustomData: func(prevReqData *api.RequestData, prevPageResponse *http.Response) (interface{}, errors.Error) {
-			var resp asanaTaskListResponse
-			err := api.UnmarshalResponse(prevPageResponse, &resp)
-			if err != nil {
-				return nil, err
-			}
-			if resp.NextPage != nil && resp.NextPage.Offset != "" {
-				return resp.NextPage.Offset, nil
-			}
-			return nil, nil
-		},
+		GetNextPageCustomData: getNextAsanaPageOffset,
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var w asanaTaskListResponse
 			err := api.UnmarshalResponse(res, &w)

backend/plugins/asana/tasks/user_collector.go

@@ -52,30 +52,15 @@ func CollectUser(taskCtx plugin.SubTaskContext) errors.Error {
 			Table: rawUserTable,
 		},
 		ApiClient:   data.ApiClient,
-		PageSize:    100,
+		PageSize:    asanaPageSize,
 		UrlTemplate: "projects/{{ .Params.ProjectId }}/members",
 		Query: func(reqData *api.RequestData) (url.Values, errors.Error) {
 			query := url.Values{}
 			query.Set("opt_fields", "gid,name,email,resource_type,photo.image_128x128")
-			query.Set("limit", "100")
-			if reqData.CustomData != nil {
-				if offset, ok := reqData.CustomData.(string); ok && offset != "" {
-					query.Set("offset", offset)
-				}
-			}
+			setAsanaPageQuery(query, reqData)
 			return query, nil
 		},
-		GetNextPageCustomData: func(prevReqData *api.RequestData, prevPageResponse *http.Response) (interface{}, errors.Error) {
-			var resp asanaListResponse
-			err := api.UnmarshalResponse(prevPageResponse, &resp)
-			if err != nil {
-				return nil, err
-			}
-			if resp.NextPage != nil && resp.NextPage.Offset != "" {
-				return resp.NextPage.Offset, nil
-			}
-			return nil, nil
-		},
+		GetNextPageCustomData: getNextAsanaPageOffset,
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var resp asanaListResponse
 			err := api.UnmarshalResponse(res, &resp)