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