refactor: rename database tables for notification entities to improve clarity and consistency

feat: add Logbull remote logging via slog handler, tee to stdout
fix: wire retry into router, fix Docker config path, fix condition error handling, remove dead code
2026-07-02 16:04:18 +08:00 · 2026-06-27 15:17:59 +08:00 · 2026-06-27 13:32:33 +08:00 · 2026-06-27 13:29:33 +08:00 · 2026-06-27 13:28:27 +08:00 · 2026-06-27 13:26:13 +08:00
24 changed files with 1302 additions and 78 deletions
@@ -2,3 +2,4 @@
 bin/
 *.log
 .DS_Store
+server
@@ -0,0 +1,16 @@
+FROM golang:1.22-alpine AS builder
+
+WORKDIR /app
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+RUN CGO_ENABLED=0 GOOS=linux go build -o /server ./cmd/server
+
+FROM alpine:3.20
+
+RUN apk add --no-cache ca-certificates tzdata
+COPY --from=builder /server /usr/local/bin/server
+COPY config/config.yaml /config/config.yaml
+
+EXPOSE 8080
+ENTRYPOINT ["server"]
@@ -1,21 +1,248 @@
 package main

 import (
+	"context"
+	"fmt"
 	"log/slog"
+	"net/http"
 	"os"
+	"os/signal"
+	"syscall"
+	"time"

+	"aiaa-notification-service/internal/adapter"
+	"aiaa-notification-service/internal/cache"
 	"aiaa-notification-service/internal/config"
+	"aiaa-notification-service/internal/engine"
+	"aiaa-notification-service/internal/handler"
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+	"github.com/logbull/logbull-go/logbull"
 )

 func main() {
-	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
-	slog.SetDefault(logger)
-
+	// Load config first (need logbull settings for logger setup)
 	cfg, err := config.Load("config/config.yaml")
 	if err != nil {
 		slog.Error("failed to load config", "error", err)
 		os.Exit(1)
 	}

-	slog.Info("config loaded", "port", cfg.Server.Port)
+	// Setup dual logger: stdout JSON + Logbull remote
+	stdoutHandler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})
+
+	var logbullHandler slog.Handler
+	if cfg.Logbull.Host != "" {
+		lbCfg := logbull.Config{
+			Host:      cfg.Logbull.Host,
+			ProjectID: cfg.Logbull.ProjectID,
+			APIKey:    cfg.Logbull.APIKey,
+		}
+		switch cfg.Logbull.LogLevel {
+		case "DEBUG":
+			lbCfg.LogLevel = logbull.DEBUG
+		case "WARNING":
+			lbCfg.LogLevel = logbull.WARNING
+		case "ERROR":
+			lbCfg.LogLevel = logbull.ERROR
+		default:
+			lbCfg.LogLevel = logbull.INFO
+		}
+		h, err := logbull.NewSlogHandler(lbCfg)
+		if err != nil {
+			slog.Warn("logbull init failed, using stdout only", "error", err)
+		} else {
+			logbullHandler = h
+		}
+	}
+
+	var baseHandler slog.Handler
+	if logbullHandler != nil {
+		baseHandler = &teeHandler{handlers: []slog.Handler{stdoutHandler, logbullHandler}}
+	} else {
+		baseHandler = stdoutHandler
+	}
+
+	logger := slog.New(baseHandler)
+	slog.SetDefault(logger)
+
+	// Flush logbull on exit
+	defer func() {
+		if lbh, ok := logbullHandler.(interface{ Flush() }); ok {
+			lbh.Flush()
+		}
+		if lbh, ok := logbullHandler.(interface{ Shutdown() }); ok {
+			lbh.Shutdown()
+		}
+		time.Sleep(2 * time.Second) // allow async flush
+	}()
+
+	// Connect MySQL
+	st, err := store.NewStore(cfg.Database)
+	if err != nil {
+		slog.Error("failed to connect to MySQL", "error", err)
+		os.Exit(1)
+	}
+	defer st.Close()
+
+	// Connect Redis
+	var redisCache *cache.Cache
+	if cfg.Redis.Host != "" {
+		redisCache, err = cache.NewCache(cfg.Redis)
+		if err != nil {
+			slog.Warn("redis connection failed, running without cache", "error", err)
+			redisCache = nil
+		}
+	} else {
+		slog.Warn("redis not configured, running without cache")
+	}
+	if redisCache != nil {
+		defer redisCache.Close()
+	}
+
+	// Build engine
+	matcher := engine.NewMatcher(st, redisCache)
+	renderer := engine.NewRenderer()
+
+	// Build sender factory
+	senderFactory := func(channelType string) (adapter.ChannelSender, error) {
+		return adapter.NewSender(channelType, &cfg.SMTP)
+	}
+	router := engine.NewRouter(st, redisCache, senderFactory)
+
+	// Build handlers
+	notifyH := handler.NewNotifyHandler(st, redisCache, matcher, renderer, router)
+	sourceH := handler.NewSourceHandler(st, redisCache)
+	templateH := handler.NewTemplateHandler(st, redisCache)
+	channelH := handler.NewChannelHandler(st, redisCache)
+	ruleH := handler.NewRuleHandler(st, redisCache)
+	msgLogH := handler.NewMessageLogHandler(st)
+
+	// Setup Gin
+	gin.SetMode(gin.ReleaseMode)
+	r := gin.New()
+	r.Use(gin.Recovery())
+
+	// Health check
+	r.GET("/health", func(c *gin.Context) {
+		c.JSON(http.StatusOK, gin.H{"status": "ok"})
+	})
+
+	api := r.Group("/api/v1")
+
+	// Notify endpoint (source auth + rate limit)
+	notifyGroup := api.Group("/notify")
+	notifyGroup.Use(handler.SourceAuth(st))
+	if redisCache != nil {
+		notifyGroup.Use(handler.RateLimit(redisCache, cfg.RateLimit.Default))
+	}
+	notifyGroup.POST("", notifyH.Handle)
+
+	// Management endpoints (admin auth)
+	admin := api.Group("")
+	admin.Use(handler.AdminAuth(cfg.Server.AdminKey))
+
+	// Sources
+	admin.POST("/sources", sourceH.Create)
+	admin.GET("/sources", sourceH.List)
+	admin.GET("/sources/:id", sourceH.Get)
+	admin.PUT("/sources/:id", sourceH.Update)
+	admin.DELETE("/sources/:id", sourceH.Delete)
+
+	// Templates
+	admin.POST("/templates", templateH.Create)
+	admin.GET("/templates", templateH.List)
+	admin.GET("/templates/:id", templateH.Get)
+	admin.PUT("/templates/:id", templateH.Update)
+	admin.DELETE("/templates/:id", templateH.Delete)
+
+	// Channels
+	admin.POST("/channels", channelH.Create)
+	admin.GET("/channels", channelH.List)
+	admin.GET("/channels/:id", channelH.Get)
+	admin.PUT("/channels/:id", channelH.Update)
+	admin.DELETE("/channels/:id", channelH.Delete)
+
+	// Rules
+	admin.POST("/rules", ruleH.Create)
+	admin.GET("/rules", ruleH.List)
+	admin.GET("/rules/:id", ruleH.Get)
+	admin.PUT("/rules/:id", ruleH.Update)
+	admin.DELETE("/rules/:id", ruleH.Delete)
+	admin.PATCH("/rules/:id/enable", ruleH.Enable)
+	admin.PATCH("/rules/:id/disable", ruleH.Disable)
+	admin.PATCH("/rules/:id/channels/:channel_id/enable", ruleH.EnableChannel)
+	admin.PATCH("/rules/:id/channels/:channel_id/disable", ruleH.DisableChannel)
+
+	// Message Logs
+	admin.GET("/message-logs", msgLogH.List)
+
+	// Start server
+	srv := &http.Server{
+		Addr:    fmt.Sprintf(":%d", cfg.Server.Port),
+		Handler: r,
+	}
+
+	go func() {
+		slog.Info("server starting", "port", cfg.Server.Port)
+		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			slog.Error("server error", "error", err)
+			os.Exit(1)
+		}
+	}()
+
+	// Graceful shutdown
+	quit := make(chan os.Signal, 1)
+	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
+	<-quit
+	slog.Info("shutting down...")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := srv.Shutdown(ctx); err != nil {
+		slog.Error("forced shutdown", "error", err)
+	}
+	slog.Info("server stopped")
+}
+
+// teeHandler fans out log records to multiple handlers (stdout + remote).
+type teeHandler struct {
+	handlers []slog.Handler
+}
+
+func (t *teeHandler) Enabled(ctx context.Context, level slog.Level) bool {
+	for _, h := range t.handlers {
+		if h.Enabled(ctx, level) {
+			return true
+		}
+	}
+	return false
+}
+
+func (t *teeHandler) Handle(ctx context.Context, r slog.Record) error {
+	for _, h := range t.handlers {
+		if h.Enabled(ctx, r.Level) {
+			if err := h.Handle(ctx, r.Clone()); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (t *teeHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
+	handlers := make([]slog.Handler, len(t.handlers))
+	for i, h := range t.handlers {
+		handlers[i] = h.WithAttrs(attrs)
+	}
+	return &teeHandler{handlers: handlers}
+}
+
+func (t *teeHandler) WithGroup(name string) slog.Handler {
+	handlers := make([]slog.Handler, len(t.handlers))
+	for i, h := range t.handlers {
+		handlers[i] = h.WithGroup(name)
+	}
+	return &teeHandler{handlers: handlers}
 }
@@ -24,3 +24,9 @@ smtp:

 rate_limit:
  default: 100
+
+logbull:
+  host: "https://log.516886.xyz"
+  project_id: "42a3fef0-2fd6-4ce4-80c5-f6bb6ecc2013"
+  api_key: "lb_60701971723797ed0374aa3896078fe5"
+  log_level: "INFO"
@@ -0,0 +1,48 @@
+version: '3.8'
+
+services:
+  mysql:
+    image: mysql:8.0
+    environment:
+      MYSQL_ROOT_PASSWORD: rootpass
+      MYSQL_DATABASE: notification
+      MYSQL_USER: notify
+      MYSQL_PASSWORD: notify
+    ports:
+      - "3306:3306"
+    volumes:
+      - mysql_data:/var/lib/mysql
+    healthcheck:
+      test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+
+  redis:
+    image: redis:7-alpine
+    ports:
+      - "6379:6379"
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
+
+  api:
+    build: .
+    ports:
+      - "8080:8080"
+    environment:
+      - DB_PASSWORD=notify
+      - SMTP_PASSWORD=
+      - NOTIFY_DATABASE_HOST=mysql
+      - NOTIFY_REDIS_HOST=redis
+    depends_on:
+      mysql:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+    restart: unless-stopped
+
+volumes:
+  mysql_data:
@@ -3,26 +3,57 @@ module aiaa-notification-service
 go 1.26.2

 require (
+	github.com/gin-gonic/gin v1.12.0
 	github.com/go-sql-driver/mysql v1.10.0
 	github.com/jmoiron/sqlx v1.4.0
+	github.com/logbull/logbull-go v0.2.0
 	github.com/redis/go-redis/v9 v9.21.0
 	github.com/spf13/viper v1.21.0
 )

 require (
 	filippo.io/edwards25519 v1.2.0 // indirect
+	github.com/bytedance/gopkg v0.1.3 // indirect
+	github.com/bytedance/sonic v1.15.0 // indirect
+	github.com/bytedance/sonic/loader v0.5.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
+	github.com/gabriel-vasile/mimetype v1.4.12 // indirect
+	github.com/gin-contrib/sse v1.1.0 // indirect
+	github.com/go-playground/locales v0.14.1 // indirect
+	github.com/go-playground/universal-translator v0.18.1 // indirect
+	github.com/go-playground/validator/v10 v10.30.1 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
+	github.com/goccy/go-json v0.10.5 // indirect
+	github.com/goccy/go-yaml v1.19.2 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
+	github.com/leodido/go-urn v1.4.0 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
+	github.com/quic-go/qpack v0.6.0 // indirect
+	github.com/quic-go/quic-go v0.59.0 // indirect
 	github.com/sagikazarmark/locafero v0.11.0 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect
 	github.com/spf13/afero v1.15.0 // indirect
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
+	github.com/ugorji/go/codec v1.3.1 // indirect
+	go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.28.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/sys v0.30.0 // indirect
-	golang.org/x/text v0.28.0 // indirect
+	golang.org/x/arch v0.22.0 // indirect
+	golang.org/x/crypto v0.48.0 // indirect
+	golang.org/x/net v0.51.0 // indirect
+	golang.org/x/sys v0.41.0 // indirect
+	golang.org/x/text v0.34.0 // indirect
+	google.golang.org/protobuf v1.36.10 // indirect
 )
@@ -5,43 +5,90 @@ github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
 github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M=
+github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM=
+github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE=
+github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k=
+github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE=
+github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M=
+github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
 github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
+github.com/gabriel-vasile/mimetype v1.4.12 h1:e9hWvmLYvtp846tLHam2o++qitpguFiYCKbn0w9jyqw=
+github.com/gabriel-vasile/mimetype v1.4.12/go.mod h1:d+9Oxyo1wTzWdyVUPMmXFvp4F9tea18J8ufA774AB3s=
+github.com/gin-contrib/sse v1.1.0 h1:n0w2GMuUpWDVp7qSpvze6fAu9iRxJY4Hmj6AmBOU05w=
+github.com/gin-contrib/sse v1.1.0/go.mod h1:hxRZ5gVpWMT7Z0B0gSNYqqsSCNIJMjzvm6fqCz9vjwM=
+github.com/gin-gonic/gin v1.12.0 h1:b3YAbrZtnf8N//yjKeU2+MQsh2mY5htkZidOM7O0wG8=
+github.com/gin-gonic/gin v1.12.0/go.mod h1:VxccKfsSllpKshkBWgVgRniFFAzFb9csfngsqANjnLc=
+github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
+github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
+github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
+github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
+github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
+github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
+github.com/go-playground/validator/v10 v10.30.1 h1:f3zDSN/zOma+w6+1Wswgd9fLkdwy06ntQJp0BBvFG0w=
+github.com/go-playground/validator/v10 v10.30.1/go.mod h1:oSuBIQzuJxL//3MelwSLD5hc2Tu889bF0Idm9Dg26cM=
 github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
 github.com/go-sql-driver/mysql v1.10.0 h1:Q+1LV8DkHJvSYAdR83XzuhDaTykuDx0l6fkXxoWCWfw=
 github.com/go-sql-driver/mysql v1.10.0/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk=
 github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs=
 github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
-github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
-github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=
+github.com/goccy/go-json v0.10.5/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
+github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM=
+github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o=
 github.com/jmoiron/sqlx v1.4.0/go.mod h1:ZrZ7UsYB/weZdl2Bxg6jCRO9c3YHl8r3ahlKmRT4JLY=
-github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE=
-github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
+github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ=
+github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/logbull/logbull-go v0.2.0 h1:lzY7otH0cF7U4BqC8EJj7xipmzsCAIZS8Tmc1sHQ6rg=
+github.com/logbull/logbull-go v0.2.0/go.mod h1:6uoWbECvFyvg3DSGUd7y9Ufn8CisBaM//sDTLAjfaJI=
+github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
+github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU=
 github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8=
+github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII=
+github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw=
+github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
 github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E=
 github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
-github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
-github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
 github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw=
 github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U=
 github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I=
@@ -52,22 +99,57 @@ github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
 github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.21.0 h1:x5S+0EU27Lbphp4UKm1C+1oQO+rKx36vfCoaVebLFSU=
 github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjbTCAY=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
+github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
+github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
+github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY=
+github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4=
 github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
 github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
+go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE=
+go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
 go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
+go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo=
+go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q=
 go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
-golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
-golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng=
-golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU=
+golang.org/x/arch v0.22.0 h1:c/Zle32i5ttqRXjdLyyHZESLD/bB90DCU1g9l/0YBDI=
+golang.org/x/arch v0.22.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A=
+golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts=
+golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos=
+golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
+golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
+golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=
+golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA=
+google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
+google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
@@ -14,6 +14,7 @@ type Config struct {
 	Redis     RedisConfig     `mapstructure:"redis"`
 	SMTP      SMTPConfig      `mapstructure:"smtp"`
 	RateLimit RateLimitConfig `mapstructure:"rate_limit"`
+	Logbull   LogbullConfig   `mapstructure:"logbull"`
 }

 type ServerConfig struct {
@@ -57,6 +58,13 @@ type RateLimitConfig struct {
 	Default int `mapstructure:"default"`
 }

+type LogbullConfig struct {
+	Host      string `mapstructure:"host"`
+	ProjectID string `mapstructure:"project_id"`
+	APIKey    string `mapstructure:"api_key"`
+	LogLevel  string `mapstructure:"log_level"`
+}
+
 func Load(path string) (*Config, error) {
 	v := viper.New()
 	v.SetConfigFile(path)
@@ -9,16 +9,10 @@ import (
 	"aiaa-notification-service/internal/adapter"
 	"aiaa-notification-service/internal/cache"
 	"aiaa-notification-service/internal/model"
+	"aiaa-notification-service/internal/retry"
 	"aiaa-notification-service/internal/store"
 )

-type SendResult struct {
-	ChannelType string `json:"channel_type"`
-	ChannelID   int    `json:"channel_id"`
-	Success     bool   `json:"success"`
-	Error       string `json:"error,omitempty"`
-}
-
 type Router struct {
 	store         *store.Store
 	cache         *cache.Cache
@@ -51,8 +45,11 @@ func (r *Router) Route(ctx context.Context, rule *model.Rule, title, content str
 			if ch.Config != nil {
 				cfg = *ch.Config
 			}
-			if err := s.Send(title, content, cfg); err != nil {
-				slog.Error("send failed", "channel_type", ch.Type, "channel_id", ch.ID, "error", err)
+			r := retry.DefaultRetrier()
+			if err := r.Do(context.Background(), func() error {
+				return s.Send(title, content, cfg)
+			}); err != nil {
+				slog.Error("send failed after retries", "channel_type", ch.Type, "channel_id", ch.ID, "error", err)
 			} else {
 				slog.Info("sent", "channel_type", ch.Type, "channel_id", ch.ID)
 			}
@@ -0,0 +1,91 @@
+package handler
+
+import (
+	"encoding/json"
+	"net/http"
+	"strconv"
+
+	"aiaa-notification-service/internal/cache"
+	"aiaa-notification-service/internal/model"
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+)
+
+type ChannelHandler struct {
+	store *store.Store
+	cache *cache.Cache
+}
+
+func NewChannelHandler(s *store.Store, c *cache.Cache) *ChannelHandler {
+	return &ChannelHandler{store: s, cache: c}
+}
+
+type createChannelReq struct {
+	Name   string          `json:"name"   binding:"required"`
+	Type   string          `json:"type"   binding:"required"`
+	Config json.RawMessage `json:"config" binding:"required"`
+	Status int             `json:"status"`
+}
+
+func (h *ChannelHandler) Create(c *gin.Context) {
+	var req createChannelReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if req.Status == 0 {
+		req.Status = 1
+	}
+	raw := json.RawMessage(req.Config)
+	ch := &model.Channel{Name: req.Name, Type: req.Type, Config: &raw, Status: req.Status}
+	if err := h.store.CreateChannel(c.Request.Context(), ch); err != nil {
+		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusCreated, ch)
+}
+
+func (h *ChannelHandler) List(c *gin.Context) {
+	channels, err := h.store.ListChannels(c.Request.Context())
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, channels)
+}
+
+func (h *ChannelHandler) Get(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	ch, err := h.store.GetChannel(c.Request.Context(), id)
+	if err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "channel not found"})
+		return
+	}
+	c.JSON(http.StatusOK, ch)
+}
+
+func (h *ChannelHandler) Update(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	var req createChannelReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	raw := json.RawMessage(req.Config)
+	ch := &model.Channel{Name: req.Name, Type: req.Type, Config: &raw, Status: req.Status}
+	if err := h.store.UpdateChannel(c.Request.Context(), id, ch); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *ChannelHandler) Delete(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	if err := h.store.DeleteChannel(c.Request.Context(), id); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
@@ -0,0 +1,37 @@
+package handler
+
+import (
+	"net/http"
+
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+)
+
+type MessageLogHandler struct {
+	store *store.Store
+}
+
+func NewMessageLogHandler(s *store.Store) *MessageLogHandler {
+	return &MessageLogHandler{store: s}
+}
+
+func (h *MessageLogHandler) List(c *gin.Context) {
+	var filter store.MessageLogFilter
+	if err := c.ShouldBindQuery(&filter); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	logs, total, err := h.store.ListMessageLogs(c.Request.Context(), filter)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusOK, gin.H{
+		"data":  logs,
+		"total": total,
+		"page":  filter.Page,
+	})
+}
@@ -0,0 +1,81 @@
+package handler
+
+import (
+	"net/http"
+	"strings"
+
+	"aiaa-notification-service/internal/cache"
+	"aiaa-notification-service/internal/model"
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+)
+
+// AdminAuth checks the admin key for management API endpoints.
+func AdminAuth(adminKey string) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		key := extractBearer(c)
+		if key == "" || key != adminKey {
+			c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
+			return
+		}
+		c.Next()
+	}
+}
+
+// SourceAuth identifies the source by its API key and sets it in context.
+func SourceAuth(s *store.Store) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		key := extractBearer(c)
+		if key == "" {
+			c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing api key"})
+			return
+		}
+		source, err := s.GetSourceByAPIKey(c.Request.Context(), key)
+		if err != nil {
+			c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid api key"})
+			return
+		}
+		c.Set("source", source)
+		c.Next()
+	}
+}
+
+// RateLimit applies per-source rate limiting.
+func RateLimit(cache *cache.Cache, defaultLimit int) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		source, exists := c.Get("source")
+		if !exists {
+			c.Next()
+			return
+		}
+		src := source.(*model.Source)
+		allowed, retryAfter, err := cache.CheckRateLimit(c.Request.Context(), src.ID, defaultLimit)
+		if err != nil {
+			// Redis error — allow pass through
+			c.Next()
+			return
+		}
+		if !allowed {
+			c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{
+				"error":       "rate_limit_exceeded",
+				"message":     "too many requests",
+				"retry_after": retryAfter,
+			})
+			return
+		}
+		c.Next()
+	}
+}
+
+func extractBearer(c *gin.Context) string {
+	auth := c.GetHeader("Authorization")
+	if auth == "" {
+		return ""
+	}
+	parts := strings.SplitN(auth, " ", 2)
+	if len(parts) != 2 || !strings.EqualFold(parts[0], "bearer") {
+		return ""
+	}
+	return parts[1]
+}
@@ -0,0 +1,145 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"aiaa-notification-service/internal/cache"
+	"aiaa-notification-service/internal/condition"
+	"aiaa-notification-service/internal/engine"
+	"aiaa-notification-service/internal/model"
+	"aiaa-notification-service/internal/parser"
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+)
+
+type NotifyHandler struct {
+	store    *store.Store
+	cache    *cache.Cache // used by engine
+	matcher  *engine.Matcher
+	renderer *engine.Renderer
+	router   *engine.Router
+}
+
+func NewNotifyHandler(s *store.Store, c *cache.Cache, m *engine.Matcher, r *engine.Renderer, rt *engine.Router) *NotifyHandler {
+	return &NotifyHandler{store: s, cache: c, matcher: m, renderer: r, router: rt}
+}
+
+func (h *NotifyHandler) Handle(c *gin.Context) {
+	src := c.MustGet("source").(*model.Source)
+
+	// 1. Read raw body
+	body, err := io.ReadAll(c.Request.Body)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "read body failed"})
+		return
+	}
+
+	// 2. Parse message
+	p, err := parser.NewParser(src.ParseMode, src.ParsePattern)
+	if err != nil {
+		c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "parser setup: " + err.Error()})
+		return
+	}
+	msg, err := p.Parse(body)
+	if err != nil {
+		c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "parse failed: " + err.Error()})
+		return
+	}
+
+	// 3. Match rule
+	rule, err := h.matcher.Match(c.Request.Context(), src.ID, msg.Event)
+	if err != nil {
+		// No matching rule → 200 with matched: false
+		c.JSON(http.StatusOK, gin.H{"matched": false})
+		return
+	}
+
+	// 4. Evaluate conditions
+	if rule.Conditions != nil {
+		var conds []model.Condition
+		if err := json.Unmarshal(*rule.Conditions, &conds); err != nil {
+			slog.Error("failed to unmarshal rule conditions", "rule_id", rule.ID, "error", err)
+			c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "invalid rule conditions"})
+			return
+		}
+		if !condition.Evaluate(conds, msg.Data) {
+			c.JSON(http.StatusOK, gin.H{
+				"matched":  true,
+				"filtered": true,
+				"reason":   "condition not met",
+			})
+			return
+		}
+	}
+
+	// 5. Get template content
+	tmpl, err := h.store.GetTemplate(c.Request.Context(), rule.TemplateID)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "template not found"})
+		return
+	}
+
+	// 6. Render template
+	content, err := h.renderer.Render(tmpl.Content, msg.Data)
+	if err != nil {
+		c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "template render failed: " + err.Error()})
+		return
+	}
+
+	// 7. Route to channels
+	title := src.Name + ": " + msg.Event
+	channels := h.router.Route(c.Request.Context(), rule, title, content)
+
+	// 8. Log message (best effort, detached context)
+	go func() {
+		payloadJSON, _ := json.Marshal(msg.Data)
+		ctx := context.Background()
+		for _, chName := range channels {
+			chID := parseChannelID(chName)
+			ml := &model.MessageLog{
+				RuleID:    rule.ID,
+				ChannelID: chID,
+				Source:    src.Name,
+				Event:     msg.Event,
+				Payload:   payloadJSON,
+				Content:   content,
+				Status:    "pending",
+			}
+			if err := h.store.CreateMessageLog(ctx, ml); err != nil {
+				slog.Warn("failed to create message log", "error", err)
+			}
+		}
+	}()
+
+	slog.Info("notification accepted",
+		"source", src.Name,
+		"event", msg.Event,
+		"channels", channels,
+	)
+
+	c.JSON(http.StatusOK, gin.H{
+		"matched":  true,
+		"channels": channels,
+		"accepted": true,
+	})
+}
+
+// parseChannelID extracts the numeric channel ID from a channel name formatted as "type:id".
+func parseChannelID(chName string) int {
+	idx := strings.LastIndex(chName, ":")
+	if idx < 0 || idx == len(chName)-1 {
+		return 0
+	}
+	id, err := strconv.Atoi(chName[idx+1:])
+	if err != nil {
+		return 0
+	}
+	return id
+}
@@ -0,0 +1,212 @@
+package handler
+
+import (
+	"encoding/json"
+	"net/http"
+	"strconv"
+
+	"aiaa-notification-service/internal/cache"
+	"aiaa-notification-service/internal/model"
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+)
+
+type RuleHandler struct {
+	store *store.Store
+	cache *cache.Cache
+}
+
+func NewRuleHandler(s *store.Store, c *cache.Cache) *RuleHandler {
+	return &RuleHandler{store: s, cache: c}
+}
+
+type createRuleReq struct {
+	SourceName   string            `json:"source_name"   binding:"required"`
+	Event        string            `json:"event"         binding:"required"`
+	TemplateName string            `json:"template_name" binding:"required"`
+	Channels     []string          `json:"channels"`
+	Conditions   []model.Condition `json:"conditions,omitempty"`
+	Enabled      int               `json:"enabled"`
+}
+
+func (h *RuleHandler) Create(c *gin.Context) {
+	var req createRuleReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if req.Enabled == 0 {
+		req.Enabled = 1
+	}
+
+	// Resolve names -> IDs
+	src, err := h.store.GetSourceByName(c.Request.Context(), req.SourceName)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "source not found: " + req.SourceName})
+		return
+	}
+	tmpl, err := h.store.GetTemplateByName(c.Request.Context(), req.TemplateName)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "template not found: " + req.TemplateName})
+		return
+	}
+	channelIDs, err := resolveChannelNames(h.store, c, req.Channels)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	var condsJSON *json.RawMessage
+	if len(req.Conditions) > 0 {
+		data, _ := json.Marshal(req.Conditions)
+		raw := json.RawMessage(data)
+		condsJSON = &raw
+	}
+
+	rule := &model.Rule{
+		SourceID:   src.ID,
+		Event:      req.Event,
+		TemplateID: tmpl.ID,
+		Conditions: condsJSON,
+		Enabled:    req.Enabled,
+	}
+
+	if err := h.store.CreateRule(c.Request.Context(), rule, channelIDs); err != nil {
+		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
+		return
+	}
+
+	c.JSON(http.StatusCreated, rule)
+}
+
+func (h *RuleHandler) List(c *gin.Context) {
+	rules, err := h.store.ListRules(c.Request.Context())
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, rules)
+}
+
+func (h *RuleHandler) Get(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	rule, err := h.store.GetRule(c.Request.Context(), id)
+	if err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "rule not found"})
+		return
+	}
+	c.JSON(http.StatusOK, rule)
+}
+
+func (h *RuleHandler) Update(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	var req createRuleReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	src, err := h.store.GetSourceByName(c.Request.Context(), req.SourceName)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "source not found"})
+		return
+	}
+	tmpl, err := h.store.GetTemplateByName(c.Request.Context(), req.TemplateName)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "template not found"})
+		return
+	}
+	channelIDs, err := resolveChannelNames(h.store, c, req.Channels)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	var condsJSON *json.RawMessage
+	if len(req.Conditions) > 0 {
+		data, _ := json.Marshal(req.Conditions)
+		raw := json.RawMessage(data)
+		condsJSON = &raw
+	}
+
+	rule := &model.Rule{
+		SourceID:   src.ID,
+		Event:      req.Event,
+		TemplateID: tmpl.ID,
+		Conditions: condsJSON,
+		Enabled:    req.Enabled,
+	}
+
+	if err := h.store.UpdateRule(c.Request.Context(), id, rule, channelIDs); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Invalidate cache
+	if h.cache != nil {
+		_ = h.cache.InvalidateRule(c.Request.Context(), src.ID, req.Event)
+		_ = h.cache.InvalidateChannels(c.Request.Context(), id)
+	}
+
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *RuleHandler) Delete(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	if err := h.store.DeleteRule(c.Request.Context(), id); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *RuleHandler) Enable(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	if err := h.store.SetRuleEnabled(c.Request.Context(), id, true); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *RuleHandler) Disable(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	if err := h.store.SetRuleEnabled(c.Request.Context(), id, false); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *RuleHandler) EnableChannel(c *gin.Context) {
+	ruleID, _ := strconv.Atoi(c.Param("id"))
+	channelID, _ := strconv.Atoi(c.Param("channel_id"))
+	if err := h.store.SetRuleChannelEnabled(c.Request.Context(), ruleID, channelID, true); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *RuleHandler) DisableChannel(c *gin.Context) {
+	ruleID, _ := strconv.Atoi(c.Param("id"))
+	channelID, _ := strconv.Atoi(c.Param("channel_id"))
+	if err := h.store.SetRuleChannelEnabled(c.Request.Context(), ruleID, channelID, false); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func resolveChannelNames(s *store.Store, c *gin.Context, names []string) ([]int, error) {
+	var ids []int
+	for _, name := range names {
+		ch, err := s.GetChannelByName(c.Request.Context(), name)
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, ch.ID)
+	}
+	return ids, nil
+}
@@ -0,0 +1,109 @@
+package handler
+
+import (
+	"net/http"
+	"strconv"
+
+	"aiaa-notification-service/internal/cache"
+	"aiaa-notification-service/internal/model"
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+)
+
+type SourceHandler struct {
+	store *store.Store
+	cache *cache.Cache
+}
+
+func NewSourceHandler(s *store.Store, c *cache.Cache) *SourceHandler {
+	return &SourceHandler{store: s, cache: c}
+}
+
+type createSourceReq struct {
+	Name         string `json:"name"          binding:"required"`
+	ParseMode    string `json:"parse_mode"`
+	ParsePattern string `json:"parse_pattern"`
+	Status       int    `json:"status"`
+}
+
+func (h *SourceHandler) Create(c *gin.Context) {
+	var req createSourceReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	if req.ParseMode == "" {
+		req.ParseMode = "json"
+	}
+	if req.Status == 0 {
+		req.Status = 1
+	}
+
+	src := &model.Source{
+		Name:         req.Name,
+		ParseMode:    req.ParseMode,
+		ParsePattern: req.ParsePattern,
+		Status:       req.Status,
+	}
+	if err := h.store.CreateSource(c.Request.Context(), src); err != nil {
+		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusCreated, src)
+}
+
+func (h *SourceHandler) List(c *gin.Context) {
+	sources, err := h.store.ListSources(c.Request.Context())
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, sources)
+}
+
+func (h *SourceHandler) Get(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	src, err := h.store.GetSource(c.Request.Context(), id)
+	if err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "source not found"})
+		return
+	}
+	c.JSON(http.StatusOK, src)
+}
+
+func (h *SourceHandler) Update(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	var req createSourceReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+
+	src := &model.Source{
+		Name:         req.Name,
+		ParseMode:    req.ParseMode,
+		ParsePattern: req.ParsePattern,
+		Status:       req.Status,
+	}
+	if err := h.store.UpdateSource(c.Request.Context(), id, src); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	// Invalidate cache
+	if h.cache != nil {
+		_ = h.cache.InvalidateBySource(c.Request.Context(), id)
+	}
+
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *SourceHandler) Delete(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	if err := h.store.DeleteSource(c.Request.Context(), id); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
@@ -0,0 +1,88 @@
+package handler
+
+import (
+	"net/http"
+	"strconv"
+
+	"aiaa-notification-service/internal/cache"
+	"aiaa-notification-service/internal/model"
+	"aiaa-notification-service/internal/store"
+
+	"github.com/gin-gonic/gin"
+)
+
+type TemplateHandler struct {
+	store *store.Store
+	cache *cache.Cache
+}
+
+func NewTemplateHandler(s *store.Store, c *cache.Cache) *TemplateHandler {
+	return &TemplateHandler{store: s, cache: c}
+}
+
+type createTemplateReq struct {
+	Name    string `json:"name"    binding:"required"`
+	Content string `json:"content" binding:"required"`
+}
+
+func (h *TemplateHandler) Create(c *gin.Context) {
+	var req createTemplateReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	tmpl := &model.Template{Name: req.Name, Content: req.Content}
+	if err := h.store.CreateTemplate(c.Request.Context(), tmpl); err != nil {
+		c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusCreated, tmpl)
+}
+
+func (h *TemplateHandler) List(c *gin.Context) {
+	templates, err := h.store.ListTemplates(c.Request.Context())
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, templates)
+}
+
+func (h *TemplateHandler) Get(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	tmpl, err := h.store.GetTemplate(c.Request.Context(), id)
+	if err != nil {
+		c.JSON(http.StatusNotFound, gin.H{"error": "template not found"})
+		return
+	}
+	c.JSON(http.StatusOK, tmpl)
+}
+
+func (h *TemplateHandler) Update(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	var req createTemplateReq
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	tmpl := &model.Template{Name: req.Name, Content: req.Content}
+	if err := h.store.UpdateTemplate(c.Request.Context(), id, tmpl); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+
+	if h.cache != nil {
+		_ = h.cache.InvalidateByTemplate(c.Request.Context(), id)
+	}
+
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *TemplateHandler) Delete(c *gin.Context) {
+	id, _ := strconv.Atoi(c.Param("id"))
+	if err := h.store.DeleteTemplate(c.Request.Context(), id); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
@@ -0,0 +1,45 @@
+package retry
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+)
+
+type Retrier struct {
+	maxRetries int
+	backoff    []time.Duration
+}
+
+func NewRetrier(maxRetries int, backoff []time.Duration) *Retrier {
+	return &Retrier{maxRetries: maxRetries, backoff: backoff}
+}
+
+// DefaultRetrier returns a retrier with 3 attempts, exponential backoff: 1s, 5s, 30s.
+func DefaultRetrier() *Retrier {
+	return NewRetrier(3, []time.Duration{1 * time.Second, 5 * time.Second, 30 * time.Second})
+}
+
+func (r *Retrier) Do(ctx context.Context, fn func() error) error {
+	var lastErr error
+	for attempt := 0; attempt <= r.maxRetries; attempt++ {
+		if attempt > 0 {
+			delay := r.backoff[attempt-1]
+			slog.Info("retrying", "attempt", attempt, "delay", delay)
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-time.After(delay):
+			}
+		}
+
+		err := fn()
+		if err == nil {
+			return nil
+		}
+		lastErr = err
+		slog.Warn("attempt failed", "attempt", attempt, "error", err)
+	}
+	return fmt.Errorf("all %d attempts failed, last error: %w", r.maxRetries+1, lastErr)
+}
@@ -13,7 +13,7 @@ func (s *Store) CreateChannel(ctx context.Context, ch *model.Channel) error {
 	if err != nil {
 		return fmt.Errorf("marshal channel config: %w", err)
 	}
-	query := `INSERT INTO channel (name, type, config, status) VALUES (?, ?, ?, ?)`
+	query := `INSERT INTO notification_channel (name, type, config, status) VALUES (?, ?, ?, ?)`
 	result, err := s.DB.ExecContext(ctx, query, ch.Name, ch.Type, configJSON, ch.Status)
 	if err != nil {
 		return fmt.Errorf("create channel: %w", err)
@@ -26,7 +26,7 @@ func (s *Store) CreateChannel(ctx context.Context, ch *model.Channel) error {
 func (s *Store) GetChannel(ctx context.Context, id int) (*model.Channel, error) {
 	var ch model.Channel
 	var configBytes []byte
-	row := s.DB.QueryRowContext(ctx, `SELECT id, name, type, config, status, created_at, updated_at FROM channel WHERE id = ?`, id)
+	row := s.DB.QueryRowContext(ctx, `SELECT id, name, type, config, status, created_at, updated_at FROM notification_channel WHERE id = ?`, id)
 	if err := row.Scan(&ch.ID, &ch.Name, &ch.Type, &configBytes, &ch.Status, &ch.CreatedAt, &ch.UpdatedAt); err != nil {
 		return nil, fmt.Errorf("get channel %d: %w", id, err)
 	}
@@ -38,7 +38,7 @@ func (s *Store) GetChannel(ctx context.Context, id int) (*model.Channel, error)
 func (s *Store) GetChannelByName(ctx context.Context, name string) (*model.Channel, error) {
 	var ch model.Channel
 	var configBytes []byte
-	row := s.DB.QueryRowContext(ctx, `SELECT id, name, type, config, status, created_at, updated_at FROM channel WHERE name = ?`, name)
+	row := s.DB.QueryRowContext(ctx, `SELECT id, name, type, config, status, created_at, updated_at FROM notification_channel WHERE name = ?`, name)
 	if err := row.Scan(&ch.ID, &ch.Name, &ch.Type, &configBytes, &ch.Status, &ch.CreatedAt, &ch.UpdatedAt); err != nil {
 		return nil, fmt.Errorf("get channel by name %s: %w", name, err)
 	}
@@ -48,7 +48,7 @@ func (s *Store) GetChannelByName(ctx context.Context, name string) (*model.Chann
 }

 func (s *Store) ListChannels(ctx context.Context) ([]model.Channel, error) {
-	rows, err := s.DB.QueryContext(ctx, `SELECT id, name, type, config, status, created_at, updated_at FROM channel ORDER BY id`)
+	rows, err := s.DB.QueryContext(ctx, `SELECT id, name, type, config, status, created_at, updated_at FROM notification_channel ORDER BY id`)
 	if err != nil {
 		return nil, fmt.Errorf("list channels: %w", err)
 	}
@@ -73,7 +73,7 @@ func (s *Store) UpdateChannel(ctx context.Context, id int, ch *model.Channel) er
 	if err != nil {
 		return fmt.Errorf("marshal channel config: %w", err)
 	}
-	query := `UPDATE channel SET name=?, type=?, config=?, status=? WHERE id=?`
+	query := `UPDATE notification_channel SET name=?, type=?, config=?, status=? WHERE id=?`
 	_, err = s.DB.ExecContext(ctx, query, ch.Name, ch.Type, configJSON, ch.Status, id)
 	if err != nil {
 		return fmt.Errorf("update channel %d: %w", id, err)
@@ -82,7 +82,7 @@ func (s *Store) UpdateChannel(ctx context.Context, id int, ch *model.Channel) er
 }

 func (s *Store) DeleteChannel(ctx context.Context, id int) error {
-	_, err := s.DB.ExecContext(ctx, `DELETE FROM channel WHERE id = ?`, id)
+	_, err := s.DB.ExecContext(ctx, `DELETE FROM notification_channel WHERE id = ?`, id)
 	if err != nil {
 		return fmt.Errorf("delete channel %d: %w", id, err)
 	}
@@ -16,7 +16,7 @@ type MessageLogFilter struct {
 }

 func (s *Store) CreateMessageLog(ctx context.Context, ml *model.MessageLog) error {
-	query := `INSERT INTO message_log (rule_id, channel_id, source, event, payload, content, status, retry_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+	query := `INSERT INTO notification_message_log (rule_id, channel_id, source, event, payload, content, status, retry_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
 	result, err := s.DB.ExecContext(ctx, query, ml.RuleID, ml.ChannelID, ml.Source, ml.Event, ml.Payload, ml.Content, ml.Status, ml.RetryCount)
 	if err != nil {
 		return fmt.Errorf("create message_log: %w", err)
@@ -27,7 +27,7 @@ func (s *Store) CreateMessageLog(ctx context.Context, ml *model.MessageLog) erro
 }

 func (s *Store) UpdateMessageLog(ctx context.Context, id int64, status string, response, errMsg *string) error {
-	query := `UPDATE message_log SET status=?, response=?, error_msg=? WHERE id=?`
+	query := `UPDATE notification_message_log SET status=?, response=?, error_msg=? WHERE id=?`
 	_, err := s.DB.ExecContext(ctx, query, status, response, errMsg, id)
 	return err
 }
@@ -49,7 +49,7 @@ func (s *Store) ListMessageLogs(ctx context.Context, filter MessageLogFilter) ([
 	}

 	var count int
-	countQuery := "SELECT COUNT(*) FROM message_log " + where
+	countQuery := "SELECT COUNT(*) FROM notification_message_log " + where
 	if err := s.DB.GetContext(ctx, &count, countQuery, args...); err != nil {
 		return nil, 0, err
 	}
@@ -63,7 +63,7 @@ func (s *Store) ListMessageLogs(ctx context.Context, filter MessageLogFilter) ([
 	offset := (filter.Page - 1) * filter.PageSize

 	var logs []model.MessageLog
-	query := "SELECT * FROM message_log " + where + " ORDER BY id DESC LIMIT ? OFFSET ?"
+	query := "SELECT * FROM notification_message_log " + where + " ORDER BY id DESC LIMIT ? OFFSET ?"
 	args = append(args, filter.PageSize, offset)
 	if err := s.DB.SelectContext(ctx, &logs, query, args...); err != nil {
 		return nil, 0, err
@@ -16,7 +16,7 @@ func (s *Store) CreateRule(ctx context.Context, r *model.Rule, channelIDs []int)
 	}
 	defer tx.Rollback()

-	query := `INSERT INTO rule (source_id, event, template_id, conditions, enabled) VALUES (?, ?, ?, ?, ?)`
+	query := `INSERT INTO notification_rule (source_id, event, template_id, conditions, enabled) VALUES (?, ?, ?, ?, ?)`
 	condsJSON, err := marshalJSON(r.Conditions)
 	if err != nil {
 		return fmt.Errorf("marshal conditions: %w", err)
@@ -29,7 +29,7 @@ func (s *Store) CreateRule(ctx context.Context, r *model.Rule, channelIDs []int)
 	r.ID = int(id)

 	for _, chID := range channelIDs {
-		_, err := tx.ExecContext(ctx, `INSERT INTO rule_channel (rule_id, channel_id, enabled) VALUES (?, ?, 1)`, r.ID, chID)
+		_, err := tx.ExecContext(ctx, `INSERT INTO notification_rule_channel (rule_id, channel_id, enabled) VALUES (?, ?, 1)`, r.ID, chID)
 		if err != nil {
 			return fmt.Errorf("add rule_channel: %w", err)
 		}
@@ -40,7 +40,7 @@ func (s *Store) CreateRule(ctx context.Context, r *model.Rule, channelIDs []int)
 func (s *Store) GetRule(ctx context.Context, id int) (*model.Rule, error) {
 	var r model.Rule
 	var condsBytes []byte
-	row := s.DB.QueryRowContext(ctx, `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM rule WHERE id = ?`, id)
+	row := s.DB.QueryRowContext(ctx, `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM notification_rule WHERE id = ?`, id)
 	if err := row.Scan(&r.ID, &r.SourceID, &r.Event, &r.TemplateID, &condsBytes, &r.Enabled, &r.CreatedAt, &r.UpdatedAt); err != nil {
 		return nil, fmt.Errorf("get rule %d: %w", id, err)
 	}
@@ -54,7 +54,7 @@ func (s *Store) GetRule(ctx context.Context, id int) (*model.Rule, error) {
 func (s *Store) GetRuleBySourceEvent(ctx context.Context, sourceID int, event string) (*model.Rule, error) {
 	var r model.Rule
 	var condsBytes []byte
-	query := `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM rule WHERE source_id = ? AND event = ? AND enabled = 1`
+	query := `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM notification_rule WHERE source_id = ? AND event = ? AND enabled = 1`
 	row := s.DB.QueryRowContext(ctx, query, sourceID, event)
 	if err := row.Scan(&r.ID, &r.SourceID, &r.Event, &r.TemplateID, &condsBytes, &r.Enabled, &r.CreatedAt, &r.UpdatedAt); err != nil {
 		return nil, fmt.Errorf("get rule by source+event: %w", err)
@@ -67,7 +67,7 @@ func (s *Store) GetRuleBySourceEvent(ctx context.Context, sourceID int, event st
 }

 func (s *Store) ListRules(ctx context.Context) ([]model.Rule, error) {
-	rows, err := s.DB.QueryContext(ctx, `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM rule ORDER BY id`)
+	rows, err := s.DB.QueryContext(ctx, `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM notification_rule ORDER BY id`)
 	if err != nil {
 		return nil, fmt.Errorf("list rules: %w", err)
 	}
@@ -86,18 +86,18 @@ func (s *Store) UpdateRule(ctx context.Context, id int, r *model.Rule, channelID
 	if err != nil {
 		return fmt.Errorf("marshal conditions: %w", err)
 	}
-	_, err = tx.ExecContext(ctx, `UPDATE rule SET source_id=?, event=?, template_id=?, conditions=?, enabled=? WHERE id=?`,
+	_, err = tx.ExecContext(ctx, `UPDATE notification_rule SET source_id=?, event=?, template_id=?, conditions=?, enabled=? WHERE id=?`,
 		r.SourceID, r.Event, r.TemplateID, condsJSON, r.Enabled, id)
 	if err != nil {
 		return fmt.Errorf("update rule: %w", err)
 	}

 	if channelIDs != nil {
-		if _, err := tx.ExecContext(ctx, `DELETE FROM rule_channel WHERE rule_id = ?`, id); err != nil {
+		if _, err := tx.ExecContext(ctx, `DELETE FROM notification_rule_channel WHERE rule_id = ?`, id); err != nil {
 			return fmt.Errorf("delete rule channels: %w", err)
 		}
 		for _, chID := range channelIDs {
-			_, err := tx.ExecContext(ctx, `INSERT INTO rule_channel (rule_id, channel_id, enabled) VALUES (?, ?, 1)`, id, chID)
+			_, err := tx.ExecContext(ctx, `INSERT INTO notification_rule_channel (rule_id, channel_id, enabled) VALUES (?, ?, 1)`, id, chID)
 			if err != nil {
 				return fmt.Errorf("add rule_channel: %w", err)
 			}
@@ -107,7 +107,7 @@ func (s *Store) UpdateRule(ctx context.Context, id int, r *model.Rule, channelID
 }

 func (s *Store) DeleteRule(ctx context.Context, id int) error {
-	_, err := s.DB.ExecContext(ctx, `DELETE FROM rule WHERE id = ?`, id)
+	_, err := s.DB.ExecContext(ctx, `DELETE FROM notification_rule WHERE id = ?`, id)
 	if err != nil {
 		return fmt.Errorf("delete rule %d: %w", id, err)
 	}
@@ -119,7 +119,7 @@ func (s *Store) SetRuleEnabled(ctx context.Context, id int, enabled bool) error
 	if enabled {
 		v = 1
 	}
-	_, err := s.DB.ExecContext(ctx, `UPDATE rule SET enabled = ? WHERE id = ?`, v, id)
+	_, err := s.DB.ExecContext(ctx, `UPDATE notification_rule SET enabled = ? WHERE id = ?`, v, id)
 	if err != nil {
 		return fmt.Errorf("set rule enabled %d: %w", id, err)
 	}
@@ -128,7 +128,7 @@ func (s *Store) SetRuleEnabled(ctx context.Context, id int, enabled bool) error

 func (s *Store) GetRuleChannels(ctx context.Context, ruleID int) ([]model.RuleChannel, error) {
 	var rcs []model.RuleChannel
-	err := s.DB.SelectContext(ctx, &rcs, `SELECT id, rule_id, channel_id, enabled FROM rule_channel WHERE rule_id = ? AND enabled = 1`, ruleID)
+	err := s.DB.SelectContext(ctx, &rcs, `SELECT id, rule_id, channel_id, enabled FROM notification_rule_channel WHERE rule_id = ? AND enabled = 1`, ruleID)
 	if err != nil {
 		return nil, fmt.Errorf("get rule channels: %w", err)
 	}
@@ -140,7 +140,7 @@ func (s *Store) SetRuleChannelEnabled(ctx context.Context, ruleID, channelID int
 	if enabled {
 		v = 1
 	}
-	_, err := s.DB.ExecContext(ctx, `UPDATE rule_channel SET enabled = ? WHERE rule_id = ? AND channel_id = ?`, v, ruleID, channelID)
+	_, err := s.DB.ExecContext(ctx, `UPDATE notification_rule_channel SET enabled = ? WHERE rule_id = ? AND channel_id = ?`, v, ruleID, channelID)
 	if err != nil {
 		return fmt.Errorf("set rule channel enabled %d/%d: %w", ruleID, channelID, err)
 	}
@@ -17,7 +17,7 @@ func generateAPIKey() string {

 func (s *Store) CreateSource(ctx context.Context, src *model.Source) error {
 	src.APIKey = generateAPIKey()
-	query := `INSERT INTO source (name, api_key, parse_mode, parse_pattern, status) VALUES (?, ?, ?, ?, ?)`
+	query := `INSERT INTO notification_source (name, api_key, parse_mode, parse_pattern, status) VALUES (?, ?, ?, ?, ?)`
 	result, err := s.DB.ExecContext(ctx, query, src.Name, src.APIKey, src.ParseMode, src.ParsePattern, src.Status)
 	if err != nil {
 		return fmt.Errorf("create source: %w", err)
@@ -29,7 +29,7 @@ func (s *Store) CreateSource(ctx context.Context, src *model.Source) error {

 func (s *Store) GetSource(ctx context.Context, id int) (*model.Source, error) {
 	var src model.Source
-	err := s.DB.GetContext(ctx, &src, `SELECT * FROM source WHERE id = ?`, id)
+	err := s.DB.GetContext(ctx, &src, `SELECT * FROM notification_source WHERE id = ?`, id)
 	if err != nil {
 		return nil, fmt.Errorf("get source %d: %w", id, err)
 	}
@@ -38,7 +38,7 @@ func (s *Store) GetSource(ctx context.Context, id int) (*model.Source, error) {

 func (s *Store) GetSourceByAPIKey(ctx context.Context, apiKey string) (*model.Source, error) {
 	var src model.Source
-	err := s.DB.GetContext(ctx, &src, `SELECT * FROM source WHERE api_key = ? AND status = 1`, apiKey)
+	err := s.DB.GetContext(ctx, &src, `SELECT * FROM notification_source WHERE api_key = ? AND status = 1`, apiKey)
 	if err != nil {
 		return nil, fmt.Errorf("get source by api_key: %w", err)
 	}
@@ -47,7 +47,7 @@ func (s *Store) GetSourceByAPIKey(ctx context.Context, apiKey string) (*model.So

 func (s *Store) GetSourceByName(ctx context.Context, name string) (*model.Source, error) {
 	var src model.Source
-	err := s.DB.GetContext(ctx, &src, `SELECT * FROM source WHERE name = ?`, name)
+	err := s.DB.GetContext(ctx, &src, `SELECT * FROM notification_source WHERE name = ?`, name)
 	if err != nil {
 		return nil, fmt.Errorf("get source by name %s: %w", name, err)
 	}
@@ -56,7 +56,7 @@ func (s *Store) GetSourceByName(ctx context.Context, name string) (*model.Source

 func (s *Store) ListSources(ctx context.Context) ([]model.Source, error) {
 	var sources []model.Source
-	err := s.DB.SelectContext(ctx, &sources, `SELECT * FROM source ORDER BY id`)
+	err := s.DB.SelectContext(ctx, &sources, `SELECT * FROM notification_source ORDER BY id`)
 	if err != nil {
 		return nil, fmt.Errorf("list sources: %w", err)
 	}
@@ -64,7 +64,7 @@ func (s *Store) ListSources(ctx context.Context) ([]model.Source, error) {
 }

 func (s *Store) UpdateSource(ctx context.Context, id int, src *model.Source) error {
-	query := `UPDATE source SET name=?, parse_mode=?, parse_pattern=?, status=? WHERE id=?`
+	query := `UPDATE notification_source SET name=?, parse_mode=?, parse_pattern=?, status=? WHERE id=?`
 	_, err := s.DB.ExecContext(ctx, query, src.Name, src.ParseMode, src.ParsePattern, src.Status, id)
 	if err != nil {
 		return fmt.Errorf("update source %d: %w", id, err)
@@ -73,7 +73,7 @@ func (s *Store) UpdateSource(ctx context.Context, id int, src *model.Source) err
 }

 func (s *Store) DeleteSource(ctx context.Context, id int) error {
-	_, err := s.DB.ExecContext(ctx, `DELETE FROM source WHERE id = ?`, id)
+	_, err := s.DB.ExecContext(ctx, `DELETE FROM notification_source WHERE id = ?`, id)
 	if err != nil {
 		return fmt.Errorf("delete source %d: %w", id, err)
 	}
@@ -8,7 +8,7 @@ import (
 )

 func (s *Store) CreateTemplate(ctx context.Context, t *model.Template) error {
-	query := `INSERT INTO template (name, content) VALUES (?, ?)`
+	query := `INSERT INTO notification_template (name, content) VALUES (?, ?)`
 	result, err := s.DB.ExecContext(ctx, query, t.Name, t.Content)
 	if err != nil {
 		return fmt.Errorf("create template: %w", err)
@@ -20,7 +20,7 @@ func (s *Store) CreateTemplate(ctx context.Context, t *model.Template) error {

 func (s *Store) GetTemplate(ctx context.Context, id int) (*model.Template, error) {
 	var t model.Template
-	err := s.DB.GetContext(ctx, &t, `SELECT * FROM template WHERE id = ?`, id)
+	err := s.DB.GetContext(ctx, &t, `SELECT * FROM notification_template WHERE id = ?`, id)
 	if err != nil {
 		return nil, fmt.Errorf("get template %d: %w", id, err)
 	}
@@ -29,7 +29,7 @@ func (s *Store) GetTemplate(ctx context.Context, id int) (*model.Template, error

 func (s *Store) GetTemplateByName(ctx context.Context, name string) (*model.Template, error) {
 	var t model.Template
-	err := s.DB.GetContext(ctx, &t, `SELECT * FROM template WHERE name = ?`, name)
+	err := s.DB.GetContext(ctx, &t, `SELECT * FROM notification_template WHERE name = ?`, name)
 	if err != nil {
 		return nil, fmt.Errorf("get template by name %s: %w", name, err)
 	}
@@ -38,7 +38,7 @@ func (s *Store) GetTemplateByName(ctx context.Context, name string) (*model.Temp

 func (s *Store) ListTemplates(ctx context.Context) ([]model.Template, error) {
 	var templates []model.Template
-	err := s.DB.SelectContext(ctx, &templates, `SELECT * FROM template ORDER BY id`)
+	err := s.DB.SelectContext(ctx, &templates, `SELECT * FROM notification_template ORDER BY id`)
 	if err != nil {
 		return nil, fmt.Errorf("list templates: %w", err)
 	}
@@ -46,7 +46,7 @@ func (s *Store) ListTemplates(ctx context.Context) ([]model.Template, error) {
 }

 func (s *Store) UpdateTemplate(ctx context.Context, id int, t *model.Template) error {
-	query := `UPDATE template SET name=?, content=? WHERE id=?`
+	query := `UPDATE notification_template SET name=?, content=? WHERE id=?`
 	_, err := s.DB.ExecContext(ctx, query, t.Name, t.Content, id)
 	if err != nil {
 		return fmt.Errorf("update template %d: %w", id, err)
@@ -55,7 +55,7 @@ func (s *Store) UpdateTemplate(ctx context.Context, id int, t *model.Template) e
 }

 func (s *Store) DeleteTemplate(ctx context.Context, id int) error {
-	_, err := s.DB.ExecContext(ctx, `DELETE FROM template WHERE id = ?`, id)
+	_, err := s.DB.ExecContext(ctx, `DELETE FROM notification_template WHERE id = ?`, id)
 	if err != nil {
 		return fmt.Errorf("delete template %d: %w", id, err)
 	}
@@ -1,6 +1,6 @@
-DROP TABLE IF EXISTS message_log;
-DROP TABLE IF EXISTS rule_channel;
-DROP TABLE IF EXISTS rule;
-DROP TABLE IF EXISTS channel;
-DROP TABLE IF EXISTS template;
-DROP TABLE IF EXISTS source;
+DROP TABLE IF EXISTS notification_message_log;
+DROP TABLE IF EXISTS notification_rule_channel;
+DROP TABLE IF EXISTS notification_rule;
+DROP TABLE IF EXISTS notification_channel;
+DROP TABLE IF EXISTS notification_template;
+DROP TABLE IF EXISTS notification_source;
@@ -1,4 +1,4 @@
-CREATE TABLE source (
+CREATE TABLE notification_source (
    id            INT AUTO_INCREMENT PRIMARY KEY,
    name          VARCHAR(64) NOT NULL,
    api_key       VARCHAR(128) NOT NULL,
@@ -11,7 +11,7 @@ CREATE TABLE source (
    UNIQUE KEY uk_api_key (api_key)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-CREATE TABLE template (
+CREATE TABLE notification_template (
    id         INT AUTO_INCREMENT PRIMARY KEY,
    name       VARCHAR(64) NOT NULL,
    content    TEXT NOT NULL,
@@ -20,7 +20,7 @@ CREATE TABLE template (
    UNIQUE KEY uk_name (name)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-CREATE TABLE channel (
+CREATE TABLE notification_channel (
    id         INT AUTO_INCREMENT PRIMARY KEY,
    name       VARCHAR(32) NOT NULL,
    type       VARCHAR(32) NOT NULL,
@@ -31,7 +31,7 @@ CREATE TABLE channel (
    UNIQUE KEY uk_name (name)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-CREATE TABLE rule (
+CREATE TABLE notification_rule (
    id          INT AUTO_INCREMENT PRIMARY KEY,
    source_id   INT NOT NULL,
    event       VARCHAR(64) NOT NULL,
@@ -41,21 +41,21 @@ CREATE TABLE rule (
    created_at  DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at  DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_source_event (source_id, event),
-    FOREIGN KEY (source_id)   REFERENCES source(id),
-    FOREIGN KEY (template_id) REFERENCES template(id)
+    FOREIGN KEY (source_id)   REFERENCES notification_source(id),
+    FOREIGN KEY (template_id) REFERENCES notification_template(id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-CREATE TABLE rule_channel (
+CREATE TABLE notification_rule_channel (
    id         INT AUTO_INCREMENT PRIMARY KEY,
    rule_id    INT NOT NULL,
    channel_id INT NOT NULL,
    enabled    TINYINT NOT NULL DEFAULT 1,
    UNIQUE KEY uk_rule_channel (rule_id, channel_id),
-    FOREIGN KEY (rule_id)    REFERENCES rule(id) ON DELETE CASCADE,
-    FOREIGN KEY (channel_id) REFERENCES channel(id)
+    FOREIGN KEY (rule_id)    REFERENCES notification_rule(id) ON DELETE CASCADE,
+    FOREIGN KEY (channel_id) REFERENCES notification_channel(id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-CREATE TABLE message_log (
+CREATE TABLE notification_message_log (
    id          BIGINT AUTO_INCREMENT PRIMARY KEY,
    rule_id     INT NOT NULL,
    channel_id  INT NOT NULL,
Author	SHA1	Message	Date
ryan	a0d26fa3a3	refactor: rename database tables for notification entities to improve clarity and consistency	2026-07-02 16:04:18 +08:00
ryan	c5225d8c97	feat: add Logbull remote logging via slog handler, tee to stdout	2026-06-27 15:17:59 +08:00
ryan	835b92ec00	fix: wire retry into router, fix Docker config path, fix condition error handling, remove dead code	2026-06-27 13:32:33 +08:00
ryan	062014773c	fix: remove compiled binary, update gitignore	2026-06-27 13:29:33 +08:00
ryan	fa7f7490f5	feat: main assembly — wire all components, gin router, graceful shutdown	2026-06-27 13:28:27 +08:00
ryan	2271a54139	feat: retry mechanism with exponential backoff	2026-06-27 13:26:13 +08:00
ryan	8e0c95edd9	feat: rule CRUD with enable/disable and message log query	2026-06-27 13:23:37 +08:00
ryan	682ca3df81	feat: management handlers — source, template, channel CRUD	2026-06-27 13:23:10 +08:00
ryan	2cf6601be0	feat: notify handler — parse, match, evaluate, render, route	2026-06-27 13:22:14 +08:00
ryan	127cf12656	feat: auth and rate limit middleware	2026-06-27 13:20:56 +08:00