big ol move around

cli.go

@@ -10,8 +10,7 @@ import (
 	"slices"
 	"strings"
 
-	dlyrs "github.com/orca-telemetry/core/internal/datalayers"
-	envs "github.com/orca-telemetry/core/internal/envs"
+	"github.com/orca-telemetry/core/internal"
 )
 
 type cliFlags struct {
@@ -138,7 +137,7 @@ func validateFlags(flags cliFlags) error {
 	return nil
 }
 
-func validateConfig(config *envs.Config) error {
+func validateConfig(config *internal.Config) error {
 	if config.Platform == "" {
 		return fmt.Errorf("platform cannot be determined from connection string")
 	}
@@ -175,8 +174,8 @@ func runCLI(flags cliFlags) {
 		return
 	}
 
-	// get singleton  configuration
-	config := envs.GetConfig()
+	// get singleton configuration
+	config := internal.GetConfig()
 
 	// validate configuration
 	if err := validateConfig(config); err != nil {
@@ -201,7 +200,7 @@ func runCLI(flags cliFlags) {
 	slog.Info("premigration")
 	if flags.migrate {
 		slog.Info("migrating datalayer", "platform", config.Platform)
-		err := dlyrs.MigrateDatalayer(config.Platform, config.ConnectionString)
+		err := MigrateDatalayer(config.Platform, config.ConnectionString)
 		if err != nil {
 			slog.Error("could not migrate the datalayer, exiting", "error", err)
 			os.Exit(1)

internal/datalayers/export_test.go → export_test.go

@@ -1,7 +1,7 @@
 // Bridge package that implements useful mocking functionality for
 // Orca processors
 
-package datalayers
+package main
 
 import (
 	"context"
@@ -27,7 +27,7 @@ func (s *mockOrcaProcessorServer) ExecuteDagPart(req *pb.ExecutionRequest, strea
 	slog.Debug("Received ExecuteDagPart request", "exec_id", req.GetExecId())
 
 	// simulate processing each algorithm in the request
-	for i, execution := range req.GetAlgorithmExecutions() {
+	for _, execution := range req.GetAlgorithmExecutions() {
 
 		// create a mock result for this algorithm
 		result := &pb.ExecutionResult{
@@ -49,7 +49,6 @@ func (s *mockOrcaProcessorServer) ExecuteDagPart(req *pb.ExecutionRequest, strea
 			return status.Errorf(codes.Internal, "failed to send result: %v", err)
 		}
 
-		slog.Debug("sent result for algorithm", "result_num", i+1, "algorithm_num", len(req.GetAlgorithms()), "algorithm_name", execution.GetAlgorithm().GetName())
 	}
 
 	slog.Debug("completed ExecuteDagPart", "exec_id", req.GetExecId())

internal/types/main.go → internal/db/errors.go

@@ -1,26 +1,7 @@
-package datalayers
+package db
 
 import (
-	"context"
 	"fmt"
-
-	pb "github.com/orca-telemetry/contract/go"
-)
-
-// the interface that all datalayers must implement to be compatible with Orca
-type (
-	Tx interface {
-		Rollback(ctx context.Context)
-		Commit(ctx context.Context) error
-	}
-	Datalayer interface {
-		WithTx(ctx context.Context) (Tx, error)
-
-		// Core level operations
-		RegisterProcessor(ctx context.Context, proc *pb.ProcessorRegistration) error
-		EmitWindow(ctx context.Context, window *pb.Window) (pb.WindowEmitStatus, error)
-		Expose(ctx context.Context, settings *pb.ExposeSettings) (*pb.InternalState, error)
-	}
 )
 
 // custom errors

internal/datalayers/postgresql/funcs.go → internal/db/funcs.go

@@ -1,4 +1,4 @@
-package postgresql
+package db
 
 import (
 	"context"
@@ -13,7 +13,6 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/jackc/pgx/v5/pgxpool"
 	pb "github.com/orca-telemetry/contract/go"
-	types "github.com/orca-telemetry/core/internal/types"
 )
 
 type Datalayer struct {
@@ -53,23 +52,13 @@ func NewClient(ctx context.Context, connStr string) (*Datalayer, error) {
 	}, nil
 }
 
-func (d *Datalayer) WithTx(ctx context.Context) (types.Tx, error) {
-	tx, err := d.conn.Begin(ctx)
-	if err != nil {
-		slog.Error("could not start transaction", "error", err)
-		return nil, err
-	}
-	return &PgTx{tx: tx}, nil
-}
-
 func (d *Datalayer) createProcessor(
 	ctx context.Context,
-	tx types.Tx,
+	tx pgx.Tx,
 	proc *pb.ProcessorRegistration,
 ) error {
-	pgTx := tx.(*PgTx)
 
-	qtx := d.queries.WithTx(pgTx.tx)
+	qtx := d.queries.WithTx(tx)
 
 	err := qtx.CreateProcessor(ctx, CreateProcessorParams{
 		Name:             proc.GetName(),
@@ -86,11 +75,10 @@ func (d *Datalayer) createProcessor(
 
 func (d *Datalayer) createMetadataField(
 	ctx context.Context,
-	tx types.Tx,
+	tx pgx.Tx,
 	metadataField *pb.MetadataField,
 ) (int64, error) {
-	pgTx := tx.(*PgTx)
-	qtx := d.queries.WithTx(pgTx.tx)
+	qtx := d.queries.WithTx(tx)
 	metadataFieldId, err := qtx.CreateMetadataField(ctx, CreateMetadataFieldParams{
 		Name:        metadataField.GetName(),
 		Description: metadataField.GetDescription(),
@@ -108,11 +96,10 @@ func (d *Datalayer) createMetadataField(
 
 func (d *Datalayer) readMetadataFieldsByWindowType(
 	ctx context.Context,
-	tx types.Tx,
+	tx pgx.Tx,
 	windowTypeId int64,
 ) ([]*pb.MetadataField, error) {
-	pgTx := tx.(*PgTx)
-	qtx := d.queries.WithTx(pgTx.tx)
+	qtx := d.queries.WithTx(tx)
 
 	metadataFields, err := qtx.ReadMetadataFieldsByWindowType(ctx, windowTypeId)
 
@@ -133,11 +120,10 @@ func (d *Datalayer) readMetadataFieldsByWindowType(
 
 func (d *Datalayer) createWindowType(
 	ctx context.Context,
-	tx types.Tx,
+	tx pgx.Tx,
 	windowType *pb.WindowType,
 ) (int64, error) {
-	pgTx := tx.(*PgTx)
-	qtx := d.queries.WithTx(pgTx.tx)
+	qtx := d.queries.WithTx(tx)
 	windowTypeId, err := qtx.CreateWindowType(ctx, CreateWindowTypeParams{
 		Name:        windowType.GetName(),
 		Version:     windowType.GetVersion(),
@@ -152,12 +138,11 @@ func (d *Datalayer) createWindowType(
 
 func (d *Datalayer) createMetadataFieldBridge(
 	ctx context.Context,
-	tx types.Tx,
+	tx pgx.Tx,
 	windowTypeId int64,
 	metadataFieldId int64,
 ) error {
-	pgTx := tx.(*PgTx)
-	qtx := d.queries.WithTx(pgTx.tx)
+	qtx := d.queries.WithTx(tx)
 	err := qtx.CreateWindowTypeMetadataFieldBridge(ctx, CreateWindowTypeMetadataFieldBridgeParams{
 		WindowTypeID:     windowTypeId,
 		MetadataFieldsID: metadataFieldId,
@@ -171,12 +156,11 @@ func (d *Datalayer) createMetadataFieldBridge(
 
 func (d *Datalayer) addAlgorithm(
 	ctx context.Context,
-	tx types.Tx,
+	tx pgx.Tx,
 	algo *pb.Algorithm,
 	proc *pb.ProcessorRegistration,
 ) error {
-	pgTx := tx.(*PgTx)
-	qtx := d.queries.WithTx(pgTx.tx)
+	qtx := d.queries.WithTx(tx)
 
 	// create algos
 	var resultType ResultType
@@ -218,12 +202,12 @@ func (d *Datalayer) addAlgorithm(
 
 func (d *Datalayer) addOverwriteAlgorithmDependency(
 	ctx context.Context,
-	tx types.Tx,
+	tx pgx.Tx,
 	algo *pb.Algorithm,
 	proc *pb.ProcessorRegistration,
 ) error {
-	pgTx := tx.(*PgTx)
-	qtx := d.queries.WithTx(pgTx.tx)
+	qtx := d.queries.WithTx(tx)
+
 	// get algorithm id
 	algoId, err := qtx.ReadAlgorithmId(ctx, ReadAlgorithmIdParams{
 		AlgorithmName:    algo.GetName(),
@@ -264,7 +248,7 @@ func (d *Datalayer) addOverwriteAlgorithmDependency(
 					"to_algo",
 					algo,
 				)
-				return &types.CircularDependencyError{
+				return &CircularDependencyError{
 					FromAlgoName:      algoDependentOn.GetName(),
 					FromAlgoVersion:   algoDependentOn.GetVersion(),
 					FromAlgoProcessor: algoDependentOn.GetProcessorName(),