29169681577

Committed 11 Jul 2026 09:57PM UTC coverage: 66.889% (+1.3%) from 65.611%

Build # 29169681577

Build Type

push

github

Committed by

didil

Commit Message

feat:rate limiting

Coverage Stats

122 of 148 new or added lines in 9 files covered. (82.43%)

1303 of 1948 relevant lines covered (66.89%)

3.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/cmd/api/main.go

package main

import (
        "context"
        "flag"
        "fmt"
        "log"
        "net/http"
        "os"
        "os/signal"
        "sync"
        "syscall"

        "github.com/didil/inhooks/pkg/lib"
        "github.com/didil/inhooks/pkg/server"
        "github.com/didil/inhooks/pkg/server/handlers"
        "github.com/didil/inhooks/pkg/services"
        "github.com/didil/inhooks/pkg/supervisor"
        versionpkg "github.com/didil/inhooks/pkg/version"
        "go.uber.org/zap"
)

var (
        version = "dev"
)

func main() {
        versionpkg.SetVersion(version)

        // handle version command
        isVersionCmd := flag.Bool("version", false, "print the version")
        flag.Parse()
        if *isVersionCmd {
                fmt.Println(version)
                os.Exit(0)
        }

        // start server
        err := lib.LoadEnv()
        if err != nil {
                log.Fatalf("failed to load env: %v", err)
        }

        appConf, err := lib.InitAppConfig(context.Background())
        if err != nil {
                log.Fatalf("failed to process config: %v", err)
        }

        logger, err := lib.NewLogger(appConf)
        if err != nil {
                log.Fatalf("failed to initialize logger: %v", err)
        }

        logger.Info("starting Inhooks", zap.String("version", version))

        inhooksConfigSvc := services.NewInhooksConfigService(logger, appConf)
        logger.Info("loading inhooks config", zap.String("inhooksConfigFile", appConf.InhooksConfigFile))

        err = inhooksConfigSvc.Load(appConf.InhooksConfigFile)
        if err != nil {
                logger.Fatal("failed to load inhooks config", zap.Error(err))
        }

        timeSvc := services.NewTimeService()

        messageBuilder := services.NewMessageBuilder(timeSvc)

        redisClient, err := lib.InitRedisClient(appConf)
        if err != nil {
                logger.Fatal("failed to init redis client", zap.Error(err))
        }
        redisStore, err := services.NewRedisStore(redisClient, appConf.Redis.InhooksDBName)
        if err != nil {
                logger.Fatal("failed to init redis store", zap.Error(err))
        }

        messageEnqueuer := services.NewMessageEnqueuer(redisStore, timeSvc)
        messageFetcher := services.NewMessageFetcher(redisStore, timeSvc)
        messageVerifier := services.NewMessageVerifier()
        messageTransformer := services.NewMessageTransformer(&appConf.Transform)

        app := handlers.NewApp(
                handlers.WithLogger(logger),
                handlers.WithInhooksConfigService(inhooksConfigSvc),
                handlers.WithMessageBuilder(messageBuilder),
                handlers.WithMessageEnqueuer(messageEnqueuer),
                handlers.WithMessageVerifier(messageVerifier),
                handlers.WithMessageTransformer(messageTransformer),
        )

        r := server.NewRouter(app)

        addr := fmt.Sprintf("%s:%d", appConf.Server.Host, appConf.Server.Port)
        httpServer := http.Server{
                Addr:    addr,
                Handler: r,
        }

        wg := &sync.WaitGroup{}

        wg.Add(1)
        go func() {
                logger.Info("listening ...", zap.String("addr", addr))
                err = httpServer.ListenAndServe()
                if err != nil && err != http.ErrServerClosed {
                        logger.Fatal("listener failure", zap.Error(err))
                }
                logger.Info("http server shutdown")
                wg.Done()
        }()

        httpClient := lib.NewHttpClient(appConf)

        messageProcessor := services.NewMessageProcessor(httpClient)
        retryCalculator := services.NewRetryCalculator()
        processingResultsSvc := services.NewProcessingResultsService(timeSvc, redisStore, retryCalculator)
        schedulerSvc := services.NewSchedulerService(redisStore, timeSvc)

        processingRecoverySvc, err := services.NewProcessingRecoveryService(redisStore)
        if err != nil {
                logger.Fatal("failed to init ProcessingRecoveryService", zap.Error(err))
        }

        cleanupSvc := services.NewCleanupService(redisStore, timeSvc)

        queueMetricsSvc := services.NewQueueMetricsService(redisStore)

        rateLimiter := services.NewTokenBucketLimiter(redisStore, timeSvc, logger)

        svisor := supervisor.NewSupervisor(
                supervisor.WithLogger(logger),
                supervisor.WithMessageFetcher(messageFetcher),
                supervisor.WithAppConfig(appConf),
                supervisor.WithInhooksConfigService(inhooksConfigSvc),
                supervisor.WithMessageProcessor(messageProcessor),
                supervisor.WithProcessingResultsService(processingResultsSvc),
                supervisor.WithSchedulerService(schedulerSvc),
                supervisor.WithProcessingRecoveryService(processingRecoverySvc),
                supervisor.WithCleanupService(cleanupSvc),
                supervisor.WithMessageTransformer(messageTransformer),
                supervisor.WithQueueMetricsService(queueMetricsSvc),
                supervisor.WithRateLimiter(rateLimiter),
        )

        wg.Add(1)
        go func() {
                logger.Info("starting supervisor ...")
                svisor.Start()
                logger.Info("supervisor shutdown")
                wg.Done()
        }()

        sigs := make(chan os.Signal, 1)
        signal.Notify(sigs, os.Interrupt, syscall.SIGTERM)

        sig := <-sigs
        logger.Info("received shutdown signal, shutting down process", zap.String("signal", sig.String()))

        svisor.Shutdown()

        serverShutdownContext, cancel := context.WithTimeout(context.Background(), appConf.Server.ShutdownGracePeriod)
        defer cancel()
        err = httpServer.Shutdown(serverShutdownContext)
        if err != nil {
                logger.Fatal("http server shutdown failed", zap.Error(err))
        }

        wg.Wait()
}

