Add the discovery flag and method, along with model package

We want to be able to output schemas when we run in discovery mode so
this PR adds that flag along with the associated method to do so.

It just goes through the streams and calls Schema() along with adding a
few other fields. In the future we will also need to generate metadata
as part of that (and consume the file back in).

The other change here is the start of pulling out the models we use to
generate schemas and serialize data into their own package.

This PR just pulls out the schema itself - but a followup will add a
bunch of models representing entities in our API that we expose. Will
just follow a simple pattern of having a schema / serialize method on
them.

Author: rliddler

cmd/tap-incident/cmd/app.go

@@ -28,9 +28,10 @@ var (
 	app = kingpin.New("tap-incident", "Extract data from incident.io for use with Singer").Version(versionStanza())
 
 	// Global flags
-	debug       = app.Flag("debug", "Enable debug logging").Default("false").Bool()
-	configFile  = app.Flag("config", "Configuration file").ExistingFile()
-	catalogFile = app.Flag("catalog", "If set, allows filtering which streams would be synced").ExistingFile()
+	debug         = app.Flag("debug", "Enable debug logging").Default("false").Bool()
+	configFile    = app.Flag("config", "Configuration file").ExistingFile()
+	catalogFile   = app.Flag("catalog", "If set, allows filtering which streams would be synced").ExistingFile()
+	discoveryMode = app.Flag("discover", "If set, only outputs the catalog and exits").Default("false").Bool()
 )
 
 func Run(ctx context.Context) (err error) {
@@ -79,7 +80,11 @@ func Run(ctx context.Context) (err error) {
 	// can be streamed separately.
 	ol := tap.NewOutputLogger(os.Stdout)
 
-	err = tap.Run(ctx, logger, ol, cl)
+	if *discoveryMode {
+		err = tap.Discover(ctx, logger, ol)
+	} else {
+		err = tap.Sync(ctx, logger, ol, cl)
+	}
 	if err != nil {
 		return err
 	}

go.mod

@@ -5,6 +5,7 @@ go 1.21
 require (
 	github.com/alecthomas/kingpin/v2 v2.3.2
 	github.com/deepmap/oapi-codegen v1.12.4
+	github.com/fatih/structs v1.1.0
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-kit/log v0.2.0
 	github.com/go-ozzo/ozzo-validation v3.6.0+incompatible

go.sum

@@ -20,6 +20,8 @@ github.com/deepmap/oapi-codegen v1.12.4/go.mod h1:3lgHGMu6myQ2vqbbTXH2H1o4eXFTGn
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.12.0 h1:mRhaKNwANqRgUBGKmnI5ZxEk7QXmjQeCcuYFMX2bfcc=
 github.com/fatih/color v1.12.0/go.mod h1:ELkj/draVOlAH/xkhN6mQ50Qd0MPOk5AAr3maGEBuJM=
+github.com/fatih/structs v1.1.0 h1:Q7juDM0QtcnhCpeyLGQKyg4TOIghuNXrkL32pHAUMxo=
+github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
 github.com/getkin/kin-openapi v0.107.0 h1:bxhL6QArW7BXQj8NjXfIJQy680NsMKd25nwhvpCXchg=
 github.com/getkin/kin-openapi v0.107.0/go.mod h1:9Dhr+FasATJZjS4iOLvB0hkaxgYdulrNYm2e9epLWOo=
 github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=

model/schema.go

@@ -0,0 +1,90 @@
+package model
+
+import (
+	"strings"
+
+	"github.com/fatih/structs"
+)
+
+// Schema is a JSON schema for a stream.
+type Schema struct {
+	// Type is the type of the schema, e.g. "object" - for some reason singer docs
+	// have this as an array and often nullable eg: `"type": ["null", "object"]`
+	Type []string `json:"type"`
+	// HasAdditionalProperties indicates whether the schema allows additional properties
+	// not defined in the schema.
+	HasAdditionalProperties bool `json:"additionalProperties"`
+	// Properties is a map of property names to their schema.
+	Properties map[string]Property `json:"properties"`
+}
+
+// Property is a property in a JSON schema.
+type Property struct {
+	// Types is a list of types that this property can be, e.g. "string" or "integer".
+	Types []string `json:"type"`
+	// CustomFormat	is a custom format for this property, e.g. "date-time".
+	CustomFormat string `json:"format,omitempty"`
+	// For nested structures a property can have its own properties.
+	Properties map[string]Property `json:"properties,omitempty"`
+	// For array structures we define the type of the items in the array
+	Items *ArrayItem `json:"items,omitempty"`
+}
+
+// ArrayItem is the type and properties of an item in an array.
+type ArrayItem struct {
+	Type       string              `json:"type"`
+	Properties map[string]Property `json:"properties,omitempty"`
+}
+
+func (s Property) IsBoolean() bool {
+	return s.hasType("boolean")
+}
+
+func (s Property) IsNumber() bool {
+	return s.hasType("number")
+}
+
+func (s Property) IsInteger() bool {
+	return s.hasType("integer")
+}
+
+func (s Property) hasType(typeName string) bool {
+	for _, t := range s.Types {
+		if strings.EqualFold(t, typeName) {
+			return true
+		}
+	}
+	return false
+}
+
+func (s Property) IsDateTime() bool {
+	return s.CustomFormat == "date-time"
+}
+
+// As a shortcut for simple leaf nodes we can just dump everything (we can also dump everything higher level probably too)
+// If we're just going to dump everything why bother with serialisers? Good question.
+// a) Initial thoughts were that we want some control on the fields we output potentially - for example
+// ignoring deprecated fields.
+// b) We also might need to be cleverer when it comes to catalog config that enables / disables optional fields
+// in the output.
+//
+// Keeping this as a single callsite so it's easy to find where we're doing this in future.
+func DumpToMap(input interface{}) map[string]any {
+	structs.DefaultTagName = "json"
+	return structs.Map(input)
+}
+
+func AsOptional(p Property) Property {
+	p.Types = append(p.Types, "null")
+	return p
+}
+
+func AsArray(p Property) Property {
+	return Property{
+		Types: []string{"array"},
+		Items: &ArrayItem{
+			Type:       "object",
+			Properties: p.Properties,
+		},
+	}
+}

tap/catalog.go

@@ -0,0 +1,43 @@
+package tap
+
+import "github.com/incident-io/singer-tap/model"
+
+// A catalog can contain several streams or "entries"
+type CatalogEntry struct {
+	// Name of the stream
+	Stream string `json:"stream"`
+
+	// Unique identifier for this stream
+	// Allows for multiple sources that have duplicate stream names
+	TapStreamID string `json:"tap_stream_id"`
+
+	// The given schema for this stream
+	Schema model.Schema `json:"schema"`
+
+	// Optional metadata for this stream
+	// Metadata *[]Metadata `json:"metadata,omitempty"`
+}
+
+// Actual catalog that we export
+// contains an array of all our streams
+type Catalog struct {
+	Streams []CatalogEntry `json:"streams"`
+}
+
+func NewCatalog(streams map[string]Stream) *Catalog {
+	entries := []CatalogEntry{}
+
+	for name, stream := range streams {
+		catalogEntry := CatalogEntry{
+			Stream:      name,
+			TapStreamID: name,
+			Schema:      *stream.Output().Schema,
+		}
+
+		entries = append(entries, catalogEntry)
+	}
+
+	return &Catalog{
+		Streams: entries,
+	}
+}

tap/output.go

@@ -4,7 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"strings"
+
+	"github.com/incident-io/singer-tap/model"
 )
 
 // OutputType is the type of Singer tap output for each message.
@@ -29,7 +30,7 @@ type Output struct {
 	// Stream is the name of the stream, e.g. "users"
 	Stream string `json:"stream,omitempty"`
 	// Schema is the schema of the stream, if Type == "SCHEMA", in JSON schema format.
-	Schema *Schema `json:"schema,omitempty"`
+	Schema *model.Schema `json:"schema,omitempty"`
 	// Record is a record from the stream, if Type == "RECORD".
 	Record map[string]any `json:"record,omitempty"`
 	// TimeExtracted is the time that this record was extracted, if Type == "RECORD".
@@ -46,50 +47,6 @@ type Output struct {
 	BookmarkProperties []string `json:"bookmark_properties,omitempty"`
 }
 
-// Schema is a JSON schema for a stream.
-type Schema struct {
-	// Type is the type of the schema, e.g. "object"
-	Type []string `json:"type"`
-	// HasAdditionalProperties indicates whether the schema allows additional properties
-	// not defined in the schema.
-	HasAdditionalProperties bool `json:"additionalProperties"`
-	// Properties is a map of property names to their schema.
-	Properties map[string]Property `json:"properties"`
-}
-
-// Property is a property in a JSON schema.
-type Property struct {
-	// Types is a list of types that this property can be, e.g. "string" or "integer".
-	Types []string `json:"type"`
-	// CustomFormat	is a custom format for this property, e.g. "date-time".
-	CustomFormat string `json:"format,omitempty"`
-}
-
-func (s Property) IsBoolean() bool {
-	return s.hasType("boolean")
-}
-
-func (s Property) IsNumber() bool {
-	return s.hasType("number")
-}
-
-func (s Property) IsInteger() bool {
-	return s.hasType("integer")
-}
-
-func (s Property) hasType(typeName string) bool {
-	for _, t := range s.Types {
-		if strings.EqualFold(t, typeName) {
-			return true
-		}
-	}
-	return false
-}
-
-func (s Property) IsDateTime() bool {
-	return s.CustomFormat == "date-time"
-}
-
 // OutputLogger is a logger that logs to STDOUT in the format expected by the downstream
 // Singer target.
 type OutputLogger struct {
@@ -113,3 +70,17 @@ func (o *OutputLogger) Log(op *Output) error {
 
 	return nil
 }
+
+func (o *OutputLogger) CataLog(catalog *Catalog) error {
+	data, err := json.Marshal(catalog)
+	if err != nil {
+		return err
+	}
+
+	_, err = fmt.Fprintln(o.w, string(data))
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

tap/stream_incidents.go

@@ -5,6 +5,7 @@ import (
 
 	kitlog "github.com/go-kit/log"
 	"github.com/incident-io/singer-tap/client"
+	"github.com/incident-io/singer-tap/model"
 	"github.com/pkg/errors"
 	"github.com/samber/lo"
 )
@@ -20,9 +21,9 @@ func (s *StreamIncidents) Output() *Output {
 	return &Output{
 		Type:   OutputTypeSchema,
 		Stream: "incidents",
-		Schema: &Schema{
+		Schema: &model.Schema{
 			Type: []string{"object"},
-			Properties: map[string]Property{
+			Properties: map[string]model.Property{
 				"id": {
 					Types: []string{"string"},
 				},

tap/tap.go

@@ -8,7 +8,7 @@ import (
 	"github.com/incident-io/singer-tap/client"
 )
 
-func Run(ctx context.Context, logger kitlog.Logger, ol *OutputLogger, cl *client.ClientWithResponses) error {
+func Sync(ctx context.Context, logger kitlog.Logger, ol *OutputLogger, cl *client.ClientWithResponses) error {
 	for name, stream := range streams {
 		logger := kitlog.With(logger, "stream", name)
 
@@ -40,3 +40,13 @@ func Run(ctx context.Context, logger kitlog.Logger, ol *OutputLogger, cl *client
 
 	return nil
 }
+
+func Discover(ctx context.Context, logger kitlog.Logger, ol *OutputLogger) error {
+	catalog := NewCatalog(streams)
+
+	if err := ol.CataLog(catalog); err != nil {
+		return err
+	}
+
+	return nil
+}