feat: init repo

backend/app/errorx/error.go

@@ -0,0 +1,147 @@
+package errorx
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"runtime"
+	"strings"
+
+	"github.com/go-jet/jet/v2/qrm"
+	"github.com/gofiber/fiber/v3"
+	"github.com/gofiber/fiber/v3/binder"
+	"github.com/gofiber/utils/v2"
+	log "github.com/sirupsen/logrus"
+)
+
+func Middleware(c fiber.Ctx) error {
+	err := c.Next()
+	if err != nil {
+		return Wrap(err).Response(c)
+	}
+	return err
+}
+
+type Response struct {
+	isFormat bool
+	err      error
+	params   []any
+	sql      string
+	file     string
+
+	StatusCode int    `json:"-" xml:"-"`
+	Code       int    `json:"code" xml:"code"`
+	Message    string `json:"message" xml:"message"`
+	Data       any    `json:"data,omitempty" xml:"data"`
+}
+
+func New(code, statusCode int, message string) *Response {
+	return &Response{
+		isFormat:   true,
+		StatusCode: statusCode,
+		Code:       code,
+		Message:    message,
+	}
+}
+
+func (r *Response) Sql(sql string) *Response {
+	r.sql = sql
+	return r
+}
+
+func (r *Response) from(err *Response) *Response {
+	r.Code = err.Code
+	r.Message = err.Message
+	r.StatusCode = err.StatusCode
+	return r
+}
+
+func (r *Response) Params(params ...any) *Response {
+	r.params = params
+	if _, file, line, ok := runtime.Caller(1); ok {
+		r.file = fmt.Sprintf("%s:%d", file, line)
+	}
+	return r
+}
+
+func Wrap(err error) *Response {
+	if e, ok := err.(*Response); ok {
+		return e
+	}
+	return &Response{err: err}
+}
+
+func (r *Response) Wrap(err error) *Response {
+	r.err = err
+	return r
+}
+
+func (r *Response) format() {
+	r.isFormat = true
+	if errors.Is(r.err, qrm.ErrNoRows) {
+		r.from(RecordNotExists)
+		return
+	}
+
+	if e, ok := r.err.(*fiber.Error); ok {
+		r.Code = e.Code
+		r.Message = e.Message
+		r.StatusCode = e.Code
+		return
+	}
+
+	if r.err != nil {
+		msg := r.err.Error()
+		if strings.Contains(msg, "duplicate key value") || strings.Contains(msg, "unique constraint") {
+			r.from(RecordDuplicated)
+			return
+		}
+
+		r.Code = http.StatusInternalServerError
+		r.StatusCode = http.StatusInternalServerError
+		r.Message = msg
+	}
+	return
+}
+
+func (r *Response) Error() string {
+	if !r.isFormat {
+		r.format()
+	}
+
+	return fmt.Sprintf("[%d] %s", r.Code, r.Message)
+}
+
+func (r *Response) Response(ctx fiber.Ctx) error {
+	if !r.isFormat {
+		r.format()
+	}
+
+	contentType := utils.ToLower(utils.UnsafeString(ctx.Request().Header.ContentType()))
+	contentType = binder.FilterFlags(utils.ParseVendorSpecificContentType(contentType))
+
+	log.
+		WithError(r.err).
+		WithField("file", r.file).
+		WithField("sql", r.sql).
+		WithField("params", r.params).
+		Errorf("response error: %+v", r)
+
+	// Parse body accordingly
+	switch contentType {
+	case fiber.MIMETextXML, fiber.MIMEApplicationXML:
+		return ctx.Status(r.StatusCode).XML(r)
+	case fiber.MIMETextHTML, fiber.MIMETextPlain:
+		return ctx.Status(r.StatusCode).SendString(r.Message)
+	default:
+		return ctx.Status(r.StatusCode).JSON(r)
+	}
+}
+
+var (
+	RecordDuplicated = New(1001, http.StatusBadRequest, "记录重复")
+	RecordNotExists  = New(http.StatusNotFound, http.StatusNotFound, "记录不存在")
+	BadRequest       = New(http.StatusBadRequest, http.StatusBadRequest, "请求错误")
+	Unauthorized     = New(http.StatusUnauthorized, http.StatusUnauthorized, "未授权")
+	InternalErr      = New(http.StatusInternalServerError, http.StatusInternalServerError, "内部错误")
+)

