# Premium DM PrivateLink Support via SinkEndpoint Reuse

## 1. Background & Motivation

Premium DM previously only supported **PUBLIC** connections to upstream MySQL data sources. Users with strict network security requirements need PrivateLink connectivity. Rather than building a new PrivateLink management system from scratch, we **reuse the existing `BizSinkEndpoint` infrastructure** (originally built for CDC changefeed downstream connections) to manage PrivateLink endpoints for Premium DM source connections.

### Implementation Status: ✅ Complete

All components are implemented on the `feat/premium-dm-privatelink` branch (PR #3347).

### Design Principles

1. SinkEndpoint (`BizSinkEndpoint`) is a **shared, tenant-scoped** PrivateLink endpoint management resource
2. `PremiumDMSource.EndpointID` and `EndpointDomain` are **deprecated** — always look up from `BizSinkEndpoint` table at runtime instead of caching
3. `BizSinkEndpointUsage` is extended with `OwnerID` (string) and `UsageType` fields for polymorphic usage tracking; the existing `BizChangefeedID` (uint64) is kept for backward compatibility but deprecated
4. Usage records are cleaned up **both** on task deletion **and** on precheck completion

## 2. Architecture

```
┌─────────────────────────────────────────────────────────────────────┐
│                       Console (Frontend)                            │
│  1. Create SinkEndpoint (existing SinkEndpointService API)          │
│  2. Create PremiumMigration with conn_type=PRIVATELINK,             │
│     sink_endpoint_id=<biz_sink_endpoint.id>                         │
│  3. Run Precheck with conn_type=PRIVATELINK,                        │
│     sink_endpoint_id=<biz_sink_endpoint.id>                         │
└────────────────────┬────────────────────────────────────────────────┘
                     │
        ┌────────────▼─────────────────────────┐
        │   PremiumDMService Controller         │
        │  - Validate SinkEndpoint available    │
        │  - resolvePrivateLinkTargets():       │
        │    Load BizSinkEndpoint → set         │
        │    ConnProfile.Target = EndpointAddr  │
        │  - Create BizSinkEndpointUsage record │
        │    (OwnerID + UsageType for DM/CDC)   │
        └────────────┬─────────────────────────┘
                     │
        ┌────────────▼─────────────────────────┐
        │    PremiumDMTaskWorker                 │
        │  OnTaskSourceCreating:                 │
        │    Use s.ConnProfile.Target directly   │
        │    (already resolved by controller)    │
        │    → pass as Host to DM agent          │
        │  OnTaskDeleting:                       │
        │    Delete BizSinkEndpointUsage         │
        └────────────┬─────────────────────────┘
                     │
        ┌────────────▼─────────────────────────┐
        │    PrecheckWorker                      │
        │  On precheck start:                    │
        │    Resolve endpoint dynamically        │
        │    → use as From.Host in SubTaskConfig │
        │  On precheck finish:                   │
        │    Delete BizSinkEndpointUsage         │
        └────────────┬─────────────────────────┘
                     │
        ┌────────────▼─────────────────────────┐
        │    BizSinkEndpoint (shared)            │
        │  Status: available                     │
        │  EndpointAddress: vpce-xxx.region.xxx  │
        │  Managed by SinkEndpoint sync workers  │
        └──────────────────────────────────────┘
```

## 3. Detailed Changes (Actual Implementation)

### 3.1 Proto: No Changes Required

The proto already has all necessary fields:
- `PremiumDMConnType.PRIVATELINK = 2`
- `PremiumDMConnProfile.sink_endpoint_id` (int64, oneof with host)
- `PremiumSupportedSource.conn_methods` (array)

### 3.2 Entity Changes

#### 3.2.1 `PremiumDMSource` — Deprecate EndpointID/EndpointDomain

`PremiumDMSource.EndpointID` and `PremiumDMSource.EndpointDomain` are **deprecated**. Not used by new code. The endpoint address is resolved at task-creation time by the controller and stored in `PremiumConnProfile.Target`.

**No schema changes** — columns stay in DB for backward compatibility.

#### 3.2.2 `PremiumConnProfile` — Separate `SinkEndpointID` field

```go
// entity/premium_dm.go
type PremiumConnProfile struct {
    Type           PremiumConnType  `json:"type"`
    Target         string           `json:"target"`               // For PUBLIC: host; For PRIVATELINK: resolved EndpointAddress
    SinkEndpointID uint64           `json:"sink_endpoint_id,string"` // For PRIVATELINK: BizSinkEndpoint ID (separate field)
    Port           int32            `json:"port"`
    User           string           `json:"user"`
    Password       string           `json:"password"`
    Security       *PremiumSecurity `json:"security"`
}
```

**Key decision**: `SinkEndpointID` is a **separate `uint64` field**, NOT stored as a string in `Target`. `Target` stores the **resolved endpoint address** (for PRIVATELINK, this is `BizSinkEndpoint.EndpointAddress` set by the controller at creation time).

#### 3.2.3 `BizSinkEndpointUsage` — Add `OwnerID` + `UsageType`, deprecate `BizChangefeedID`

```go
// entity/entity.go
type BizSinkEndpointUsageType string

const (
    BizSinkEndpointUsageTypeChangefeed       BizSinkEndpointUsageType = "changefeed"
    BizSinkEndpointUsageTypePremiumDMTask     BizSinkEndpointUsageType = "premium_dm_task"
    BizSinkEndpointUsageTypePremiumDMPrecheck BizSinkEndpointUsageType = "premium_dm_precheck"
)

type BizSinkEndpointUsage struct {
    Base
    SoftDelete
    BizSinkEndpointID ID                       `gorm:"...;uniqueIndex:uniq_endpoint_owner_type,priority:1;"`
    BizChangefeedID   ID                       `gorm:"..."` // DEPRECATED: kept for backward compatibility, NOT in new unique index
    OwnerID           string                   `gorm:"type:varchar(255);not null;uniqueIndex:uniq_endpoint_owner_type,priority:2;"` // NEW: polymorphic owner reference
    UsageType         BizSinkEndpointUsageType `gorm:"type:varchar(64);not null;default:'changefeed';uniqueIndex:uniq_endpoint_owner_type,priority:3;"`
}
```

**Key decisions:**
- `BizChangefeedID` (uint64) is **kept as-is** (deprecated) — no type change. Existing CDC code still writes to it for backward compatibility
- **NEW `OwnerID` (string) field** added for polymorphic usage tracking:
  - For `changefeed`: stores the changefeed ID as string (backfilled from `BizChangefeedID` by migration)
  - For `premium_dm_task`: stores the DM task string ID (e.g., `"dmtsk-xxx"`)
  - For `premium_dm_precheck`: stores the precheck string ID (e.g., `"dmpck-xxx"`)
- Unique index renamed from `uniq_endpoint_type_id` → `uniq_endpoint_owner_type` on `(biz_sink_endpoint_id, owner_id, usage_type)`

### 3.3 Controller Layer Changes

#### 3.3.1 `premium_dm_vo.go` — VO Conversions

**`V1beta1ConnProfileToEntity`**: PRIVATELINK case stores `SinkEndpointID` as separate uint64; `Target` is NOT set (will be resolved by controller):

```go
case v1beta2.PremiumDMConnType_PRIVATELINK:
    return entity.PremiumConnProfile{
        Type:           entity.PremiumConnTypePrivateLink,
        SinkEndpointID: uint64(pb.GetSinkEndpointId()),
        Port:           pb.GetPort(),
        User:           pb.GetUser(),
        Password:       pb.GetPassword(),
        Security:       V1beta1SecurityToEntity(pb.GetSecurity()),
    }
```

**`EntityConnProfileToV1beta1`**: PRIVATELINK case reads `SinkEndpointID` directly (no string parsing):

```go
case entity.PremiumConnTypePrivateLink:
    return &v1beta2.PremiumDMConnProfile{
        ConnType: v1beta2.PremiumDMConnType_PRIVATELINK,
        Endpoint: &v1beta2.PremiumDMConnProfile_SinkEndpointId{SinkEndpointId: int64(cp.SinkEndpointID)},
        Port:     cp.Port,
        User:     cp.User,
        Password: "",
        Security: EntitySecurityToV1beta1(cp.Security),
    }
```

#### 3.3.2 `premium_dm.go` — Controller

**DI Dependencies added**: `SinkEndpointModel` (existing interface) injected into `PremiumDMService`.

**`CreatePremiumMigration`**: After converting req to entity sources:
1. `resolvePrivateLinkTargets(ctx, sources)` — for each PRIVATELINK source, calls `resolveSinkEndpointAddress(ctx, sinkEndpointID)` which loads BizSinkEndpoint, validates status=`available`, and sets `ConnProfile.Target = EndpointAddress`
2. `createSinkEndpointUsagesForReqSources(ctx, sources, taskID, premium_dm_task)` — creates `BizSinkEndpointUsage` records with `OwnerID=taskID` and `UsageType=premium_dm_task`

**`CreatePremiumMigrationPrecheck`**: Similarly:
1. `resolvePrivateLinkTargets(ctx, sources)` — resolve endpoint addresses
2. `createSinkEndpointUsagesForReqSources(ctx, sources, precheckID, premium_dm_precheck)` — creates usage records

**`GetSupportedSources`**: Returns `[PUBLIC, PRIVATELINK]` as `conn_methods` for MySQL source type.

**`execSQLOnSource`**: When `conn_type=PRIVATE_LINK`:
1. `resolveSinkEndpointAddress(ctx, cp.SinkEndpointID)` — resolves address dynamically (at exec time, NOT from stored Target)
2. Passes resolved address via `agentpb.ConnProfile_EndpointId{EndpointId: endpointAddr}`
3. Sets `ConnType: agentpb.ConnType_PRIVATE_LINK`

**Helper functions on `PremiumDMService`:**
- `resolveSinkEndpointAddress(ctx, sinkEndpointID int64) (string, error)` — loads BizSinkEndpoint, validates available status
- `resolvePrivateLinkTargets(ctx, sources []entity.PremiumDMSource)` — iterates sources, resolves PRIVATELINK targets
- `createSinkEndpointUsagesForReqSources(ctx, sources, ownerID, usageType)` — creates usage records for PRIVATELINK sources

### 3.4 Worker Layer Changes

#### 3.4.1 `worker.go` — Task Worker

**DI Dependencies added**: `SinkEndpointModel` injected into `PremiumDMTaskWorker`.

**`OnTaskSourceCreating`**: Uses `s.ConnProfile.Target` **directly** — the endpoint address was already resolved by the controller at task creation time and persisted in `ConnProfile.Target`. No additional BizSinkEndpoint lookup needed at worker time. For PRIVATELINK, uses `ConnType_PRIVATE_LINK` and passes `Target` as `EndpointId`.

**`OnTaskDeleting`**: For each PRIVATELINK source, calls `deleteSinkEndpointUsageForSource(ctx, source, taskID)` which deletes usage via `SinkEndpointModel.DeleteSinkEndpointUsage` with `OwnerID=taskID` and `UsageType=premium_dm_task`.

**Helper functions on `PremiumDMTaskWorker`:**
- `resolveSinkEndpointAddress(ctx, sinkEndpointID uint64) (string, error)` — loads BizSinkEndpoint for dynamic resolution
- `deleteSinkEndpointUsageForSource(ctx, source, taskID)` — deletes usage record

#### 3.4.2 `precheck_execution.go` — Precheck Worker

**On precheck finish** (both success and failure): For each PRIVATELINK source, deletes `BizSinkEndpointUsage` where `OwnerID=precheckID` and `UsageType=premium_dm_precheck`.

### 3.5 Model Layer: Precheck Changes

#### 3.5.1 `precheck_builder.go`

**`BuildPremiumPrecheckBaseConfig`**: Uses `premiumSource.ConnProfile.Target` as `From.Host` — this was already resolved by the controller before persisting.

**`connProfileToEntity`** (local in precheck_builder): PRIVATELINK case sets `SinkEndpointID = uint64(pb.GetSinkEndpointId())` and leaves `Target` empty (to be filled by caller):

```go
case v1beta2.PremiumDMConnType_PRIVATELINK:
    return entity.PremiumConnProfile{
        Type:           entity.PremiumConnTypePrivateLink,
        Target:         "", // fill by caller
        SinkEndpointID: uint64(pb.GetSinkEndpointId()),
        Port:           pb.GetPort(),
        User:           pb.GetUser(),
        Password:       pb.GetPassword(),
        Security:       securityToEntity(pb.GetSecurity()),
    }
```

### 3.6 Service Dependency Injection

**`PremiumDMService`** (controller) receives:
- `SinkEndpointModel` — to create/delete usage records and resolve endpoints

**`PremiumDMTaskWorker`** receives:
- `SinkEndpointModel` — to resolve endpoint address and delete usage records on task deletion

**Container wiring** (`container.go:initPremiumDM`):
```go
taskWorker := premium_dm_model.NewPremiumDMTaskWorker(dao, mgmtClient, encryptor, agentCfg, container.SinkEndpointModel)
container.PremiumDMAPI = premium_dm_controller.NewPremiumDMService(
    dao, taskWorker, mgmtClient, encryptor, agentCfg, transProvider, container.SinkEndpointModel,
)
```

### 3.7 Domain Request/Response Updates

`domain/sink_endpoint.go` uses `OwnerID` (NOT `BizChangefeedID`):
```go
type CreateSinkEndpointUsageReq struct {
    SinkEndpointID uint64
    OwnerID        string                          // polymorphic: changefeed ID (as string), DM task ID, or precheck ID
    UsageType      entity.BizSinkEndpointUsageType
}

type DeleteSinkEndpointUsageReq struct {
    SinkEndpointID uint64
    OwnerID        string                          // polymorphic: changefeed ID (as string), DM task ID, or precheck ID
    UsageType      entity.BizSinkEndpointUsageType
}
```

## 4. Migration

### 4.1 DB Migration: Add `OwnerID`, Backfill, Add `UsageType`, Reindex

File: `dataflow-service-ng/migrate/record_202507171000_sink_endpoint_usage_type.go`

```go
var _ = AddDataflowDBMigrateRecord(
    "202507171000",
    func(db *gorm.DB) error {
        // Step 1: Add owner_id column (non-nullable with empty default for backfill)
        if err := db.Exec(
            "ALTER TABLE biz_sink_endpoint_usages ADD COLUMN owner_id VARCHAR(255) NOT NULL DEFAULT ''",
        ).Error; err != nil {
            return err
        }
        // Step 2: Migrate existing biz_changefeed_id (bigint) values to owner_id (string)
        if err := db.Exec(
            "UPDATE biz_sink_endpoint_usages SET owner_id = CAST(biz_changefeed_id AS CHAR) WHERE owner_id = ''",
        ).Error; err != nil {
            return err
        }
        // Step 3: Add usage_type column with default 'changefeed' (backfills existing rows)
        if err := db.Exec(
            "ALTER TABLE biz_sink_endpoint_usages ADD COLUMN usage_type VARCHAR(64) NOT NULL DEFAULT 'changefeed'",
        ).Error; err != nil {
            return err
        }
        // Step 4: Drop old unique index on (biz_sink_endpoint_id, biz_changefeed_id)
        if err := db.Exec(
            "ALTER TABLE biz_sink_endpoint_usages DROP INDEX uniq_endpoint_type_id",
        ).Error; err != nil {
            return err
        }
        // Step 5: Create new unique index on (biz_sink_endpoint_id, owner_id, usage_type)
        return db.Exec(
            "CREATE UNIQUE INDEX uniq_endpoint_owner_type ON biz_sink_endpoint_usages (biz_sink_endpoint_id, owner_id, usage_type)",
        ).Error
    },
)
```

**Migration strategy**: ADD columns + backfill (non-destructive). `BizChangefeedID` (bigint) column is NOT modified — kept for backward compatibility. Existing CDC rows get `owner_id` backfilled from `biz_changefeed_id` and `usage_type` defaulted to `'changefeed'`.

## 5. Execution Status: ✅ All Steps Complete

All steps are implemented on the `feat/premium-dm-privatelink` branch.

## 6. Risk Assessment

| Risk | Level | Mitigation |
|------|-------|------------|
| Unique index change on `biz_sink_endpoint_usages` | 🟡 Medium | Drop old + create new in single migration; existing rows get defaults |
| New `owner_id` column backfill | 🟢 Low | Non-destructive ADD COLUMN + backfill from existing `biz_changefeed_id`; no data loss |
| Endpoint not available at task creation | 🟢 Low | Validate endpoint status = `available` before accepting request |
| Precheck usage record leak | 🟢 Low | Cleanup runs on both success and failure paths; SinkEndpoint cleanup worker is safety net |
| Polymorphic OwnerID field | 🟢 Low | `UsageType` column prevents ID collisions between changefeed and DM task/precheck IDs |

## 7. Out of Scope

- **PrivateLink billing for Premium DM**: No metering/billing for Premium DM PrivateLink (consistent with current design)
- **VPC Peering**: Not supported for Premium DM
- **Auto-create SinkEndpoint**: Users must create SinkEndpoint first via existing SinkEndpointService API
- **PremiumDMVpcEndpoint table**: The existing `PremiumDMVpcEndpoint` entity is a legacy placeholder; NOT used
- **Deprecate EndpointID/EndpointDomain columns**: DB columns stay for backward compatibility; just not used by new code
- **Remove BizChangefeedID column**: Kept as deprecated uint64 for backward compatibility; `OwnerID` is the canonical field