chore: various improvements across codebase (#7)

Signed-off-by: Imtiaz Uddin <[email protected]>

Author: Imtiaz246

.github/workflows/release.yml

@@ -64,8 +64,8 @@ jobs:
         if: startsWith(github.ref, 'refs/tags/')
         with:
           files: |
-            bin/wal-listener-linux-amd64.tar.gz
-            bin/wal-listener-linux-arm64.tar.gz
-            bin/wal-listener-checksums.txt
+            bin/pgoutbox-linux-amd64.tar.gz
+            bin/pgoutbox-linux-arm64.tar.gz
+            bin/pgoutbox-checksums.txt
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.lgtm

@@ -0,0 +1,3 @@
+pattern = "(?)LGTM"
+self_approval_off = true
+ignore_maintainers_file = true

DCO

@@ -0,0 +1,36 @@
+Developer Certificate of Origin
+Version 1.1
+
+Copyright (C) 2004, 2006 The Linux Foundation and its contributors.
+660 York Street, Suite 102,
+San Francisco, CA 94110 USA
+
+Everyone is permitted to copy and distribute verbatim copies of this
+license document, but changing it is not allowed.
+
+
+Developer's Certificate of Origin 1.1
+
+By making a contribution to this project, I certify that:
+
+(a) The contribution was created in whole or in part by me and I
+    have the right to submit it under the open source license
+    indicated in the file; or
+
+(b) The contribution is based upon previous work that, to the best
+    of my knowledge, is covered under an appropriate open source
+    license and I have the right under that license to submit that
+    work with modifications, whether created in whole or in part
+    by me, under the same open source license (unless I am
+    permitted to submit under a different license), as indicated
+    in the file; or
+
+(c) The contribution was provided directly to me by some other
+    person who certified (a), (b) or (c) and I have not modified
+    it.
+
+(d) I understand and agree that this project and the contribution
+    are public and that a record of the contribution (including all
+    personal information I submit with it, including my sign-off) is
+    maintained indefinitely and may be redistributed consistent with
+    this project or the open source license(s) involved.

Dockerfile.dbg

@@ -16,7 +16,7 @@ FROM ghcr.io/appscode/dlv:1.24
 
 FROM {ARG_FROM}
 
-LABEL org.opencontainers.image.source https://github.com/bytebuilders/wal-listener
+LABEL org.opencontainers.image.source https://github.com/kubeops/pgoutbox
 
 RUN set -x \
   && apt-get update \

Dockerfile.in

@@ -14,7 +14,7 @@
 
 FROM {ARG_FROM}
 
-LABEL org.opencontainers.image.source https://github.com/bytebuilders/wal-listener
+LABEL org.opencontainers.image.source https://github.com/kubeops/pgoutbox
 
 RUN set -x \
   && apk add --update --upgrade --no-cache pcre2 ca-certificates tzdata openssh-client \

Makefile

@@ -17,12 +17,12 @@
 SHELL=/bin/bash -o pipefail
 
 PRODUCT_OWNER_NAME := appscode
-PRODUCT_NAME       := wal-listener
+PRODUCT_NAME       := pgoutbox
 ENFORCE_LICENSE    ?=
 
-GO_PKG   := github.com/ihippik
+GO_PKG   := kubeops.dev
 REPO     := $(notdir $(shell pwd))
-BIN      := wal-listener
+BIN      := pgoutbox
 COMPRESS ?= no
 
 # Where to push the docker image.
@@ -313,7 +313,7 @@ install:
 	@cd ../installer; \
 	kubectl create ns $(KUBE_NAMESPACE) || true; \
 	kubectl label ns $(KUBE_NAMESPACE) pod-security.kubernetes.io/enforce=restricted; \
-	helm upgrade -i cluster-importer charts/cluster-importer --wait \
+	helm upgrade -i pgoutbox charts/pgoutbox --wait \
 		--namespace=$(KUBE_NAMESPACE) --create-namespace \
 		--set image.tag=$(TAG_PROD) \
 		--set imagePullPolicy=$(IMAGE_PULL_POLICY) \
@@ -322,7 +322,7 @@ install:
 .PHONY: uninstall
 uninstall:
 	@cd ../installer; \
-	helm uninstall cluster-importer --namespace=$(KUBE_NAMESPACE) || true
+	helm uninstall pgoutbox --namespace=$(KUBE_NAMESPACE) || true
 
 .PHONY: purge
 purge: uninstall

README.md

@@ -1,11 +1,6 @@
-# WAL-Listener
+# pgoutbox
 
-[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
-![GitHub go.mod Go version](https://img.shields.io/github/go-mod/go-version/ihippik/wal-listener)
-![GitHub tag (latest SemVer)](https://img.shields.io/github/v/tag/ihippik/wal-listener)
-[![Publish Docker image](https://github.com/ihippik/wal-listener/actions/workflows/github-actions.yml/badge.svg)](https://github.com/ihippik/wal-listener/actions/workflows/github-actions.yml)
-
-![WAL-Listener](wal-listener.png)
+![PgOutbox](wal-listener.png)
 
 A service that helps implement the **Event-Driven architecture**.
 
@@ -77,7 +72,7 @@ You must make the following settings in the db configuration (postgresql.conf)
 * max_replication_slots >= 1
 
 The publication & slot created automatically when the service starts (for all tables and all actions).
-You can delete the default publication and create your own (name: _wal-listener_) with the necessary filtering conditions, and then the filtering will occur at the database level and not at the application level.
+You can delete the default publication and create your own (name: _pgoutbox_) with the necessary filtering conditions, and then the filtering will occur at the database level and not at the application level.
 
 https://www.postgresql.org/docs/current/sql-createpublication.html
 
@@ -150,7 +145,7 @@ for readiness `/ready`  and liveness `/healthz` probes.
 You can start the container from the project folder (configuration file is required).
 
 See `./config_example.yml` for an example configuration.
-Be sure to copy the file to the docker image in the `Dockerfile` prior to running [after the build setp](https://github.com/ihippik/wal-listener/blob/master/Dockerfile#L31)
+Be sure to copy the file to the docker image in the `Dockerfile` prior to running [after the build setp](https://kubeops.dev/pgoutbox/blob/master/Dockerfile#L31)
 ex:
 ```docker
 COPY /config.yml .
@@ -161,10 +156,3 @@ Please don't forget to delete them:
 ```shell
 docker image prune --filter label=stage=builder
 ```
-
-#### Docker Hub
-https://hub.docker.com/r/ihippik/wal-listener
-#### Example
-```shell
-docker run -v $(pwd)/config.yml:/app/config.yml ihippik/wal-listener:tag
-```

apis/config.go

@@ -0,0 +1,97 @@
+package apis
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/asaskevich/govalidator"
+	"github.com/spf13/viper"
+)
+
+type PublisherType string
+
+const (
+	PublisherTypeNats         PublisherType = "nats"
+	PublisherTypeKafka        PublisherType = "kafka"
+	PublisherTypeRabbitMQ     PublisherType = "rabbitmq"
+	PublisherTypeGooglePubSub PublisherType = "google_pubsub"
+)
+
+// Config for pgoutbox.
+type Config struct {
+	Listener  *ListenerCfg  `valid:"required" json:"listener" mapstructure:"listener"`
+	Database  *DatabaseCfg  `valid:"required" json:"database" mapstructure:"database"`
+	Publisher *PublisherCfg `valid:"required" json:"publisher" mapstructure:"publisher"`
+	Logger    *Logger       `valid:"required" json:"logger" mapstructure:"logger"`
+}
+
+// ListenerCfg path of the listener config.
+type ListenerCfg struct {
+	SlotName          string            `valid:"required" json:"slotName" mapstructure:"slotName"`
+	ServerPort        int               `json:"serverPort" mapstructure:"serverPort"`
+	AckTimeout        time.Duration     `json:"ackTimeout" mapstructure:"ackTimeout"`
+	RefreshConnection time.Duration     `json:"refreshConnection" valid:"required" mapstructure:"refreshConnection"`
+	HeartbeatInterval time.Duration     `json:"heartbeatInterval" valid:"required" mapstructure:"heartbeatInterval"`
+	Filter            FilterStruct      `json:"filter" mapstructure:"filter"`
+	TopicsMap         map[string]string `json:"topicsMap" mapstructure:"topicsMap"`
+}
+
+// PublisherCfg represent configuration for any publisher types.
+type PublisherCfg struct {
+	Type            PublisherType `valid:"required" json:"type" mapstructure:"type"`
+	Address         string        `valid:"required" json:"address" mapstructure:"address"`
+	NatsCredPath    string        `valid:"required" json:"natsCredPath" mapstructure:"natsCredPath"`
+	Topic           string        `valid:"required" json:"topic" mapstructure:"topic"`
+	TopicPrefix     string        `json:"topicPrefix" mapstructure:"topicPrefix"`
+	EnableTLS       bool          `json:"enableTLS" mapstructure:"enableTlS"`
+	ClientCert      string        `json:"clientCert" mapstructure:"clientCert"`
+	ClientKey       string        `json:"clientKey" mapstructure:"clientKey"`
+	CACert          string        `json:"CACert" mapstructure:"caCert"`
+	PubSubProjectID string        `json:"pubSubProjectID" mapstructure:"pubSubProductId"`
+}
+
+// DatabaseCfg path of the PostgreSQL DB config.
+type DatabaseCfg struct {
+	Host     string `valid:"required" json:"host" mapstructure:"host"`
+	Port     uint16 `valid:"required" json:"port" mapstructure:"port"`
+	Name     string `valid:"required" json:"name" mapstructure:"name"`
+	User     string `valid:"required" json:"user" mapstructure:"user"`
+	Password string `valid:"required" json:"password" mapstructure:"password"`
+	Debug    bool   `json:"debug" mapstructure:"debug"`
+}
+
+// FilterStruct incoming WAL message filter.
+type FilterStruct struct {
+	Tables map[string][]string `json:"tables" yaml:"tables" mapstructure:"tables"`
+}
+
+// Validate config data.
+func (c Config) Validate() error {
+	_, err := govalidator.ValidateStruct(c)
+	return err
+}
+
+// InitConfig load config from file.
+func InitConfig(path string) (*Config, error) {
+	const envPrefix = "WAL"
+
+	var conf Config
+
+	vp := viper.New()
+
+	vp.SetEnvPrefix(envPrefix)
+	vp.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
+	vp.AutomaticEnv()
+	vp.SetConfigFile(path)
+
+	if err := vp.ReadInConfig(); err != nil {
+		return nil, fmt.Errorf("error reading config: %w", err)
+	}
+
+	if err := vp.Unmarshal(&conf); err != nil {
+		return nil, fmt.Errorf("unable to decode into config struct: %w", err)
+	}
+
+	return &conf, nil
+}

internal/config/config_test.go renamed to apis/config_test.go

@@ -1,17 +1,15 @@
-package config
+package apis
 
 import (
 	"errors"
 	"testing"
 
-	scfg "github.com/ihippik/config"
-
 	"github.com/stretchr/testify/assert"
 )
 
 func TestConfig_Validate(t *testing.T) {
 	type fields struct {
-		Logger    *scfg.Logger
+		Logger    *Logger
 		Listener  *ListenerCfg
 		Database  *DatabaseCfg
 		Publisher *PublisherCfg
@@ -24,7 +22,7 @@ func TestConfig_Validate(t *testing.T) {
 		{
 			name: "success",
 			fields: fields{
-				Logger: &scfg.Logger{
+				Logger: &Logger{
 					Level: "info",
 				},
 				Listener: &ListenerCfg{
@@ -52,7 +50,7 @@ func TestConfig_Validate(t *testing.T) {
 		{
 			name: "bad listener cfg",
 			fields: fields{
-				Logger: &scfg.Logger{
+				Logger: &Logger{
 					Level: "info",
 				},
 				Listener: &ListenerCfg{
@@ -78,7 +76,7 @@ func TestConfig_Validate(t *testing.T) {
 		{
 			name: "bad db cfg",
 			fields: fields{
-				Logger: &scfg.Logger{
+				Logger: &Logger{
 					Level: "info",
 				},
 				Listener: &ListenerCfg{
@@ -104,7 +102,7 @@ func TestConfig_Validate(t *testing.T) {
 		{
 			name: "empty publisher kind",
 			fields: fields{
-				Logger: &scfg.Logger{
+				Logger: &Logger{
 					Level: "info",
 				},
 				Listener: &ListenerCfg{

internal/publisher/event.go renamed to apis/event.go

@@ -1,12 +1,9 @@
-package publisher
+package apis
 
 import (
 	"fmt"
-	"time"
-
-	"github.com/ihippik/wal-listener/v2/internal/config"
-
 	"github.com/google/uuid"
+	"time"
 )
 
 // Event structure for publishing to the NATS server.
@@ -21,8 +18,8 @@ type Event struct {
 }
 
 // SubjectName creates subject name from the prefix, schema and table name. Also using topic map from cfg.
-func (e *Event) SubjectName(cfg *config.Config) string {
-	topic := fmt.Sprintf("%s_%s", e.Schema, e.Table)
+func (e *Event) SubjectName(cfg *Config) string {
+	topic := fmt.Sprintf("schemas.%s.tables.%s", e.Schema, e.Table)
 
 	if cfg.Listener.TopicsMap != nil {
 		if t, ok := cfg.Listener.TopicsMap[topic]; ok {