Kafka streaming, projection state recovery improvements, documentation, code coverage

.github/workflows/ci.yml

@@ -9,7 +9,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v4
         with:
-          go-version: ^1.19
+          go-version: ^1.21
         id: go
 
       - run: cd cmd/mkunion; go get -v -t -d ./...
@@ -39,3 +39,12 @@ jobs:
             echo "Go tests failed after all retries."
             exit 1
           fi
+
+      - run: |
+          go test -race -coverprofile=coverage.out -covermode=atomic ./... || true
+
+      - name: Upload coverage reports to Codecov
+        uses: codecov/[email protected]
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}
+          slug: widmogrod/mkunion

.github/workflows/docs.yml

@@ -0,0 +1,29 @@
+name: docs
+on:
+  push:
+    branches:
+      - master
+      - main
+permissions:
+  contents: write
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Configure Git Credentials
+        run: |
+          git config user.name github-actions[bot]
+          git config user.email 41898282+github-actions[bot]@users.noreply.github.com
+      - uses: actions/setup-python@v5
+        with:
+          python-version: 3.x
+      - run: echo "cache_id=$(date --utc '+%V')" >> $GITHUB_ENV
+      - uses: actions/cache@v4
+        with:
+          key: mkdocs-material-${{ env.cache_id }}
+          path: .cache
+          restore-keys: |
+            mkdocs-material-
+      - run: pip install mkdocs-material
+      - run: mkdocs gh-deploy --force

cmd/mkunion/main.go

@@ -12,6 +12,7 @@ import (
 	"os"
 	"os/signal"
 	"path"
+	"sort"
 	"strings"
 	"syscall"
 )
