perf: add mcache

app/app.go

@@ -1,12 +1,16 @@
 package app
 
 import (
+	"context"
+	"encoding/json"
 	"errors"
 	"github.com/webhookx-io/webhookx/admin"
 	"github.com/webhookx-io/webhookx/admin/api"
 	"github.com/webhookx-io/webhookx/config"
 	"github.com/webhookx-io/webhookx/db"
 	"github.com/webhookx-io/webhookx/dispatcher"
+	"github.com/webhookx-io/webhookx/eventbus"
+	"github.com/webhookx-io/webhookx/mcache"
 	"github.com/webhookx-io/webhookx/pkg/cache"
 	"github.com/webhookx-io/webhookx/pkg/log"
 	"github.com/webhookx-io/webhookx/pkg/taskqueue"
@@ -15,6 +19,7 @@
 	"github.com/webhookx-io/webhookx/worker/deliverer"
 	"go.uber.org/zap"
 	"sync"
+	"time"
 )
 
 var (
@@ -35,6 +40,7 @@
 	queue      taskqueue.TaskQueue
 	dispatcher *dispatcher.Dispatcher
 	cache      cache.Cache
+	bus        eventbus.Bus
 
 	admin   *admin.Admin
 	gateway *proxy.Gateway
@@ -65,24 +71,43 @@
 	zap.ReplaceGlobals(log)
 	app.log = zap.S()
 
+	// cache
+	client := cfg.RedisConfig.GetClient()
+	app.cache = cache.NewRedisCache(client)
+
+	mcache.Set(mcache.NewMCache(&mcache.Options{
+		L1Size: 1000,
+		L1TTL:  time.Second * 10,
+		L2:     app.cache,
+	}))
+
+	app.bus = eventbus.NewDatabaseEventBus(cfg.DatabaseConfig.GetDSN(), app.log)
+	app.bus.Subscribe(eventbus.EventInvalidation, func(data []byte) {
+		maps := make(map[string]interface{})
+		if err := json.Unmarshal(data, &maps); err != nil {
+			return
+		}
+		if cacheKey, ok := maps["cache_key"]; ok {
+			err := mcache.Invalidate(context.TODO(), cacheKey.(string))
+			if err != nil {
+				app.log.Errorf("failed to invalidate cache: key=%s %v", cacheKey, err)
+			}
+		}
+	})
+
 	// db
 	db, err := db.NewDB(&cfg.DatabaseConfig)
 	if err != nil {
 		return err
 	}
 	app.db = db
 
-	client := cfg.RedisConfig.GetClient()
-
 	// queue
 	queue := taskqueue.NewRedisQueue(taskqueue.RedisTaskQueueOptions{
 		Client: client,
 	}, app.log)
 	app.queue = queue
 
-	// cache
-	app.cache = cache.NewRedisCache(client)
-
 	app.dispatcher = dispatcher.NewDispatcher(log.Sugar(), queue, db)
 
 	// worker
@@ -122,6 +147,9 @@
 		return ErrApplicationStarted
 	}
 
+	if err := app.bus.Start(); err != nil {
+		return err
+	}
 	if app.admin != nil {
 		app.admin.Start()
 	}
@@ -156,6 +184,7 @@
 		app.log.Infof("stopped")
 	}()
 