1	package main
2
3	import (
4	"context"
5	"flag"
6	"fmt"
7	"log"
8	"net/http"
9	"os"
10	"os/signal"
11	"sync"
12	"syscall"
13
14	"github.com/didil/inhooks/pkg/lib"
15	"github.com/didil/inhooks/pkg/server"
16	"github.com/didil/inhooks/pkg/server/handlers"
17	"github.com/didil/inhooks/pkg/services"
18	"github.com/didil/inhooks/pkg/supervisor"
19	versionpkg "github.com/didil/inhooks/pkg/version"
20	"go.uber.org/zap"
21	)
22
23	var (
24	version = "dev"
25	)
26
27	func main() {	×
28	versionpkg.SetVersion(version)	×
29		×
30	// handle version command	×
31	isVersionCmd := flag.Bool("version", false, "print the version")	×
32	flag.Parse()	×
33	if *isVersionCmd {	×
34	fmt.Println(version)	×
35	os.Exit(0)	×
36	}	×
37
38	// start server
39	err := lib.LoadEnv()	×
40	if err != nil {	×
41	log.Fatalf("failed to load env: %v", err)	×
42	}	×
43
44	appConf, err := lib.InitAppConfig(context.Background())	×
45	if err != nil {	×
46	log.Fatalf("failed to process config: %v", err)	×
47	}	×
48
49	logger, err := lib.NewLogger(appConf)	×
50	if err != nil {	×
51	log.Fatalf("failed to initialize logger: %v", err)	×
52	}	×
53
54	logger.Info("starting Inhooks", zap.String("version", version))	×
55		×
56	inhooksConfigSvc := services.NewInhooksConfigService(logger, appConf)	×
57	logger.Info("loading inhooks config", zap.String("inhooksConfigFile", appConf.InhooksConfigFile))	×
58		×
59	err = inhooksConfigSvc.Load(appConf.InhooksConfigFile)	×
60	if err != nil {	×
61	logger.Fatal("failed to load inhooks config", zap.Error(err))	×
62	}	×
63
64	timeSvc := services.NewTimeService()	×
65		×
66	messageBuilder := services.NewMessageBuilder(timeSvc)	×
67		×
68	redisClient, err := lib.InitRedisClient(appConf)	×
69	if err != nil {	×
70	logger.Fatal("failed to init redis client", zap.Error(err))	×
71	}	×
72	redisStore, err := services.NewRedisStore(redisClient, appConf.Redis.InhooksDBName)	×
73	if err != nil {	×
74	logger.Fatal("failed to init redis store", zap.Error(err))	×
75	}	×
76
77	messageEnqueuer := services.NewMessageEnqueuer(redisStore, timeSvc)	×
78	messageFetcher := services.NewMessageFetcher(redisStore, timeSvc)	×
79	messageVerifier := services.NewMessageVerifier()	×
80	messageTransformer := services.NewMessageTransformer(&appConf.Transform)	×
81		×
82	app := handlers.NewApp(	×
83	handlers.WithLogger(logger),	×
84	handlers.WithInhooksConfigService(inhooksConfigSvc),	×
85	handlers.WithMessageBuilder(messageBuilder),	×
86	handlers.WithMessageEnqueuer(messageEnqueuer),	×
87	handlers.WithMessageVerifier(messageVerifier),	×
88	handlers.WithMessageTransformer(messageTransformer),	×
89	)	×
90		×
91	r := server.NewRouter(app)	×
92		×
93	addr := fmt.Sprintf("%s:%d", appConf.Server.Host, appConf.Server.Port)	×
94	httpServer := http.Server{	×
95	Addr: addr,	×
96	Handler: r,	×
97	}	×
98		×
99	wg := &sync.WaitGroup{}	×
100		×
101	wg.Add(1)	×
102	go func() {	×
103	logger.Info("listening ...", zap.String("addr", addr))	×
104	err = httpServer.ListenAndServe()	×
105	if err != nil && err != http.ErrServerClosed {	×
106	logger.Fatal("listener failure", zap.Error(err))	×
107	}	×
108	logger.Info("http server shutdown")	×
109	wg.Done()	×
110	}()
111
112	httpClient := lib.NewHttpClient(appConf)	×
113		×
114	messageProcessor := services.NewMessageProcessor(httpClient)	×
115	retryCalculator := services.NewRetryCalculator()	×
116	processingResultsSvc := services.NewProcessingResultsService(timeSvc, redisStore, retryCalculator)	×
117	schedulerSvc := services.NewSchedulerService(redisStore, timeSvc)	×
118		×
119	processingRecoverySvc, err := services.NewProcessingRecoveryService(redisStore)	×
120	if err != nil {	×
121	logger.Fatal("failed to init ProcessingRecoveryService", zap.Error(err))	×
122	}	×
123
124	cleanupSvc := services.NewCleanupService(redisStore, timeSvc)	×
125		×
126	queueMetricsSvc := services.NewQueueMetricsService(redisStore)	×
127		×
NEW 128	rateLimiter := services.NewTokenBucketLimiter(redisStore, timeSvc, logger)	×
NEW 129		×
130	svisor := supervisor.NewSupervisor(	×
131	supervisor.WithLogger(logger),	×
132	supervisor.WithMessageFetcher(messageFetcher),	×
133	supervisor.WithAppConfig(appConf),	×
134	supervisor.WithInhooksConfigService(inhooksConfigSvc),	×
135	supervisor.WithMessageProcessor(messageProcessor),	×
136	supervisor.WithProcessingResultsService(processingResultsSvc),	×
137	supervisor.WithSchedulerService(schedulerSvc),	×
138	supervisor.WithProcessingRecoveryService(processingRecoverySvc),	×
139	supervisor.WithCleanupService(cleanupSvc),	×
140	supervisor.WithMessageTransformer(messageTransformer),	×
141	supervisor.WithQueueMetricsService(queueMetricsSvc),	×
NEW 142	supervisor.WithRateLimiter(rateLimiter),	×
143	)	×
144		×
145	wg.Add(1)	×
146	go func() {	×
147	logger.Info("starting supervisor ...")	×
148	svisor.Start()	×
149	logger.Info("supervisor shutdown")	×
150	wg.Done()	×
151	}()	×
152
153	sigs := make(chan os.Signal, 1)	×
154	signal.Notify(sigs, os.Interrupt, syscall.SIGTERM)	×
155		×
156	sig := <-sigs	×
157	logger.Info("received shutdown signal, shutting down process", zap.String("signal", sig.String()))	×
158		×
159	svisor.Shutdown()	×
160		×
161	serverShutdownContext, cancel := context.WithTimeout(context.Background(), appConf.Server.ShutdownGracePeriod)	×
162	defer cancel()	×
163	err = httpServer.Shutdown(serverShutdownContext)	×
164	if err != nil {	×
165	logger.Fatal("http server shutdown failed", zap.Error(err))	×
166	}	×
167
168	wg.Wait()	×
169	}

didil / inhooks / 29169681577

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous