1package app
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "os"
9 "sort"
10 "text/tabwriter"
11 "time"
12
13 "github.com/bnema/vev/internal/adapters/brokerconfig"
14 "github.com/bnema/vev/internal/adapters/brokerstore"
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/ports"
17 "github.com/bnema/vev/internal/protocol"
18 "github.com/bnema/vev/internal/protocol/catalogue"
19)
20
21const (
22 hostActionAdd = "add"
23 hostActionRm = "rm"
24 hostActionList = "list"
25)
26
27// Remote carriages selectable with `host add --transport`.
28const (
29 hostTransportQUIC = "quic"
30 hostTransportSSH = "ssh"
31)
32
33// hostTransportRoutes maps each CLI transport to the broker policy transport.
34var hostTransportRoutes = map[string]ports.BrokerRouteKind{
35 hostTransportQUIC: ports.BrokerRouteSSHQUIC,
36 hostTransportSSH: ports.BrokerRouteSSHStdio,
37}
38
39type remoteHostDeps struct {
40 connect func(context.Context) (ports.BrokerService, error)
41 connectExisting func(context.Context) (ports.BrokerService, error)
42 stdout io.Writer
43}
44
45func defaultRemoteHostDeps() remoteHostDeps {
46 return remoteHostDeps{connect: connectProductionBroker, connectExisting: connectExistingBroker, stdout: os.Stdout}
47}
48
49func (d remoteHostDeps) withDefaults() remoteHostDeps {
50 if d.connect == nil {
51 d.connect = connectProductionBroker
52 }
53 if d.connectExisting == nil {
54 d.connectExisting = connectExistingBroker
55 }
56 if d.stdout == nil {
57 d.stdout = os.Stdout
58 }
59 return d
60}
61
62func runHostCommand(ctx context.Context, cmd command, deps remoteHostDeps) error {
63 deps = deps.withDefaults()
64 connect := deps.connect
65 if cmd.hostAction == hostActionList {
66 connect = deps.connectExisting
67 }
68 service, err := connect(ctx)
69 if err != nil {
70 if cmd.hostAction == hostActionList && errors.Is(err, errBrokerAbsent) {
71 return listDisconnectedHosts(deps.stdout)
72 }
73 return err
74 }
75 defer func() { _ = service.Close() }()
76 switch cmd.hostAction {
77 case hostActionAdd:
78 if err := domain.ValidateRemoteHostTarget(cmd.hostTarget); err != nil {
79 return err
80 }
81 policy := remoteBrokerPolicy(cmd.hostTransport)
82 _, err = service.AddHost(ctx, cmd.hostTarget, policy)
83 if isHostConflict(err) {
84 if stored := storedHostTransport(service.Snapshot(), cmd.hostTarget); stored != "" && stored != cmd.hostTransport {
85 return fmt.Errorf("host %q is already registered with transport %q; run `vev host rm %s` first, then add it again: %w", cmd.hostTarget, stored, cmd.hostTarget, err)
86 }
87 }
88 return err
89 case hostActionRm:
90 if err := domain.ValidateRemoteHostTarget(cmd.hostTarget); err != nil {
91 return err
92 }
93 var registration domain.RemoteRegistration
94 for _, daemon := range service.Snapshot().Daemons {
95 if !daemon.Local && daemon.Endpoint == cmd.hostTarget {
96 registration = daemon.Registration
97 break
98 }
99 }
100 if registration.Endpoint == "" {
101 return fmt.Errorf("unknown host %q", cmd.hostTarget)
102 }
103 removed, err := service.RemoveHost(ctx, registration)
104 if err != nil {
105 return err
106 }
107 if !removed {
108 return fmt.Errorf("unknown host %q", cmd.hostTarget)
109 }
110 return nil
111 case hostActionList:
112 return printBrokerHosts(deps.stdout, service.Snapshot())
113 default:
114 return usagef("unknown host action %q", cmd.hostAction)
115 }
116}
117
118// isHostConflict matches a host conflict both in-process and after it crossed
119// broker IPC, where only the error code survives.
120func isHostConflict(err error) bool {
121 if errors.Is(err, ports.ErrBrokerHostConflict) {
122 return true
123 }
124 var brokerErr ports.BrokerError
125 return errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorHostConflict
126}
127
128// storedHostTransport returns the CLI transport name registered for target,
129// or "" when it cannot be determined.
130func storedHostTransport(snapshot ports.BrokerSnapshot, target string) string {
131 for _, daemon := range snapshot.Daemons {
132 if daemon.Local || daemon.Endpoint != target {
133 continue
134 }
135 for name, route := range hostTransportRoutes {
136 if daemon.Policy.Transport == string(route) {
137 return name
138 }
139 }
140 }
141 return ""
142}
143
144// listDisconnectedHosts reads configured endpoints without dialing any remote.
145func listDisconnectedHosts(w io.Writer) error {
146 state := productionBrokerLayout().State
147 hosts, err := brokerstore.ReadHosts(state)
148 var endpoints []string
149 if errors.Is(err, os.ErrNotExist) {
150 // Before the first broker run, the config is the only membership.
151 config, configErr := brokerconfig.LoadPath(productionBrokerLayout(), productionBrokerConfigPath())
152 if errors.Is(configErr, os.ErrNotExist) {
153 _, err = fmt.Fprintln(w, "no hosts")
154 return err
155 }
156 if configErr != nil {
157 return configErr
158 }
159 endpoints = config.Endpoints()
160 } else if err != nil {
161 return err
162 } else {
163 for _, host := range hosts.Hosts {
164 endpoints = append(endpoints, host.Registration.Endpoint)
165 }
166 sort.Strings(endpoints)
167 }
168 if len(endpoints) == 0 {
169 _, err = fmt.Fprintln(w, "no hosts")
170 return err
171 }
172 for _, endpoint := range endpoints {
173 if _, err := fmt.Fprintf(w, "%s: not connected (no broker running)\n", endpoint); err != nil {
174 return err
175 }
176 }
177 return nil
178}
179
180func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error {
181 if err := snapshot.Validate(); err != nil {
182 return fmt.Errorf("reading broker catalogue: %w", err)
183 }
184 var endpoints []string
185 for _, daemon := range snapshot.Daemons {
186 if !daemon.Local {
187 endpoints = append(endpoints, daemon.Endpoint)
188 }
189 }
190 sort.Strings(endpoints)
191 if len(endpoints) == 0 {
192 _, err := fmt.Fprintln(w, "no hosts")
193 return err
194 }
195 tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
196 _, _ = fmt.Fprintln(tw, "TARGET\tSOURCE\tSTATUS")
197 for _, endpoint := range endpoints {
198 status := "not connected"
199 for _, daemon := range snapshot.Daemons {
200 if !daemon.Local && daemon.Endpoint == endpoint {
201 status, _ = observedHostStatus(daemon, time.Now())
202 break
203 }
204 }
205 _, _ = fmt.Fprintf(tw, "%s\tbroker\t%s\n", endpoint, status)
206 }
207 return tw.Flush()
208}
209
210func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer) error {
211 return renderBrokerSnapshotList(cmd, snapshot, stdout, time.Now())
212}
213
214func renderBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer, now time.Time) error {
215 if err := snapshot.Validate(); err != nil {
216 return fmt.Errorf("reading broker catalogue: %w", err)
217 }
218 var sessions []protocol.SessionInfo
219 var statuses []string
220 found := cmd.listHost == ""
221 for _, daemon := range snapshot.Daemons {
222 if cmd.listHost != "" && (daemon.Local || daemon.Endpoint != cmd.listHost) {
223 continue
224 }
225 found = true
226 status, showSessions := observedHostStatus(daemon, now)
227 if daemon.Local {
228 showSessions = daemon.Availability == domain.RemoteAvailabilityReachable
229 }
230 if !daemon.Local && status != "reachable" {
231 statuses = append(statuses, fmt.Sprintf("%s: %s", daemon.Endpoint, status))
232 }
233 if !showSessions {
234 continue
235 }
236 for i, session := range catalogSessionsAsInfo(daemon.Endpoint, daemon.Sessions) {
237 if daemon.Local {
238 session.Name = daemon.Sessions[i].Name
239 }
240 sessions = append(sessions, session)
241 }
242 }
243 if !found {
244 return fmt.Errorf("unknown host %q", cmd.listHost)
245 }
246 for _, status := range statuses {
247 if _, err := fmt.Fprintln(stdout, status); err != nil {
248 return err
249 }
250 }
251 if len(sessions) == 0 && len(statuses) > 0 {
252 return nil
253 }
254 printSessions(stdout, sessions)
255 return nil
256}
257
258// observedHostStatus never treats cached inventory as proof of current reachability.
259func observedHostStatus(daemon ports.BrokerDaemonObservation, now time.Time) (string, bool) {
260 if daemon.Availability == domain.RemoteAvailabilityNoDaemon {
261 return "no vev daemon — Enter in the picker to create a session", false
262 }
263 if daemon.Availability == domain.RemoteAvailabilityUnknown {
264 return "not yet observed", false
265 }
266 if daemon.Availability == domain.RemoteAvailabilityIncompatible {
267 return "incompatible (identity changed? use vev host rm/add)", false
268 }
269 if daemon.Availability != domain.RemoteAvailabilityReachable {
270 return "down (" + daemon.Availability.String() + ")", false
271 }
272 if daemon.LastSuccess.IsZero() || now.Sub(daemon.LastSuccess) > 15*time.Second {
273 if daemon.LastSuccess.IsZero() {
274 return "stale (last observation unknown)", false
275 }
276 return fmt.Sprintf("stale (%s since last observation)", now.Sub(daemon.LastSuccess).Round(time.Second)), false
277 }
278 return "reachable", true
279}
280
281func catalogSessionsAsInfo(host string, sessions []catalogue.RemoteCatalogSession) []protocol.SessionInfo {
282 out := make([]protocol.SessionInfo, 0, len(sessions))
283 for _, session := range sessions {
284 info := protocol.SessionInfo{Name: domain.RemoteSessionDisplay(session.Name, host), Ephemeral: session.Ephemeral, Tabs: catalogue.SaturateUint16(catalogue.CatalogTabCount(session)), Attached: session.Attached}
285 switch session.State {
286 case catalogue.RemoteCatalogSessionUp:
287 info.State = protocol.SessionUp
288 case catalogue.RemoteCatalogSessionDown:
289 info.State = protocol.SessionDown
290 default:
291 info.State = protocol.SessionBroken
292 }
293 out = append(out, info)
294 }
295 return out
296}