# Implementation Plan: Integrate OpenAPI Changefeed Cases into E2E Framework

## Summary

Add an `apiMode` top-level parameter (`console` | `openapi`) to the e2e test framework. When `apiMode=openapi`, existing changefeed cases use the OpenAPI client for shared operations (get, list, pause, resume) instead of the console API client. The standalone `open_api.go` is deleted and its unique tests (event filters) are preserved as a separate case.

---

## Analysis

### Current State

| File | Role |
|------|------|
| `internal/config/config.go` | Config struct with `Domain` field, `InitConfig()`, `Validate()` |
| `run_test.go` | CLI flag parsing (`-domain`), `runTestSuite()` calls `InitConfig(env, domain)` |
| `cases/dedicated/changefeed/open_api.go` | Standalone OpenAPI test case (get, list, pause, resume, event filters) |
| `cases/dedicated/changefeed/kafka_smoke.go` | Console API smoke test (create, scale, pause, resume, edit, checkpoint) |
| `cases/dedicated/changefeed/scale_changefeed.go` | Console-only scale test |
| `cases/dedicated/common/changfeed_validator.go` | Validator wrapping `CDCClientWrapper` (console API only) |
| `internal/tidbcloud/dedicated/consoleapi/cdc_client_wrapper.go` | Console API client — all methods require `*TiDBDedicatedEntity` |
| `internal/tidbcloud/dedicated/openapi/open_api_client_wrapper.go` | OpenAPI client — most methods only need `changefeedID` |

### Key Differences Between APIs

| Operation | Console API | OpenAPI | Shared? |
|-----------|------------|---------|---------|
| Create | `CreateChangefeed(cluster, spec)` | N/A | No (console-only) |
| Delete | `DeleteChangefeed(cluster, id)` | N/A | No (console-only) |
| Pause | `PauseChangefeed(cluster, id)` | `PauseChangefeed(id)` | Yes |
| Resume | `ResumeChangefeed(cluster, id)` | `ResumeChangefeed(id)` | Yes |
| Get | `GetChangefeed(cluster, id)` → `*GetConnectorOK` | `GetChangefeed(id)` → `*GetChangefeedOK` | Yes (different types) |
| List | `ListChangefeed(cluster)` → `*ListConnectorsOK` | `ListChangefeed(clusterID)` → `*ListChangefeedsOK` | Yes (different types) |
| Scale | `ScaleChangefeed(cluster, id, type)` | N/A | No (console-only) |
| Edit | `EditChangefeed(cluster, id, spec)` | N/A | No (console-only) |
| TestConnection | `TestConnection(cluster, downstream)` | N/A | No (console-only) |
| Event Filters | N/A | `SetEventFilters(id, body)` / `GetEventFilters(id)` | No (openapi-only) |

### Operation Classification

- **Both APIs**: pause, resume, get, list
- **Console-only**: create, delete, scale, edit, test connection, describe schema, kernel config
- **OpenAPI-only**: event filters (set/get)

---

## Design Decision: Interface + Adapter Pattern

**Chosen approach**: Define a `ChangefeedClient` Go interface for shared operations with concrete adapter types wrapping each API client.

**Rationale**:
1. The two API clients have fundamentally different method signatures and return types — an interface + adapter cleanly normalizes these differences
2. Existing case files use method chaining (`k.Expect(...).RetryDo(...)`) — the interface preserves this pattern
3. Avoids if/else branching in every case file for every API call
4. Type-safe at compile time; adapter handles runtime type conversion
5. Follows Go idioms (interface segregation, adapter pattern)

**Rejected alternatives**:
- **Full abstraction (one giant interface)**: Too much surface area; console-only and openapi-only methods don't belong in a shared interface
- **if/else in cases**: Prone to copy-paste errors; every case file needs duplicated branching logic
- **Parallel case files**: Contradicts the "reuse existing cases" requirement

---

## Execution Steps

### Step 1: Add `apiMode` Config Field

**File**: `internal/config/config.go`

1. Add constants:
```go
const (
    ApiModeConsole  = "console"
    ApiModeOpenApi  = "openapi"
)

var AllApiModes = []string{ApiModeConsole, ApiModeOpenApi}
```

2. Add field to `config` struct:
```go
type config struct {
    // ... existing fields ...
    ApiMode string `yaml:"apiMode"`
}
```

3. Add `Validate()` check (after domain validation):
```go
if t.ApiMode == "" {
    t.ApiMode = ApiModeConsole
}
if !pie.Contains(AllApiModes, t.ApiMode) {
    panic(fmt.Sprintf("invalid apiMode '%s', must be one of: %v", t.ApiMode, AllApiModes))
}
```

