contracts

Single source of truth for MathTrail event schemas, generated Go client code, AsyncAPI specification, and AsyncAPI HTML documentation portal.

Overview

This repository defines the event contracts for the MathTrail EDA stack. It contains:

• Protobuf schemas — strongly-typed event definitions managed with buf
• Generated Go code — committed .pb.go files, importable as a Go module without running buf locally
• AsyncAPI v3 specification — machine-readable channel and message contracts
• AsyncAPI HTML portal — documentation portal built from the AsyncAPI spec via @asyncapi/generator, deployed to the cluster at /observability/eventcatalog

Event Pipeline

flowchart TD
    PG[(PostgreSQL)]
    CDC[RisingWave\nNative CDC]
    MV["RisingWave\nMaterialized Views\n(SQL transformations)"]
    AMQ["AutoMQ\n(S3 storage)"]
    APR["Apicurio Registry\n(Protobuf schemas)"]
    MA["Microservice\n(SASL/SCRAM-SHA-512)"]

    PG -->|"publication + replication slot"| CDC
    CDC --> MV
    MV -->|"Confluent Wire Format\n+ CloudEvents Binary Mode headers"| AMQ
    APR -.->|"schema registration"| AMQ
    AMQ -->|"Confluent wire format decode\nCE header parse"| MA
    APR -.->|"TopicValidator: schema_id validation"| MA

Loading

Quick Start

Open in devcontainer — buf, Go, Node.js, and just are pre-installed.

# Generate Go code from proto files
just generate

# Lint proto schemas
just lint

# Check for breaking changes against main
just breaking

# Format proto files
just fmt

# Full pipeline: generate + lint + go mod tidy
just build

Wire Format

Every message on AutoMQ uses Confluent Wire Format:

[0x00]  magic byte
[4 bytes]  schema_id (big-endian uint32, from Apicurio)
[n bytes]  protobuf binary payload

CloudEvents attributes are carried as Kafka record headers (Binary Mode):

Header key	Example value
ce_specversion	1.0
ce_type	com.mathtrail.students.onboarding.ready
ce_source	/mathtrail/risingwave
ce_id	550e8400-e29b-41d4-a716-446655440000
ce_time	2024-01-15T12:34:56Z

Using as a Go Module

Generated Go code is committed to gen/go/ and importable without running buf:

import (
    studentsv1 "github.com/mathtrail/contracts/gen/go/students/v1"
    "google.golang.org/protobuf/proto"
)

var msg studentsv1.StudentOnboardingReady
if err := proto.Unmarshal(rawBytes, &msg); err != nil {
    return err
}

Local development (before tagging a release) — add a replace directive to go.mod:

replace github.com/mathtrail/contracts => ../contracts

CI/CD — use a tagged release (v0.1.0) so the replace directive is not needed.

Adding a New Schema

1. Create or update a .proto file under proto/<domain>/v1/events.proto
2. Keep the schema self-contained — no import directives, use string for timestamps
3. Run just generate to regenerate Go code
4. Run just lint and just breaking to validate
5. Add the new subject to infra-streaming/infra/local/helm/apicurio/templates/schema-registration.yaml
6. Add the new channel and message to asyncapi/mathtrail-events.yaml
7. Commit — CI will validate lint, breaking changes, and AsyncAPI portal build

Schema Registration

Schemas are registered in Apicurio Registry via the Confluent compat v7 API on cluster startup:

POST /apis/ccompat/v7/subjects/{subject}/versions
Content-Type: application/vnd.schemaregistry.v1+json
Body: {"schemaType": "PROTOBUF", "schema": "<escaped .proto content>"}

Subject naming: {package}.{MessageName} — e.g. students.v1.StudentOnboardingReady. This ensures the subject name matches exactly what RisingWave uses in FORMAT PLAIN ENCODE PROTOBUF.

AsyncAPI Portal

The HTML portal is built from asyncapi/mathtrail-events.yaml using @asyncapi/generator and deployed to the cluster:

http://<cluster>/observability/eventcatalog

To build the Docker image locally:

just build-push-image k3d-mathtrail-registry.localhost:5050/eventcatalog:latest

CI builds and pushes the image on every merge to main.