This is a microservices project based on go-micro, It is suitable for lightweight mobile apps or small game backends.
- API gateway / HTTP handler: Gin
- Microservices framework: go-micro
- Database: pgx (PostgreSQL driver)
- Cache: go-redis (Redis client)
- Cache: Redis
- Database: PostgreSQL
- Service discovery & configuration: etcd
The framework now includes a minimal reminder worker built on PostgreSQL and the Mesa lifecycle.
- Store reminder records in PostgreSQL
- Poll due reminders in the background
- Deliver reminders through a user-supplied notifier
- Retry failed deliveries with backoff
- Recover stale processing reminders after restart
- Cancel or reschedule reminders by key
mesa, err := wego.New(
wego.WithDSN(dsn),
)
if err != nil {
panic(err)
}
svc := reminder.NewService(
mesa.DB,
reminder.NotifierFunc(func(ctx context.Context, r *reminder.Reminder) error {
// send email, push, webhook, etc.
return nil
}),
)
// Run versioned migrations before starting Mesa (normally in deployment tooling).
if err := reminder.Migrate(context.Background(), mesa.DB); err != nil {
panic(err)
}
if err := mesa.RegisterComponent(svc); err != nil {
panic(err)
}
reminderSvc, ok := wego.GetComponentAs[*reminder.Service](mesa, svc.Name())
if !ok {
panic("reminder component not found")
}
_, err := reminderSvc.Create(context.Background(), reminder.CreateParams{
Key: "order-123-pay-deadline",
UserID: "42",
Channel: "in_app",
Payload: `{"title":"Payment due soon"}`,
ScheduleAt: time.Now().Add(10 * time.Minute),
})Reminder schema changes are explicit: call reminder.Migrate(ctx, db) before the worker starts. Service.Start never creates or alters tables.
item, err := reminderSvc.GetByKey(ctx, "order-123-pay-deadline")
if errors.Is(err, reminder.ErrNotFound) {
// handle a missing reminder
}
page, err := reminderSvc.List(ctx, reminder.ListParams{
UserID: "42",
Statuses: []reminder.Status{reminder.StatusPending},
Limit: 50,
})
nextPage, err := reminderSvc.List(ctx, reminder.ListParams{
UserID: "42",
Limit: 50,
Cursor: page.NextCursor,
})
_ = item
_ = nextPageCreate returns ErrAlreadyExists for duplicate keys. Cancel and reschedule return ErrNotFound for unknown keys and ErrStateConflict when the current state does not allow the operation. Canceling an already canceled reminder succeeds.
When reminder scenes grow, you can register handlers by channel, by payload type, or by an exact channel + type route.
dispatcher := reminder.NewDispatcherNotifier()
_ = dispatcher.RegisterChannel("in_app", reminder.NotifierFunc(func(ctx context.Context, r *reminder.Reminder) error {
// generic in-app delivery
return nil
}))
_ = dispatcher.RegisterType("payment_due", reminder.NotifierFunc(func(ctx context.Context, r *reminder.Reminder) error {
// shared payment_due template logic
return nil
}))
_ = dispatcher.Register("sms", "payment_due", reminder.NotifierFunc(func(ctx context.Context, r *reminder.Reminder) error {
// exact route for sms + payment_due
return nil
}))
_ = dispatcher.SetFallback(reminder.NotifierFunc(func(ctx context.Context, r *reminder.Reminder) error {
// last-resort handler
return nil
}))
mesa, err := wego.New(
wego.WithDSN(dsn),
)
if err != nil {
panic(err)
}
svc := reminder.NewService(mesa.DB, dispatcher)
if err := mesa.RegisterComponent(svc); err != nil {
panic(err)
}
component, ok := mesa.GetComponent(svc.Name())
if !ok {
panic("component not found")
}
reminderSvc := component.(*reminder.Service)
_, _ = reminderSvc.Create(ctx, reminder.CreateParams{
Key: "order-123-pay-deadline",
UserID: "42",
Channel: "sms",
Payload: `{"type":"payment_due","title":"Payment due soon"}`,
ScheduleAt: time.Now().Add(10 * time.Minute),
})Dispatch priority is:
- exact
channel + type - payload
type channel- fallback
Mesa now acts as a generic component registry.
if err := mesa.RegisterComponent(svc); err != nil {
panic(err)
}
component, ok := mesa.GetComponent("reminder")
if !ok {
panic("component not found")
}
reminderSvc, ok := wego.GetComponentAs[*reminder.Service](mesa, "reminder")
if !ok {
panic("unexpected component type")
}
_ = component
_ = reminderSvcWithComponents(...) lets Mesa auto-register built-in components in one place.
mesa, err := wego.New(
wego.WithDSN(dsn),
wego.WithRedisConfig(wego.RedisConfig{
Addr: "127.0.0.1:6379",
}),
wego.WithComponents(wego.ComponentsConfig{
Reminder: wego.ReminderComponentConfig{
Enabled: true,
Notifier: reminder.NotifierFunc(func(ctx context.Context, r *reminder.Reminder) error {
return nil
}),
PollInterval: 2 * time.Second,
},
PubSub: wego.PubSubComponentConfig{
Enabled: true,
StreamPrefix: "wego:events",
},
}),
)
if err != nil {
panic(err)
}
reminderSvc := mesa.MustGetReminder()
pubsubSvc := mesa.MustGetPubSub()
_ = reminderSvc
_ = pubsubSvcpubsub provides a Redis Streams based component for cross-instance publish and subscribe with consumer groups.
- Publish messages to a topic backed by Redis Streams
- Register subscriptions before Mesa runtime starts
- Competing consumers within one consumer group
- Fan-out delivery through different consumer groups
- Retry failed deliveries with configurable backoff
- Ack on success and dead-letter on max delivery attempts
mesa, err := wego.New(
wego.WithDSN(dsn),
wego.WithRedisConfig(wego.RedisConfig{
Addr: "127.0.0.1:6379",
}),
wego.WithComponents(wego.ComponentsConfig{
PubSub: wego.PubSubComponentConfig{
Enabled: true,
},
}),
)
if err != nil {
panic(err)
}
pubsubSvc := mesa.MustGetPubSub()
pubsubSvc.MustSubscribe(pubsub.Subscription{
Topic: "order.created",
Group: "billing",
Handler: pubsub.HandlerFunc(func(ctx context.Context, delivery *pubsub.Delivery) error {
fmt.Println("received:", delivery.Message.Type, string(delivery.Message.Data))
return nil
}),
})
_, err := pubsubSvc.Publish(context.Background(), "order.created", pubsub.Message{
Key: "order-123",
Type: "order.created",
Data: []byte(`{"id":"123"}`),
})
if err != nil {
panic(err)
}Runtime rules are:
- subscriptions must be registered before
mesa.Run(ctx) - handlers must be idempotent because delivery is at least once
- one
topic + groupmaps to one handler within a process - use different consumer groups when the same topic needs fan-out
- handler failures are re-published through an internal retry queue using the configured backoff policy
- retry queues are isolated by
topic + group; scheduling a retry and acknowledging the source entry are atomic on Redis - auto-registration only happens when
wego.WithComponents(...)orwego.WithPubSub(...)is provided and Redis is configured
Pub/Sub can also be configured with framework-level fields such as Name, RetryPolicy, StreamPrefix, StreamMaxLen, StreamBlock, StreamReadCount, and dead-letter options through wego.PubSubComponentConfig.
The Redis-backed integration tests for pubsub are opt-in and use the integration build tag.
Required environment variables:
WEGO_REDIS_TEST_ADDR: Redis address such as127.0.0.1:6379
Optional environment variables:
WEGO_REDIS_TEST_PASSWORD: Redis passwordWEGO_REDIS_TEST_DB: Redis database index, defaults to15
Run the default unit tests:
go test ./pubsubRun the Redis integration tests:
WEGO_REDIS_TEST_ADDR=127.0.0.1:6379 go test -tags=integration ./pubsubRun the full repository test suite without integration tests:
go test ./...If WEGO_REDIS_TEST_ADDR is not set or Redis is unreachable, the integration tests are skipped.
PostgreSQL-backed Reminder tests are opt-in. Each test creates and removes an isolated schema and applies the production migration.
WEGO_POSTGRES_TEST_DSN='postgres://user:password@127.0.0.1:5432/land_contract?sslmode=disable' go test -tags=integration ./remindertransport/tcp provides a generic TCP server implementation that matches the existing transport.Server lifecycle.
tcpServer := tcp.NewTCPServer(
tcp.WithHost(net.ParseIP("0.0.0.0")),
tcp.WithPort(9000),
tcp.WithHandler(func(ctx context.Context, conn net.Conn) {
defer conn.Close()
buf := make([]byte, 1024)
n, err := conn.Read(buf)
if err != nil {
return
}
_, _ = conn.Write(buf[:n])
}),
)
mesa, err := wego.New(
wego.WithDSN(dsn),
wego.WithServers(tcpServer),
)
if err != nil {
panic(err)
}go get github.com/Jinchenyuan/wego@main
New only initializes configured dependencies and returns connection or configuration errors. Run accepts a context, and Shutdown is safe to call more than once.
When the default HTTP server is enabled with WithHttpPort, it exposes GET /livez and GET /readyz. The same status is available through mesa.Liveness() and mesa.Readiness(ctx).
mesa, err := wego.New(wego.WithHttpPort(8080))
if err != nil {
return err
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()
return mesa.Run(ctx)wego.Config is the typed production configuration surface. Applications remain
responsible for loading environment variables, files, and secrets, then call
Validate or Options before creating Mesa:
cfg := wego.Config{
Profile: wego.Profile{Name: "orders"},
HTTP: wego.HTTPConfig{Port: 8080},
Redis: wego.RedisConfig{Addr: "127.0.0.1:6379"},
}
opts, err := cfg.Options()
if err != nil {
return err
}
mesa, err := wego.New(opts...)Configuration errors include the invalid field path and may be inspected with
errors.As(err, *wego.ConfigError).
middleware.Authenticate accepts a context-aware Authenticator. Successful
authentication stores a Principal in the request context for handlers to read
with middleware.PrincipalFromContext. The existing AuthMiddleware remains
available for header ID plus cached-token schemes.
The default HTTP transport uses bounded header, request, response, idle, body,
and header sizes. Override them without breaking the legacy WithHttpPort API:
m, err := wego.New(wego.WithHTTPConfig(wego.HTTPConfig{
Port: 8080,
ReadHeaderTimeout: 3 * time.Second,
ReadTimeout: 10 * time.Second,
WriteTimeout: 20 * time.Second,
IdleTimeout: 60 * time.Second,
MaxHeaderBytes: 1 << 20,
MaxBodyBytes: 8 << 20,
RequestTimeout: 15 * time.Second,
}))/livez, /readyz, and /metrics are exposed by the default server. Logs are
newline-delimited JSON. Redis TLS can be configured with
RedisConfig.TLSConfig; TLS termination for public HTTP traffic is expected at
the Kubernetes Ingress. Reusable CORS, security-header, and per-instance rate
limit middleware is available in middleware.
Tracing is configured per Mesa instance and is disabled by default. When enabled, Wego exports spans through OTLP gRPC and automatically instruments the HTTP server, go-micro handlers/clients, PostgreSQL, Redis, Pub/Sub, and Reminder delivery. W3C trace context is propagated through HTTP headers and Pub/Sub message headers.
cfg := wego.Config{
Profile: wego.Profile{Name: "orders"},
Telemetry: telemetry.Config{
Enabled: true,
OTLPEndpoint: "otel-collector:4317",
Insecure: true, // use false with TLS credentials at the collector
TraceSampleRatio: 0.1,
},
}
opts, err := cfg.Options()
if err != nil {
return err
}
mesa, err := wego.New(opts...)The default /metrics endpoint remains available and includes bounded-label
request, database, Redis, Pub/Sub, and Reminder counters and duration summaries.
Telemetry never records request bodies, message payloads, credentials, user
IDs, or reminder keys.
examples/http-server is a minimal application that uses the Mesa lifecycle,
registers GET /hello, and includes Docker and Helm deployment assets. Run it
locally from the repository root:
go run ./examples/http-server
curl http://127.0.0.1:8080/helloThe default server also exposes /livez, /readyz, and /metrics. Override
the port with WEGO_HTTP_PORT.
Build the image with the repository root as the Docker context:
docker build -f examples/http-server/Dockerfile -t wego-http-server-example .
docker run --rm -p 8080:8080 wego-http-server-exampleInstall the example chart after publishing the image to a registry:
helm upgrade --install wego-http-server \
examples/http-server/deploy/helm/http-server \
--set image.repository=registry.example.com/wego-http-server-example \
--set image.tag=1.0.0Import packages to your .go files.
"github.com/Jinchenyuan/wego"
"github.com/Jinchenyuan/wego/logger"
"github.com/Jinchenyuan/wego/middleware"
"github.com/Jinchenyuan/wego/transport"
"github.com/Jinchenyuan/wego/transport/micro"Run a mesa instance with your configuration.
// read config from file
cfg, err := config.Read("config.toml")
if err != nil {
fmt.Printf("failed to read config: %v\n", err)
return
}
m, err := wego.New(
wego.WithEtcdConfig(clientv3.Config{
Endpoints: cfg.Etcd.Endpoints,
DialTimeout: 5 * time.Second,
Username: cfg.Etcd.User,
Password: cfg.Etcd.Password,
}),
wego.WithHttpPort(cfg.Http.Port),
wego.WithDSN(cfg.PostgreSQL.DSN),
wego.WithLogLevel(logger.ParseLevel(cfg.Log.Level)),
wego.WithRedisConfig(wego.RedisConfig{
Addr: cfg.Redis.Addr,
Password: cfg.Redis.Password,
DB: cfg.Redis.DB,
}),
wego.WithProfile(wego.Profile{
Name: cfg.Profile.Name,
}),
)
if err != nil {
fmt.Printf("failed to initialize mesa: %v\n", err)
return
}
ginhandler.SetAuthMiddleware(middleware.AuthMiddleware("account", func(id string) string {
cacheToken, err := m.Redis.Get(context.Background(), fmt.Sprintf("token:%s", id)).Result()
if err != nil {
return ""
}
return cacheToken
}, cfg.Http.ExcludeAuthPaths...))
ginhandler.Registry()
ms := m.GetServerByType(transport.MICRO_SERVER).(*micro.Service)
ms.NewServiceClients(serviceclient.Registry)
if err := m.Run(context.Background()); err != nil {
fmt.Printf("failed to run mesa: %v\n", err)
}example project see: https://github.com/Jinchenyuan/weserver