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
27type remoteHostDeps struct {
28 connect func(context.Context) (ports.BrokerService, error)
29 connectExisting func(context.Context) (ports.BrokerService, error)
30 stdout io.Writer
31}
32
33func defaultRemoteHostDeps() remoteHostDeps {
34 return remoteHostDeps{connect: connectProductionBroker, connectExisting: connectExistingBroker, stdout: os.Stdout}
35}
36
37func (d remoteHostDeps) withDefaults() remoteHostDeps {
38 if d.connect == nil {
39 d.connect = connectProductionBroker
40 }
41 if d.connectExisting == nil {
42 d.connectExisting = connectExistingBroker
43 }
44 if d.stdout == nil {
45 d.stdout = os.Stdout
46 }
47 return d
48}
49
50func runHostCommand(ctx context.Context, cmd command, deps remoteHostDeps) error {
51 deps = deps.withDefaults()
52 connect := deps.connect
53 if cmd.hostAction == hostActionList {
54 connect = deps.connectExisting
55 }
56 service, err := connect(ctx)
57 if err != nil {
58 if cmd.hostAction == hostActionList && errors.Is(err, errBrokerAbsent) {
59 return listDisconnectedHosts(deps.stdout)
60 }
61 return err
62 }
63 defer func() { _ = service.Close() }()
64 switch cmd.hostAction {
65 case hostActionAdd:
66 if err := domain.ValidateRemoteHostTarget(cmd.hostTarget); err != nil {
67 return err
68 }
69 transport, err := remoteTransportModeFromEnv(os.Getenv(envRemoteTransport))
70 if err != nil {
71 return err
72 }
73 _, err = service.AddHost(ctx, cmd.hostTarget, remoteBrokerPolicy(transport))
74 return err
75 case hostActionRm:
76 if err := domain.ValidateRemoteHostTarget(cmd.hostTarget); err != nil {
77 return err
78 }
79 var registration domain.RemoteRegistration
80 for _, daemon := range service.Snapshot().Daemons {
81 if !daemon.Local && daemon.Endpoint == cmd.hostTarget {
82 registration = daemon.Registration
83 break
84 }
85 }
86 if registration.Endpoint == "" {
87 return fmt.Errorf("vev: unknown host %q", cmd.hostTarget)
88 }
89 removed, err := service.RemoveHost(ctx, registration)
90 if err != nil {
91 return err
92 }
93 if !removed {
94 return fmt.Errorf("vev: unknown host %q", cmd.hostTarget)
95 }
96 return nil
97 case hostActionList:
98 return printBrokerHosts(deps.stdout, service.Snapshot())
99 default:
100 return usagef("unknown host action %q", cmd.hostAction)
101 }
102}
103
104// listDisconnectedHosts reads configured endpoints without dialing any remote.
105func listDisconnectedHosts(w io.Writer) error {
106 state := productionBrokerLayout().State
107 hosts, err := brokerstore.ReadHosts(state)
108 var endpoints []string
109 if errors.Is(err, os.ErrNotExist) {
110 // Before the first broker run, the config is the only membership.
111 config, configErr := brokerconfig.LoadPath(productionBrokerLayout(), productionBrokerConfigPath())
112 if errors.Is(configErr, os.ErrNotExist) {
113 _, err = fmt.Fprintln(w, "no hosts")
114 return err
115 }
116 if configErr != nil {
117 return configErr
118 }
119 endpoints = config.Endpoints()
120 } else if err != nil {
121 return err
122 } else {
123 for _, host := range hosts.Hosts {
124 endpoints = append(endpoints, host.Registration.Endpoint)
125 }
126 sort.Strings(endpoints)
127 }
128 if len(endpoints) == 0 {
129 _, err = fmt.Fprintln(w, "no hosts")
130 return err
131 }
132 for _, endpoint := range endpoints {
133 if _, err := fmt.Fprintf(w, "%s: not connected (no broker running)\n", endpoint); err != nil {
134 return err
135 }
136 }
137 return nil
138}
139
140func printBrokerHosts(w io.Writer, snapshot ports.BrokerSnapshot) error {
141 if err := snapshot.Validate(); err != nil {
142 return fmt.Errorf("vev: reading broker catalogue: %w", err)
143 }
144 var endpoints []string
145 for _, daemon := range snapshot.Daemons {
146 if !daemon.Local {
147 endpoints = append(endpoints, daemon.Endpoint)
148 }
149 }
150 sort.Strings(endpoints)
151 if len(endpoints) == 0 {
152 _, err := fmt.Fprintln(w, "no hosts")
153 return err
154 }
155 tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
156 _, _ = fmt.Fprintln(tw, "TARGET\tSOURCE\tSTATUS")
157 for _, endpoint := range endpoints {
158 status := "not connected"
159 for _, daemon := range snapshot.Daemons {
160 if !daemon.Local && daemon.Endpoint == endpoint {
161 status, _ = observedHostStatus(daemon, time.Now())
162 break
163 }
164 }
165 _, _ = fmt.Fprintf(tw, "%s\tbroker\t%s\n", endpoint, status)
166 }
167 return tw.Flush()
168}
169
170func runBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer) error {
171 return renderBrokerSnapshotList(cmd, snapshot, stdout, time.Now())
172}
173
174func renderBrokerSnapshotList(cmd command, snapshot ports.BrokerSnapshot, stdout io.Writer, now time.Time) error {
175 if err := snapshot.Validate(); err != nil {
176 return fmt.Errorf("vev: reading broker catalogue: %w", err)
177 }
178 var sessions []protocol.SessionInfo
179 var statuses []string
180 found := cmd.listHost == ""
181 for _, daemon := range snapshot.Daemons {
182 if cmd.listHost != "" && (daemon.Local || daemon.Endpoint != cmd.listHost) {
183 continue
184 }
185 found = true
186 status, showSessions := observedHostStatus(daemon, now)
187 if daemon.Local {
188 showSessions = daemon.Availability == domain.RemoteAvailabilityReachable
189 }
190 if !daemon.Local && status != "reachable" {
191 statuses = append(statuses, fmt.Sprintf("%s: %s", daemon.Endpoint, status))
192 }
193 if !showSessions {
194 continue
195 }
196 for i, session := range catalogSessionsAsInfo(daemon.Endpoint, daemon.Sessions) {
197 if daemon.Local {
198 session.Name = daemon.Sessions[i].Name
199 }
200 sessions = append(sessions, session)
201 }
202 }
203 if !found {
204 return fmt.Errorf("vev: unknown host %q", cmd.listHost)
205 }
206 for _, status := range statuses {
207 if _, err := fmt.Fprintln(stdout, status); err != nil {
208 return err
209 }
210 }
211 if len(sessions) == 0 && len(statuses) > 0 {
212 return nil
213 }
214 printSessions(stdout, sessions)
215 return nil
216}
217
218// observedHostStatus never treats cached inventory as proof of current reachability.
219func observedHostStatus(daemon ports.BrokerDaemonObservation, now time.Time) (string, bool) {
220 if daemon.Availability == domain.RemoteAvailabilityNoDaemon {
221 return "no vev daemon — Enter in the picker to create a session", false
222 }
223 if daemon.Availability == domain.RemoteAvailabilityUnknown {
224 return "not yet observed", false
225 }
226 if daemon.Availability == domain.RemoteAvailabilityIncompatible {
227 return "incompatible (identity changed? use vev host rm/add)", false
228 }
229 if daemon.Availability != domain.RemoteAvailabilityReachable {
230 return "down (" + daemon.Availability.String() + ")", false
231 }
232 if daemon.LastSuccess.IsZero() || now.Sub(daemon.LastSuccess) > 15*time.Second {
233 if daemon.LastSuccess.IsZero() {
234 return "stale (last observation unknown)", false
235 }
236 return fmt.Sprintf("stale (%s since last observation)", now.Sub(daemon.LastSuccess).Round(time.Second)), false
237 }
238 return "reachable", true
239}
240
241func catalogSessionsAsInfo(host string, sessions []catalogue.RemoteCatalogSession) []protocol.SessionInfo {
242 out := make([]protocol.SessionInfo, 0, len(sessions))
243 for _, session := range sessions {
244 info := protocol.SessionInfo{Name: domain.RemoteSessionDisplay(session.Name, host), Ephemeral: session.Ephemeral, Tabs: catalogue.SaturateUint16(catalogue.CatalogTabCount(session)), Attached: session.Attached}
245 switch session.State {
246 case catalogue.RemoteCatalogSessionUp:
247 info.State = protocol.SessionUp
248 case catalogue.RemoteCatalogSessionDown:
249 info.State = protocol.SessionDown
250 default:
251 info.State = protocol.SessionBroken
252 }
253 out = append(out, info)
254 }
255 return out
256}