diff --git a/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_worklogs.csv b/backend/plugins/tempo/e2e/raw_tables/_raw_tempo_api_worklogs.csv new file mode 100644 index 00000000000..9897ad35f1d --- /dev/null +++ b/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 diff --git a/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_worklogs_extracted.csv b/backend/plugins/tempo/e2e/snapshot_tables/_tool_tempo_worklogs_extracted.csv new file mode 100644 index 00000000000..c61ae2ab16e --- /dev/null +++ b/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","" diff --git a/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv b/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv index a100e9640dd..417b0f122f2 100644 --- a/backend/plugins/tempo/e2e/snapshot_tables/issue_worklogs.csv +++ b/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, diff --git a/backend/plugins/tempo/e2e/worklog_test.go b/backend/plugins/tempo/e2e/worklog_test.go index 1b08af889d6..3d2d5a74039 100644 --- a/backend/plugins/tempo/e2e/worklog_test.go +++ b/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) diff --git a/backend/plugins/tempo/tasks/worklog_collector.go b/backend/plugins/tempo/tasks/worklog_collector.go index 528b0b8cf44..0571956ec9f 100644 --- a/backend/plugins/tempo/tasks/worklog_collector.go +++ b/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 +} diff --git a/backend/plugins/tempo/tasks/worklog_collector_test.go b/backend/plugins/tempo/tasks/worklog_collector_test.go new file mode 100644 index 00000000000..d484d0d8869 --- /dev/null +++ b/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")) + }) +} diff --git a/backend/plugins/tempo/tasks/worklog_convertor.go b/backend/plugins/tempo/tasks/worklog_convertor.go index b5c62aa7bbf..bd6c4f25554 100644 --- a/backend/plugins/tempo/tasks/worklog_convertor.go +++ b/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 -} diff --git a/backend/plugins/tempo/tasks/worklog_extractor.go b/backend/plugins/tempo/tasks/worklog_extractor.go index 19b82c7f4b0..b8a0427d455 100644 --- a/backend/plugins/tempo/tasks/worklog_extractor.go +++ b/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,