feat(cursor): add stateful extract and docs

jbsmith7741 · jbsmith7741 · commit 1759930eefdf · 2026-07-28T11:30:46.000-06:00
* Switch usage events and daily usage extractors to StatefulApiExtractor with versioned one-time full re-extract.
  * Chunk filtered-usage-events into 30-day ranges and add plugin README.

Signed-off-by: Joshua Smith &lt;jbsmith7741@gmail.com&gt;
diff --git a/backend/plugins/cursor/README.md b/backend/plugins/cursor/README.md
@@ -0,0 +1,169 @@
+<!--
+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.
+-->
+# Cursor Plugin (Usage & Cost)
+
+This plugin ingests **Cursor team usage, billing, and adoption metrics** from the [Cursor Admin API](https://cursor.com/docs/account/teams/admin-api) and stores them in DevLake tool-layer tables for Grafana dashboards and SQL analysis.
+
+It follows the same structure and patterns as other DevLake AI usage plugins (notably `backend/plugins/gh-copilot`).
+
+## What it collects
+
+**Cursor Admin API endpoints:**
+
+| Endpoint | Method | Data |
+|----------|--------|------|
+| `/teams/members` | GET | Team roster |
+| `/teams/spend` | POST | Per-user billing cycle spend |
+| `/teams/filtered-usage-events` | POST | Event-level usage and charges |
+| `/teams/daily-usage-data` | POST | Per-user per-day adoption metrics |
+
+**Stored data (tool layer):**
+
+| Table | Description |
+|-------|-------------|
+| `_tool_cursor_members` | Team member roster (email, name, role) |
+| `_tool_cursor_usage_events` | Billable usage events with model, tokens, and charged amounts |
+| `_tool_cursor_user_spend` | Per-user spend for the current billing cycle (on-demand and included) |
+| `_tool_cursor_daily_usage` | Daily adoption metrics: completions, requests by feature, tab acceptance, line edits |
+
+Data is collected in the **Raw → Tool** layers only. There is no domain-layer converter in this plugin; Grafana dashboards query `_tool_cursor_*` tables directly.
+
+## Data flow
+
+```mermaid
+flowchart LR
+  API[Cursor Admin API]
+  RAW[(Raw tables\n_raw_cursor_*)]
+  TOOL[(Tool tables\n_tool_cursor_*)]
+  GRAF[Grafana Dashboards]
+
+  API --> RAW --> TOOL --> GRAF
+```
+
+**Pipeline subtasks (in order):**
+
+1. `collectMembers` → `extractMembers`
+2. `collectUsageEvents` → `extractUsageEvents`
+3. `collectUserSpend` → `extractUserSpend`
+4. `collectDailyUsage` → `extractDailyUsage`
+
+## Repository layout
+
+- `api/` — REST layer for connections, scopes, and scope configs
+- `impl/` — plugin meta, blueprint v200, connection helpers
+- `models/` — tool-layer models and migration scripts
+- `tasks/` — collectors, extractors, and pipeline registration
+- `service/` — connection test logic (Admin API permission probes)
+- `e2e/` — E2E fixtures and golden CSV assertions
+
+## Setup
+
+### Prerequisites
+
+- A Cursor **Team or Business** plan with Admin API access
+- A **Team Admin API key** from the Cursor dashboard (Dashboard → API Keys)
+- Do **not** use a User API key from Settings → Integrations — user keys cannot access `/teams/*` endpoints
+
+Authentication uses HTTP Basic auth: the API key as username with an empty password.
+
+### 1) Create a connection
+
+1. DevLake UI → **Connections → Add Connection → Cursor**
+2. Fill in:
+   - **Name**: e.g. `Cursor Production Team`
+   - **Endpoint**: defaults to `https://api.cursor.com`
+   - **Token**: Team Admin API key
+   - **Rate Limit**: defaults to 1,200 requests/hour (Cursor documents 20 requests/minute)
+3. Click **Test Connection**. DevLake probes `/teams/members`, `/teams/spend`, and `/teams/filtered-usage-events` and reports which endpoints the key can access.
+4. Save the connection.
+
+When updating an existing connection, omit the token field to keep the encrypted value already stored in DevLake.
+
+### 2) Add a scope
+
+Cursor data is team-level. Add a **Team** scope for the connection. The default scope ID is `team`.
+
+### 3) Create a blueprint
+
+Use a blueprint plan like:
+
+```json
+[
+  [
+    {
+      "plugin": "cursor",
+      "options": {
+        "connectionId": 1,
+        "scopeId": "team"
+      }
+    }
+  ]
+]
+```
+
+Run the blueprint on a daily schedule to keep usage and cost data current.
+
+### Collection behavior
+
+- **Initial backfill**: usage events and daily usage collect up to **90 days** of history on the first run.
+- **Incremental runs**: subsequent runs use the pipeline sync policy (`TimeAfter`) to collect only new data.
+- **Date range chunking**: both `/teams/daily-usage-data` and `/teams/filtered-usage-events` requests are split into **30-day** chunks (API limit for daily usage; applied to usage events for resilience).
+- **Extract**: `extractUsageEvents` and `extractDailyUsage` use **StatefulApiExtractor** (incremental by default). A config version bump triggers a one-time full re-extract after upgrade.
+- **Rate limiting**: collectors honor `Retry-After` response headers and respect the configured `rateLimitPerHour`.
+
+## Dashboards
+
+Grafana dashboard JSON lives under `grafana/dashboards/mysql/`:
+
+| Dashboard | File | UID |
+|-----------|------|-----|
+| Cursor Usage & Cost | `cursor-usage.json` | `cursor_usage` |
+| AI Cost Efficiency (Cursor panels) | `ai-cost-efficiency.json` | — |
+| Multi-AI Comparison (Cursor panels) | `multi-ai-comparison.json` | — |
+
+See `grafana/dashboards/mysql/CursorREADME.md` for dashboard prerequisites, variables, and panel descriptions.
+
+## Error handling
+
+| Symptom | Likely cause |
+|---------|--------------|
+| **401 Unauthorized** on test connection | Invalid API key, or a User API key instead of a Team Admin key |
+| **403 Forbidden** on spend or usage events | Key lacks permission for that Admin API endpoint |
+| **429 Too Many Requests** | Rate limit exceeded — lower `rateLimitPerHour` or wait for `Retry-After` |
+| Empty usage events after successful run | Selected time range has no billable events, or sync policy excludes the date range |
+| Reconciliation delta on dashboard | Expected when comparing event-level charges to billing-cycle spend snapshots; see dashboard notes |
+
+Tokens are sanitized before persisting. Connection test results include a `permissions` object showing which Admin API endpoints succeeded.
+
+## Limitations
+
+- **Team/Business Admin API only** — Enterprise-only Analytics API endpoints (`/analytics/*`) are not collected in this plugin.
+- **Tool layer only** — no domain-layer tables; cross-plugin joins (Jira, GitHub PRs, etc.) are done in Grafana SQL or separate tooling.
+- **Team-level scope** — one scope per connection represents the whole team; per-team multi-tenant collection is not supported.
+- **Beta** — the plugin is marked beta in Config UI while the Admin API surface continues to evolve.
+
+## Testing
+
+```sh
+# Unit tests
+cd backend && go test ./plugins/cursor/...
+
+# E2E (requires E2E_DB_URL)
+make e2e-test
+```
+
+E2E fixtures live in `backend/plugins/cursor/e2e/raw_tables/` and `e2e/snapshot_tables/`.
diff --git a/backend/plugins/cursor/tasks/collector_utils.go b/backend/plugins/cursor/tasks/collector_utils.go
@@ -95,8 +95,8 @@ func computeUsageTimeRangeMs(since *time.Time, now time.Time) (int64, int64) {
 	return start.UnixMilli(), end.UnixMilli()
 }
 
-// splitDailyUsageTimeRangeMs splits [startMs, endMs] into chunks of at most maxDays for
-// POST /teams/daily-usage-data (API limit: date range cannot exceed 30 days).
+// splitDailyUsageTimeRangeMs splits [startMs, endMs] into chunks of at most maxDays.
+// Used by daily-usage-data (30-day API limit) and filtered-usage-events collectors.
 func splitDailyUsageTimeRangeMs(startMs, endMs int64, maxDays int) []cursorTimeRangeInput {
 	if startMs >= endMs || maxDays <= 0 {
 		return nil
diff --git a/backend/plugins/cursor/tasks/collector_utils_test.go b/backend/plugins/cursor/tasks/collector_utils_test.go
@@ -83,6 +83,16 @@ func TestSplitDailyUsageTimeRangeMsEmptyRange(t *testing.T) {
 	}
 }
 
+func TestNewUsageEventsDateRangeIteratorChunksLongRanges(t *testing.T) {
+	start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+	end := time.Date(2026, 4, 1, 12, 0, 0, 0, time.UTC)
+	chunks := splitDailyUsageTimeRangeMs(start.UnixMilli(), end.UnixMilli(), cursorDailyUsageMaxDays)
+
+	if len(chunks) != 4 {
+		t.Fatalf("expected 4 chunks for ~90-day usage-events span, got %d", len(chunks))
+	}
+}
+
 func TestDailyUsagePostBodyUsesEpochMilliseconds(t *testing.T) {
 	body := dailyUsagePostBody(&helper.RequestData{
 		Input: cursorTimeRangeInput{StartDateMs: 1710720000000, EndDateMs: 1710892800000},
diff --git a/backend/plugins/cursor/tasks/daily_usage_extractor.go b/backend/plugins/cursor/tasks/daily_usage_extractor.go
@@ -18,7 +18,6 @@ limitations under the License.
 package tasks
 
 import (
-	"encoding/json"
 	"strings"
 	"time"
 
@@ -60,18 +59,9 @@ func ExtractDailyUsage(taskCtx plugin.SubTaskContext) errors.Error {
 		return errors.Default.New("task data is not CursorTaskData")
 	}
 
-	extractor, err := helper.NewApiExtractor(helper.ApiExtractorArgs{
-		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx:     taskCtx,
-			Table:   rawDailyUsageTable,
-			Options: rawParamsFromTaskData(data),
-		},
-		Extract: func(row *helper.RawData) ([]interface{}, errors.Error) {
-			var record dailyUsageRecord
-			if err := errors.Convert(json.Unmarshal(row.Data, &record)); err != nil {
-				return nil, err
-			}
-
+	extractor, err := helper.NewStatefulApiExtractor(&helper.StatefulApiExtractorArgs[dailyUsageRecord]{
+		SubtaskCommonArgs: cursorSubtaskCommonArgs(taskCtx, data, rawDailyUsageTable),
+		Extract: func(record *dailyUsageRecord, _ *helper.RawData) ([]any, errors.Error) {
 			userId := strings.TrimSpace(record.UserId)
 			if userId == "" {
 				userId = strings.TrimSpace(record.Email)
@@ -113,7 +103,7 @@ func ExtractDailyUsage(taskCtx plugin.SubTaskContext) errors.Error {
 				LinesAdded:               record.AcceptedLinesAdded,
 				LinesDeleted:             record.AcceptedLinesDeleted,
 			}
-			return []interface{}{usage}, nil
+			return []any{usage}, nil
 		},
 	})
 	if err != nil {
diff --git a/backend/plugins/cursor/tasks/extract_utils.go b/backend/plugins/cursor/tasks/extract_utils.go
@@ -0,0 +1,38 @@
+/*
+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 tasks
+
+import (
+	"github.com/apache/incubator-devlake/core/plugin"
+	helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
+)
+
+// cursorExtractorVersion bumps when extract logic changes in a way that requires
+// a one-time full re-extract (SubtaskStateManager compares SubtaskConfig).
+const cursorExtractorVersion = 2
+
+func cursorSubtaskCommonArgs(taskCtx plugin.SubTaskContext, data *CursorTaskData, rawTable string) *helper.SubtaskCommonArgs {
+	return &helper.SubtaskCommonArgs{
+		SubTaskContext: taskCtx,
+		Table:          rawTable,
+		Params:         rawParamsFromTaskData(data),
+		SubtaskConfig: map[string]any{
+			"extractorVersion": cursorExtractorVersion,
+		},
+	}
+}
diff --git a/backend/plugins/cursor/tasks/usage_events_collector.go b/backend/plugins/cursor/tasks/usage_events_collector.go
@@ -26,10 +26,12 @@ import (
 	helper "github.com/apache/incubator-devlake/helpers/pluginhelper/api"
 )
 
-func newTimeRangeIterator(since *time.Time) *helper.QueueIterator {
-	iter := helper.NewQueueIterator()
+func newUsageEventsDateRangeIterator(since *time.Time) *helper.QueueIterator {
 	startMs, endMs := computeUsageTimeRangeMs(since, time.Now().UTC())
-	iter.Push(cursorTimeRangeInput{StartDateMs: startMs, EndDateMs: endMs})
+	iter := helper.NewQueueIterator()
+	for _, chunk := range splitDailyUsageTimeRangeMs(startMs, endMs, cursorDailyUsageMaxDays) {
+		iter.Push(chunk)
+	}
 	return iter
 }
 
@@ -90,7 +92,7 @@ func CollectUsageEvents(taskCtx plugin.SubTaskContext) errors.Error {
 
 	err = collector.InitCollector(helper.ApiCollectorArgs{
 		ApiClient:             apiClient,
-		Input:                 newTimeRangeIterator(collector.GetSince()),
+		Input:                 newUsageEventsDateRangeIterator(collector.GetSince()),
 		Method:                http.MethodPost,
 		UrlTemplate:           "teams/filtered-usage-events",
 		PageSize:              cursorApiPageSize,
diff --git a/backend/plugins/cursor/tasks/usage_events_extractor.go b/backend/plugins/cursor/tasks/usage_events_extractor.go
@@ -18,7 +18,6 @@ limitations under the License.
 package tasks
 
 import (
-	"encoding/json"
 	"strings"
 
 	"github.com/apache/incubator-devlake/core/errors"
@@ -60,18 +59,9 @@ func ExtractUsageEvents(taskCtx plugin.SubTaskContext) errors.Error {
 		return errors.Default.New("task data is not CursorTaskData")
 	}
 
-	extractor, err := helper.NewApiExtractor(helper.ApiExtractorArgs{
-		RawDataSubTaskArgs: helper.RawDataSubTaskArgs{
-			Ctx:   taskCtx,
-			Table: rawUsageEventsTable,
-			Options: rawParamsFromTaskData(data),
-		},
-		Extract: func(row *helper.RawData) ([]interface{}, errors.Error) {
-			var record usageEventRecord
-			if err := errors.Convert(json.Unmarshal(row.Data, &record)); err != nil {
-				return nil, err
-			}
-
+	extractor, err := helper.NewStatefulApiExtractor(&helper.StatefulApiExtractorArgs[usageEventRecord]{
+		SubtaskCommonArgs: cursorSubtaskCommonArgs(taskCtx, data, rawUsageEventsTable),
+		Extract: func(record *usageEventRecord, _ *helper.RawData) ([]any, errors.Error) {
 			eventTime, err := parseEventTimestampMs(record.Timestamp)
 			if err != nil {
 				return nil, err
@@ -108,7 +98,7 @@ func ExtractUsageEvents(taskCtx plugin.SubTaskContext) errors.Error {
 				event.CacheWriteTokens = record.TokenUsage.CacheWriteTokens
 				event.TotalCents = record.TokenUsage.TotalCents
 			}
-			return []interface{}{event}, nil
+			return []any{event}, nil
 		},
 	})
 	if err != nil {