4. Add CLI override in `InitConfig()` (follows `domain` pattern):
```go
func InitConfig(env, domain, apiMode string) {
    // ... existing code ...
    if apiMode != "" {
        GlobalConfig.ApiMode = apiMode
        fmt.Printf("ApiMode overridden from flag: %s\n", apiMode)
    }
    GlobalConfig.Validate()
}
```

5. Add accessor:
```go
func IsOpenApiMode() bool {
    return GlobalConfig.ApiMode == ApiModeOpenApi
}
```

### Step 2: Add CLI Flag

**File**: `run_test.go`

1. Add variable:
```go
var apiMode string
```

2. Add flag in `parseArgsWithDefaults()`:
```go
flag.StringVar(&apiMode, "apiMode", ApiModeConsole, fmt.Sprintf("the API mode for changefeed cases: %v", AllApiModes))
```

3. Update `runTestSuite()` call:
```go
InitConfig(env, domain, apiMode)
```

### Step 3: Define `ChangefeedClient` Interface

**New file**: `cases/dedicated/common/changefeed_client.go`

```go
package common

import "time"

// ChangefeedInfo is the normalized changefeed state across console and OpenAPI.
type ChangefeedInfo struct {
    ID           string
    Name         string
    State        string // "Running", "Paused", etc.
    CheckpointTso string
    CreateTime   time.Time
}

// ChangefeedClient abstracts shared changefeed operations.
// Console API methods require cluster context (passed at adapter creation).
// OpenAPI methods use global auth config.
type ChangefeedClient interface {
    PauseChangefeed(changefeedID string) error
    ResumeChangefeed(changefeedID string) error
    GetChangefeed(changefeedID string) (*ChangefeedInfo, error)
    ListChangefeed() ([]*ChangefeedInfo, error)
}
```

### Step 4: Console API Adapter

**New file**: `cases/dedicated/common/console_changefeed_adapter.go`

```go
package common

import (
    "github.com/tidbcloud/dataflow-service-e2e/internal/model"
    "github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi"
)

type ConsoleChangefeedAdapter struct {
    cdc     *consoleapi.CDCClientWrapper
    cluster *model.TiDBDedicatedEntity
}

func NewConsoleChangefeedAdapter(cdc *consoleapi.CDCClientWrapper, cluster *model.TiDBDedicatedEntity) *ConsoleChangefeedAdapter {
    return &ConsoleChangefeedAdapter{cdc: cdc, cluster: cluster}
}

func (a *ConsoleChangefeedAdapter) PauseChangefeed(changefeedID string) error {
    return a.cdc.PauseChangefeed(a.cluster, changefeedID)
}

func (a *ConsoleChangefeedAdapter) ResumeChangefeed(changefeedID string) error {
    return a.cdc.ResumeChangefeed(a.cluster, changefeedID)
}

func (a *ConsoleChangefeedAdapter) GetChangefeed(changefeedID string) (*ChangefeedInfo, error) {
    resp, err := a.cdc.GetChangefeed(a.cluster, changefeedID)
    if err != nil {
        return nil, err
    }
    state := ""
    if resp.Payload.Status != nil {
        state = string(*resp.Payload.Status)
    }
    return &ChangefeedInfo{
        ID:           *resp.Payload.ConnectorID,
        Name:         resp.Payload.Name,
        State:        state,
        CheckpointTso: resp.Payload.CheckpointTso,
    }, nil
}

func (a *ConsoleChangefeedAdapter) ListChangefeed() ([]*ChangefeedInfo, error) {
    resp, err := a.cdc.ListChangefeed(a.cluster)
    if err != nil {
        return nil, err
    }
    var result []*ChangefeedInfo
    for _, item := range resp.Payload.Items {
        state := ""
        if item.Status != nil {
            state = string(*item.Status)
        }
        result = append(result, &ChangefeedInfo{
            ID:           *item.ConnectorID,
            Name:         item.Name,
            State:        state,
            CheckpointTso: item.CheckpointTso,
        })
    }
    return result, nil
}
```

### Step 5: OpenAPI Adapter

**New file**: `cases/dedicated/common/openapi_changefeed_adapter.go`

