From 46f44896315dc81b0e7ccd17e5df32fe5c794776 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Fri, 9 Oct 2026 19:13:52 +1000 Subject: [PATCH] refactor: route the remaining fire-and-forget vpc.* producers through projection CreateVPC/DeleteVPC, CreateSubnet/DeleteSubnet, AttachInternetGateway/ DetachInternetGateway, AddNATGateway/RemoveNATGateway, AddIGWRoute/ RemoveIGWRoute, GateSubnetEgress/UngateSubnetEgress and RemoveSystemEgress were still publishing through natsmsg.PublishEvent or a raw nc.Publish call with a bare subject string, left over from the mechanical slice 3 pass. Each now goes through its named projection.Client method instead, so every vpc.* producer in the tree uses the typed client. --- spinifex/domains/ec2/igw/service_impl.go | 27 +++--------- spinifex/domains/ec2/natgw/service_impl.go | 6 +-- .../domains/ec2/routetable/service_impl.go | 43 +++++++------------ spinifex/domains/ec2/vpc/service_impl.go | 18 ++++++-- .../domains/network/systemvpc/systemvpc.go | 4 +- spinifex/handlers/eks/service_impl.go | 4 +- 6 files changed, 44 insertions(+), 58 deletions(-) diff --git a/spinifex/domains/ec2/igw/service_impl.go b/spinifex/domains/ec2/igw/service_impl.go index d4cec7cd4..302faf069 100644 --- a/spinifex/domains/ec2/igw/service_impl.go +++ b/spinifex/domains/ec2/igw/service_impl.go @@ -15,6 +15,7 @@ import ( networkv1 "github.com/mulgadc/spinifex/contracts/network/v1" "github.com/mulgadc/spinifex/spinifex/bootstrap/config" ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/domains/network/projection" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" @@ -412,18 +413,10 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D } // Publish event for vpcd to clean up OVN external switch + gateway + NAT - if s.natsConn != nil { - event := networkv1.InternetGatewayEvent{ - InternetGatewayId: igwID, - VpcId: vpcID, - } - eventData, err := json.Marshal(event) - if err != nil { - slog.WarnContext(ctx, "Failed to marshal IGW detach event", "error", err) - } else if err := s.natsConn.Publish(networkv1.InternetGatewayDetachSubject, eventData); err != nil { - slog.WarnContext(ctx, "Failed to publish IGW detach event", "error", err) - } - } + projection.New(s.natsConn).DetachInternetGateway(networkv1.InternetGatewayEvent{ + InternetGatewayId: igwID, + VpcId: vpcID, + }) // After detach the VPC's LR external gateway and router-wide default // route are removed; any subnet whose effective RT still points at the @@ -440,15 +433,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D // publishAttach asks vpcd to create the OVN external switch, gateway and SNAT. func (s *IGWServiceImpl) publishAttach(ctx context.Context, igwID, vpcID string) { - if s.natsConn == nil { - return - } - eventData, err := json.Marshal(networkv1.InternetGatewayEvent{InternetGatewayId: igwID, VpcId: vpcID}) - if err != nil { - slog.WarnContext(ctx, "Failed to marshal IGW attach event", "error", err) - } else if err := s.natsConn.Publish(networkv1.InternetGatewayAttachSubject, eventData); err != nil { - slog.WarnContext(ctx, "Failed to publish IGW attach event", "error", err) - } + projection.New(s.natsConn).AttachInternetGateway(networkv1.InternetGatewayEvent{InternetGatewayId: igwID, VpcId: vpcID}) } // CreateAttachedInternetGateway creates igwID already attached to vpcID, for a diff --git a/spinifex/domains/ec2/natgw/service_impl.go b/spinifex/domains/ec2/natgw/service_impl.go index 43ea32784..ec6f5a265 100644 --- a/spinifex/domains/ec2/natgw/service_impl.go +++ b/spinifex/domains/ec2/natgw/service_impl.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "github.com/mulgadc/spinifex/spinifex/domains/ec2/tagmirror" - "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats" "log/slog" "strings" "time" @@ -16,6 +15,7 @@ import ( networkv1 "github.com/mulgadc/spinifex/contracts/network/v1" ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip" ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/domains/network/projection" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" @@ -473,7 +473,7 @@ func natgwMatchesFilters(record *NatGatewayRecord, filters map[string][]string) // PublishAddEvent publishes a vpc.add-nat-gateway event for vpcd to create the SNAT rule. // Called by the route table service when CreateRoute targets a NAT GW. func (s *NatGatewayServiceImpl) PublishAddEvent(vpcId, natGatewayId, publicIp, subnetCidr string) { - natsmsg.PublishEvent(s.natsConn, networkv1.NATGatewayAddSubject, networkv1.NATGatewayEvent{ + projection.New(s.natsConn).AddNATGateway(networkv1.NATGatewayEvent{ VpcId: vpcId, NatGatewayId: natGatewayId, PublicIp: publicIp, @@ -483,7 +483,7 @@ func (s *NatGatewayServiceImpl) PublishAddEvent(vpcId, natGatewayId, publicIp, s // PublishDeleteEvent publishes a vpc.delete-nat-gateway event for vpcd to remove the SNAT rule. func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp, subnetCidr string) { - natsmsg.PublishEvent(s.natsConn, networkv1.NATGatewayDeleteSubject, networkv1.NATGatewayEvent{ + projection.New(s.natsConn).RemoveNATGateway(networkv1.NATGatewayEvent{ VpcId: vpcId, NatGatewayId: natGatewayId, PublicIp: publicIp, diff --git a/spinifex/domains/ec2/routetable/service_impl.go b/spinifex/domains/ec2/routetable/service_impl.go index 7c61ccb1e..29cce3bd8 100644 --- a/spinifex/domains/ec2/routetable/service_impl.go +++ b/spinifex/domains/ec2/routetable/service_impl.go @@ -19,6 +19,7 @@ import ( ec2igw "github.com/mulgadc/spinifex/spinifex/domains/ec2/igw" ec2natgw "github.com/mulgadc/spinifex/spinifex/domains/ec2/natgw" ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/domains/network/projection" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" @@ -332,26 +333,18 @@ func (s *RouteTableServiceImpl) publishSubnetEgressGateDecision(ctx context.Cont "vpcId", vpcID, "subnetId", subnetID, "err", err) return } + client := projection.New(s.natsConn) topic := networkv1.SubnetEgressGateSubject - var data []byte if hasEgress { topic = networkv1.SubnetEgressUngateSubject - data, err = json.Marshal(networkv1.SubnetEgressUngateEvent{ + client.UngateSubnetEgress(networkv1.SubnetEgressUngateEvent{ VpcId: vpcID, SubnetId: subnetID, DestinationCidr: destCidr, }) } else { - data, err = json.Marshal(networkv1.SubnetEgressGateEvent{ + client.GateSubnetEgress(networkv1.SubnetEgressGateEvent{ VpcId: vpcID, SubnetId: subnetID, DestinationCidr: destCidr, }) } - if err != nil { - slog.WarnContext(ctx, "subnet egress gate: marshal failed", "topic", topic, "err", err) - return - } - if err := s.natsConn.Publish(topic, data); err != nil { - slog.WarnContext(ctx, "subnet egress gate: publish failed", "topic", topic, "subnetId", subnetID, "err", err) - return - } slog.InfoContext(ctx, "subnet egress gate decision published", "topic", topic, "vpcId", vpcID, "subnetId", subnetID, "destinationCidr", destCidr) } @@ -1464,21 +1457,19 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventForSubnet(ctx context.Cont slog.WarnContext(ctx, "NAT GW event: subnet unmarshal failed", "topic", topic, "subnetId", subnetID, "err", err) return } - data, err := json.Marshal(networkv1.NATGatewayEvent{ + evt := networkv1.NATGatewayEvent{ VpcId: vpcID, NatGatewayId: natgwID, PublicIp: publicIp, SubnetCidr: subnet.CidrBlock, SubnetId: subnetID, DestinationCidr: destCidr, - }) - if err != nil { - slog.WarnContext(ctx, "NAT GW event: marshal failed", "topic", topic, "err", err) - return } - if err := s.natsConn.Publish(topic, data); err != nil { - slog.WarnContext(ctx, "NAT GW event: publish failed", "topic", topic, "subnetId", subnetID, "err", err) - return + client := projection.New(s.natsConn) + if topic == networkv1.NATGatewayDeleteSubject { + client.RemoveNATGateway(evt) + } else { + client.AddNATGateway(evt) } slog.InfoContext(ctx, "NAT GW event published", "topic", topic, "subnetCidr", subnet.CidrBlock, "publicIp", publicIp, "subnetId", subnetID, "destinationCidr", destCidr) } @@ -1533,19 +1524,17 @@ func (s *RouteTableServiceImpl) publishIGWRouteEventForSubnet(ctx context.Contex if s.natsConn == nil { return } - data, err := json.Marshal(networkv1.IGWRouteEvent{ + evt := networkv1.IGWRouteEvent{ VpcId: vpcID, SubnetId: subnetID, DestinationCidr: destCidr, InternetGatewayId: igwID, - }) - if err != nil { - slog.WarnContext(ctx, "IGW route event: marshal failed", "topic", topic, "err", err) - return } - if err := s.natsConn.Publish(topic, data); err != nil { - slog.WarnContext(ctx, "IGW route event: publish failed", "topic", topic, "subnetId", subnetID, "err", err) - return + client := projection.New(s.natsConn) + if topic == networkv1.IGWRouteDeleteSubject { + client.RemoveIGWRoute(evt) + } else { + client.AddIGWRoute(evt) } slog.InfoContext(ctx, "IGW route event published", "topic", topic, "subnetId", subnetID, "destinationCidr", destCidr, "igwId", igwID) } diff --git a/spinifex/domains/ec2/vpc/service_impl.go b/spinifex/domains/ec2/vpc/service_impl.go index 34cef656d..d620459b0 100644 --- a/spinifex/domains/ec2/vpc/service_impl.go +++ b/spinifex/domains/ec2/vpc/service_impl.go @@ -6,7 +6,6 @@ import ( "errors" "fmt" "github.com/mulgadc/spinifex/spinifex/domains/ec2/tagmirror" - "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats" "log/slog" "net" "strconv" @@ -17,6 +16,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" networkv1 "github.com/mulgadc/spinifex/contracts/network/v1" "github.com/mulgadc/spinifex/spinifex/bootstrap/config" + "github.com/mulgadc/spinifex/spinifex/domains/network/projection" "github.com/mulgadc/spinifex/spinifex/foundation/aws/arn" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters" @@ -1661,10 +1661,22 @@ func (s *VPCServiceImpl) getSubnet(ctx context.Context, accountID, subnetId stri // publishVPCEvent publishes a VPC lifecycle event to NATS for vpcd consumption. // This is fire-and-forget; errors are logged but do not fail the API response. func (s *VPCServiceImpl) publishVPCEvent(topic, vpcId, cidrBlock string, vni int64) { - natsmsg.PublishEvent(s.natsConn, topic, networkv1.VPCEvent{VpcId: vpcId, CidrBlock: cidrBlock, VNI: vni}) + evt := networkv1.VPCEvent{VpcId: vpcId, CidrBlock: cidrBlock, VNI: vni} + client := projection.New(s.natsConn) + if topic == networkv1.VPCDeleteSubject { + client.DeleteVPC(evt) + return + } + client.CreateVPC(evt) } // publishSubnetEvent publishes a subnet lifecycle event to NATS for vpcd consumption. func (s *VPCServiceImpl) publishSubnetEvent(topic, subnetId, vpcId, cidrBlock string) { - natsmsg.PublishEvent(s.natsConn, topic, networkv1.SubnetEvent{SubnetId: subnetId, VpcId: vpcId, CidrBlock: cidrBlock}) + evt := networkv1.SubnetEvent{SubnetId: subnetId, VpcId: vpcId, CidrBlock: cidrBlock} + client := projection.New(s.natsConn) + if topic == networkv1.SubnetDeleteSubject { + client.DeleteSubnet(evt) + return + } + client.CreateSubnet(evt) } diff --git a/spinifex/domains/network/systemvpc/systemvpc.go b/spinifex/domains/network/systemvpc/systemvpc.go index 7dac2d77d..9c59f6070 100644 --- a/spinifex/domains/network/systemvpc/systemvpc.go +++ b/spinifex/domains/network/systemvpc/systemvpc.go @@ -16,7 +16,6 @@ import ( "context" "errors" "fmt" - "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats" "hash/fnv" "log/slog" "net/netip" @@ -25,6 +24,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" networkv1 "github.com/mulgadc/spinifex/contracts/network/v1" + "github.com/mulgadc/spinifex/spinifex/domains/network/projection" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/foundation/aws/tags" "github.com/nats-io/nats.go" @@ -678,6 +678,6 @@ func gcTopology(ctx context.Context, nc *nats.Conn, name, vpcID string) { if nc == nil || vpcID == "" { return } - natsmsg.PublishEvent(nc, networkv1.VPCDeleteSubject, networkv1.VPCEvent{VpcId: vpcID}) + projection.New(nc).DeleteVPC(networkv1.VPCEvent{VpcId: vpcID}) slog.InfoContext(ctx, "systemvpc Delete: republished vpc.delete for OVN GC", "name", name, "vpc", vpcID) } diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index 3e223149b..34df3d3f8 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -8,7 +8,6 @@ import ( eksv1 "github.com/mulgadc/spinifex/contracts/eks/v1" networkv1 "github.com/mulgadc/spinifex/contracts/network/v1" awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers" - "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats" "log/slog" "maps" "net" @@ -29,6 +28,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/domains/ec2/systeminstance" "github.com/mulgadc/spinifex/spinifex/domains/eks/access" "github.com/mulgadc/spinifex/spinifex/domains/eks/addon" + "github.com/mulgadc/spinifex/spinifex/domains/network/projection" resourcearn "github.com/mulgadc/spinifex/spinifex/foundation/aws/arn" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency" @@ -1318,7 +1318,7 @@ func (s *EKSServiceImpl) purgeClusterInfra(ctx context.Context, accountID, name if len(meta.ResourcesVpcConfig.SubnetIds) > 0 { subnetID = meta.ResourcesVpcConfig.SubnetIds[0] } - natsmsg.PublishEvent(s.deps.NATSConn, networkv1.SystemEgressDeleteSubject, networkv1.SystemEgressEvent{ + projection.New(s.deps.NATSConn).RemoveSystemEgress(networkv1.SystemEgressEvent{ VpcId: meta.ResourcesVpcConfig.VpcId, SubnetId: subnetID, InstanceIp: meta.ControlPlaneENIIP,