diff --git a/CHANGELOG.md b/CHANGELOG.md index ec40fe3..cf48272 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,8 @@ ### Fixed +- 修复钉钉新登记来源仍按旧离线目录身份取件的问题：从当前来源配置读取空间坐标，实时枚举所在目录并严格匹配节点身份；同名文件不替代、目录不完整不误报删除。修复解析监控遗漏 `version_id` 取件任务及缺少业务元数据的失败记录，正确显示失败原因、重试入口与历史。 + - 修复通过带端口的 HTTP 地址访问问答运营时，正常同源操作被误报“拒绝跨站配置请求”：网页代理保留原始 Host 及端口，使缺少 Fetch Metadata 的浏览器也能正确执行 Origin 校验。保留后端跨站、异端口、伪造转发头与缺失操作意图的拒绝保护。 - **登记钉钉知识库失败（HTTP 409 `space_registration_conflict`）的真实原因与修复**：`source_connections.name` 有唯一约束，而早期路径登记的钉钉来源（`base_url` 为空、id 与 dws 现在推导的确定性 id 不同）被**归档后仍占着名字槽位** ⇒ 重新登记时 INSERT 撞唯一约束；处理器又把**所有**存储层错误都翻译成同一个 409，真实原因被吞掉（线上只能看到一个无细节的错误码）。现在：① 同名来源**已归档**时，把它改名腾位（`…（已下架 · <短ID>）`，归档行不显示、数据与审计一行不动）；② 同名来源**仍在使用**时返回 `space_name_conflict` + 可行动提示（先「去掉来源」）；③ 同一登记键指向别的空间且该行仍在用时返回 `space_identity_conflict`（已归档的行则允许就地改写 —— 这就是"去掉后再用 dws 重新获取"）；④ 其余内部故障改回 **500** `space_registration_failed`，不再伪装成冲突；前端优先显示后端 `hint`，并为这两个新码补了中文说明。 diff --git a/internal/platform/postgres/artifact_fetch_manifest_integration_test.go b/internal/platform/postgres/artifact_fetch_manifest_integration_test.go new file mode 100644 index 0000000..6b424d9 --- /dev/null +++ b/internal/platform/postgres/artifact_fetch_manifest_integration_test.go @@ -0,0 +1,53 @@ +package postgres + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + "knowledge-governance/internal/workflows/artifactfetch" +) + +func TestArtifactManifestCurrentSourceIntegration(t *testing.T) { + url := isolatedIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL")) + if url == "" { + t.Skip("TEST_DATABASE_URL is not set") + } + ctx := context.Background() + pool, err := pgxpool.New(ctx, url) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + if err = ApplyMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + if _, err = pool.Exec(ctx, `INSERT INTO source_connections(id,source_type,name,base_url) VALUES('conn-live','dingtalk','Renamed KB','spaces/workspace1?space=workspace1'); +INSERT INTO catalog_import_intents(source_connection_id,node_id,visibility,decided_by,relative_path,name,locator_hash,fetch_state) VALUES('conn-live','sn-live','internal','tester','Folder/Manual.adoc','Manual.adoc',repeat('a',64),'pending')`); err != nil { + t.Fatal(err) + } + manifest, err := NewStore(pool).LoadArtifactFetchManifest(ctx, "conn-live", "sn-live") + if err != nil { + t.Fatal(err) + } + dir := t.TempDir() + script := filepath.Join(dir, "manifest_probe.py") + if err = os.WriteFile(script, []byte("import json,sys\nm=json.loads(sys.argv[1])\nprint(json.dumps({'ok':m.get('source_root')=='spaces/workspace1?space=workspace1' and m.get('connection_id')=='conn-live','node_id':m['node_id']}))\n"), 0600); err != nil { + t.Fatal(err) + } + python, err := exec.LookPath("python3") + if err != nil { + t.Fatal(err) + } + runner, err := artifactfetch.NewCommandRunner(python, script, dir, "", 4096) + if err != nil { + t.Fatal(err) + } + result, err := runner.Run(ctx, manifest) + if err != nil || !result.OK { + t.Fatalf("current source coordinates lost across DB/Go/Python: %#v %v", result, err) + } +} diff --git a/internal/platform/postgres/artifact_fetch_store.go b/internal/platform/postgres/artifact_fetch_store.go index 6eaea42..e0a6c14 100644 --- a/internal/platform/postgres/artifact_fetch_store.go +++ b/internal/platform/postgres/artifact_fetch_store.go @@ -22,12 +22,13 @@ func loadArtifactFetchManifest(ctx context.Context, db interface { var manifest artifactfetch.Manifest var fetchState string err := db.QueryRow(ctx, `SELECT connection.source_type,connection.name,intent.relative_path,intent.name, - intent.size_bytes,intent.modified_at,intent.locator_hash,intent.fetch_state + intent.size_bytes,intent.modified_at,intent.locator_hash,intent.fetch_state, + CASE WHEN connection.source_type='dingtalk' THEN connection.base_url ELSE '' END FROM catalog_import_intents intent JOIN source_connections connection ON connection.id=intent.source_connection_id WHERE intent.source_connection_id=$1 AND intent.node_id=$2`, connectionID, nodeID). Scan(&manifest.SourceType, &manifest.SourceName, &manifest.RelativePath, &manifest.Name, - &manifest.SizeBytes, &manifest.ModifiedAt, &manifest.LocatorHash, &fetchState) + &manifest.SizeBytes, &manifest.ModifiedAt, &manifest.LocatorHash, &fetchState, &manifest.SourceRoot) if errors.Is(err, pgx.ErrNoRows) { return artifactfetch.Manifest{}, artifactfetch.ErrManifestGone } diff --git a/internal/platform/postgres/monitor_tasks.go b/internal/platform/postgres/monitor_tasks.go index 23c52b6..7885c68 100644 --- a/internal/platform/postgres/monitor_tasks.go +++ b/internal/platform/postgres/monitor_tasks.go @@ -54,7 +54,7 @@ const monitorTasksCTE = `WITH history_publications AS ( WHERE $7<>'' AND (hv.source_item_id,hp.test_kb_id)=(SELECT v.source_item_id,p.test_kb_id FROM test_publications p JOIN source_item_versions v ON v.id=p.source_item_version_id WHERE p.id=$7) ), history_events AS ( SELECT updated_at AS at,'版本状态' AS kind,source_item_version_id || ' · ' || processing_state || ' / ' || retrieval_state AS detail FROM history_publications - UNION ALL SELECT f.created_at,'失败记录',f.job_type || ' · ' || f.error_code || ' · 尝试 ' || f.attempt FROM job_failures f WHERE f.publication_id IN(SELECT id FROM history_publications) OR (f.job_type='artifact.fetch' AND f.business_version_id IN(SELECT source_item_version_id FROM history_publications)) + UNION ALL SELECT f.created_at,'失败记录',f.job_type || ' · ' || f.error_code || ' · 尝试 ' || f.attempt FROM job_failures f LEFT JOIN jobs fj ON fj.id=f.job_id WHERE f.publication_id IN(SELECT id FROM history_publications) OR (f.job_type='artifact.fetch' AND COALESCE(NULLIF(f.business_version_id,''),fj.payload->>'version_id') IN(SELECT source_item_version_id FROM history_publications)) UNION ALL SELECT r.created_at,'重试记录',r.reason || ' · ' || r.outcome FROM job_retry_operations r WHERE r.publication_id IN(SELECT id FROM history_publications) ), latest AS ( SELECT DISTINCT ON(v.source_item_id,p.test_kb_id) p.*,v.source_item_id,v.content_object_id @@ -70,14 +70,14 @@ const monitorTasksCTE = `WITH history_publications AS ( CASE WHEN p.retrieval_state IN ('disabling','disabled') OR (p.knowledge_id<>'' AND si.policy_state='blocked') OR j.type='test.remove' THEN 'remove' WHEN EXISTS(SELECT 1 FROM test_publications prev JOIN source_item_versions pv ON pv.id=prev.source_item_version_id WHERE pv.source_item_id=si.id AND prev.test_kb_id=p.test_kb_id AND prev.id<>p.id AND prev.created_at>'publication_id'=p.id OR (j.type='artifact.fetch' AND j.payload->>'source_item_version_id'=p.source_item_version_id)) + (j.payload->>'publication_id'=p.id OR (j.type='artifact.fetch' AND j.payload->>'version_id'=p.source_item_version_id)) AND j.type IN ('artifact.fetch','test.upload','test.poll','test.remove') ORDER BY CASE WHEN j.state IN ('pending','running','retry') THEN 0 ELSE 1 END,j.updated_at DESC,j.id DESC LIMIT 1) j ON TRUE - LEFT JOIN LATERAL(SELECT f.id,f.job_type,f.error_code,f.created_at FROM job_failures f WHERE f.state='open' AND (f.publication_id=p.id OR (f.job_type='artifact.fetch' AND f.business_version_id=p.source_item_version_id)) ORDER BY f.created_at DESC,f.id DESC LIMIT 1) f ON TRUE + LEFT JOIN LATERAL(SELECT f.id,f.job_type,f.error_code,f.last_error,f.created_at FROM job_failures f LEFT JOIN jobs fj ON fj.id=f.job_id WHERE f.state='open' AND (f.publication_id=p.id OR (f.job_type='artifact.fetch' AND COALESCE(NULLIF(f.business_version_id,''),fj.payload->>'version_id')=p.source_item_version_id)) ORDER BY f.created_at DESC,f.id DESC LIMIT 1) f ON TRUE ), classified AS ( SELECT *, CASE WHEN action='remove' AND retrieval_state='disabled' THEN 'completed' @@ -85,7 +85,7 @@ const monitorTasksCTE = `WITH history_publications AS ( WHEN job_state IN ('pending','retry') AND job_attempt>0 THEN 'retrying' WHEN job_state='running' THEN 'running' WHEN job_state IN ('pending','retry') THEN 'waiting' - WHEN failure_id<>'' OR processing_state IN ('failed','stalled') OR ingest_state='failed' OR retrieval_state='failed' THEN 'attention' + WHEN job_state='failed' OR failure_id<>'' OR processing_state IN ('failed','stalled') OR ingest_state='failed' OR retrieval_state='failed' THEN 'attention' WHEN processing_state='cancelled' THEN 'cancelled' WHEN action='remove' THEN 'waiting' WHEN processing_state IN ('processing','finalizing') THEN 'running' diff --git a/internal/platform/postgres/monitor_tasks_integration_test.go b/internal/platform/postgres/monitor_tasks_integration_test.go index d3ba7e7..8c5736e 100644 --- a/internal/platform/postgres/monitor_tasks_integration_test.go +++ b/internal/platform/postgres/monitor_tasks_integration_test.go @@ -102,4 +102,30 @@ func TestMonitorTasksUnifiedPaginationIntegration(t *testing.T) { if err = store.RetryMonitorRemoval(ctx, "mt-remove", "tester", now); err != ErrPublicationStateConflict { t.Fatalf("changed policy accepted: %v", err) } + t.Run("fetch payload identity without business metadata", func(t *testing.T) { + insertReviewFixture(t, ctx, pool, 7, "mt-fetch", now) + exec(`UPDATE test_publications SET processing_state='planned',ingest_state='pending',knowledge_id='' WHERE id='mt-fetch'`) + exec(`INSERT INTO jobs(id,type,idempotency_key,payload,state,attempt,last_error) VALUES('mt-fetch-job','artifact.fetch','mt-fetch-key','{"version_id":"sv-review-7","source_connection_id":"conn-manual","node_id":"sn-fetch"}','failed',1,'DingTalk node not found in current inventory')`) + exec(`INSERT INTO job_failures(id,job_id,job_type,error_code,attempt,max_attempts_at_failure,last_error) VALUES('mt-fetch-failure','mt-fetch-job','artifact.fetch','artifact_invalid',1,8,'DingTalk node not found in current inventory')`) + check := func(status string, retryable int) { + t.Helper() + page, err := store.ListMonitorTasks(ctx, MonitorTaskFilter{ID: "mt-fetch", Status: status, Stage: "fetch", Limit: 1}) + if err != nil || page.Total != 1 || len(page.Items) != 1 || page.Retryable != retryable { + t.Fatalf("fetch status %s: %#v %v", status, page, err) + } + if status == "attention" { + row := page.Items[0] + if row.FailureID != "mt-fetch-failure" || row.RetryKind != "fetch" || row.Detail != "DingTalk node not found in current inventory" || len(row.History) < 2 { + t.Fatalf("failure details/history: %#v", row) + } + } + } + check("attention", 1) + exec(`UPDATE jobs SET state='pending',attempt=0 WHERE id='mt-fetch-job'`) + check("waiting", 0) + exec(`UPDATE jobs SET state='running',attempt=1 WHERE id='mt-fetch-job'`) + check("running", 0) + exec(`UPDATE jobs SET state='retry' WHERE id='mt-fetch-job'`) + check("retrying", 0) + }) } diff --git a/internal/workflows/artifactfetch/handler.go b/internal/workflows/artifactfetch/handler.go index 109b217..59fbb3b 100644 --- a/internal/workflows/artifactfetch/handler.go +++ b/internal/workflows/artifactfetch/handler.go @@ -49,6 +49,7 @@ type Manifest struct { Request SourceType string SourceName string + SourceRoot string RelativePath string Name string SizeBytes *int64 @@ -218,7 +219,7 @@ func stderrDetail(tail string) string { func (runner *CommandRunner) Run(ctx context.Context, manifest Manifest) (Result, error) { manifestJSON, err := json.Marshal(map[string]any{ "spec_id": manifest.VersionID, "node_id": manifest.NodeID, "connection_id": manifest.ConnectionID, - "source_type": manifest.SourceType, "source_name": manifest.SourceName, + "source_type": manifest.SourceType, "source_name": manifest.SourceName, "source_root": manifest.SourceRoot, "relative_path": manifest.RelativePath, "name": manifest.Name, "size_bytes": manifest.SizeBytes, "locator_hash": manifest.LocatorHash, }) diff --git a/scripts/artifact_fetch.py b/scripts/artifact_fetch.py index cbd720b..1c00c2e 100644 --- a/scripts/artifact_fetch.py +++ b/scripts/artifact_fetch.py @@ -371,8 +371,55 @@ def dingtalk_local_export(manifest): return str(candidate), fresh +def dingtalk_live_raw_key(manifest): + """Resolve in the configured source; names locate folders, never file identity.""" + from live_catalog_refresh import dingtalk_root_key + from source_catalog_import import node_id as catalog_node_id, safe_relative_path + from dws_dingtalk_scan import normalise + + match = re.fullmatch(r"spaces/([A-Za-z0-9_-]+)\?space=([A-Za-z0-9_-]+)", manifest['source_root']) + if not match: + raise FetchError('DingTalk source root is invalid', retryable=False, code='source_identity') + root_id, space_id = match.groups() + try: + source_key = 'dingtalk-' + dingtalk_root_key(manifest['connection_id'], space_id) + parts = safe_relative_path(manifest['relative_path']).split('/') + except (OSError, ValueError, RuntimeError): + raise FetchError('DingTalk source identity cannot be resolved', retryable=False, code='source_identity') + folder = '' + for index in range(len(parts)): + drive = space_id.isdigit() + args = ['drive', '+list', '--space-id', space_id] if drive else ['wiki', '+node-list', '--workspace', space_id] + args += ['--page-all', '--limit', '50', '--max-items', '10000', '--page-limit', '200'] + if folder: + args += ['--folder', folder] + payload, verdict = dws_run(args, timeout=180) + if not verdict['ok']: + raise _dws_error(verdict) + data = payload.get('data', {}) if isinstance(payload, dict) else {} + entries = data.get('files' if drive else 'nodes') + if not isinstance(entries, list) or data.get('hasMore') is not False or data.get('autoPageComplete') is not True: + raise FetchError('DingTalk directory response is incomplete', retryable=True, code='source_lookup_incomplete') + matches = [] + for entry in entries: + raw, node, is_folder = normalise(entry, folder or root_id) + if index == len(parts) - 1: + if not is_folder and catalog_node_id(source_key, raw) == manifest['node_id']: + matches.append(raw) + elif is_folder and node['name'] == parts[index]: + matches.append(raw) + if len(matches) != 1: + # A missing/ambiguous path is not evidence that the source file was deleted. + raise FetchError('DingTalk node identity unresolved; refresh 来源目录 before retrying', + retryable=False, code='source_node_unresolved') + folder = matches[0] + return folder, 'file' + + def dingtalk_raw_keys(manifest): - """Recover the dentry uuid behind a node id via the scanner inventory.""" + """Use current source coordinates; retain inventory support for old manifests.""" + if manifest.get('source_root'): + return dingtalk_live_raw_key(manifest) state_path = pathlib.Path("/sources/dingtalk/data/dingtalk-internal/inventory-state.json") state = json.loads(state_path.read_text(encoding="utf-8")) roots = state.get("roots") or {} diff --git a/scripts/artifact_fetch_test.py b/scripts/artifact_fetch_test.py index c25841b..1f5e7bc 100644 --- a/scripts/artifact_fetch_test.py +++ b/scripts/artifact_fetch_test.py @@ -69,6 +69,78 @@ class DingTalkOnlineDocumentExportTest(unittest.TestCase): download.assert_called_once_with("https://oss.example/result.docx", {}, 180, 1_000_000) +class DingTalkCurrentIdentityTest(unittest.TestCase): + def test_current_source_resolves_live_node_not_same_named_legacy_node(self): + from source_catalog_import import node_id, stable_connection_id + source = 'dingtalk-space-workspace1' + manifest = { + 'source_root': 'spaces/workspace1?space=workspace1', + 'connection_id': stable_connection_id(source), + 'source_name': 'Renamed KB', 'node_id': node_id(source, 'current-raw'), + 'relative_path': 'Folder/Manual.adoc', 'name': 'Manual.adoc', + } + legacy = {'roots': {'space-legacy': {'live_space_name': 'Renamed KB', 'nodes': { + 'wrong-raw': {'name': 'Manual.adoc', 'dentry_type': 'file'}, + }}}} + pages = [ + [{'nodeId': 'folder-raw', 'name': 'Folder', 'nodeType': 'folder'}], + [{'nodeId': 'wrong-raw', 'name': 'Manual', 'nodeType': 'file', 'extension': 'adoc'}, + {'nodeId': 'current-raw', 'name': 'Manual', 'nodeType': 'file', 'extension': 'adoc'}], + ] + responses = [({'data': {'nodes': page, 'hasMore': False, 'autoPageComplete': True}}, {'ok': True}) for page in pages] + with mock.patch.object(pathlib.Path, 'read_text', return_value=json.dumps(legacy)), \ + mock.patch.object(artifact_fetch, 'dws_run', side_effect=responses) as listing: + self.assertEqual(artifact_fetch.dingtalk_raw_keys(manifest), ('current-raw', 'file')) + self.assertEqual(listing.call_count, 2) + self.assertIn('workspace1', listing.call_args_list[0].args[0]) + self.assertEqual(listing.call_args_list[1].args[0][-2:], ['--folder', 'folder-raw']) + + + def test_identity_and_incomplete_listing_fail_closed_without_false_deletion(self): + from source_catalog_import import node_id, stable_connection_id + source = 'dingtalk-space-workspace1' + manifest = {'source_root': 'spaces/workspace1?space=workspace1', + 'connection_id': stable_connection_id(source), 'node_id': node_id(source, 'right'), + 'name': 'Manual.adoc', 'relative_path': 'Manual.adoc'} + wrong = {'nodeId': 'wrong', 'name': 'Manual.adoc', 'nodeType': 'file'} + cases = [ + ({'nodes': [wrong], 'hasMore': False, 'autoPageComplete': True}, 'source_node_unresolved'), + ({'nodes': [], 'hasMore': True, 'autoPageComplete': False}, 'source_lookup_incomplete'), + ({'nodes': [], 'hasMore': False}, 'source_lookup_incomplete'), + ] + for data, code in cases: + with self.subTest(code=code, data=data), mock.patch.object(artifact_fetch, 'dws_run', return_value=({'data': data}, {'ok': True})): + with self.assertRaises(artifact_fetch.FetchError) as failure: + artifact_fetch.dingtalk_raw_keys(manifest) + self.assertNotIsInstance(failure.exception, artifact_fetch.MissingSource) + self.assertEqual(failure.exception.code, code) + with mock.patch.object(artifact_fetch, 'dws_run') as call: + with self.assertRaises(artifact_fetch.FetchError): + artifact_fetch.dingtalk_raw_keys({**manifest, 'source_root': 'invalid'}) + call.assert_not_called() + verdict = artifact_fetch.dws_error_map.classify({'error': {'message': 'invalidParameter.authCode.notFound'}}) + with mock.patch.object(artifact_fetch, 'dws_run', return_value=(None, verdict)): + with self.assertRaises(artifact_fetch.FetchError) as failure: + artifact_fetch.dingtalk_raw_keys(manifest) + self.assertEqual(failure.exception.code, 'credentials') + self.assertFalse(failure.exception.retryable) + + def test_legacy_namespace_and_drive_keep_their_own_identity(self): + from source_catalog_import import node_id, stable_connection_id + for space, root_key, drive in [('workspace1', 'space-old', False), ('12345', 'space-12345', True)]: + source = 'dingtalk-' + root_key + manifest = {'source_root': f'spaces/{space}?space={space}', + 'connection_id': stable_connection_id(source), 'node_id': node_id(source, 'right'), + 'name': 'Manual.adoc', 'relative_path': 'Manual.adoc'} + data = {'files' if drive else 'nodes': [{'nodeId': 'right', 'name': 'Manual', 'type': 'file'}], + 'hasMore': False, 'autoPageComplete': True} + with self.subTest(space=space), \ + mock.patch.object(pathlib.Path, 'read_text', return_value=json.dumps({'roots': {'space-old': {}}})), \ + mock.patch.object(artifact_fetch, 'dws_run', return_value=({'data': data}, {'ok': True})) as call: + self.assertEqual(artifact_fetch.dingtalk_raw_keys(manifest), ('right', 'file')) + self.assertEqual(call.call_args.args[0][0], 'drive' if drive else 'wiki') + + class DingTalkDwsFetcherTest(unittest.TestCase): """dws 取件器：路由、回落语义、以及"结果 JSON 形状不得变化"的契约锁定。"""