From f6bfa96d93a2ec4ce679daf5ae78c3193d224a91 Mon Sep 17 00:00:00 2001 From: Andriy Oblivantsev Date: Fri, 21 Aug 2026 20:38:46 +0100 Subject: [PATCH] =?UTF-8?q?feat(stack):=20bin/stack/sync.go=20=E2=80=94=20?= =?UTF-8?q?deterministic=20sync=20wave=20orchestrator=20(#75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixed-order wave: mail-sync → mail-import → chats (--with-chats) → contact-brain → git-brain → contact-crm. --only subset keeps canonical order; per-step RUN/OK/FAIL/SKIP report; exit 1 on any failure. Ladybug single-writer handling: local brain-serve/search are quiesced (SIGTERM) around write steps and restarted via bin/stack/serve-brain. Shebang tools invoked with their own tags (mail_import; brain via zig). --- bin/stack/sync.go | 198 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 198 insertions(+) create mode 100644 bin/stack/sync.go diff --git a/bin/stack/sync.go b/bin/stack/sync.go new file mode 100644 index 0000000..4b80c38 --- /dev/null +++ b/bin/stack/sync.go @@ -0,0 +1,198 @@ +// usr/bin/env go run "$0" "$@"; exit +// +// bin/stack/sync.go - deterministic sync wave: run every ingest/reconcile step +// in fixed order, stream each tool's output, print a summary, exit non-zero on +// any failure. +// +// ./bin/stack/sync.go # default wave +// ./bin/stack/sync.go --with-chats --contacts ~/contacts.vcf +// ./bin/stack/sync.go --only mail,crm # subset, fixed order kept +// +// Steps (fixed order): +// 1. mail-sync bin/mail/sync.go (skipped without creds) +// 2. mail-import bin/mail/import.go +// 3. chats chat/sync telegram+linkedin (--with-chats) +// 4. contact-brain brain/import-contact.go (--contacts PATH) +// 5. git-brain brain/import-git.go (--git-root DIR) +// 6. contact-crm onlyoffice/import-contact.go (--contacts PATH) +// +// Idempotent by construction: every step is a reconcile/upsert. Bulk rebuild +// is deliberately NOT part of the wave (expensive) — use brain/index.go. +package main + +import ( + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "time" + + cliparse "github.com/eSlider/2dph/pkg/cli" + "github.com/eSlider/2dph/pkg/repo" +) + +func main() { + os.Exit(run(os.Args[1:])) +} + +func run(args []string) int { + var ( + only string + contacts string + gitRoot string + withChats bool + dryRun bool + ) + p := cliparse.New("stack-sync") + p.Description = "deterministic sync wave: mail/chats/contacts/git → brain + OO CRM" + p.String(&only, "", "only", "comma-separated subset: mail,chats,contacts,git,crm") + p.String(&contacts, "", "contacts", "address-book file/dir for brain+CRM reconcile") + p.String(&gitRoot, "", "git-root", "dir of git repos to import into the brain") + p.Bool(&withChats, "", "with-chats", "include telegram/linkedin chat sync") + p.Bool(&dryRun, "", "dry-run", "print the wave without executing") + if err := cliparse.Parse(p, args); err != nil { + return cliparse.Fail(err) + } + + want := map[string]bool{} + if only != "" { + for _, s := range strings.Split(only, ",") { + want[strings.TrimSpace(s)] = true + } + } + included := func(name string) bool { return len(want) == 0 || want[name] } + + steps := []struct { + name string + args []string + skip string + }{ + {"mail", []string{"bin/mail/sync.go"}, ""}, + {"mail-import", []string{"bin/mail/import.go", "--from-raw", "var/mail"}, ""}, + {"chats", []string{"bin/chat/sync.go"}, skipUnless(withChats, "--with-chats")}, + {"contact-brain", contactStep("bin/brain/import-contact.go", contacts), skipUnless(contacts != "", "--contacts")}, + {"git-brain", gitStep(gitRoot), skipUnless(gitRoot != "", "--git-root")}, + {"contact-crm", contactStep("bin/onlyoffice/import-contact.go", contacts), skipUnless(contacts != "", "--contacts")}, + } + + var failed int + fmt.Fprintln(os.Stderr, "stack-sync: wave start") + + // Ladybug allows a single writer: local brain-serve/brain-search processes + // hold kb.lbug. Quiesce them around write steps, restart after the wave. + writeSteps := map[string]bool{"contact-brain": true, "git-brain": true} + needWrite := false + for _, st := range steps { + if writeSteps[st.name] && included(st.name) && st.skip == "" && !dryRun { + needWrite = true + } + } + stopped := false + if needWrite && brainServing() { + fmt.Fprintln(os.Stderr, " quiesce STOP brain-serve/search (single-writer)") + quiesceBrain() + stopped = true + } + defer func() { + if stopped { + restartBrain() + } + }() + + for _, st := range steps { + if !included(st.name) { + continue + } + if st.skip != "" { + fmt.Fprintf(os.Stderr, " %-14s SKIP (%s)\n", st.name, st.skip) + continue + } + if dryRun { + fmt.Fprintf(os.Stderr, " %-14s DRY %v\n", st.name, st.args) + continue + } + fmt.Fprintf(os.Stderr, " %-14s RUN %v\n", st.name, st.args) + cmd := runner(st.args[0], st.args[1:]) + cmd.Stdout, cmd.Stderr = os.Stderr, os.Stderr + if err := cmd.Run(); err != nil { + failed++ + fmt.Fprintf(os.Stderr, " %-14s FAIL %v\n", st.name, err) + continue + } + fmt.Fprintf(os.Stderr, " %-14s OK\n", st.name) + } + fmt.Fprintf(os.Stderr, "stack-sync: wave done (failures=%d)\n", failed) + if failed > 0 { + return 1 + } + return 0 +} + +// runner builds the exec for a shebang tool. The `//usr/bin/env` trick is not +// kernel-executable, so we reproduce each tool's own invocation: +// - brain/* need Zig CGO (bin/cgo/zig go run -tags=system_ladybug) +// - some tools carry build tags (mail/import → mail_import) +// - everything else is plain `go run` +func runner(tool string, args []string) *exec.Cmd { + dir := "." + if i := strings.LastIndex(tool, "/"); i > 0 { + dir = tool[:i] + } + switch { + case strings.HasPrefix(tool, "bin/brain/"): + script := fmt.Sprintf(`exec "%s/../cgo/zig" go run -tags=system_ladybug %q "$@"`, dir, tool) + argv := append([]string{"-c", script, "--"}, args...) + return exec.Command("bash", argv...) + case tool == "bin/mail/import.go": + return exec.Command("go", append([]string{"run", "-tags=mail_import", tool}, args...)...) + default: + return exec.Command("go", append([]string{"run", tool}, args...)...) + } +} + +func contactStep(tool, contacts string) []string { + a := []string{tool, "--sources", contacts} + return a +} + +func gitStep(root string) []string { + return []string{"bin/brain/import-git.go", "--root", root} +} + +func skipUnless(cond bool, what string) string { + if cond { + return "" + } + return what +} + +// brainServing reports whether local built brain processes are running. +func brainServing() bool { + out, err := exec.Command("pgrep", "-f", `bin-build/brain-(serve|search)`).Output() + return err == nil && len(strings.TrimSpace(string(out))) > 0 +} + +// quiesceBrain SIGTERMs the local brain serve/search processes and waits for +// them to release kb.lbug (bounded). +func quiesceBrain() { + _ = exec.Command("pkill", "-TERM", "-f", `bin-build/brain-serve`).Run() + _ = exec.Command("pkill", "-TERM", "-f", `bin-build/brain-search serve`).Run() + for i := 0; i < 30; i++ { + if !brainServing() { + time.Sleep(500 * time.Millisecond) // let the kernel release fds + return + } + time.Sleep(200 * time.Millisecond) + } + fmt.Fprintln(os.Stderr, " quiesce WARN brain processes still running") +} + +// restartBrain brings the API + embedding daemon back via the stack script. +func restartBrain() { + fmt.Fprintln(os.Stderr, " restart brain-serve/search") + root := repo.Root() + cmd := exec.Command("bash", filepath.Join(root, "bin", "stack", "serve-brain")) + cmd.Dir = root + _ = cmd.Run() +}