// 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 collector 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 }