1package app
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "time"
9
10 "github.com/bnema/vev/internal/adapters/brokeripc"
11 "github.com/bnema/vev/internal/adapters/ipc"
12 "github.com/bnema/vev/internal/adapters/lifecycle"
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/pkg/safedir"
15)
16
17const productionBrokerStartupTimeout = 10 * time.Second
18
19// connectExistingBroker dials without ensuring or spawning a broker.
20func connectExistingBroker(ctx context.Context) (ports.BrokerService, error) {
21 path := brokeripc.SocketPath(productionBrokerLayout().Runtime)
22 service, err := brokeripc.NewConnector(path, brokeripc.Config{PassiveSubscribe: true}).Connect(ctx)
23 if err != nil {
24 if backendAbsent(err) {
25 return nil, absentError(path)
26 }
27 return nil, err
28 }
29 if err := awaitBrokerPublication(ctx, service); err != nil {
30 _ = service.Close()
31 return nil, err
32 }
33 return service, nil
34}
35
36// connectProductionBroker connects to the sole per-user production broker.
37// Process election and idle lifetime remain owned by the existing broker
38// launcher and broker.Supervisor; callers receive only the semantic façade.
39func connectProductionBroker(ctx context.Context) (ports.BrokerService, error) {
40 if err := ensureProductionBrokerConfig(); err != nil {
41 return nil, err
42 }
43 // Validate the shared parent before creating broker-owned descendants,
44 // especially when a long XDG root maps directly into /tmp.
45 if err := safedir.EnsurePrivate(ipc.SocketDir()); err != nil {
46 return nil, fmt.Errorf("secure broker runtime parent: %w", err)
47 }
48 layout := productionBrokerLayout()
49 request := brokerStatusRequest{
50 layout: layout, socketPath: brokeripc.SocketPath(layout.Runtime),
51 deadline: time.Now().Add(productionBrokerStartupTimeout), timeout: productionBrokerStartupTimeout,
52 root: layout.Root,
53 }
54 deps := defaultBrokerStatusDeps()
55 deps.spawn = func(ctx context.Context, _ string, _ time.Duration, _ bool) error {
56 return spawnProductionBrokerLauncher(ctx)
57 }
58 connect := func() (ports.BrokerService, error) {
59 if _, err := ensureBrokerReady(ctx, request, deps); err != nil {
60 return nil, fmt.Errorf("ensure broker: %w", err)
61 }
62 return brokeripc.NewConnector(request.socketPath, brokeripc.Config{}).Connect(ctx)
63 }
64 service, err := connect()
65 if errors.Is(err, brokeripc.ErrBrokerRetired) {
66 owner, acquireErr := lifecycle.Acquire(ctx, layout.Runtime, 50*time.Millisecond)
67 if acquireErr != nil {
68 return nil, fmt.Errorf("wait for retiring broker: %w", acquireErr)
69 }
70 if releaseErr := owner.Release(); releaseErr != nil {
71 return nil, fmt.Errorf("release broker ownership: %w", releaseErr)
72 }
73 request.deadline = time.Now().Add(productionBrokerStartupTimeout)
74 service, err = connect()
75 }
76 if err != nil {
77 return nil, fmt.Errorf("connect broker: %w", err)
78 }
79 if _, stale := brokeripc.StaleBuild(service); stale {
80 fmt.Fprintln(os.Stderr, `vev: the running broker is from another build; run "vev kill --all" to restart it`)
81 }
82 if err := awaitBrokerPublication(ctx, service); err != nil {
83 _ = service.Close()
84 return nil, fmt.Errorf("await broker publication: %w", err)
85 }
86 return service, nil
87}
88
89func awaitBrokerPublication(ctx context.Context, service ports.BrokerService) error {
90 sub, err := service.Subscribe()
91 if err != nil {
92 return err
93 }
94 defer sub.Close()
95 for {
96 snapshot := service.Snapshot()
97 if snapshot.Epoch != 0 {
98 return snapshot.Validate()
99 }
100 select {
101 case <-ctx.Done():
102 return ctx.Err()
103 case <-service.Done():
104 return service.Err()
105 case <-sub.Changed():
106 }
107 }
108}