From 2b90a62159b3d186431718b1fd573c43fc5a2b19 Mon Sep 17 00:00:00 2001
From: Kunghim <kunghim@163.com>
Date: Sun, 9 Aug 2026 14:11:19 +0800
Subject: [PATCH] fix(sync): constrain time-series targets to append writes

---
 .../data-sync/DataSyncDeliveryStage.test.tsx  | 38 +++++++++++++++++++
 .../data-sync/DataSyncTaskEditor.tsx          | 20 +++++++++-
 .../data-sync/DataSyncWorkbenchShell.tsx      |  1 +
 frontend/src/components/data-sync/gateway.ts  |  1 +
 frontend/src/components/data-sync/model.ts    |  1 +
 frontend/src/components/data-sync/text.ts     |  4 ++
 .../src/components/data-sync/wailsDto.test.ts |  3 ++
 frontend/src/components/data-sync/wailsDto.ts |  5 +++
 .../src/components/data-sync/wailsGateway.ts  |  1 +
 internal/app/data_sync_job_approval_test.go   | 19 ++++++++++
 internal/app/data_sync_job_preflight.go       | 27 +++++++++++++
 internal/sync/migration_capability.go         |  7 ++++
 internal/sync/migration_capability_test.go    | 22 +++++++++++
 13 files changed, 148 insertions(+), 1 deletion(-)

diff --git a/frontend/src/components/data-sync/DataSyncDeliveryStage.test.tsx b/frontend/src/components/data-sync/DataSyncDeliveryStage.test.tsx
index 06c260d0..abfdee55 100644
--- a/frontend/src/components/data-sync/DataSyncDeliveryStage.test.tsx
+++ b/frontend/src/components/data-sync/DataSyncDeliveryStage.test.tsx
@@ -19,6 +19,7 @@ const supportedCapability: DataSyncRouteCapability = {
   supportsAutoCreate: true,
   supportsAutoAddColumns: true,
   requiresExistingTarget: false,
+  supportsMutations: true,
   supportsCdc: true,
 };
 
@@ -140,6 +141,43 @@ describe('DataSyncTaskEditor delivery stage', () => {
     });
   });
 
