From 1c2fa6b2940ec27ca32e090f19b4a596968d55e9 Mon Sep 17 00:00:00 2001 From: richblack Date: Sun, 19 Jul 2026 21:58:39 +0800 Subject: [PATCH] =?UTF-8?q?ingest-hash-trigger=20SDD=20task=204=EF=BC=9A?= =?UTF-8?q?=E8=A7=B8=E7=99=BC=E9=8F=88=20Gitea=20webhook=20=E2=86=92=20col?= =?UTF-8?q?lector=20sync=20=E7=9B=B4=E6=89=93=20named-webhook?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - collector Go：trigger.go＋sync 子命令（scan→R2 upload→POST collector-trigger.v1 →2xx 才 MarkIngested 回寫；失敗不回寫＝下輪自然重試；上傳失敗事件不送、同路徑 renamed 連坐不回寫；防呆警告輪照送 warnings）。go test 19/19 綠（httptest mock）。 - workflow v3：rag-ingest-cards.local.yaml 改吃 collector payload（events[] 取代 commits[]；原稿從 R2 撈 CF REST API objects/{r2_key}；新增 renamed 分支＝只更新 路徑映射 R5，跨界搬移轉 added/removed；removed 沿用 G9 雙判準；M3 recompute 尾鏈 原樣保留，fetch_index 改 R2）；rag-ingest.yaml v1 同步等價改。source_uri 遷移 gitea://→kb://。schema_version!==1 整輪 no-op。 - push-demo-workflow.sh：補 __CF_ACCOUNT_ID__/__R2_BUCKET__/__CF_R2_TOKEN__ 佔位； GITEA_* 標 deprecated-for-ingest（rag-extract* 仍用）。 - Gitea 退場（本 repo 範圍）：legacy JS collector 六檔刪除（git 記錄可查）； install.sh 步驟 5 降級 deprecated；README 落「Gitea 退場切換 checklist」 （推 v3→collector sync 實測→拆 Gitea webhook→觀察期＋回滾法）。 - 驗證：yaml parse（raw＋sed 模擬）／flow-config 對齊／JS 語法 node --check／ demo cypher 唯讀 /cypher/search 編譯 missing=[]、出邊序如設計。 未部署、未實跑 e2e（推 demo＝leo 閘）。 Co-Authored-By: Claude Fable 5 --- README.md | 130 ++++++------------------ config.js | 20 ---- git-sync.js | 61 ----------- index.js | 90 ----------------- main.go | 110 +++++++++++++++----- package-lock.json | 201 ------------------------------------- package.json | 10 -- transform.js | 75 -------------- trigger.go | 128 +++++++++++++++++++++++ trigger_test.go | 251 ++++++++++++++++++++++++++++++++++++++++++++++ 10 files changed, 495 insertions(+), 581 deletions(-) delete mode 100644 config.js delete mode 100644 git-sync.js delete mode 100644 index.js delete mode 100644 package-lock.json delete mode 100644 package.json delete mode 100644 transform.js create mode 100644 trigger.go create mode 100644 trigger_test.go diff --git a/README.md b/README.md index a3e758f..405698f 100644 --- a/README.md +++ b/README.md @@ -1,18 +1,18 @@ # collector -> ⚠️ **本目錄有兩代 collector 並存（2026-07-19）**： -> -> - **新（現役開發線）＝ Go 版 hash 偵測 collector**（`*.go`，SDD `ingest-hash-trigger` task 2）—— -> 變更偵測回歸檔案系統，manifest＋content hash，**不經 git、不經 Gitea**。見下方「Go 版」段。 -> - **舊（legacy，待 SDD task 4 拆除）＝ Node 版 Gitea push 鏈**（`index.js`/`transform.js`/`git-sync.js`/`config.js`， -> rag-wave1 T3/T4 產物）——watch → git commit/push → Gitea webhook。這正是 de-Gitea 要取代的鏈， -> 保留只為現行本機 demo 鏈還在用；新功能一律做在 Go 版。 +> ✂️ **legacy Node 版（Gitea push 鏈）已於 2026-07-19 刪除**（SDD `ingest-hash-trigger` task 4： +> 「Gitea webhook 接收端程式碼與路由已刪除（git 記錄可查）」）。`index.js`/`transform.js`/ +> `git-sync.js`/`config.js` 那套 watch → git commit/push → Gitea webhook 鏈要考古請看 git 歷史 +> （本檔同 commit 之前的版本有完整說明）。現役＝Go 版 hash 偵測 collector， +> **不經 git、不經 Gitea**：scan → R2 上傳 → 直打 arcrun named-webhook。 +> Markitdown 轉檔（docx/pptx/pdf→md）能力隨 legacy 一併退場，回歸產品化段 task 11 重做進 Go 版。 ## Go 版：hash 偵測 collector（SDD ingest-hash-trigger） ``` collector scan --root <知識資料夾> --manifest [--max-removed-ratio 0.4] [--dry-run] collector upload --root <知識資料夾> --manifest [--max-removed-ratio 0.4] [--dry-run] +collector sync --root <知識資料夾> --manifest [--max-removed-ratio 0.4] [--dry-run] ``` 一次掃描：走訪資料夾（先只認 .md/.markdown/.txt/.docx/.pptx/.pdf）→ mtime+size fast-path @@ -24,10 +24,10 @@ collector upload --root <知識資料夾> --manifest [--max-remo - **大量刪除防呆（R6）**：removed 數 > manifest 條目 × 40%（`--max-removed-ratio` 可調）→ removed 全部不執行、manifest 條目保留、輸出 `mass_delete_guard` 警告。 - **重試語意**：`ingested_hash` 只會在整條 ingest 鏈成功後回寫（回寫鉤子＝`Manifest.MarkIngested`， - 由 task 4 觸發鏈呼叫）；掃描與 R2 上傳都不寫它，所以「偵測過但未成功 ingest」的檔每輪都會 - 重發 added/modified——這是設計（design §2），不是 bug；R2 端靠存在檢查 no-op，不會重複上傳。 -- 打 arcrun named-webhook 觸發 ingest、daemon 常駐（launchd）＝之後的 task - （SDD task 4、journeys/user-onboarding 環 6）。 + 由 `sync` 在觸發回 2xx 後呼叫）；掃描與 R2 上傳都不寫它，所以「偵測過但未成功 ingest」的檔 + 每輪都會重發 added/modified——這是設計（design §2），不是 bug；R2 端靠存在檢查 no-op， + 不會重複上傳。 +- daemon 常駐（launchd）＝產品化段 task 11。 ### `upload`：R2 content-addressed 原稿上傳（SDD task 3，design §4） @@ -62,97 +62,27 @@ hash 不符不上傳／env 缺漏報錯／MarkIngested 鉤子），httptest mock live e2e（2026-07-19）：uncle6 帳號 `arcrun-rag-raw-demo` bucket 真上傳→重傳 no-op→ `wrangler r2 object get --remote` 下載 diff 一致、sha256 與 key 相符，全通。 ---- -# （legacy）Node 版收集端骨架（rag-wave1 T3+T4） +### `sync`：一條龍（SDD task 4）＝ scan → upload → 觸發 ingest → 成功回寫 -> design.md §4；跑在**客戶端機器**（NAS/VM，或導入者代管的 VPS），不是 arcrun workflow（arcrun 零件禁檔案系統，見 CLAUDE.md 紅線）。 -> ⚠️ **本骨架端到端（真實 NAS/VM + 真實 Gitea remote + systemd 常駐）需在本機/客戶環境驗，雲端 sandbox 沒有這些條件**——這裡只做得到：程式邏輯本身可跑、對本機臨時目錄的煙測（見下方「已驗證」）。 +`sync`＝`upload` 再加最後一步：把整輪的 collector-trigger.v1 payload POST 到 +`ARCRUN_TRIGGER_URL`（arcrun named-webhook 完整 URL，如 +`{cypher}/webhooks/named/{ns}/rag_ingest/trigger`——端點是 arcrun 原生觸發機制， +design 鐵律段明言保留；被刪掉的是「Gitea push 事件」這個來源語意）。 -## 做什麼 +- **回寫語意**：HTTP 2xx 才對本輪送出的 added/modified/renamed 檔呼叫 `Manifest.MarkIngested`； + 非 2xx／網路錯不回寫＝下輪自然重試（exit 1）。 +- **上傳失敗的事件不送**：schema 約定 added/modified 的 `r2_key`＝原稿已在 R2，上傳失敗還送 + ＝叫消費端去 404 → 該事件本輪擋下（`dispatch.dropped_paths`），下輪重試補送；同路徑的 + renamed 也不回寫（防「內容從未上 R2 卻被標 ingested」）。 +- **防呆警告輪照送**：mass_delete_guard 觸發時 removed 事件已被壓下，但 payload 連同 + `warnings[]` 照送——消費端看得到警告、不執行下架；notify 呈現歸 collector 端輸出／daemon。 +- **無變更輪不發送**（`skipped_no_changes`）。 +- 輸出 JSON：`{"trigger": <實際送出的 payload>, "uploads": [...], "dispatch": + {status, http_status?, error?, marked_count, dropped_paths?}}`。 -``` -watch 知識資料夾 ──(新增/修改)──► transform（.md passthrough／docx,pptx,pdf→Markitdown）──► 寫入 target repo 工作目錄 - ──(刪除) ──► target repo 移除對應檔 - │ - git add/commit/push - │ - Gitea push webhook（既有機制，非本骨架自建） - │ - arcrun ingest workflow（另案，如 km_wiki_ingest_drain 同款模式） -``` +設定＝upload 的三個環境變數＋`ARCRUN_TRIGGER_URL`（皆絕不落 repo/code）。 -**關鍵設計判斷**：collector 的責任止於「commit + push」。design.md §4 說的「打 ingest webhook」＝ Gitea 自己的 push webhook 機制（同 `Leo/Arcrun` `registry/examples/km-wiki-ingest` 已驗證的模式：Gitea push webhook → arcrun workflow），**不是** collector 自己再打一支 HTTP webhook。collector 不需要知道 ingest 的內部細節（哪個 workflow、哪個 KBDB），它只管「客戶檔案的忠實鏡像」進 Gitea repo。 - -## 事件驅動、不輪詢 - -用 `chokidar`（Node）watch 檔案系統事件（inotify/FSEvents，非 polling），對齊「單一 repo、事件驅動」的鐵律。debounce 視窗（預設 3 秒）把同一批變更合併成一次 commit，避免每個檔案獨立 commit 洗歷史。 - -## 刪檔語意 - -檔案在來源資料夾被刪除 → collector 在 target repo 對應路徑也 `git rm` → commit → push。**deprecated 標記不是 collector 的責任**——ingest workflow 收到 Gitea push event 裡的 `removed` 檔案清單後，自己去 KBDB 把對應 entry 標 `status: deprecated`（append-only，不物理刪，見 km-wiki-ingest description.md 冪等設計）。collector 只管檔案鏡像忠實，不碰 KBDB。 - -## 轉檔（T4） - -`transform.js`：`.md`/`.markdown` passthrough（frontmatter 補 `source_path` 溯源欄位）；`.docx`/`.pptx`/`.pdf` 呼叫 `markitdown` CLI（子行程，需部署機器已 `pip install markitdown[docx,pptx,pdf]`）轉出 md，輸出路徑副檔名換成 `.md`，同時把**原檔**複製進 target repo 的 `assets/originals/<相對路徑>`（design.md §4「原檔進 LFS」）。`git-sync.js` 的 `ensureGitAttributes()` 在啟動時冪等寫入 `.gitattributes`（`assets/originals/**/*.{pdf,docx,pptx} filter=lfs ...`）。 - -⚠️ **LFS 是否真的生效待本機驗**：`.gitattributes` 宣告本身雲端驗過內容正確，但 LFS smudge/clean filter 要部署機器裝了 `git-lfs` 並 `git lfs install` 才真的把大檔案存進 LFS store（否則 git 仍會把二進位檔案當一般 blob 存進版控歷史——功能上檔案還是會進 repo，只是沒享受到 LFS 的空間/頻寬優化）。雲端 sandbox 沒有 `git-lfs` 二進位，這段驗不到。 - -其餘格式（xlsx、圖片等）不在 design.md §4 第一波承諾範圍，仍丟 `NotImplementedError` 並記警告日誌，誠實不假裝轉好。 - -## 檔案 - -| 檔案 | 職責 | -|---|---| -| `index.js` | 進入點：watch＋debounce＋事件分派＋原檔複製 | -| `transform.js` | 檔案 → md 轉換：`.md` passthrough／`docx,pptx,pdf` 走 Markitdown | -| `git-sync.js` | target repo 的 git add/commit/push 封裝＋`.gitattributes` LFS 宣告 | -| `config.js` | 環境變數讀取（`WATCH_DIR`/`TARGET_REPO_DIR`/`DEBOUNCE_MS`） | - -## 設定（環境變數） - -| 變數 | 說明 | 預設 | -|---|---|---| -| `WATCH_DIR` | 客戶知識資料夾（來源） | 必填 | -| `TARGET_REPO_DIR` | 已 clone 好、有 push 權限的 Gitea repo 工作目錄（去向） | 必填 | -| `TARGET_SUBDIR` | 在 target repo 內落地的子目錄 | `collected/` | -| `DEBOUNCE_MS` | 合併變更的等待視窗 | `3000` | -| `GIT_AUTHOR_NAME` / `GIT_AUTHOR_EMAIL` | commit 署名 | `collector` / `collector@localhost` | - -## 部署（客戶端機器，設計稿——本輪未實際跑 systemd） - -```ini -# /etc/systemd/system/arcrun-rag-collector.service（範本，未部署未測） -[Unit] -Description=arcrun-rag collector -After=network.target - -[Service] -Environment=WATCH_DIR=/mnt/knowledge -Environment=TARGET_REPO_DIR=/opt/collector-repo -ExecStart=/usr/bin/node /opt/arcrun-rag/collector/index.js -Restart=always -User=collector - -[Install] -WantedBy=multi-user.target -``` - -`TARGET_REPO_DIR` 需事先 `git clone`＋設好有 push 權限的 remote（credential 用該機器的 git credential helper 或 SSH key，不是本骨架管的事）。 - -## 已驗證（雲端 sandbox 能做到的部分） - -- `node --check` 語法檢查全過。 -- 本機臨時目錄煙測（非真實客戶環境，兩輪）：起一個 scratch git repo 當 target、一個 scratch 資料夾當 watch 來源： - 1. **T3 事件機制**：新增/修改/刪除 `.md` 檔案 → collector 正確偵測、寫入/移除、debounce 後產生一次 commit，`git log` 驗到 commit 內容與變更一致。 - 2. **T4 Markitdown**：用 `python-docx` 生一份真實 `.docx`（含標題+段落）丟進 watch 資料夾 → collector 呼叫 `markitdown` 轉出真實 md 內容（人工核對文字與原檔一致）、原檔複製進 `assets/originals/`、`.gitattributes` 正確寫入三種格式的 LFS 宣告、整批 commit+push 送達 bare remote（`git log` 驗證）。 - - **不含**（待本機/客戶環境）：真實 Gitea remote push（需真實 token+repo）、真實 git-lfs smudge/clean filter（sandbox 無 `git-lfs` 二進位）、systemd 常駐、NAS/VM 環境、Gitea push webhook 是否真觸發 ingest workflow。 - -## 待本機/客戶環境驗（端到端） - -1. 真實客戶知識資料夾 watch（含各類真實 docx/pptx/pdf 樣本，非合成測試檔）。 -2. 真實 Gitea remote push（含憑證管理）。 -3. 真實 git-lfs 安裝＋`git lfs track`，確認大檔案真的走 LFS store 而非塞進一般 blob 歷史。 -4. push 後確認 Gitea push webhook 真觸發 ingest workflow（另案）。 -5. systemd 常駐穩定性（重開機自動起、崩潰自動重啟）。 +測試：`go test ./...` 19/19——掃描 7＋上傳 7＋sync/trigger 5（成功回寫＋無變更輪不重發／ +失敗不回寫＋修復後重試成功／防呆警告輪照送 warnings 零事件／上傳失敗事件擋下＋renamed +連坐不回寫／URL env 驗證），全部 httptest mock（**未實跑雲端 e2e**——等切換日與 leo 一起驗）。 diff --git a/config.js b/config.js deleted file mode 100644 index f11fed7..0000000 --- a/config.js +++ /dev/null @@ -1,20 +0,0 @@ -function requireEnv(name) { - const v = process.env[name]; - if (!v) { - throw new Error(`collector: 缺少環境變數 ${name}（見 README.md 設定表）`); - } - return v; -} - -function loadConfig() { - return { - watchDir: requireEnv('WATCH_DIR'), - targetRepoDir: requireEnv('TARGET_REPO_DIR'), - targetSubdir: process.env.TARGET_SUBDIR || 'collected/', - debounceMs: parseInt(process.env.DEBOUNCE_MS || '3000', 10), - gitAuthorName: process.env.GIT_AUTHOR_NAME || 'collector', - gitAuthorEmail: process.env.GIT_AUTHOR_EMAIL || 'collector@localhost', - }; -} - -module.exports = { loadConfig, requireEnv }; diff --git a/git-sync.js b/git-sync.js deleted file mode 100644 index 1464ab3..0000000 --- a/git-sync.js +++ /dev/null @@ -1,61 +0,0 @@ -const fs = require('fs'); -const path = require('path'); -const { execFileSync } = require('child_process'); - -// design.md §4「原檔進 LFS」：assets/originals/ 底下的常見二進位格式走 git-lfs。 -// ⚠️ 這裡只寫 .gitattributes 宣告（冪等、不重複寫）——LFS 是否真的生效取決於部署機器 -// 有沒有裝 git-lfs 並 `git lfs install`；雲端 sandbox 沒有 git-lfs 二進位，只能驗到 -// .gitattributes 內容正確，驗不到真實 LFS smudge/clean filter，見 README「待本機驗」。 -const LFS_PATTERNS = ['assets/originals/**/*.pdf', 'assets/originals/**/*.docx', 'assets/originals/**/*.pptx']; - -function ensureGitAttributes(repoDir) { - const gaPath = path.join(repoDir, '.gitattributes'); - const existing = fs.existsSync(gaPath) ? fs.readFileSync(gaPath, 'utf8') : ''; - const missing = LFS_PATTERNS.filter((p) => !existing.includes(p)); - if (missing.length === 0) return false; - const lines = missing.map((p) => `${p} filter=lfs diff=lfs merge=lfs -text`); - const next = existing.length && !existing.endsWith('\n') ? `${existing}\n` : existing; - fs.writeFileSync(gaPath, next + lines.join('\n') + '\n', 'utf8'); - return true; -} - -/** - * target repo 的 git add/commit/push 封裝（design.md §4：git commit/push 觸發既有 - * Gitea push webhook → ingest workflow；本模組不打任何 ingest HTTP 端點）。 - */ -class GitSync { - constructor(config) { - this.repoDir = config.targetRepoDir; - this.authorName = config.gitAuthorName; - this.authorEmail = config.gitAuthorEmail; - ensureGitAttributes(this.repoDir); - } - - _git(args) { - return execFileSync('git', args, { - cwd: this.repoDir, - env: { - ...process.env, - GIT_AUTHOR_NAME: this.authorName, - GIT_AUTHOR_EMAIL: this.authorEmail, - GIT_COMMITTER_NAME: this.authorName, - GIT_COMMITTER_EMAIL: this.authorEmail, - }, - encoding: 'utf8', - }); - } - - /** 一批變更合併成一次 commit + push；無變更則不 commit（冪等，避免空 commit 洗歷史）。 */ - commitAndPush(summary) { - this._git(['add', '-A']); - const status = this._git(['status', '--porcelain']); - if (!status.trim()) { - return { committed: false }; - } - this._git(['commit', '-m', summary]); - this._git(['push']); - return { committed: true }; - } -} - -module.exports = { GitSync }; diff --git a/index.js b/index.js deleted file mode 100644 index 5770f1e..0000000 --- a/index.js +++ /dev/null @@ -1,90 +0,0 @@ -const fs = require('fs'); -const path = require('path'); -const chokidar = require('chokidar'); - -const { loadConfig } = require('./config'); -const { transformFile, NotImplementedError } = require('./transform'); -const { GitSync } = require('./git-sync'); - -function ensureDir(p) { - fs.mkdirSync(p, { recursive: true }); -} - -function outputPathFor(config, relOutputPath) { - return path.join(config.targetRepoDir, config.targetSubdir, relOutputPath); -} - -function handleAddOrChange(config, srcPath) { - try { - const { relOutputPath, content, originalCopy } = transformFile(srcPath, config.watchDir); - const outPath = outputPathFor(config, relOutputPath); - ensureDir(path.dirname(outPath)); - fs.writeFileSync(outPath, content, 'utf8'); - console.log(`[collector] 寫入 ${relOutputPath}`); - - if (originalCopy) { - const origOutPath = outputPathFor(config, originalCopy.relPath); - ensureDir(path.dirname(origOutPath)); - fs.copyFileSync(srcPath, origOutPath); - console.log(`[collector] 原檔複製 ${originalCopy.relPath}（LFS 追蹤需部署機器已裝 git-lfs）`); - } - } catch (e) { - if (e instanceof NotImplementedError) { - console.warn(`[collector] 跳過（待 T4）：${e.message}`); - } else { - console.error(`[collector] 轉檔失敗 ${srcPath}：${e.message}`); - } - } -} - -function handleUnlink(config, srcPath) { - const relSrc = path.relative(config.watchDir, srcPath); - const outPath = outputPathFor(config, relSrc); - if (fs.existsSync(outPath)) { - fs.unlinkSync(outPath); - console.log(`[collector] 移除 ${relSrc}（來源已刪除，deprecated 標記交給 ingest workflow）`); - } -} - -function startCollector(config, { git = new GitSync(config) } = {}) { - ensureDir(path.join(config.targetRepoDir, config.targetSubdir)); - - let pendingCount = 0; - let debounceTimer = null; - - const flush = () => { - if (pendingCount === 0) return; - const n = pendingCount; - pendingCount = 0; - const result = git.commitAndPush(`collector: 同步 ${n} 項變更`); - if (result.committed) { - console.log(`[collector] 已 commit+push（${n} 項變更）`); - } - }; - - const schedule = () => { - pendingCount += 1; - if (debounceTimer) clearTimeout(debounceTimer); - debounceTimer = setTimeout(flush, config.debounceMs); - }; - - const watcher = chokidar.watch(config.watchDir, { - ignoreInitial: false, // 首次啟動＝首灌，掃過整個資料夾 - persistent: true, - }); - - watcher - .on('add', (p) => { handleAddOrChange(config, p); schedule(); }) - .on('change', (p) => { handleAddOrChange(config, p); schedule(); }) - .on('unlink', (p) => { handleUnlink(config, p); schedule(); }); - - console.log(`[collector] watch 中：${config.watchDir} → ${config.targetRepoDir}/${config.targetSubdir}`); - return watcher; -} - -if (require.main === module) { - const config = loadConfig(); - startCollector(config); -} - -module.exports = { startCollector }; diff --git a/main.go b/main.go index 2cc8088..8476478 100644 --- a/main.go +++ b/main.go @@ -6,6 +6,8 @@ // [--max-removed-ratio 0.4] [--dry-run] // collector upload --root <知識資料夾> --manifest \ // [--max-removed-ratio 0.4] [--dry-run] +// collector sync --root <知識資料夾> --manifest \ +// [--max-removed-ratio 0.4] [--dry-run] // // scan：走訪 root、對照 manifest、把事件（符合 schemas/collector-trigger.v1.schema.json） // 以 JSON 輸出到 stdout，並更新 manifest（--dry-run 不寫）。 @@ -15,7 +17,12 @@ // R2_BUCKET（絕不落 repo）；--dry-run 只列出會上傳的 key（planned），不碰網路不寫 manifest。 // 任一上傳失敗＝exit 1（manifest 照存：content_hash 反映現況、ingested_hash 不動＝可重試）。 // -// daemon 常駐（launchd）/打 named-webhook 觸發 ingest＝之後的 task（SDD task 4、產品化段）。 +// sync（task 4）＝scan＋upload＋把整輪 payload POST 到 ARCRUN_TRIGGER_URL（arcrun +// named-webhook 觸發 ingest cypher workflow）。HTTP 2xx 才對本輪送出的 added/modified/ +// renamed 事件回寫 Manifest.MarkIngested；失敗不回寫（下輪自然重試）。上傳失敗的事件 +// 不隨 payload 送出（r2_key 語意＝原稿已在 R2）。防呆警告輪照送 warnings、不含下架事件。 +// +// daemon 常駐（launchd）＝之後的 task（產品化段 task 11）。 package main import ( @@ -24,8 +31,14 @@ import ( "fmt" "os" "path/filepath" + "time" ) +type runMode struct { + withUpload bool + withTrigger bool +} + func main() { if len(os.Args) < 2 { usage() @@ -33,9 +46,11 @@ func main() { } switch os.Args[1] { case "scan": - os.Exit(run(os.Args[2:], false)) + os.Exit(run(os.Args[2:], runMode{})) case "upload": - os.Exit(run(os.Args[2:], true)) + os.Exit(run(os.Args[2:], runMode{withUpload: true})) + case "sync": + os.Exit(run(os.Args[2:], runMode{withUpload: true, withTrigger: true})) default: usage() os.Exit(2) @@ -46,16 +61,18 @@ func usage() { fmt.Fprintln(os.Stderr, `用法: collector scan --root --manifest [--max-removed-ratio 0.4] [--dry-run] collector upload --root --manifest [--max-removed-ratio 0.4] [--dry-run] -upload 需環境變數: CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET`) + collector sync --root --manifest [--max-removed-ratio 0.4] [--dry-run] +upload 需環境變數: CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET +sync 另需: ARCRUN_TRIGGER_URL（named-webhook 觸發完整 URL）`) } -// run 是 scan/upload 共用主體；withUpload＝true 時掃描後把 added/modified 原稿上傳 R2。 -func run(args []string, withUpload bool) int { +// run 是 scan/upload/sync 共用主體。 +func run(args []string, mode runMode) int { fs := flag.NewFlagSet("collector", flag.ExitOnError) root := fs.String("root", "", "知識資料夾根路徑（必填）") manifestPath := fs.String("manifest", "", "manifest JSON 檔路徑（必填；不存在會建新）") ratio := fs.Float64("max-removed-ratio", DefaultMaxRemovedRatio, "大量刪除防呆門檻（removed 數 > manifest 條目 × 本值 → 全部不下架、只發警告）") - dryRun := fs.Bool("dry-run", false, "只輸出事件（upload 模式另列 planned 上傳清單），不更新 manifest、不碰網路") + dryRun := fs.Bool("dry-run", false, "只輸出事件（upload/sync 模式另列 planned 清單），不更新 manifest、不碰網路") if err := fs.Parse(args); err != nil { return 2 } @@ -64,15 +81,26 @@ func run(args []string, withUpload bool) int { return 2 } - // upload 模式先驗設定（fail fast：缺 env 連掃都不掃，不留半套狀態）。 + // 先驗設定（fail fast：缺 env 連掃都不掃，不留半套狀態）。 var client *R2Client - if withUpload && !*dryRun { - cfg, err := LoadR2ConfigFromEnv() - if err != nil { - fmt.Fprintln(os.Stderr, "collector:", err) - return 2 + var triggerURL string + if !*dryRun { + if mode.withUpload { + cfg, err := LoadR2ConfigFromEnv() + if err != nil { + fmt.Fprintln(os.Stderr, "collector:", err) + return 2 + } + client = NewR2Client(cfg) + } + if mode.withTrigger { + u, err := LoadTriggerURLFromEnv() + if err != nil { + fmt.Fprintln(os.Stderr, "collector:", err) + return 2 + } + triggerURL = u } - client = NewR2Client(cfg) } absRoot, err := filepath.Abs(*root) @@ -96,8 +124,10 @@ func run(args []string, withUpload bool) int { return fail(err) } + exitCode := 0 + var uploads []UploadResult - if withUpload { + if mode.withUpload { if *dryRun { uploads = []UploadResult{} for _, ev := range payload.Events { @@ -107,6 +137,38 @@ func run(args []string, withUpload bool) int { } } else { uploads = UploadChanged(absRoot, payload.Events, client) + for _, r := range uploads { + if r.Status == "failed" { + exitCode = 1 // 有敗＝非零退出；ingested_hash 未動＝下輪自然重試 + } + } + } + } + + // sync：POST 觸發 → 2xx 才回寫 MarkIngested（在 m.Save 之前，回寫才進得了檔）。 + outPayload := payload + var dispatch *TriggerResult + if mode.withTrigger { + if *dryRun { + dispatch = &TriggerResult{Status: "planned"} + } else { + sendable, dropped := BuildSendablePayload(payload, uploads) + outPayload = sendable + dispatch = &TriggerResult{DroppedPaths: dropped} + if len(sendable.Events) == 0 && len(sendable.Warnings) == 0 { + dispatch.Status = "skipped_no_changes" // 無變更輪不發送（schema 註明空發也合法，但沒必要） + } else { + status, terr := SendTrigger(triggerURL, sendable, nil) + dispatch.HTTPStatus = status + if terr != nil { + dispatch.Status = "failed" + dispatch.Error = terr.Error() + exitCode = 1 + } else { + dispatch.Status = "sent" + dispatch.MarkedCount = MarkIngestedEvents(m, sendable.Events, dropped, time.Now().Unix()) + } + } } } @@ -116,18 +178,18 @@ func run(args []string, withUpload bool) int { } } - exitCode := 0 - var out any = payload - if withUpload { + var out any = outPayload + if mode.withTrigger { + out = struct { + Trigger *TriggerPayload `json:"trigger"` + Uploads []UploadResult `json:"uploads"` + Dispatch *TriggerResult `json:"dispatch"` + }{outPayload, uploads, dispatch} + } else if mode.withUpload { out = struct { Trigger *TriggerPayload `json:"trigger"` Uploads []UploadResult `json:"uploads"` - }{payload, uploads} - for _, r := range uploads { - if r.Status == "failed" { - exitCode = 1 // 有敗＝非零退出；manifest 已存、ingested_hash 未動＝下輪自然重試 - } - } + }{outPayload, uploads} } data, err := json.MarshalIndent(out, "", " ") if err != nil { diff --git a/package-lock.json b/package-lock.json deleted file mode 100644 index 8f1ea3a..0000000 --- a/package-lock.json +++ /dev/null @@ -1,201 +0,0 @@ -{ - "name": "arcrun-rag-collector", - "version": "0.1.0", - "lockfileVersion": 3, - "requires": true, - "packages": { - "": { - "name": "arcrun-rag-collector", - "version": "0.1.0", - "dependencies": { - "chokidar": "^3.6.0" - } - }, - "node_modules/anymatch": { - "version": "3.1.3", - "resolved": "https://registry.npmjs.org/anymatch/-/anymatch-3.1.3.tgz", - "integrity": "sha512-KMReFUr0B4t+D+OBkjR3KYqvocp2XaSzO55UcB6mgQMd3KbcE+mWTyvVV7D/zsdEbNnV6acZUutkiHQXvTr1Rw==", - "license": "ISC", - "dependencies": { - "normalize-path": "^3.0.0", - "picomatch": "^2.0.4" - }, - "engines": { - "node": ">= 8" - } - }, - "node_modules/binary-extensions": { - "version": "2.3.0", - "resolved": "https://registry.npmjs.org/binary-extensions/-/binary-extensions-2.3.0.tgz", - "integrity": "sha512-Ceh+7ox5qe7LJuLHoY0feh3pHuUDHAcRUeyL2VYghZwfpkNIy/+8Ocg0a3UuSoYzavmylwuLWQOf3hl0jjMMIw==", - "license": "MIT", - "engines": { - "node": ">=8" - }, - "funding": { - "url": "https://github.com/sponsors/sindresorhus" - } - }, - "node_modules/braces": { - "version": "3.0.3", - "resolved": "https://registry.npmjs.org/braces/-/braces-3.0.3.tgz", - "integrity": "sha512-yQbXgO/OSZVD2IsiLlro+7Hf6Q18EJrKSEsdoMzKePKXct3gvD8oLcOQdIzGupr5Fj+EDe8gO/lxc1BzfMpxvA==", - "license": "MIT", - "dependencies": { - "fill-range": "^7.1.1" - }, - "engines": { - "node": ">=8" - } - }, - "node_modules/chokidar": { - "version": "3.6.0", - "resolved": "https://registry.npmjs.org/chokidar/-/chokidar-3.6.0.tgz", - "integrity": "sha512-7VT13fmjotKpGipCW9JEQAusEPE+Ei8nl6/g4FBAmIm0GOOLMua9NDDo/DWp0ZAxCr3cPq5ZpBqmPAQgDda2Pw==", - "license": "MIT", - "dependencies": { - "anymatch": "~3.1.2", - "braces": "~3.0.2", - "glob-parent": "~5.1.2", - "is-binary-path": "~2.1.0", - "is-glob": "~4.0.1", - "normalize-path": "~3.0.0", - "readdirp": "~3.6.0" - }, - "engines": { - "node": ">= 8.10.0" - }, - "funding": { - "url": "https://paulmillr.com/funding/" - }, - "optionalDependencies": { - "fsevents": "~2.3.2" - } - }, - "node_modules/fill-range": { - "version": "7.1.1", - "resolved": "https://registry.npmjs.org/fill-range/-/fill-range-7.1.1.tgz", - "integrity": "sha512-YsGpe3WHLK8ZYi4tWDg2Jy3ebRz2rXowDxnld4bkQB00cc/1Zw9AWnC0i9ztDJitivtQvaI9KaLyKrc+hBW0yg==", - "license": "MIT", - "dependencies": { - "to-regex-range": "^5.0.1" - }, - "engines": { - "node": ">=8" - } - }, - "node_modules/fsevents": { - "version": "2.3.3", - "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", - "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", - "hasInstallScript": true, - "license": "MIT", - "optional": true, - "os": [ - "darwin" - ], - "engines": { - "node": "^8.16.0 || ^10.6.0 || >=11.0.0" - } - }, - "node_modules/glob-parent": { - "version": "5.1.2", - "resolved": "https://registry.npmjs.org/glob-parent/-/glob-parent-5.1.2.tgz", - "integrity": "sha512-AOIgSQCepiJYwP3ARnGx+5VnTu2HBYdzbGP45eLw1vr3zB3vZLeyed1sC9hnbcOc9/SrMyM5RPQrkGz4aS9Zow==", - "license": "ISC", - "dependencies": { - "is-glob": "^4.0.1" - }, - "engines": { - "node": ">= 6" - } - }, - "node_modules/is-binary-path": { - "version": "2.1.0", - "resolved": "https://registry.npmjs.org/is-binary-path/-/is-binary-path-2.1.0.tgz", - "integrity": "sha512-ZMERYes6pDydyuGidse7OsHxtbI7WVeUEozgR/g7rd0xUimYNlvZRE/K2MgZTjWy725IfelLeVcEM97mmtRGXw==", - "license": "MIT", - "dependencies": { - "binary-extensions": "^2.0.0" - }, - "engines": { - "node": ">=8" - } - }, - "node_modules/is-extglob": { - "version": "2.1.1", - "resolved": "https://registry.npmjs.org/is-extglob/-/is-extglob-2.1.1.tgz", - "integrity": "sha512-SbKbANkN603Vi4jEZv49LeVJMn4yGwsbzZworEoyEiutsN3nJYdbO36zfhGJ6QEDpOZIFkDtnq5JRxmvl3jsoQ==", - "license": "MIT", - "engines": { - "node": ">=0.10.0" - } - }, - "node_modules/is-glob": { - "version": "4.0.3", - "resolved": "https://registry.npmjs.org/is-glob/-/is-glob-4.0.3.tgz", - "integrity": "sha512-xelSayHH36ZgE7ZWhli7pW34hNbNl8Ojv5KVmkJD4hBdD3th8Tfk9vYasLM+mXWOZhFkgZfxhLSnrwRr4elSSg==", - "license": "MIT", - "dependencies": { - "is-extglob": "^2.1.1" - }, - "engines": { - "node": ">=0.10.0" - } - }, - "node_modules/is-number": { - "version": "7.0.0", - "resolved": "https://registry.npmjs.org/is-number/-/is-number-7.0.0.tgz", - "integrity": "sha512-41Cifkg6e8TylSpdtTpeLVMqvSBEVzTttHvERD741+pnZ8ANv0004MRL43QKPDlK9cGvNp6NZWZUBlbGXYxxng==", - "license": "MIT", - "engines": { - "node": ">=0.12.0" - } - }, - "node_modules/normalize-path": { - "version": "3.0.0", - "resolved": "https://registry.npmjs.org/normalize-path/-/normalize-path-3.0.0.tgz", - "integrity": "sha512-6eZs5Ls3WtCisHWp9S2GUy8dqkpGi4BVSz3GaqiE6ezub0512ESztXUwUB6C6IKbQkY2Pnb/mD4WYojCRwcwLA==", - "license": "MIT", - "engines": { - "node": ">=0.10.0" - } - }, - "node_modules/picomatch": { - "version": "2.3.2", - "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-2.3.2.tgz", - "integrity": "sha512-V7+vQEJ06Z+c5tSye8S+nHUfI51xoXIXjHQ99cQtKUkQqqO1kO/KCJUfZXuB47h/YBlDhah2H3hdUGXn8ie0oA==", - "license": "MIT", - "engines": { - "node": ">=8.6" - }, - "funding": { - "url": "https://github.com/sponsors/jonschlinkert" - } - }, - "node_modules/readdirp": { - "version": "3.6.0", - "resolved": "https://registry.npmjs.org/readdirp/-/readdirp-3.6.0.tgz", - "integrity": "sha512-hOS089on8RduqdbhvQ5Z37A0ESjsqz6qnRcffsMU3495FuTdqSm+7bhJ29JvIOsBDEEnan5DPu9t3To9VRlMzA==", - "license": "MIT", - "dependencies": { - "picomatch": "^2.2.1" - }, - "engines": { - "node": ">=8.10.0" - } - }, - "node_modules/to-regex-range": { - "version": "5.0.1", - "resolved": "https://registry.npmjs.org/to-regex-range/-/to-regex-range-5.0.1.tgz", - "integrity": "sha512-65P7iz6X5yEr1cwcgvQxbbIw7Uk3gOy5dIdtZ4rDveLqhrdJP+Li/Hx6tyK0NEb+2GCyneCMJiGqrADCSNk8sQ==", - "license": "MIT", - "dependencies": { - "is-number": "^7.0.0" - }, - "engines": { - "node": ">=8.0" - } - } - } -} diff --git a/package.json b/package.json deleted file mode 100644 index 2b5cd61..0000000 --- a/package.json +++ /dev/null @@ -1,10 +0,0 @@ -{ - "name": "arcrun-rag-collector", - "version": "0.1.0", - "description": "收集端骨架（rag-wave1 T3）：watch 客戶知識資料夾 → 轉檔 → git commit/push", - "main": "index.js", - "private": true, - "dependencies": { - "chokidar": "^3.6.0" - } -} diff --git a/transform.js b/transform.js deleted file mode 100644 index d47d22a..0000000 --- a/transform.js +++ /dev/null @@ -1,75 +0,0 @@ -const fs = require('fs'); -const path = require('path'); -const { execFileSync } = require('child_process'); - -class NotImplementedError extends Error {} - -const MARKDOWN_EXT = new Set(['.md', '.markdown']); - -// design.md §4 明列的第一波轉檔格式：docx/pptx/pdf。其餘格式仍走 NotImplementedError， -// 不擴大承諾範圍（品質不承諾是 design 原話，但格式範圍先照 SDD 講定的三種）。 -const MARKITDOWN_EXT = new Set(['.docx', '.pptx', '.pdf']); - -/** - * 檔案 → md 轉換（T4：Markitdown adapter）。 - * srcPath: 來源檔案絕對路徑（客戶知識資料夾內） - * 回傳：{ relOutputPath, content, originalCopy? } - * - originalCopy（僅 Markitdown 轉檔路徑）：{ relPath }，供 index.js 把原檔複製進 - * target repo 的 assets/originals/（design.md §4：「原檔進 LFS」——LFS 追蹤本身 - * 依賴部署機器裝 git-lfs 並 `git lfs track`，本模組只管把原檔放進約定路徑， - * 不假裝在雲端 sandbox 驗過真實 LFS push，見 README「待本機驗」）。 - */ -function transformFile(srcPath, watchDir) { - const ext = path.extname(srcPath).toLowerCase(); - const relSrc = path.relative(watchDir, srcPath); - - if (MARKDOWN_EXT.has(ext)) { - const raw = fs.readFileSync(srcPath, 'utf8'); - const content = stampSourcePath(raw, relSrc); - return { relOutputPath: relSrc, content }; - } - - if (MARKITDOWN_EXT.has(ext)) { - const md = convertWithMarkitdown(srcPath); - const content = stampSourcePath(md, relSrc); - const relOutputPath = relSrc.slice(0, -ext.length) + '.md'; - return { - relOutputPath, - content, - originalCopy: { relPath: path.join('assets/originals', relSrc) }, - }; - } - - // 其餘格式（xlsx/圖片/...）— 不在 design.md §4 第一波承諾範圍，誠實丟未實作。 - throw new NotImplementedError( - `${relSrc}：格式 ${ext} 不在第一波 Markitdown 承諾範圍（docx/pptx/pdf），本骨架先跳過`, - ); -} - -/** 呼叫 markitdown CLI 轉檔（品質不承諾，design.md §4 原話）。 */ -function convertWithMarkitdown(srcPath) { - try { - return execFileSync('markitdown', [srcPath], { - encoding: 'utf8', - maxBuffer: 20 * 1024 * 1024, - }); - } catch (e) { - throw new Error(`markitdown 轉檔失敗（${srcPath}）：${e.message}`); - } -} - -/** md frontmatter 補 source_path（溯源用，design.md §4：「md frontmatter 記原檔路徑」）。 */ -function stampSourcePath(raw, relSrc) { - const stamp = `source_path: "${relSrc}"`; - if (raw.startsWith('---\n')) { - const end = raw.indexOf('\n---', 4); - if (end !== -1) { - const fm = raw.slice(4, end); - if (fm.includes('source_path:')) return raw; // 已有就不重複塞 - return `---\n${stamp}\n${fm}\n---${raw.slice(end + 4)}`; - } - } - return `---\n${stamp}\n---\n\n${raw}`; -} - -module.exports = { transformFile, NotImplementedError }; diff --git a/trigger.go b/trigger.go new file mode 100644 index 0000000..6c48005 --- /dev/null +++ b/trigger.go @@ -0,0 +1,128 @@ +// trigger.go — 把一輪掃描的 collector-trigger.v1 payload POST 到 arcrun named-webhook +// （SDD ingest-hash-trigger task 4：觸發鏈從 Gitea push webhook 改為 collector 直打）。 +// +// 目標端點＝arcrun 原生 named-webhook 觸發機制（design 鐵律段明言保留）： +// +// POST {cypher}/webhooks/named/{ns}/rag_ingest/trigger +// +// 完整 URL 走環境變數 ARCRUN_TRIGGER_URL（絕不落 repo）。語意： +// - HTTP 2xx ＝本輪觸發成功 → 對「實際送出」的 added/modified/renamed 事件回寫 +// Manifest.MarkIngested（design §2 的回寫鉤子，至此才第一次被呼叫）。 +// - 非 2xx／網路錯 ＝不回寫（ingested_hash 不動）→ 下輪掃描自然重發＝重試， +// R2 端靠存在檢查 no-op、ingest 端靠 source_hash 冪等（design §5），無腦重試安全。 +// - 上傳失敗的 added/modified 事件「不」隨 payload 送出（schema 約定 r2_key＝原稿已在 +// R2；沒上去就送＝消費端 fetch 必 404）——下輪重試補送。renamed/removed 不依賴 R2，照送。 +// - 防呆警告輪（mass_delete_guard）：removed 事件已被 collector 壓下，但 payload 連同 +// warnings 照送（消費端／執行紀錄看得到警告，不執行下架）。 +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" +) + +// LoadTriggerURLFromEnv 讀 ARCRUN_TRIGGER_URL（sync 模式必填）。 +func LoadTriggerURLFromEnv() (string, error) { + u := os.Getenv("ARCRUN_TRIGGER_URL") + if u == "" { + return "", fmt.Errorf("sync 缺環境變數：ARCRUN_TRIGGER_URL（named-webhook 觸發完整 URL，設定只走環境變數，絕不寫進 repo/code）") + } + if !strings.HasPrefix(u, "http://") && !strings.HasPrefix(u, "https://") { + return "", fmt.Errorf("ARCRUN_TRIGGER_URL 必須是完整 URL（http/https），得到：%s", u) + } + return u, nil +} + +// TriggerResult 是本輪觸發的結果，隨 sync 輸出 stdout。 +type TriggerResult struct { + Status string `json:"status"` // sent | skipped_no_changes | failed | planned（--dry-run） + HTTPStatus int `json:"http_status,omitempty"` + Error string `json:"error,omitempty"` + MarkedCount int `json:"marked_count"` // 本輪成功回寫 ingested_hash 的檔數 + DroppedPaths []string `json:"dropped_paths,omitempty"` // 因上傳失敗被擋下、未隨 payload 送出的事件路徑 +} + +// BuildSendablePayload 依上傳結果過濾掃描 payload： +// added/modified 只有上傳成功（uploaded/skipped_exists）才隨 payload 送出—— +// schema 的 r2_key 語意＝「原稿已在 R2」，上傳失敗還送＝叫消費端去 404。 +// renamed/removed 不依賴 R2 物件，一律保留；warnings 原樣保留（防呆輪照送）。 +// 回傳（可送出的 payload 副本, 被擋下的路徑清單）。 +func BuildSendablePayload(p *TriggerPayload, uploads []UploadResult) (*TriggerPayload, []string) { + okPaths := map[string]bool{} + for _, u := range uploads { + if u.Status == "uploaded" || u.Status == "skipped_exists" { + okPaths[u.Path] = true + } + } + sendable := *p + sendable.Events = []Event{} + var dropped []string + for _, ev := range p.Events { + if ev.Type == "added" || ev.Type == "modified" { + if !okPaths[ev.Path] { + dropped = append(dropped, ev.Path) + continue + } + } + sendable.Events = append(sendable.Events, ev) + } + return &sendable, dropped +} + +// SendTrigger 把 payload POST 到 named-webhook。回傳 HTTP 狀態碼；非 2xx 視為錯誤。 +// timeout 放寬到 300s：named-webhook 觸發的 ingest workflow 可能同步跑（demo 實測 20-30s+）。 +func SendTrigger(url string, p *TriggerPayload, hc *http.Client) (int, error) { + if hc == nil { + hc = &http.Client{Timeout: 300 * time.Second} + } + body, err := json.Marshal(p) + if err != nil { + return 0, err + } + req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return 0, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := hc.Do(req) + if err != nil { + return 0, fmt.Errorf("觸發 POST 失敗（不回寫 manifest，下輪自然重試）：%w", err) + } + defer resp.Body.Close() + snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return resp.StatusCode, fmt.Errorf("觸發回 HTTP %d（不回寫 manifest，下輪自然重試）：%s", + resp.StatusCode, strings.TrimSpace(string(snippet))) + } + return resp.StatusCode, nil +} + +// MarkIngestedEvents 在觸發成功（2xx）後，對「實際送出」的 added/modified/renamed 事件 +// 回寫 ingested_hash。droppedPaths＝本輪因上傳失敗被擋下的路徑——同路徑若另有 renamed +// 事件（改名＋內容從未成功 ingest 的檔會同輪補發 added），該 renamed 也不得回寫， +// 否則原稿永遠上不了 R2 卻被標成已 ingest。回傳成功回寫的檔數。 +func MarkIngestedEvents(m *Manifest, sentEvents []Event, droppedPaths []string, at int64) int { + droppedSet := map[string]bool{} + for _, p := range droppedPaths { + droppedSet[p] = true + } + n := 0 + for _, ev := range sentEvents { + switch ev.Type { + case "added", "modified", "renamed": + if droppedSet[ev.Path] { + continue + } + if m.MarkIngested(ev.Path, ev.SourceHash, at) { + n++ + } + } + } + return n +} diff --git a/trigger_test.go b/trigger_test.go new file mode 100644 index 0000000..29da073 --- /dev/null +++ b/trigger_test.go @@ -0,0 +1,251 @@ +package main + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "testing" + "time" +) + +// mockTrigger 用 httptest 模擬 arcrun named-webhook 觸發端點。 +type mockTrigger struct { + mu sync.Mutex + status int // 回應狀態碼（預設 200） + payloads []*TriggerPayload +} + +func newMockTrigger(t *testing.T) (*httptest.Server, *mockTrigger) { + t.Helper() + m := &mockTrigger{status: http.StatusOK} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + if ct := r.Header.Get("Content-Type"); !strings.HasPrefix(ct, "application/json") { + t.Errorf("Content-Type 應為 application/json，得到 %q", ct) + } + body, err := io.ReadAll(r.Body) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + var p TriggerPayload + if err := json.Unmarshal(body, &p); err != nil { + t.Errorf("payload 不是合法 JSON: %v", err) + } + m.mu.Lock() + m.payloads = append(m.payloads, &p) + st := m.status + m.mu.Unlock() + w.WriteHeader(st) + if st >= 500 { + io.WriteString(w, `{"error":"mock 觸發失敗"}`) + } else { + io.WriteString(w, `{"success":true,"execution_id":"mock-exec"}`) + } + })) + t.Cleanup(srv.Close) + return srv, m +} + +// syncRound 模擬 sync 的一輪：scan → upload（真 mock R2）→ 過濾 → POST → 2xx 才回寫。 +// 與 main.run 的 sync 分支同邏輯（單測不跑 CLI 殼，直接組核心函式）。 +func syncRound(t *testing.T, root string, m *Manifest, r2 *R2Client, url string) (*TriggerPayload, []UploadResult, *TriggerResult) { + t.Helper() + p := mustScan(t, root, m) + uploads := UploadChanged(root, p.Events, r2) + sendable, dropped := BuildSendablePayload(p, uploads) + res := &TriggerResult{DroppedPaths: dropped} + if len(sendable.Events) == 0 && len(sendable.Warnings) == 0 { + res.Status = "skipped_no_changes" + return sendable, uploads, res + } + status, err := SendTrigger(url, sendable, nil) + res.HTTPStatus = status + if err != nil { + res.Status = "failed" + res.Error = err.Error() + return sendable, uploads, res + } + res.Status = "sent" + res.MarkedCount = MarkIngestedEvents(m, sendable.Events, dropped, time.Now().Unix()) + return sendable, uploads, res +} + +// 情境 1：成功回寫——上傳全成、觸發 2xx → added 事件的檔回寫 ingested_hash， +// 下一輪掃描歸零（不重發）；伺服器收到的 payload 帶 schema_version/folder_id/r2_key。 +func TestSyncSuccessMarksIngested(t *testing.T) { + _, _, r2 := newMockR2(t) + srv, trig := newMockTrigger(t) + root := t.TempDir() + content := "sync 測試內容\n" + writeFile(t, root, "a.md", content, baseTime) + m := newTestManifest() + + _, uploads, res := syncRound(t, root, m, r2, srv.URL) + + if len(uploads) != 1 || uploads[0].Status != "uploaded" { + t.Fatalf("要 1 筆 uploaded: %+v", uploads) + } + if res.Status != "sent" || res.MarkedCount != 1 || len(res.DroppedPaths) != 0 { + t.Fatalf("要 sent＋marked 1: %+v", res) + } + e := m.Entries["a.md"] + if e.IngestedHash != hashOf(content) || e.IngestedAt == 0 { + t.Fatalf("觸發 2xx 後應回寫 ingested_hash: %+v", e) + } + // 伺服器收到的 payload 符合約定 + if len(trig.payloads) != 1 { + t.Fatalf("要恰好 1 發觸發: %d", len(trig.payloads)) + } + got := trig.payloads[0] + if got.SchemaVersion != 1 || got.FolderID == "" { + t.Fatalf("payload 頭欄位不對: %+v", got) + } + if len(got.Events) != 1 || got.Events[0].Type != "added" || got.Events[0].R2Key == "" { + t.Fatalf("要 1 筆帶 r2_key 的 added: %+v", got.Events) + } + // 下一輪：無變更＝零事件、不再觸發 + _, _, res2 := syncRound(t, root, m, r2, srv.URL) + if res2.Status != "skipped_no_changes" || len(trig.payloads) != 1 { + t.Fatalf("無變更輪不該再觸發: %+v（觸發數 %d）", res2, len(trig.payloads)) + } +} + +// 情境 2：觸發失敗（500）——不回寫 ingested_hash，下一輪同檔重發＝自然重試； +// 修好後（200）重試成功才回寫。 +func TestSyncFailureNoMarkThenRetry(t *testing.T) { + _, _, r2 := newMockR2(t) + srv, trig := newMockTrigger(t) + trig.status = http.StatusInternalServerError + root := t.TempDir() + content := "會先失敗的內容\n" + writeFile(t, root, "b.md", content, baseTime) + m := newTestManifest() + + _, _, res := syncRound(t, root, m, r2, srv.URL) + if res.Status != "failed" || res.Error == "" || res.HTTPStatus != 500 { + t.Fatalf("要 failed＋錯誤訊息: %+v", res) + } + if m.Entries["b.md"].IngestedHash != "" { + t.Fatal("觸發失敗不得回寫 ingested_hash") + } + + // 修好 → 下一輪重發（R2 端 skipped_exists no-op）→ 2xx → 回寫 + trig.status = http.StatusOK + sendable, uploads, res2 := syncRound(t, root, m, r2, srv.URL) + if res2.Status != "sent" || res2.MarkedCount != 1 { + t.Fatalf("重試輪應 sent＋marked 1: %+v", res2) + } + if len(uploads) != 1 || uploads[0].Status != "skipped_exists" { + t.Fatalf("重試輪 R2 應 no-op: %+v", uploads) + } + if len(sendable.Events) != 1 || sendable.Events[0].Type != "added" { + t.Fatalf("重試輪應重發 added: %+v", sendable.Events) + } + if m.Entries["b.md"].IngestedHash != hashOf(content) { + t.Fatal("重試成功後應回寫 ingested_hash") + } +} + +// 情境 3：防呆警告輪（mass_delete_guard）——removed 事件被壓下，但 payload 連同 +// warnings 照送（消費端看得到警告、不執行下架）；沒有任何回寫。 +func TestSyncGuardRoundStillSendsWarnings(t *testing.T) { + _, _, r2 := newMockR2(t) + srv, trig := newMockTrigger(t) + root := t.TempDir() + m := newTestManifest() + // 先建 3 檔並完成一輪成功 sync（全部標 ingested） + for _, n := range []string{"a.md", "b.md", "c.md"} { + writeFile(t, root, n, n+" content\n", baseTime) + } + if _, _, res := syncRound(t, root, m, r2, srv.URL); res.Status != "sent" || res.MarkedCount != 3 { + t.Fatalf("前置輪應全成: %+v", res) + } + // 刪 2/3（67% > 40% 門檻）→ 防呆 + os.Remove(root + "/a.md") + os.Remove(root + "/b.md") + + sendable, _, res := syncRound(t, root, m, r2, srv.URL) + if res.Status != "sent" { + t.Fatalf("警告輪應照送: %+v", res) + } + if len(sendable.Events) != 0 { + t.Fatalf("防呆輪不得夾帶任何事件（removed 已壓下）: %+v", sendable.Events) + } + if len(sendable.Warnings) != 1 || sendable.Warnings[0].Code != "mass_delete_guard" { + t.Fatalf("要 mass_delete_guard 警告: %+v", sendable.Warnings) + } + got := trig.payloads[len(trig.payloads)-1] + if len(got.Warnings) != 1 || got.Warnings[0].Code != "mass_delete_guard" || len(got.Events) != 0 { + t.Fatalf("伺服器端收到的警告輪不對: %+v", got) + } + if res.MarkedCount != 0 { + t.Fatalf("警告輪無內容事件，不該回寫: %+v", res) + } +} + +// 情境 4（純函式）：上傳失敗的 added/modified 不隨 payload 送出；同路徑的 renamed +// 也不得回寫（否則內容從未上 R2 卻被標 ingested）。 +func TestBuildSendableDropsFailedUploads(t *testing.T) { + h1, h2 := hashOf("one"), hashOf("two") + p := &TriggerPayload{ + SchemaVersion: 1, FolderID: "f", + Events: []Event{ + {Type: "added", Path: "ok.md", SourceHash: h1, R2Key: r2KeyOf(h1)}, + {Type: "added", Path: "bad.md", SourceHash: h2, R2Key: r2KeyOf(h2)}, + {Type: "renamed", Path: "bad.md", OldPath: "old-bad.md", SourceHash: h2}, + {Type: "removed", Path: "gone.md", SourceHash: hashOf("g")}, + }, + } + uploads := []UploadResult{ + {Path: "ok.md", Status: "uploaded"}, + {Path: "bad.md", Status: "failed", Error: "mock"}, + } + sendable, dropped := BuildSendablePayload(p, uploads) + if len(dropped) != 1 || dropped[0] != "bad.md" { + t.Fatalf("要擋下 bad.md: %v", dropped) + } + types := []string{} + for _, ev := range sendable.Events { + types = append(types, ev.Type+":"+ev.Path) + } + want := "added:ok.md renamed:bad.md removed:gone.md" + if strings.Join(types, " ") != want { + t.Fatalf("送出清單不對：%v（要 %s）", types, want) + } + // 回寫：ok.md 回寫；bad.md 的 renamed 因同路徑被擋下也不回寫 + m := newTestManifest() + m.Entries["ok.md"] = &ManifestEntry{ContentHash: h1} + m.Entries["bad.md"] = &ManifestEntry{ContentHash: h2} + n := MarkIngestedEvents(m, sendable.Events, dropped, 42) + if n != 1 || m.Entries["ok.md"].IngestedHash != h1 { + t.Fatalf("只該回寫 ok.md: n=%d %+v", n, m.Entries["ok.md"]) + } + if m.Entries["bad.md"].IngestedHash != "" { + t.Fatal("上傳失敗檔的 renamed 不得回寫 ingested_hash") + } +} + +// 附加：ARCRUN_TRIGGER_URL 缺漏／格式錯，清楚報錯。 +func TestLoadTriggerURL(t *testing.T) { + t.Setenv("ARCRUN_TRIGGER_URL", "") + if _, err := LoadTriggerURLFromEnv(); err == nil || !strings.Contains(err.Error(), "ARCRUN_TRIGGER_URL") { + t.Fatalf("缺 env 應報含變數名的錯: %v", err) + } + t.Setenv("ARCRUN_TRIGGER_URL", "not-a-url") + if _, err := LoadTriggerURLFromEnv(); err == nil { + t.Fatal("非完整 URL 應報錯") + } + t.Setenv("ARCRUN_TRIGGER_URL", "https://example.com/webhooks/named/demo/rag_ingest/trigger") + u, err := LoadTriggerURLFromEnv() + if err != nil || u == "" { + t.Fatalf("合法 URL 應通過: %v", err) + } +}