diff --git a/go-runtime/compile/schema.go b/go-runtime/compile/schema.go
index 09cc41815e..58a268d3b3 100644
--- a/go-runtime/compile/schema.go
+++ b/go-runtime/compile/schema.go
@@ -5,12 +5,8 @@ import (
 	"go/ast"
 	"go/token"
 	"go/types"
-	"path"
-	"reflect"
 	"strconv"
 	"strings"
-	"unicode"
-	"unicode/utf8"
 
 	"github.com/alecthomas/types/optional"
 	"golang.org/x/exp/maps"
@@ -26,36 +22,20 @@ import (
 var (
 	fset = token.NewFileSet()
 
-	ftlCallFuncPath         = "github.com/TBD54566975/ftl/go-runtime/ftl.Call"
-	ftlFSMFuncPath          = "github.com/TBD54566975/ftl/go-runtime/ftl.FSM"
-	ftlTransitionFuncPath   = "github.com/TBD54566975/ftl/go-runtime/ftl.Transition"
-	ftlStartFuncPath        = "github.com/TBD54566975/ftl/go-runtime/ftl.Start"
-	ftlPostgresDBFuncPath   = "github.com/TBD54566975/ftl/go-runtime/ftl.PostgresDatabase"
-	ftlTopicFuncPath        = "github.com/TBD54566975/ftl/go-runtime/ftl.Topic"
-	ftlSubscriptionFuncPath = "github.com/TBD54566975/ftl/go-runtime/ftl.Subscription"
-	aliasFieldTag           = "json"
+	ftlCallFuncPath       = "github.com/TBD54566975/ftl/go-runtime/ftl.Call"
+	ftlFSMFuncPath        = "github.com/TBD54566975/ftl/go-runtime/ftl.FSM"
+	ftlTransitionFuncPath = "github.com/TBD54566975/ftl/go-runtime/ftl.Transition"
+	ftlStartFuncPath      = "github.com/TBD54566975/ftl/go-runtime/ftl.Start"
+	ftlPostgresDBFuncPath = "github.com/TBD54566975/ftl/go-runtime/ftl.PostgresDatabase"
 )
 
 // NativeNames is a map of top-level declarations to their native Go names.
 type NativeNames map[schema.Node]string
 
-func noEndColumnErrorf(pos token.Pos, format string, args ...interface{}) *schema.Error {
-	return tokenErrorf(pos, "", format, args...)
-}
-
 func unexpectedDirectiveErrorf(dir directive, format string, args ...interface{}) *schema.Error {
 	return schema.Errorf(dir.GetPosition(), 0, format, args...)
 }
 
-func tokenErrorf(pos token.Pos, tokenText string, format string, args ...interface{}) *schema.Error {
-	goPos := goPosToSchemaPos(pos)
-	endColumn := goPos.Column
-	if len(tokenText) > 0 {
-		endColumn += utf8.RuneCountInString(tokenText)
-	}
-	return schema.Errorf(goPosToSchemaPos(pos), endColumn, format, args...)
-}
-
 func errorf(node ast.Node, format string, args ...interface{}) *schema.Error {
 	pos, endCol := goNodePosToSchemaPos(node)
 	return schema.Errorf(pos, endCol, format, args...)
@@ -92,10 +72,6 @@ func legacyExtractModuleSchema(dir string, sch *schema.Schema, out *extract.Resu
 			continue
 		}
 		pctx := newParseContext(pkg, pkgs, sch, out)
-		err := extractInitialDecls(pctx)
-		if err != nil {
-			return err
-		}
 		for _, file := range pkg.Syntax {
 			err := goast.Visit(file, func(stack []ast.Node, next func() error) (err error) {
 				node := stack[len(stack)-1]
@@ -118,30 +94,6 @@ func legacyExtractModuleSchema(dir string, sch *schema.Schema, out *extract.Resu
 	return nil
 }
 
-// extractInitialDecls traverses the package's AST and extracts declarations needed up front (topics)
-//
-// We need to know the stack when visiting a topic decl, but the subscription may occur first.
-// In this case there is no way for the subscription to make the topic exported.
-func extractInitialDecls(pctx *parseContext) error {
-	for _, file := range pctx.pkg.Syntax {
-		err := goast.Visit(file, func(stack []ast.Node, next func() error) (err error) {
-			switch node := stack[len(stack)-1].(type) {
-			case *ast.CallExpr:
-				_, fn := deref[*types.Func](pctx.pkg, node.Fun)
-				if fn != nil && fn.FullName() == ftlTopicFuncPath {
-					extractTopicDecl(pctx, node, stack)
-				}
-			}
-
-			return next()
-		})
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func extractStringLiteralArg(node *ast.CallExpr, argIndex int) (string, *schema.Error) {
 	if argIndex >= len(node.Args) {
 		return "", errorf(node, "expected string argument at index %d", argIndex)
@@ -162,55 +114,6 @@ func extractStringLiteralArg(node *ast.CallExpr, argIndex int) (string, *schema.
 	return s, nil
 }
 
-// extractTopicDecl expects: _ = ftl.Topic[EventType]("name_literal")
-func extractTopicDecl(pctx *parseContext, node *ast.CallExpr, stack []ast.Node) {
-	name, nameErr := extractStringLiteralArg(node, 0)
-	if nameErr != nil {
-		pctx.errors.add(nameErr)
-		return
-	}
-
-	varDecl, ok := varDeclForStack(stack)
-	if !ok {
-		pctx.errors.add(errorf(node, "expected topic declaration to be assigned to a variable"))
-		return
-	} else if len(varDecl.Specs) == 0 {
-		pctx.errors.add(errorf(node, "expected topic declaration to have at least 1 spec"))
-		return
-	}
-	topicVarPos := goPosToSchemaPos(varDecl.Specs[0].Pos())
-
-	comments, directives := commentsAndDirectivesForVar(pctx, varDecl, stack)
-	export := false
-	for _, dir := range directives {
-		if _, ok := dir.(*directiveExport); ok {
-			export = true
-		} else {
-			pctx.errors.add(unexpectedDirectiveErrorf(dir, "unexpected directive %q attached for topic", dir))
-		}
-	}
-
-	// Check for duplicates
-	_, endCol := goNodePosToSchemaPos(node)
-	for _, d := range pctx.module.Decls {
-		existing, ok := d.(*schema.Topic)
-		if ok && existing.Name == name {
-			pctx.errors.add(errorf(node, "duplicate topic registration at %d:%d-%d", existing.Pos.Line, existing.Pos.Column, endCol))
-			return
-		}
-	}
-
-	topic := &schema.Topic{
-		Pos:      goPosToSchemaPos(node.Pos()),
-		Name:     name,
-		Export:   export,
-		Comments: comments,
-		Event:    nil, // event is nil until we the next pass, when we can visit the full graph
-	}
-	pctx.module.Decls = append(pctx.module.Decls, topic)
-	pctx.topicsByPos[topicVarPos] = topic
-}
-
 func visitCallExpr(pctx *parseContext, node *ast.CallExpr, stack []ast.Node) {
 	_, fn := deref[*types.Func](pctx.pkg, node.Fun)
 	if fn == nil {
@@ -225,12 +128,6 @@ func visitCallExpr(pctx *parseContext, node *ast.CallExpr, stack []ast.Node) {
 
 	case ftlPostgresDBFuncPath:
 		parseDatabaseDecl(pctx, node, schema.PostgresDatabaseType)
-
-	case ftlTopicFuncPath:
-		parseTopicDecl(pctx, node)
-
-	case ftlSubscriptionFuncPath:
-		parseSubscriptionDecl(pctx, node)
 	}
 }
 
@@ -421,103 +318,6 @@ func parseDatabaseDecl(pctx *parseContext, node *ast.CallExpr, dbType string) {
 	pctx.module.Decls = append(pctx.module.Decls, decl)
 }
 
-// parseTopicDecl expects: _ = ftl.Topic[EventType]("name_literal")
-func parseTopicDecl(pctx *parseContext, node *ast.CallExpr) {
-	// already extracted topic in the initial pass of the ast graph
-	// we did not do event type resolution yet, so we need to do that now
-	name, nameErr := extractStringLiteralArg(node, 0)
-	if nameErr != nil {
-		// error already added in previous pass
-		return
-	}
-
-	var topic *schema.Topic
-	for _, d := range pctx.module.Decls {
-		if d, ok := d.(*schema.Topic); ok && d.Name == name {
-			topic = d
-		}
-	}
-
-	// update topic's event type
-	indexExpr, ok := node.Fun.(*ast.IndexExpr)
-	if !ok {
-		pctx.errors.add(errorf(node, "must have an event type as a type parameter"))
-		return
-	}
-	typeParamType, ok := visitType(pctx, node.Pos(), pctx.pkg.TypesInfo.TypeOf(indexExpr.Index), topic.Export).Get()
-	if !ok {
-		pctx.errors.add(errorf(node, "invalid event type"))
-		return
-	}
-	topic.Event = typeParamType
-}
-
-// parseSubscriptionDecl expects: var _ = ftl.Subscription(topicHandle, "name_literal")
-func parseSubscriptionDecl(pctx *parseContext, node *ast.CallExpr) {
-	var name string
-	var topicRef *schema.Ref
-	if len(node.Args) != 2 {
-		pctx.errors.add(errorf(node, "subscription registration must have a topic"))
-		return
-	}
-	if topicIdent, ok := node.Args[0].(*ast.Ident); ok {
-		// Topic is within module
-		// we will find the subscription name from the string literal parameter
-		object := pctx.pkg.TypesInfo.ObjectOf(topicIdent)
-		topic, ok := pctx.topicsByPos[goPosToSchemaPos(object.Pos())]
-		if !ok {
-			pctx.errors.add(errorf(node, "could not find topic declaration for topic variable"))
-			return
-		}
-		topicRef = &schema.Ref{
-			Module: pctx.module.Name,
-			Name:   topic.Name,
-		}
-	} else if topicSelExp, ok := node.Args[0].(*ast.SelectorExpr); ok {
-		// External topic
-		// we will derive subscription name from generated variable name
-		moduleIdent, moduleOk := topicSelExp.X.(*ast.Ident)
-		if !moduleOk {
-			pctx.errors.add(errorf(node, "subscription registration must have a topic"))
-			return
-		}
-		varName := topicSelExp.Sel.Name
-		topicRef = &schema.Ref{
-			Module: moduleIdent.Name,
-			Name:   strings.ToLower(string(varName[0])) + varName[1:],
-		}
-	} else {
-		pctx.errors.add(errorf(node, "subscription registration must have a topic"))
-		return
-	}
-
-	// name
-	var schemaErr *schema.Error
-	name, schemaErr = extractStringLiteralArg(node, 1)
-	if schemaErr != nil {
-		pctx.errors.add(schemaErr)
-		return
-	}
-
-	// Check for duplicates
-	_, endCol := goNodePosToSchemaPos(node)
-	for _, d := range pctx.module.Decls {
-		existing, ok := d.(*schema.Subscription)
-		if ok && existing.Name == name {
-			pctx.errors.add(errorf(node, "duplicate subscription registration at %d:%d-%d", existing.Pos.Line, existing.Pos.Column, endCol))
-			return
-		}
-	}
-
-	decl := &schema.Subscription{
-		Pos:   goPosToSchemaPos(node.Pos()),
-		Name:  name,
-		Topic: topicRef,
-	}
-
-	pctx.module.Decls = append(pctx.module.Decls, decl)
-}
-
 // varDeclForCall finds the variable being set in the stack
 func varDeclForStack(stack []ast.Node) (varDecl *ast.GenDecl, ok bool) {
 	for i := len(stack) - 1; i >= 0; i-- {
@@ -566,322 +366,6 @@ func ftlModuleFromGoModule(pkgPath string) optional.Option[string] {
 	return optional.Some(strings.TrimSuffix(parts[1], "_test"))
 }
 
-func visitStruct(pctx *parseContext, pos token.Pos, tnode types.Type, isExported bool) optional.Option[*schema.Ref] {
-	named, ok := tnode.(*types.Named)
-	if !ok {
-		pctx.errors.add(noEndColumnErrorf(pos, "expected named type but got %s", tnode))
-		return optional.None[*schema.Ref]()
-	}
-	nodePath := named.Obj().Pkg().Path()
-	if !pctx.isPathInPkg(nodePath) {
-		destModule, ok := ftlModuleFromGoModule(nodePath).Get()
-		if !ok {
-			pctx.errors.add(tokenErrorf(pos, nodePath, "struct declared in non-FTL module %s", nodePath))
-			return optional.None[*schema.Ref]()
-		}
-		dataRef := &schema.Ref{
-			Pos:    goPosToSchemaPos(pos),
-			Module: destModule,
-			Name:   named.Obj().Name(),
-		}
-		for i := range named.TypeArgs().Len() {
-			if typeArg, ok := visitType(pctx, pos, named.TypeArgs().At(i), isExported).Get(); ok {
-				// Fully qualify the Ref if needed
-				if ref, okArg := typeArg.(*schema.Ref); okArg {
-					if ref.Module == "" {
-						ref.Module = destModule
-					}
-					typeArg = ref
-				}
-				dataRef.TypeParameters = append(dataRef.TypeParameters, typeArg)
-			}
-		}
-		return optional.Some[*schema.Ref](dataRef)
-	}
-
-	out := &schema.Data{
-		Pos:    goPosToSchemaPos(pos),
-		Name:   strcase.ToUpperCamel(named.Obj().Name()),
-		Export: isExported,
-	}
-	pctx.nativeNames[out] = named.Obj().Name()
-	dataRef := &schema.Ref{
-		Pos:    goPosToSchemaPos(pos),
-		Module: pctx.module.Name,
-		Name:   out.Name,
-	}
-	for i := range named.TypeParams().Len() {
-		param := named.TypeParams().At(i)
-		out.TypeParameters = append(out.TypeParameters, &schema.TypeParameter{
-			Pos:  goPosToSchemaPos(pos),
-			Name: param.Obj().Name(),
-		})
-		typeArgs := named.TypeArgs()
-		if typeArgs == nil {
-			continue
-		}
-		if typeArg, ok := visitType(pctx, pos, typeArgs.At(i), isExported).Get(); ok {
-			dataRef.TypeParameters = append(dataRef.TypeParameters, typeArg)
-		}
-	}
-
-	// If the struct is generic, we need to use the origin type to get the
-	// fields.
-	if named.TypeParams().Len() > 0 {
-		named = named.Origin()
-	}
-
-	// Find type declaration so we can extract comments.
-	namedPos := named.Obj().Pos()
-	pkg, path, _ := pctx.pathEnclosingInterval(namedPos, namedPos)
-	if pkg != nil {
-		for i := len(path) - 1; i >= 0; i-- {
-			// We have to check both the type spec and the gen decl because the
-			// type could be declared as either "type Foo struct { ... }" or
-			// "type ( Foo struct { ... } )"
-			switch path := path[i].(type) {
-			case *ast.TypeSpec:
-				if path.Doc != nil {
-					out.Comments = parseComments(path.Doc)
-				}
-			case *ast.GenDecl:
-				if path.Doc != nil {
-					out.Comments = parseComments(path.Doc)
-				}
-			}
-		}
-	}
-
-	s, ok := named.Underlying().(*types.Struct)
-	if !ok {
-		pctx.errors.add(tokenErrorf(pos, named.String(), "expected struct but got %s", named))
-		return optional.None[*schema.Ref]()
-	}
-
-	fieldErrors := false
-	for i := range s.NumFields() {
-		f := s.Field(i)
-		if ft, ok := visitType(pctx, f.Pos(), f.Type(), isExported).Get(); ok {
-			// Check if field is exported
-			if len(f.Name()) > 0 && unicode.IsLower(rune(f.Name()[0])) {
-				pctx.errors.add(tokenErrorf(f.Pos(), f.Name(),
-					"struct field %s must be exported by starting with an uppercase letter", f.Name()))
-				fieldErrors = true
-			}
-
-			// Extract the JSON tag and split it to get just the field name
-			tagContent := reflect.StructTag(s.Tag(i)).Get(aliasFieldTag)
-			tagParts := strings.Split(tagContent, ",")
-			jsonFieldName := ""
-			if len(tagParts) > 0 {
-				jsonFieldName = tagParts[0]
-			}
-
-			var metadata []schema.Metadata
-			if jsonFieldName != "" {
-				metadata = append(metadata, &schema.MetadataAlias{
-					Pos:   goPosToSchemaPos(f.Pos()),
-					Kind:  schema.AliasKindJSON,
-					Alias: jsonFieldName,
-				})
-			}
-			out.Fields = append(out.Fields, &schema.Field{
-				Pos:      goPosToSchemaPos(f.Pos()),
-				Name:     strcase.ToLowerCamel(f.Name()),
-				Type:     ft,
-				Metadata: metadata,
-			})
-		} else {
-			pctx.errors.add(tokenErrorf(f.Pos(), f.Name(), "unsupported type %q for field %q", f.Type(), f.Name()))
-			fieldErrors = true
-		}
-	}
-	if fieldErrors {
-		return optional.None[*schema.Ref]()
-	}
-
-	pctx.module.AddData(out)
-	return optional.Some[*schema.Ref](dataRef)
-}
-
-func visitType(pctx *parseContext, pos token.Pos, tnode types.Type, isExported bool) optional.Option[schema.Type] {
-	if tparam, ok := tnode.(*types.TypeParam); ok {
-		return optional.Some[schema.Type](&schema.Ref{Pos: goPosToSchemaPos(pos), Name: tparam.Obj().Id()})
-	}
-
-	if named, ok := tnode.(*types.Named); ok {
-		// Handle refs to type aliases and enums, rather than the underlying type.
-		decl, ok := pctx.getDeclForTypeName(named.Obj().Name()).Get()
-		if ok {
-			switch decl.(type) {
-			case *schema.TypeAlias, *schema.Enum:
-				return visitNamedRef(pctx, pos, named, isExported)
-			case *schema.Data, *schema.Verb, *schema.Config, *schema.Secret, *schema.Database, *schema.FSM, *schema.Topic, *schema.Subscription:
-			}
-		}
-	}
-
-	switch underlying := tnode.Underlying().(type) {
-	case *types.Basic:
-		if named, ok := tnode.(*types.Named); ok {
-			if !pctx.isPathInPkg(named.Obj().Pkg().Path()) {
-				// external named types get treated as refs
-				return visitNamedRef(pctx, pos, named, isExported)
-			}
-			// internal named types without decls are treated as basic types
-		}
-		switch underlying.Kind() {
-		case types.String:
-			return optional.Some[schema.Type](&schema.String{Pos: goPosToSchemaPos(pos)})
-
-		case types.Int:
-			return optional.Some[schema.Type](&schema.Int{Pos: goPosToSchemaPos(pos)})
-
-		case types.Bool:
-			return optional.Some[schema.Type](&schema.Bool{Pos: goPosToSchemaPos(pos)})
-
-		case types.Float64:
-			return optional.Some[schema.Type](&schema.Float{Pos: goPosToSchemaPos(pos)})
-
-		default:
-			return optional.None[schema.Type]()
-		}
-
-	case *types.Struct:
-		named, ok := tnode.(*types.Named)
-		if !ok {
-			pctx.errors.add(noEndColumnErrorf(pos, "expected named type but got %s", tnode))
-			return optional.None[schema.Type]()
-		}
-
-		// Special-cased types.
-		switch named.Obj().Pkg().Path() + "." + named.Obj().Name() {
-		case "time.Time":
-			return optional.Some[schema.Type](&schema.Time{Pos: goPosToSchemaPos(pos)})
-
-		case "github.com/TBD54566975/ftl/go-runtime/ftl.Unit":
-			return optional.Some[schema.Type](&schema.Unit{Pos: goPosToSchemaPos(pos)})
-
-		case "github.com/TBD54566975/ftl/go-runtime/ftl.Option":
-			if underlying, ok := visitType(pctx, pos, named.TypeArgs().At(0), isExported).Get(); ok {
-				return optional.Some[schema.Type](&schema.Optional{Pos: goPosToSchemaPos(pos), Type: underlying})
-			}
-			return optional.None[schema.Type]()
-
-		default:
-			nodePath := named.Obj().Pkg().Path()
-			if !pctx.isPathInPkg(nodePath) && !strings.HasPrefix(nodePath, "ftl/") {
-				pctx.errors.add(noEndColumnErrorf(pos, "unsupported external type %s", nodePath+"."+named.Obj().Name()))
-				return optional.None[schema.Type]()
-			}
-			if ref, ok := visitStruct(pctx, pos, tnode, isExported).Get(); ok {
-				return optional.Some[schema.Type](ref)
-			}
-			return optional.None[schema.Type]()
-		}
-
-	case *types.Map:
-		return visitMap(pctx, pos, underlying, isExported)
-
-	case *types.Slice:
-		return visitSlice(pctx, pos, underlying, isExported)
-
-	case *types.Interface:
-		if underlying.String() == "any" {
-			return optional.Some[schema.Type](&schema.Any{Pos: goPosToSchemaPos(pos)})
-		}
-		if named, ok := tnode.(*types.Named); ok {
-			return visitNamedRef(pctx, pos, named, isExported)
-		}
-		return optional.None[schema.Type]()
-
-	default:
-		return optional.None[schema.Type]()
-	}
-}
-
-func visitNamedRef(pctx *parseContext, pos token.Pos, named *types.Named, isExported bool) optional.Option[schema.Type] {
-	if named.Obj().Pkg() == nil {
-		return optional.None[schema.Type]()
-	}
-
-	// Update the visibility of the reference if the referencer is exported (ensuring refs are transitively
-	// exported as needed).
-	if isExported {
-		if decl, ok := pctx.getDeclForTypeName(named.Obj().Name()).Get(); ok {
-			pctx.markAsExported(decl)
-		}
-	}
-
-	nodePath := named.Obj().Pkg().Path()
-	destModule := pctx.module.Name
-	if !pctx.isPathInPkg(nodePath) {
-		if !strings.HasPrefix(named.Obj().Pkg().Path(), "ftl/") {
-			pctx.errors.add(noEndColumnErrorf(pos,
-				"unsupported external type %q", named.Obj().Pkg().Path()+"."+named.Obj().Name()))
-			return optional.None[schema.Type]()
-		}
-		base := path.Dir(pctx.pkg.PkgPath)
-		destModule = path.Base(strings.TrimPrefix(nodePath, base+"/"))
-	}
-	ref := &schema.Ref{
-		Pos:    goPosToSchemaPos(pos),
-		Module: destModule,
-		Name:   strcase.ToUpperCamel(named.Obj().Name()),
-	}
-	return optional.Some[schema.Type](ref)
-}
-
-func visitMap(pctx *parseContext, pos token.Pos, tnode *types.Map, isExported bool) optional.Option[schema.Type] {
-	key, ok := visitType(pctx, pos, tnode.Key(), isExported).Get()
-	if !ok {
-		return optional.None[schema.Type]()
-	}
-
-	value, ok := visitType(pctx, pos, tnode.Elem(), isExported).Get()
-	if !ok {
-		return optional.None[schema.Type]()
-	}
-
-	return optional.Some[schema.Type](&schema.Map{
-		Pos:   goPosToSchemaPos(pos),
-		Key:   key,
-		Value: value,
-	})
-}
-
-func visitSlice(pctx *parseContext, pos token.Pos, tnode *types.Slice, isExported bool) optional.Option[schema.Type] {
-	// If it's a []byte, treat it as a Bytes type.
-	if basic, ok := tnode.Elem().Underlying().(*types.Basic); ok && basic.Kind() == types.Byte {
-		return optional.Some[schema.Type](&schema.Bytes{Pos: goPosToSchemaPos(pos)})
-	}
-	value, ok := visitType(pctx, pos, tnode.Elem(), isExported).Get()
-	if !ok {
-		return optional.None[schema.Type]()
-	}
-
-	return optional.Some[schema.Type](&schema.Array{
-		Pos:     goPosToSchemaPos(pos),
-		Element: value,
-	})
-}
-
-// Lazy load the compile-time reference from a package.
-func mustLoadRef(pkg, name string) types.Object {
-	pkgs, err := packages.Load(&packages.Config{Fset: fset, Mode: packages.NeedTypes}, pkg)
-	if err != nil {
-		panic(err)
-	}
-	if len(pkgs) != 1 {
-		panic("expected one package")
-	}
-	obj := pkgs[0].Types.Scope().Lookup(name)
-	if obj == nil {
-		panic("interface not found")
-	}
-	return obj
-}
-
 func deref[T types.Object](pkg *packages.Package, node ast.Expr) (string, T) {
 	var obj T
 	switch node := node.(type) {
diff --git a/go-runtime/compile/schema_test.go b/go-runtime/compile/schema_test.go
index 153211cff1..abf3c665de 100644
--- a/go-runtime/compile/schema_test.go
+++ b/go-runtime/compile/schema_test.go
@@ -3,8 +3,6 @@ package compile
 import (
 	"context"
 	"fmt"
-	"go/token"
-	"go/types"
 	"os"
 	"path/filepath"
 	"strings"
@@ -13,10 +11,7 @@ import (
 	"github.com/alecthomas/assert/v2"
 	"github.com/alecthomas/participle/v2/lexer"
 
-	"github.com/TBD54566975/golang-tools/go/packages"
-
 	"github.com/TBD54566975/ftl/backend/schema"
-	extract "github.com/TBD54566975/ftl/go-runtime/schema"
 	"github.com/TBD54566975/ftl/internal/errors"
 	"github.com/TBD54566975/ftl/internal/exec"
 	"github.com/TBD54566975/ftl/internal/log"
@@ -495,35 +490,6 @@ func TestParsedirectives(t *testing.T) {
 	}
 }
 
-func TestParseTypesTime(t *testing.T) {
-	timeRef := mustLoadRef("time", "Time").Type()
-	pctx := newParseContext(nil, []*packages.Package{}, &schema.Schema{}, &extract.Result{Module: &schema.Module{}})
-	parsed, ok := visitType(pctx, token.NoPos, timeRef, false).Get()
-	assert.True(t, ok)
-	_, ok = parsed.(*schema.Time)
-	assert.True(t, ok)
-}
-
-func TestParseBasicTypes(t *testing.T) {
-	tests := []struct {
-		name     string
-		input    types.Type
-		expected schema.Type
-	}{
-		{name: "String", input: types.Typ[types.String], expected: &schema.String{}},
-		{name: "Int", input: types.Typ[types.Int], expected: &schema.Int{}},
-		{name: "Bool", input: types.Typ[types.Bool], expected: &schema.Bool{}},
-		{name: "Float64", input: types.Typ[types.Float64], expected: &schema.Float{}},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			parsed, ok := visitType(nil, token.NoPos, tt.input, false).Get()
-			assert.True(t, ok)
-			assert.Equal(t, tt.expected, parsed)
-		})
-	}
-}
-
 func normaliseString(s string) string {
 	return strings.TrimSpace(strings.Join(slices.Map(strings.Split(s, "\n"), strings.TrimSpace), "\n"))
 }
diff --git a/go-runtime/schema/common/common.go b/go-runtime/schema/common/common.go
index 24fec17a86..a25a42eb8c 100644
--- a/go-runtime/schema/common/common.go
+++ b/go-runtime/schema/common/common.go
@@ -21,12 +21,14 @@ import (
 
 var (
 	// FtlUnitTypePath is the path to the FTL unit type.
-	FtlUnitTypePath   = "github.com/TBD54566975/ftl/go-runtime/ftl.Unit"
+	FtlUnitTypePath = "github.com/TBD54566975/ftl/go-runtime/ftl.Unit"
+	// FtlOptionTypePath is the path to the FTL option type.
 	FtlOptionTypePath = "github.com/TBD54566975/ftl/go-runtime/ftl.Option"
 
 	extractorRegistery = xsync.NewMapOf[reflect.Type, ExtractDeclFunc[schema.Decl, ast.Node]]()
 )
 
+// NewExtractor creates a new schema element extractor.
 func NewExtractor(name string, factType analysis.Fact, run func(*analysis.Pass) (interface{}, error)) *analysis.Analyzer {
 	if !reflect.TypeOf(factType).Implements(reflect.TypeOf((*SchemaFact)(nil)).Elem()) {
 		panic(fmt.Sprintf("factType %T does not implement SchemaFact", factType))
@@ -41,8 +43,13 @@ func NewExtractor(name string, factType analysis.Fact, run func(*analysis.Pass)
 	}
 }
 
+// ExtractDeclFunc extracts a schema declaration from the given node.
 type ExtractDeclFunc[T schema.Decl, N ast.Node] func(pass *analysis.Pass, node N, object types.Object) optional.Option[T]
 
+// NewDeclExtractor creates a new schema declaration extractor and registers its extraction function with
+// the common extractor registry.
+// The registry provides functions for extracting schema declarations by type and is used to extract
+// transitive declarations in a separate pass from the decl extraction pass.
 func NewDeclExtractor[T schema.Decl, N ast.Node](name string, extractFunc ExtractDeclFunc[T, N]) *analysis.Analyzer {
 	type Tag struct{} // Tag uniquely identifies the fact type for this extractor.
 	dType := reflect.TypeFor[T]()
@@ -60,10 +67,12 @@ func NewDeclExtractor[T schema.Decl, N ast.Node](name string, extractFunc Extrac
 	return NewExtractor(name, (*DefaultFact[Tag])(nil), runExtractDeclsFunc[T, N](extractFunc))
 }
 
+// ExtractorResult contains the results of an extraction pass.
 type ExtractorResult struct {
 	Facts []analysis.ObjectFact
 }
 
+// NewExtractorResult creates a new ExtractorResult with all object facts from this pass.
 func NewExtractorResult(pass *analysis.Pass) ExtractorResult {
 	return ExtractorResult{Facts: pass.AllObjectFacts()}
 }
@@ -107,6 +116,7 @@ func runExtractDeclsFunc[T schema.Decl, N ast.Node](extractFunc ExtractDeclFunc[
 	}
 }
 
+// ExtractComments extracts the comments from the given comment group.
 func ExtractComments(doc *ast.CommentGroup) []string {
 	if doc == nil {
 		return nil
@@ -118,6 +128,7 @@ func ExtractComments(doc *ast.CommentGroup) []string {
 	return comments
 }
 
+// ExtractType extracts the schema type for the given Go type.
 func ExtractType(pass *analysis.Pass, pos token.Pos, tnode types.Type) optional.Option[schema.Type] {
 	if tnode == nil {
 		return optional.None[schema.Type]()
@@ -196,6 +207,7 @@ func ExtractType(pass *analysis.Pass, pos token.Pos, tnode types.Type) optional.
 	}
 }
 
+// ExtractFuncForDecl returns the registered extraction function for the given declaration type.
 func ExtractFuncForDecl(t schema.Decl) (ExtractDeclFunc[schema.Decl, ast.Node], error) {
 	if f, ok := extractorRegistery.Load(reflect.TypeOf(t)); ok {
 		return f, nil
@@ -203,11 +215,13 @@ func ExtractFuncForDecl(t schema.Decl) (ExtractDeclFunc[schema.Decl, ast.Node],
 	return nil, fmt.Errorf("no extractor registered for %T", t)
 }
 
+// GoPosToSchemaPos converts a Go token.Pos to a schema.Position.
 func GoPosToSchemaPos(fset *token.FileSet, pos token.Pos) schema.Position {
 	p := fset.Position(pos)
 	return schema.Position{Filename: p.Filename, Line: p.Line, Column: p.Column, Offset: p.Offset}
 }
 
+// FtlModuleFromGoPackage returns the FTL module name from the given Go package path.
 func FtlModuleFromGoPackage(pkgPath string) (string, error) {
 	parts := strings.Split(pkgPath, "/")
 	if parts[0] != "ftl" {
@@ -216,6 +230,7 @@ func FtlModuleFromGoPackage(pkgPath string) (string, error) {
 	return strings.TrimSuffix(parts[1], "_test"), nil
 }
 
+// IsType returns true if the given type is of the specified type.
 func IsType[T types.Type](t types.Type) bool {
 	if _, ok := t.(*types.Named); ok {
 		t = t.Underlying()
@@ -224,6 +239,7 @@ func IsType[T types.Type](t types.Type) bool {
 	return ok
 }
 
+// IsPathInPkg returns true if the given path is in the package.
 func IsPathInPkg(pkg *types.Package, path string) bool {
 	if path == pkg.Path() {
 		return true
@@ -231,6 +247,7 @@ func IsPathInPkg(pkg *types.Package, path string) bool {
 	return strings.HasPrefix(path, pkg.Path()+"/")
 }
 
+// GetObjectForNode returns the types.Object for the given node.
 func GetObjectForNode(typesInfo *types.Info, node ast.Node) optional.Option[types.Object] {
 	var obj types.Object
 	switch n := node.(type) {
@@ -382,6 +399,7 @@ func extractSlice(pass *analysis.Pass, pos token.Pos, tnode *types.Slice) option
 	})
 }
 
+// ExtractTypeForNode extracts the schema type for the given node.
 func ExtractTypeForNode(pass *analysis.Pass, obj types.Object, node ast.Node, index types.Type) optional.Option[schema.Type] {
 	switch typ := node.(type) {
 	// Selector expression e.g. ftl.Unit, ftl.Option, foo.Bar
@@ -453,6 +471,7 @@ func ExtractTypeForNode(pass *analysis.Pass, obj types.Object, node ast.Node, in
 	return optional.None[schema.Type]()
 }
 
+// IsSelfReference returns true if the schema reference refers to this object itself.
 func IsSelfReference(pass *analysis.Pass, obj types.Object, t schema.Type) bool {
 	ref, ok := t.(*schema.Ref)
 	if !ok {
@@ -465,6 +484,7 @@ func IsSelfReference(pass *analysis.Pass, obj types.Object, t schema.Type) bool
 	return ref.Module == moduleName && strcase.ToUpperCamel(obj.Name()) == ref.Name
 }
 
+// GetNativeName returns the fully qualified name of the object, e.g. "github.com/TBD54566975/ftl/go-runtime/ftl.Unit".
 func GetNativeName(obj types.Object) string {
 	fqName := obj.Pkg().Path()
 	if parts := strings.Split(obj.Pkg().Path(), "/"); parts[len(parts)-1] != obj.Pkg().Name() {
@@ -473,10 +493,12 @@ func GetNativeName(obj types.Object) string {
 	return fqName + "." + obj.Name()
 }
 
+// IsExternalType returns true if the object is from an external package.
 func IsExternalType(obj types.Object) bool {
 	return !strings.HasPrefix(obj.Pkg().Path(), "ftl/")
 }
 
+// GetDeclTypeName returns the name of the declaration type, e.g. "verb" for *schema.Verb.
 func GetDeclTypeName(d schema.Decl) string {
 	typeStr := reflect.TypeOf(d).String()
 	lastDotIndex := strings.LastIndex(typeStr, ".")
@@ -509,6 +531,55 @@ func Deref[T types.Object](pass *analysis.Pass, node ast.Expr) (string, T) {
 	}
 }
 
+// CallExprFromVar extracts a call expression from a variable declaration, if present.
+func CallExprFromVar(node *ast.GenDecl) optional.Option[*ast.CallExpr] {
+	if node.Tok != token.VAR {
+		return optional.None[*ast.CallExpr]()
+	}
+	if len(node.Specs) != 1 {
+		return optional.None[*ast.CallExpr]()
+	}
+	vs, ok := node.Specs[0].(*ast.ValueSpec)
+	if !ok {
+		return optional.None[*ast.CallExpr]()
+	}
+	if len(vs.Values) != 1 {
+		return optional.None[*ast.CallExpr]()
+	}
+	callExpr, ok := vs.Values[0].(*ast.CallExpr)
+	if !ok {
+		return optional.None[*ast.CallExpr]()
+	}
+	return optional.Some(callExpr)
+}
+
+// FuncPathEquals checks if the function call expression is a call to the given path.
+func FuncPathEquals(pass *analysis.Pass, callExpr *ast.CallExpr, path string) bool {
+	_, fn := Deref[*types.Func](pass, callExpr.Fun)
+	if fn == nil {
+		return false
+	}
+	if fn.FullName() != path {
+		return false
+	}
+	return fn.FullName() == path
+}
+
+// ApplyMetadata applies the extracted metadata to the object, if present. Returns true if metadata was found and
+// applied.
+func ApplyMetadata[T schema.Decl](pass *analysis.Pass, obj types.Object, apply func(md *ExtractedMetadata)) bool {
+	if md, ok := GetFactForObject[*ExtractedMetadata](pass, obj).Get(); ok {
+		if _, ok = md.Type.(T); !ok && md.Type != nil {
+			NoEndColumnErrorf(pass, obj.Pos(), "schema declaration contains conflicting directives")
+			return false
+		}
+		apply(md)
+		return true
+	}
+	return false
+}
+
+// ExtractStringLiteralArg extracts a string literal argument from a call expression at the given index.
 func ExtractStringLiteralArg(pass *analysis.Pass, node *ast.CallExpr, argIndex int) string {
 	if argIndex >= len(node.Args) {
 		Errorf(pass, node, "expected string argument at index %d", argIndex)
diff --git a/go-runtime/schema/common/directive.go b/go-runtime/schema/common/directive.go
index 76fb2f9993..f37cd2f6fa 100644
--- a/go-runtime/schema/common/directive.go
+++ b/go-runtime/schema/common/directive.go
@@ -274,6 +274,9 @@ func (d *DirectiveExport) GetPosition() token.Pos {
 	return d.Pos
 }
 func (*DirectiveExport) MustAnnotate() []ast.Node { return []ast.Node{&ast.GenDecl{}} }
+func (d *DirectiveExport) IsExported() bool {
+	return d.Export
+}
 
 // DirectiveTypeMap is used to declare a native type to deserialize to in a given runtime.
 type DirectiveTypeMap struct {
diff --git a/go-runtime/schema/configsecret/analyzer.go b/go-runtime/schema/configsecret/analyzer.go
index 2f60064bcf..aac5567436 100644
--- a/go-runtime/schema/configsecret/analyzer.go
+++ b/go-runtime/schema/configsecret/analyzer.go
@@ -27,18 +27,15 @@ type Fact = common.DefaultFact[Tag]
 func Extract(pass *analysis.Pass) (interface{}, error) {
 	in := pass.ResultOf[inspect.Analyzer].(*inspector.Inspector) //nolint:forcetypeassert
 	nodeFilter := []ast.Node{
-		(*ast.ValueSpec)(nil),
+		(*ast.GenDecl)(nil),
 	}
 	in.Preorder(nodeFilter, func(n ast.Node) {
-		node := n.(*ast.ValueSpec) //nolint:forcetypeassert
-		obj, ok := common.GetObjectForNode(pass.TypesInfo, node).Get()
+		node := n.(*ast.GenDecl) //nolint:forcetypeassert
+		callExpr, ok := common.CallExprFromVar(node).Get()
 		if !ok {
 			return
 		}
-		if len(node.Values) != 1 {
-			return
-		}
-		callExpr, ok := node.Values[0].(*ast.CallExpr)
+		obj, ok := common.GetObjectForNode(pass.TypesInfo, node).Get()
 		if !ok {
 			return
 		}
diff --git a/go-runtime/schema/data/analyzer.go b/go-runtime/schema/data/analyzer.go
index fff96949f3..b0a1a0ab73 100644
--- a/go-runtime/schema/data/analyzer.go
+++ b/go-runtime/schema/data/analyzer.go
@@ -49,14 +49,10 @@ func extractData(pass *analysis.Pass, pos token.Pos, named *types.Named) optiona
 		Pos:  common.GoPosToSchemaPos(fset, pos),
 		Name: strcase.ToUpperCamel(named.Obj().Name()),
 	}
-	if md, ok := common.GetFactForObject[*common.ExtractedMetadata](pass, named.Obj()).Get(); ok {
-		if _, ok = md.Type.(*schema.Data); !ok && md.Type != nil {
-			return optional.None[*schema.Data]()
-		}
+	common.ApplyMetadata[*schema.Data](pass, named.Obj(), func(md *common.ExtractedMetadata) {
 		out.Comments = md.Comments
 		out.Export = md.IsExported
-	}
-
+	})
 	for i := range named.TypeParams().Len() {
 		param := named.TypeParams().At(i)
 		out.TypeParameters = append(out.TypeParameters, &schema.TypeParameter{
diff --git a/go-runtime/schema/enum/analyzer.go b/go-runtime/schema/enum/analyzer.go
index cbee770530..c60eb13b5a 100644
--- a/go-runtime/schema/enum/analyzer.go
+++ b/go-runtime/schema/enum/analyzer.go
@@ -57,11 +57,10 @@ func Extract(pass *analysis.Pass, node *ast.TypeSpec, obj types.Object) optional
 		Variants: valueVariants,
 		Type:     typ,
 	}
-	if md, ok := common.GetFactForObject[*common.ExtractedMetadata](pass, obj).Get(); ok {
+	common.ApplyMetadata[*schema.Enum](pass, obj, func(md *common.ExtractedMetadata) {
 		e.Comments = md.Comments
 		e.Export = md.IsExported
-	}
-
+	})
 	return optional.Some(e)
 
 }
diff --git a/go-runtime/schema/extract.go b/go-runtime/schema/extract.go
index b155defecb..6332b68846 100644
--- a/go-runtime/schema/extract.go
+++ b/go-runtime/schema/extract.go
@@ -6,7 +6,10 @@ import (
 
 	"github.com/TBD54566975/ftl/go-runtime/schema/call"
 	"github.com/TBD54566975/ftl/go-runtime/schema/configsecret"
+	"github.com/TBD54566975/ftl/go-runtime/schema/data"
 	"github.com/TBD54566975/ftl/go-runtime/schema/enum"
+	"github.com/TBD54566975/ftl/go-runtime/schema/subscription"
+	"github.com/TBD54566975/ftl/go-runtime/schema/topic"
 	"github.com/TBD54566975/ftl/go-runtime/schema/typeenum"
 	"github.com/TBD54566975/ftl/go-runtime/schema/typeenumvariant"
 	"github.com/TBD54566975/ftl/go-runtime/schema/valueenumvariant"
@@ -16,7 +19,6 @@ import (
 
 	"github.com/TBD54566975/ftl/backend/schema"
 	"github.com/TBD54566975/ftl/go-runtime/schema/common"
-	"github.com/TBD54566975/ftl/go-runtime/schema/data"
 	"github.com/TBD54566975/ftl/go-runtime/schema/finalize"
 	"github.com/TBD54566975/ftl/go-runtime/schema/initialize"
 	"github.com/TBD54566975/ftl/go-runtime/schema/metadata"
@@ -40,8 +42,8 @@ var Extractors = [][]*analysis.Analyzer{
 		inspect.Analyzer,
 	},
 	{
-		metadata.Extractor,
 		call.Extractor,
+		metadata.Extractor,
 	},
 	{
 		// must run before typeenumvariant.Extractor; typeenum.Extractor determines all possible discriminator
@@ -49,17 +51,20 @@ var Extractors = [][]*analysis.Analyzer{
 		typeenum.Extractor,
 	},
 	{
-		typealias.Extractor,
-		verb.Extractor,
-		data.Extractor,
 		configsecret.Extractor,
-		valueenumvariant.Extractor,
+		data.Extractor,
+		topic.Extractor,
+		typealias.Extractor,
 		typeenumvariant.Extractor,
+		valueenumvariant.Extractor,
+		verb.Extractor,
 	},
 	{
 		// must run after valueenumvariant.Extractor and typeenumvariant.Extractor;
 		// visits a node and aggregates its enum variants if present
 		enum.Extractor,
+		// must run after topic.Extractor
+		subscription.Extractor,
 	},
 	{
 		transitive.Extractor,
@@ -158,20 +163,11 @@ func combineAllPackageResults(results map[*analysis.Analyzer][]any, diagnostics
 		}
 		copyFailedRefs(refResults, fr.Failed)
 		for decl, obj := range fr.Extracted {
-			typename := common.GetDeclTypeName(decl)
-			var key string
-			switch d := decl.(type) {
-			case *schema.Config:
-				key = typename + d.Name + ":" + d.Type.String()
-			case *schema.Secret:
-				key = typename + d.Name + ":" + d.Type.String()
-			default:
-				key = typename + d.GetName()
-			}
+			key := getDeclKey(decl)
 			if value, ok := declKeys[key]; ok && value.A != obj {
 				// decls redeclared in subpackage
 				combined.Errors = append(combined.Errors, schema.Errorf(decl.Position(), decl.Position().Column,
-					"duplicate %s declaration for %q; already declared at %q", typename,
+					"duplicate %s declaration for %q; already declared at %q", common.GetDeclTypeName(decl),
 					combined.Module.Name+"."+decl.GetName(), value.B))
 				continue
 			}
@@ -354,3 +350,19 @@ func goQualifiedNameForWidenedType(obj types.Object, metadata []schema.Metadata)
 	}
 	return nativeName, nil
 }
+
+// criteria for uniqueness within a given decl type
+// used for detecting duplicate declarations
+func getDeclKey(decl schema.Decl) string {
+	typename := common.GetDeclTypeName(decl)
+	switch d := decl.(type) {
+	case *schema.Config:
+		return fmt.Sprintf("%s-%s-%s", typename, d.Name, d.Type)
+	case *schema.Secret:
+		return fmt.Sprintf("%s-%s-%s", typename, d.Name, d.Type)
+	case *schema.Topic:
+		return fmt.Sprintf("%s-%s-%s", typename, d.Name, d.Event)
+	default:
+		return fmt.Sprintf("%s-%s", typename, d.GetName())
+	}
+}
diff --git a/go-runtime/schema/subscription/analyzer.go b/go-runtime/schema/subscription/analyzer.go
new file mode 100644
index 0000000000..2622626f86
--- /dev/null
+++ b/go-runtime/schema/subscription/analyzer.go
@@ -0,0 +1,116 @@
+package subscription
+
+import (
+	"go/ast"
+	"go/types"
+	"strings"
+
+	"github.com/TBD54566975/ftl/backend/schema"
+	"github.com/TBD54566975/ftl/go-runtime/schema/common"
+	"github.com/TBD54566975/golang-tools/go/analysis"
+	"github.com/TBD54566975/golang-tools/go/analysis/passes/inspect"
+	"github.com/TBD54566975/golang-tools/go/ast/inspector"
+	"github.com/alecthomas/types/optional"
+)
+
+const (
+	ftlSubscriptionFuncPath = "github.com/TBD54566975/ftl/go-runtime/ftl.Subscription"
+)
+
+// Extractor extracts subscriptions.
+var Extractor = common.NewExtractor("subscription", (*Fact)(nil), Extract)
+
+type Tag struct{} // Tag uniquely identifies the fact type for this extractor.
+type Fact = common.DefaultFact[Tag]
+
+func Extract(pass *analysis.Pass) (interface{}, error) {
+	in := pass.ResultOf[inspect.Analyzer].(*inspector.Inspector) //nolint:forcetypeassert
+	nodeFilter := []ast.Node{
+		(*ast.GenDecl)(nil),
+	}
+	in.Preorder(nodeFilter, func(n ast.Node) {
+		node := n.(*ast.GenDecl) //nolint:forcetypeassert
+		callExpr, ok := common.CallExprFromVar(node).Get()
+		if !ok {
+			return
+		}
+		if !common.FuncPathEquals(pass, callExpr, ftlSubscriptionFuncPath) {
+			return
+		}
+		obj, ok := common.GetObjectForNode(pass.TypesInfo, node).Get()
+		if !ok {
+			return
+		}
+		if s, ok := extractSubscription(pass, obj, callExpr).Get(); ok {
+			common.MarkSchemaDecl(pass, obj, s)
+		}
+	})
+	return common.NewExtractorResult(pass), nil
+}
+
+// expects: var _ = ftl.Subscription(topicHandle, "name_literal")
+func extractSubscription(pass *analysis.Pass, obj types.Object, node *ast.CallExpr) optional.Option[*schema.Subscription] {
+	var topicRef *schema.Ref
+	if len(node.Args) != 2 {
+		common.Errorf(pass, node, "subscription registration must have exactly two arguments")
+		return optional.None[*schema.Subscription]()
+	}
+	if topicIdent, ok := node.Args[0].(*ast.Ident); ok {
+		// Topic is within module
+		// we will find the subscription name from the string literal parameter
+		object := pass.TypesInfo.ObjectOf(topicIdent)
+		fact, ok := common.GetFactForObject[*common.ExtractedDecl](pass, object).Get()
+		if !ok || fact.Decl == nil {
+			common.Errorf(pass, node, "could not find topic declaration for topic variable")
+			return optional.None[*schema.Subscription]()
+		}
+		topic, ok := fact.Decl.(*schema.Topic)
+		if !ok {
+			common.Errorf(pass, node, "could not find topic declaration for topic variable")
+			return optional.None[*schema.Subscription]()
+		}
+
+		moduleName, err := common.FtlModuleFromGoPackage(pass.Pkg.Path())
+		if err != nil {
+			return optional.None[*schema.Subscription]()
+		}
+		topicRef = &schema.Ref{
+			Module: moduleName,
+			Name:   topic.Name,
+		}
+	} else if topicSelExp, ok := node.Args[0].(*ast.SelectorExpr); ok {
+		// External topic
+		// we will derive subscription name from generated variable name
+		moduleIdent, moduleOk := topicSelExp.X.(*ast.Ident)
+		if !moduleOk {
+			common.Errorf(pass, node, "subscription registration must have a topic")
+			return optional.None[*schema.Subscription]()
+		}
+		varName := topicSelExp.Sel.Name
+		if varName == "" {
+			common.Errorf(pass, node, "subscription registration must have a topic")
+			return optional.None[*schema.Subscription]()
+		}
+		name := strings.ToLower(string(varName[0]))
+		if len(varName) > 1 {
+			name += varName[1:]
+		}
+		topicRef = &schema.Ref{
+			Module: moduleIdent.Name,
+			Name:   name,
+		}
+	} else {
+		common.Errorf(pass, node, "subscription registration must have a topic")
+		return optional.None[*schema.Subscription]()
+	}
+
+	subscription := &schema.Subscription{
+		Pos:   common.GoPosToSchemaPos(pass.Fset, node.Pos()),
+		Name:  common.ExtractStringLiteralArg(pass, node, 1),
+		Topic: topicRef,
+	}
+	common.ApplyMetadata[*schema.Subscription](pass, obj, func(md *common.ExtractedMetadata) {
+		subscription.Comments = md.Comments
+	})
+	return optional.Some(subscription)
+}
diff --git a/go-runtime/schema/topic/analyzer.go b/go-runtime/schema/topic/analyzer.go
new file mode 100644
index 0000000000..177b8f9deb
--- /dev/null
+++ b/go-runtime/schema/topic/analyzer.go
@@ -0,0 +1,73 @@
+package topic
+
+import (
+	"go/ast"
+	"go/types"
+
+	"github.com/TBD54566975/ftl/backend/schema"
+	"github.com/TBD54566975/ftl/go-runtime/schema/common"
+	"github.com/TBD54566975/golang-tools/go/analysis"
+	"github.com/TBD54566975/golang-tools/go/analysis/passes/inspect"
+	"github.com/TBD54566975/golang-tools/go/ast/inspector"
+	"github.com/alecthomas/types/optional"
+)
+
+const (
+	ftlTopicFuncPath = "github.com/TBD54566975/ftl/go-runtime/ftl.Topic"
+)
+
+// Extractor extracts topics.
+var Extractor = common.NewExtractor("topic", (*Fact)(nil), Extract)
+
+type Tag struct{} // Tag uniquely identifies the fact type for this extractor.
+type Fact = common.DefaultFact[Tag]
+
+func Extract(pass *analysis.Pass) (interface{}, error) {
+	in := pass.ResultOf[inspect.Analyzer].(*inspector.Inspector) //nolint:forcetypeassert
+	nodeFilter := []ast.Node{
+		(*ast.GenDecl)(nil),
+	}
+	in.Preorder(nodeFilter, func(n ast.Node) {
+		node := n.(*ast.GenDecl) //nolint:forcetypeassert
+		callExpr, ok := common.CallExprFromVar(node).Get()
+		if !ok {
+			return
+		}
+		if !common.FuncPathEquals(pass, callExpr, ftlTopicFuncPath) {
+			return
+		}
+		obj, ok := common.GetObjectForNode(pass.TypesInfo, node).Get()
+		if !ok {
+			return
+		}
+		if topic, ok := extractTopic(pass, node, callExpr, obj).Get(); ok {
+			common.MarkSchemaDecl(pass, obj, topic)
+		}
+	})
+	return common.NewExtractorResult(pass), nil
+}
+
+// expects: _ = ftl.Topic[EventType]("name_literal")
+func extractTopic(pass *analysis.Pass, node *ast.GenDecl, callExpr *ast.CallExpr, obj types.Object) optional.Option[*schema.Topic] {
+	indexExpr, ok := callExpr.Fun.(*ast.IndexExpr)
+	if !ok {
+		common.Errorf(pass, node, "must have an event type as a type parameter")
+		return optional.None[*schema.Topic]()
+	}
+	typeParamType, ok := common.ExtractType(pass, node.Pos(), pass.TypesInfo.TypeOf(indexExpr.Index)).Get()
+	if !ok {
+		common.Errorf(pass, node, "unsupported event type")
+		return optional.None[*schema.Topic]()
+	}
+
+	topic := &schema.Topic{
+		Pos:   common.GoPosToSchemaPos(pass.Fset, node.Pos()),
+		Name:  common.ExtractStringLiteralArg(pass, callExpr, 0),
+		Event: typeParamType,
+	}
+	common.ApplyMetadata[*schema.Subscription](pass, obj, func(md *common.ExtractedMetadata) {
+		topic.Comments = md.Comments
+		topic.Export = md.IsExported
+	})
+	return optional.Some(topic)
+}
diff --git a/go-runtime/schema/typealias/analyzer.go b/go-runtime/schema/typealias/analyzer.go
index d89d4ab86a..9ccd0697b5 100644
--- a/go-runtime/schema/typealias/analyzer.go
+++ b/go-runtime/schema/typealias/analyzer.go
@@ -27,7 +27,7 @@ func Extract(pass *analysis.Pass, node *ast.TypeSpec, obj types.Object) optional
 		Name: strcase.ToUpperCamel(obj.Name()),
 		Type: schType,
 	}
-	if md, ok := common.GetFactForObject[*common.ExtractedMetadata](pass, obj).Get(); ok {
+	if common.ApplyMetadata[*schema.TypeAlias](pass, obj, func(md *common.ExtractedMetadata) {
 		alias.Comments = md.Comments
 		alias.Export = md.IsExported
 		alias.Metadata = md.Metadata
@@ -36,7 +36,7 @@ func Extract(pass *analysis.Pass, node *ast.TypeSpec, obj types.Object) optional
 			hasGoTypeMap := false
 			nativeName := qualifiedNameFromSelectorExpr(pass, node.Type)
 			if nativeName == "" {
-				return optional.None[*schema.TypeAlias]()
+				return
 			}
 			for _, m := range md.Metadata {
 				if mt, ok := m.(*schema.MetadataTypeMap); ok {
@@ -46,7 +46,7 @@ func Extract(pass *analysis.Pass, node *ast.TypeSpec, obj types.Object) optional
 					if nativeName != mt.NativeName {
 						common.Errorf(pass, node, "declared type %s in typemap does not match native type %s",
 							mt.NativeName, nativeName)
-						return optional.None[*schema.TypeAlias]()
+						return
 					}
 					hasGoTypeMap = true
 				} else {
@@ -63,8 +63,9 @@ func Extract(pass *analysis.Pass, node *ast.TypeSpec, obj types.Object) optional
 				})
 			}
 			alias.Type = &schema.Any{}
-			return optional.Some(alias)
 		}
+	}) {
+		return optional.Some(alias)
 	} else if _, ok := alias.Type.(*schema.Any); ok &&
 		!strings.HasPrefix(qualifiedNameFromSelectorExpr(pass, node.Type), "ftl") {
 		alias.Metadata = append(alias.Metadata, &schema.MetadataTypeMap{
diff --git a/go-runtime/schema/typeenum/analyzer.go b/go-runtime/schema/typeenum/analyzer.go
index 1dc6864c41..7d08d9eb82 100644
--- a/go-runtime/schema/typeenum/analyzer.go
+++ b/go-runtime/schema/typeenum/analyzer.go
@@ -43,7 +43,7 @@ func Extract(pass *analysis.Pass) (interface{}, error) {
 			Pos:  common.GoPosToSchemaPos(pass.Fset, node.Pos()),
 			Name: strcase.ToUpperCamel(node.Name.Name),
 		}
-		if md, ok := common.GetFactForObject[*common.ExtractedMetadata](pass, obj).Get(); ok {
+		common.ApplyMetadata[*schema.Enum](pass, obj, func(md *common.ExtractedMetadata) {
 			enum.Comments = md.Comments
 			enum.Export = md.IsExported
 
@@ -62,7 +62,7 @@ func Extract(pass *analysis.Pass) (interface{}, error) {
 				}
 				common.MarkNeedsExtraction(pass, obj)
 			}
-		}
+		})
 		if iType.NumMethods() > 0 {
 			common.MarkMaybeTypeEnum(pass, obj, enum)
 		}
diff --git a/go-runtime/schema/verb/analyzer.go b/go-runtime/schema/verb/analyzer.go
index 4095b441b2..7bcad54d71 100644
--- a/go-runtime/schema/verb/analyzer.go
+++ b/go-runtime/schema/verb/analyzer.go
@@ -17,17 +17,16 @@ import (
 var Extractor = common.NewDeclExtractor[*schema.Verb, *ast.FuncDecl]("verb", Extract)
 
 func Extract(pass *analysis.Pass, root *ast.FuncDecl, obj types.Object) optional.Option[*schema.Verb] {
-	md, ok := common.GetFactForObject[*common.ExtractedMetadata](pass, obj).Get()
-	if !ok {
-		return optional.None[*schema.Verb]()
-	}
-
 	verb := &schema.Verb{
-		Pos:      common.GoPosToSchemaPos(pass.Fset, root.Pos()),
-		Name:     strcase.ToLowerCamel(root.Name.Name),
-		Comments: md.Comments,
-		Export:   md.IsExported,
-		Metadata: md.Metadata,
+		Pos:  common.GoPosToSchemaPos(pass.Fset, root.Pos()),
+		Name: strcase.ToLowerCamel(root.Name.Name),
+	}
+	if !common.ApplyMetadata[*schema.Verb](pass, obj, func(md *common.ExtractedMetadata) {
+		verb.Comments = md.Comments
+		verb.Export = md.IsExported
+		verb.Metadata = md.Metadata
+	}) {
+		return optional.None[*schema.Verb]()
 	}
 
 	fnt := obj.(*types.Func)             //nolint:forcetypeassert
@@ -64,9 +63,14 @@ func Extract(pass *analysis.Pass, root *ast.FuncDecl, obj types.Object) optional
 }
 
 func checkSignature(pass *analysis.Pass, node *ast.FuncDecl, sig *types.Signature) (req, resp optional.Option[*types.Var]) {
+	if expVerbName := strcase.ToUpperCamel(node.Name.Name); node.Name.Name != expVerbName {
+		common.Errorf(pass, node, "unexpected verb name %q, did you mean to use %q instead?", node.Name.Name,
+			expVerbName)
+		return optional.None[*types.Var](), optional.None[*types.Var]()
+	}
+
 	params := sig.Params()
 	results := sig.Results()
-
 	if params.Len() > 2 {
 		common.Errorf(pass, node, "must have at most two parameters (context.Context, struct)")
 	}