Merge pull request #19 from abskrj/fix/nango-connection-reconcile

fix: reconcile Nango connection IDs when webhook races ahead

Author: abskrj

services/control-plane/internal/api/connections_test.go

@@ -144,10 +144,10 @@ func setupWithNango(t *testing.T) *testEnv {
 			_ = json.NewDecoder(r.Body).Decode(&req)
 			if len(req.AllowedIntegrations) > 0 {
 				if _, ok := configuredIntegrations[req.AllowedIntegrations[0]]; !ok {
-				w.WriteHeader(http.StatusBadRequest)
-				_, _ = w.Write([]byte(`{"error":"provider config not found"}`))
-				return
-			}
+					w.WriteHeader(http.StatusBadRequest)
+					_, _ = w.Write([]byte(`{"error":"provider config not found"}`))
+					return
+				}
 			}
 			w.Header().Set("Content-Type", "application/json")
 			w.WriteHeader(http.StatusOK)
@@ -157,6 +157,43 @@ func setupWithNango(t *testing.T) *testEnv {
 		case r.Method == http.MethodDelete && strings.HasPrefix(r.URL.Path, "/connection/"):
 			w.WriteHeader(http.StatusNoContent)
 
+		// GET /connections — returns connection records for reconciliation after OAuth.
+		case r.Method == http.MethodGet && r.URL.Path == "/connections":
+			type conn struct {
+				ConnectionID      string            `json:"connection_id"`
+				ProviderConfigKey string            `json:"provider_config_key"`
+				Provider          string            `json:"provider"`
+				Tags              map[string]string `json:"tags"`
+				Created           string            `json:"created"`
+			}
+			connections := []conn{
+				{
+					ConnectionID:      "mock-connection-github",
+					ProviderConfigKey: "github",
+					Provider:          "github",
+					Tags:              map[string]string{"velane_alias": "default"},
+					Created:           "2026-01-01T00:00:00Z",
+				},
+				{
+					ConnectionID:      "mock-connection-figma",
+					ProviderConfigKey: "figma",
+					Provider:          "figma",
+					Tags:              map[string]string{"velane_alias": "default"},
+					Created:           "2026-01-01T00:00:00Z",
+				},
+			}
+			for key, provider := range configuredIntegrations {
+				connections = append(connections, conn{
+					ConnectionID:      "mock-connection-" + key,
+					ProviderConfigKey: key,
+					Provider:          provider,
+					Created:           "2026-01-02T00:00:00Z",
+				})
+			}
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(http.StatusOK)
+			_ = json.NewEncoder(w).Encode(map[string]any{"connections": connections})
+
 		// GET /providers — returns a minimal provider list.
 		case r.Method == http.MethodGet && r.URL.Path == "/providers":
 			w.Header().Set("Content-Type", "application/json")
@@ -279,6 +316,26 @@ func TestConnections_Record(t *testing.T) {
 	}
 }
 
