feat: allow arbitrary types in requests/responses (#763)

This is a prerequisite for PubSub, but we've always wanted to loosen
this restriction anyway. It was only in place temporarily until we get
concurrent deployed versions working, but I think we can just guide
people until that point.

There are still restrictions for `@Ingress` verbs in that the request
must still be a data structure, because the routing needs something to
map the path parameters and query parameters onto.

backend/controller/console.go

@@ -62,10 +62,14 @@ func (c *ConsoleService) GetModules(ctx context.Context, req *connect.Request[pb
 			case *schema.Verb:
 				//nolint:forcetypeassert
 				v := decl.ToProto().(*schemapb.Verb)
-				verbSchema := schema.VerbToSchema(v)
+				verbSchema := schema.VerbToSchema(v) // TODO: include all of the types  that the verb references
+				requestData, ok := verbSchema.Request.(*schema.DataRef)
+				if !ok {
+					return nil, fmt.Errorf("expected request to be a data ref, got %T", verbSchema.Request)
+				}
 				dataRef := schema.DataRef{
-					Module: verbSchema.Request.Module,
-					Name:   verbSchema.Request.Name,
+					Module: requestData.Module,
+					Name:   requestData.Name,
 				}
 				jsonRequestSchema, err := schema.DataToJSONSchema(sch, dataRef)
 				if err != nil {
@@ -80,6 +84,7 @@ func (c *ConsoleService) GetModules(ctx context.Context, req *connect.Request[pb
 					Schema:            verbSchema.String(),
 					JsonRequestSchema: string(jsonData),
 				})
+
 			case *schema.Data:
 				//nolint:forcetypeassert
 				d := decl.ToProto().(*schemapb.Data)

backend/controller/ingress/ingress.go

@@ -12,6 +12,7 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/TBD54566975/ftl/backend/common/slices"
 	"github.com/TBD54566975/ftl/backend/controller/dal"
@@ -73,7 +74,7 @@ func ValidateCallBody(body []byte, verbRef *schema.VerbRef, sch *schema.Schema)
 		return fmt.Errorf("HTTP request body is not valid JSON: %w", err)
 	}
 
-	return validateRequestMap(verb.Request, []string{verb.Request.String()}, requestMap, sch)
+	return validateValue(verb.Request, []string{verb.Request.String()}, requestMap, sch)
 }
 
 type HTTPRequest struct {
@@ -117,12 +118,16 @@ func ValidateAndExtractRequestBody(route *dal.IngressRoute, r *http.Request, sch
 			return nil, err
 		}
 	} else {
-		requestMap, err := buildRequestMap(route, r, verb.Request, sch)
+		request, ok := verb.Request.(*schema.DataRef)
+		if !ok {
+			return nil, fmt.Errorf("verb %s input must be a data structure", verb.Name)
+		}
+		requestMap, err := buildRequest(route, r, request, sch)
 		if err != nil {
 			return nil, err
 		}
 
-		err = validateRequestMap(verb.Request, []string{verb.Request.String()}, requestMap, sch)
+		err = validateRequestMap(request, []string{verb.Request.String()}, requestMap, sch)
 		if err != nil {
 			return nil, err
 		}
@@ -136,7 +141,7 @@ func ValidateAndExtractRequestBody(route *dal.IngressRoute, r *http.Request, sch
 	return body, nil
 }
 
-func buildRequestMap(route *dal.IngressRoute, r *http.Request, dataRef *schema.DataRef, sch *schema.Schema) (map[string]any, error) {
+func buildRequest(route *dal.IngressRoute, r *http.Request, dataRef *schema.DataRef, sch *schema.Schema) (map[string]any, error) {
 	requestMap := map[string]any{}
 	matchSegments(route.Path, r.URL.Path, func(segment, value string) {
 		requestMap[segment] = value
@@ -205,6 +210,24 @@ func validateRequestMap(dataRef *schema.DataRef, path path, request map[string]a
 func validateValue(fieldType schema.Type, path path, value any, sch *schema.Schema) error {
 	var typeMatches bool
 	switch fieldType := fieldType.(type) {
+	case *schema.Unit:
+		rv := reflect.ValueOf(value)
+		if rv.Kind() != reflect.Map || rv.Len() != 0 {
+			return fmt.Errorf("%s must be an empty map", path)
+		}
+		return nil
+
+	case *schema.Time:
+		str, ok := value.(string)
+		if !ok {
+			return fmt.Errorf("time %s must be an RFC3339 formatted string", path)
+		}
+		_, err := time.Parse(time.RFC3339Nano, str)
+		if err != nil {
+			return fmt.Errorf("time %s must be an RFC3339 formatted string: %w", path, err)
+		}
+		return nil
+
 	case *schema.Int:
 		switch value := value.(type) {
 		case float64:
@@ -214,6 +237,7 @@ func validateValue(fieldType schema.Type, path path, value any, sch *schema.Sche
 				typeMatches = true
 			}
 		}
+
 	case *schema.Float:
 		switch value := value.(type) {
 		case float64:
@@ -223,8 +247,10 @@ func validateValue(fieldType schema.Type, path path, value any, sch *schema.Sche
 				typeMatches = true
 			}
 		}
+
 	case *schema.String:
 		_, typeMatches = value.(string)
+
 	case *schema.Bool:
 		switch value := value.(type) {
 		case bool:
@@ -234,6 +260,7 @@ func validateValue(fieldType schema.Type, path path, value any, sch *schema.Sche
 				typeMatches = true
 			}
 		}
+
 	case *schema.Array:
 		rv := reflect.ValueOf(value)
 		if rv.Kind() != reflect.Slice {
@@ -248,6 +275,7 @@ func validateValue(fieldType schema.Type, path path, value any, sch *schema.Sche
 			}
 		}
 		typeMatches = true
+
 	case *schema.Map:
 		rv := reflect.ValueOf(value)
 		if rv.Kind() != reflect.Map {
@@ -266,13 +294,15 @@ func validateValue(fieldType schema.Type, path path, value any, sch *schema.Sche
 			}
 		}
 		typeMatches = true
+
 	case *schema.DataRef:
 		if valueMap, ok := value.(map[string]any); ok {
 			if err := validateRequestMap(fieldType, path, valueMap, sch); err != nil {
 				return err
 			}
 			typeMatches = true
 		}
+
 	case *schema.Bytes:
 		_, typeMatches = value.([]byte)
 		if bodyStr, ok := value.(string); ok {
@@ -282,15 +312,13 @@ func validateValue(fieldType schema.Type, path path, value any, sch *schema.Sche
 			}
 			typeMatches = true
 		}
+
 	case *schema.Optional:
 		if value == nil {
 			typeMatches = true
 		} else {
 			return validateValue(fieldType.Type, path, value, sch)
 		}
-
-	default:
-		return fmt.Errorf("%s has unsupported type %T", path, fieldType)
 	}
 
 	if !typeMatches {

backend/schema/bool.go

@@ -14,10 +14,7 @@ type Bool struct {
 
 var _ Type = (*Bool)(nil)
 
-func (*Bool) schemaChildren() []Node { return nil }
-func (*Bool) schemaType()            {}
-func (*Bool) String() string         { return "Bool" }
-
-func (b *Bool) ToProto() proto.Message {
-	return &schemapb.Bool{Pos: posToProto(b.Pos)}
-}
+func (*Bool) schemaChildren() []Node   { return nil }
+func (*Bool) schemaType()              {}
+func (*Bool) String() string           { return "Bool" }
+func (b *Bool) ToProto() proto.Message { return &schemapb.Bool{Pos: posToProto(b.Pos)} }

backend/schema/builtin.go

@@ -0,0 +1,33 @@
+package schema
+
+// BuiltinsSource is the schema source code for built-in types.
+var BuiltinsSource = `
+// Built-in types for FTL.
+builtin module builtin {
+  // HTTP request structure used for HTTP ingress verbs.
+  data HttpRequest {
+    method String
+    path String
+    pathParameters {String: String}
+    query {String: [String]}
+    headers {String: [String]}
+    body Bytes
+  }
+
+  // HTTP response structure used for HTTP ingress verbs.
+  data HttpResponse {
+    status Int
+    headers {String: [String]}
+    body Bytes
+  }
+}
+`
+
+// Builtins returns a [Module] containing built-in types.
+func Builtins() *Module {
+	module, err := ParseModuleString("builtins.ftl", BuiltinsSource)
+	if err != nil {
+		panic("failed to parse builtins: " + err.Error())
+	}
+	return module
+}

backend/schema/jsonschema.go

@@ -40,6 +40,10 @@ func DataToJSONSchema(schema *Schema, dataRef DataRef) (*jsonschema.Schema, erro
 
 func nodeToJSSchema(node Node, dataRefs map[DataRef]bool) *jsonschema.Schema {
 	switch node := node.(type) {
+	case *Unit:
+		st := jsonschema.Object
+		return &jsonschema.Schema{Type: &jsonschema.Type{SimpleTypes: &st}}
+
 	case *Data:
 		st := jsonschema.Object
 		schema := &jsonschema.Schema{

backend/schema/normalise.go

@@ -8,6 +8,9 @@ func Normalise[T Node](n T) T {
 	var zero Position
 	var ni Node = n
 	switch c := ni.(type) {
+	case *Unit:
+		c.Pos = zero
+
 	case *Schema:
 		c.Pos = zero
 		c.Modules = normaliseSlice(c.Modules)

backend/schema/parser.go

@@ -13,10 +13,13 @@ import (
 
 var (
 	declUnion            = []Decl{&Data{}, &Verb{}}
-	nonOptionalTypeUnion = []Type{&Int{}, &Float{}, &String{}, &Bytes{}, &Bool{}, &Time{}, &Array{}, &Map{} /*&VerbRef{},*/, &DataRef{}}
-	typeUnion            = append(nonOptionalTypeUnion, &Optional{})
-	metadataUnion        = []Metadata{&MetadataCalls{}, &MetadataIngress{}}
-	ingressUnion         = []IngressPathComponent{&IngressPathLiteral{}, &IngressPathParameter{}}
+	nonOptionalTypeUnion = []Type{
+		&Int{}, &Float{}, &String{}, &Bytes{}, &Bool{}, &Time{}, &Array{},
+		&Map{}, &DataRef{}, &Unit{},
+	}
+	typeUnion     = append(nonOptionalTypeUnion, &Optional{})
+	metadataUnion = []Metadata{&MetadataCalls{}, &MetadataIngress{}}
+	ingressUnion  = []IngressPathComponent{&IngressPathLiteral{}, &IngressPathParameter{}}
 
 	// Used by protobuf generation.
 	unions = map[reflect.Type][]reflect.Type{

backend/schema/protobuf_enc.go

@@ -73,6 +73,9 @@ func ingressListToProto(nodes []IngressPathComponent) []*schemapb.IngressPathCom
 
 func typeToProto(t Type) *schemapb.Type {
 	switch t := t.(type) {
+	case *Unit:
+		return &schemapb.Type{Value: &schemapb.Type_Unit{Unit: t.ToProto().(*schemapb.Unit)}}
+
 	case *VerbRef, *SourceRef, *SinkRef:
 		panic("unreachable")
 

backend/schema/unit.go

@@ -0,0 +1,20 @@
+package schema
+
+import (
+	"google.golang.org/protobuf/reflect/protoreflect"
+
+	schemapb "github.com/TBD54566975/ftl/protos/xyz/block/ftl/v1/schema"
+)
+
+type Unit struct {
+	Pos Position `parser:"" protobuf:"1,optional"`
+
+	Unit bool `parser:"@'Unit'" protobuf:"-"`
+}
+
+var _ Type = (*Unit)(nil)
+
+func (u *Unit) schemaType()                        {}
+func (u *Unit) String() string                     { return "Unit" }
+func (u *Unit) ToProto() protoreflect.ProtoMessage { return &schemapb.Unit{Pos: posToProto(u.Pos)} }
+func (u *Unit) schemaChildren() []Node             { return nil }

backend/schema/validate.go

@@ -17,42 +17,10 @@ var (
 	// keywords associated with these types.
 	reservedIdentNames = map[string]bool{
 		"Int": true, "Float": true, "String": true, "Bytes": true, "Bool": true,
-		"Time": true, "Map": true, "Array": true,
+		"Time": true, "Unit": true,
 	}
-
-	// BuiltinsSource is the schema source code for built-in types.
-	BuiltinsSource = `
-// Built-in types for FTL.
-builtin module builtin {
-  // HTTP request structure used for HTTP ingress verbs.
-  data HttpRequest {
-    method String
-    path String
-    pathParameters {String: String}
-    query {String: [String]}
-    headers {String: [String]}
-    body Bytes
-  }
-
-  // HTTP response structure used for HTTP ingress verbs.
-  data HttpResponse {
-    status Int
-    headers {String: [String]}
-    body Bytes
-  }
-}
-`
 )
 
-// Builtins returns a [Module] containing built-in types.
-func Builtins() *Module {
-	module, err := ParseModuleString("builtins.ftl", BuiltinsSource)
-	if err != nil {
-		panic("failed to parse builtins: " + err.Error())
-	}
-	return module
-}
-
 // MustValidate panics if a schema is invalid.
 //
 // This is useful for testing.
@@ -215,7 +183,7 @@ func ValidateModule(module *Module) error {
 			*Time, *Map, *Module, *Schema, *String, *Bytes, *VerbRef,
 			*MetadataCalls, *MetadataIngress, IngressPathComponent,
 			*IngressPathLiteral, *IngressPathParameter, *Optional,
-			*SourceRef, *SinkRef:
+			*SourceRef, *SinkRef, *Unit:
 		case Type, Metadata, Decl: // Union sql.
 		}
 		return next()

backend/schema/verb.go

@@ -15,11 +15,21 @@ type Verb struct {
 
 	Comments []string   `parser:"@Comment*" protobuf:"3"`
 	Name     string     `parser:"'verb' @Ident" protobuf:"2"`
-	Request  *DataRef   `parser:"'(' @@ ')'" protobuf:"4"`
-	Response *DataRef   `parser:"@@" protobuf:"5"`
+	Request  Type       `parser:"'(' @@ ')'" protobuf:"4"`
+	Response Type       `parser:"@@" protobuf:"5"`
 	Metadata []Metadata `parser:"@@*" protobuf:"6"`
 }
 
+// verb mySink(Req) Unit
+// verb mySource(Unit) Req
+//
+// func MySource(context.Context) (Req, error) {}
+//
+// func Checkout(ctx context.Context, req CheckoutRequest) (CheckoutResponse, error) {
+//    addresses, err := ftl.Call(ctx, GetAddress, req.User)
+//    addresses, err := ftl.Call(ctx, GetAddress, GetAddressRequest{User: req.User})
+// 	  return CheckoutResponse{}, nil
+
 var _ Decl = (*Verb)(nil)
 
 func (v *Verb) schemaDecl() {}
@@ -65,8 +75,8 @@ func (v *Verb) ToProto() proto.Message {
 		Pos:      posToProto(v.Pos),
 		Name:     v.Name,
 		Comments: v.Comments,
-		Request:  v.Request.ToProto().(*schemapb.DataRef),  //nolint:forcetypeassert
-		Response: v.Response.ToProto().(*schemapb.DataRef), //nolint:forcetypeassert
+		Request:  typeToProto(v.Request),
+		Response: typeToProto(v.Response),
 		Metadata: metadataListToProto(v.Metadata),
 	}
 }
@@ -76,8 +86,8 @@ func VerbToSchema(s *schemapb.Verb) *Verb {
 		Pos:      posFromProto(s.Pos),
 		Name:     s.Name,
 		Comments: s.Comments,
-		Request:  dataRefToSchema(s.Request),
-		Response: dataRefToSchema(s.Response),
+		Request:  typeToSchema(s.Request),
+		Response: typeToSchema(s.Response),
 		Metadata: metadataListToSchema(s.Metadata),
 	}
 }