1package broker
2
3import (
4 "context"
5 "errors"
6
7 "github.com/bnema/vev/internal/ports"
8)
9
10// RouteAddress derives the opaque pool address of one durable route. The
11// carriage adapter owns the derivation; routing only compares its results.
12type RouteAddress func(ports.BrokerRouteSpec) (string, error)
13
14// Routes is the broker's single route authorization policy. Local routing is
15// provisioned by composition; remote routing is read from committed registry
16// membership, never from an advisory snapshot or a startup import. A remote
17// target is authorized twice against the same committed record: once when it is
18// resolved, and again immediately before dialing, so a removed, replaced,
19// rebound, or rerouted host can never authorize a stale target.
20//
21// Hosts may be assigned after the registry is built, but before it runs.
22type Routes struct {
23 Local ports.BrokerRouteAuthority
24 Hosts ports.BrokerHostAuthorityReader
25 Address RouteAddress
26}
27
28var _ ports.BrokerRouteAuthority = (*Routes)(nil)
29
30var errRouteAuthorityChanged = ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "route authority changed"}
31
32func (r *Routes) ResolveDialTarget(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
33 if err := ctx.Err(); err != nil {
34 return ports.BrokerDialTarget{}, err
35 }
36 if request.Local {
37 if r.Local == nil {
38 return ports.BrokerDialTarget{}, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "no local route"}
39 }
40 return r.Local.ResolveDialTarget(ctx, request)
41 }
42 if r.Hosts == nil || r.Address == nil {
43 return ports.BrokerDialTarget{}, errors.New("broker: incomplete route authority")
44 }
45 record, found, err := r.Hosts.LookupHost(ctx, request.Endpoint)
46 if err != nil {
47 return ports.BrokerDialTarget{}, err
48 }
49 if !found || !record.Registration.Equal(request.Registration) {
50 return ports.BrokerDialTarget{}, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "unknown or stale registration"}
51 }
52 if !record.Policy.Compatible(request.Policy) {
53 return ports.BrokerDialTarget{}, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}
54 }
55 address, err := r.Address(record.Route)
56 if err != nil {
57 return ports.BrokerDialTarget{}, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
58 }
59 target := ports.BrokerDialTarget{
60 Fence: ports.BrokerEndpointFence{Registration: record.Registration},
61 Policy: record.Policy, Address: address, StartMode: request.StartMode,
62 ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: record.Identity, Bound: record.Identity != ""},
63 }
64 return target, target.Validate()
65}
66
67// AuthorizeRemoteDial rechecks a resolved remote target against the committed
68// record immediately before dialing and returns the route to dial. Any change
69// to registration, policy, identity, or route since resolution refuses it.
70func (r *Routes) AuthorizeRemoteDial(ctx context.Context, target ports.BrokerDialTarget) (ports.BrokerRouteSpec, error) {
71 if target.Fence.Local {
72 return ports.BrokerRouteSpec{}, errors.New("broker: local target has no remote route")
73 }
74 if r.Hosts == nil || r.Address == nil {
75 return ports.BrokerRouteSpec{}, errors.New("broker: incomplete route authority")
76 }
77 record, found, err := r.Hosts.LookupHost(ctx, target.Fence.Registration.Endpoint)
78 if err != nil {
79 return ports.BrokerRouteSpec{}, err
80 }
81 if !found || !record.Registration.Equal(target.Fence.Registration) || record.Policy != target.Policy || record.Identity != target.ExpectedIdentity.Identity {
82 return ports.BrokerRouteSpec{}, errRouteAuthorityChanged
83 }
84 address, err := r.Address(record.Route)
85 if err != nil {
86 return ports.BrokerRouteSpec{}, err
87 }
88 if address != target.Address {
89 return ports.BrokerRouteSpec{}, errRouteAuthorityChanged
90 }
91 return record.Route, nil
92}