Migration
This is my new public git source viewer. Source coming soon. Please contact with concerns/bugs.
Contact me
internal/core/watch.go
d658fcc4ea9f3bb562fd00dce8f863934560c56c
· 5.1 KB · 196 lines
raw
| 1 | package core |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "log/slog" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "sync" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/fsnotify/fsnotify" |
| 12 | |
| 13 | "gsg/internal/gitx" |
| 14 | ) |
| 15 | |
| 16 | // discoverConcurrency bounds how many repositories are opened at once during |
| 17 | // the initial scan. Each one is a handful of git forks plus, for |
| 18 | // keyring-bearing repos, a gpg import -- unbounded parallelism would fork |
| 19 | // thousands of processes at once on a large install. |
| 20 | const discoverConcurrency = 16 |
| 21 | |
| 22 | // watcher owns every repo directly under root: which ones are currently |
| 23 | // open, and the fsnotify watch that tells it when to look again. reconcile |
| 24 | // is the single entry point for loading, reloading, or dropping a repo -- |
| 25 | // used identically by the initial scan and by every later filesystem event. |
| 26 | type watcher struct { |
| 27 | site *Site |
| 28 | root string |
| 29 | fsw *fsnotify.Watcher |
| 30 | |
| 31 | mu sync.Mutex |
| 32 | repos map[string]*gitx.Repo // open repos, by directory name -- reused across reconciles |
| 33 | dirName map[string]string // git-dir path -> directory name, for event dispatch |
| 34 | |
| 35 | debMu sync.Mutex |
| 36 | debounce map[string]*time.Timer |
| 37 | } |
| 38 | |
| 39 | // Start discovers every repository under site.Root, then watches root and |
| 40 | // each repo's git-dir for changes for the life of the process. It returns |
| 41 | // once the initial scan completes, the same as the old discover() did. |
| 42 | func Start(site *Site) error { |
| 43 | fsw, err := fsnotify.NewWatcher() |
| 44 | if err != nil { |
| 45 | return fmt.Errorf("config watch: %w", err) |
| 46 | } |
| 47 | if err := fsw.Add(site.Root); err != nil { |
| 48 | fsw.Close() |
| 49 | return fmt.Errorf("config watch: %w", err) |
| 50 | } |
| 51 | |
| 52 | w := &watcher{ |
| 53 | site: site, |
| 54 | root: site.Root, |
| 55 | fsw: fsw, |
| 56 | repos: map[string]*gitx.Repo{}, |
| 57 | dirName: map[string]string{}, |
| 58 | debounce: map[string]*time.Timer{}, |
| 59 | } |
| 60 | |
| 61 | entries, err := os.ReadDir(site.Root) |
| 62 | if err != nil { |
| 63 | fsw.Close() |
| 64 | return err |
| 65 | } |
| 66 | |
| 67 | sem := make(chan struct{}, discoverConcurrency) |
| 68 | var wg sync.WaitGroup |
| 69 | for _, e := range entries { |
| 70 | if !e.IsDir() { |
| 71 | continue |
| 72 | } |
| 73 | wg.Add(1) |
| 74 | go func(name string) { |
| 75 | defer wg.Done() |
| 76 | sem <- struct{}{} |
| 77 | defer func() { <-sem }() |
| 78 | w.reconcile(name) |
| 79 | }(e.Name()) |
| 80 | } |
| 81 | wg.Wait() |
| 82 | |
| 83 | if len(site.RepoList()) == 0 { |
| 84 | fsw.Close() |
| 85 | return fmt.Errorf("no repositories found under %s", site.Root) |
| 86 | } |
| 87 | |
| 88 | go w.run() |
| 89 | return nil |
| 90 | } |
| 91 | |
| 92 | // reconcile loads, reloads, or drops exactly one repo, named by its |
| 93 | // directory under root. It is existence-driven rather than event-driven -- |
| 94 | // it doesn't ask what kind of filesystem event caused this call, only |
| 95 | // whether root/name currently exists and is a valid repo -- which is what |
| 96 | // lets the same function serve both the initial concurrent scan and the |
| 97 | // one-at-a-time runtime event loop with no special-casing between them. |
| 98 | func (w *watcher) reconcile(name string) { |
| 99 | dir := filepath.Join(w.root, name) |
| 100 | |
| 101 | w.mu.Lock() |
| 102 | r, known := w.repos[name] |
| 103 | w.mu.Unlock() |
| 104 | |
| 105 | if _, err := os.Stat(dir); err != nil { |
| 106 | if known { |
| 107 | w.mu.Lock() |
| 108 | delete(w.repos, name) |
| 109 | w.fsw.Remove(r.Dir) |
| 110 | delete(w.dirName, r.Dir) |
| 111 | w.mu.Unlock() |
| 112 | w.site.remove(r.Slug) |
| 113 | } |
| 114 | return |
| 115 | } |
| 116 | |
| 117 | if !known { |
| 118 | var err error |
| 119 | if r, err = gitx.Open(dir); err != nil { |
| 120 | return // not a repo (or a bare clone still being populated) |
| 121 | } |
| 122 | w.mu.Lock() |
| 123 | w.repos[name] = r |
| 124 | w.fsw.Add(r.Dir) // catches this repo's own config/description writes from here on |
| 125 | w.dirName[r.Dir] = name |
| 126 | w.mu.Unlock() |
| 127 | } |
| 128 | |
| 129 | info := LoadRepoInfo(r) |
| 130 | |
| 131 | // Converge the keyring to whatever's configured now, on this call just |
| 132 | // like any other -- no "only import the first time" special case. |
| 133 | if info.GPGKeyring != r.KeyringPath() { |
| 134 | if info.GPGKeyring == "" { |
| 135 | r.ClearKeyring() |
| 136 | } else if err := r.UseKeyring(info.GPGKeyring); err != nil { |
| 137 | slog.Warn("gpg keyring: verification disabled", "repo", info.Slug, "err", err) |
| 138 | info.GPGKeyring = "" |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | w.site.upsert(Handle{Repo: r, Info: info}) |
| 143 | } |
| 144 | |
| 145 | // run consumes filesystem events for the life of the process. A repo dir |
| 146 | // appearing, disappearing, or being renamed under root is dispatched by its |
| 147 | // (new) name; a write to a known repo's config or description file is |
| 148 | // dispatched by that repo's name. |
| 149 | func (w *watcher) run() { |
| 150 | for { |
| 151 | select { |
| 152 | case ev, ok := <-w.fsw.Events: |
| 153 | if !ok { |
| 154 | return |
| 155 | } |
| 156 | if filepath.Dir(ev.Name) == w.root { |
| 157 | w.scheduleReconcile(filepath.Base(ev.Name)) |
| 158 | continue |
| 159 | } |
| 160 | if base := filepath.Base(ev.Name); base == "config" || base == "description" { |
| 161 | w.mu.Lock() |
| 162 | name := w.dirName[filepath.Dir(ev.Name)] |
| 163 | w.mu.Unlock() |
| 164 | if name != "" { |
| 165 | w.scheduleReconcile(name) |
| 166 | } |
| 167 | } |
| 168 | case err, ok := <-w.fsw.Errors: |
| 169 | if !ok { |
| 170 | return |
| 171 | } |
| 172 | slog.Warn("config watch error", "err", err) |
| 173 | } |
| 174 | } |
| 175 | } |
| 176 | |
| 177 | // scheduleReconcile coalesces bursts of events for the same repo -- a single |
| 178 | // `git config` write is a lockfile create plus a rename, i.e. at least two |
| 179 | // raw events for one logical change -- into a single reconcile call. |
| 180 | func (w *watcher) scheduleReconcile(name string) { |
| 181 | const debounceDelay = 200 * time.Millisecond |
| 182 | |
| 183 | w.debMu.Lock() |
| 184 | defer w.debMu.Unlock() |
| 185 | |
| 186 | if t, ok := w.debounce[name]; ok { |
| 187 | t.Reset(debounceDelay) |
| 188 | return |
| 189 | } |
| 190 | w.debounce[name] = time.AfterFunc(debounceDelay, func() { |
| 191 | w.debMu.Lock() |
| 192 | delete(w.debounce, name) |
| 193 | w.debMu.Unlock() |
| 194 | w.reconcile(name) |
| 195 | }) |
| 196 | } |