+func TestConnections_RecordBackfillsNangoConnectionID(t *testing.T) {
+	env := setupWithNango(t)
+	path := "/v1/tenant/connections"
+	profile := configureProfile(t, env, "github", "default", true)
+
+	rec := env.do(t, http.MethodPost, path, env.manageKey, map[string]any{
+		"provider":              "github",
+		"credential_profile_id": profile["id"],
+	})
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("status = %d; want 201\nbody: %s", rec.Code, rec.Body.String())
+	}
+
+	body := decodeJSON(t, rec)
+	want := "mock-connection-" + profile["nango_provider_config_key"].(string)
+	if body["nango_connection_id"] != want {
+		t.Errorf("nango_connection_id = %q; want %q", body["nango_connection_id"], want)
+	}
+}
+
 func TestConnections_ListAfterRecord(t *testing.T) {
 	env := setupWithNango(t)
 	path := "/v1/tenant/connections"

services/control-plane/internal/api/handlers/connections.go

@@ -145,6 +145,44 @@ func (h *ConnectionsHandler) RecordConnection(w http.ResponseWriter, r *http.Req
 		writeError(w, http.StatusInternalServerError, "failed to record connection")
 		return
 	}
+	if conn.NangoConnectionID == "" && h.nango != nil {
+		nangoConnID, err := h.nango.FindConnectionID(r.Context(), tenant.ID, providerConfigKey, req.Alias)
+		if err != nil {
+			h.log.Warn("record connection: failed to reconcile nango connection id",
+				zap.String("provider", req.Provider),
+				zap.String("provider_config_key", providerConfigKey),
+				zap.String("alias", req.Alias),
+				zap.Error(err),
+			)
+		} else if nangoConnID != "" {
+			if updated, err := h.store.UpdateNangoConnectionIDByProviderConfigKey(r.Context(), tenant.ID, providerConfigKey, nangoConnID); err != nil {
+				h.log.Warn("record connection: failed to store reconciled nango connection id",
+					zap.String("provider", req.Provider),
+					zap.String("provider_config_key", providerConfigKey),
+					zap.String("alias", req.Alias),
+					zap.String("nango_connection_id", nangoConnID),
+					zap.Error(err),
+				)
+			} else {
+				conn = updated
+				if err := h.nango.PatchConnectionMetadata(r.Context(), nangoConnID, providerConfigKey, map[string]any{
+					"velane_alias":     conn.Alias,
+					"velane_tenant_id": tenant.ID,
+				}); err != nil {
+					h.log.Warn("record connection: failed to patch nango connection metadata (non-fatal)",
+						zap.String("nango_connection_id", nangoConnID),
+						zap.Error(err),
+					)
+				}
+			}
+		} else {
+			h.log.Warn("record connection: nango connection id not found during reconciliation",
+				zap.String("provider", req.Provider),
+				zap.String("provider_config_key", providerConfigKey),
+				zap.String("alias", req.Alias),
+			)
+		}
+	}
 
 	if h.auditor != nil {
 		actorID, actorType := resolveActor(r)
@@ -314,5 +352,9 @@ func (h *ConnectionsHandler) Proxy(w http.ResponseWriter, r *http.Request) {
 	if providerConfigKey == "" {
 		providerConfigKey = provider
 	}
+	if conn.NangoConnectionID == "" {
+		writeError(w, http.StatusBadRequest, "connection is not fully linked; reconnect provider")
+		return
+	}
 	h.nango.Proxy(w, r, conn.NangoConnectionID, providerConfigKey, path)
 }

services/control-plane/internal/nango/client.go

@@ -81,6 +81,102 @@ func (c *Client) CreateConnectSession(ctx context.Context, tenantID, tenantName,
 	return out.Data.Token, nil
 }
 
+// FindConnectionID returns the newest Nango connection matching Velane's tenant,
+// provider config, and alias tags.
+func (c *Client) FindConnectionID(ctx context.Context, tenantID, providerConfigKey, alias string) (string, error) {
+	if alias == "" {
+		alias = "default"
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+"/connections", nil)
+	if err != nil {
+		return "", err
+	}
+	c.setAuth(req)
+
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("nango FindConnectionID: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 300 {
+		raw, _ := io.ReadAll(resp.Body)
+		return "", fmt.Errorf("nango FindConnectionID %d: %s", resp.StatusCode, raw)
+	}
+
+	type nangoConnection struct {
+		ConnectionID      string            `json:"connection_id"`
+		ProviderConfigKey string            `json:"provider_config_key"`
+		Tags              map[string]string `json:"tags"`
+		Metadata          map[string]any    `json:"metadata"`
+		Created           string            `json:"created"`
+	}
+	var envelope struct {
+		Connections []nangoConnection `json:"connections"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
+		return "", fmt.Errorf("nango FindConnectionID decode: %w", err)
+	}
+
+	var newest nangoConnection
+	var newestAt time.Time
+	for _, conn := range envelope.Connections {
+		if conn.ConnectionID == "" || conn.ProviderConfigKey != providerConfigKey {
+			continue
+		}
+		if !nangoConnectionMatchesTenant(conn.Tags, conn.Metadata, tenantID) {
+			continue
+		}
+		if !nangoConnectionMatchesAlias(conn.Tags, conn.Metadata, alias) {
+			continue
+		}
+
+		createdAt, err := time.Parse(time.RFC3339Nano, conn.Created)
+		if err != nil {
+			if newest.ConnectionID == "" {
+				newest = conn
+			}
+			continue
+		}
+		if newest.ConnectionID == "" || createdAt.After(newestAt) {
+			newest = conn
+			newestAt = createdAt
+		}
+	}
+	return newest.ConnectionID, nil
+}
+
+func nangoConnectionMatchesTenant(tags map[string]string, metadata map[string]any, tenantID string) bool {
+	if tenantID == "" {
+		return true
+	}
+	if tags["end_user_id"] == tenantID || tags["organization_id"] == tenantID {
+		return true
+	}
+	if metadata != nil {
+		if value, ok := metadata["velane_tenant_id"].(string); ok && value != "" {
+			return value == tenantID
+		}
+	}
+	return tags["end_user_id"] == "" && tags["organization_id"] == ""
+}
+
+func nangoConnectionMatchesAlias(tags map[string]string, metadata map[string]any, alias string) bool {
+	if alias == "" {
+		alias = "default"
+	}
+	if value := tags["velane_alias"]; value != "" {
+		return value == alias
+	}
+	if metadata != nil {
+		if value, ok := metadata["velane_alias"].(string); ok && value != "" {
+			return value == alias
+		}
+	}
+	return true
+}
+
 // PatchConnectionMetadata updates metadata on an existing Nango connection without
 // overwriting fields not included in the patch (PATCH semantics).
 func (c *Client) PatchConnectionMetadata(ctx context.Context, connectionID, providerConfigKey string, metadata map[string]any) error {