@@ -26,7 +27,7 @@ func main() {
 	var app *cli.App
 	app = &cli.App{
 		Name:                   shared.Program,
-		Description:            "VisitorGenerator union type and visitor pattern gor golang",
+		Description:            "Strongly typed union type in golang.",
 		EnableBashCompletion:   true,
 		UseShortOptionHandling: true,
 		Flags: []cli.Flag{
@@ -35,17 +36,6 @@ func main() {
 				Aliases:  []string{"n"},
 				Required: false,
 			},
-			&cli.StringFlag{
-				Name:     "skip-extension",
-				Aliases:  []string{"skip-ext"},
-				Value:    "",
-				Required: false,
-			},
-			&cli.StringFlag{
-				Name:     "include-extension",
-				Aliases:  []string{"inc-ext"},
-				Required: false,
-			},
 			&cli.StringSliceFlag{
 				Name:      "input-go-file",
 				Aliases:   []string{"i", "input"},
@@ -59,8 +49,8 @@ func main() {
 				Value:    false,
 			},
 			&cli.BoolFlag{
-				Name:     "no-compact",
-				Required: false,
+				Name:  "type-registry",
+				Value: false,
 			},
 		},
 		Action: func(c *cli.Context) error {
@@ -82,13 +72,19 @@ func main() {
 				cli.ShowAppHelpAndExit(c, 1)
 			}
 
+			packages := make(map[string]*shape.InferredInfo)
+
 			for _, sourcePath := range sourcePaths {
 				// file name without extension
 				inferred, err := shape.InferFromFile(sourcePath)
 				if err != nil {
 					return err
 				}
 
+				if _, ok := packages[inferred.PackageImportName()]; !ok {
+					packages[inferred.PackageImportName()] = inferred
+				}
+
 				contents, err := GenerateUnions(inferred)
 				if err != nil {
 					return fmt.Errorf("failed generating union in %s: %w", sourcePath, err)
@@ -118,14 +114,42 @@ func main() {
 				if err != nil {
 					return fmt.Errorf("failed saving shape in %s: %w", sourcePath, err)
 				}
+			}
+
+			if c.Bool("type-registry") {
+				for _, inferred := range packages {
+					dir := path.Dir(inferred.FileName())
 
+					// walk through all *.go files in the same directory
+					// and generate type registry for all inferred packages
+					// in the same directory
+
+					indexed, err := shape.NewIndexTypeInDir(dir)
+					if err != nil {
+						return fmt.Errorf("mkunion: failed indexing types in directory %s: %w", dir, err)
+					}
+
+					if len(indexed.IndexedShapes()) > 0 {
+						contents, err := GenerateTypeRegistry(indexed)
+						if err != nil {
+							return fmt.Errorf("mkunion: failed walking through directory %s: %w", dir, err)
+						}
+
+						regPath := path.Join(dir, "types.go")
+						err = SaveFile(contents, regPath, "reg_gen")
+						if err != nil {
+							return fmt.Errorf("mkunion: failed saving type registry in %s: %w", regPath, err)
+						}
+					}
+				}
 			}
 
 			return nil
 		},
 		Commands: []*cli.Command{
 			{
-				Name: "match",
+				Name:        "match",
+				Description: "Generate custom pattern matching function",
 				Flags: []cli.Flag{
 					&cli.StringFlag{
 						Name:     "name",
@@ -172,7 +196,8 @@ func main() {
 				},
 			},
 			{
-				Name: "shape-export",
+				Name:        "shape-export",
+				Description: "Generate typescript types from golang types, and enable end-to-end type safety.",
 				Flags: []cli.Flag{
 					&cli.StringFlag{
 						Name:        "language",
@@ -270,6 +295,10 @@ func GenerateUnions(inferred *shape.InferredInfo) (bytes.Buffer, error) {
 		}
 		shapesContents.Write(contents)
 
+		if shape.TagHasOption(union.Tags, "mkunion", "noserde") {
+			continue
+		}
+
 		genSerde := generators.NewSerdeJSONUnion(union)
 		genSerde.SkipImportsAndPackage(true)
 
@@ -485,3 +514,59 @@ func SaveFile(contents bytes.Buffer, sourcePath string, infix string) error {
 	}
 	return nil
 }
+
+func GenerateTypeRegistry(inferred *shape.IndexedTypeWalker) (bytes.Buffer, error) {
+	packageName := inferred.PackageName()
+
+	contents := bytes.Buffer{}
+	contents.WriteString("// Code generated by mkunion. DO NOT EDIT.\n")
+	contents.WriteString(fmt.Sprintf("package %s\n\n", packageName))
+
+	found := inferred.ExpandedShapes()
+	if len(found) == 0 {
+		return contents, nil
+	}
+
+	sortedKeys := make([]string, 0, len(found))
+	for k := range found {
+		sortedKeys = append(sortedKeys, k)
+	}
+
+	sort.Strings(sortedKeys)
+
+	maps := []generators.PkgMap{
+		{
+			"shared": "github.com/widmogrod/mkunion/x/shared",
+		},
+	}
+	for _, key := range sortedKeys {
+		inst := found[key]
+		next := shape.ExtractPkgImportNamesForTypeInitialisation(inst)
+		maps = append(maps, next)
+	}
+	pkgMap := generators.MergePkgMaps(maps...)
+	delete(pkgMap, packageName)
+
+	contents.WriteString(generators.GenerateImports(pkgMap))
+
+	contents.WriteString("func init() {\n")
+	// generate type registry
+
+	for _, key := range sortedKeys {
+		inst := found[key]
+		instantiatedTypeName := shape.ToGoTypeName(inst,
+			shape.WithInstantiation(),
+			shape.WithRootPackage(packageName),
+		)
+		fullTypeName := shape.ToGoTypeName(inst,
+			shape.WithInstantiation(),
+			shape.WithPkgImportName(),
+		)
+
+		contents.WriteString(fmt.Sprintf("\tshared.TypeRegistryStore[%s](%q)\n", instantiatedTypeName, fullTypeName))
+	}
+
+	contents.WriteString("}\n")
+
+	return contents, nil
+}

dev/bootstrap.sh

@@ -36,10 +36,14 @@ echo "export AWS_ENDPOINT_URL=http://localhost:4566" >> $envrc_file
 echo "export OPENSEARCH_ADDRESS=http://localhost:9200" >> $envrc_file
 echo "export OPENSEARCH_USERNAME=admin" >> $envrc_file
 echo "export AWS_SQS_QUEUE_URL=$queue_url" >> $envrc_file
+echo "export KAFKA_SERVERS=localhost:9092" >> $envrc_file
 
 echo "Localstack is UI is at port"
 echo "http://localhost:8080"
 
+echo "Kafka UI is at port"
+echo "http://localhost:9088"
+
 ## check if it should stream logs, or just end
 ## don't trigger trap on exit
 if [ "$1" == "-nologs" ]; then

dev/compose.yml

@@ -38,7 +38,56 @@ services:
     volumes:
       - data01:/usr/share/opensearch/data
 
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.5.3
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  kafka:
+    image: confluentinc/cp-kafka:7.5.3
+    depends_on:
+      - zookeeper
+    ports:
+      - 9092:9092
+
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_NUM_PARTITIONS: 3
+      KAFKA_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://0.0.0.0:9092'
+
+    healthcheck:
+      test: [ "CMD", "kafka-topics", "--bootstrap-server", "kafka:9092", "--list" ]
+      interval: 5s
+      timeout: 10s
+      retries: 30
+      start_period: 10s
 
+  kafka-ui:
+    container_name: kafka-ui
+    image: provectuslabs/kafka-ui:latest
+    ports:
+      - 9088:8080
+    depends_on:
+      - kafka
+#      - schema-registry0
+#      - kafka-connect0
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
+      KAFKA_CLUSTERS_0_METRICS_PORT: 9997
+#      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry0:8085
+#      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: first
+#      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://kafka-connect0:8083
+      DYNAMIC_CONFIG_ENABLED: 'true'  # not necessary, added for tests
+      KAFKA_CLUSTERS_0_AUDIT_TOPICAUDITENABLED: 'true'
+      KAFKA_CLUSTERS_0_AUDIT_CONSOLEAUDITENABLED: 'true'
 volumes:
   data01:
     name: tmp

dev/docs.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+set -e
+
+cwd=$(dirname "$0")
+project_root=$(dirname "$cwd")
+
+if [ "$1" == "run" ]; then
+  docker run --rm -it -p 8000:8000 -v ${project_root}:/docs squidfunk/mkdocs-material
+elif [ "$1" == "build" ]; then
+  docker run --rm -it -v ${project_root}:/docs squidfunk/mkdocs-material build
+else
+  echo "Usage: $0 [run|build]"
+fi

docs/development/development.md

@@ -0,0 +1,35 @@
+---
+title: Contributing and development
+---
+
+# Contributing and development
+## Development
+
+Checkout repo and run:
+```
+./dev/bootstrap.sh
+```
+
+This command starts docker container with all necessary tools to develop and test `mkunion` project.
+
+In separate terminal run:
+```
+go generate ./...
+go test ./...
+```
+
+This will generate code and run tests.
+
+Note: Some tests are flaky (yes I know, I'm working on it), so if you see some test failing, please run it again.
+
+## Documentation
+
+To preview documentation run:
+```
+ ./dev/docs.sh run
+```
+
+## Contributing
+
+If you want to contribute to `mkunion` project, please open issue first to discuss your idea.
+I have opinions about how `mkunion` should work, how I want to evolve it, and I want to make sure that your idea fits into the project.