fix(tempo): paginate worklog collection, honour the sync range, and emit ids that join jira

backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_worklogs.csv

@@ -0,0 +1,3 @@
+id,params,data,url,input,created_at
+1,"{""ConnectionId"":1,""TeamId"":4}","{""tempoWorklogId"":12345,""issue"":{""id"":10001},""timeSpentSeconds"":3600,""billableSeconds"":3600,""startDate"":""2024-01-15"",""startTime"":""09:00:00"",""description"":""Working on task"",""author"":{""accountId"":""abc123""},""createdAt"":""2024-01-15T10:00:00Z"",""updatedAt"":""2024-01-15T10:00:00Z""}",https://api.tempo.io/4/worklogs/team/4?from=2024-01-01&limit=1000&offset=0,null,2024-01-17 10:00:00.000
+2,"{""ConnectionId"":1,""TeamId"":4}","{""tempoWorklogId"":12346,""issue"":{""id"":10002},""timeSpentSeconds"":7200,""billableSeconds"":7200,""startDate"":""2024-01-16"",""startTime"":""14:00:00"",""description"":""Internal meeting"",""author"":{""accountId"":""712020:def-456""},""createdAt"":""2024-01-16T16:00:00Z"",""updatedAt"":""2024-01-16T16:00:00Z""}",https://api.tempo.io/4/worklogs/team/4?from=2024-01-01&limit=1000&offset=0,null,2024-01-17 10:00:00.000

backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_worklogs_extracted.csv

@@ -0,0 +1,3 @@
+"connection_id","tempo_worklog_id","team_id","issue_id","author_account_id","time_spent_seconds","start_date","start_time","_raw_data_params","_raw_data_table","_raw_data_id","_raw_data_remark"
+"1","12345","4","10001","abc123","3600","2024-01-15","09:00:00","{""ConnectionId"":1,""TeamId"":4}","_raw_tempo_api_worklogs","1",""
+"1","12346","4","10002","712020:def-456","7200","2024-01-16","14:00:00","{""ConnectionId"":1,""TeamId"":4}","_raw_tempo_api_worklogs","2",""

backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv

@@ -1,3 +1,3 @@
 id,author_id,comment,time_spent_minutes,logged_date,started_date,issue_id,_raw_data_params,_raw_data_table,_raw_data_id,_raw_data_remark
-tempo:TempoWorklog:12345,abc123,Working on task,60,2024-01-15T10:00:00.000+00:00,2024-01-15T09:00:00.000+00:00,jira:JiraIssues:1:10001,,,0,
-tempo:TempoWorklog:12346,abc123,Internal meeting,120,2024-01-16T16:00:00.000+00:00,2024-01-16T14:00:00.000+00:00,jira:JiraIssues:1:10002,,,0,
+tempo:TempoWorklog:12345,jira:JiraAccount:1:abc123,Working on task,60,2024-01-15T10:00:00.000+00:00,2024-01-15T09:00:00.000+00:00,jira:JiraIssue:1:10001,,,0,
+tempo:TempoWorklog:12346,jira:JiraAccount:1:abc123,Internal meeting,120,2024-01-16T16:00:00.000+00:00,2024-01-16T14:00:00.000+00:00,jira:JiraIssue:1:10002,,,0,

backend/plugins/tempo/e2e/worklog_test.go

@@ -47,6 +47,36 @@ func TestWorklogDataFlow(t *testing.T) {
 	)
 }
 
