From fdb7a6748451a8c79d32c933f712babc2f14599c Mon Sep 17 00:00:00 2001 From: richblack Date: Sun, 19 Jul 2026 21:27:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(ingest-hash-trigger):=20collector=20R2=20c?= =?UTF-8?q?ontent-addressed=20=E5=8E=9F=E7=A8=BF=E4=B8=8A=E5=82=B3?= =?UTF-8?q?=EF=BC=88SDD=20task=203=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - collector upload 子命令＝scan＋把 added/modified 原稿傳 R2（CF REST API Bearer token，key=raw/，對齊 collector-trigger.v1 的 r2_key） - 冪等：存在檢查命中＝skipped_exists 不 PUT；CF API objects 端點不支援 HEAD（live 實測 405）→ 改 GET＋Range: bytes=0-0 - 完整性：上傳前重算 hash 核對 key，不符＝failed 不上傳 - 失敗語意：failed → exit 1，ingested_hash 不動＝下輪自動重試； 回寫鉤子 Manifest.MarkIngested 留給 task 4 - 設定只走環境變數 CF_ACCOUNT_ID/CF_API_TOKEN/R2_BUCKET，不落 repo - go test 14/14 綠（httptest mock 對齊真 API：HEAD 405）；live e2e 全通 （arcrun-rag-raw-demo：真上傳→重傳 no-op→下載 diff 一致＋sha256==key） Co-Authored-By: Claude Fable 5 --- README.md | 45 ++++++++-- main.go | 121 +++++++++++++++++++------ manifest.go | 15 ++++ upload.go | 209 +++++++++++++++++++++++++++++++++++++++++++ upload_test.go | 237 +++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 595 insertions(+), 32 deletions(-) create mode 100644 upload.go create mode 100644 upload_test.go diff --git a/README.md b/README.md index 06f9fec..a3e758f 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,8 @@ ## Go 版：hash 偵測 collector（SDD ingest-hash-trigger） ``` -collector scan --root <知識資料夾> --manifest [--max-removed-ratio 0.4] [--dry-run] +collector scan --root <知識資料夾> --manifest [--max-removed-ratio 0.4] [--dry-run] +collector upload --root <知識資料夾> --manifest [--max-removed-ratio 0.4] [--dry-run] ``` 一次掃描：走訪資料夾（先只認 .md/.markdown/.txt/.docx/.pptx/.pdf）→ mtime+size fast-path @@ -22,12 +23,44 @@ collector scan --root <知識資料夾> --manifest [--max-remove （只更新路徑映射，不 retire、不重萃）；再分 added / modified / removed。 - **大量刪除防呆（R6）**：removed 數 > manifest 條目 × 40%（`--max-removed-ratio` 可調）→ removed 全部不執行、manifest 條目保留、輸出 `mass_delete_guard` 警告。 -- **重試語意**：`ingested_hash` 只會在（未來的）上傳/ingest 成功後回寫；本階段永不寫它， - 所以「偵測過但未成功 ingest」的檔每輪都會重發 added/modified——這是設計（design §2），不是 bug。 -- **本階段不接網路**：daemon 常駐（launchd）、R2 上傳、打 arcrun named-webhook＝之後的 task - （SDD task 3/4、journeys/user-onboarding 環 6）。 +- **重試語意**：`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）。 -測試：`go test ./...`（added / modified / removed / renamed / 大量刪除防呆五情境＋fast-path＋manifest 往返）。 +### `upload`：R2 content-addressed 原稿上傳（SDD task 3，design §4） + +`upload`＝`scan`＋把本輪 **added/modified** 的原稿上傳 R2（renamed/removed 內容未變/已留底，不上傳）。 + +- **走 Cloudflare REST API**（`PUT /accounts/{account_id}/r2/buckets/{bucket}/objects/{key}`， + Bearer token）——不用 S3 sigv4，token 模型跟產品其他部分一致（客戶本來就有 CF API token）。 +- **key＝`raw/`**（不含 `sha256:` 前綴，對齊 `schemas/collector-trigger.v1.schema.json` 的 `r2_key`）。 +- **冪等**：每 key 先做存在檢查，已存在＝`skipped_exists` 不重傳（content-addressed 天然去重）。 + ⚠️ CF REST API 的 objects 端點**不支援 HEAD**（2026-07-19 live 實測回 405）， + 存在檢查走 `GET`＋`Range: bytes=0-0`（存在＝200/206 只讀 1 byte，404＝不存在）。 +- **完整性**：上傳前重算 sha256 核對事件 hash；檔案在掃描後被改動＝該筆 `failed` 不上傳 + （不能把新內容塞進舊 hash 的 key），下輪重掃自然帶新 hash。 +- **失敗語意**：任一筆 `failed` → exit code 1；manifest 照存（content_hash 反映現況、 + `ingested_hash` 不動）＝下輪自動重試。上傳成功也**不**標 ingested——上傳只是鏈的第一環。 +- `--dry-run`：不碰網路、不寫 manifest，只列 `planned` 上傳清單。 +- 輸出 JSON：`{"trigger": , "uploads": [{path, r2_key, status, error?}]}`， + status＝`uploaded`／`skipped_exists`／`failed`／`planned`。 + +設定（**只走環境變數，絕不落 repo/code**）： + +| 變數 | 說明 | +|---|---| +| `CF_ACCOUNT_ID` | Cloudflare 帳號 ID | +| `CF_API_TOKEN` | 有該 bucket R2 read+write 權的 API token | +| `R2_BUCKET` | 目的 bucket 名（demo＝`arcrun-rag-raw-demo`） | +| `CF_API_BASE` | 選填，API 基底覆蓋（測試用；預設 `https://api.cloudflare.com/client/v4`） | + +測試：`go test ./...`——掃描七項（五情境＋fast-path＋manifest 往返）＋上傳七項 +（新檔上傳／同 hash 重傳 no-op／上傳失敗不標 ingested＋重試／非內容事件不上傳／ +hash 不符不上傳／env 缺漏報錯／MarkIngested 鉤子），httptest mock 對齊真 API 行為（HEAD 405）。 +live e2e（2026-07-19）：uncle6 帳號 `arcrun-rag-raw-demo` bucket 真上傳→重傳 no-op→ +`wrangler r2 object get --remote` 下載 diff 一致、sha256 與 key 相符，全通。 --- diff --git a/main.go b/main.go index 6faddc1..2cc8088 100644 --- a/main.go +++ b/main.go @@ -1,14 +1,21 @@ -// collector — hash 差異偵測 collector（SDD ingest-hash-trigger，Go 版骨架）。 +// collector — hash 差異偵測 collector（SDD ingest-hash-trigger，Go 版）。 // // 用法： // -// collector scan --root <知識資料夾> --manifest \ +// collector scan --root <知識資料夾> --manifest \ +// [--max-removed-ratio 0.4] [--dry-run] +// collector upload --root <知識資料夾> --manifest \ // [--max-removed-ratio 0.4] [--dry-run] // -// 一次掃描：走訪 root、對照 manifest、把事件（符合 -// schemas/collector-trigger.v1.schema.json）以 JSON 輸出到 stdout，並更新 manifest -// （--dry-run 不寫）。daemon 常駐/launchd/R2 上傳/打 named-webhook＝之後的 task， -// 本階段不接網路。 +// scan：走訪 root、對照 manifest、把事件（符合 schemas/collector-trigger.v1.schema.json） +// 以 JSON 輸出到 stdout，並更新 manifest（--dry-run 不寫）。 +// +// upload（task 3）＝scan＋把 added/modified 的原稿上傳 R2（content-addressed， +// key=raw/，design §4）。設定走環境變數 CF_ACCOUNT_ID / CF_API_TOKEN / +// R2_BUCKET（絕不落 repo）；--dry-run 只列出會上傳的 key（planned），不碰網路不寫 manifest。 +// 任一上傳失敗＝exit 1（manifest 照存：content_hash 反映現況、ingested_hash 不動＝可重試）。 +// +// daemon 常駐（launchd）/打 named-webhook 觸發 ingest＝之後的 task（SDD task 4、產品化段）。 package main import ( @@ -20,55 +27,117 @@ import ( ) func main() { - if len(os.Args) < 2 || os.Args[1] != "scan" { - fmt.Fprintln(os.Stderr, "用法: collector scan --root --manifest [--max-removed-ratio 0.4] [--dry-run]") + if len(os.Args) < 2 { + usage() os.Exit(2) } - fs := flag.NewFlagSet("scan", flag.ExitOnError) + switch os.Args[1] { + case "scan": + os.Exit(run(os.Args[2:], false)) + case "upload": + os.Exit(run(os.Args[2:], true)) + default: + usage() + os.Exit(2) + } +} + +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`) +} + +// run 是 scan/upload 共用主體；withUpload＝true 時掃描後把 added/modified 原稿上傳 R2。 +func run(args []string, withUpload bool) 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, "只輸出事件，不更新 manifest") - if err := fs.Parse(os.Args[2:]); err != nil { - os.Exit(2) + dryRun := fs.Bool("dry-run", false, "只輸出事件（upload 模式另列 planned 上傳清單），不更新 manifest、不碰網路") + if err := fs.Parse(args); err != nil { + return 2 } if *root == "" || *manifestPath == "" { fmt.Fprintln(os.Stderr, "錯誤：--root 與 --manifest 皆為必填") - os.Exit(2) + return 2 } + + // upload 模式先驗設定（fail fast：缺 env 連掃都不掃，不留半套狀態）。 + var client *R2Client + if withUpload && !*dryRun { + cfg, err := LoadR2ConfigFromEnv() + if err != nil { + fmt.Fprintln(os.Stderr, "collector:", err) + return 2 + } + client = NewR2Client(cfg) + } + absRoot, err := filepath.Abs(*root) if err != nil { - fatal(err) + return fail(err) } absManifest, err := filepath.Abs(*manifestPath) if err != nil { - fatal(err) + return fail(err) } m, err := LoadManifest(absManifest, absRoot) if err != nil { - fatal(err) + return fail(err) } payload, err := Scan(absRoot, m, ScanOptions{ MaxRemovedRatio: *ratio, SkipPaths: map[string]bool{absManifest: true}, // manifest 若住在 root 底下，不掃自己 }) if err != nil { - fatal(err) + return fail(err) } - if !*dryRun { - if err := m.Save(absManifest); err != nil { - fatal(err) + + var uploads []UploadResult + if withUpload { + if *dryRun { + uploads = []UploadResult{} + for _, ev := range payload.Events { + if ev.Type == "added" || ev.Type == "modified" { + uploads = append(uploads, UploadResult{Path: ev.Path, R2Key: ev.R2Key, Status: "planned"}) + } + } + } else { + uploads = UploadChanged(absRoot, payload.Events, client) } } - out, err := json.MarshalIndent(payload, "", " ") - if err != nil { - fatal(err) + + if !*dryRun { + if err := m.Save(absManifest); err != nil { + return fail(err) + } } - fmt.Println(string(out)) + + exitCode := 0 + var out any = payload + if 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 未動＝下輪自然重試 + } + } + } + data, err := json.MarshalIndent(out, "", " ") + if err != nil { + return fail(err) + } + fmt.Println(string(data)) + return exitCode } -func fatal(err error) { +func fail(err error) int { fmt.Fprintln(os.Stderr, "collector:", err) - os.Exit(1) + return 1 } diff --git a/manifest.go b/manifest.go index 4bf7e0a..50cbb6c 100644 --- a/manifest.go +++ b/manifest.go @@ -75,6 +75,21 @@ func LoadManifest(path, root string) (*Manifest, error) { return &m, nil } +// MarkIngested 是「整條 ingest 鏈成功」後的回寫鉤子（design §2）：把該路徑的 +// ingested_hash 接上當時送出的 source_hash。**R2 上傳成功不呼叫它**——上傳只是鏈的 +// 第一環，要等 task 4（named-webhook → ingest workflow）確認成功才回寫；在那之前 +// 同檔每輪重發 added/modified＝設計內重試，R2 端靠 HEAD no-op 天然冪等、零浪費。 +// 回傳 false＝路徑已不在 manifest（例如回報前檔案又被改名/刪除），呼叫端自行決定忽略或告警。 +func (m *Manifest) MarkIngested(path, sourceHash string, at int64) bool { + e, ok := m.Entries[path] + if !ok { + return false + } + e.IngestedHash = sourceHash + e.IngestedAt = at + return true +} + // Save 原子寫入（temp + rename），避免掃描中斷留半個 JSON。 func (m *Manifest) Save(path string) error { data, err := json.MarshalIndent(m, "", " ") diff --git a/upload.go b/upload.go new file mode 100644 index 0000000..f741be7 --- /dev/null +++ b/upload.go @@ -0,0 +1,209 @@ +// upload.go — R2 content-addressed 原稿上傳（SDD ingest-hash-trigger design §4，task 3）。 +// +// 走 Cloudflare REST API（非 S3 sigv4——token 模型跟產品其他部分一致，客戶本來就有 CF API token）： +// +// GET/PUT https://api.cloudflare.com/client/v4/accounts/{account_id}/r2/buckets/{bucket}/objects/{key} +// Authorization: Bearer +// +// key＝`raw/`（不含 `sha256:` 前綴，對齊 schemas/collector-trigger.v1.schema.json 的 r2_key）。 +// 冪等：先做存在檢查（GET＋Range 1 byte；端點不支援 HEAD，見 Exists 註解）， +// 已存在＝no-op 跳過 PUT（content-addressed 天然去重，design §4）。 +// 設定走環境變數 CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET，絕不落 repo。 +package main + +import ( + "fmt" + "io" + "mime" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "time" +) + +// DefaultCFAPIBase 是 Cloudflare REST API 基底；測試用 CF_API_BASE 指到 httptest server。 +const DefaultCFAPIBase = "https://api.cloudflare.com/client/v4" + +type R2Config struct { + AccountID string + APIToken string + Bucket string + BaseURL string // 空＝DefaultCFAPIBase +} + +// LoadR2ConfigFromEnv 讀 CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET（必填）與 CF_API_BASE（選填，測試用）。 +func LoadR2ConfigFromEnv() (R2Config, error) { + cfg := R2Config{ + AccountID: os.Getenv("CF_ACCOUNT_ID"), + APIToken: os.Getenv("CF_API_TOKEN"), + Bucket: os.Getenv("R2_BUCKET"), + BaseURL: os.Getenv("CF_API_BASE"), + } + var missing []string + if cfg.AccountID == "" { + missing = append(missing, "CF_ACCOUNT_ID") + } + if cfg.APIToken == "" { + missing = append(missing, "CF_API_TOKEN") + } + if cfg.Bucket == "" { + missing = append(missing, "R2_BUCKET") + } + if len(missing) > 0 { + return cfg, fmt.Errorf("R2 上傳缺環境變數：%s（設定只走環境變數，絕不寫進 repo/code）", strings.Join(missing, ", ")) + } + return cfg, nil +} + +type R2Client struct { + cfg R2Config + hc *http.Client +} + +func NewR2Client(cfg R2Config) *R2Client { + if cfg.BaseURL == "" { + cfg.BaseURL = DefaultCFAPIBase + } + return &R2Client{cfg: cfg, hc: &http.Client{Timeout: 120 * time.Second}} +} + +func (c *R2Client) objectURL(key string) string { + segs := strings.Split(key, "/") + for i, s := range segs { + segs[i] = url.PathEscape(s) + } + return fmt.Sprintf("%s/accounts/%s/r2/buckets/%s/objects/%s", + strings.TrimSuffix(c.cfg.BaseURL, "/"), + url.PathEscape(c.cfg.AccountID), + url.PathEscape(c.cfg.Bucket), + strings.Join(segs, "/")) +} + +func (c *R2Client) do(req *http.Request) (*http.Response, error) { + req.Header.Set("Authorization", "Bearer "+c.cfg.APIToken) + return c.hc.Do(req) +} + +// Exists 查同 key 是否已在 bucket（no-op 依據）。 +// ⚠️ CF REST API 的 objects 端點不支援 HEAD（2026-07-19 live 實測回 405）， +// 改用 GET＋`Range: bytes=0-0`：存在＝200/206（最多讀 1 byte 就丟）、404＝不存在、其他＝錯誤。 +func (c *R2Client) Exists(key string) (bool, error) { + req, err := http.NewRequest(http.MethodGet, c.objectURL(key), nil) + if err != nil { + return false, err + } + req.Header.Set("Range", "bytes=0-0") + resp, err := c.do(req) + if err != nil { + return false, err + } + defer resp.Body.Close() + io.Copy(io.Discard, io.LimitReader(resp.Body, 1024)) // Range 若未被支援也只讀一小段就收 + switch resp.StatusCode { + case http.StatusOK, http.StatusPartialContent: + return true, nil + case http.StatusNotFound: + return false, nil + default: + return false, fmt.Errorf("存在檢查 GET %s 非預期狀態 %d", key, resp.StatusCode) + } +} + +// Put 上傳物件。key 必為 raw/（呼叫端保證 key＝內容 hash）。 +func (c *R2Client) Put(key string, body io.Reader, size int64, contentType string) error { + req, err := http.NewRequest(http.MethodPut, c.objectURL(key), body) + if err != nil { + return err + } + req.ContentLength = size + if contentType == "" { + contentType = "application/octet-stream" + } + req.Header.Set("Content-Type", contentType) + resp, err := c.do(req) + if err != nil { + return err + } + defer resp.Body.Close() + snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("PUT %s 失敗（HTTP %d）：%s", key, resp.StatusCode, strings.TrimSpace(string(snippet))) + } + return nil +} + +// UploadResult 是單一事件的上傳結果，隨 payload 一起輸出 stdout 給呼叫端/日誌。 +type UploadResult struct { + Path string `json:"path"` + R2Key string `json:"r2_key"` + Status string `json:"status"` // uploaded | skipped_exists | failed | planned（--dry-run） + Error string `json:"error,omitempty"` +} + +// UploadChanged 把本輪 added/modified 事件的原稿上傳 R2（renamed/removed 內容未變/已留底，不上傳）。 +// +// - 冪等：每 key 先做存在檢查（Exists），存在＝skipped_exists 不 PUT。 +// - 上傳前重算 sha256 核對事件 hash——content-addressed 鐵律：key 與內容不符，寧可失敗也不上傳 +// （檔案在掃描後被改動＝本輪跳過，下輪重掃自然帶新 hash）。 +// - 失敗只記結果、不動 manifest：ingested_hash 是「整條 ingest 鏈成功」後才回寫的 +// （回寫鉤子＝Manifest.MarkIngested，由 task 4 觸發鏈呼叫；上傳成功 ≠ ingest 完成）。 +func UploadChanged(root string, events []Event, c *R2Client) []UploadResult { + results := []UploadResult{} + for _, ev := range events { + if ev.Type != "added" && ev.Type != "modified" { + continue + } + res := UploadResult{Path: ev.Path, R2Key: ev.R2Key} + full := filepath.Join(root, filepath.FromSlash(ev.Path)) + + h, err := hashFile(full) + if err != nil { + res.Status, res.Error = "failed", "讀檔失敗："+err.Error() + results = append(results, res) + continue + } + if h != ev.SourceHash { + res.Status = "failed" + res.Error = "檔案在掃描後被改動（hash 不符 key），本輪不上傳；下輪重掃會帶新 hash" + results = append(results, res) + continue + } + + exists, err := c.Exists(ev.R2Key) + if err != nil { + res.Status, res.Error = "failed", err.Error() + results = append(results, res) + continue + } + if exists { + res.Status = "skipped_exists" + results = append(results, res) + continue + } + + f, err := os.Open(full) + if err != nil { + res.Status, res.Error = "failed", "開檔失敗："+err.Error() + results = append(results, res) + continue + } + info, err := f.Stat() + if err != nil { + f.Close() + res.Status, res.Error = "failed", err.Error() + results = append(results, res) + continue + } + err = c.Put(ev.R2Key, f, info.Size(), mime.TypeByExtension(strings.ToLower(filepath.Ext(full)))) + f.Close() + if err != nil { + res.Status, res.Error = "failed", err.Error() + } else { + res.Status = "uploaded" + } + results = append(results, res) + } + return results +} diff --git a/upload_test.go b/upload_test.go new file mode 100644 index 0000000..ab352b8 --- /dev/null +++ b/upload_test.go @@ -0,0 +1,237 @@ +package main + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +const ( + testAccount = "acct-test" + testBucket = "bkt-test" + testToken = "tok-test" // 假 token，只給 httptest mock 驗 header 用 +) + +// mockR2 用 httptest 模擬 Cloudflare R2 REST API 的 objects 端點。 +// 對齊 2026-07-19 live 實測行為：HEAD 回 405（真 API 不支援）、存在檢查走 GET。 +type mockR2 struct { + mu sync.Mutex + objects map[string][]byte + existsCount int // GET（存在檢查）次數 + putCount int + failPut bool // true＝PUT 一律回 500（模擬上傳失敗） +} + +func newMockR2(t *testing.T) (*httptest.Server, *mockR2, *R2Client) { + t.Helper() + m := &mockR2{objects: map[string][]byte{}} + prefix := "/client/v4/accounts/" + testAccount + "/r2/buckets/" + testBucket + "/objects/" + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer "+testToken { + w.WriteHeader(http.StatusForbidden) + return + } + if !strings.HasPrefix(r.URL.Path, prefix) { + t.Errorf("非預期路徑: %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + key := strings.TrimPrefix(r.URL.Path, prefix) + m.mu.Lock() + defer m.mu.Unlock() + switch r.Method { + case http.MethodHead: // 真 API 行為：objects 端點不支援 HEAD + w.WriteHeader(http.StatusMethodNotAllowed) + case http.MethodGet: + m.existsCount++ + body, ok := m.objects[key] + if !ok { + w.WriteHeader(http.StatusNotFound) + io.WriteString(w, `{"success":false,"errors":[{"code":10007,"message":"object not found"}]}`) + return + } + if r.Header.Get("Range") != "" && len(body) > 0 { + w.WriteHeader(http.StatusPartialContent) + w.Write(body[:1]) + return + } + w.Write(body) + case http.MethodPut: + m.putCount++ + if m.failPut { + w.WriteHeader(http.StatusInternalServerError) + io.WriteString(w, `{"success":false,"errors":[{"code":10000,"message":"mock 上傳失敗"}]}`) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + m.objects[key] = body + io.WriteString(w, `{"success":true}`) + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } + })) + t.Cleanup(srv.Close) + client := NewR2Client(R2Config{ + AccountID: testAccount, APIToken: testToken, Bucket: testBucket, + BaseURL: srv.URL + "/client/v4", + }) + return srv, m, client +} + +// 情境 1：新檔上傳——PUT 到 raw/、body 與檔案內容一致、帶 Bearer token； +// 上傳成功也「不」標 ingested（ingested_hash 由 task 4 整鏈成功後經 MarkIngested 回寫）。 +func TestUploadNew(t *testing.T) { + _, mock, client := newMockR2(t) + root := t.TempDir() + content := "上傳測試內容 v1\n" + writeFile(t, root, "a.md", content, baseTime) + m := newTestManifest() + p := mustScan(t, root, m) + + results := UploadChanged(root, p.Events, client) + + if len(results) != 1 || results[0].Status != "uploaded" { + t.Fatalf("要 1 筆 uploaded，得到: %+v", results) + } + wantKey := "raw/" + strings.TrimPrefix(hashOf(content), "sha256:") + if results[0].R2Key != wantKey { + t.Fatalf("r2_key 不對: %s", results[0].R2Key) + } + if got, ok := mock.objects[wantKey]; !ok || string(got) != content { + t.Fatalf("R2 端物件內容不符: ok=%v got=%q", ok, string(got)) + } + if mock.putCount != 1 || mock.existsCount != 1 { + t.Fatalf("要 1 存在檢查 + 1 PUT，得到 exists=%d put=%d", mock.existsCount, mock.putCount) + } + if m.Entries["a.md"].IngestedHash != "" { + t.Fatal("上傳成功 ≠ ingest 完成，掃描/上傳階段不得寫 ingested_hash") + } +} + +// 情境 2：同 hash 重傳＝no-op——存在檢查命中就不 PUT（content-addressed 冪等，design §4）。 +func TestUploadExistingNoOp(t *testing.T) { + _, mock, client := newMockR2(t) + root := t.TempDir() + content := "同一份內容\n" + writeFile(t, root, "b.md", content, baseTime) + key := "raw/" + strings.TrimPrefix(hashOf(content), "sha256:") + mock.objects[key] = []byte(content) // 模擬先前已上傳過同 hash（可能來自別的路徑/別台機器） + + m := newTestManifest() + p := mustScan(t, root, m) + results := UploadChanged(root, p.Events, client) + + if len(results) != 1 || results[0].Status != "skipped_exists" { + t.Fatalf("要 skipped_exists，得到: %+v", results) + } + if mock.putCount != 0 { + t.Fatalf("存在檢查命中後不得 PUT，putCount=%d", mock.putCount) + } + if mock.existsCount != 1 { + t.Fatalf("要恰好 1 次存在檢查，得到 %d", mock.existsCount) + } +} + +// 情境 3：上傳失敗（PUT 500）——結果標 failed 帶錯誤、絕不標 ingested； +// manifest 的 content_hash 照掃描更新（ingested_hash 仍空＝下輪自然重試，design §2）。 +func TestUploadFailedNotIngested(t *testing.T) { + _, mock, client := newMockR2(t) + mock.failPut = true + root := t.TempDir() + writeFile(t, root, "c.md", "會失敗的內容\n", baseTime) + m := newTestManifest() + p := mustScan(t, root, m) + + results := UploadChanged(root, p.Events, client) + + if len(results) != 1 || results[0].Status != "failed" || results[0].Error == "" { + t.Fatalf("要 failed＋錯誤訊息，得到: %+v", results) + } + e := m.Entries["c.md"] + if e == nil || e.IngestedHash != "" || e.IngestedAt != 0 { + t.Fatalf("上傳失敗不得標 ingested: %+v", e) + } + if e.ContentHash != hashOf("會失敗的內容\n") { + t.Fatalf("content_hash 應照掃描更新（重試靠 ingested_hash 空）: %+v", e) + } + // 失敗後同檔重掃＝事件重發（重試語意） + p2 := mustScan(t, root, m) + if len(p2.Events) != 1 || p2.Events[0].Type != "added" { + t.Fatalf("失敗後下輪應重發 added: %+v", p2.Events) + } +} + +// 附加：renamed/removed 事件不上傳（內容未變/已留底，design §3+§4）。 +func TestUploadSkipsNonContentEvents(t *testing.T) { + _, mock, client := newMockR2(t) + root := t.TempDir() + results := UploadChanged(root, []Event{ + {Type: "renamed", Path: "x.md", OldPath: "y.md", SourceHash: hashOf("x")}, + {Type: "removed", Path: "z.md", SourceHash: hashOf("z")}, + }, client) + if len(results) != 0 { + t.Fatalf("renamed/removed 不該有上傳結果: %+v", results) + } + if mock.existsCount != 0 || mock.putCount != 0 { + t.Fatalf("不該碰網路: head=%d put=%d", mock.existsCount, mock.putCount) + } +} + +// 附加：content-addressed 完整性——檔案在掃描後被改動（hash 不符 key）＝failed 不上傳， +// 不能把新內容塞進舊 hash 的 key。 +func TestUploadHashMismatch(t *testing.T) { + _, mock, client := newMockR2(t) + root := t.TempDir() + writeFile(t, root, "d.md", "掃描時內容\n", baseTime) + m := newTestManifest() + p := mustScan(t, root, m) + + // 掃描後、上傳前檔案被改動 + writeFile(t, root, "d.md", "上傳前偷偷改了\n", baseTime.Add(time.Second)) + results := UploadChanged(root, p.Events, client) + + if len(results) != 1 || results[0].Status != "failed" { + t.Fatalf("hash 不符應 failed: %+v", results) + } + if mock.putCount != 0 { + t.Fatalf("hash 不符不得 PUT: %d", mock.putCount) + } +} + +// 附加：設定缺環境變數＝清楚報缺哪幾個。 +func TestLoadR2ConfigMissing(t *testing.T) { + t.Setenv("CF_ACCOUNT_ID", "") + t.Setenv("CF_API_TOKEN", "") + t.Setenv("R2_BUCKET", "b") + _, err := LoadR2ConfigFromEnv() + if err == nil { + t.Fatal("缺 env 應報錯") + } + msg := err.Error() + if !strings.Contains(msg, "CF_ACCOUNT_ID") || !strings.Contains(msg, "CF_API_TOKEN") || strings.Contains(msg, "R2_BUCKET,") { + t.Fatalf("錯誤訊息應列出缺的變數: %s", msg) + } +} + +// 附加：MarkIngested 回寫鉤子（task 4 用）——存在的路徑回寫成功、消失的路徑回 false。 +func TestMarkIngested(t *testing.T) { + m := newTestManifest() + m.Entries["a.md"] = &ManifestEntry{ContentHash: hashOf("a"), Size: 1, Mtime: 2} + if !m.MarkIngested("a.md", hashOf("a"), 99) { + t.Fatal("存在的路徑應回寫成功") + } + if m.Entries["a.md"].IngestedHash != hashOf("a") || m.Entries["a.md"].IngestedAt != 99 { + t.Fatalf("回寫值不對: %+v", m.Entries["a.md"]) + } + if m.MarkIngested("gone.md", hashOf("g"), 1) { + t.Fatal("不存在的路徑應回 false") + } +}