backend/app/events/event_demo.go

@@ -0,0 +1,71 @@
+package events
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"git.ipao.vip/rogeecn/atom/contracts"
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/sirupsen/logrus"
+)
+
+var _ contracts.EventHandler = (*UserRegister)(nil)
+
+type Event struct {
+	ID int `json:"id"`
+}
+
+type ProcessedEvent struct {
+	ProcessedID int       `json:"processed_id"`
+	Time        time.Time `json:"time"`
+}
+
+// @provider(event)
+type UserRegister struct {
+	log *logrus.Entry `inject:"false"`
+}
+
+func (u *UserRegister) Prepare() error {
+	return nil
+}
+
+// Handler implements contracts.EventHandler.
+func (u *UserRegister) Handler(msg *message.Message) ([]*message.Message, error) {
+	consumedPayload := Event{}
+	err := json.Unmarshal(msg.Payload, &consumedPayload)
+	if err != nil {
+		// When a handler returns an error, the default behavior is to send a Nack (negative-acknowledgement).
+		// The message will be processed again.
+		//
+		// You can change the default behaviour by using middlewares, like Retry or PoisonQueue.
+		// You can also implement your own middleware.
+		return nil, err
+	}
+
+	fmt.Printf("received event %+v\n", consumedPayload)
+
+	newPayload, err := json.Marshal(ProcessedEvent{
+		ProcessedID: consumedPayload.ID,
+		Time:        time.Now(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	newMessage := message.NewMessage(watermill.NewUUID(), newPayload)
+
+	return nil, nil
+	return []*message.Message{newMessage}, nil
+}
+
+// PublishToTopic implements contracts.EventHandler.
+func (u *UserRegister) PublishToTopic() string {
+	return "event:processed"
+}
+
+// Topic implements contracts.EventHandler.
+func (u *UserRegister) Topic() string {
+	return "event:user-register"
+}

backend/app/events/provider.gen.go

@@ -0,0 +1,27 @@
+package events
+
+import (
+	"backend/providers/events"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	"git.ipao.vip/rogeecn/atom/utils/opt"
+)
+
+func Provide(opts ...opt.Option) error {
+	if err := container.Container.Provide(func(
+		__event *events.PubSub,
+	) (contracts.Initial, error) {
+		obj := &UserRegister{}
+		if err := obj.Prepare(); err != nil {
+			return nil, err
+		}
+		__event.Handle("handler:UserRegister", obj.Topic(), obj.PublishToTopic(), obj.Handler)
+
+		return obj, nil
+	}, atom.GroupInitial); err != nil {
+		return err
+	}
+	return nil
+}

backend/app/grpc/users/handler.go

@@ -0,0 +1,26 @@
+package users
+
+import (
+	"context"
+
+	userv1 "backend/pkg/proto/user/v1"
+)
+
+// @provider(grpc) userv1.RegisterUserServiceServer
+type Users struct {
+	userv1.UnimplementedUserServiceServer
+}
+
+func (u *Users) ListUsers(ctx context.Context, in *userv1.ListUsersRequest) (*userv1.ListUsersResponse, error) {
+	// userv1.UserServiceServer
+	return &userv1.ListUsersResponse{}, nil
+}
+
+// GetUser implements userv1.UserServiceServer
+func (u *Users) GetUser(ctx context.Context, in *userv1.GetUserRequest) (*userv1.GetUserResponse, error) {
+	return &userv1.GetUserResponse{
+		User: &userv1.User{
+			Id: in.Id,
+		},
+	}, nil
+}

backend/app/grpc/users/provider.gen.go

@@ -0,0 +1,25 @@
+package users
+
+import (
+	userv1 "backend/pkg/proto/user/v1"
+	"backend/providers/grpc"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	"git.ipao.vip/rogeecn/atom/utils/opt"
+)
+
+func Provide(opts ...opt.Option) error {
+	if err := container.Container.Provide(func(
+		__grpc *grpc.Grpc,
+	) (contracts.Initial, error) {
+		obj := &Users{}
+		userv1.RegisterUserServiceServer(__grpc.Server, obj)
+
+		return obj, nil
+	}, atom.GroupInitial); err != nil {
+		return err
+	}
+	return nil
+}

backend/app/jobs/demo_cron.go

@@ -0,0 +1,50 @@
+package jobs
+
+import (
+	"time"
+
+	_ "git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	"github.com/riverqueue/river"
+	"github.com/sirupsen/logrus"
+)
+
+var _ contracts.CronJob = (*CronJob)(nil)
+
+// @provider contracts.CronJob atom.GroupCronJob
+type CronJob struct {
+	log *logrus.Entry `inject:"false"`
+}
+
+func (cron *CronJob) Prepare() error {
+	cron.log = logrus.WithField("module", "cron")
+	return nil
+}
+
+func (cron *CronJob) Description() string {
+	return "hello world cron job"
+}
+
+// InsertOpts implements contracts.CronJob.
+func (cron *CronJob) InsertOpts() *river.InsertOpts {
+	return nil
+}
+
+// JobArgs implements contracts.CronJob.
+func (cron *CronJob) JobArgs() []river.JobArgs {
+	return []river.JobArgs{
+		SortArgs{
+			Strings: []string{"a", "c", "b", "d"},
+		},
+	}
+}
+
+// Periodic implements contracts.CronJob.
+func (cron *CronJob) Periodic() time.Duration {
+	return time.Second * 10
+}
+
+// RunOnStart implements contracts.CronJob.
+func (cron *CronJob) RunOnStart() bool {
+	return true
+}

backend/app/jobs/demo_job.go

@@ -0,0 +1,53 @@
+package jobs
+
+import (
+	"context"
+	"sort"
+	"time"
+
+	_ "git.ipao.vip/rogeecn/atom"
+	_ "git.ipao.vip/rogeecn/atom/contracts"
+	. "github.com/riverqueue/river"
+	log "github.com/sirupsen/logrus"
+)
+
+// provider:[except|only] [returnType] [group]
+
+var (
+	_ JobArgs               = SortArgs{}
+	_ JobArgsWithInsertOpts = SortArgs{}
+)
+
+type SortArgs struct {
+	Strings []string `json:"strings"`
+}
+
+// InsertOpts implements JobArgsWithInsertOpts.
+func (s SortArgs) InsertOpts() InsertOpts {
+	return InsertOpts{
+		Queue:    QueueDefault,
+		Priority: PriorityDefault,
+	}
+}
+
+func (SortArgs) Kind() string {
+	return "sort"
+}
+
+var _ Worker[SortArgs] = (*SortWorker)(nil)
+
+// @provider(job)
+type SortWorker struct {
+	WorkerDefaults[SortArgs]
+}
+
+func (w *SortWorker) Work(ctx context.Context, job *Job[SortArgs]) error {
+	sort.Strings(job.Args.Strings)
+
+	log.Infof("[%s] Sorted strings: %v\n", time.Now().Format(time.TimeOnly), job.Args.Strings)
+	return nil
+}
+
+func (w *SortWorker) NextRetry(job *Job[SortArgs]) time.Time {
+	return time.Now().Add(5 * time.Second)
+}

backend/app/jobs/provider.gen.go

@@ -0,0 +1,37 @@
+package jobs
+
+import (
+	"backend/providers/job"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	"git.ipao.vip/rogeecn/atom/utils/opt"
+	"github.com/riverqueue/river"
+)
+
+func Provide(opts ...opt.Option) error {
+	if err := container.Container.Provide(func() (contracts.CronJob, error) {
+		obj := &CronJob{}
+		if err := obj.Prepare(); err != nil {
+			return nil, err
+		}
+
+		return obj, nil
+	}, atom.GroupCronJob); err != nil {
+		return err
+	}
+	if err := container.Container.Provide(func(
+		__job *job.Job,
+	) (contracts.Initial, error) {
+		obj := &SortWorker{}
+		if err := river.AddWorkerSafely(__job.Workers, obj); err != nil {
+			return nil, err
+		}
+
+		return obj, nil
+	}, atom.GroupInitial); err != nil {
+		return err
+	}
+	return nil
+}

backend/app/middlewares/m_check_ua.go

@@ -0,0 +1,17 @@
+package middlewares
+
+import (
+	"strings"
+
+	"github.com/gofiber/fiber/v3"
+)
+
+func (m *Middlewares) CheckUA(ctx fiber.Ctx) error {
+	keyword := strings.ToLower("MicroMessenger")
+	userAgent := ctx.GetReqHeaders()["User-Agent"][0]
+
+	if strings.Contains(userAgent, keyword) {
+		return ctx.SendString("")
+	}
+	return ctx.Next()
+}

backend/app/middlewares/m_jwt_parse.go

@@ -0,0 +1,37 @@
+package middlewares
+
+import (
+	"time"
+
+	"backend/app/errorx"
+
+	"github.com/gofiber/fiber/v3"
+	log "github.com/sirupsen/logrus"
+)
+
+func (f *Middlewares) ParseJWT(c fiber.Ctx) error {
+	tokens := c.GetReqHeaders()["Authorization"]
+	if len(tokens) == 0 {
+		queryToken := c.Query("token")
+		tokens = []string{queryToken}
+		if len(tokens) == 0 {
+			return c.Next()
+		}
+	}
+
+	token := tokens[0]
+	claim, err := f.jwt.Parse(token)
+	if err != nil {
+		c.Cookie(&fiber.Cookie{
+			Name:     "token",
+			Value:    "",
+			Expires:  time.Now().Add(-1 * time.Hour),
+			HTTPOnly: true,
+		})
+		log.Errorf("failed to parse jwt from token: %s", token)
+		return errorx.Unauthorized
+	}
+	_ = claim
+
+	return c.Next()
+}

backend/app/middlewares/m_wechat_auth.go

@@ -0,0 +1,66 @@
+package middlewares
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"backend/providers/wechat"
+
+	"github.com/gofiber/fiber/v3"
+	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
+)
+
+const StatePrefix = "sns_basic_auth"
+
+func (f *Middlewares) WeChatAuth(c fiber.Ctx) error {
+	log := log.WithField("module", "middleware.AuthUserInfo")
+	log.Debugf("%s, query: %v", c.OriginalURL(), c.Queries())
+	state := c.Query("state")
+	code := c.Query("code")
+	log.Debugf("code: %s, state: %s", code, state)
+
+	jwtToken := c.Cookies("token")
+	if jwtToken != "" {
+		log.Debugf("jwtToken: %s", jwtToken)
+
+		if _, err := f.jwt.Parse(jwtToken); err != nil {
+			log.WithError(err).Error("failed to parse jwt token")
+
+			c.Cookie(&fiber.Cookie{
+				Name:     "token",
+				Value:    "",
+				Expires:  time.Now().Add(-1 * time.Hour),
+				HTTPOnly: true,
+			})
+			return c.Redirect().To(c.Path())
+		}
+	}
+
+	if state == "" && code == "" {
+		url := string(c.Request().URI().FullURI())
+		url = strings.ReplaceAll(url, "http", "https")
+		url = strings.ReplaceAll(url, c.BaseURL(), *f.app.BaseURI)
+
+		log.WithField("module", "middleware.SilentAuth").Debug("redirect_uri: ", url)
+
+		to, err := f.client.ScopeAuthorizeURL(
+			wechat.ScopeAuthorizeURLWithRedirectURI(url),
+			wechat.ScopeAuthorizeURLWithState(fmt.Sprintf("%s_%d", StatePrefix, time.Now().UnixNano())),
+		)
+		if err != nil {
+			return errors.Wrap(err, "failed to get wechat auth url")
+		}
+		log.WithField("module", "middleware.SilentAuth").Debug("redirectTo: ", to.String())
+
+		return c.Redirect().To(to.String())
+
+	}
+
+	if !strings.HasPrefix(state, StatePrefix) || code == "" {
+		return errors.New("invalid request")
+	}
+
+	return c.Next()
+}

backend/app/middlewares/m_wechat_verify.go

@@ -0,0 +1,33 @@
+package middlewares
+
+import (
+	"github.com/gofiber/fiber/v3"
+	log "github.com/sirupsen/logrus"
+)
+
+// 此方法用于微信首次接入时的数据验证
+func (f *Middlewares) WeChatVerify(c fiber.Ctx) error {
+	// get the query parameters
+	signature := c.Query("signature")
+	timestamp := c.Query("timestamp")
+	nonce := c.Query("nonce")
+	echostr := c.Query("echostr")
+
+	if signature == "" || timestamp == "" || nonce == "" || echostr == "" {
+		return c.Next()
+	}
+
+	log.WithField("method", "Verify").WithFields(log.Fields{
+		"signature": signature,
+		"timestamp": timestamp,
+		"nonce":     nonce,
+		"echostr":   echostr,
+	}).Debug("begin verify signature")
+
+	// verify the signature
+	if err := f.client.Verify(signature, timestamp, nonce); err != nil {
+		return c.SendString(err.Error())
+	}
+
+	return c.SendString(echostr)
+}

backend/app/middlewares/mid_debug.go

@@ -0,0 +1,9 @@
+package middlewares
+
+import (
+	"github.com/gofiber/fiber/v3"
+)
+
+func (f *Middlewares) DebugMode(c fiber.Ctx) error {
+	return c.Next()
+}

backend/app/middlewares/middlewares.go

@@ -0,0 +1,25 @@
+package middlewares
+
+import (
+	"backend/providers/app"
+	"backend/providers/jwt"
+	"backend/providers/storage"
+	"backend/providers/wechat"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// @provider
+type Middlewares struct {
+	log *log.Entry `inject:"false"`
+
+	app         *app.Config
+	storagePath *storage.Config
+	jwt         *jwt.JWT
+	client      *wechat.Client
+}
+
+func (f *Middlewares) Prepare() error {
+	f.log = log.WithField("module", "middleware")
+	return nil
+}

backend/app/middlewares/provider.gen.go

@@ -0,0 +1,20 @@
+package middlewares
+
+import (
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/utils/opt"
+)
+
+func Provide(opts ...opt.Option) error {
+	if err := container.Container.Provide(func() (*Middlewares, error) {
+		obj := &Middlewares{}
+		if err := obj.Prepare(); err != nil {
+			return nil, err
+		}
+
+		return obj, nil
+	}); err != nil {
+		return err
+	}
+	return nil
+}

backend/app/requests/pagination.go

@@ -0,0 +1,30 @@
+package requests
+
+import "github.com/samber/lo"
+
+type Pager struct {
+	Pagination `json:",inline"`
+	Total      int64       `json:"total"`
+	Items      interface{} `json:"items"`
+}
+
+type Pagination struct {
+	Page  int `json:"page" form:"page" query:"page"`
+	Limit int `json:"limit" form:"limit" query:"limit"`
+}
+
+func (filter *Pagination) Offset() int {
+	return (filter.Page - 1) * filter.Limit
+}
+
+func (filter *Pagination) Format() *Pagination {
+	if filter.Page <= 0 {
+		filter.Page = 1
+	}
+
+	if !lo.Contains([]int{10, 20, 50, 100}, filter.Limit) {
+		filter.Limit = 10
+	}
+
+	return filter
+}

backend/app/requests/sort.go

@@ -0,0 +1,41 @@
+package requests
+
+import (
+	"strings"
+
+	"github.com/samber/lo"
+)
+
+type SortQueryFilter struct {
+	Asc  *string `json:"asc" form:"asc"`
+	Desc *string `json:"desc" form:"desc"`
+}
+
+func (s *SortQueryFilter) AscFields() []string {
+	if s.Asc == nil {
+		return nil
+	}
+	return strings.Split(*s.Asc, ",")
+}
+
+func (s *SortQueryFilter) DescFields() []string {
+	if s.Desc == nil {
+		return nil
+	}
+	return strings.Split(*s.Desc, ",")
+}
+
+func (s *SortQueryFilter) DescID() *SortQueryFilter {
+	if s.Desc == nil {
+		s.Desc = lo.ToPtr("id")
+	}
+
+	items := s.DescFields()
+	if lo.Contains(items, "id") {
+		return s
+	}
+
+	items = append(items, "id")
+	s.Desc = lo.ToPtr(strings.Join(items, ","))
+	return s
+}

backend/app/service/event/event.go

@@ -0,0 +1,58 @@
+package event
+
+import (
+	"context"
+
+	"backend/app/events"
+	"backend/app/service"
+	"backend/providers/app"
+	providerEvents "backend/providers/events"
+	"backend/providers/postgres"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"go.uber.org/dig"
+)
+
+func defaultProviders() container.Providers {
+	return service.Default(container.Providers{
+		postgres.DefaultProvider(),
+	}...)
+}
+
+func Command() atom.Option {
+	return atom.Command(
+		atom.Name("event"),
+		atom.Short("start event processor"),
+		atom.RunE(Serve),
+		atom.Providers(
+			defaultProviders().
+				With(
+					events.Provide,
+				),
+		),
+	)
+}
+
+type Service struct {
+	dig.In
+
+	App      *app.Config
+	PubSub   *providerEvents.PubSub
+	Initials []contracts.Initial `group:"initials"`
+}
+
+func Serve(cmd *cobra.Command, args []string) error {
+	return container.Container.Invoke(func(ctx context.Context, svc Service) error {
+		log.SetFormatter(&log.JSONFormatter{})
+
+		if svc.App.IsDevMode() {
+			log.SetLevel(log.DebugLevel)
+		}
+
+		return svc.PubSub.Serve(ctx)
+	})
+}

backend/app/service/grpc/grpc.go

@@ -0,0 +1,58 @@
+package grpc
+
+import (
+	"backend/app/grpc/users"
+	"backend/app/service"
+	_ "backend/docs"
+	"backend/providers/app"
+	"backend/providers/grpc"
+	"backend/providers/postgres"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"go.uber.org/dig"
+)
+
+func defaultProviders() container.Providers {
+	return service.Default(container.Providers{
+		postgres.DefaultProvider(),
+		grpc.DefaultProvider(),
+	}...)
+}
+
+func Command() atom.Option {
+	return atom.Command(
+		atom.Name("grpc"),
+		atom.Short("run grpc server"),
+		atom.RunE(Serve),
+		atom.Providers(
+			defaultProviders().
+				With(
+					users.Provide,
+				),
+		),
+	)
+}
+
+type Service struct {
+	dig.In
+
+	App      *app.Config
+	Grpc     *grpc.Grpc
+	Initials []contracts.Initial `group:"initials"`
+}
+
+func Serve(cmd *cobra.Command, args []string) error {
+	return container.Container.Invoke(func(svc Service) error {
+		log.SetFormatter(&log.JSONFormatter{})
+
+		if svc.App.IsDevMode() {
+			log.SetLevel(log.DebugLevel)
+		}
+
+		return svc.Grpc.Serve()
+	})
+}

backend/app/service/http/http.go

@@ -0,0 +1,85 @@
+package http
+
+import (
+	"backend/app/errorx"
+	"backend/app/jobs"
+	"backend/app/middlewares"
+	"backend/app/service"
+	_ "backend/docs"
+	"backend/providers/app"
+	"backend/providers/hashids"
+	"backend/providers/http"
+	"backend/providers/http/swagger"
+	"backend/providers/job"
+	"backend/providers/jwt"
+	"backend/providers/postgres"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	"github.com/gofiber/fiber/v3/middleware/favicon"
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"go.uber.org/dig"
+)
+
+func defaultProviders() container.Providers {
+	return service.Default(container.Providers{
+		http.DefaultProvider(),
+		postgres.DefaultProvider(),
+		jwt.DefaultProvider(),
+		hashids.DefaultProvider(),
+		job.DefaultProvider(),
+	}...)
+}
+
+func Command() atom.Option {
+	return atom.Command(
+		atom.Name("serve"),
+		atom.Short("run http server"),
+		atom.RunE(Serve),
+		atom.Providers(
+			defaultProviders().
+				With(
+					jobs.Provide,
+				),
+		),
+	)
+}
+
+type Service struct {
+	dig.In
+
+	App *app.Config
+	Job *job.Job
+
+	Middlewares *middlewares.Middlewares
+	Http        *http.Service
+
+	Initials []contracts.Initial   `group:"initials"`
+	Routes   []contracts.HttpRoute `group:"routes"`
+}
+
+func Serve(cmd *cobra.Command, args []string) error {
+	return container.Container.Invoke(func(svc Service) error {
+		log.SetFormatter(&log.JSONFormatter{})
+
+		if svc.App.IsDevMode() {
+			log.SetLevel(log.DebugLevel)
+			svc.Http.Engine.Get("/swagger/*", swagger.HandlerDefault)
+		}
+
+		svc.Http.Engine.Use(svc.Middlewares.WeChatVerify)
+		svc.Http.Engine.Use(errorx.Middleware)
+		svc.Http.Engine.Use(favicon.New(favicon.Config{
+			Data: []byte{},
+		}))
+
+		group := svc.Http.Engine.Group("")
+		for _, route := range svc.Routes {
+			route.Register(group)
+		}
+
+		return svc.Http.Serve()
+	})
+}

backend/app/service/queue/error.go

@@ -0,0 +1,24 @@
+package queue
+
+import (
+	"context"
+
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/rivertype"
+	log "github.com/sirupsen/logrus"
+)
+
+type CustomErrorHandler struct{}
+
+func (*CustomErrorHandler) HandleError(ctx context.Context, job *rivertype.JobRow, err error) *river.ErrorHandlerResult {
+	log.Infof("Job errored with: %s\n", err)
+	return nil
+}
+
+func (*CustomErrorHandler) HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *river.ErrorHandlerResult {
+	log.Infof("Job panicked with: %v\n", panicVal)
+	log.Infof("Stack trace: %s\n", trace)
+	return &river.ErrorHandlerResult{
+		SetCancelled: true,
+	}
+}

backend/app/service/queue/river.go

@@ -0,0 +1,94 @@
+package queue
+
+import (
+	"context"
+
+	"backend/app/jobs"
+	"backend/app/service"
+	"backend/providers/app"
+	"backend/providers/job"
+	"backend/providers/postgres"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"git.ipao.vip/rogeecn/atom/contracts"
+	"github.com/riverqueue/river"
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"go.uber.org/dig"
+)
+
+func defaultProviders() container.Providers {
+	return service.Default(container.Providers{
+		postgres.DefaultProvider(),
+		job.DefaultProvider(),
+	}...)
+}
+
+func Command() atom.Option {
+	return atom.Command(
+		atom.Name("queue"),
+		atom.Short("start queue processor"),
+		atom.RunE(Serve),
+		atom.Providers(
+			defaultProviders().
+				With(
+					jobs.Provide,
+				),
+		),
+	)
+}
+
+type Service struct {
+	dig.In
+
+	App      *app.Config
+	Job      *job.Job
+	Initials []contracts.Initial `group:"initials"`
+	CronJobs []contracts.CronJob `group:"cron_jobs"`
+}
+
+func Serve(cmd *cobra.Command, args []string) error {
+	return container.Container.Invoke(func(ctx context.Context, svc Service) error {
+		log.SetFormatter(&log.JSONFormatter{})
+
+		if svc.App.IsDevMode() {
+			log.SetLevel(log.DebugLevel)
+		}
+
+		client, err := svc.Job.Client()
+		if err != nil {
+			return err
+		}
+
+		for _, cronJob := range svc.CronJobs {
+			log.
+				WithField("module", "cron").
+				WithField("name", cronJob.Description()).
+				WithField("duration", cronJob.Periodic().Seconds()).
+				Info("registering cron job")
+
+			for _, jobArgs := range cronJob.JobArgs() {
+				client.PeriodicJobs().Add(
+					river.NewPeriodicJob(
+						river.PeriodicInterval(cronJob.Periodic()),
+						func() (river.JobArgs, *river.InsertOpts) {
+							return jobArgs, cronJob.InsertOpts()
+						},
+						&river.PeriodicJobOpts{
+							RunOnStart: cronJob.RunOnStart(),
+						},
+					),
+				)
+			}
+		}
+
+		if err := client.Start(ctx); err != nil {
+			return err
+		}
+		defer client.StopAndCancel(ctx)
+
+		<-ctx.Done()
+		return nil
+	})
+}

backend/app/service/service.go

@@ -0,0 +1,15 @@
+package service
+
+import (
+	"backend/providers/app"
+	"backend/providers/events"
+
+	"git.ipao.vip/rogeecn/atom/container"
+)
+
+func Default(providers ...container.ProviderContainer) container.Providers {
+	return append(container.Providers{
+		app.DefaultProvider(),
+		events.DefaultProvider(),
+	}, providers...)
+}

backend/app/service/testx/testing.go

@@ -0,0 +1,29 @@
+package testx
+
+import (
+	"os"
+	"testing"
+
+	"git.ipao.vip/rogeecn/atom"
+	"git.ipao.vip/rogeecn/atom/container"
+	"github.com/rogeecn/fabfile"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func Default(providers ...container.ProviderContainer) container.Providers {
+	return append(container.Providers{}, providers...)
+}
+
+func Serve(providers container.Providers, t *testing.T, invoke any) {
+	Convey("tests boot up", t, func() {
+		file := fabfile.MustFind("config.toml")
+
+		localEnv := os.Getenv("ENV_LOCAL")
+		if localEnv != "" {
+			file = fabfile.MustFind("config." + localEnv + ".toml")
+		}
+
+		So(atom.LoadProviders(file, providers), ShouldBeNil)
+		So(container.Container.Invoke(invoke), ShouldBeNil)
+	})
+}