+  it('constrains append-only targets to inserts and removes delete propagation', async () => {
+    const mapping = {
+      ...createDataSyncTableMapping('timeseries:mapping:1', 'orders', 'orders'),
+      keyColumns: ['id'],
+    };
+    const base = createDataSyncTaskDraft({ id: 'timeseries', kind: 'reconcile' });
+    const task = reviseDataSyncTask(base, {
+      source: endpoint('source'),
+      target: endpoint('target'),
+      mappings: [mapping],
+      delivery: {
+        ...base.delivery,
+        writeMode: 'upsert',
+        retryLimit: 3,
+        propagateDeletes: true,
+      },
+    });
+    const { renderer, onPatch } = await renderDelivery(task, vi.fn(), {
+      ...supportedCapability,
+      supportsMutations: false,
+    });
+
+    expect(JSON.stringify(renderer.toJSON())).toContain('当前时序目标仅支持追加写入');
+    expect(renderer.root.findAllByProps({ 'data-delete-propagation': 'true' })).toHaveLength(0);
+    const upsert = renderer.root
+      .findAllByType('option')
+      .find((option) => option.props.value === 'upsert')!;
+    expect(upsert.props.disabled).toBe(true);
+    expect(onPatch).toHaveBeenCalledWith({
+      delivery: expect.objectContaining({
+        writeMode: 'append',
+        retryLimit: 0,
+        propagateDeletes: false,
+      }),
+    });
+  });
+
   it('reveals migration schema controls only for supported implicit same-name mappings', async () => {
     const implicitMapping = {
       ...createDataSyncTableMapping('migration:mapping:1', 'public.orders', 'public.orders'),
diff --git a/frontend/src/components/data-sync/DataSyncTaskEditor.tsx b/frontend/src/components/data-sync/DataSyncTaskEditor.tsx
index 2f75ce61..8e0052da 100644
--- a/frontend/src/components/data-sync/DataSyncTaskEditor.tsx
+++ b/frontend/src/components/data-sync/DataSyncTaskEditor.tsx
@@ -300,6 +300,8 @@ const DeliveryStage: React.FC<{
       (task.kind !== 'cdc' || capability.supportsCdc === true));
   const rowIsolationAvailable =
     routeCanWrite && canUseDataSyncRowErrorIsolation(task);
+  const appendOnlyTarget =
+    capability.level !== 'unknown' && capability.supportsMutations === false;
   const enabledMappings = task.mappings.filter((mapping) => mapping.enabled);
   const allEnabledMappingsHaveKeys =
     enabledMappings.length > 0 &&
@@ -308,6 +310,7 @@ const DeliveryStage: React.FC<{
     );
   const canPropagateDeletes =
     routeCanWrite &&
+    !appendOnlyTarget &&
     task.delivery.writeMode === 'upsert' &&
     ((task.kind === 'reconcile' &&
       task.incremental.mode === 'snapshot' &&
@@ -370,6 +373,10 @@ const DeliveryStage: React.FC<{
     if (!canPropagateDeletes && task.delivery.propagateDeletes) {
       patch.propagateDeletes = false;
     }
+    if (appendOnlyTarget && task.delivery.writeMode !== 'append') {
+      patch.writeMode = 'append';
+      patch.retryLimit = 0;
+    }
     if (
       structureCapabilityResolved &&
       !canAutoAddColumns &&
@@ -391,6 +398,7 @@ const DeliveryStage: React.FC<{
     canAutoAddColumns,
     canCreateIndexes,
     canPropagateDeletes,
+    appendOnlyTarget,
     onPatch,
     readOnly,
     rowIsolationAvailable,
@@ -400,6 +408,7 @@ const DeliveryStage: React.FC<{
     task.delivery.createIndexes,
     task.delivery.errorPolicy,
     task.delivery.propagateDeletes,
+    task.delivery.writeMode,
   ]);
 
   if (readOnly) {
@@ -428,6 +437,11 @@ const DeliveryStage: React.FC<{
         </div>
       </header>
       <div className="gn-data-sync-delivery-main">
+        {appendOnlyTarget ? (
+          <p className="gn-data-sync-inline-note" role="note" data-append-only-target="true">
+            {t('delivery.append_only_target_note')}
+          </p>
+        ) : null}
         <Field label={t('delivery.write_mode')}>
           <select
             className="gn-data-sync-control"
@@ -451,7 +465,11 @@ const DeliveryStage: React.FC<{
             >
               {t('delivery.write.append')}
             </option>
-            <option value="upsert">{t('delivery.write.upsert')}</option>
+            <option value="upsert" disabled={appendOnlyTarget}>
+              {appendOnlyTarget
+                ? t('delivery.write.upsert_unavailable')
+                : t('delivery.write.upsert')}
+            </option>
             <option value="overwrite" disabled>
               {t('delivery.write.overwrite_unavailable')}
             </option>
diff --git a/frontend/src/components/data-sync/DataSyncWorkbenchShell.tsx b/frontend/src/components/data-sync/DataSyncWorkbenchShell.tsx
index dd05cf3d..74eeacf1 100644
--- a/frontend/src/components/data-sync/DataSyncWorkbenchShell.tsx
+++ b/frontend/src/components/data-sync/DataSyncWorkbenchShell.tsx
@@ -46,6 +46,7 @@ const EMPTY_CAPABILITY: DataSyncRouteCapability = {
   level: 'unknown',
   canExecute: false,
   supportsAutoCreate: false,
+  supportsMutations: false,
   supportsCdc: false,
 };
 
diff --git a/frontend/src/components/data-sync/gateway.ts b/frontend/src/components/data-sync/gateway.ts
index f910f0d8..ec3cb62a 100644
--- a/frontend/src/components/data-sync/gateway.ts
+++ b/frontend/src/components/data-sync/gateway.ts
@@ -178,6 +178,7 @@ const unresolvedCapability: DataSyncRouteCapability = {
   level: 'unknown',
   canExecute: false,
   supportsAutoCreate: false,
+  supportsMutations: false,
   supportsCdc: false,
 };
 
diff --git a/frontend/src/components/data-sync/model.ts b/frontend/src/components/data-sync/model.ts
index 55b6fbec..4d122d6c 100644
--- a/frontend/src/components/data-sync/model.ts
+++ b/frontend/src/components/data-sync/model.ts
@@ -275,6 +275,7 @@ export type DataSyncRouteCapability = {
   supportsAutoCreate: boolean;
   supportsAutoAddColumns?: boolean;
   requiresExistingTarget?: boolean;
+  supportsMutations?: boolean;
   supportsCdc: boolean;
 };
 
diff --git a/frontend/src/components/data-sync/text.ts b/frontend/src/components/data-sync/text.ts
index 016efaf8..bbd9adad 100644
--- a/frontend/src/components/data-sync/text.ts
+++ b/frontend/src/components/data-sync/text.ts
@@ -162,6 +162,7 @@ const zhCN = {
   'delivery.write.append': '追加',
   'delivery.write.append_desc': '每次运行都新增记录；重复运行可能产生重复数据，系统会自动关闭重试。',
   'delivery.write.upsert': '新增或更新',
+  'delivery.write.upsert_unavailable': '新增或更新（当前目标不支持）',
   'delivery.write.upsert_desc': '按主键或唯一键识别同一行：不存在则新增，已存在则更新。',
   'delivery.write.overwrite': '覆盖',
   'delivery.write.overwrite_unavailable': '覆盖（当前任务系统暂不支持）',
@@ -182,6 +183,7 @@ const zhCN = {
   'delivery.retry_backoff': '重试退避（毫秒）',
   'delivery.row_isolation_note': '当前配置必须遇错停止。逐行跳过只适用于写入已有关系表，且不能同时自动建表、修改结构、同步删除或覆盖写入。',
   'delivery.append_retry_note': '追加写入可能已部分落库，为避免重复写入，自动重试固定为 0。',
+  'delivery.append_only_target_note': '当前时序目标仅支持追加写入（INSERT）；更新和删除不会同步。',
   'delivery.propagate_deletes': '传播源端删除',
   'delivery.delete_policy_title': '目标端删除策略',
   'delivery.delete_risk_badge': '高风险',
@@ -576,6 +578,7 @@ const enUS: Record<DataSyncWorkbenchTextKey, string> = {
   'delivery.write.append': 'Append',
   'delivery.write.append_desc': 'Each run inserts new rows. Re-running may create duplicates, so automatic retries are disabled.',
   'delivery.write.upsert': 'Insert or update',
+  'delivery.write.upsert_unavailable': 'Insert or update (not supported by this target)',
   'delivery.write.upsert_desc': 'Use a primary or unique key to identify a row: insert when missing and update when present.',
   'delivery.write.overwrite': 'Overwrite',
   'delivery.write.overwrite_unavailable': 'Overwrite (not yet supported for jobs)',
@@ -596,6 +599,7 @@ const enUS: Record<DataSyncWorkbenchTextKey, string> = {
   'delivery.retry_backoff': 'Retry backoff (ms)',
   'delivery.row_isolation_note': 'This configuration must stop on error. Row-level skip is available only for existing relational targets without auto-create, schema changes, delete propagation, or overwrite.',
   'delivery.append_retry_note': 'Append writes may be partially committed. Automatic retries stay at zero to prevent duplicates.',
+  'delivery.append_only_target_note': 'This time-series target supports append/INSERT-only delivery; updates and deletes are not synchronized.',
   'delivery.propagate_deletes': 'Propagate source deletes',
   'delivery.delete_policy_title': 'Target deletion policy',
   'delivery.delete_risk_badge': 'High risk',
diff --git a/frontend/src/components/data-sync/wailsDto.test.ts b/frontend/src/components/data-sync/wailsDto.test.ts
index aa062236..d071191e 100644
--- a/frontend/src/components/data-sync/wailsDto.test.ts
+++ b/frontend/src/components/data-sync/wailsDto.test.ts
@@ -103,6 +103,7 @@ describe('data sync Wails DTO boundary', () => {
         supportsAutoCreate: true,
         supportsAutoAddColumns: true,
         requiresExistingTarget: true,
+        supportsMutations: true,
       }),
     ).toMatchObject({
       level: 'full',
@@ -110,6 +111,7 @@ describe('data sync Wails DTO boundary', () => {
       supportsAutoCreate: true,
       supportsAutoAddColumns: true,
       requiresExistingTarget: true,
+      supportsMutations: true,
     });
     expect(
       decodeRouteCapability({
@@ -120,6 +122,7 @@ describe('data sync Wails DTO boundary', () => {
     ).toMatchObject({
       supportsAutoAddColumns: false,
       requiresExistingTarget: false,
+      supportsMutations: false,
     });
   });
 
diff --git a/frontend/src/components/data-sync/wailsDto.ts b/frontend/src/components/data-sync/wailsDto.ts
index 901a4c1c..1dc13802 100644
--- a/frontend/src/components/data-sync/wailsDto.ts
+++ b/frontend/src/components/data-sync/wailsDto.ts
@@ -831,6 +831,10 @@ export const decodeRouteCapability = (value: unknown): DataSyncRouteCapability =
       capability.requiresExistingTarget,
       'DataSyncCapabilityResolve.data.requiresExistingTarget',
     ),
+    supportsMutations: optionalBoolean(
+      capability.supportsMutations,
+      'DataSyncCapabilityResolve.data.supportsMutations',
+    ),
     supportsCdc: false,
   };
 };
@@ -899,6 +903,7 @@ export const decodeDataSyncPreflight = (
         supportsAutoCreate: false,
         supportsAutoAddColumns: false,
         requiresExistingTarget: false,
+        supportsMutations: false,
         supportsCdc: false,
       };
     }
diff --git a/frontend/src/components/data-sync/wailsGateway.ts b/frontend/src/components/data-sync/wailsGateway.ts
index 2e9c8541..939d532d 100644
--- a/frontend/src/components/data-sync/wailsGateway.ts
+++ b/frontend/src/components/data-sync/wailsGateway.ts
@@ -140,6 +140,7 @@ const UNKNOWN_CAPABILITY: DataSyncRouteCapability = {
   level: 'unknown',
   canExecute: false,
   supportsAutoCreate: false,
+  supportsMutations: false,
   supportsCdc: false,
 };
 
diff --git a/internal/app/data_sync_job_approval_test.go b/internal/app/data_sync_job_approval_test.go
index 80235f28..ddabd808 100644
--- a/internal/app/data_sync_job_approval_test.go
+++ b/internal/app/data_sync_job_approval_test.go
@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"GoNavi-Wails/internal/sync"
 	"GoNavi-Wails/internal/syncjob"
 )
 
@@ -162,3 +163,21 @@ func TestDataSyncJobPreflightDiscardsCallerSuppliedApproval(t *testing.T) {
 		t.Fatal("preflight must not trust or echo a caller-supplied approval")
 	}
 }
+
+func TestAppendOnlyTargetPreflightIssuesBlockMutations(t *testing.T) {
+	definition := approvalTestDefinition()
+	definition.Options.SyncMode = "insert_update"
+	definition.Options.PropagateDeletes = true
+	issues := appendOnlyTargetPreflightIssues(definition, sync.MigrationCapability{
+		TargetType:        "tdengine",
+		SupportsMutations: false,
+	})
+	if len(issues) != 2 || issues[0].Code != "append_only_target_requires_insert_only" || issues[1].Code != "append_only_target_delete_unsupported" {
+		t.Fatalf("unexpected append-only target issues: %#v", issues)
+	}
+
+	definition.Kind = syncjob.JobKindCompare
+	if issues := appendOnlyTargetPreflightIssues(definition, sync.MigrationCapability{TargetType: "tdengine"}); len(issues) != 0 {
+		t.Fatalf("compare task must not be blocked by write capability: %#v", issues)
+	}
+}
diff --git a/internal/app/data_sync_job_preflight.go b/internal/app/data_sync_job_preflight.go
index cb4ce401..d0e68979 100644
--- a/internal/app/data_sync_job_preflight.go
+++ b/internal/app/data_sync_job_preflight.go
@@ -58,6 +58,7 @@ func (a *App) preflightDataSyncJob(input syncjob.JobDefinition, now time.Time) D
 		message := fmt.Sprintf("migration route %s -> %s is %s", result.Capability.SourceType, result.Capability.TargetType, result.Capability.SupportLevel)
 		result.Issues = append(result.Issues, preflightIssue("route_unsupported", DataSyncJobPreflightBlocker, "endpoints", message, ""))
 	}
+	result.Issues = append(result.Issues, appendOnlyTargetPreflightIssues(definition, result.Capability)...)
 	if definition.Kind != syncjob.JobKindCompare {
 		for _, mapping := range definition.Mappings {
 			if !mapping.Enabled {
@@ -225,6 +226,32 @@ func (a *App) preflightDataSyncJob(input syncjob.JobDefinition, now time.Time) D
 	return finishDataSyncJobPreflight(result)
 }
 
+func appendOnlyTargetPreflightIssues(definition syncjob.JobDefinition, capability sync.MigrationCapability) []DataSyncJobPreflightIssue {
+	if definition.Kind == syncjob.JobKindCompare || capability.SupportsMutations {
+		return nil
+	}
+	issues := make([]DataSyncJobPreflightIssue, 0, 2)
+	if !strings.EqualFold(definition.Options.SyncMode, "insert_only") {
+		issues = append(issues, preflightIssue(
+			"append_only_target_requires_insert_only",
+			DataSyncJobPreflightBlocker,
+			"delivery",
+			fmt.Sprintf("%s targets support append/INSERT-only delivery; updates are not supported", capability.TargetType),
+			"",
+		))
+	}
+	if definition.Options.PropagateDeletes {
+		issues = append(issues, preflightIssue(
+			"append_only_target_delete_unsupported",
+			DataSyncJobPreflightBlocker,
+			"delivery",
+			fmt.Sprintf("%s targets support append/INSERT-only delivery; deletes are not supported", capability.TargetType),
+			"",
+		))
+	}
+	return issues
+}
+
 func (a *App) preflightDataSyncMappings(definition syncjob.JobDefinition, source, target resolvedDataSyncJobEndpoint) []DataSyncJobPreflightIssue {
 	issues := make([]DataSyncJobPreflightIssue, 0)
 	for _, mapping := range definition.Mappings {
diff --git a/internal/sync/migration_capability.go b/internal/sync/migration_capability.go
index 8e4a8403..7c3963ea 100644
--- a/internal/sync/migration_capability.go
+++ b/internal/sync/migration_capability.go
@@ -16,6 +16,7 @@ type MigrationCapability struct {
 	SupportsAutoCreate     bool                  `json:"supportsAutoCreate"`
 	SupportsAutoAddColumns bool                  `json:"supportsAutoAddColumns"`
 	RequiresExistingTarget bool                  `json:"requiresExistingTarget"`
+	SupportsMutations      bool                  `json:"supportsMutations"`
 }
 
 // ResolveMigrationCapability returns the migration runtime's effective support
@@ -28,6 +29,12 @@ func ResolveMigrationCapability(sourceConfig connection.ConnectionConfig, target
 		TargetType:  targetType,
 		SourceModel: classifyMigrationDataModel(sourceType),
 		TargetModel: classifyMigrationDataModel(targetType),
+		// Most writable targets support the job engine's update/delete semantics.
+		// Time-series targets are deliberately narrowed below.
+		SupportsMutations: true,
+	}
+	if targetType == "tdengine" || targetType == "iotdb" {
+		capability.SupportsMutations = false
 	}
 	syncConfig := SyncConfig{
 		SourceConfig: sourceConfig,
diff --git a/internal/sync/migration_capability_test.go b/internal/sync/migration_capability_test.go
index a1f5ff46..000da719 100644
--- a/internal/sync/migration_capability_test.go
+++ b/internal/sync/migration_capability_test.go
@@ -21,6 +21,7 @@ func TestResolveMigrationCapability_MySQLToPostgresUsesFullPlanner(t *testing.T)
 		SupportsAutoCreate:     true,
 		SupportsAutoAddColumns: true,
 		RequiresExistingTarget: false,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -44,6 +45,7 @@ func TestResolveMigrationCapability_PostgresToKingbaseUsesSameFamilyPlanner(t *t
 		SupportsAutoCreate:     true,
 		SupportsAutoAddColumns: true,
 		RequiresExistingTarget: false,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -67,6 +69,7 @@ func TestResolveMigrationCapability_OracleToSQLServerUsesExistingTargetCompatibi
 		SupportsAutoCreate:     false,
 		SupportsAutoAddColumns: false,
 		RequiresExistingTarget: true,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -90,6 +93,7 @@ func TestResolveMigrationCapability_MongoToOracleReportsPlannedNonExecutablePath
 		SupportsAutoCreate:     false,
 		SupportsAutoAddColumns: false,
 		RequiresExistingTarget: true,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -113,6 +117,7 @@ func TestResolveMigrationCapability_RedisToMongoReportsFullKeyspaceBridge(t *tes
 		SupportsAutoCreate:     true,
 		SupportsAutoAddColumns: false,
 		RequiresExistingTarget: false,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -136,6 +141,7 @@ func TestResolveMigrationCapability_MongoToRedisReportsFullKeyspaceBridge(t *tes
 		SupportsAutoCreate:     true,
 		SupportsAutoAddColumns: false,
 		RequiresExistingTarget: false,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -159,6 +165,7 @@ func TestResolveMigrationCapability_CustomPostgresUsesResolvedDriverFamily(t *te
 		SupportsAutoCreate:     true,
 		SupportsAutoAddColumns: true,
 		RequiresExistingTarget: false,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -182,6 +189,7 @@ func TestResolveMigrationCapability_KafkaToQdrantIsUnsupported(t *testing.T) {
 		SupportsAutoCreate:     false,
 		SupportsAutoAddColumns: false,
 		RequiresExistingTarget: true,
+		SupportsMutations:      true,
 	}
 
 	if got != want {
@@ -200,6 +208,20 @@ func TestResolveMigrationCapability_RedisToNonMongoTargetIsUnsupported(t *testin
 	}
 }
 
+func TestResolveMigrationCapability_TimeSeriesTargetsAreAppendOnly(t *testing.T) {
+	for _, targetType := range []string{"tdengine", "iotdb"} {
+		t.Run(targetType, func(t *testing.T) {
+			capability := ResolveMigrationCapability(
+				connection.ConnectionConfig{Type: "mysql"},
+				connection.ConnectionConfig{Type: targetType},
+			)
+			if !capability.CanExecute || capability.SupportsMutations {
+				t.Fatalf("expected %s target to be executable and append-only, got %+v", targetType, capability)
+			}
+		})
+	}
+}
+
 func TestResolveMigrationCapability_GoldenDBUsesMySQLPlannerFamily(t *testing.T) {
 	got := ResolveMigrationCapability(
 		connection.ConnectionConfig{Type: "goldendb"},