```go
package common

import (
    "github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/openapi"
    openapi_model "github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/openapi/changefeed/model"
)

type OpenAPIChangefeedAdapter struct {
    cli       *openapi.ChangefeedOpenApiClientWrapper
    clusterID string
}

func NewOpenAPIChangefeedAdapter(cli *openapi.ChangefeedOpenApiClientWrapper, clusterID string) *OpenAPIChangefeedAdapter {
    return &OpenAPIChangefeedAdapter{cli: cli, clusterID: clusterID}
}

func (a *OpenAPIChangefeedAdapter) PauseChangefeed(changefeedID string) error {
    return a.cli.PauseChangefeed(changefeedID)
}

func (a *OpenAPIChangefeedAdapter) ResumeChangefeed(changefeedID string) error {
    return a.cli.ResumeChangefeed(changefeedID)
}

func (a *OpenAPIChangefeedAdapter) GetChangefeed(changefeedID string) (*ChangefeedInfo, error) {
    resp, err := a.cli.GetChangefeed(changefeedID)
    if err != nil {
        return nil, err
    }
    p := resp.Payload
    state := ""
    if p.State != nil {
        state = string(*p.State)
    }
    return &ChangefeedInfo{
        ID:           p.ID,
        Name:         p.Name,
        State:        state,
        CheckpointTso: p.CheckpointTso,
        CreateTime:   p.CreateTime,
    }, nil
}

func (a *OpenAPIChangefeedAdapter) ListChangefeed() ([]*ChangefeedInfo, error) {
    resp, err := a.cli.ListChangefeed(a.clusterID)
    if err != nil {
        return nil, err
    }
    var result []*ChangefeedInfo
    for _, cf := range resp.Payload.Changefeeds {
        state := ""
        if cf.State != nil {
            state = string(*cf.State)
        }
        result = append(result, &ChangefeedInfo{
            ID:           cf.ID,
            Name:         cf.Name,
            State:        state,
            CheckpointTso: cf.CheckpointTso,
            CreateTime:   cf.CreateTime,
        })
    }
    return result, nil
}
```

### Step 6: Add Helper to `BaseCase`

**File**: `cases/base_case.go`

Add a helper that creates the appropriate `ChangefeedClient` based on `apiMode`:

```go
func (k *BaseCase) NewChangefeedClient(cluster *model.TiDBDedicatedEntity, lgr log.SimpleLogger) common.ChangefeedClient {
    if config.IsOpenApiMode() {
        openApiCli := openapi.NewChangefeedOpenApiClientWrapper(lgr)
        return common.NewOpenAPIChangefeedAdapter(openApiCli, cluster.ClusterID)
    }
    cdcCli := consoleapi.NewCDCClientWrapper(lgr)
    return common.NewConsoleChangefeedAdapter(cdcCli, cluster)
}
```

**Note**: This introduces a dependency from `cases` package to `common` and `openapi` packages. Check for circular imports. If problematic, move the helper to a standalone function in `common/` or accept it as a case-level method.

**Alternative** (if circular import): Each case creates the adapter directly in `Run()`:
```go
var cfClient common.ChangefeedClient
if config.IsOpenApiMode() {
    openApiCli := openapi.NewChangefeedOpenApiClientWrapper(k.Logger())
    cfClient = common.NewOpenAPIChangefeedAdapter(openApiCli, tidbCluster.ClusterID)
} else {
    cdcCli := consoleapi.NewCDCClientWrapper(k.Logger())
    cfClient = common.NewConsoleChangefeedAdapter(cdcCli, tidbCluster)
}
```

### Step 7: Update Case Files

#### 7a. `kafka_smoke.go` — Replace pause/resume with interface

Changes:
- Add `cfClient common.ChangefeedClient` field
- Initialize `cfClient` in `Run()` based on `apiMode`
- Replace `cdcApiWrapper.PauseChangefeed(tidbCluster, id)` → `cfClient.PauseChangefeed(id)`
- Replace `cdcApiWrapper.ResumeChangefeed(tidbCluster, id)` → `cfClient.ResumeChangefeed(id)`
- Wrap console-only steps (Scale, Edit, TestConnection, CheckpointTs) with skip guard:
  ```go
  if !config.IsOpenApiMode() {
      k.RetryDo("scale changefeed", func() error { ... })
  }
  ```
- Keep `cfValidator` for create/delete (always console API)

#### 7b. `scale_changefeed.go` — Skip entire case in openapi mode

Scale is console-only. Add early return at top of `Run()`:
```go
func (k *scaleChangefeedCase) Run() {
    if config.IsOpenApiMode() {
        k.Logger().Info("Skipping scale_changefeed: scale is console-only")
        return
    }
    // ... existing code unchanged ...
}
```

#### 7c. `test_connection.go` — Skip in openapi mode

TestConnection is console-only. Same pattern as scale.

#### 7d. `cdc_recreate.go` — Skip in openapi mode

Recreate is console-only.

#### 7e. `changefeed_config_operations.go` — Skip in openapi mode

Config operations are console-only.

#### 7f. `ticdc_cluster_config_operations.go` — Skip in openapi mode

TiCDC cluster config is console-only.

#### 7g. Other smoke cases (mysql, s3, gcs, azblob, pulsar, serverless, kafka_avro, kafka_private_link, kafka_kernel_configuration, kafka_validate_configuration)

Same pattern as `kafka_smoke.go`:
- Add `cfClient` field
- Replace pause/resume with `cfClient` calls
- Wrap console-only operations with skip guard
- Keep `cfValidator` for create/delete

### Step 8: Delete `open_api.go` and Create Event Filter Case

