arcrun-collector/main.go

// collector — hash 差異偵測 collector（SDD ingest-hash-trigger，Go 版）。
//
// 用法：
//
//	collector scan   --root <知識資料夾> --manifest <manifest.json 路徑> \
//	    [--max-removed-ratio 0.4] [--dry-run]
//	collector upload --root <知識資料夾> --manifest <manifest.json 路徑> \
//	    [--max-removed-ratio 0.4] [--dry-run]
//	collector sync   --root <知識資料夾> --manifest <manifest.json 路徑> \
//	    [--max-removed-ratio 0.4] [--dry-run]
//
// 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/<sha256hex>，design §4）。設定走環境變數 CF_ACCOUNT_ID / CF_API_TOKEN /
// R2_BUCKET（絕不落 repo）；--dry-run 只列出會上傳的 key（planned），不碰網路不寫 manifest。
// 任一上傳失敗＝exit 1（manifest 照存：content_hash 反映現況、ingested_hash 不動＝可重試）。
//
// 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 (
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"path/filepath"
	"time"
)

type runMode struct {
	withUpload  bool
	withTrigger bool
}

func main() {
	if len(os.Args) < 2 {
		usage()
		os.Exit(2)
	}
	switch os.Args[1] {
	case "scan":
		os.Exit(run(os.Args[2:], runMode{}))
	case "upload":
		os.Exit(run(os.Args[2:], runMode{withUpload: true}))
	case "sync":
		os.Exit(run(os.Args[2:], runMode{withUpload: true, withTrigger: true}))
	case "direct":
		os.Exit(runDirect(os.Args[2:]))
	default:
		usage()
		os.Exit(2)
	}
}

// newFlagSet 是子命令共用的 flag.FlagSet 建構子（ExitOnError＝解析失敗直接退出）。
func newFlagSet() *flag.FlagSet {
	return flag.NewFlagSet("collector", flag.ExitOnError)
}

func usage() {
	fmt.Fprintln(os.Stderr, `用法:
  collector scan   --root <dir> --manifest <file> [--max-removed-ratio 0.4] [--dry-run]
  collector upload --root <dir> --manifest <file> [--max-removed-ratio 0.4] [--dry-run]
  collector sync   --root <dir> --manifest <file> [--max-removed-ratio 0.4] [--dry-run]
  collector direct --config <config.json> [--once] [--dry-run]
upload 需環境變數: CF_ACCOUNT_ID / CF_API_TOKEN / R2_BUCKET
sync   另需:      ARCRUN_TRIGGER_URL（named-webhook 觸發完整 URL）
direct（無 R2）:  監看資料夾 → 讀檔內容直送實例 rag_ingest_direct workflow；設定走 --config JSON`)
}

// 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/sync 模式另列 planned 清單），不更新 manifest、不碰網路")
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if *root == "" || *manifestPath == "" {
		fmt.Fprintln(os.Stderr, "錯誤：--root 與 --manifest 皆為必填")
		return 2
	}

	// 先驗設定（fail fast：缺 env 連掃都不掃，不留半套狀態）。
	var client *R2Client
	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
		}
	}

	absRoot, err := filepath.Abs(*root)
	if err != nil {
		return fail(err)
	}
	absManifest, err := filepath.Abs(*manifestPath)
	if err != nil {
		return fail(err)
	}

	m, err := LoadManifest(absManifest, absRoot)
	if err != nil {
		return fail(err)
	}
	payload, err := Scan(absRoot, m, ScanOptions{
		MaxRemovedRatio: *ratio,
		SkipPaths:       map[string]bool{absManifest: true}, // manifest 若住在 root 底下，不掃自己
	})
	if err != nil {
		return fail(err)
	}

	exitCode := 0

	var uploads []UploadResult
	if mode.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)
			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())
				}
			}
		}
	}

	if !*dryRun {
		if err := m.Save(absManifest); err != nil {
			return fail(err)
		}
	}

	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"`
		}{outPayload, uploads}
	}
	data, err := json.MarshalIndent(out, "", "  ")
	if err != nil {
		return fail(err)
	}
	fmt.Println(string(data))
	return exitCode
}

func fail(err error) int {
	fmt.Fprintln(os.Stderr, "collector:", err)
	return 1
}