+	_ = app.bus.Stop()
 	// TODO: timeout
 	if app.admin != nil {
 		app.admin.Stop()

config/config.go

@@ -3,6 +3,7 @@ package config
 import (
 	"encoding/json"
 	"github.com/creasty/defaults"
+	uuid "github.com/satori/go.uuid"
 	"github.com/webhookx-io/webhookx/pkg/envconfig"
 	"gopkg.in/yaml.v3"
 	"os"
@@ -11,6 +12,7 @@ import (
 var (
 	VERSION = "dev"
 	COMMIT  = "unknown"
+	NODE    = uuid.NewV4().String()
 )
 
 var cfg Config

config/database.go

@@ -1,27 +1,30 @@
 package config
 
 import (
-	"database/sql"
 	"fmt"
 )
 
 type DatabaseConfig struct {
+	DSN      string
 	Host     string `yaml:"host" default:"localhost"`
 	Port     uint32 `yaml:"port" default:"5432"`
 	Username string `yaml:"username" default:"webhookx"`
 	Password string `yaml:"password" default:""`
 	Database string `yaml:"database" default:"webhookx"`
 }
 
-func (cfg DatabaseConfig) GetDB() (*sql.DB, error) {
-	dsn := fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=disable",
-		cfg.Username,
-		cfg.Password,
-		cfg.Host,
-		cfg.Port,
-		cfg.Database,
-	)
-	return sql.Open("postgres", dsn)
+func (cfg DatabaseConfig) GetDSN() string {
+	dsn := cfg.DSN
+	if dsn == "" {
+		dsn = fmt.Sprintf("postgres://%s:%s@%s:%d/%s?sslmode=disable",
+			cfg.Username,
+			cfg.Password,
+			cfg.Host,
+			cfg.Port,
+			cfg.Database,
+		)
+	}
+	return dsn
 }
 
 func (cfg DatabaseConfig) Validate() error {

constants/constants.go

@@ -2,6 +2,7 @@ package constants
 
 import (
 	"github.com/webhookx-io/webhookx/config"
+	"strings"
 	"time"
 )
 
@@ -26,6 +27,29 @@ const (
 	RequeueInterval = time.Second * 60
 )
 
+type CacheKey string
+
+func (c CacheKey) Build(id string) string {
+	var sb strings.Builder
+	sb.WriteString(Namespace)
+	sb.WriteString(":")
+	sb.WriteString(string(c))
+	sb.WriteString(":")
+	sb.WriteString(id)
+	return sb.String()
+}
+
+const (
+	Namespace             string   = "webhookx"
+	EventCacheKey         CacheKey = "events"
+	EndpointCacheKey      CacheKey = "endpoints"
+	SourceCacheKey        CacheKey = "sources"
+	WorkspaceCacheKey     CacheKey = "workspaces"
+	AttemptCacheKey       CacheKey = "attempts"
+	PluginCacheKey        CacheKey = "plugins"
+	AttemptDetailCacheKey CacheKey = "attempt_details"
+)
+
 var (
 	DefaultResponseHeaders = map[string]string{
 		"Content-Type": "application/json",

db/dao/attempt_dao.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	sq "github.com/Masterminds/squirrel"
 	"github.com/jmoiron/sqlx"
+	"github.com/webhookx-io/webhookx/constants"
 	"github.com/webhookx-io/webhookx/db/entities"
 	"github.com/webhookx-io/webhookx/pkg/types"
 )
@@ -22,8 +23,15 @@ type AttemptResult struct {
 }
 
 func NewAttemptDao(db *sqlx.DB, workspace bool) AttemptDAO {
+	opts := Options{
+		Table:          "attempts",
+		EntityName:     "Attempt",
+		Workspace:      workspace,
+		CachePropagate: false,
+		CacheKey:       constants.AttemptCacheKey,
+	}
 	return &attemptDao{
-		DAO: NewDAO[entities.Attempt]("attempts", db, workspace),
+		DAO: NewDAO[entities.Attempt](db, opts),
 	}
 }
 

db/dao/attempt_detail_dao.go

@@ -2,6 +2,7 @@ package dao
 
 import (
 	"context"
+	"github.com/webhookx-io/webhookx/constants"
 	"time"
 
 	"github.com/jmoiron/sqlx"
@@ -13,8 +14,15 @@ type attemptDetailDao struct {
 }
 
 func NewAttemptDetailDao(db *sqlx.DB, workspace bool) AttemptDetailDAO {
+	opts := Options{
+		Table:          "attempt_details",
+		EntityName:     "AttemptDetail",
+		Workspace:      workspace,
+		CachePropagate: false,
+		CacheKey:       constants.AttemptDetailCacheKey,
+	}
 	return &attemptDetailDao{
-		DAO: NewDAO[entities.AttemptDetail]("attempt_details", db, workspace),
+		DAO: NewDAO[entities.AttemptDetail](db, opts),
 	}
 }