package refscan import ( "encoding/json" "errors" "fmt" "os" "os/exec" "path/filepath" "sort" "strings" "sync" "sync/atomic" "time" "bugabinga.net/luci/internal/inbox" "bugabinga.net/luci/internal/trigger" ) const zeroRev = "0000000000000000000000000000000000000000" var ( pollMu sync.Mutex lastEventID atomic.Int64 writeInbox = inbox.WriteID saveRefs = save ) type RepoErrors struct { errs map[string]error } func (e *RepoErrors) Error() string { names := make([]string, 0, len(e.errs)) for name := range e.errs { names = append(names, name) } sort.Strings(names) messages := make([]string, 0, len(names)) for _, name := range names { messages = append(messages, e.errs[name].Error()) } return strings.Join(messages, "\n") } func (e *RepoErrors) Failed(repo string) bool { _, failed := e.errs[repo] return failed } type PollConfig struct { DataDir string RepoRoots []string InboxDir string } type Ref struct { Repo string `json:"repo"` Name string `json:"name"` Rev string `json:"rev"` } type outboxEvent struct { ID string `json:"id"` Event trigger.PushEvent `json:"event"` } type outbox struct { Current []Ref `json:"current"` Events []outboxEvent `json:"events"` } func Poll(cfg PollConfig) ([]string, error) { pollMu.Lock() defer pollMu.Unlock() statePath := filepath.Join(cfg.DataDir, "state", "refs.json") outboxPath := filepath.Join(cfg.DataDir, "state", "refs.outbox.json") previous, existed, err := load(statePath) if err != nil { return nil, err } ids, err := recoverOutbox(cfg, statePath, outboxPath) if err != nil { return ids, err } if existed { previous, _, err = load(statePath) if err != nil { return ids, err } } current, scanErr := Scan(cfg.RepoRoots) if current == nil { return ids, scanErr } var repoErrs *RepoErrors if errors.As(scanErr, &repoErrs) { for _, ref := range previous { if repoErrs.Failed(ref.Repo) { current = append(current, ref) } } sortRefs(current) } if !existed { return ids, errors.Join(scanErr, saveRefs(statePath, current)) } events := Events(previous, current) if len(events) == 0 { return ids, errors.Join(scanErr, saveRefs(statePath, current)) } pending := outbox{Current: current, Events: make([]outboxEvent, len(events))} for i, event := range events { pending.Events[i] = outboxEvent{ID: nextEventID(), Event: event} } if err := saveOutbox(outboxPath, pending); err != nil { return ids, errors.Join(scanErr, err) } published, err := recoverOutbox(cfg, statePath, outboxPath) ids = append(ids, published...) return ids, errors.Join(scanErr, err) } // recoverOutbox publishes durable occurrences before advancing refs.json. // Replaying an outbox reuses its IDs, so an interrupted A-B-A-B sequence // retains every transition without creating a second run for one occurrence. func recoverOutbox(cfg PollConfig, statePath, outboxPath string) ([]string, error) { pending, found, err := loadOutbox(outboxPath) if err != nil || !found { return nil, err } ids := make([]string, 0, len(pending.Events)) for _, item := range pending.Events { err := writeInbox(cfg.InboxDir, item.ID, item.Event) if err == nil { ids = append(ids, item.ID) continue } if !errors.Is(err, os.ErrExist) { return ids, err } } if err := inbox.Sync(cfg.InboxDir); err != nil { return ids, err } if err := saveRefs(statePath, pending.Current); err != nil { return ids, err } if err := os.Remove(outboxPath); err != nil && !errors.Is(err, os.ErrNotExist) { return ids, err } return ids, syncDir(filepath.Dir(outboxPath)) } func nextEventID() string { for { previous := lastEventID.Load() next := time.Now().UnixNano() if next <= previous { next = previous + 1 } if lastEventID.CompareAndSwap(previous, next) { return fmt.Sprintf("%d", next) } } } func Scan(roots []string) ([]Ref, error) { refs := []Ref{} repoErrs := &RepoErrors{errs: map[string]error{}} for _, root := range roots { entries, err := os.ReadDir(root) if err != nil { if os.IsNotExist(err) { continue } return nil, err } for _, entry := range entries { if !entry.IsDir() { continue } slug := strings.TrimSuffix(entry.Name(), ".git") if !cleanSlug(slug) { continue } repoPath := filepath.Join(root, entry.Name()) if _, err := os.Stat(filepath.Join(repoPath, "HEAD")); err != nil { continue } repoRefs, err := scanRepo(repoPath, slug) if err != nil { repoErrs.errs[slug] = err continue } refs = append(refs, repoRefs...) } } sortRefs(refs) if len(repoErrs.errs) != 0 { return refs, repoErrs } return refs, nil } func Events(previous []Ref, current []Ref) []trigger.PushEvent { prev := map[string]Ref{} for _, ref := range previous { prev[key(ref)] = ref } events := []trigger.PushEvent{} for _, ref := range current { old := zeroRev if before, ok := prev[key(ref)]; ok { old = before.Rev } if old == ref.Rev { continue } events = append(events, trigger.PushEvent{Kind: "push", Repo: ref.Repo, Old: old, New: ref.Rev, Ref: ref.Name}) } return events } func scanRepo(repoPath, slug string) ([]Ref, error) { cmd := exec.Command("git", "--git-dir", repoPath, "for-each-ref", "--format=%(refname) %(objectname) %(*objectname)", "refs/heads", "refs/tags") out, err := cmd.CombinedOutput() if err != nil { return nil, fmt.Errorf("git scan %s: %w: %s", repoPath, err, strings.TrimSpace(string(out))) } refs := []Ref{} for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { if strings.TrimSpace(line) == "" { continue } fields := strings.Fields(line) if len(fields) < 2 || len(fields) > 3 { return nil, fmt.Errorf("bad ref line %q", line) } rev := fields[1] if len(fields) == 3 { rev = fields[2] } refs = append(refs, Ref{Repo: slug, Name: fields[0], Rev: rev}) } return refs, nil } func load(path string) ([]Ref, bool, error) { data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { return nil, false, nil } return nil, false, err } var refs []Ref if err := json.Unmarshal(data, &refs); err != nil { return nil, false, err } return refs, true, nil } func loadOutbox(path string) (outbox, bool, error) { data, err := os.ReadFile(path) if errors.Is(err, os.ErrNotExist) { return outbox{}, false, nil } if err != nil { return outbox{}, false, err } var pending outbox if err := json.Unmarshal(data, &pending); err != nil { return outbox{}, false, err } for _, item := range pending.Events { if item.ID == "" || filepath.Base(item.ID) != item.ID { return outbox{}, false, fmt.Errorf("unsafe ref outbox id %q", item.ID) } if err := item.Event.Validate(); err != nil { return outbox{}, false, err } } return pending, true, nil } func saveOutbox(path string, pending outbox) error { return saveJSON(path, pending) } func save(path string, refs []Ref) error { return saveJSON(path, refs) } func saveJSON(path string, value any) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return err } data, err := json.MarshalIndent(value, "", " ") if err != nil { return err } data = append(data, '\n') file, err := os.OpenFile(path+".tmp", os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) if err != nil { return err } tmp := file.Name() if _, err := file.Write(data); err != nil { file.Close() _ = os.Remove(tmp) return err } if err := file.Sync(); err != nil { file.Close() _ = os.Remove(tmp) return err } if err := file.Close(); err != nil { _ = os.Remove(tmp) return err } if err := os.Rename(tmp, path); err != nil { _ = os.Remove(tmp) return err } return syncDir(filepath.Dir(path)) } func syncDir(path string) error { dir, err := os.Open(path) if err != nil { return err } defer dir.Close() return dir.Sync() } func sortRefs(refs []Ref) { sort.Slice(refs, func(i, j int) bool { if refs[i].Repo != refs[j].Repo { return refs[i].Repo < refs[j].Repo } return refs[i].Name < refs[j].Name }) } func key(ref Ref) string { return ref.Repo + "\x00" + ref.Name } func cleanSlug(slug string) bool { return strings.TrimSpace(slug) != "" && slug == filepath.Base(slug) && slug != "." && slug != ".." && !strings.Contains(slug, string(filepath.Separator)) }