SY-4331: Move Channel Functionality to the Service Layer

core/cmd/start/start.go

@@ -201,24 +201,24 @@ func BootupCore(ctx context.Context, onServerStarted chan struct{}, cfgs ...Core
 	)
 
 	if distributionLayer, err = distribution.OpenLayer(ctx, distribution.LayerConfig{
-		Instrumentation:      cfg.Child("distribution"),
-		AdvertiseAddress:     cfg.listenAddress,
-		PeerAddresses:        cfg.peers,
-		AspenTransport:       aspenTransport,
-		Transport:            distTransport,
-		Verifier:             cfg.verifier,
-		Storage:              storageLayer,
-		ValidateChannelNames: cfg.validateChannelNames,
+		Instrumentation:  cfg.Child("distribution"),
+		AdvertiseAddress: cfg.listenAddress,
+		PeerAddresses:    cfg.peers,
+		AspenTransport:   aspenTransport,
+		Transport:        distTransport,
+		Storage:          storageLayer,
 	}); !ok(err, distributionLayer) {
 		return err
 	}
 
 	if serviceLayer, err = service.OpenLayer(ctx, service.LayerConfig{
-		Instrumentation: cfg.Child("service"),
-		Distribution:    distributionLayer,
-		Security:        securityProvider,
-		Storage:         storageLayer,
-		RootCredentials: cfg.rootCredentials,
+		Instrumentation:      cfg.Child("service"),
+		Distribution:         distributionLayer,
+		Security:             securityProvider,
+		Storage:              storageLayer,
+		RootCredentials:      cfg.rootCredentials,
+		Verifier:             cfg.verifier,
+		ValidateChannelNames: cfg.validateChannelNames,
 	}); !ok(err, serviceLayer) {
 		return err
 	}
@@ -243,8 +243,8 @@ func BootupCore(ctx context.Context, onServerStarted chan struct{}, cfgs ...Core
 	if transportLayer, err = transport.NewLayer(transport.LayerConfig{
 		Instrumentation: cfg.Child("transport"),
 		API:             apiLayer,
+		Channel:         serviceLayer.Channel,
 		Router:          r,
-		Channel:         distributionLayer.Channel,
 	}); !ok(err, nil) {
 		return err
 	}

core/pkg/api/arc/arc_suite_test.go

@@ -24,7 +24,7 @@ import (
 	"github.com/synnaxlabs/synnax/pkg/service/access/rbac"
 	"github.com/synnaxlabs/synnax/pkg/service/access/rbac/policy"
 	"github.com/synnaxlabs/synnax/pkg/service/access/rbac/role"
-	arc "github.com/synnaxlabs/synnax/pkg/service/arc"
+	"github.com/synnaxlabs/synnax/pkg/service/arc"
 	"github.com/synnaxlabs/synnax/pkg/service/auth"
 	"github.com/synnaxlabs/synnax/pkg/service/channel"
 	"github.com/synnaxlabs/synnax/pkg/service/label"
@@ -77,6 +77,16 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 		Label:    labelSvc,
 		Search:   searchIdx,
 	}))
+	channelSvc := MustOpen(channel.OpenService(ctx, channel.ServiceConfig{
+		Channel:      node.Channel,
+		DB:           node.DB,
+		HostResolver: node.Cluster,
+		Ontology:     node.Ontology,
+		Group:        node.Group,
+		Status:       statSvc,
+		Search:       node.Search,
+	}))
+
 	rackSvc := MustOpen(rack.OpenService(ctx, rack.ServiceConfig{
 		DB:                  db,
 		Ontology:            otg,
@@ -97,7 +107,7 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 	arcSvc = MustOpen(arc.OpenService(ctx, arc.ServiceConfig{
 		DB:       db,
 		Ontology: otg,
-		Channel:  channel.Wrap(node.Channel),
+		Channel:  channelSvc,
 		Task:     taskSvc,
 		Search:   searchIdx,
 	}))

core/pkg/api/channel/BUILD.bazel

@@ -6,7 +6,7 @@ proto_library(
     srcs = ["pb/channel.proto"],
     visibility = ["//visibility:public"],
     deps = [
-        "//core/pkg/distribution/channel:channel_proto",
+        "//core/pkg/service/channel:channel_proto",
         "//core/pkg/service/status:status_proto",
         "//x/go/control:control_proto",
     ],
@@ -18,7 +18,7 @@ cpp_grpc_library(
     protos = [":channel_proto"],
     visibility = ["//visibility:public"],
     deps = [
-        "//core/pkg/distribution/channel:channel_grpc",
+        "//core/pkg/service/channel:channel_grpc",
         "//core/pkg/service/status:status_grpc",
         "//x/go/control:control_grpc",
     ],

core/pkg/api/channel/channel.go

@@ -329,7 +329,7 @@ func (s *Service) Delete(
 		if err := s.access.NewEnforcer(tx).Enforce(ctx, access.Request{
 			Subject: auth.GetSubject(ctx),
 			Action:  access.ActionDelete,
-			Objects: req.Keys.OntologyIDs(),
+			Objects: channel.OntologyIDsFromKeys(req.Keys),
 		}); err != nil {
 			return types.Nil{}, err
 		}
@@ -370,7 +370,7 @@ func (s *Service) Rename(
 	if err := s.access.NewEnforcer(tx).Enforce(ctx, access.Request{
 		Subject: auth.GetSubject(ctx),
 		Action:  access.ActionUpdate,
-		Objects: req.Keys.OntologyIDs(),
+		Objects: channel.OntologyIDsFromKeys(req.Keys),
 	}); err != nil {
 		return types.Nil{}, err
 	}

core/pkg/api/channel/pb/channel.proto

@@ -13,7 +13,7 @@ syntax = "proto3";
 
 package api.channel.pb;
 
-import "core/pkg/distribution/channel/pb/channel.proto";
+import "core/pkg/service/channel/pb/channel.proto";
 import "core/pkg/service/status/pb/status.proto";
 import "x/go/control/pb/control.proto";
 
@@ -52,7 +52,7 @@ message Channel {
   string expression = 10;
   // operations contains optional aggregation operations (min, max, avg) applied to
   // channel data over time or triggered by a reset channel.
-  repeated .distribution.channel.pb.Operation operations = 11;
+  repeated .service.channel.pb.Operation operations = 11;
   // concurrency sets the policy for concurrent writes to the channel's data. Only virtual
   // channels can have a policy of shared concurrency.
   .x.control.pb.Concurrency concurrency = 12;