+func TestWorklogExtraction(t *testing.T) {
+	var plugin impl.Tempo
+	dataflowTester := e2ehelper.NewDataFlowTester(t, "tempo", plugin)
+
+	taskData := &tasks.TempoTaskData{
+		Options: &tasks.TempoOptions{
+			ConnectionId: 1,
+			TeamId:       4,
+		},
+	}
+
+	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_tempo_api_worklogs.csv", "_raw_tempo_api_worklogs")
+	dataflowTester.FlushTabler(&models.TempoWorklog{})
+	dataflowTester.Subtask(tasks.ExtractWorklogsMeta, taskData)
+	dataflowTester.VerifyTable(
+		models.TempoWorklog{},
+		"./snapshot_tables/_tool_tempo_worklogs_extracted.csv",
+		e2ehelper.ColumnWithRawData(
+			"connection_id",
+			"tempo_worklog_id",
+			"team_id",
+			"issue_id",
+			"author_account_id",
+			"time_spent_seconds",
+			"start_date",
+			"start_time",
+		),
+	)
+}
+
 func TestTeamDataFlow(t *testing.T) {
 	var plugin impl.Tempo
 	dataflowTester := e2ehelper.NewDataFlowTester(t, "tempo", plugin)

backend/plugins/tempo/tasks/worklog_collector.go

@@ -42,13 +42,19 @@ var CollectWorklogsMeta = plugin.SubTaskMeta{
 func CollectWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 	data := taskCtx.GetData().(*TempoTaskData)
 
-	apiCollector, err := api.NewStatefulApiCollector(api.RawDataSubTaskArgs{
+	// Scope raw data and collector state per team, like the scope itself
+	// (TempoTeam.GetParams): otherwise every team scope of a connection shares
+	// one collector state, and the second team collected in a pipeline would
+	// run incrementally from the first team's start time.
+	rawDataSubTaskArgs := api.RawDataSubTaskArgs{
 		Ctx: taskCtx,
 		Params: models.TempoApiParams{
 			ConnectionId: data.Options.ConnectionId,
+			TeamId:       data.Options.TeamId,
 		},
 		Table: RAW_WORKLOG_TABLE,
-	})
+	}
+	apiCollector, err := api.NewStatefulApiCollector(rawDataSubTaskArgs)
 	if err != nil {
 		return err
 	}
@@ -59,60 +65,16 @@ func CollectWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 	}
 
 	err = apiCollector.InitCollector(api.ApiCollectorArgs{
-		RawDataSubTaskArgs: api.RawDataSubTaskArgs{
-			Ctx: taskCtx,
-			Params: models.TempoApiParams{
-				ConnectionId: data.Options.ConnectionId,
-			},
-			Table: RAW_WORKLOG_TABLE,
-		},
-		ApiClient:   data.ApiClient,
-		UrlTemplate: urlTemplate,
-		PageSize:    1000,
-		GetTotalPages: func(res *http.Response, args *api.ApiCollectorArgs) (int, errors.Error) {
-			var response struct {
-				Metadata struct {
-					Count  int `json:"count"`
-					Limit  int `json:"limit"`
-					Total  int `json:"total"`
-					Offset int `json:"offset"`
-				} `json:"metadata"`
-			}
-			if err := api.UnmarshalResponse(res, &response); err != nil {
-				return 0, err
-			}
-			totalPages := (response.Metadata.Total + args.PageSize - 1) / args.PageSize
-			return totalPages, nil
-		},
+		RawDataSubTaskArgs: rawDataSubTaskArgs,
+		ApiClient:          data.ApiClient,
+		UrlTemplate:        urlTemplate,
+		PageSize:           1000,
+		// No GetTotalPages: Tempo v4 pagination metadata (PageableMetadata)
+		// has count/offset/limit/next/previous but no total, so pages are
+		// fetched until one comes back short.
 		Query: func(reqData *api.RequestData) (url.Values, errors.Error) {
-			query := url.Values{}
-			pager := reqData.Pager
-			if pager == nil {
-				pager = &api.Pager{Page: 1, Skip: 0, Size: 1000}
-			}
-			query.Set("offset", strconv.Itoa(pager.Skip))
-			query.Set("limit", strconv.Itoa(pager.Size))
-
-			if data.Options.TeamId != 0 {
-				fromDate := data.Options.FromDate
-				toDate := data.Options.ToDate
-				if fromDate == "" {
-					since := time.Now().AddDate(0, 0, -90)
-					fromDate = since.Format("2006-01-02")
-				}
-				if toDate == "" {
-					toDate = time.Now().Format("2006-01-02")
-				}
-				query.Set("from", fromDate)
-				query.Set("to", toDate)
-			} else {
-				if apiCollector.IsIncremental() && apiCollector.GetSince() != nil {
-					since := apiCollector.GetSince()
-					query.Set("updatedFrom", since.Format(time.RFC3339))
-				}
-			}
-
-			return query, nil
+			return buildWorklogQuery(data.Options, reqData.Pager,
+				apiCollector.IsIncremental(), apiCollector.GetSince()), nil
 		},
 		ResponseParser: func(res *http.Response) ([]json.RawMessage, errors.Error) {
 			var response struct {
@@ -132,3 +94,31 @@ func CollectWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 
 	return apiCollector.Execute()
 }
+
+// buildWorklogQuery builds the query for one page of worklogs. Explicit
+// fromDate/toDate options win; otherwise an incremental run asks for what
+// changed since the last successful collection (updatedFrom) and a full sync
+// starts at the sync policy's timeAfter (from).
+func buildWorklogQuery(opts *TempoOptions, pager *api.Pager, incremental bool, since *time.Time) url.Values {
+	if pager == nil {
+		pager = &api.Pager{Page: 1, Skip: 0, Size: 1000}
+	}
+	query := url.Values{}
+	query.Set("offset", strconv.Itoa(pager.Skip))
+	query.Set("limit", strconv.Itoa(pager.Size))
+
+	switch {
+	case opts.FromDate != "" || opts.ToDate != "":
+		if opts.FromDate != "" {
+			query.Set("from", opts.FromDate)
+		}
+		if opts.ToDate != "" {
+			query.Set("to", opts.ToDate)
+		}
+	case since != nil && incremental:
+		query.Set("updatedFrom", since.UTC().Format(time.RFC3339))
+	case since != nil:
+		query.Set("from", since.UTC().Format("2006-01-02"))
+	}
+	return query
+}

backend/plugins/tempo/tasks/worklog_collector_test.go

@@ -0,0 +1,70 @@
+/*
+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 (
+	"testing"
+	"time"
+
+	"github.com/apache/devlake/helpers/pluginhelper/api"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBuildWorklogQuery(t *testing.T) {
+	pager := &api.Pager{Page: 3, Skip: 2000, Size: 1000}
+	timeAfter := time.Date(2026, 1, 28, 0, 0, 0, 0, time.UTC)
+	lastSuccess := time.Date(2026, 9, 22, 0, 0, 13, 0, time.UTC)
+
+	t.Run("pages with offset and limit", func(t *testing.T) {
+		q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, false, nil)
+		assert.Equal(t, "2000", q.Get("offset"))
+		assert.Equal(t, "1000", q.Get("limit"))
+	})
+
+	t.Run("full sync starts at timeAfter, without a rolling window", func(t *testing.T) {
+		q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, false, &timeAfter)
+		assert.Equal(t, "2026-01-28", q.Get("from"))
+		assert.Empty(t, q.Get("to"))
+		assert.Empty(t, q.Get("updatedFrom"))
+	})
+
+	t.Run("incremental sync fetches what changed since the last success", func(t *testing.T) {
+		q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, true, &lastSuccess)
+		assert.Equal(t, "2026-09-22T00:00:13Z", q.Get("updatedFrom"))
+		assert.Empty(t, q.Get("from"))
+	})
+
+	t.Run("global endpoint gets the same bounds", func(t *testing.T) {
+		q := buildWorklogQuery(&TempoOptions{}, pager, false, &timeAfter)
+		assert.Equal(t, "2026-01-28", q.Get("from"))
+	})
+
+	t.Run("explicit fromDate/toDate options win", func(t *testing.T) {
+		opts := &TempoOptions{TeamId: 4, FromDate: "2026-03-01", ToDate: "2026-03-31"}
+		q := buildWorklogQuery(opts, pager, true, &lastSuccess)
+		assert.Equal(t, "2026-03-01", q.Get("from"))
+		assert.Equal(t, "2026-03-31", q.Get("to"))
+		assert.Empty(t, q.Get("updatedFrom"))
+	})
+
+	t.Run("no bounds when nothing is known", func(t *testing.T) {
+		q := buildWorklogQuery(&TempoOptions{TeamId: 4}, pager, false, nil)
+		assert.Empty(t, q.Get("from"))
+		assert.Empty(t, q.Get("updatedFrom"))
+	})
+}

backend/plugins/tempo/tasks/worklog_convertor.go

@@ -47,16 +47,14 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 
 	logger.Info("converting Tempo worklogs to domain layer")
 
-	issueIdMapping, err := buildIssueIdMapping(db, connectionId)
-	if err != nil {
-		return errors.Default.Wrap(err, "failed to build issue ID mapping")
-	}
-
 	clauses := []dal.Clause{
 		dal.Select("*"),
 		dal.From("_tool_tempo_worklogs"),
 		dal.Where("connection_id = ?", connectionId),
 	}
+	if data.Options.TeamId != 0 {
+		clauses = append(clauses, dal.Where("team_id = ?", data.Options.TeamId))
+	}
 	cursor, err := db.Cursor(clauses...)
 	if err != nil {
 		return errors.Default.Wrap(err, "failed to query Tempo worklogs")
@@ -68,6 +66,7 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 			Ctx: taskCtx,
 			Params: models.TempoApiParams{
 				ConnectionId: connectionId,
+				TeamId:       data.Options.TeamId,
 			},
 			Table: RAW_WORKLOG_TABLE,
 		},
@@ -76,11 +75,14 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 		Convert: func(inputRow interface{}) ([]interface{}, errors.Error) {
 			tempoWorklog := inputRow.(*models.TempoWorklog)
 
-			domainIssueId := ""
-			if domainId, ok := issueIdMapping[tempoWorklog.IssueId]; ok {
-				domainIssueId = domainId
-			} else {
-				domainIssueId = fmt.Sprintf("jira:JiraIssues:%d:%d", connectionId, tempoWorklog.IssueId)
+			// Must match the domain ids the jira plugin generates with didgen
+			// for JiraIssue / JiraAccount (plugins may not import each other),
+			// so worklogs join issues and accounts. Like before, this assumes
+			// the Tempo connection id equals the Jira connection id.
+			domainIssueId := fmt.Sprintf("jira:JiraIssue:%d:%d", connectionId, tempoWorklog.IssueId)
+			authorId := ""
+			if tempoWorklog.AuthorAccountId != "" {
+				authorId = fmt.Sprintf("jira:JiraAccount:%d:%s", connectionId, tempoWorklog.AuthorAccountId)
 			}
 
 			domainWorklogId := fmt.Sprintf("tempo:TempoWorklog:%d", tempoWorklog.TempoWorklogId)
@@ -105,7 +107,7 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 					Id: domainWorklogId,
 				},
 				IssueId:          domainIssueId,
-				AuthorId:         tempoWorklog.AuthorAccountId,
+				AuthorId:         authorId,
 				TimeSpentMinutes: timeSpentMinutes,
 				StartedDate:      startedDate,
 				LoggedDate:       loggedDate,
@@ -122,34 +124,3 @@ func ConvertWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 
 	return converter.Execute()
 }
-
-func buildIssueIdMapping(db dal.Dal, connectionId uint64) (map[int64]string, errors.Error) {
-	mapping := make(map[int64]string)
-
-	if !db.HasTable("_tool_jira_issues") {
-		return mapping, nil
-	}
-
-	clauses := []dal.Clause{
-		dal.Select("issue_id"),
-		dal.From("_tool_jira_issues"),
-		dal.Where("connection_id = ?", connectionId),
-	}
-
-	rows, err := db.Cursor(clauses...)
-	if err != nil {
-		return nil, errors.Default.Wrap(err, "failed to query jira issues")
-	}
-	defer rows.Close()
-
-	for rows.Next() {
-		var issueId uint64
-		if err := rows.Scan(&issueId); err != nil {
-			return nil, errors.Default.Wrap(err, "failed to scan issue")
-		}
-		domainId := fmt.Sprintf("jira:JiraIssues:%d:%d", connectionId, issueId)
-		mapping[int64(issueId)] = domainId
-	}
-
-	return mapping, nil
-}

backend/plugins/tempo/tasks/worklog_extractor.go

@@ -61,6 +61,7 @@ func ExtractWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 			Ctx: taskCtx,
 			Params: models.TempoApiParams{
 				ConnectionId: data.Options.ConnectionId,
+				TeamId:       data.Options.TeamId,
 			},
 			Table: RAW_WORKLOG_TABLE,
 		},
@@ -74,6 +75,7 @@ func ExtractWorklogs(taskCtx plugin.SubTaskContext) errors.Error {
 			worklog := &models.TempoWorklog{
 				ConnectionId:     data.Options.ConnectionId,
 				TempoWorklogId:   apiWorklog.TempoWorklogId,
+				TeamId:           data.Options.TeamId,
 				IssueId:          apiWorklog.Issue.Id,
 				TimeSpentSeconds: apiWorklog.TimeSpentSeconds,
 				BillableSeconds:  apiWorklog.BillableSeconds,