INTG-3237 - Issue Assignees feed (#462)

pkg/api/export_feeds_test.go

@@ -43,6 +43,7 @@ func TestExporterFeedClient_ExportFeeds_should_create_all_schemas_to_file(t *tes
 	filesEqualish(t, "mocks/set_1/schemas/schedule_occurrences.csv", filepath.Join(exporter.ExportPath, "schedule_occurrences.csv"))
 
 	filesEqualish(t, "mocks/set_1/schemas/training_course_progresses.csv", filepath.Join(exporter.ExportPath, "training_course_progresses.csv"))
+	filesEqualish(t, "mocks/set_1/schemas/issue_assignees.csv", filepath.Join(exporter.ExportPath, "issue_assignees.csv"))
 }
 
 func TestExporterFeedClient_ExportFeeds_should_export_all_feeds_to_file(t *testing.T) {

pkg/api/mock_feeds_test.go

@@ -159,6 +159,11 @@ func initMockFeedsSet1(httpClient *http.Client) {
 		Get("/feed/training-course-progress").
 		Reply(200).
 		File("mocks/set_1/feed_training_course_progress_1.json")
+
+	gock.New("http://localhost:9999").
+		Get("/feed/issue_assignees").
+		Reply(200).
+		File("mocks/set_1/feed_issue_assignees_1.json")
 }
 
 func initMockFeedsSet2(httpClient *http.Client) {
@@ -269,6 +274,11 @@ func initMockFeedsSet2(httpClient *http.Client) {
 		Get("/feed/training-course-progress").
 		Reply(200).
 		File("mocks/set_1/feed_training_course_progress_1.json")
+
+	gock.New("http://localhost:9999").
+		Get("/feed/issue_assignees").
+		Reply(200).
+		File("mocks/set_1/feed_issue_assignees_1.json")
 }
 
 func initMockFeedsSet3(httpClient *http.Client) {
@@ -373,6 +383,11 @@ func initMockFeedsSet3(httpClient *http.Client) {
 		Get("/feed/training-course-progress").
 		Reply(200).
 		File("mocks/set_1/feed_training_course_progress_1.json")
+
+	gock.New("http://localhost:9999").
+		Get("/feed/issue_assignees").
+		Reply(200).
+		File("mocks/set_1/feed_issue_assignees_1.json")
 }
 
 func initMockIssuesFeed(httpClient *http.Client) {

pkg/api/mocks/set_1/feed_issue_assignees_1.json

@@ -0,0 +1,26 @@
+{
+  "metadata": {
+    "next_page": null,
+    "next_page_token": ""
+  },
+  "data": [
+    {
+      "id": "test-1",
+      "issue_id": "issue-1",
+      "assignee_id": "user-1",
+      "name": "Test User",
+      "organisation_id": "role-1",
+      "modified_at": "2024-03-13T03:19:34Z",
+      "type": "user"
+    },
+    {
+      "id": "test-2",
+      "issue_id": "issue-1",
+      "assignee_id": "user-1",
+      "name": "13 Mar",
+      "organisation_id": "role-1",
+      "modified_at": "2024-03-13T03:20:56Z",
+      "type": "user"
+    }
+  ]
+}

pkg/api/mocks/set_1/outputs/issue_assignees.csv

@@ -0,0 +1,3 @@
+id,issue_id,assignee_id,name,organisation_id,modified_at,type
+test-1,issue-1,user-1,Test User,role-1,--date--,user
+test-2,issue-2,user-1,13 Mar,role-1,--date--,user

pkg/api/mocks/set_1/schemas/formatted/issue_assignees.txt

@@ -0,0 +1,11 @@
++-----------------+----------+-------------+
+|      NAME       |   TYPE   | PRIMARY KEY |
++-----------------+----------+-------------+
+| id              | TEXT     | true        |
+| issue_id        | TEXT     |             |
+| assignee_id     | TEXT     |             |
+| name            | TEXT     |             |
+| organisation_id | TEXT     |             |
+| modified_at     | datetime |             |
+| type            | TEXT     |             |
++-----------------+----------+-------------+

pkg/api/mocks/set_1/schemas/issue_assignees.csv

@@ -0,0 +1 @@
+id,issue_id,assignee_id,name,organisation_id,modified_at,type

pkg/internal/feed/feed_exporter.go

@@ -310,6 +310,10 @@ func (e *ExporterFeedClient) GetFeeds() []Feed {
 			Limit:            e.configuration.ExportCourseProgressLimit,
 			CompletionStatus: "COMPLETION_STATUS_COMPLETED",
 		},
+		&IssueAssigneeFeed{
+			Incremental: false, // IssueAssignee doesn't support modified after filters
+			Limit:       e.configuration.ExportIssueLimit,
+		},
 	}
 }
 

pkg/internal/feed/feed_issue_assignee.go

@@ -0,0 +1,137 @@
+package feed
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/MickStanciu/go-fn/fn"
+	"github.com/SafetyCulture/safetyculture-exporter/pkg/httpapi"
+	"github.com/SafetyCulture/safetyculture-exporter/pkg/internal/events"
+	"github.com/SafetyCulture/safetyculture-exporter/pkg/internal/util"
+	"github.com/SafetyCulture/safetyculture-exporter/pkg/logger"
+	"time"
+)
+
+type IssueAssignee struct {
+	ID             string    `json:"id" csv:"id" gorm:"primarykey;column:id;size:36"`
+	IssueID        string    `json:"issue_id" csv:"issue_id" gorm:"index;column:issue_id;size:36"`
+	AssigneeID     string    `json:"assignee_id" csv:"assignee_id" gorm:"index;column:assignee_id;size:36"`
+	Name           string    `json:"name" csv:"name"`
+	OrganisationID string    `json:"organisation_id" csv:"organisation_id" gorm:"index;column:organisation_id;size:36"`
+	ModifiedAt     time.Time `json:"modified_at" csv:"modified_at"`
+	Type           string    `json:"type" csv:"type"`
+}
+
+type IssueAssigneeFeed struct {
+	Incremental bool
+	Limit       int
+}
+
+func (f *IssueAssigneeFeed) Name() string {
+	return "issue_assignees"
+}
+
+func (f *IssueAssigneeFeed) writeRows(exporter Exporter, rows []*IssueAssignee) error {
+	batchSize := exporter.ParameterLimit() / (len(f.Columns()) + 4)
+	err := util.SplitSliceInBatch(batchSize, rows, func(batch []*IssueAssignee) error {
+		issueIDs := fn.Map(batch, func(a *IssueAssignee) string { return a.IssueID })
+
+		if err := exporter.DeleteRowsIfExist(f, "issue_id IN (?)", issueIDs); err != nil {
+			return fmt.Errorf("delete rows: %w", err)
+		}
+
+		if err := exporter.WriteRows(f, batch); err != nil {
+			return events.WrapEventError(err, "write rows")
+		}
+		return nil
+	})
+
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// Export exports the feed to the supplied exporter
+func (f *IssueAssigneeFeed) Export(ctx context.Context, apiClient *httpapi.Client, exporter Exporter, orgID string) error {
+	l := logger.GetLogger().With("feed", f.Name(), "org_id", orgID)
+	status := GetExporterStatus()
+
+	if err := exporter.InitFeed(f, &InitFeedOptions{
+		// Delete data if incremental refresh is disabled so there is no duplicates
+		Truncate: !f.Incremental,
+	}); err != nil {
+		return events.WrapEventError(err, "init feed")
+	}
+
+	drainFn := func(resp *GetFeedResponse) error {
+		var rows []*IssueAssignee
+
+		if err := json.Unmarshal(resp.Data, &rows); err != nil {
+			return events.NewEventErrorWithMessage(err, events.ErrorSeverityError, events.ErrorSubSystemDataIntegrity, false, "map data")
+		}
+
+		numRows := len(rows)
+		if numRows != 0 {
+			if err := f.writeRows(exporter, rows); err != nil {
+				return err
+			}
+		}
+
+		status.IncrementStatus(f.Name(), int64(numRows), apiClient.Duration.Milliseconds())
+
+		l.With(
+			"downloaded", status.ReadCounter(f.Name()),
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
+		return nil
+	}
+
+	req := &GetFeedRequest{
+		InitialURL: "/feed/issue_assignees",
+		Params: GetFeedParams{
+			Limit: f.Limit,
+		},
+	}
+	if err := DrainFeed(ctx, apiClient, req, drainFn); err != nil {
+		return events.WrapEventError(err, fmt.Sprintf("feed %q", f.Name()))
+
+	}
+	return exporter.FinaliseExport(f, &[]IssueAssignee{})
+}
+
+func (f *IssueAssigneeFeed) HasRemainingInformation() bool {
+	return false
+}
+
+func (f *IssueAssigneeFeed) Model() interface{} {
+	return IssueAssignee{}
+}
+
+func (f *IssueAssigneeFeed) RowsModel() interface{} {
+	return &[]*IssueAssignee{}
+}
+
+func (f *IssueAssigneeFeed) PrimaryKey() []string {
+	return []string{"id"}
+}
+
+func (f *IssueAssigneeFeed) Columns() []string {
+	return []string{
+		"issue_id",
+		"assignee_id",
+		"type",
+		"name",
+		"organisation_id",
+		"modified_at",
+	}
+}
+
+func (f *IssueAssigneeFeed) Order() string {
+	return "issue_id, assignee_id"
+}
+
+func (f *IssueAssigneeFeed) CreateSchema(exporter Exporter) error {
+	return exporter.CreateSchema(f, &[]*IssueAssignee{})
+}

pkg/internal/feed/fixtures/schemas/formatted/issue_assignees.txt

@@ -0,0 +1,11 @@
++-----------------+----------+-------------+
+|      NAME       |   TYPE   | PRIMARY KEY |
++-----------------+----------+-------------+
+| id              | TEXT     | true        |
+| issue_id        | TEXT     |             |
+| assignee_id     | TEXT     |             |
+| name            | TEXT     |             |
+| organisation_id | TEXT     |             |
+| modified_at     | datetime |             |
+| type            | TEXT     |             |
++-----------------+----------+-------------+