feat(agents): publish agent.updated

.github/workflows/ci.yml

@@ -25,7 +25,7 @@ jobs:
           version: 1.66.0
 
       - name: Generate protobuf bindings
-        run: buf generate buf.build/agynio/api --path agynio/api/agents/v1 --path agynio/api/authorization/v1 --path agynio/api/identity/v1
+        run: buf generate buf.build/agynio/api --path agynio/api/agents/v1 --path agynio/api/authorization/v1 --path agynio/api/identity/v1 --path agynio/api/notifications/v1
 
       - name: Run tests
         run: go test ./...

Dockerfile

@@ -25,7 +25,8 @@ COPY buf.gen.yaml buf.yaml ./
 RUN buf generate buf.build/agynio/api \
     --path agynio/api/agents/v1 \
     --path agynio/api/authorization/v1 \
-    --path agynio/api/identity/v1
+    --path agynio/api/identity/v1 \
+    --path agynio/api/notifications/v1
 
 COPY . .
 

cmd/agents-service/main.go

@@ -12,6 +12,7 @@ import (
 
 	agentsv1 "github.com/agynio/agents/.gen/go/agynio/api/agents/v1"
 	authorizationv1 "github.com/agynio/agents/.gen/go/agynio/api/authorization/v1"
+	notificationsv1 "github.com/agynio/agents/.gen/go/agynio/api/notifications/v1"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -68,8 +69,16 @@ func run() error {
 		_ = identityConn.Close()
 	}()
 	identity := server.NewIdentityWriter(identityConn)
+	notificationsConn, err := grpc.NewClient(cfg.NotificationsServiceAddress, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return fmt.Errorf("connect to notifications service: %w", err)
+	}
+	defer func() {
+		_ = notificationsConn.Close()
+	}()
+	notificationsClient := notificationsv1.NewNotificationsServiceClient(notificationsConn)
 
-	agentsv1.RegisterAgentsServiceServer(grpcServer, server.New(store.New(pool), authzClient, identity))
+	agentsv1.RegisterAgentsServiceServer(grpcServer, server.New(store.New(pool), authzClient, identity, notificationsClient))
 
 	lis, err := net.Listen("tcp", cfg.GRPCAddress)
 	if err != nil {

devspace.yaml

@@ -52,7 +52,7 @@ functions:
                     sleep 1; elapsed=$((elapsed + 1))
                     [ "$elapsed" -ge 120 ] && { echo "ERROR: sync timeout" >&2; exit 1; }
                   done
-                  buf generate buf.build/agynio/api --path agynio/api/agents/v1 --path agynio/api/authorization/v1 --path agynio/api/identity/v1
+                  buf generate buf.build/agynio/api --path agynio/api/agents/v1 --path agynio/api/authorization/v1 --path agynio/api/identity/v1 --path agynio/api/notifications/v1
                   exec go run ./cmd/agents-service
               volumeMounts:
                 - name: data
@@ -135,7 +135,7 @@ pipelines:
       exec_container \
         --label-selector "app.kubernetes.io/name=agents-e2e" \
         -n ${AGENTS_NAMESPACE} \
-        -- bash -c 'cd /opt/app/data && buf generate buf.build/agynio/api --path agynio/api/agents/v1 --path agynio/api/authorization/v1 --path agynio/api/identity/v1 && go test -v -count=1 -tags e2e ./test/e2e/'
+        -- bash -c 'cd /opt/app/data && buf generate buf.build/agynio/api --path agynio/api/agents/v1 --path agynio/api/authorization/v1 --path agynio/api/identity/v1 --path agynio/api/notifications/v1 && go test -v -count=1 -tags e2e ./test/e2e/'
       EXIT_CODE=$?
       stop_dev e2e-runner
       purge_deployments e2e-runner

internal/config/config.go

@@ -10,6 +10,7 @@ type Config struct {
 	DatabaseURL                 string
 	AuthorizationServiceAddress string
 	IdentityServiceAddress      string
+	NotificationsServiceAddress string
 }
 
 func FromEnv() (Config, error) {
@@ -30,5 +31,9 @@ func FromEnv() (Config, error) {
 	if cfg.IdentityServiceAddress == "" {
 		cfg.IdentityServiceAddress = "identity:50051"
 	}
+	cfg.NotificationsServiceAddress = os.Getenv("NOTIFICATIONS_SERVICE_ADDRESS")
+	if cfg.NotificationsServiceAddress == "" {
+		cfg.NotificationsServiceAddress = "notifications:50051"
+	}
 	return cfg, nil
 }

internal/server/notifications.go

@@ -0,0 +1,83 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	notificationsv1 "github.com/agynio/agents/.gen/go/agynio/api/notifications/v1"
+	"github.com/google/uuid"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+const agentUpdatedEvent = "agent.updated"
+
+func (s *Server) publishAgentUpdated(ctx context.Context, agentID uuid.UUID, organizationID uuid.UUID) {
+	payload, err := structpb.NewStruct(map[string]any{
+		"agent_id":        agentID.String(),
+		"organization_id": organizationID.String(),
+	})
+	if err != nil {
+		log.Printf("agents: build agent.updated payload: %v", err)
+		return
+	}
+	_, err = s.notifications.Publish(ctx, &notificationsv1.PublishRequest{
+		Event:   agentUpdatedEvent,
+		Rooms:   []string{fmt.Sprintf("agent:%s", agentID)},
+		Payload: payload,
+		Source:  "agents",
+	})
+	if err != nil {
+		log.Printf("agents: publish agent.updated: %v", err)
+	}
+}
+
+func (s *Server) publishAgentUpdatedByID(ctx context.Context, agentID uuid.UUID) {
+	agent, err := s.store.GetAgent(ctx, agentID)
+	if err != nil {
+		log.Printf("agents: fetch agent for notification: %v", err)
+		return
+	}
+	s.publishAgentUpdated(ctx, agent.Meta.ID, agent.OrganizationID)
+}
+
+func (s *Server) resolveAgentID(ctx context.Context, agentID *uuid.UUID, mcpID *uuid.UUID, hookID *uuid.UUID) (uuid.UUID, error) {
+	if agentID != nil {
+		return *agentID, nil
+	}
+	if mcpID != nil {
+		mcp, err := s.store.GetMcp(ctx, *mcpID)
+		if err != nil {
+			return uuid.UUID{}, err
+		}
+		return mcp.AgentID, nil
+	}
+	if hookID != nil {
+		hook, err := s.store.GetHook(ctx, *hookID)
+		if err != nil {
+			return uuid.UUID{}, err
+		}
+		return hook.AgentID, nil
+	}
+	return uuid.UUID{}, fmt.Errorf("missing target identifier")
+}
+
+func (s *Server) publishAgentUpdatedForVolume(ctx context.Context, volumeID uuid.UUID) {
+	agentIDs, err := s.store.ListAgentIDsForVolume(ctx, volumeID)
+	if err != nil {
+		log.Printf("agents: list volume agents: %v", err)
+		return
+	}
+	for _, agentID := range agentIDs {
+		s.publishAgentUpdatedByID(ctx, agentID)
+	}
+}
+
+func (s *Server) publishAgentUpdatedForTarget(ctx context.Context, agentID *uuid.UUID, mcpID *uuid.UUID, hookID *uuid.UUID) {
+	resolvedID, err := s.resolveAgentID(ctx, agentID, mcpID, hookID)
+	if err != nil {
+		log.Printf("agents: resolve agent for notification: %v", err)
+		return
+	}
+	s.publishAgentUpdatedByID(ctx, resolvedID)
+}