adds connection id verification to hook

backend/internal/cmds/mgmt/serve/connect/cmd.go

@@ -479,9 +479,9 @@ func serve(ctx context.Context) error {
 	)
 
 	jobhookOpts := []jobhooks.Option{}
-	if getIsNeosyncCloud() || eelicense.IsValid() {
-		jobhookOpts = append(jobhookOpts, jobhooks.WithEnabled())
-	}
+	// if getIsNeosyncCloud() || eelicense.IsValid() {
+	jobhookOpts = append(jobhookOpts, jobhooks.WithEnabled())
+	// }
 
 	jobhookService := jobhooks.New(
 		db,

backend/internal/dtomaps/job-hooks.go

@@ -18,18 +18,11 @@ func ToJobHookDto(
 		priority = uint32(input.Priority)
 	}
 
-	var config *mgmtv1alpha1.JobHookConfig
+	config := &mgmtv1alpha1.JobHookConfig{}
 	err := config.UnmarshalJSON(input.Config)
 	if err != nil {
 		return nil, err
 	}
-	// if input.Config != nil {
-	// 	var err error
-	// 	config, err = input.Config.ToDto()
-	// 	if err != nil {
-	// 		return nil, err
-	// 	}
-	// }
 
 	output := &mgmtv1alpha1.JobHook{
 		Id:              neosyncdb.UUIDString(input.ID),

backend/internal/ee/hooks/jobs/service.go

@@ -2,6 +2,7 @@ package jobhooks
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 
@@ -236,10 +237,22 @@ func (s *Service) CreateJobHook(
 		return nil, err
 	}
 
-	// todo: verify all connections are within the account as well as the job
 	hookReq := req.GetHook()
 	logger.Debug(fmt.Sprintf("attempting to create new job hook %q", hookReq.GetName()))
 
+	isValid, err := s.verifyHookHasValidConnections(
+		ctx,
+		hookReq.GetConfig(),
+		jobuuid,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("unable to validate if job hook has valid connections: %w", err)
+	}
+	if !isValid {
+		logger.Debug("job hook creation did not pass connection id verification")
+		return nil, nucleuserrors.NewBadRequest("connection id specified in hook is not a part of job")
+	}
+
 	config, err := hookReq.GetConfig().MarshalJSON()
 	if err != nil {
 		return nil, fmt.Errorf("unable to map config to valid json for db storage: %w", err)
@@ -263,6 +276,7 @@ func (s *Service) CreateJobHook(
 	if err != nil {
 		return nil, fmt.Errorf("unable to create job hook: %w", err)
 	}
+	logger.Debug("created job hook")
 
 	dto, err := dtomaps.ToJobHookDto(&hook)
 	if err != nil {
@@ -304,3 +318,26 @@ func safeInt32(v uint32) (int32, error) {
 	}
 	return int32(v), nil //nolint:gosec // safe due to check above
 }
+
+func (s *Service) verifyHookHasValidConnections(
+	ctx context.Context,
+	config *mgmtv1alpha1.JobHookConfig,
+	jobuuid pgtype.UUID,
+) (bool, error) {
+	switch cfg := config.GetConfig().(type) {
+	case *mgmtv1alpha1.JobHookConfig_Sql:
+		if cfg.Sql == nil {
+			return false, errors.New("job hook config was type Sql, but the config was nil")
+		}
+		connuuid, err := neosyncdb.ToUuid(cfg.Sql.GetConnectionId())
+		if err != nil {
+			return false, err
+		}
+		return s.db.Q.DoesJobHaveConnectionId(ctx, s.db.Db, db_queries.DoesJobHaveConnectionIdParams{
+			JobId:        jobuuid,
+			ConnectionId: connuuid,
+		})
+	default:
+		return false, fmt.Errorf("job hook config %T is not currently supported when checking valid connections", cfg)
+	}
+}

backend/sql/postgresql/queries/jobs.sql

@@ -127,3 +127,45 @@ SELECT account_id
 FROM neosync_api.jobs
 WHERE id = $1
 LIMIT 1;
+
+-- name: DoesJobHaveConnectionId :one
+SELECT EXISTS (
+    SELECT 1
+    FROM (
+        -- Check direct associations in the job_destination_connection_associations table
+        SELECT connection_id
+        FROM neosync_api.job_destination_connection_associations
+        WHERE job_id = sqlc.arg('jobId')
+            AND connection_id = sqlc.arg('connectionId')
+
+        UNION
+
+        -- Check connection IDs embedded in the jobs table connection_options
+        SELECT connection_id
+        FROM (
+            SELECT CASE
+                -- Generate options FK source connection
+                WHEN connection_options->'generateOptions'->>'fkSourceConnectionId' IS NOT NULL THEN
+                    (connection_options->'generateOptions'->>'fkSourceConnectionId')::uuid
+                -- Postgres connection
+                WHEN connection_options->'postgresOptions'->>'connectionId' IS NOT NULL THEN
+                    (connection_options->'postgresOptions'->>'connectionId')::uuid
+                -- MSSQL connection
+                WHEN connection_options->'mssqlOptions'->>'connectionId' IS NOT NULL THEN
+                    (connection_options->'mssqlOptions'->>'connectionId')::uuid
+                -- MySQL connection
+                WHEN connection_options->'mysqlOptions'->>'connectionId' IS NOT NULL THEN
+                    (connection_options->'mysqlOptions'->>'connectionId')::uuid
+                -- Mongo connection
+                WHEN connection_options->'mongoOptions'->>'connectionId' IS NOT NULL THEN
+                    (connection_options->'mongoOptions'->>'connectionId')::uuid
+                -- DynamoDB connection
+                WHEN connection_options->'dynamoDBOptions'->>'connectionId' IS NOT NULL THEN
+                    (connection_options->'dynamoDBOptions'->>'connectionId')::uuid
+            END AS connection_id
+            FROM neosync_api.jobs
+            WHERE id = sqlc.arg('jobId')
+        ) embedded_connections
+        WHERE connection_id = sqlc.arg('connectionId')
+    ) all_connections
+);

backend/sql/postgresql/schema/20241121225729_add-job-hooks.up.sql

@@ -17,13 +17,25 @@ CREATE TABLE IF NOT EXISTS neosync_api.job_hooks (
 
   hook_timing text GENERATED ALWAYS AS (
     CASE
-      WHEN config->'sql'->>'pre_sync' IS NOT NULL THEN 'pre_sync'
-      WHEN config->'sql'->>'post_sync' IS NOT NULL THEN 'post_sync'
+      WHEN config->'sql'->'timing'->>'preSync' IS NOT NULL THEN 'preSync'
+      WHEN config->'sql'->'timing'->>'postSync' IS NOT NULL THEN 'postSync'
       ELSE NULL
     END
   ) STORED,
   CONSTRAINT hook_timing_not_null CHECK (hook_timing IS NOT NULL), -- Ensures we always have valid hook timings
 
+  connection_id uuid GENERATED ALWAYS AS (
+    CASE
+      WHEN config ? 'sql' THEN (config->'sql'->>'connectionId')::uuid
+      ELSE NULL
+    END
+  ) STORED,
+
+  CONSTRAINT fk_job_hooks_connection
+    FOREIGN KEY (connection_id)
+    REFERENCES neosync_api.connections(id)
+    ON DELETE RESTRICT,
+
   CONSTRAINT fk_job_hooks_job
     FOREIGN KEY (job_id)
     REFERENCES neosync_api.jobs(id)