Add postgres store (#18)

.github/workflows/ci.yaml

@@ -31,10 +31,23 @@ jobs:
         version: v1.54
         args: --verbose --timeout=3m
 
+    - name: Install docker-compose
+      run: |
+        sudo apt-get update
+        sudo apt-get -y install docker-compose
+
+    - name: Start docker-compose
+      run: |
+        docker-compose up -d
+
     - name: Run go test and generate coverage report
       run: |
         go test -v -coverprofile=coverage.out -coverpkg=./... ./...
 
+    - name: Stop docker-compose
+      run: |
+        docker-compose down
+
     - name: Upload coverage report to Codecov
       uses: codecov/codecov-action@v3 # nosemgrep
       with:

Dockerfile

@@ -0,0 +1,17 @@
+FROM golang:1.21 AS builder
+
+WORKDIR /app
+COPY . .
+
+ENV CGO_ENABLED 0
+RUN go build -o resonate .
+
+FROM scratch
+
+WORKDIR /app
+COPY --from=builder /app/resonate .
+
+EXPOSE 8001
+EXPOSE 50051
+
+ENTRYPOINT ["./resonate"]

cmd/serve.go

@@ -12,7 +12,7 @@ import (
 	"github.com/resonatehq/resonate/internal/api"
 	"github.com/resonatehq/resonate/internal/app/coroutines"
 	"github.com/resonatehq/resonate/internal/app/subsystems/aio/network"
-	"github.com/resonatehq/resonate/internal/app/subsystems/aio/store/sqlite"
+	"github.com/resonatehq/resonate/internal/app/subsystems/aio/store/postgres"
 	"github.com/resonatehq/resonate/internal/app/subsystems/api/grpc"
 	"github.com/resonatehq/resonate/internal/app/subsystems/api/http"
 	"github.com/resonatehq/resonate/internal/kernel/system"
@@ -57,7 +57,21 @@ var serveCmd = &cobra.Command{
 
 		// instatiate aio subsystems
 		network := network.New(10 * time.Second)
-		store, err := sqlite.New(sqlite.Config{Path: "resonate.db"})
+		// store, err := sqlite.New(sqlite.Config{Path: "resonate.db"})
+		// if err != nil {
+		// 	return err
+		// }
+
+		store, err := postgres.New(postgres.Config{
+			Host:            os.Getenv("POSTGRES_HOST"),
+			Port:            os.Getenv("POSTGRES_PORT"),
+			Username:        os.Getenv("POSTGRES_USER"),
+			Password:        os.Getenv("POSTGRES_PASSWORD"),
+			Database:        os.Getenv("POSTGRES_DB"),
+			MaxOpenConns:    3,
+			MaxIdleConns:    3,
+			ConnMaxIdleTime: 0,
+		})
 		if err != nil {
 			return err
 		}
@@ -66,7 +80,7 @@ var serveCmd = &cobra.Command{
 		api.AddSubsystem(http)
 		api.AddSubsystem(grpc)
 		aio.AddSubsystem(types.Network, network, 100, 1, 3)
-		aio.AddSubsystem(types.Store, store, 100, 10, 1)
+		aio.AddSubsystem(types.Store, store, 100, 10, 3)
 
 		// start api/aio
 		if err := api.Start(); err != nil {

docker-compose.yml

@@ -0,0 +1,32 @@
+version: '3'
+services:
+  postgres:
+    image: postgres:15
+    container_name: postgres
+    environment:
+      POSTGRES_USER: username
+      POSTGRES_PASSWORD: password
+      POSTGRES_DB: resonate
+    ports:
+    - "5432:5432"
+    read_only: true
+    tmpfs: /var/run/postgresql:rw,nosuid,nodev,noexec,mode=1777
+    security_opt:
+    - no-new-privileges:true
+
+  # app:
+  #   build:
+  #     context: "."
+  #   container_name: resonate
+  #   command: ["serve"]
+  #   depends_on:
+  #   - postgres
+  #   environment:
+  #     POSTGRES_HOST: postgres
+  #     POSTGRES_PORT: 5432
+  #     POSTGRES_USER: username
+  #     POSTGRES_PASSWORD: password
+  #     POSTGRES_DB: resonate
+  #   ports:
+  #   - "8001:8001"
+  #   - "50051:50051"

go.mod

@@ -4,7 +4,9 @@ go 1.21
 
 require (
 	github.com/gin-gonic/gin v1.9.1
+	github.com/lib/pq v1.10.9
 	github.com/mattn/go-sqlite3 v1.14.16
+	github.com/prometheus/client_golang v1.16.0
 	github.com/spf13/cobra v1.7.0
 	github.com/stretchr/testify v1.8.3
 	google.golang.org/grpc v1.56.1
@@ -27,17 +29,18 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.4 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.2.4 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.16.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect
 	github.com/prometheus/procfs v0.10.1 // indirect
+	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.11 // indirect

go.sum

@@ -9,6 +9,7 @@ github.com/chenzhuoyu/base64x v0.0.0-20211019084208-fb5309c8db06/go.mod h1:DH46F
 github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 h1:qSGYFH7+jGhDF8vLC+iwCD4WpbV1EBDSzWkJODFLams=
 github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311/go.mod h1:b583jCggY9gE99b6G5LEC39OIiVsWj+R97kbl5odCEk=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -44,8 +45,14 @@ github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHm
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.4 h1:acbojRNwl3o09bUq+yDCtZFc1aiwaAAxtcn8YkZXnvk=
 github.com/klauspost/cpuid/v2 v2.2.4/go.mod h1:RVVoqg1df56z8g3pUjL/3lE5UfnlrJX8tyFgg4nqhuY=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/leodido/go-urn v1.2.4 h1:XlAE/cm/ms7TE/VMVoduSpNBoyc2dOxHs5MZSwAN63Q=
 github.com/leodido/go-urn v1.2.4/go.mod h1:7ZrI8mTSeBSHl/UaRyKQW1qZeMgak41ANeCNaVckg+4=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
 github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-sqlite3 v1.14.16 h1:yOQRA0RpS5PFz/oikGwBEqvAWhWg5ufRz4ETLjwpU1Y=
@@ -69,6 +76,8 @@ github.com/prometheus/common v0.42.0 h1:EKsfXEYo4JpWMHH5cg+KOUWeuJSov1Id8zGR8eeI
 github.com/prometheus/common v0.42.0/go.mod h1:xBwqVerjNdUDjgODMpudtOMwlOwf2SaTr1yjz4b7Zbc=
 github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+PymziUAg=
 github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
+github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
+github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/spf13/cobra v1.7.0 h1:hyqWnYt1ZQShIddO5kBpj3vu05/++x6tJ6dg8EC572I=
 github.com/spf13/cobra v1.7.0/go.mod h1:uLxZILRyS/50WlhOIKD7W6V5bgeIt+4sICxh6uRMrb0=
@@ -112,8 +121,9 @@ google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp0
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
 google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/app/coroutines/cancelPromise.go

@@ -15,6 +15,9 @@ func CancelPromise(t int64, req *types.Request, res func(*types.Response, error)
 		if req.CancelPromise.Value.Headers == nil {
 			req.CancelPromise.Value.Headers = map[string]string{}
 		}
+		if req.CancelPromise.Value.Data == nil {
+			req.CancelPromise.Value.Data = []byte{}
+		}
 
 		submission := &types.Submission{
 			Kind: types.Store,
@@ -79,7 +82,7 @@ func CancelPromise(t int64, req *types.Request, res func(*types.Response, error)
 										Value: promise.Value{
 											Headers: map[string]string{},
 											Ikey:    nil,
-											Data:    nil,
+											Data:    []byte{},
 										},
 										Timeout:     p.Timeout,
 										Tags:        p.Tags,
@@ -132,12 +135,19 @@ func CancelPromise(t int64, req *types.Request, res func(*types.Response, error)
 										Id: req.CancelPromise.Id,
 									},
 								},
+								{
+									Kind: types.StoreDeleteSubscriptions,
+									DeleteSubscriptions: &types.DeleteSubscriptionsCommand{
+										PromiseId: req.CancelPromise.Id,
+									},
+								},
 							}
 
 							for _, record := range records {
 								commands = append(commands, &types.Command{
 									Kind: types.StoreCreateNotification,
 									CreateNotification: &types.CreateNotificationCommand{
+										Id:          record.Id,
 										PromiseId:   record.PromiseId,
 										Url:         record.Url,
 										RetryPolicy: record.RetryPolicy,

internal/app/coroutines/createPromise.go

@@ -8,14 +8,16 @@ import (
 	"github.com/resonatehq/resonate/internal/kernel/types"
 	"github.com/resonatehq/resonate/internal/util"
 	"github.com/resonatehq/resonate/pkg/promise"
-	"github.com/resonatehq/resonate/pkg/subscription"
 )
 
 func CreatePromise(t int64, req *types.Request, res func(*types.Response, error)) *scheduler.Coroutine {
 	return scheduler.NewCoroutine(fmt.Sprintf("CreatePromise(id=%s)", req.CreatePromise.Id), "CreatePromise", func(s *scheduler.Scheduler, c *scheduler.Coroutine) {
 		if req.CreatePromise.Param.Headers == nil {
 			req.CreatePromise.Param.Headers = map[string]string{}
 		}
+		if req.CreatePromise.Param.Data == nil {
+			req.CreatePromise.Param.Data = []byte{}
+		}
 		if req.CreatePromise.Tags == nil {
 			req.CreatePromise.Tags = map[string]string{}
 		}
@@ -49,51 +51,29 @@ func CreatePromise(t int64, req *types.Request, res func(*types.Response, error)
 			util.Assert(result.RowsReturned == 0 || result.RowsReturned == 1, "result must return 0 or 1 rows")
 
 			if result.RowsReturned == 0 {
-				commands := []*types.Command{
-					{
-						Kind: types.StoreCreatePromise,
-						CreatePromise: &types.CreatePromiseCommand{
-							Id:        req.CreatePromise.Id,
-							Param:     req.CreatePromise.Param,
-							Timeout:   req.CreatePromise.Timeout,
-							Tags:      req.CreatePromise.Tags,
-							CreatedOn: t,
-						},
-					},
-					{
-						Kind: types.StoreCreateTimeout,
-						CreateTimeout: &types.CreateTimeoutCommand{
-							Id:   req.CreatePromise.Id,
-							Time: req.CreatePromise.Timeout,
-						},
-					},
-				}
-
-				for _, s := range req.CreatePromise.Subscriptions {
-					// default retry policy
-					if s.RetryPolicy == nil {
-						s.RetryPolicy = &subscription.RetryPolicy{
-							Delay:    30,
-							Attempts: 3,
-						}
-					}
-
-					commands = append(commands, &types.Command{
-						Kind: types.StoreCreateSubscription,
-						CreateSubscription: &types.CreateSubscriptionCommand{
-							PromiseId:   req.CreatePromise.Id,
-							Url:         s.Url,
-							RetryPolicy: s.RetryPolicy,
-							CreatedOn:   t,
-						},
-					})
-				}
-
 				submission := &types.Submission{
 					Kind: types.Store,
 					Store: &types.StoreSubmission{
 						Transaction: &types.Transaction{
-							Commands: commands,
+							Commands: []*types.Command{
+								{
+									Kind: types.StoreCreatePromise,
+									CreatePromise: &types.CreatePromiseCommand{
+										Id:        req.CreatePromise.Id,
+										Param:     req.CreatePromise.Param,
+										Timeout:   req.CreatePromise.Timeout,
+										Tags:      req.CreatePromise.Tags,
+										CreatedOn: t,
+									},
+								},
+								{
+									Kind: types.StoreCreateTimeout,
+									CreateTimeout: &types.CreateTimeoutCommand{
+										Id:   req.CreatePromise.Id,
+										Time: req.CreatePromise.Timeout,
+									},
+								},
+							},
 						},
 					},
 				}
@@ -163,7 +143,7 @@ func CreatePromise(t int64, req *types.Request, res func(*types.Response, error)
 									Value: promise.Value{
 										Headers: map[string]string{},
 										Ikey:    nil,
-										Data:    nil,
+										Data:    []byte{},
 									},
 									Timeout:     p.Timeout,
 									Tags:        p.Tags,