1package app
2
3import (
4 "context"
5 "errors"
6 "fmt"
7
8 "github.com/bnema/vev/internal/adapters/brokerconfig"
9 "github.com/bnema/vev/internal/adapters/daemonmux"
10 "github.com/bnema/vev/internal/adapters/sessionwire"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/usecase/broker"
13)
14
15// Broker-owned local observation composition.
16//
17// The broker observes its own machine daemon through the same daemonmux
18// carriage a local logical stream is dialed over. broker.LocalDaemonProbe owns
19// the observation itself; this file only proves the configured route is a local
20// Unix socket (so probing can never spawn a process) and wires the broker's
21// already owned endpoint connector and session codec into it.
22
23// localCarrierDialer establishes one raw framed daemonmux carriage for one
24// resolved dial target. It is the same seam the pooled mux connector uses.
25type localCarrierDialer func(ctx context.Context, target ports.BrokerDialTarget) (daemonmux.RawFramedTransport, error)
26
27func newDynamicLocalRouteProbe(route brokerconfig.Route, policy ports.BrokerPolicy, epoch ports.BrokerEpoch, connector ports.BrokerEndpointConnector, loadIdentity broker.LocalIdentityLoader) (*broker.LocalDaemonProbe, error) {
28 // Probing must never spawn a process, so only a Unix route this process
29 // dials directly is acceptable. brokerconfig refuses any other kind at load
30 // time; this re-checks the exported binding a caller may have built itself,
31 // because the dial seam routes an ssh address to a child process.
32 if !route.IsLocal() {
33 return nil, errors.New("broker local probe requires a local route")
34 }
35 if err := route.Validate(); err != nil {
36 return nil, fmt.Errorf("broker local probe: %w", err)
37 }
38 return broker.NewLocalDaemonProbe(epoch, route.Address(), policy, connector, sessionwire.BrokerCodec{}, loadIdentity)
39}
40
41// brokerLocalProbe builds the broker-owned local probe over one validated
42// configuration and the broker's already owned endpoint connector.
43func brokerLocalProbe(config *brokerconfig.Config, epoch ports.BrokerEpoch, connector ports.BrokerEndpointConnector, loaders ...broker.LocalIdentityLoader) (*broker.LocalDaemonProbe, error) {
44 binding, ok := config.LocalBinding()
45 if !ok {
46 return nil, errors.New("broker local probe requires a provisioned local binding")
47 }
48 loader := broker.LocalIdentityLoader(func() (ports.BrokerDaemonIdentity, error) { return binding.Identity, nil })
49 if len(loaders) != 0 {
50 loader = loaders[0]
51 }
52 return newDynamicLocalRouteProbe(binding.Route, binding.Policy, epoch, connector, loader)
53}
54
55// brokerLocalObservation composes the registry's local producer: the
56// configured display origin and policy become the local entry's authority,
57// and the probe supplies observed state only.
58func brokerLocalObservation(config *brokerconfig.Config, epoch ports.BrokerEpoch, connector ports.BrokerEndpointConnector, loaders ...broker.LocalIdentityLoader) (*broker.LocalObservation, error) {
59 binding, ok := config.LocalBinding()
60 if !ok {
61 return nil, errors.New("broker local observation requires a provisioned local binding")
62 }
63 probe, err := brokerLocalProbe(config, epoch, connector, loaders...)
64 if err != nil {
65 return nil, err
66 }
67 return &broker.LocalObservation{DisplayOrigin: binding.DisplayOrigin, Policy: binding.Policy, Probe: probe}, nil
68}