refactor: typesafe encrypted DB columns

backend/controller/dal/async_calls.go

@@ -115,7 +115,7 @@ func (d *DAL) AcquireAsyncCall(ctx context.Context) (call *AsyncCall, err error)
 		return nil, fmt.Errorf("failed to parse origin key %q: %w", row.Origin, err)
 	}
 
-	decryptedRequest, err := d.decrypt(encryption.AsyncSubKey, row.Request)
+	decryptedRequest, err := d.decrypt(&row.Request)
 	if err != nil {
 		return nil, fmt.Errorf("failed to decrypt async call request: %w", err)
 	}
@@ -158,11 +158,12 @@ func (d *DAL) CompleteAsyncCall(ctx context.Context,
 	didScheduleAnotherCall = false
 	switch result := result.(type) {
 	case either.Left[[]byte, string]: // Successful response.
-		encryptedResult, err := d.encrypt(encryption.AsyncSubKey, result.Get())
+		var encryptedResult encryption.EncryptedAsyncColumn
+		err := d.encrypt(result.Get(), &encryptedResult)
 		if err != nil {
 			return false, fmt.Errorf("failed to encrypt async call result: %w", err)
 		}
-		_, err = tx.db.SucceedAsyncCall(ctx, encryptedResult, call.ID)
+		_, err = tx.db.SucceedAsyncCall(ctx, optional.Some(encryptedResult), call.ID)
 		if err != nil {
 			return false, dalerrs.TranslatePGError(err) //nolint:wrapcheck
 		}
@@ -227,7 +228,7 @@ func (d *DAL) LoadAsyncCall(ctx context.Context, id int64) (*AsyncCall, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse origin key %q: %w", row.Origin, err)
 	}
-	request, err := d.decrypt(encryption.AsyncSubKey, row.Request)
+	request, err := d.decrypt(&row.Request)
 	if err != nil {
 		return nil, fmt.Errorf("failed to decrypt async call request: %w", err)
 	}

backend/controller/dal/dal.go

@@ -709,10 +709,11 @@ func (d *DAL) SetDeploymentReplicas(ctx context.Context, key model.DeploymentKey
 			return dalerrs.TranslatePGError(err)
 		}
 	}
-	payload, err := d.encryptJSON(encryption.TimelineSubKey, map[string]interface{}{
+	var payload encryption.EncryptedTimelineColumn
+	err = d.encryptJSON(map[string]interface{}{
 		"prev_min_replicas": deployment.MinReplicas,
 		"min_replicas":      minReplicas,
-	})
+	}, &payload)
 	if err != nil {
 		return fmt.Errorf("failed to encrypt payload for InsertDeploymentUpdatedEvent: %w", err)
 	}
@@ -782,10 +783,11 @@ func (d *DAL) ReplaceDeployment(ctx context.Context, newDeploymentKey model.Depl
 		}
 	}
 
-	payload, err := d.encryptJSON(encryption.TimelineSubKey, map[string]any{
+	var payload encryption.EncryptedTimelineColumn
+	err = d.encryptJSON(map[string]any{
 		"min_replicas": int32(minReplicas),
 		"replaced":     replacedDeploymentKey,
-	})
+	}, &payload)
 	if err != nil {
 		return fmt.Errorf("replace deployment failed to encrypt payload: %w", err)
 	}
@@ -1057,7 +1059,8 @@ func (d *DAL) InsertLogEvent(ctx context.Context, log *LogEvent) error {
 		"error":      log.Error,
 		"stack":      log.Stack,
 	}
-	encryptedPayload, err := d.encryptJSON(encryption.TimelineSubKey, payload)
+	var encryptedPayload encryption.EncryptedTimelineColumn
+	err := d.encryptJSON(payload, &encryptedPayload)
 	if err != nil {
 		return fmt.Errorf("failed to encrypt log payload: %w", err)
 	}
