fix(github_graphql): remove stale local issue rows when an issue no longer resolves (#8820)

* fix(github_graphql): remove stale local issue rows

* fix(github_graphql): prevent stale issue cleanup from crossing repository boundaries

The previous missing-issue cleanup only keyed deletes by connection and issue identifiers, which could reach too broadly when an issue was transferred between repositories under the same DevLake connection. This change scopes cleanup to the source repository by carrying RepoId through the refresh path and requiring the source RawDataOrigin parameters before deleting stale tool rows.

Constraint: Keep the fix inside the existing github_graphql collector path without schema changes
Rejected: Continue deleting by connection_id + github_id only | could remove destination-repository rows after transfers
Rejected: Disable stale cleanup for all missing issues | would leave deleted-issue failures recurring
Confidence: medium
Scope-risk: moderate
Directive: Preserve source-scope guards if future cleanup paths touch transferred issues again
Tested: go test ./plugins/github_graphql/tasks; go build ./helpers/pluginhelper/api ./plugins/github_graphql/tasks
Not-tested: End-to-end transfer scenario against a live GitHub repository

Author: squatboy

backend/helpers/pluginhelper/api/graphql_collector.go

@@ -276,14 +276,18 @@ func (collector *GraphqlCollector) fetchAsync(reqData *GraphqlRequestData, handl
 	}
 	if len(dataErrors) > 0 {
 		if !collector.args.IgnoreQueryErrors {
+			hasNonIgnorableDataErrors := false
 			for _, dataError := range dataErrors {
-				if strings.Contains(dataError.Error(), "Could not resolve to an Issue") {
-					logger.Warn(nil, "Issue may have been transferred.")
-				} else {
-					collector.checkError(errors.Default.Wrap(dataError, `graphql query got error`))
+				if isIgnorableGraphqlQueryError(dataError) {
+					logger.Warn(nil, "Issue may have been transferred or deleted.")
+					continue
 				}
+				hasNonIgnorableDataErrors = true
+				collector.checkError(errors.Default.Wrap(dataError, `graphql query got error`))
+			}
+			if hasNonIgnorableDataErrors {
+				return
 			}
-			return
 		}
 		// else: error will deal by ResponseParserWithDataErrors
 	}
@@ -344,4 +348,8 @@ func (collector *GraphqlCollector) HasError() bool {
 	return len(collector.workerErrors) > 0
 }
 
+func isIgnorableGraphqlQueryError(err error) bool {
+	return err != nil && strings.Contains(err.Error(), "Could not resolve to an Issue")
+}
+
 var _ plugin.SubTask = (*GraphqlCollector)(nil)

backend/helpers/pluginhelper/api/graphql_collector_test.go

@@ -0,0 +1,31 @@
+/*
+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 api
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsIgnorableGraphqlQueryError(t *testing.T) {
+	assert.True(t, isIgnorableGraphqlQueryError(errors.New("Could not resolve to an Issue with the number of 17.")))
+	assert.False(t, isIgnorableGraphqlQueryError(errors.New("some other graphql error")))
+	assert.False(t, isIgnorableGraphqlQueryError(nil))
+}

backend/plugins/github_graphql/tasks/issue_collector.go

@@ -20,11 +20,14 @@ package tasks
 import (
 	"encoding/json"
 	"reflect"
+	"sort"
 	"strings"
 	"time"
 
 	"github.com/apache/incubator-devlake/core/dal"
 	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/log"
+	"github.com/apache/incubator-devlake/core/models/common"
 	"github.com/apache/incubator-devlake/core/plugin"
 	"github.com/apache/incubator-devlake/core/utils"
 	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
@@ -49,7 +52,8 @@ type GraphqlQueryIssueWrapper struct {
 }
 
 type GraphqlQueryIssueDetailWrapper struct {
-	RateLimit struct {
+	requestedIssues map[int]missingGithubIssueRef
+	RateLimit       struct {
 		Cost int
 	}
 	Repository struct {
@@ -84,6 +88,22 @@ type GraphqlQueryIssue struct {
 	} `graphql:"labels(first: 100)"`
 }
 
+type missingGithubIssueRef struct {
+	ConnectionId  uint64
+	RepoId        int
+	GithubId      int
+	Number        int
+	RawDataOrigin common.RawDataOrigin
+}
+
+type missingGithubIssueCleanupScope struct {
+	ConnectionId  uint64
+	RepoId        int
+	GithubId      int
+	Number        int
+	RawDataOrigin common.RawDataOrigin
+}
+
 var CollectIssuesMeta = plugin.SubTaskMeta{
 	Name:             "Collect Issues",
 	EntryPoint:       CollectIssues,
@@ -175,12 +195,20 @@ func CollectIssues(taskCtx plugin.SubTaskContext) errors.Error {
 			ownerName := strings.Split(data.Options.Name, "/")
 			inputIssues := reqData.Input.([]interface{})
 			outputIssues := []map[string]interface{}{}
+			query.requestedIssues = make(map[int]missingGithubIssueRef, len(inputIssues))
 			for _, i := range inputIssues {
 				inputIssue := i.(*models.GithubIssue)
 				outputIssues = append(outputIssues, map[string]interface{}{
 					`number`: graphql.Int(inputIssue.Number),
 				})
 				issueUpdatedAt[inputIssue.Number] = inputIssue.GithubUpdatedAt
+				query.requestedIssues[inputIssue.Number] = missingGithubIssueRef{
+					ConnectionId:  inputIssue.ConnectionId,
+					RepoId:        inputIssue.RepoId,
+					GithubId:      inputIssue.GithubId,
+					Number:        inputIssue.Number,
+					RawDataOrigin: inputIssue.RawDataOrigin,
+				}
 			}
 			variables := map[string]interface{}{
 				"issue": outputIssues,
@@ -193,10 +221,17 @@ func CollectIssues(taskCtx plugin.SubTaskContext) errors.Error {
 			query := queryWrapper.(*GraphqlQueryIssueDetailWrapper)
 			issues := query.Repository.Issues
 			for _, rawL := range issues {
+				if rawL.DatabaseId == 0 || rawL.Number == 0 {
+					continue
+				}
 				if rawL.UpdatedAt.After(issueUpdatedAt[rawL.Number]) {
 					messages = append(messages, errors.Must1(json.Marshal(rawL)))
 				}
 			}
+			missingIssues := findMissingGithubIssues(query.requestedIssues, issues)
+			if len(missingIssues) > 0 {
+				err = cleanupMissingGithubIssues(db, taskCtx.GetLogger(), missingIssues)
+			}
 			return
 		},
 	})
@@ -206,3 +241,159 @@ func CollectIssues(taskCtx plugin.SubTaskContext) errors.Error {
 
 	return apiCollector.Execute()
 }
+
+func findMissingGithubIssues(requestedIssues map[int]missingGithubIssueRef, resolvedIssues []GraphqlQueryIssue) []missingGithubIssueRef {
+	if len(requestedIssues) == 0 {
+		return nil
+	}
+
+	resolvedNumbers := make(map[int]struct{}, len(resolvedIssues))
+	for _, issue := range resolvedIssues {
+		if issue.DatabaseId == 0 || issue.Number == 0 {
+			continue
+		}
+		resolvedNumbers[issue.Number] = struct{}{}
+	}
+
+	missingIssues := make([]missingGithubIssueRef, 0)
+	for number, issue := range requestedIssues {
+		if _, ok := resolvedNumbers[number]; ok {
+			continue
+		}
+		missingIssues = append(missingIssues, issue)
+	}
+	sort.Slice(missingIssues, func(i, j int) bool {
+		return missingIssues[i].Number < missingIssues[j].Number
+	})
+	return missingIssues
+}
+
+func cleanupMissingGithubIssues(db dal.Dal, logger log.Logger, issues []missingGithubIssueRef) errors.Error {
+	var allErrors []error
+	for _, issue := range issues {
+		scope, ok := buildMissingGithubIssueCleanupScope(issue)
+		if !ok {
+			logger.Warn(nil, "GitHub issue #%d no longer resolves from the source API, but source scope is incomplete so stale cleanup is skipped", issue.Number)
+			continue
+		}
+		logger.Warn(nil, "GitHub issue #%d no longer resolves from the source API, deleting stale local data for the current repository scope", issue.Number)
+		err := cleanupMissingGithubIssue(db, scope)
+		if err != nil {
+			allErrors = append(allErrors, err)
+		}
+	}
+	return errors.Default.Combine(allErrors)
+}
+
+func buildMissingGithubIssueCleanupScope(issue missingGithubIssueRef) (*missingGithubIssueCleanupScope, bool) {
+	if issue.ConnectionId == 0 || issue.RepoId == 0 || issue.GithubId == 0 || issue.RawDataOrigin.RawDataTable == "" || issue.RawDataOrigin.RawDataParams == "" {
+		return nil, false
+	}
+	return &missingGithubIssueCleanupScope{
+		ConnectionId:  issue.ConnectionId,
+		RepoId:        issue.RepoId,
+		GithubId:      issue.GithubId,
+		Number:        issue.Number,
+		RawDataOrigin: issue.RawDataOrigin,
+	}, true
+}
+
+func (scope *missingGithubIssueCleanupScope) issueScopedClauses() []dal.Clause {
+	return []dal.Clause{
+		dal.Where(
+			"connection_id = ? AND issue_id = ? AND _raw_data_table = ? AND _raw_data_params = ?",
+			scope.ConnectionId,
+			scope.GithubId,
+			scope.RawDataOrigin.RawDataTable,
+			scope.RawDataOrigin.RawDataParams,
+		),
+	}
+}
+
+func (scope *missingGithubIssueCleanupScope) assigneeScopedClauses() []dal.Clause {
+	return []dal.Clause{
+		dal.Where(
+			"connection_id = ? AND repo_id = ? AND issue_id = ? AND _raw_data_table = ? AND _raw_data_params = ?",
+			scope.ConnectionId,
+			scope.RepoId,
+			scope.GithubId,
+			scope.RawDataOrigin.RawDataTable,
+			scope.RawDataOrigin.RawDataParams,
+		),
+	}
+}
+
+func (scope *missingGithubIssueCleanupScope) githubIssueScopedClauses() []dal.Clause {
+	return []dal.Clause{
+		dal.Where(
+			"connection_id = ? AND repo_id = ? AND github_id = ? AND _raw_data_table = ? AND _raw_data_params = ?",
+			scope.ConnectionId,
+			scope.RepoId,
+			scope.GithubId,
+			scope.RawDataOrigin.RawDataTable,
+			scope.RawDataOrigin.RawDataParams,
+		),
+	}
+}
+
+func (scope *missingGithubIssueCleanupScope) rawDataScopedClauses() []dal.Clause {
+	if scope.RawDataOrigin.RawDataId == 0 {
+		return nil
+	}
+	return []dal.Clause{dal.Where("id = ?", scope.RawDataOrigin.RawDataId)}
+}
+
+func cleanupMissingGithubIssue(db dal.Dal, scope *missingGithubIssueCleanupScope) errors.Error {
+	deleteByIssueId := func(model any, table string) errors.Error {
+		clauses := append([]dal.Clause{dal.From(table)}, scope.issueScopedClauses()...)
+		err := db.Delete(model, clauses...)
+		if err != nil {
+			return errors.Default.Wrap(err, "failed to delete stale github issue data from "+table)
+		}
+		return nil
+	}
+
+	err := deleteByIssueId(&models.GithubIssueComment{}, models.GithubIssueComment{}.TableName())
+	if err != nil {
+		return err
+	}
+	err = deleteByIssueId(&models.GithubIssueEvent{}, models.GithubIssueEvent{}.TableName())
+	if err != nil {
+		return err
+	}
+	err = deleteByIssueId(&models.GithubIssueLabel{}, models.GithubIssueLabel{}.TableName())
+	if err != nil {
+		return err
+	}
+	err = db.Delete(
+		&models.GithubIssueAssignee{},
+		append([]dal.Clause{dal.From(models.GithubIssueAssignee{}.TableName())}, scope.assigneeScopedClauses()...)...,
+	)
+	if err != nil {
+		return errors.Default.Wrap(err, "failed to delete stale github issue assignees")
+	}
+	err = db.Delete(
+		&models.GithubPrIssue{},
+		append([]dal.Clause{dal.From(models.GithubPrIssue{}.TableName())}, scope.issueScopedClauses()...)...,
+	)
+	if err != nil {
+		return errors.Default.Wrap(err, "failed to delete stale github pull request issue links")
+	}
+	err = db.Delete(
+		&models.GithubIssue{},
+		append([]dal.Clause{dal.From(models.GithubIssue{}.TableName())}, scope.githubIssueScopedClauses()...)...,
+	)
+	if err != nil {
+		return errors.Default.Wrap(err, "failed to delete stale github issue")
+	}
+	if rawDataClauses := scope.rawDataScopedClauses(); len(rawDataClauses) > 0 {
+		err = db.Delete(
+			&api.RawData{},
+			append([]dal.Clause{dal.From(scope.RawDataOrigin.RawDataTable)}, rawDataClauses...)...,
+		)
+		if err != nil {
+			return errors.Default.Wrap(err, "failed to delete stale raw github issue")
+		}
+	}
+	return nil
+}