Add Nats Jetstream

README.md

@@ -1,6 +1,6 @@
 # Simian Go
 
-[![Coverage Status](https://coveralls.io/repos/github/simiancreative/simiango/badge.svg?branch=master)](https://coveralls.io/github/simiancreative/simiango?branch=master)
+[![codecov](https://codecov.io/gh/simiancreative/simiango/graph/badge.svg?token=OYPECDAI3X)](https://codecov.io/gh/simiancreative/simiango)
 [![tests](https://github.com/simiancreative/simiango/workflows/CI/badge.svg)](https://github.com/simiancreative/simiango/actions)
 [![Godoc](http://img.shields.io/badge/godoc-reference-blue.svg?style=flat)](https://godoc.org/github.com/simiancreative/simiango) 
 [![license](http://img.shields.io/badge/license-MIT-red.svg?style=flat)](https://github.com/simiancreative/simiango/blob/master/LICENSE)

go.mod

@@ -1,6 +1,8 @@
 module github.com/simiancreative/simiango
 
-go 1.21.7
+go 1.22
+
+toolchain go1.22.10
 
 require (
 	github.com/AlecAivazis/survey/v2 v2.3.5
@@ -28,6 +30,8 @@ require (
 	github.com/joho/godotenv v1.4.0
 	github.com/kr/pretty v0.3.0
 	github.com/mandrigin/gin-spa v0.0.0-20200212133200-790d0c0c7335
+	github.com/nats-io/nats-server/v2 v2.10.24
+	github.com/nats-io/nats.go v1.38.0
 	github.com/p768lwy3/gin-server-timing v0.0.0-20200316080543-ab69795cf847
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.13.0
@@ -72,7 +76,7 @@ require (
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect
-	github.com/klauspost/compress v1.16.0 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
@@ -82,10 +86,14 @@ require (
 	github.com/mattn/go-runewidth v0.0.13 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b // indirect
+	github.com/minio/highwayhash v1.0.3 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/muesli/reflow v0.2.1-0.20210115123740-9e1d0d53df68 // indirect
 	github.com/muesli/termenv v0.11.1-0.20220204035834-5ac8409525e0 // indirect
+	github.com/nats-io/jwt/v2 v2.7.3 // indirect
+	github.com/nats-io/nkeys v0.4.9 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/onsi/gomega v1.27.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.1 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15 // indirect
@@ -103,6 +111,7 @@ require (
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/term v0.27.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
+	golang.org/x/time v0.8.0 // indirect
 	google.golang.org/protobuf v1.34.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -351,8 +351,8 @@ github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNU
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.15.7/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
-github.com/klauspost/compress v1.16.0 h1:iULayQNOReoYUe+1qtKOqw9CwJv3aNQu8ivo7lw1HU4=
-github.com/klauspost/compress v1.16.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM=
 github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
@@ -414,6 +414,8 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0j
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b h1:j7+1HpAFS1zy5+Q4qx1fWh90gTKwiN4QCGoY9TWyyO4=
 github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE=
+github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q=
+github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
 github.com/mitchellh/mapstructure v0.0.0-20170523030023-d0303fe80992/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
@@ -429,6 +431,16 @@ github.com/muesli/termenv v0.11.1-0.20220204035834-5ac8409525e0 h1:STjmj0uFfRryL
 github.com/muesli/termenv v0.11.1-0.20220204035834-5ac8409525e0/go.mod h1:Bd5NYQ7pd+SrtBSrSNoBBmXlcY8+Xj4BMJgh8qcZrvs=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/nats-io/jwt/v2 v2.7.3 h1:6bNPK+FXgBeAqdj4cYQ0F8ViHRbi7woQLq4W29nUAzE=
+github.com/nats-io/jwt/v2 v2.7.3/go.mod h1:GvkcbHhKquj3pkioy5put1wvPxs78UlZ7D/pY+BgZk4=
+github.com/nats-io/nats-server/v2 v2.10.24 h1:KcqqQAD0ZZcG4yLxtvSFJY7CYKVYlnlWoAiVZ6i/IY4=
+github.com/nats-io/nats-server/v2 v2.10.24/go.mod h1:olvKt8E5ZlnjyqBGbAXtxvSQKsPodISK5Eo/euIta4s=
+github.com/nats-io/nats.go v1.38.0 h1:A7P+g7Wjp4/NWqDOOP/K6hfhr54DvdDQUznt5JFg9XA=
+github.com/nats-io/nats.go v1.38.0/go.mod h1:IGUM++TwokGnXPs82/wCuiHS02/aKrdYUQkU8If6yjw=
+github.com/nats-io/nkeys v0.4.9 h1:qe9Faq2Gxwi6RZnZMXfmGMZkg3afLLOtrU+gDZJ35b0=
+github.com/nats-io/nkeys v0.4.9/go.mod h1:jcMqs+FLG+W5YO36OX6wFIFcmpdAns+w1Wm6D3I/evE=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
@@ -799,6 +811,7 @@ golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
 golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
@@ -828,6 +841,8 @@ golang.org/x/time v0.0.0-20170424234030-8be79e1e0910/go.mod h1:tRJNPiyCQ0inRvYxb
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg=
+golang.org/x/time v0.8.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=

messaging/natsjs/argo-event.go

@@ -0,0 +1,73 @@
+package natsjs
+
+import (
+	"encoding/base64"
+	"encoding/json"
+
+	"github.com/simiancreative/simiango/errors"
+)
+
+func ArgoEventFromString(str string) (ArgoEvent, error) {
+	var e ArgoEvent
+	err := json.Unmarshal([]byte(str), &e)
+	return e, err
+}
+
+func ArgoUnmarshalEvent(dest interface{}, str string) (err error) {
+	newEvent, err := ArgoEventFromString(str)
+	if err != nil {
+		return errors.Wrap(err, "failed to unmarshal event")
+	}
+
+	str, err = newEvent.JSONBody()
+	if err != nil {
+		return errors.Wrap(err, "failed to extract JSON body")
+	}
+
+	err = json.Unmarshal([]byte(str), &dest)
+	if err != nil {
+		return errors.Wrap(err, "failed to unmarshal event")
+	}
+
+	return nil
+}
+
+// EventContext represents the context of an event
+type ArgoEventContext struct {
+	ID              string `json:"id"`
+	Source          string `json:"source"`
+	SpecVersion     string `json:"specversion"`
+	Type            string `json:"type"`
+	DataContentType string `json:"datacontenttype"`
+	Subject         string `json:"subject"`
+	Time            string `json:"time"`
+}
+
+// Event represents an event received from an argo event source
+type ArgoEvent struct {
+	Context ArgoEventContext `json:"context"`
+	Data    string           `json:"data"`
+}
+
+type ArgoEventData struct {
+	Subject string      `json:"subject"`
+	Body    interface{} `json:"body"`
+}
+
+func (e ArgoEvent) JSONBody() (string, error) {
+	var data ArgoEventData
+
+	bytes, err := base64.StdEncoding.DecodeString(e.Data)
+	if err != nil {
+		return "", err
+	}
+
+	err = json.Unmarshal(bytes, &data)
+	if err != nil {
+		return "", err
+	}
+
+	bytes, err = json.Marshal(data.Body)
+
+	return string(bytes), err
+}

messaging/natsjs/argo-event_test.go

@@ -0,0 +1,182 @@
+package natsjs_test
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"testing"
+
+	"github.com/simiancreative/simiango/messaging/natsjs"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestArgoEventFromString(t *testing.T) {
+	testEvent := natsjs.ArgoEvent{
+		Context: natsjs.ArgoEventContext{
+			ID:              "testID",
+			Source:          "testSource",
+			SpecVersion:     "testSpecVersion",
+			Type:            "testType",
+			DataContentType: "testDataContentType",
+			Subject:         "testSubject",
+			Time:            "testTime",
+		},
+		Data: "testData",
+	}
+
+	bytes, _ := json.Marshal(testEvent)
+	str := string(bytes)
+
+	event, err := natsjs.ArgoEventFromString(str)
+
+	assert.NoError(t, err)
+	assert.Equal(t, testEvent.Context.ID, event.Context.ID)
+
+	// Test with invalid JSON
+	_, err = natsjs.ArgoEventFromString("{invalid json}")
+	assert.Error(t, err)
+}
+
+func TestJSONBody(t *testing.T) {
+	testData := natsjs.ArgoEventData{
+		Subject: "testSubject",
+		Body:    "testBody",
+	}
+
+	bytes, _ := json.Marshal(testData)
+	str := base64.StdEncoding.EncodeToString(bytes)
+
+	testEvent := natsjs.ArgoEvent{
+		Context: natsjs.ArgoEventContext{
+			ID:              "testID",
+			Source:          "testSource",
+			SpecVersion:     "testSpecVersion",
+			Type:            "testType",
+			DataContentType: "testDataContentType",
+			Subject:         "testSubject",
+			Time:            "testTime",
+		},
+		Data: str,
+	}
+
+	body, err := testEvent.JSONBody()
+
+	assert.NoError(t, err)
+	assert.Equal(t, `"testBody"`, body)
+
+	// Test with invalid base64 data
+	testEvent.Data = "invalid base64"
+
+	_, err = testEvent.JSONBody()
+	assert.Error(t, err)
+}
+
+func TestBadJSONBody(t *testing.T) {
+	str := base64.StdEncoding.EncodeToString([]byte("invalidJSON"))
+
+	testEvent := natsjs.ArgoEvent{
+		Context: natsjs.ArgoEventContext{
+			ID:              "testID",
+			Source:          "testSource",
+			SpecVersion:     "testSpecVersion",
+			Type:            "testType",
+			DataContentType: "testDataContentType",
+			Subject:         "testSubject",
+			Time:            "testTime",
+		},
+		Data: str,
+	}
+
+	body, err := testEvent.JSONBody()
+	assert.Error(t, err)
+	assert.Equal(t, "", body)
+}
+
+func TestArgoUnmarshalEvent(t *testing.T) {
+	// Sample JSON string representing an ArgoEvent
+	sampleJSON := `{
+		"context": {
+			"id": "1234",
+			"source": "source",
+			"specversion": "1.0",
+			"type": "type",
+			"datacontenttype": "application/json",
+			"subject": "subject",
+			"time": "2023-10-01T00:00:00Z"
+		},
+		"data": "eyJzdWJqZWN0IjoiZXhhbXBsZSIsImJvZHkiOnsia2V5IjoidmFsdWUifX0="
+	}`
+
+	// Destination variable
+	var dest map[string]interface{}
+
+	// Call the function
+	err := natsjs.ArgoUnmarshalEvent(&dest, sampleJSON)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	// Verify the results
+	expectedBody := map[string]interface{}{"key": "value"}
+	if dest["key"] != expectedBody["key"] {
+		t.Errorf("Expected body %v, got %v", expectedBody, dest)
+	}
+}
+
+func TestArgoUnmarshalEvent_Errors(t *testing.T) {
+	// Test case 1: Invalid JSON string
+	invalidJSON := `{"context": { "id": "1234" "source": "source" }}` // Missing comma
+	var dest1 map[string]interface{}
+	err := natsjs.ArgoUnmarshalEvent(&dest1, invalidJSON)
+	assert.Error(t, err)
+
+	// Test case 2: Invalid base64 data
+	invalidBase64 := `{
+		"context": {
+			"id": "1234",
+			"source": "source",
+			"specversion": "1.0",
+			"type": "type",
+			"datacontenttype": "application/json",
+			"subject": "subject",
+			"time": "2023-10-01T00:00:00Z"
+		},
+		"data": "invalid_base64_data"
+	}`
+	var dest2 map[string]interface{}
+	err = natsjs.ArgoUnmarshalEvent(&dest2, invalidBase64)
+	assert.Error(t, err)
+
+	// test case 3: invalid json body
+	invalidjsonbody := `{
+		"context": {
+			"id": "1234",
+			"source": "source",
+			"specversion": "1.0",
+			"type": "type",
+			"datacontenttype": "application/json",
+			"subject": "subject",
+			"time": "2023-10-01t00:00:00z"
+		},
+		"data": "W30K" // {"subject":"example","body":{"foo":"bar"}}
+	}`
+	var dest3 map[string]interface{}
+	err = natsjs.ArgoUnmarshalEvent(&dest3, invalidjsonbody)
+	assert.Error(t, err)
+
+	// test case 3: invalid json body
+	invalidjsonMessage := `{
+		"context": {
+			"id": "1234",
+			"source": "source",
+			"specversion": "1.0",
+			"type": "type",
+			"datacontenttype": "application/json",
+			"subject": "subject",
+			"time": "2023-10-01t00:00:00z"
+		},
+		"data": "eyJzdWJqZWN0IjoiZXhhbXBsZSIsImJvZHkiOiJ7XSJ9Cg=="
+	}`
+	var dest4 map[string]interface{}
+	err = natsjs.ArgoUnmarshalEvent(&dest4, invalidjsonMessage)
+	assert.Error(t, err)
+}