Refactor code to avoid cyclic references

generate.go

@@ -1,6 +1,7 @@
 package main
 
-//go:generate mockgen  -destination=./tests/mocks/stream_reconciler_factory.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers/stream_class StreamReconcilerFactory
-//go:generate mockgen  -destination=./tests/mocks/cache_provider.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers/stream_class CacheProvider
-//go:generate mockgen  -destination=./tests/mocks/stream_reconciler.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers StreamReconciler
+//go:generate mockgen -destination=./tests/mocks/stream_reconciler_factory.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers/stream_class StreamReconcilerFactory
+//go:generate mockgen -destination=./tests/mocks/cache_provider.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers/stream_class CacheProvider
+//go:generate mockgen -destination=./tests/mocks/stream_reconciler.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers StreamReconciler
 //go:generate mockgen -destination=./tests/mocks/controller.go -package=mocks sigs.k8s.io/controller-runtime/pkg/controller Controller
+//go:generate mockgen -destination=./tests/mocks/job_builder.go -package=mocks github.com/SneaksAndData/arcane-operator/services/controllers/stream JobBuilder

services/controllers/stream/basic_job_configurator.go

@@ -1,15 +1,20 @@
 package stream
 
 import (
+	"github.com/SneaksAndData/arcane-operator/services"
 	batchv1 "k8s.io/api/batch/v1"
 )
 
-var _ JobConfigurator = &BasicJobConfigurator{}
+var _ services.JobConfigurator = &BasicJobConfigurator{}
 
 type BasicJobConfigurator struct {
 	definition Definition
 }
 
+func (f BasicJobConfigurator) AddNext(configurator *services.JobConfigurator) services.JobConfigurator {
+	panic("not implemented")
+}
+
 func (f BasicJobConfigurator) ConfigureJob(job *batchv1.Job) error {
 	panic("not implemented")
 }

services/controllers/stream/job_builder.go

@@ -0,0 +1,11 @@
+package stream
+
+import (
+	"context"
+	"github.com/SneaksAndData/arcane-operator/services"
+	batchv1 "k8s.io/api/batch/v1"
+)
+
+type JobBuilder interface {
+	BuildJob(ctx context.Context, jobType services.JobTemplateType, configurator services.JobConfigurator) (*batchv1.Job, error)
+}

services/controllers/stream/stream_definition.go

@@ -55,9 +55,6 @@ type Definition interface {
 
 	// ToOwnerReference converts the stream definition to an owner reference.
 	ToOwnerReference() v1.OwnerReference
-
-	// JobConfigurator returns a JobConfigurator for the stream definition.
-	JobConfigurator() JobConfigurator
 }
 
 func fromUnstructured(obj *unstructured.Unstructured) (Definition, error) {

services/controllers/stream/unstructured_wrapper.go

@@ -5,14 +5,18 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"github.com/SneaksAndData/arcane-operator/services"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 )
 
-var _ Definition = (*unstructuredWrapper)(nil)
+var (
+	_ Definition                       = (*unstructuredWrapper)(nil)
+	_ services.JobConfiguratorProvider = (*unstructuredWrapper)(nil)
+)
 
 type unstructuredWrapper struct {
 	underlying      *unstructured.Unstructured
@@ -116,7 +120,7 @@ func (u *unstructuredWrapper) ToOwnerReference() metav1.OwnerReference {
 	}
 }
 
-func (u *unstructuredWrapper) JobConfigurator() JobConfigurator {
+func (u *unstructuredWrapper) JobConfigurator() services.JobConfigurator {
 	return NewFromStreamDefinition(u)
 }
 

services/job_configurator.go

@@ -0,0 +1,22 @@
+package services
+
+import batchv1 "k8s.io/api/batch/v1"
+
+type JobTemplateType string
+
+const (
+	BackfillJobTemplate  JobTemplateType = "backfill"
+	StreamingJobTemplate JobTemplateType = "streaming"
+)
+
+// JobConfigurator defines an interface for configuring Kubernetes Jobs. Each implementer
+// can modify the Job object and chain to the next configurator in the sequence.
+type JobConfigurator interface {
+
+	// ConfigureJob modifies the provided Job object according to the configurator's logic.
+	ConfigureJob(job *batchv1.Job) error
+
+	// AddNext sets the next JobConfigurator in the chain.
+	// Returns self to allow for method chaining.
+	AddNext(configurator *JobConfigurator) JobConfigurator
+}

services/job_configurator_provider.go

@@ -0,0 +1,8 @@
+package services
+
+// JobConfiguratorProvider defines an interface for types that can provide a JobConfigurator.
+type JobConfiguratorProvider interface {
+
+	// JobConfigurator returns a JobConfigurator for the current instance.
+	JobConfigurator() JobConfigurator
+}