From 931c4026db2e19caeca18dcd55bbefa0b8b40358 Mon Sep 17 00:00:00 2001 From: Aleks Petrov Date: Mon, 13 Jul 2026 20:55:16 +0200 Subject: [PATCH] =?UTF-8?q?fix(linear):=20SyncAdapter=20correctness=20?= =?UTF-8?q?=E2=80=94=20state=20name=E2=86=92id,=20paginated=20idem=20scan,?= =?UTF-8?q?=20on-or-after=20delta,=20UpdateFields=20validation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Post-merge review of #94 found four defects in sdk/integrations/linear/sync.go: TransitionState sent a state NAME straight through as a GraphQL stateId (round-trip broken); AddComment's idempotency scan only checked the first comments page, letting retries duplicate-post once a marker fell off page 1; ListUpdatedSince used gt+second-truncated timestamps, silently dropping issues updated exactly at the watermark; and UpdateFields silently ignored unknown keys and wrongly-typed values instead of rejecting the patch. --- sdk/integrations/linear/sync.go | 240 +++++++++++++++++++++------ sdk/integrations/linear/sync_test.go | 178 +++++++++++++++++++- 2 files changed, 359 insertions(+), 59 deletions(-) diff --git a/sdk/integrations/linear/sync.go b/sdk/integrations/linear/sync.go index 8839f56..df19e51 100644 --- a/sdk/integrations/linear/sync.go +++ b/sdk/integrations/linear/sync.go @@ -3,6 +3,7 @@ package linear import ( "context" "fmt" + "regexp" "strings" "sync" "time" @@ -10,6 +11,15 @@ import ( "github.com/qf-studio/studio-sdk/sdk/core" ) +// uuidPattern matches a Linear-style UUID (e.g. workflow state or issue id), +// distinguishing an already-resolved id from a human-readable state name +// passed into TransitionState. +var uuidPattern = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`) + +func looksLikeUUID(s string) bool { + return uuidPattern.MatchString(s) +} + // Compile-time interface assertion. var _ core.SyncCapable = (*SyncAdapter)(nil) @@ -27,13 +37,20 @@ type SyncAdapter struct { teamIDMu sync.RWMutex teamIDCache map[string]string + + // workflowStateMu/workflowStateCache cache team-key -> (state name -> + // state id) so repeated TransitionState calls with a state name don't + // re-fetch the team's workflow states on every call. + workflowStateMu sync.RWMutex + workflowStateCache map[string]map[string]string } // NewSyncAdapter creates a SyncAdapter backed by client. func NewSyncAdapter(client *Client) *SyncAdapter { return &SyncAdapter{ - client: client, - teamIDCache: make(map[string]string), + client: client, + teamIDCache: make(map[string]string), + workflowStateCache: make(map[string]map[string]string), } } @@ -102,15 +119,18 @@ func priorityRankFromNormalized(p string) int { } // ListUpdatedSince returns issues in the team keyed by projectID whose -// updatedAt is strictly greater than since, one page at a time following the -// #85 cursor-following convention (page is the opaque "after" cursor). +// updatedAt is on or after since (inclusive), one page at a time following +// the #85 cursor-following convention (page is the opaque "after" cursor). +// since is formatted with nanosecond precision so an issue updated exactly at +// the watermark is never dropped; callers are expected to dedupe returned +// snapshots by NativeID across successive calls at the same watermark. func (s *SyncAdapter) ListUpdatedSince(ctx context.Context, projectID string, since time.Time, page core.Cursor) ([]core.IssueSnapshot, core.Cursor, error) { query := ` query SyncIssuesSince($teamId: String!, $since: DateTimeOrDuration!, $first: Int!, $after: String) { issues( filter: { team: { key: { eq: $teamId } } - updatedAt: { gt: $since } + updatedAt: { gte: $since } } first: $first after: $after @@ -124,7 +144,7 @@ func (s *SyncAdapter) ListUpdatedSince(ctx context.Context, projectID string, si variables := map[string]interface{}{ "teamId": projectID, - "since": since.UTC().Format(time.RFC3339), + "since": since.UTC().Format(time.RFC3339Nano), "first": listIssuesPageSize, } if page != "" { @@ -201,42 +221,59 @@ func (s *SyncAdapter) GetIssue(ctx context.Context, nativeID string) (core.Issue // UpdateFields applies a partial field patch to nativeID via the Linear // issueUpdate mutation. Recognized fields keys: "title" (string), // "description" (string), "labels" ([]string of label names), "priority" -// (a core.Priority* normalized string). Unrecognized keys are ignored. +// (a core.Priority* normalized string). An unrecognized key, or a recognized +// key with the wrong value type, is rejected with an error and no field is +// applied. func (s *SyncAdapter) UpdateFields(ctx context.Context, nativeID string, fields core.FieldPatch) (core.IssueSnapshot, error) { + for key := range fields { + switch key { + case "title", "description", "priority", "labels": + default: + return core.IssueSnapshot{}, fmt.Errorf("unrecognized field key %q", key) + } + } + variables := map[string]interface{}{"id": nativeID} if v, ok := fields["title"]; ok { - if str, ok := v.(string); ok { - variables["title"] = str + str, ok := v.(string) + if !ok { + return core.IssueSnapshot{}, fmt.Errorf("field %q must be a string, got %T", "title", v) } + variables["title"] = str } if v, ok := fields["description"]; ok { - if str, ok := v.(string); ok { - variables["description"] = str + str, ok := v.(string) + if !ok { + return core.IssueSnapshot{}, fmt.Errorf("field %q must be a string, got %T", "description", v) } + variables["description"] = str } if v, ok := fields["priority"]; ok { - if str, ok := v.(string); ok { - variables["priority"] = priorityRankFromNormalized(str) + str, ok := v.(string) + if !ok { + return core.IssueSnapshot{}, fmt.Errorf("field %q must be a string, got %T", "priority", v) } + variables["priority"] = priorityRankFromNormalized(str) } if v, ok := fields["labels"]; ok { - names := toStringSlice(v) - if names != nil { - issue, err := s.client.GetIssue(ctx, nativeID) + names, ok := toStringSlice(v) + if !ok { + return core.IssueSnapshot{}, fmt.Errorf("field %q must be a []string, got %T", "labels", v) + } + issue, err := s.client.GetIssue(ctx, nativeID) + if err != nil { + return core.IssueSnapshot{}, fmt.Errorf("failed to resolve team for label update on %s: %w", nativeID, err) + } + labelIDs := make([]string, 0, len(names)) + for _, name := range names { + id, err := s.client.GetOrCreateLabel(ctx, issue.Team.Key, name, "#8b949e") if err != nil { - return core.IssueSnapshot{}, fmt.Errorf("failed to resolve team for label update on %s: %w", nativeID, err) + return core.IssueSnapshot{}, fmt.Errorf("failed to resolve label %q: %w", name, err) } - labelIDs := make([]string, 0, len(names)) - for _, name := range names { - id, err := s.client.GetOrCreateLabel(ctx, issue.Team.Key, name, "#8b949e") - if err != nil { - return core.IssueSnapshot{}, fmt.Errorf("failed to resolve label %q: %w", name, err) - } - labelIDs = append(labelIDs, id) - } - variables["labelIds"] = labelIDs + labelIDs = append(labelIDs, id) } + variables["labelIds"] = labelIDs } if len(variables) == 1 { @@ -270,9 +307,78 @@ func (s *SyncAdapter) UpdateFields(ctx context.Context, nativeID string, fields return toSnapshot(result.IssueUpdate.Issue.toIssue()), nil } -// TransitionState moves nativeID to providerState (a workflow state id). +// TransitionState moves nativeID to providerState, which may be either a +// Linear workflow-state UUID (as returned in core.IssueSnapshot's internal +// representation, or by GetTeamDoneStateID) or a workflow-state NAME (the +// value core.IssueSnapshot.State actually carries). A UUID is passed straight +// through to Client.UpdateIssueState; a name is first resolved to the state's +// id via a per-team workflow-state lookup, since the underlying GraphQL +// mutation only accepts a stateId. func (s *SyncAdapter) TransitionState(ctx context.Context, nativeID, providerState string) error { - return s.client.UpdateIssueState(ctx, nativeID, providerState) + stateID := providerState + if !looksLikeUUID(providerState) { + issue, err := s.client.GetIssue(ctx, nativeID) + if err != nil { + return fmt.Errorf("failed to resolve issue %s to look up its team: %w", nativeID, err) + } + id, err := s.resolveWorkflowStateID(ctx, issue.Team.Key, providerState) + if err != nil { + return fmt.Errorf("failed to resolve state %q for team %s: %w", providerState, issue.Team.Key, err) + } + stateID = id + } + return s.client.UpdateIssueState(ctx, nativeID, stateID) +} + +// resolveWorkflowStateID resolves stateName to its workflow-state id within +// teamKey's team, caching all of the team's states on first lookup. +func (s *SyncAdapter) resolveWorkflowStateID(ctx context.Context, teamKey, stateName string) (string, error) { + s.workflowStateMu.RLock() + if states, ok := s.workflowStateCache[teamKey]; ok { + id, ok := states[stateName] + s.workflowStateMu.RUnlock() + if ok { + return id, nil + } + return "", fmt.Errorf("no workflow state named %q found for team %s", stateName, teamKey) + } + s.workflowStateMu.RUnlock() + + query := ` + query SyncResolveWorkflowStates($teamKey: String!) { + workflowStates(filter: { team: { key: { eq: $teamKey } } }) { + nodes { id name } + } + } + ` + + var result struct { + WorkflowStates struct { + Nodes []struct { + ID string `json:"id"` + Name string `json:"name"` + } `json:"nodes"` + } `json:"workflowStates"` + } + + if err := s.client.Execute(ctx, query, map[string]interface{}{"teamKey": teamKey}, &result); err != nil { + return "", err + } + + states := make(map[string]string, len(result.WorkflowStates.Nodes)) + for _, n := range result.WorkflowStates.Nodes { + states[n.Name] = n.ID + } + + s.workflowStateMu.Lock() + s.workflowStateCache[teamKey] = states + s.workflowStateMu.Unlock() + + id, ok := states[stateName] + if !ok { + return "", fmt.Errorf("no workflow state named %q found for team %s", stateName, teamKey) + } + return id, nil } // syncCommentMarker returns the idempotency marker embedded in comment @@ -281,42 +387,63 @@ func syncCommentMarker(idemKey string) string { return fmt.Sprintf("", idemKey) } +// syncCommentsPageSize is the GraphQL page size used when scanning an +// issue's comments for a prior idempotency marker in AddComment. +const syncCommentsPageSize = 100 + // AddComment posts body as a comment on nativeID, embedding a // "" marker. If a comment with the same marker // already exists on the issue, AddComment is a no-op — this makes retried -// syncs idempotent. +// syncs idempotent. The existing-comments scan follows the comments +// connection's pagination cursor to completion, so a marker sitting beyond +// the first page is still found. func (s *SyncAdapter) AddComment(ctx context.Context, nativeID, body, idemKey string) error { marker := syncCommentMarker(idemKey) query := ` - query SyncIssueComments($id: String!) { + query SyncIssueComments($id: String!, $first: Int!, $after: String) { issue(id: $id) { - comments { + comments(first: $first, after: $after) { nodes { id body } + pageInfo { hasNextPage endCursor } } } } ` - var result struct { - Issue struct { - Comments struct { - Nodes []struct { - ID string `json:"id"` - Body string `json:"body"` - } `json:"nodes"` - } `json:"comments"` - } `json:"issue"` - } + after := "" + for { + variables := map[string]interface{}{"id": nativeID, "first": syncCommentsPageSize} + if after != "" { + variables["after"] = after + } - if err := s.client.Execute(ctx, query, map[string]interface{}{"id": nativeID}, &result); err != nil { - return fmt.Errorf("failed to list comments for %s: %w", nativeID, err) - } + var result struct { + Issue struct { + Comments struct { + Nodes []struct { + ID string `json:"id"` + Body string `json:"body"` + } `json:"nodes"` + PageInfo pageInfo `json:"pageInfo"` + } `json:"comments"` + } `json:"issue"` + } + + if err := s.client.Execute(ctx, query, variables, &result); err != nil { + return fmt.Errorf("failed to list comments for %s: %w", nativeID, err) + } + + for _, c := range result.Issue.Comments.Nodes { + if strings.Contains(c.Body, marker) { + return nil + } + } - for _, c := range result.Issue.Comments.Nodes { - if strings.Contains(c.Body, marker) { - return nil + if !result.Issue.Comments.PageInfo.HasNextPage { + break } + after = result.Issue.Comments.PageInfo.EndCursor } return s.client.AddComment(ctx, nativeID, body+"\n\n"+marker) @@ -421,20 +548,23 @@ func (s *SyncAdapter) resolveTeamID(ctx context.Context, teamKey string) (string // toStringSlice converts a FieldPatch value expected to be a string slice, // accepting both []string (native Go callers) and []interface{} (values -// that passed through JSON unmarshaling). Returns nil if v is neither. -func toStringSlice(v interface{}) []string { +// that passed through JSON unmarshaling). Returns ok=false if v is neither, +// or if a []interface{} contains a non-string element. +func toStringSlice(v interface{}) (out []string, ok bool) { switch t := v.(type) { case []string: - return t + return t, true case []interface{}: - out := make([]string, 0, len(t)) + out = make([]string, 0, len(t)) for _, item := range t { - if str, ok := item.(string); ok { - out = append(out, str) + str, ok := item.(string) + if !ok { + return nil, false } + out = append(out, str) } - return out + return out, true default: - return nil + return nil, false } } diff --git a/sdk/integrations/linear/sync_test.go b/sdk/integrations/linear/sync_test.go index c6eb5b0..ffcdce7 100644 --- a/sdk/integrations/linear/sync_test.go +++ b/sdk/integrations/linear/sync_test.go @@ -327,14 +327,19 @@ func TestSyncAdapter_UpdateFields_Labels(t *testing.T) { } } -func TestSyncAdapter_TransitionState(t *testing.T) { +func TestSyncAdapter_TransitionState_UUID(t *testing.T) { + const stateUUID = "12345678-90ab-cdef-1234-567890abcdef" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var reqBody GraphQLRequest if err := json.NewDecoder(r.Body).Decode(&reqBody); err != nil { t.Fatalf("failed to decode request: %v", err) } - if reqBody.Variables["stateId"] != "state-456" { - t.Errorf("variables[stateId] = %v, want state-456", reqBody.Variables["stateId"]) + if strings.Contains(reqBody.Query, "workflowStates") || strings.Contains(reqBody.Query, "issue(id:") { + t.Fatalf("state UUID input should not trigger a lookup query, got: %s", reqBody.Query) + } + if reqBody.Variables["stateId"] != stateUUID { + t.Errorf("variables[stateId] = %v, want %s", reqBody.Variables["stateId"], stateUUID) } _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(`{"issueUpdate": {"success": true}}`)}) })) @@ -343,11 +348,52 @@ func TestSyncAdapter_TransitionState(t *testing.T) { client := NewClientWithBaseURL(testutil.FakeLinearToken, server.URL) sa := NewSyncAdapter(client) - if err := sa.TransitionState(context.Background(), "issue-1", "state-456"); err != nil { + if err := sa.TransitionState(context.Background(), "issue-1", stateUUID); err != nil { t.Fatalf("TransitionState failed: %v", err) } } +func TestSyncAdapter_TransitionState_Name(t *testing.T) { + const resolvedStateID = "state-uuid-1" + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var reqBody GraphQLRequest + if err := json.NewDecoder(r.Body).Decode(&reqBody); err != nil { + t.Fatalf("failed to decode request: %v", err) + } + + switch { + case strings.Contains(reqBody.Query, "issue(id:"): + body := `{"issue": ` + issueNodeJSON("issue-1", "ENG-1") + `}` + _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(body)}) + case strings.Contains(reqBody.Query, "workflowStates"): + if reqBody.Variables["teamKey"] != "ENG" { + t.Errorf("variables[teamKey] = %v, want ENG", reqBody.Variables["teamKey"]) + } + body := `{"workflowStates": {"nodes": [ + {"id": "` + resolvedStateID + `", "name": "In Progress"}, + {"id": "state-uuid-2", "name": "Done"} + ]}}` + _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(body)}) + case strings.Contains(reqBody.Query, "issueUpdate"): + if reqBody.Variables["stateId"] != resolvedStateID { + t.Errorf("variables[stateId] = %v, want %s", reqBody.Variables["stateId"], resolvedStateID) + } + _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(`{"issueUpdate": {"success": true}}`)}) + default: + t.Fatalf("unexpected query: %s", reqBody.Query) + } + })) + defer server.Close() + + client := NewClientWithBaseURL(testutil.FakeLinearToken, server.URL) + sa := NewSyncAdapter(client) + + if err := sa.TransitionState(context.Background(), "issue-1", "In Progress"); err != nil { + t.Fatalf("TransitionState (by name) failed: %v", err) + } +} + func TestSyncAdapter_AddComment_Idempotent(t *testing.T) { commentCreateCalls := 0 existingBodies := []string{} @@ -396,6 +442,130 @@ func TestSyncAdapter_AddComment_Idempotent(t *testing.T) { } } +func TestSyncAdapter_AddComment_IdempotentAcrossPages(t *testing.T) { + commentCreateCalls := 0 + commentQueries := 0 + marker := syncCommentMarker("sync-key-page2") + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var reqBody GraphQLRequest + if err := json.NewDecoder(r.Body).Decode(&reqBody); err != nil { + t.Fatalf("failed to decode request: %v", err) + } + + switch { + case strings.Contains(reqBody.Query, "comments"): + commentQueries++ + if commentQueries == 1 { + if _, hasAfter := reqBody.Variables["after"]; hasAfter { + t.Errorf("first comments page should not set 'after'") + } + body := `{"issue": {"comments": { + "nodes": [{"id": "c1", "body": "unrelated comment"}], + "pageInfo": {"hasNextPage": true, "endCursor": "cpage-2"} + }}}` + _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(body)}) + return + } + if reqBody.Variables["after"] != "cpage-2" { + t.Errorf("second page should pass through the returned cursor, got %v", reqBody.Variables["after"]) + } + body := `{"issue": {"comments": { + "nodes": [{"id": "c2", "body": "Synced\n\n` + marker + `"}], + "pageInfo": {"hasNextPage": false, "endCursor": ""} + }}}` + _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(body)}) + case strings.Contains(reqBody.Query, "commentCreate"): + commentCreateCalls++ + _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(`{"commentCreate": {"success": true}}`)}) + default: + t.Fatalf("unexpected query: %s", reqBody.Query) + } + })) + defer server.Close() + + client := NewClientWithBaseURL(testutil.FakeLinearToken, server.URL) + sa := NewSyncAdapter(client) + + if err := sa.AddComment(context.Background(), "issue-1", "Synced", "sync-key-page2"); err != nil { + t.Fatalf("AddComment failed: %v", err) + } + if commentQueries != 2 { + t.Fatalf("commentQueries = %d, want 2 (marker found on page 2)", commentQueries) + } + if commentCreateCalls != 0 { + t.Fatalf("commentCreateCalls = %d, want 0 — marker on page 2 should have been found", commentCreateCalls) + } +} + +func TestSyncAdapter_ListUpdatedSince_OnOrAfter(t *testing.T) { + since := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var reqBody GraphQLRequest + if err := json.NewDecoder(r.Body).Decode(&reqBody); err != nil { + t.Fatalf("failed to decode request: %v", err) + } + if reqBody.Variables["since"] != since.Format(time.RFC3339Nano) { + t.Errorf("variables[since] = %v, want %s (RFC3339Nano)", reqBody.Variables["since"], since.Format(time.RFC3339Nano)) + } + if !strings.Contains(reqBody.Query, "gte") { + t.Errorf("query must use gte (on-or-after) semantics, got: %s", reqBody.Query) + } + + body := `{"issues": {"nodes": [` + issueNodeJSON("issue-1", "ENG-1") + `], + "pageInfo": {"hasNextPage": false, "endCursor": ""}}}` + _ = json.NewEncoder(w).Encode(GraphQLResponse{Data: json.RawMessage(body)}) + })) + defer server.Close() + + client := NewClientWithBaseURL(testutil.FakeLinearToken, server.URL) + sa := NewSyncAdapter(client) + + snaps, _, err := sa.ListUpdatedSince(context.Background(), "ENG", since, "") + if err != nil { + t.Fatalf("ListUpdatedSince failed: %v", err) + } + if len(snaps) != 1 { + t.Fatalf("snaps = %+v, want 1 (issue updated exactly at since should be included)", snaps) + } +} + +func TestSyncAdapter_UpdateFields_UnknownKey(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatalf("no request should be made when the patch has an unrecognized key") + })) + defer server.Close() + + client := NewClientWithBaseURL(testutil.FakeLinearToken, server.URL) + sa := NewSyncAdapter(client) + + _, err := sa.UpdateFields(context.Background(), "issue-1", core.FieldPatch{ + "title": "New title", + "assignedTo": "someone", + }) + if err == nil { + t.Fatal("UpdateFields with unknown key should return an error") + } +} + +func TestSyncAdapter_UpdateFields_WrongType(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatalf("no request should be made when a recognized field has the wrong type") + })) + defer server.Close() + + client := NewClientWithBaseURL(testutil.FakeLinearToken, server.URL) + sa := NewSyncAdapter(client) + + _, err := sa.UpdateFields(context.Background(), "issue-1", core.FieldPatch{ + "priority": 2, + }) + if err == nil { + t.Fatal("UpdateFields with wrongly-typed priority should return an error") + } +} + func TestSyncAdapter_CreateIssue(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { var reqBody GraphQLRequest