@@ -1137,13 +1140,14 @@ func (d *DAL) InsertCallEvent(ctx context.Context, call *CallEvent) error {
 	if pr, ok := call.ParentRequestKey.Get(); ok {
 		parentRequestKey = optional.Some(pr.String())
 	}
-	payload, err := d.encryptJSON(encryption.TimelineSubKey, map[string]any{
+	var payload encryption.EncryptedTimelineColumn
+	err := d.encryptJSON(map[string]any{
 		"duration_ms": call.Duration.Milliseconds(),
 		"request":     call.Request,
 		"response":    call.Response,
 		"error":       call.Error,
 		"stack":       call.Stack,
-	})
+	}, &payload)
 	if err != nil {
 		return fmt.Errorf("failed to encrypt call payload: %w", err)
 	}

backend/controller/dal/encryption.go

@@ -10,45 +10,45 @@ import (
 	"github.com/TBD54566975/ftl/internal/log"
 )
 
-func (d *DAL) encrypt(subKey encryption.SubKey, cleartext []byte) ([]byte, error) {
+func (d *DAL) encrypt(cleartext []byte, dest encryption.Encryptable) error {
 	if d.encryptor == nil {
-		return nil, fmt.Errorf("encryptor not set")
+		return fmt.Errorf("encryptor not set")
 	}
 
-	v, err := d.encryptor.Encrypt(subKey, cleartext)
+	err := d.encryptor.Encrypt(cleartext, dest)
 	if err != nil {
-		return nil, fmt.Errorf("failed to encrypt binary with subkey %s: %w", subKey, err)
+		return fmt.Errorf("failed to encrypt binary with subkey %s: %w", dest.SubKey(), err)
 	}
 
-	return v, nil
+	return nil
 }
 
-func (d *DAL) decrypt(subKey encryption.SubKey, encrypted []byte) ([]byte, error) {
+func (d *DAL) decrypt(encrypted encryption.Encryptable) ([]byte, error) {
 	if d.encryptor == nil {
 		return nil, fmt.Errorf("encryptor not set")
 	}
 
-	v, err := d.encryptor.Decrypt(subKey, encrypted)
+	v, err := d.encryptor.Decrypt(encrypted)
 	if err != nil {
-		return nil, fmt.Errorf("failed to decrypt binary with subkey %s: %w", subKey, err)
+		return nil, fmt.Errorf("failed to decrypt binary with subkey %s: %w", encrypted.SubKey(), err)
 	}
 
 	return v, nil
 }
 
-func (d *DAL) encryptJSON(subKey encryption.SubKey, v any) ([]byte, error) {
+func (d *DAL) encryptJSON(v any, dest encryption.Encryptable) error {
 	serialized, err := json.Marshal(v)
 	if err != nil {
-		return nil, fmt.Errorf("failed to marshal JSON: %w", err)
+		return fmt.Errorf("failed to marshal JSON: %w", err)
 	}
 
-	return d.encrypt(subKey, serialized)
+	return d.encrypt(serialized, dest)
 }
 
-func (d *DAL) decryptJSON(subKey encryption.SubKey, encrypted []byte, v any) error { //nolint:unparam
-	decrypted, err := d.decrypt(subKey, encrypted)
+func (d *DAL) decryptJSON(encrypted encryption.Encryptable, v any) error { //nolint:unparam
+	decrypted, err := d.decrypt(encrypted)
 	if err != nil {
-		return fmt.Errorf("failed to decrypt json with subkey %s: %w", subKey, err)
+		return fmt.Errorf("failed to decrypt json with subkey %s: %w", encrypted.SubKey(), err)
 	}
 
 	if err = json.Unmarshal(decrypted, v); err != nil {

backend/controller/dal/events.go

@@ -13,7 +13,6 @@ import (
 	"github.com/TBD54566975/ftl/backend/controller/sql"
 	dalerrs "github.com/TBD54566975/ftl/backend/dal"
 	"github.com/TBD54566975/ftl/backend/schema"
-	"github.com/TBD54566975/ftl/internal/encryption"
 	"github.com/TBD54566975/ftl/internal/log"
 	"github.com/TBD54566975/ftl/internal/model"
 )
@@ -349,7 +348,7 @@ func (d *DAL) transformRowsToTimelineEvents(deploymentKeys map[int64]model.Deplo
 		switch row.Type {
 		case sql.EventTypeLog:
 			var jsonPayload eventLogJSON
-			if err := d.decryptJSON(encryption.TimelineSubKey, row.Payload, &jsonPayload); err != nil {
+			if err := d.decryptJSON(&row.Payload, &jsonPayload); err != nil {
 				return nil, fmt.Errorf("failed to decrypt log event: %w", err)
 			}
 
@@ -371,7 +370,7 @@ func (d *DAL) transformRowsToTimelineEvents(deploymentKeys map[int64]model.Deplo
 
 		case sql.EventTypeCall:
 			var jsonPayload eventCallJSON
-			if err := d.decryptJSON(encryption.TimelineSubKey, row.Payload, &jsonPayload); err != nil {
+			if err := d.decryptJSON(&row.Payload, &jsonPayload); err != nil {
 				return nil, fmt.Errorf("failed to decrypt call event: %w", err)
 			}
 			var sourceVerb optional.Option[schema.Ref]
@@ -396,7 +395,7 @@ func (d *DAL) transformRowsToTimelineEvents(deploymentKeys map[int64]model.Deplo
 
 		case sql.EventTypeDeploymentCreated:
 			var jsonPayload eventDeploymentCreatedJSON
-			if err := d.decryptJSON(encryption.TimelineSubKey, row.Payload, &jsonPayload); err != nil {
+			if err := d.decryptJSON(&row.Payload, &jsonPayload); err != nil {
 				return nil, fmt.Errorf("failed to decrypt call event: %w", err)
 			}
 			out = append(out, &DeploymentCreatedEvent{
@@ -411,7 +410,7 @@ func (d *DAL) transformRowsToTimelineEvents(deploymentKeys map[int64]model.Deplo
 
 		case sql.EventTypeDeploymentUpdated:
 			var jsonPayload eventDeploymentUpdatedJSON
-			if err := d.decryptJSON(encryption.TimelineSubKey, row.Payload, &jsonPayload); err != nil {
+			if err := d.decryptJSON(&row.Payload, &jsonPayload); err != nil {
 				return nil, fmt.Errorf("failed to decrypt call event: %w", err)
 			}
 			out = append(out, &DeploymentUpdatedEvent{

backend/controller/dal/fsm.go

@@ -32,11 +32,11 @@ import (
 //
 // Note: no validation of the FSM is performed.
 func (d *DAL) StartFSMTransition(ctx context.Context, fsm schema.RefKey, instanceKey string, destinationState schema.RefKey, request []byte, encrypted bool, retryParams schema.RetryParams) (err error) {
-	var encryptedRequest []byte
+	var encryptedRequest encryption.EncryptedAsyncColumn
 	if encrypted {
-		encryptedRequest = request
+		encryptedRequest = encryption.EncryptedAsyncColumn(request)
 	} else {
-		encryptedRequest, err = d.encrypt(encryption.AsyncSubKey, request)
+		err = d.encrypt(request, &encryptedRequest)
 		if err != nil {
 			return fmt.Errorf("failed to encrypt FSM request: %w", err)
 		}
@@ -146,7 +146,8 @@ func (d *DAL) PopNextFSMEvent(ctx context.Context, fsm schema.RefKey, instanceKe
 }
 
 func (d *DAL) SetNextFSMEvent(ctx context.Context, fsm schema.RefKey, instanceKey string, nextState schema.RefKey, request json.RawMessage, requestType schema.Type) error {
-	encryptedRequest, err := d.encryptJSON(encryption.AsyncSubKey, request)
+	var encryptedRequest encryption.EncryptedAsyncColumn
+	err := d.encryptJSON(request, &encryptedRequest)
 	if err != nil {
 		return fmt.Errorf("failed to encrypt FSM request: %w", err)
 	}

backend/controller/dal/pubsub.go

@@ -21,7 +21,8 @@ import (
 )
 
 func (d *DAL) PublishEventForTopic(ctx context.Context, module, topic, caller string, payload []byte) error {
-	encryptedPayload, err := d.encrypt(encryption.AsyncSubKey, payload)
+	var encryptedPayload encryption.EncryptedAsyncColumn
+	err := d.encrypt(payload, &encryptedPayload)
 	if err != nil {
 		return fmt.Errorf("failed to encrypt payload: %w", err)
 	}

backend/controller/sql/schema/20240815003340_typesafe_encrypted_columns.sql

@@ -0,0 +1,15 @@
+-- migrate:up
+
+CREATE DOMAIN encrypted_timeline AS BYTEA;
+
+ALTER TABLE timeline
+    ALTER COLUMN payload TYPE encrypted_timeline;
+
+CREATE DOMAIN encrypted_async AS BYTEA;
+
+ALTER TABLE async_calls
+    ALTER COLUMN request TYPE encrypted_async,
+    ALTER COLUMN response TYPE encrypted_async;
+
+-- migrate:down
+