Merge pull request #21 from gitopia/handle-handler-error

Do not exit the server in case of event handler error

Author: faza

app/consumer/persistence.go

@@ -0,0 +1,59 @@
+package consumer
+
+import (
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/pkg/errors"
+)
+
+const (
+	OFFSET_FILE = ".o"
+	FILE_PERM   = 0666
+)
+
+type Client struct {
+	f *os.File
+}
+
+func NewClient(topic string) (Client, error) {
+	file, err := os.OpenFile("_"+topic+OFFSET_FILE, os.O_RDWR|os.O_CREATE, FILE_PERM)
+	if err != nil {
+		return Client{}, errors.Wrap(err, "error opening offset file")
+	}
+	return Client{
+		f: file,
+	}, nil
+}
+
+// write intensive use case. save all the offsets
+func (c Client) Commit(offset uint64) error {
+	err := c.f.Truncate(0)
+	if err != nil {
+		return errors.Wrap(err, "error truncating offset file")
+	}
+	_, err = c.f.Seek(0, 0)
+	if err != nil {
+		return errors.Wrap(err, "error overwriting offset file")
+	}
+	_, err = fmt.Fprintf(c.f, "%d", offset)
+	if err != nil {
+		return errors.Wrap(err, "error formatting offset file")
+	}
+	// ensure all data s written to disk. no data loss at the cost of latency and CPU resources
+	err = c.f.Sync()
+	if err != nil {
+		return errors.Wrap(err, "error syncing offset file")
+	}
+	return nil
+}
+
+func (c Client) Offset() (uint64, error) {
+	n := uint64(0)
+	_, err := fmt.Fscanf(c.f, "%d", &n)
+	if err != nil && err != io.EOF {
+		return 0, errors.Wrap(err, "error reading offset file")
+	}
+	return n, nil
+}

app/gitopia.go

@@ -6,7 +6,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/spf13/viper"
 
-	"github.com/gitopia/gitopia-ipfs-bridge/logger"
+	"github.com/gitopia/git-server/logger"
 	"github.com/gitopia/gitopia/x/gitopia/types"
 	"github.com/tendermint/starport/starport/pkg/cosmosaccount"
 	"github.com/tendermint/starport/starport/pkg/cosmosclient"

app/tm/tm.go

@@ -0,0 +1,73 @@
+package tm
+
+import (
+	"context"
+
+	"github.com/gitopia/git-server/logger"
+	"github.com/pkg/errors"
+	"github.com/spf13/viper"
+	"github.com/tendermint/tendermint/rpc/jsonrpc/client"
+)
+
+const (
+	TM_WS_ENDPOINT = "/websocket"
+)
+
+type Client struct {
+	c *client.WSClient
+}
+
+type evenHandlerFunc func(context.Context, []byte) error
+
+func NewTmClient() (*Client, error) {
+	wsc, err := client.NewWS(viper.GetString("tm_addr"), TM_WS_ENDPOINT)
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating ws client")
+	}
+	err = wsc.Start()
+	if err != nil {
+		return nil, errors.Wrap(err, "error connecting to WS")
+	}
+	return &Client{
+		c: wsc,
+	}, nil
+}
+
+// processes events from tm
+// returns error on failure
+// returns error when event handler returns error
+func (c Client) Subscribe(ctx context.Context, q string, h evenHandlerFunc) (<-chan struct{}, chan error) {
+	e := make(chan error)
+	ctx, cancel := context.WithCancel(ctx)
+	go func() {
+		defer cancel()
+		err := c.c.Subscribe(ctx, q)
+		if err != nil {
+			e <- errors.Wrap(err, "error sending subscribe request")
+			return
+		}
+		for {
+			event := <-c.c.ResponsesCh
+			if event.Error != nil {
+				e <- errors.Wrap(err, "error reading from ws")
+				return
+			}
+
+			jsonBuf, err := event.Result.MarshalJSON()
+			if err != nil {
+				e <- errors.Wrap(err, "error parsing result")
+				return
+			}
+			// hack: TM sends empty event to begin with. skipping
+			if string(jsonBuf) == "{}" {
+				logger.FromContext(ctx).Info("received empty event. continuing...")
+				continue
+			}
+			err = h(ctx, jsonBuf)
+			if err != nil {
+				logger.FromContext(ctx).Error(errors.WithMessage(err, "error from event handler"))
+			}
+		}
+	}()
+	return ctx.Done(), e
+}

cmd/git-server-events/main.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/gitopia/git-server/app"
-	"github.com/gitopia/gitopia-ipfs-bridge/logger"
+	"github.com/gitopia/git-server/logger"
 	"github.com/spf13/viper"
 )
 

cmd/git-server-events/run.go

@@ -8,10 +8,10 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/gitopia/git-server/app"
+	"github.com/gitopia/git-server/app/consumer"
+	"github.com/gitopia/git-server/app/tm"
 	"github.com/gitopia/git-server/handler"
-	"github.com/gitopia/gitopia-ipfs-bridge/app/consumer"
-	"github.com/gitopia/gitopia-ipfs-bridge/app/tm"
-	"github.com/gitopia/gitopia-ipfs-bridge/logger"
+	"github.com/gitopia/git-server/logger"
 )
 
 func NewRunCmd() *cobra.Command {

config_dev.toml

@@ -1,3 +1,4 @@
+web_server_port = 5000
 gitopia_grpc_url = "grpc.gitopia.dev:9090"
 git_dir = "/var/repos"
 attachment_dir = "/var/attachments"

config_local.toml

@@ -1,3 +1,4 @@
+web_server_port = 5001
 gitopia_grpc_url = "localhost:9090"
 git_dir = "/var/repos"
 attachment_dir = "/var/attachments"

config_prod.toml

@@ -1,3 +1,4 @@
+web_server_port = 5000
 gitopia_grpc_url = "grpc.gitopia.com:9090"
 git_dir = "/var/repos"
 attachment_dir = "/var/attachments"

go.mod

@@ -6,16 +6,24 @@ require (
 	github.com/buger/jsonparser v1.1.1
 	github.com/cosmos/cosmos-sdk v0.45.1
 	github.com/gitopia/gitopia v0.13.0
-	github.com/gitopia/gitopia-ipfs-bridge v0.0.0-20220324075926-93f790f71054
 	github.com/gitopia/go-git/v5 v5.4.3-0.20211224112515-b2efd9bec92c
 	github.com/go-git/go-billy/v5 v5.3.1
+	github.com/kr/pretty v0.3.0 // indirect
 	github.com/libgit2/git2go/v33 v33.0.4
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.12.0 // indirect
 	github.com/rs/cors v1.8.2
+	github.com/rs/zerolog v1.26.1 // indirect
+	github.com/sirupsen/logrus v1.8.1
+	github.com/spf13/afero v1.8.0 // indirect
 	github.com/spf13/cobra v1.4.0
 	github.com/spf13/viper v1.10.1
 	github.com/tendermint/starport v0.19.3
+	github.com/tendermint/tendermint v0.34.15
+	golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce // indirect
+	golang.org/x/mod v0.5.1 // indirect
 	google.golang.org/grpc v1.45.0
+	gopkg.in/ini.v1 v1.66.3 // indirect
 )
 
 replace github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1