Merge pull request #57 from Prescott-Data/refactor/issue-48-service-layer

Refactor/issue 48 service layer

Author: sangalo20

nexus-broker/cmd/nexus-broker/main.go

@@ -7,6 +7,8 @@ import (
 	"time"
 
 	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/audit"
+	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/repository/postgres"
+	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/service"
 	"github.com/Prescott-Data/nexus-framework/nexus-broker/pkg/caching"
 	"github.com/Prescott-Data/nexus-framework/nexus-broker/pkg/config"
 	"github.com/Prescott-Data/nexus-framework/nexus-broker/pkg/handlers"
@@ -62,25 +64,30 @@ func main() {
 	auditSvc := audit.NewService(db)
 
 	providersHandler := handlers.NewProvidersHandler(store, auditSvc)
+
+	connRepo := postgres.NewConnectionRepository(db)
+	tokenRepo := postgres.NewTokenRepository(db)
+
+	connSvc := service.NewConnectionService(
+		connRepo,
+		tokenRepo,
+		store,
+		auditSvc,
+		cfg.BaseURL,
+		cfg.RedirectPath,
+		cfg.EncryptionKey,
+		cfg.StateKey,
+		cachingClient,
+		cfg.EnforceReturnURL,
+		cfg.AllowedReturnDomains,
+	)
+
 	consentHandler := handlers.NewConsentHandler(handlers.ConsentHandlerConfig{
-		DB:                   db,
-		BaseURL:              cfg.BaseURL,
-		RedirectPath:         cfg.RedirectPath,
-		StateKey:             cfg.StateKey,
-		HTTPClient:           cachingClient,
-		EnforceReturnURL:     cfg.EnforceReturnURL,
-		AllowedReturnDomains: cfg.AllowedReturnDomains,
+		Service: connSvc,
 	})
 	callbackHandler := handlers.NewCallbackHandler(handlers.CallbackHandlerConfig{
-		DB:                   db,
-		Audit:                auditSvc,
-		BaseURL:              cfg.BaseURL,
-		RedirectPath:         cfg.RedirectPath,
-		EncryptionKey:        cfg.EncryptionKey,
-		StateKey:             cfg.StateKey,
-		HTTPClient:           cachingClient,
-		EnforceReturnURL:     cfg.EnforceReturnURL,
-		AllowedReturnDomains: cfg.AllowedReturnDomains,
+		Service: connSvc,
+		Audit:   auditSvc,
 	})
 	auditHandler := handlers.NewAuditHandler(db)
 

nexus-broker/internal/architecture_test.go

@@ -0,0 +1,107 @@
+package architecture_test
+
+import (
+	"go/parser"
+	"go/token"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// Rule defines an architectural constraint
+type Rule struct {
+	Package     string
+	ShouldNot   []string // Substrings of import paths that are forbidden
+	Description string
+}
+
+func TestSeparationOfConcerns(t *testing.T) {
+	modulePrefix := "github.com/Prescott-Data/nexus-framework/nexus-broker"
+
+	rules := []Rule{
+		{
+			Package: "internal/domain",
+			ShouldNot: []string{
+				"net/http",
+				"github.com/jmoiron/sqlx",
+				"github.com/lib/pq",
+				modulePrefix + "/pkg/handlers",
+				modulePrefix + "/internal/service",
+				modulePrefix + "/internal/repository",
+			},
+			Description: "Domain models must be pure and ignorant of HTTP, database drivers, and other layers.",
+		},
+		{
+			Package: "internal/repository",
+			ShouldNot: []string{
+				"net/http",
+				modulePrefix + "/pkg/handlers",
+				modulePrefix + "/internal/service",
+			},
+			Description: "Repositories must not depend on HTTP handlers or business logic services.",
+		},
+		{
+			Package: "internal/service",
+			ShouldNot: []string{
+				modulePrefix + "/pkg/handlers",
+				"github.com/jmoiron/sqlx",
+			},
+			Description: "Services must not depend on HTTP handlers or raw SQL drivers (use repositories instead).",
+		},
+		{
+			Package: "pkg/handlers",
+			ShouldNot: []string{
+				modulePrefix + "/internal/repository",
+			},
+			Description: "HTTP Handlers must not bypass the Service layer to talk directly to Repositories.",
+		},
+	}
+
+	basePath := ".." // We are inside internal, so .. is the broker root
+
+	for _, rule := range rules {
+		t.Run(rule.Package, func(t *testing.T) {
+			targetDir := filepath.Join(basePath, rule.Package)
+			
+			// Walk the target directory
+			err := filepath.Walk(targetDir, func(path string, info os.FileInfo, err error) error {
+				if err != nil {
+					return err
+				}
+				
+				// Only analyze Go files, skip tests
+				if info.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+					return nil
+				}
+
+				fset := token.NewFileSet()
+				node, err := parser.ParseFile(fset, path, nil, parser.ImportsOnly)
+				if err != nil {
+					t.Fatalf("failed to parse %s: %v", path, err)
+				}
+
+				for _, imp := range node.Imports {
+					importPath := strings.Trim(imp.Path.Value, `"`)
+					
+					for _, forbidden := range rule.ShouldNot {
+						if strings.Contains(importPath, forbidden) {
+							t.Errorf("\nViolation in %s\nRule: %s\nForbidden Import Found: %s", 
+								path, rule.Description, importPath)
+						}
+					}
+				}
+				return nil
+			})
+
+			if err != nil {
+				// If the directory doesn't exist yet, we just skip (e.g. if we are running from a different root)
+				if os.IsNotExist(err) {
+					t.Logf("Directory %s does not exist, skipping.", targetDir)
+				} else {
+					t.Fatalf("error walking directory %s: %v", targetDir, err)
+				}
+			}
+		})
+	}
+}

nexus-broker/internal/domain/models.go

@@ -0,0 +1,38 @@
+package domain
+
+import (
+	"database/sql"
+	"encoding/json"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+// Connection represents a user's connection to a provider
+type Connection struct {
+	ID           uuid.UUID
+	WorkspaceID  string
+	ProviderID   uuid.UUID
+	CodeVerifier sql.NullString
+	Scopes       []string
+	ReturnURL    string
+	Status       string
+	ExpiresAt    time.Time
+}
+
+// ConnectionWithProvider joins connection and basic provider info
+type ConnectionWithProvider struct {
+	Connection
+	AuthType         string
+	AuthHeader       string
+	APIBaseURL       string
+	UserInfoEndpoint string
+	ProviderParams   *json.RawMessage
+}
+
+// Token represents an encrypted token at rest
+type Token struct {
+	ConnectionID  uuid.UUID
+	EncryptedData string
+	ExpiresAt     *time.Time
+}

nexus-broker/internal/repository/interfaces.go

@@ -0,0 +1,23 @@
+package repository
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/domain"
+)
+
+// ConnectionRepository handles database operations for connections
+type ConnectionRepository interface {
+	Create(ctx context.Context, conn *domain.Connection) error
+	GetPending(ctx context.Context, id uuid.UUID) (*domain.Connection, error)
+	GetWithProvider(ctx context.Context, id uuid.UUID) (*domain.ConnectionWithProvider, error)
+	GetReturnURL(ctx context.Context, id uuid.UUID) (string, error)
+	UpdateStatus(ctx context.Context, id uuid.UUID, status string) error
+}
+
+// TokenRepository handles database operations for tokens
+type TokenRepository interface {
+	Upsert(ctx context.Context, token *domain.Token) error
+	Get(ctx context.Context, connectionID uuid.UUID) (*domain.Token, error)
+}

nexus-broker/internal/repository/postgres/connection.go

@@ -0,0 +1,68 @@
+package postgres
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
+	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/domain"
+	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/repository"
+)
+
+type connectionRepository struct {
+	db *sqlx.DB
+}
+
+// NewConnectionRepository creates a new Postgres ConnectionRepository
+func NewConnectionRepository(db *sqlx.DB) repository.ConnectionRepository {
+	return &connectionRepository{db: db}
+}
+
+func (r *connectionRepository) Create(ctx context.Context, conn *domain.Connection) error {
+	_, err := r.db.ExecContext(ctx, `
+		INSERT INTO connections (id, workspace_id, provider_id, code_verifier, scopes, return_url, expires_at)
+		VALUES ($1, $2, $3, $4, $5, $6, $7)`,
+		conn.ID, conn.WorkspaceID, conn.ProviderID, conn.CodeVerifier, pq.Array(conn.Scopes), conn.ReturnURL, conn.ExpiresAt)
+	return err
+}
+
+func (r *connectionRepository) GetPending(ctx context.Context, id uuid.UUID) (*domain.Connection, error) {
+	var conn domain.Connection
+	err := r.db.QueryRowContext(ctx, `
+		SELECT id, code_verifier, return_url, provider_id, scopes
+		FROM connections
+		WHERE id = $1 AND status = 'pending' AND expires_at > NOW()`, id).
+		Scan(&conn.ID, &conn.CodeVerifier, &conn.ReturnURL, &conn.ProviderID, pq.Array(&conn.Scopes))
+	if err != nil {
+		return nil, err
+	}
+	return &conn, nil
+}
+
+func (r *connectionRepository) GetWithProvider(ctx context.Context, id uuid.UUID) (*domain.ConnectionWithProvider, error) {
+	var conn domain.ConnectionWithProvider
+	err := r.db.QueryRowContext(ctx, `
+		SELECT c.id, c.provider_id, c.status, c.scopes, c.return_url,
+		       p.auth_type, COALESCE(p.auth_header, ''), COALESCE(p.api_base_url, ''), COALESCE(p.user_info_endpoint, ''), p.params
+		FROM connections c
+		JOIN provider_profiles p ON p.id = c.provider_id
+		WHERE c.id = $1`, id).
+		Scan(&conn.ID, &conn.ProviderID, &conn.Status, pq.Array(&conn.Scopes), &conn.ReturnURL,
+			&conn.AuthType, &conn.AuthHeader, &conn.APIBaseURL, &conn.UserInfoEndpoint, &conn.ProviderParams)
+	if err != nil {
+		return nil, err
+	}
+	return &conn, nil
+}
+
+func (r *connectionRepository) GetReturnURL(ctx context.Context, id uuid.UUID) (string, error) {
+	var returnURL string
+	err := r.db.QueryRowContext(ctx, "SELECT return_url FROM connections WHERE id = $1", id).Scan(&returnURL)
+	return returnURL, err
+}
+
+func (r *connectionRepository) UpdateStatus(ctx context.Context, id uuid.UUID, status string) error {
+	_, err := r.db.ExecContext(ctx, "UPDATE connections SET status = $1, updated_at = NOW() WHERE id = $2", status, id)
+	return err
+}

nexus-broker/internal/repository/postgres/token.go

@@ -0,0 +1,43 @@
+package postgres
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+	"github.com/jmoiron/sqlx"
+	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/domain"
+	"github.com/Prescott-Data/nexus-framework/nexus-broker/internal/repository"
+)
+
+type tokenRepository struct {
+	db *sqlx.DB
+}
+
+// NewTokenRepository creates a new Postgres TokenRepository
+func NewTokenRepository(db *sqlx.DB) repository.TokenRepository {
+	return &tokenRepository{db: db}
+}
+
+func (r *tokenRepository) Upsert(ctx context.Context, token *domain.Token) error {
+	_, err := r.db.ExecContext(ctx, `
+		INSERT INTO tokens (connection_id, encrypted_data, expires_at)
+		VALUES ($1, $2, $3)
+		ON CONFLICT (connection_id)
+		DO UPDATE SET
+			encrypted_data = EXCLUDED.encrypted_data,
+			expires_at     = EXCLUDED.expires_at,
+			created_at     = NOW()`,
+		token.ConnectionID, token.EncryptedData, token.ExpiresAt)
+	return err
+}
+
+func (r *tokenRepository) Get(ctx context.Context, connectionID uuid.UUID) (*domain.Token, error) {
+	var token domain.Token
+	err := r.db.QueryRowContext(ctx, "SELECT encrypted_data, expires_at FROM tokens WHERE connection_id = $1", connectionID).
+		Scan(&token.EncryptedData, &token.ExpiresAt)
+	if err != nil {
+		return nil, err
+	}
+	token.ConnectionID = connectionID
+	return &token, nil
+}