1. **Delete** `cases/dedicated/changefeed/open_api.go`

2. **Create** `cases/dedicated/changefeed/openapi_event_filters.go` — contains only the event filter tests (OpenAPI-only):
   - `NewOpenAPIEventFiltersCase(cloudProvider)` constructor
   - `Tags()` returns CDC domain
   - `Run()` creates a changefeed via console API, then tests event filters via OpenAPI
   - Skip guard at top: `if !config.IsOpenApiMode() { return }`

3. **Update registry** (`cases_registry_dedicated.go`):
   - Replace `NewChangefeedOpenApiCase` registration with `NewOpenAPIEventFiltersCase`
   - Keep in weeklyRegression groups (only runs when `apiMode=openapi`)

### Step 9: Update `ChangfeedValidator` (Optional, Minimal)

The validator internally uses `CDCClientWrapper` for all operations. For this phase, **do not refactor the validator** — it continues to use console API directly for:
- `CreateDummyChangefeedWithPublicKafka` (always console)
- `DeleteChangefeedByName` (always console)
- `AreChangefeedsRunning` (uses console List)
- `AllChangefeedsNotInListByNames` (uses console List)
- `VerifyCheckpointTsAdvanceNormal` (uses console List)

The validator is used for setup/teardown which always uses console API. Cases use `cfClient` for the actual test operations (pause/resume/get/list).

**Future**: If OpenAPI gains create/delete endpoints, add a validator variant that uses the interface.

### Step 10: Update YAML Configs

**File**: `config.common.yaml` (and env-specific configs if needed)

```yaml
apiMode: "console"  # default; override with -apiMode=openapi flag
```

---

## File Change Summary

| File | Action | Description |
|------|--------|-------------|
| `internal/config/config.go` | Modify | Add `ApiMode` field, constants, validation, `IsOpenApiMode()` |
| `run_test.go` | Modify | Add `-apiMode` CLI flag, pass to `InitConfig()` |
| `cases/dedicated/common/changefeed_client.go` | **Create** | `ChangefeedClient` interface + `ChangefeedInfo` type |
| `cases/dedicated/common/console_changefeed_adapter.go` | **Create** | Console API adapter |
| `cases/dedicated/common/openapi_changefeed_adapter.go` | **Create** | OpenAPI adapter |
| `cases/dedicated/changefeed/open_api.go` | **Delete** | Standalone case removed |
| `cases/dedicated/changefeed/openapi_event_filters.go` | **Create** | Event filter tests (openapi-only case) |
| `cases/dedicated/changefeed/kafka_smoke.go` | Modify | Use `ChangefeedClient` for pause/resume; skip console-only steps |
| `cases/dedicated/changefeed/scale_changefeed.go` | Modify | Skip entire case in openapi mode |
| `cases/dedicated/changefeed/test_connection.go` | Modify | Skip in openapi mode |
| `cases/dedicated/changefeed/cdc_recreate.go` | Modify | Skip in openapi mode |
| `cases/dedicated/changefeed/changefeed_config_operations.go` | Modify | Skip in openapi mode |
| `cases/dedicated/changefeed/ticdc_cluster_config_operations.go` | Modify | Skip in openapi mode |
| Other smoke cases (8 files) | Modify | Use `ChangefeedClient` for pause/resume; skip console-only steps |
| `internal/case_driver/cases_registry_dedicated.go` | Modify | Replace `NewChangefeedOpenApiCase` with `NewOpenAPIEventFiltersCase` |
| `config.common.yaml` | Modify | Add `apiMode: "console"` |

---

## Usage

```bash
# Default (console mode, unchanged behavior)
go test -v ./... -env=staging -domain=cdc

# OpenAPI mode
go test -v ./... -env=staging -domain=cdc -apiMode=openapi

# Specific case group with openapi
go test -v ./... -env=staging -domain=cdc -apiMode=openapi -caseGroup=awsWeeklyRegression
```

---

## Risk Assessment

| Risk | Mitigation |
|------|-----------|
| Circular import (`cases` → `common` → `cases`) | Create adapters in `common/`; helper function in `common/` not `cases/base_case.go` |
| OpenAPI state constants differ from console | Both use string "Running"/"Paused" — adapter uses string conversion, not enum comparison |
| Cases silently skipped in openapi mode | Log skip messages; report skipped count in test output |
| `ChangfeedValidator` still uses console API internally | Expected for this phase; validator handles setup/teardown which is always console |
| Config YAML backward compatibility | Default `apiMode: "console"` preserves existing behavior |

---

## Verification

1. `go build ./...` in `test/e2e/` — no compile errors
2. Run with default flags — identical behavior to current (all console API)
3. Run with `-apiMode=openapi` — shared operations use OpenAPI; console-only steps are skipped
4. Event filter tests only execute in openapi mode
5. `make check` passes