feat: gitlab extractors and convertors support incremental mode

backend/helpers/pluginhelper/api/api_extractor_stateful.go

@@ -18,6 +18,7 @@ limitations under the License.
 package api
 
 import (
+	"encoding/json"
 	"reflect"
 
 	"github.com/apache/incubator-devlake/core/dal"
@@ -27,9 +28,10 @@ import (
 )
 
 // StatefulApiExtractorArgs is a struct that contains the arguments for a stateful api extractor
-type StatefulApiExtractorArgs struct {
+type StatefulApiExtractorArgs[InputType any] struct {
 	*SubtaskCommonArgs
-	Extract func(row *RawData) ([]any, errors.Error)
+	BeforeExtract func(issue *InputType, stateManager *SubtaskStateManager) errors.Error
+	Extract       func(body *InputType, row *RawData) ([]any, errors.Error)
 }
 
 // StatefulApiExtractor is a struct that manages the stateful API extraction process.
@@ -41,7 +43,7 @@ type StatefulApiExtractorArgs struct {
 //
 // Example:
 //
-//	extractor, err := api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs{
+//	extractor, err := api.NewStatefulApiExtractor(&api.StatefulApiExtractorArgs[apiv2models.Issue]{
 //	  SubtaskCommonArgs: &api.SubtaskCommonArgs{
 //	    SubTaskContext: subtaskCtx,
 //	    Table:          RAW_ISSUE_TABLE,
@@ -54,8 +56,17 @@ type StatefulApiExtractorArgs struct {
 //	                            // Ensure that the configuration is serializable and contains only public fields.
 //	                            // It is also recommended that the configuration includes only the necessary fields used by the extractor.
 //	..},
-//	  Extract: func(row *api.RawData) ([]interface{}, errors.Error) {
-//	    return extractIssues(data, config, row, userFieldMap)
+//	  BeforeExtract: func(body *IssuesResponse, stateManager *api.SubtaskStateManager) errors.Error {
+//	    if stateManager.IsIncremental() {
+//	      // It is important to delete all existing child-records under DiffSync Mode
+//	      err := db.Delete(
+//	        &models.JiraIssueLabel{},
+//	        dal.Where("connection_id = ? AND issue_id = ?", data.Options.ConnectionId, body.Id),
+//	      )
+//	    }
+//	    return nil
+//	  },
+//	  Extract: func(apiIssue *apiv2models.Issue, row *api.RawData) ([]interface{}, errors.Error) {
 //	  },
 //	})
 //
@@ -64,25 +75,25 @@ type StatefulApiExtractorArgs struct {
 //	}
 //
 // return extractor.Execute()
-type StatefulApiExtractor struct {
-	*StatefulApiExtractorArgs
+type StatefulApiExtractor[InputType any] struct {
+	*StatefulApiExtractorArgs[InputType]
 	*SubtaskStateManager
 }
 
 // NewStatefulApiExtractor creates a new StatefulApiExtractor
-func NewStatefulApiExtractor(args *StatefulApiExtractorArgs) (*StatefulApiExtractor, errors.Error) {
+func NewStatefulApiExtractor[InputType any](args *StatefulApiExtractorArgs[InputType]) (*StatefulApiExtractor[InputType], errors.Error) {
 	stateManager, err := NewSubtaskStateManager(args.SubtaskCommonArgs)
 	if err != nil {
 		return nil, err
 	}
-	return &StatefulApiExtractor{
+	return &StatefulApiExtractor[InputType]{
 		StatefulApiExtractorArgs: args,
 		SubtaskStateManager:      stateManager,
 	}, nil
 }
 
 // Execute sub-task
-func (extractor *StatefulApiExtractor) Execute() errors.Error {
+func (extractor *StatefulApiExtractor[InputType]) Execute() errors.Error {
 	// load data from database
 	db := extractor.GetDal()
 	logger := extractor.GetLogger()
@@ -135,7 +146,20 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
 			return errors.Default.Wrap(err, "error fetching row")
 		}
 
-		results, err := extractor.Extract(row)
+		body := new(InputType)
+		err = errors.Convert(json.Unmarshal(row.Data, body))
+		if err != nil {
+			return err
+		}
+
+		if extractor.BeforeExtract != nil {
+			err = extractor.BeforeExtract(body, extractor.SubtaskStateManager)
+			if err != nil {
+				return err
+			}
+		}
+
+		results, err := extractor.Extract(body, row)
 		if err != nil {
 			return errors.Default.Wrap(err, "error calling plugin Extract implementation")
 		}
@@ -169,4 +193,4 @@ func (extractor *StatefulApiExtractor) Execute() errors.Error {
 	return extractor.SubtaskStateManager.Close()
 }
 
-var _ plugin.SubTask = (*StatefulApiExtractor)(nil)
+var _ plugin.SubTask = (*StatefulApiExtractor[any])(nil)

backend/helpers/pluginhelper/api/api_rawdata.go

@@ -34,7 +34,7 @@ type RawData struct {
 	Data      []byte
 	Url       string
 	Input     json.RawMessage `gorm:"type:json"`
-	CreatedAt time.Time
+	CreatedAt time.Time       `gorm:"index"`
 }
 
 type TaskOptions interface {

backend/helpers/pluginhelper/api/data_convertor_stateful.go

@@ -27,9 +27,10 @@ import (
 
 type StatefulDataConverterArgs[InputType any] struct {
 	*SubtaskCommonArgs
-	Input     func(*SubtaskStateManager) (dal.Rows, errors.Error)
-	Convert   func(row *InputType) ([]any, errors.Error)
-	BatchSize int
+	Input         func(*SubtaskStateManager) (dal.Rows, errors.Error)
+	BeforeConvert func(issue *InputType, stateManager *SubtaskStateManager) errors.Error
+	Convert       func(row *InputType) ([]any, errors.Error)
+	BatchSize     int
 }
 
 // StatefulDataConverter is a struct that manages the stateful data conversion process.
@@ -47,7 +48,7 @@ type StatefulDataConverterArgs[InputType any] struct {
 //
 // Example:
 //
-// converter, err := api.NewStatefulDataConverter[models.JiraIssue](&api.StatefulDataConverterArgs[models.JiraIssue]{
+// converter, err := api.NewStatefulDataConverter(&api.StatefulDataConverterArgs[models.JiraIssue]{
 // 	SubtaskCommonArgs: &api.SubtaskCommonArgs{
 // 		SubTaskContext: subtaskCtx,
 // 		Table:          RAW_ISSUE_TABLE,
@@ -78,6 +79,15 @@ type StatefulDataConverterArgs[InputType any] struct {
 // 		}
 // 		return db.Cursor(clauses...)
 // 	},
+//	BeforeConvert: func(jiraIssue *models.GitlabMergeRequest, stateManager *api.SubtaskStateManager) errors.Error {
+//		// It is important to delete all existing child-records under DiffSync Mode
+//		issueId := issueIdGen.Generate(data.Options.ConnectionId, jiraIssue.IssueId)
+//		if err := db.Delete(&ticket.IssueAssignee{}, dal.Where("issue_id = ?", issueId)); err != nil {
+//			return err
+//		}
+//		...
+//		return nil
+//	},
 // 	Convert: func(jiraIssue *models.JiraIssue) ([]interface{}, errors.Error) {
 // 	},
 // })
@@ -93,7 +103,6 @@ type StatefulDataConverter[InputType any] struct {
 }
 
 func NewStatefulDataConverter[
-	OptType any,
 	InputType any,
 ](
 	args *StatefulDataConverterArgs[InputType],
@@ -146,6 +155,13 @@ func (converter *StatefulDataConverter[InputType]) Execute() errors.Error {
 			return errors.Default.Wrap(err, "error fetching rows")
 		}
 
+		if converter.BeforeConvert != nil {
+			err = converter.BeforeConvert(inputRow, converter.SubtaskStateManager)
+			if err != nil {
+				return err
+			}
+		}
+
 		results, err := converter.Convert(inputRow)
 		if err != nil {
 			return errors.Default.Wrap(err, "error calling Converter plugin implementation")

backend/helpers/pluginhelper/api/enrich_with_regex.go

@@ -107,6 +107,14 @@ func (r *RegexEnricher) ReturnNameIfOmittedOrMatched(name string, targets ...str
 	return r.ReturnNameIfMatched(name, targets...)
 }
 
+func (r *RegexEnricher) PlainMap() map[string]string {
+	m := make(map[string]string)
+	for k, v := range r.regexpMap {
+		m[k] = v.String()
+	}
+	return m
+}
+
 // TryAdd a named regexp if given pattern is not empty
 func (r *RegexEnricher) TryAddList(name string, patterns ...string) errors.Error {
 	if _, ok := r.regexMapList[name]; ok {

backend/helpers/pluginhelper/api/subtask_state_manager.go

@@ -34,7 +34,7 @@ type SubtaskCommonArgs struct {
 	plugin.SubTaskContext
 	Table         string // raw table name
 	Params        any    // for filtering rows belonging to the scope (jira board, github repo) of the subtask
-	SubtaskConfig any    // for determining whether the subtask should run in incremental or full sync mode
+	SubtaskConfig any    // for determining whether the subtask should run in Incremental or Full-Sync mode by comparing with the previous config to see if it changed
 	BatchSize     int    // batch size for saving data
 }
 

backend/impls/dalgorm/encdec_serializer.go

@@ -91,6 +91,11 @@ func (es *EncDecSerializer) Value(ctx context.Context, field *schema.Field, dst
 		}
 		target = string(b)
 	}
+	if field.GORMDataType == "string" {
+		println("field.GORMDataType == string", field.Size)
+		gormTag, ok := field.Tag.Lookup("gorm")
+		println(ok, gormTag)
+	}
 	return plugin.Encrypt(es.encryptionSecret, target)
 }
 

backend/plugins/gitlab/e2e/mr_commits_test.go

@@ -133,37 +133,4 @@ func TestGitlabMrCommitDataFlow(t *testing.T) {
 		CSVRelPath:  "./snapshot_tables/pull_request_commits.csv",
 		IgnoreTypes: []interface{}{common.Model{}},
 	})
-
-	// verify conversion
-	dataflowTester.FlushTabler(&code.Commit{})
-	dataflowTester.FlushTabler(&code.RepoCommit{})
-	dataflowTester.Subtask(tasks.ConvertCommitsMeta, taskData)
-	dataflowTester.VerifyTable(
-		code.Commit{},
-		"./snapshot_tables/commits.csv",
-		e2ehelper.ColumnWithRawData(
-			"sha",
-			"additions",
-			"deletions",
-			"dev_eq",
-			"message",
-			"author_name",
-			"author_email",
-			"authored_date",
-			"author_id",
-			"committer_name",
-			"committer_email",
-			"committed_date",
-			"committer_id",
-		),
-	)
-
-	dataflowTester.VerifyTable(
-		code.RepoCommit{},
-		"./snapshot_tables/repo_commits.csv",
-		e2ehelper.ColumnWithRawData(
-			"repo_id",
-			"commit_sha",
-		),
-	)
 }

backend/plugins/gitlab/e2e/mr_detail_test.go

@@ -41,6 +41,8 @@ func TestGitlabMrDetailDataFlow(t *testing.T) {
 		},
 	}
 	// import raw data table
+	dataflowTester.FlushTabler(&code.PullRequestAssignee{})
+	dataflowTester.FlushTabler(&code.PullRequestReviewer{})
 	dataflowTester.ImportCsvIntoRawTable("./raw_tables/_raw_gitlab_api_merge_requests.csv",
 		"_raw_gitlab_api_merge_request_details")