ryan/aiaa-notification-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
976ec3ca70bb807c073a83b8569e76343bee54b1
aiaa-notification-server/docs/superpowers/plans/2025-06-27-notification-service-plan.md
T

3725 lines
92 KiB
Markdown
Raw Blame History

# Notification Service Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Build a pure-Go notification service — webhook in, parse (JSON/Regex/Text), match rules, evaluate conditions, render Go templates, dispatch to DingTalk/WeCom/Email/Bark.
**Architecture:** Gin HTTP → auth+ratelimit middleware → handler → parser → engine(matcher→condition→renderer→router) → adapter(ChannelSender interface) → external services. MySQL for config, Redis for cache+ratelimit.
**Tech Stack:** Go 1.22+, gin, sqlx, go-redis/v9, text/template, viper, slog, golang-migrate, MySQL 8.0, Redis 7
## Global Constraints
- Go 1.22+, module path: `aiaa-notification-service`
- MySQL 8.0 InnoDB, Redis 7
- Config via `config/config.yaml` with `${ENV_VAR}` interpolation
- Tests: `testing` stdlib + `httptest` for handlers, `testify/assert` for assertions
- Docker multi-stage: `golang:1.22-alpine` → `alpine:3.20`
## File Structure
```
notification-service/
├── cmd/server/main.go
├── config/config.yaml
├── migrations/
│ ├── 001_init.up.sql
│ └── 001_init.down.sql
├── internal/
│ ├── config/config.go
│ ├── model/model.go
│ ├── store/
│ │ ├── mysql.go
│ │ ├── source.go
│ │ ├── template.go
│ │ ├── channel.go
│ │ ├── rule.go
│ │ └── message_log.go
│ ├── cache/redis.go
│ ├── parser/
│ │ ├── parser.go ← Parser interface + factory
│ │ ├── json.go
│ │ ├── regex.go
│ │ └── text.go
│ ├── condition/
│ │ └── evaluator.go ← condition array evaluator
│ ├── adapter/
│ │ ├── adapter.go
│ │ ├── dingtalk.go
│ │ ├── wecom.go
│ │ ├── email.go
│ │ └── bark.go
│ ├── engine/
│ │ ├── matcher.go
│ │ ├── renderer.go
│ │ └── router.go
│ ├── handler/
│ │ ├── middleware.go
│ │ ├── notify.go
│ │ ├── source.go
│ │ ├── template.go
│ │ ├── channel.go
│ │ ├── rule.go
│ │ └── message_log.go
│ └── retry/retry.go
├── Dockerfile
├── docker-compose.yml
├── Makefile
└── go.mod
```
---
### Task 1: Project Scaffold & Configuration Loading
**Files:**
- Create: `go.mod`
- Create: `cmd/server/main.go` (skeleton)
- Create: `config/config.yaml`
- Create: `internal/config/config.go`
- Create: `Makefile`
**Interfaces:**
- Produces: `config.Config` struct, `config.Load(path string) (*Config, error)`
- [ ] **Step 1: Initialize Go module**
```bash
cd /Users/ryan/Documents/code/go/aiaa-notification-service
go mod init aiaa-notification-service
```
- [ ] **Step 2: Create config file**
File: `config/config.yaml`
```yaml
server:
port: 8080
admin_key: "admin-sk-change-me"
database:
host: "127.0.0.1"
port: 3306
user: "notify"
password: "${DB_PASSWORD:-notify}"
database: "notification"
redis:
host: "127.0.0.1"
port: 6379
password: ""
db: 0
smtp:
host: "smtp.example.com"
port: 587
user: "notify@example.com"
password: "${SMTP_PASSWORD:-}"
from: "Notification Service <notify@example.com>"
rate_limit:
default: 100
```
- [ ] **Step 3: Write Config loader**
File: `internal/config/config.go`
```go
package config
import (
"fmt"
"os"
"strings"
"github.com/spf13/viper"
)
type Config struct {
Server ServerConfig `mapstructure:"server"`
Database DatabaseConfig `mapstructure:"database"`
Redis RedisConfig `mapstructure:"redis"`
SMTP SMTPConfig `mapstructure:"smtp"`
RateLimit RateLimitConfig `mapstructure:"rate_limit"`
}
type ServerConfig struct {
Port int `mapstructure:"port"`
AdminKey string `mapstructure:"admin_key"`
}
type DatabaseConfig struct {
Host string `mapstructure:"host"`
Port int `mapstructure:"port"`
User string `mapstructure:"user"`
Password string `mapstructure:"password"`
Database string `mapstructure:"database"`
}
func (d DatabaseConfig) DSN() string {
return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=true&loc=Local",
d.User, d.Password, d.Host, d.Port, d.Database)
}
type RedisConfig struct {
Host string `mapstructure:"host"`
Port int `mapstructure:"port"`
Password string `mapstructure:"password"`
DB int `mapstructure:"db"`
}
func (r RedisConfig) Addr() string {
return fmt.Sprintf("%s:%d", r.Host, r.Port)
}
type SMTPConfig struct {
Host string `mapstructure:"host"`
Port int `mapstructure:"port"`
User string `mapstructure:"user"`
Password string `mapstructure:"password"`
From string `mapstructure:"from"`
}
type RateLimitConfig struct {
Default int `mapstructure:"default"`
}
func Load(path string) (*Config, error) {
v := viper.New()
v.SetConfigFile(path)
v.SetEnvPrefix("NOTIFY")
v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
v.AutomaticEnv()
if err := v.ReadInConfig(); err != nil {
return nil, fmt.Errorf("read config: %w", err)
}
// Expand ${ENV_VAR} placeholders in config values
for _, key := range v.AllKeys() {
val := v.GetString(key)
if strings.Contains(val, "${") {
expanded := expandEnv(val)
v.Set(key, expanded)
}
}
var cfg Config
if err := v.Unmarshal(&cfg); err != nil {
return nil, fmt.Errorf("unmarshal config: %w", err)
}
return &cfg, nil
}
func expandEnv(s string) string {
return os.Expand(s, func(key string) string {
// support ${VAR:-default}
if i := strings.Index(key, ":-"); i >= 0 {
name := key[:i]
def := key[i+2:]
if v, ok := os.LookupEnv(name); ok {
return v
}
return def
}
return os.Getenv(key)
})
}
```
- [ ] **Step 4: Create skeleton main.go**
File: `cmd/server/main.go`
```go
package main
import (
"log/slog"
"os"
"aiaa-notification-service/internal/config"
)
func main() {
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
slog.SetDefault(logger)
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)
}
```
- [ ] **Step 5: Create Makefile**
File: `Makefile`
```makefile
.PHONY: build run test docker-build docker-up docker-down migrate-up migrate-down
build:
go build -o bin/server ./cmd/server
run:
go run ./cmd/server
test:
go test ./internal/... -v -count=1
docker-build:
docker build -t notification-service .
docker-up:
docker-compose up -d
docker-down:
docker-compose down
migrate-up:
migrate -path migrations -database "mysql://notify:notify@tcp(127.0.0.1:3306)/notification" up
migrate-down:
migrate -path migrations -database "mysql://notify:notify@tcp(127.0.0.1:3306)/notification" down
```
- [ ] **Step 6: Install dependencies and verify build**
```bash
go mod tidy
go build ./...
```
Expected: build succeeds with no errors.
- [ ] **Step 7: Commit**
```bash
git add -A && git commit -m "feat: project scaffold, config loading, skeleton main"
```
---
### Task 2: Models & Database Migration
**Files:**
- Create: `internal/model/model.go`
- Create: `migrations/001_init.up.sql`
- Create: `migrations/001_init.down.sql`
**Interfaces:**
- Produces: `model.Source`, `model.Template`, `model.Channel`, `model.Rule`, `model.RuleChannel`, `model.MessageLog`, `model.Condition`
- [ ] **Step 1: Write Go models**
File: `internal/model/model.go`
```go
package model
import (
"database/sql"
"encoding/json"
"time"
)
type Source struct {
ID int `db:"id" json:"id"`
Name string `db:"name" json:"name"`
APIKey string `db:"api_key" json:"api_key,omitempty"`
ParseMode string `db:"parse_mode" json:"parse_mode"`
ParsePattern string `db:"parse_pattern" json:"parse_pattern,omitempty"`
Status int `db:"status" json:"status"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
}
type Template struct {
ID int `db:"id" json:"id"`
Name string `db:"name" json:"name"`
Content string `db:"content" json:"content"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
}
type Channel struct {
ID int `db:"id" json:"id"`
Name string `db:"name" json:"name"`
Type string `db:"type" json:"type"`
Config *json.RawMessage `db:"config" json:"config"`
Status int `db:"status" json:"status"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
}
type Rule struct {
ID int `db:"id" json:"id"`
SourceID int `db:"source_id" json:"source_id"`
Event string `db:"event" json:"event"`
TemplateID int `db:"template_id" json:"template_id"`
Conditions *json.RawMessage `db:"conditions" json:"conditions,omitempty"`
Enabled int `db:"enabled" json:"enabled"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
}
type Condition struct {
Field string `json:"field"`
Op string `json:"op"`
Value string `json:"value,omitempty"`
}
type RuleChannel struct {
ID int `db:"id" json:"id"`
RuleID int `db:"rule_id" json:"rule_id"`
ChannelID int `db:"channel_id" json:"channel_id"`
Enabled int `db:"enabled" json:"enabled"`
}
type MessageLog struct {
ID int64 `db:"id" json:"id"`
RuleID int `db:"rule_id" json:"rule_id"`
ChannelID int `db:"channel_id" json:"channel_id"`
Source string `db:"source" json:"source"`
Event string `db:"event" json:"event"`
Payload json.RawMessage `db:"payload" json:"payload"`
Content string `db:"content" json:"content"`
Status string `db:"status" json:"status"`
RetryCount int `db:"retry_count" json:"retry_count"`
Response sql.NullString `db:"response" json:"response,omitempty"`
ErrorMsg sql.NullString `db:"error_msg" json:"error_msg,omitempty"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
}
```
- [ ] **Step 2: Write migration SQL**
File: `migrations/001_init.up.sql`
```sql
CREATE TABLE source (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(64) NOT NULL,
api_key VARCHAR(128) NOT NULL,
parse_mode VARCHAR(16) NOT NULL DEFAULT 'json',
parse_pattern VARCHAR(512) DEFAULT NULL,
status TINYINT NOT NULL DEFAULT 1,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_name (name),
UNIQUE KEY uk_api_key (api_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE template (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(64) NOT NULL,
content TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_name (name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE channel (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(32) NOT NULL,
type VARCHAR(32) NOT NULL,
config JSON NOT NULL,
status TINYINT NOT NULL DEFAULT 1,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_name (name)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE rule (
id INT AUTO_INCREMENT PRIMARY KEY,
source_id INT NOT NULL,
event VARCHAR(64) NOT NULL,
template_id INT NOT NULL,
conditions JSON DEFAULT NULL,
enabled TINYINT NOT NULL DEFAULT 1,
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)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE 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)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE message_log (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
rule_id INT NOT NULL,
channel_id INT NOT NULL,
source VARCHAR(64) NOT NULL,
event VARCHAR(64) NOT NULL,
payload JSON NOT NULL,
content TEXT NOT NULL,
status ENUM('pending','success','failed','retrying') NOT NULL DEFAULT 'pending',
retry_count INT NOT NULL DEFAULT 0,
response TEXT DEFAULT NULL,
error_msg TEXT DEFAULT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
INDEX idx_source_event_time (source, event, created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
```
File: `migrations/001_init.down.sql`
```sql
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;
```
- [ ] **Step 3: Verify models compile**
```bash
go build ./internal/model/...
```
Expected: no errors.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: data models and database migration"
```
---
### Task 3: MySQL Store — Connection & Source CRUD
**Files:**
- Create: `internal/store/mysql.go`
- Create: `internal/store/source.go`
**Interfaces:**
- Consumes: `config.Config`, `model.Source`
- Produces: `store.NewStore(cfg *config.DatabaseConfig) (*Store, error)`, `Store.CreateSource`, `Store.GetSource`, `Store.GetSourceByAPIKey`, `Store.GetSourceByName`, `Store.ListSources`, `Store.UpdateSource`, `Store.DeleteSource`
- [ ] **Step 1: Write store connection**
File: `internal/store/mysql.go`
```go
package store
import (
"fmt"
"aiaa-notification-service/internal/config"
"github.com/jmoiron/sqlx"
_ "github.com/go-sql-driver/mysql"
)
type Store struct {
DB *sqlx.DB
}
func NewStore(cfg config.DatabaseConfig) (*Store, error) {
db, err := sqlx.Connect("mysql", cfg.DSN())
if err != nil {
return nil, fmt.Errorf("connect mysql: %w", err)
}
db.SetMaxOpenConns(25)
db.SetMaxIdleConns(5)
return &Store{DB: db}, nil
}
func (s *Store) Close() error {
return s.DB.Close()
}
```
- [ ] **Step 2: Write Source store methods**
File: `internal/store/source.go`
```go
package store
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"aiaa-notification-service/internal/model"
)
func generateAPIKey() string {
b := make([]byte, 32)
rand.Read(b)
return "sk-" + hex.EncodeToString(b)
}
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 (?, ?, ?, ?, ?)`
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)
}
id, _ := result.LastInsertId()
src.ID = int(id)
return nil
}
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)
if err != nil {
return nil, fmt.Errorf("get source %d: %w", id, err)
}
return &src, nil
}
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)
if err != nil {
return nil, fmt.Errorf("get source by api_key: %w", err)
}
return &src, nil
}
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)
if err != nil {
return nil, fmt.Errorf("get source by name %s: %w", name, err)
}
return &src, nil
}
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`)
if err != nil {
return nil, fmt.Errorf("list sources: %w", err)
}
return sources, nil
}
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=?`
_, 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)
}
return nil
}
func (s *Store) DeleteSource(ctx context.Context, id int) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM source WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("delete source %d: %w", id, err)
}
return nil
}
```
- [ ] **Step 3: Install dependencies**
```bash
go get github.com/jmoiron/sqlx github.com/go-sql-driver/mysql
go mod tidy
```
- [ ] **Step 4: Verify compilation**
```bash
go build ./internal/store/...
```
Expected: no errors.
- [ ] **Step 5: Commit**
```bash
git add -A && git commit -m "feat: store connection and source CRUD"
```
---
### Task 4: MySQL Store — Template & Channel CRUD
**Files:**
- Create: `internal/store/template.go`
- Create: `internal/store/channel.go`
**Interfaces:**
- Produces: `Store.CreateTemplate`, `Store.GetTemplate`, `Store.GetTemplateByName`, `Store.ListTemplates`, `Store.UpdateTemplate`, `Store.DeleteTemplate`
- Produces: `Store.CreateChannel`, `Store.GetChannel`, `Store.GetChannelByName`, `Store.ListChannels`, `Store.UpdateChannel`, `Store.DeleteChannel`
- [ ] **Step 1: Write Template store**
File: `internal/store/template.go`
```go
package store
import (
"context"
"fmt"
"aiaa-notification-service/internal/model"
)
func (s *Store) CreateTemplate(ctx context.Context, t *model.Template) error {
query := `INSERT INTO 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)
}
id, _ := result.LastInsertId()
t.ID = int(id)
return nil
}
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)
if err != nil {
return nil, fmt.Errorf("get template %d: %w", id, err)
}
return &t, nil
}
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)
if err != nil {
return nil, fmt.Errorf("get template by name %s: %w", name, err)
}
return &t, nil
}
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`)
if err != nil {
return nil, fmt.Errorf("list templates: %w", err)
}
return templates, nil
}
func (s *Store) UpdateTemplate(ctx context.Context, id int, t *model.Template) error {
query := `UPDATE 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)
}
return nil
}
func (s *Store) DeleteTemplate(ctx context.Context, id int) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM template WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("delete template %d: %w", id, err)
}
return nil
}
```
- [ ] **Step 2: Write Channel store**
File: `internal/store/channel.go`
```go
package store
import (
"context"
"encoding/json"
"fmt"
"aiaa-notification-service/internal/model"
)
func (s *Store) CreateChannel(ctx context.Context, ch *model.Channel) error {
configJSON, err := json.Marshal(ch.Config)
if err != nil {
return fmt.Errorf("marshal channel config: %w", err)
}
query := `INSERT INTO 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)
}
id, _ := result.LastInsertId()
ch.ID = int(id)
return nil
}
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)
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)
}
raw := json.RawMessage(configBytes)
ch.Config = &raw
return &ch, nil
}
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)
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)
}
raw := json.RawMessage(configBytes)
ch.Config = &raw
return &ch, nil
}
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`)
if err != nil {
return nil, fmt.Errorf("list channels: %w", err)
}
defer rows.Close()
var channels []model.Channel
for rows.Next() {
var ch model.Channel
var configBytes []byte
if err := rows.Scan(&ch.ID, &ch.Name, &ch.Type, &configBytes, &ch.Status, &ch.CreatedAt, &ch.UpdatedAt); err != nil {
return nil, fmt.Errorf("scan channel: %w", err)
}
raw := json.RawMessage(configBytes)
ch.Config = &raw
channels = append(channels, ch)
}
return channels, rows.Err()
}
func (s *Store) UpdateChannel(ctx context.Context, id int, ch *model.Channel) error {
configJSON, err := json.Marshal(ch.Config)
if err != nil {
return fmt.Errorf("marshal channel config: %w", err)
}
query := `UPDATE 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)
}
return nil
}
func (s *Store) DeleteChannel(ctx context.Context, id int) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM channel WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("delete channel %d: %w", id, err)
}
return nil
}
```
- [ ] **Step 3: Verify compilation**
```bash
go build ./internal/store/...
```
Expected: no errors.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: template and channel store CRUD"
```
---
### Task 5: MySQL Store — Rule & MessageLog
**Files:**
- Create: `internal/store/rule.go`
- Create: `internal/store/message_log.go`
**Interfaces:**
- Produces: `Store.CreateRule`, `Store.GetRule`, `Store.GetRuleBySourceEvent`, `Store.ListRules`, `Store.UpdateRule`, `Store.DeleteRule`, `Store.AddRuleChannel`, `Store.RemoveRuleChannel`, `Store.GetRuleChannels`, `Store.UpdateRuleChannelEnabled`
- Produces: `Store.CreateMessageLog`, `Store.UpdateMessageLog`, `Store.ListMessageLogs`, `MessageLogFilter`
- [ ] **Step 1: Write Rule store (includes RuleChannel ops)**
File: `internal/store/rule.go`
```go
package store
import (
"context"
"encoding/json"
"fmt"
"aiaa-notification-service/internal/model"
)
func (s *Store) CreateRule(ctx context.Context, r *model.Rule, channelIDs []int) error {
tx, err := s.DB.BeginTxx(ctx, nil)
if err != nil {
return fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback()
query := `INSERT INTO rule (source_id, event, template_id, conditions, enabled) VALUES (?, ?, ?, ?, ?)`
condsJSON, _ := marshalJSON(r.Conditions)
result, err := tx.ExecContext(ctx, query, r.SourceID, r.Event, r.TemplateID, condsJSON, r.Enabled)
if err != nil {
return fmt.Errorf("create rule: %w", err)
}
id, _ := result.LastInsertId()
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)
if err != nil {
return fmt.Errorf("add rule_channel: %w", err)
}
}
return tx.Commit()
}
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)
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)
}
if len(condsBytes) > 0 && string(condsBytes) != "null" {
raw := json.RawMessage(condsBytes)
r.Conditions = &raw
}
return &r, nil
}
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`
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)
}
if len(condsBytes) > 0 && string(condsBytes) != "null" {
raw := json.RawMessage(condsBytes)
r.Conditions = &raw
}
return &r, nil
}
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`)
if err != nil {
return nil, fmt.Errorf("list rules: %w", err)
}
defer rows.Close()
return scanRules(rows)
}
func (s *Store) UpdateRule(ctx context.Context, id int, r *model.Rule, channelIDs []int) error {
tx, err := s.DB.BeginTxx(ctx, nil)
if err != nil {
return fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback()
condsJSON, _ := marshalJSON(r.Conditions)
_, err = tx.ExecContext(ctx, `UPDATE 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 {
_, _ = tx.ExecContext(ctx, `DELETE FROM rule_channel WHERE rule_id = ?`, id)
for _, chID := range channelIDs {
_, err := tx.ExecContext(ctx, `INSERT INTO rule_channel (rule_id, channel_id, enabled) VALUES (?, ?, 1)`, id, chID)
if err != nil {
return fmt.Errorf("add rule_channel: %w", err)
}
}
}
return tx.Commit()
}
func (s *Store) DeleteRule(ctx context.Context, id int) error {
_, err := s.DB.ExecContext(ctx, `DELETE FROM rule WHERE id = ?`, id)
return err
}
func (s *Store) SetRuleEnabled(ctx context.Context, id int, enabled bool) error {
v := 0
if enabled {
v = 1
}
_, err := s.DB.ExecContext(ctx, `UPDATE rule SET enabled = ? WHERE id = ?`, v, id)
return err
}
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)
if err != nil {
return nil, fmt.Errorf("get rule channels: %w", err)
}
return rcs, nil
}
func (s *Store) SetRuleChannelEnabled(ctx context.Context, ruleID, channelID int, enabled bool) error {
v := 0
if enabled {
v = 1
}
_, err := s.DB.ExecContext(ctx, `UPDATE rule_channel SET enabled = ? WHERE rule_id = ? AND channel_id = ?`, v, ruleID, channelID)
return err
}
// helpers
func marshalJSON(v *json.RawMessage) (interface{}, error) {
if v == nil {
return nil, nil
}
return json.Marshal(v)
}
func scanRules(rows *sql.Rows) ([]model.Rule, error) {
var rules []model.Rule
for rows.Next() {
var r model.Rule
var condsBytes []byte
if err := rows.Scan(&r.ID, &r.SourceID, &r.Event, &r.TemplateID, &condsBytes, &r.Enabled, &r.CreatedAt, &r.UpdatedAt); err != nil {
return nil, err
}
if len(condsBytes) > 0 && string(condsBytes) != "null" {
raw := json.RawMessage(condsBytes)
r.Conditions = &raw
}
rules = append(rules, r)
}
return rules, rows.Err()
}
```
Note: `scanRules` needs `database/sql` import. Add to top of file:
```go
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"aiaa-notification-service/internal/model"
)
```
- [ ] **Step 2: Write MessageLog store**
File: `internal/store/message_log.go`
```go
package store
import (
"context"
"fmt"
"aiaa-notification-service/internal/model"
)
type MessageLogFilter struct {
Source string `form:"source"`
Event string `form:"event"`
Status string `form:"status"`
Page int `form:"page"`
PageSize int `form:"page_size"`
}
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 (?, ?, ?, ?, ?, ?, ?, ?)`
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)
}
id, _ := result.LastInsertId()
ml.ID = id
return nil
}
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=?`
_, err := s.DB.ExecContext(ctx, query, status, response, errMsg, id)
return err
}
func (s *Store) ListMessageLogs(ctx context.Context, filter MessageLogFilter) ([]model.MessageLog, int, error) {
where := "WHERE 1=1"
args := []interface{}{}
if filter.Source != "" {
where += " AND source = ?"
args = append(args, filter.Source)
}
if filter.Event != "" {
where += " AND event = ?"
args = append(args, filter.Event)
}
if filter.Status != "" {
where += " AND status = ?"
args = append(args, filter.Status)
}
var count int
countQuery := "SELECT COUNT(*) FROM message_log " + where
if err := s.DB.GetContext(ctx, &count, countQuery, args...); err != nil {
return nil, 0, err
}
if filter.Page <= 0 {
filter.Page = 1
}
if filter.PageSize <= 0 {
filter.PageSize = 20
}
offset := (filter.Page - 1) * filter.PageSize
var logs []model.MessageLog
query := "SELECT * FROM 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
}
return logs, count, nil
}
```
- [ ] **Step 3: Verify compilation**
```bash
go build ./internal/store/...
```
Expected: no errors. Fix any import issues.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: rule and message_log store CRUD"
```
---
### Task 6: Redis Cache Layer
**Files:**
- Create: `internal/cache/redis.go`
**Interfaces:**
- Consumes: `config.RedisConfig`
- Produces: `cache.NewCache(cfg config.RedisConfig) (*Cache, error)`, `Cache.GetRule`, `Cache.SetRule`, `Cache.InvalidateRule`, `Cache.GetChannels`, `Cache.SetChannels`, `Cache.InvalidateChannels`, `Cache.CheckRateLimit`, `Cache.InvalidateBySource`, `Cache.InvalidateByTemplate`
- [ ] **Step 1: Write cache layer**
File: `internal/cache/redis.go`
```go
package cache
import (
"context"
"encoding/json"
"fmt"
"strconv"
"time"
"aiaa-notification-service/internal/model"
"github.com/redis/go-redis/v9"
)
type Cache struct {
rdb *redis.Client
}
// CachedRule holds the minimal info needed after rule matching.
type CachedRule struct {
RuleID int `json:"rule_id"`
TemplateID int `json:"template_id"`
Content string `json:"content"`
Conditions string `json:"conditions"` // JSON string, empty if null
}
type CachedChannel struct {
ID int `json:"id"`
Type string `json:"type"`
Config *json.RawMessage `json:"config"`
}
func NewCache(cfg model.config.RedisConfig) (*Cache, error) {
rdb := redis.NewClient(&redis.Options{
Addr: cfg.Addr(),
Password: cfg.Password,
DB: cfg.DB,
})
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
if err := rdb.Ping(ctx).Err(); err != nil {
return nil, fmt.Errorf("redis ping: %w", err)
}
return &Cache{rdb: rdb}, nil
}
// --- Rule cache ---
func ruleKey(sourceID int, event string) string {
return fmt.Sprintf("notify:rule:%d:%s", sourceID, event)
}
func (c *Cache) GetRule(ctx context.Context, sourceID int, event string) (*CachedRule, error) {
data, err := c.rdb.Get(ctx, ruleKey(sourceID, event)).Bytes()
if err != nil {
return nil, err
}
var cr CachedRule
if err := json.Unmarshal(data, &cr); err != nil {
return nil, err
}
return &cr, nil
}
func (c *Cache) SetRule(ctx context.Context, sourceID int, event string, cr *CachedRule) error {
data, err := json.Marshal(cr)
if err != nil {
return err
}
return c.rdb.Set(ctx, ruleKey(sourceID, event), data, 5*time.Minute).Err()
}
func (c *Cache) InvalidateRule(ctx context.Context, sourceID int, event string) error {
return c.rdb.Del(ctx, ruleKey(sourceID, event)).Err()
}
// --- Channel cache ---
func channelsKey(ruleID int) string {
return fmt.Sprintf("notify:channels:%d", ruleID)
}
func (c *Cache) GetChannels(ctx context.Context, ruleID int) ([]CachedChannel, error) {
data, err := c.rdb.Get(ctx, channelsKey(ruleID)).Bytes()
if err != nil {
return nil, err
}
var channels []CachedChannel
if err := json.Unmarshal(data, &channels); err != nil {
return nil, err
}
return channels, nil
}
func (c *Cache) SetChannels(ctx context.Context, ruleID int, channels []CachedChannel) error {
data, err := json.Marshal(channels)
if err != nil {
return err
}
return c.rdb.Set(ctx, channelsKey(ruleID), data, 5*time.Minute).Err()
}
func (c *Cache) InvalidateChannels(ctx context.Context, ruleID int) error {
return c.rdb.Del(ctx, channelsKey(ruleID)).Err()
}
// --- Bulk invalidation ---
func (c *Cache) InvalidateBySource(ctx context.Context, sourceID int) error {
pattern := fmt.Sprintf("notify:rule:%d:*", sourceID)
iter := c.rdb.Scan(ctx, 0, pattern, 0).Iterator()
for iter.Next(ctx) {
c.rdb.Del(ctx, iter.Val())
}
return iter.Err()
}
func (c *Cache) InvalidateByTemplate(ctx context.Context, templateID int) error {
// Template changes mean all cached rule info is stale. Simplest: scan all rule keys.
// In production, you'd maintain a template→rule reverse index. For v1, scan is acceptable.
iter := c.rdb.Scan(ctx, 0, "notify:rule:*", 0).Iterator()
for iter.Next(ctx) {
c.rdb.Del(ctx, iter.Val())
}
return iter.Err()
}
// --- Rate limiter ---
func (c *Cache) CheckRateLimit(ctx context.Context, sourceID int, limitPerSec int) (bool, int, error) {
key := fmt.Sprintf("ratelimit:%d", sourceID)
now := time.Now().Unix()
pipe := c.rdb.Pipeline()
pipe.ZRemRangeByScore(ctx, key, "0", strconv.FormatInt(now-1, 10))
countCmd := pipe.ZCard(ctx, key)
pipe.Exec(ctx)
count := countCmd.Val()
if count >= int64(limitPerSec) {
return false, 1, nil // rate limited, retry after 1s
}
c.rdb.ZAdd(ctx, key, redis.Z{Score: float64(now), Member: strconv.FormatInt(now*1000, 10)})
c.rdb.Expire(ctx, key, 2*time.Second)
return true, 0, nil
}
func (c *Cache) Close() error {
return c.rdb.Close()
}
```
Note: the import references `model.config.RedisConfig` — need to fix this. The cache package imports `config`, not model. Update the function signature:
```go
import (
"aiaa-notification-service/internal/config"
)
func NewCache(cfg config.RedisConfig) (*Cache, error) {
```
- [ ] **Step 2: Install go-redis**
```bash
go get github.com/redis/go-redis/v9
go mod tidy
```
- [ ] **Step 3: Verify compilation**
```bash
go build ./internal/cache/...
```
Expected: no errors.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: redis cache layer with rule/channel caching and rate limiter"
```
---
### Task 7: Message Parser (JSON / Regex / Text)
**Files:**
- Create: `internal/parser/parser.go`
- Create: `internal/parser/json.go`
- Create: `internal/parser/regex.go`
- Create: `internal/parser/text.go`
**Interfaces:**
- Produces: `parser.ParsedMessage`, `parser.Parser` interface, `parser.NewParser(parseMode string, parsePattern string) (Parser, error)`
- [ ] **Step 1: Write parser interface and ParsedMessage**
File: `internal/parser/parser.go`
```go
package parser
import (
"fmt"
)
// ParsedMessage is the result of parsing an incoming notification body.
type ParsedMessage struct {
Event string
Data map[string]interface{} // field name → value, passed to template
Body string // raw body (for text mode)
}
// Parser converts a raw HTTP body into a ParsedMessage.
type Parser interface {
Parse(body []byte) (*ParsedMessage, error)
}
func NewParser(parseMode string, parsePattern string) (Parser, error) {
switch parseMode {
case "json":
return &JSONParser{}, nil
case "regex":
if parsePattern == "" {
return nil, fmt.Errorf("regex parse mode requires parse_pattern")
}
return NewRegexParser(parsePattern)
case "text":
return &TextParser{}, nil
default:
return nil, fmt.Errorf("unknown parse_mode: %s", parseMode)
}
}
```
- [ ] **Step 2: Write JSON parser**
File: `internal/parser/json.go`
```go
package parser
import (
"encoding/json"
"fmt"
)
type JSONParser struct{}
func (p *JSONParser) Parse(body []byte) (*ParsedMessage, error) {
var raw map[string]interface{}
if err := json.Unmarshal(body, &raw); err != nil {
return nil, fmt.Errorf("json parse: %w", err)
}
event, ok := raw["event"].(string)
if !ok {
return nil, fmt.Errorf("json: missing 'event' field")
}
data, _ := raw["data"].(map[string]interface{})
if data == nil {
// If no "data" key, use the entire JSON as data (minus event)
data = make(map[string]interface{})
for k, v := range raw {
if k != "event" {
data[k] = v
}
}
}
return &ParsedMessage{
Event: event,
Data: data,
Body: string(body),
}, nil
}
```
- [ ] **Step 3: Write Regex parser**
File: `internal/parser/regex.go`
```go
package parser
import (
"fmt"
"regexp"
)
type RegexParser struct {
re *regexp.Regexp
}
func NewRegexParser(pattern string) (*RegexParser, error) {
re, err := regexp.Compile(pattern)
if err != nil {
return nil, fmt.Errorf("compile regex pattern: %w", err)
}
return &RegexParser{re: re}, nil
}
func (p *RegexParser) Parse(body []byte) (*ParsedMessage, error) {
matches := p.re.FindStringSubmatch(string(body))
if matches == nil {
return nil, fmt.Errorf("regex: body does not match pattern")
}
names := p.re.SubexpNames()
data := make(map[string]interface{})
var event string
for i, name := range names {
if i == 0 || name == "" {
continue
}
if name == "event" {
event = matches[i]
} else {
data[name] = matches[i]
}
}
return &ParsedMessage{
Event: event,
Data: data,
Body: string(body),
}, nil
}
```
- [ ] **Step 4: Write Text parser**
File: `internal/parser/text.go`
```go
package parser
type TextParser struct{}
func (p *TextParser) Parse(body []byte) (*ParsedMessage, error) {
return &ParsedMessage{
Event: "default", // text mode has no explicit event
Data: map[string]interface{}{"Body": string(body)},
Body: string(body),
}, nil
}
```
- [ ] **Step 5: Write parser tests**
File: `internal/parser/parser_test.go`
```go
package parser
import (
"testing"
)
func TestJSONParser(t *testing.T) {
p := &JSONParser{}
body := []byte(`{"event":"trade.open","data":{"symbol":"BTC","price":65000}}`)
msg, err := p.Parse(body)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if msg.Event != "trade.open" {
t.Errorf("expected event 'trade.open', got '%s'", msg.Event)
}
if msg.Data["symbol"] != "BTC" {
t.Errorf("expected symbol 'BTC', got '%v'", msg.Data["symbol"])
}
}
func TestJSONParser_NoDataKey(t *testing.T) {
p := &JSONParser{}
body := []byte(`{"event":"alert","symbol":"ETH"}`)
msg, err := p.Parse(body)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if msg.Data["symbol"] != "ETH" {
t.Errorf("expected symbol 'ETH' in data, got '%v'", msg.Data["symbol"])
}
}
func TestRegexParser(t *testing.T) {
p, err := NewRegexParser(`(?P<event>\w+) (?P<message>.+)`)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
body := []byte("trade.open BTC long at 65000")
msg, err := p.Parse(body)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if msg.Event != "trade.open" {
t.Errorf("expected event 'trade.open', got '%s'", msg.Event)
}
if msg.Data["message"] != "BTC long at 65000" {
t.Errorf("expected message, got '%v'", msg.Data["message"])
}
}
func TestTextParser(t *testing.T) {
p := &TextParser{}
body := []byte("raw notification text here")
msg, err := p.Parse(body)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if msg.Data["Body"] != "raw notification text here" {
t.Errorf("expected Body in data, got '%v'", msg.Data["Body"])
}
}
```
- [ ] **Step 6: Run tests**
```bash
go test ./internal/parser/... -v
```
Expected: all tests PASS.
- [ ] **Step 7: Commit**
```bash
git add -A && git commit -m "feat: message parser with json/regex/text modes"
```
---
### Task 8: Condition Evaluator
**Files:**
- Create: `internal/condition/evaluator.go`
**Interfaces:**
- Produces: `condition.Evaluate(conditions []model.Condition, data map[string]interface{}) bool`
- [ ] **Step 1: Write condition evaluator**
File: `internal/condition/evaluator.go`
```go
package condition
import (
"fmt"
"strconv"
"strings"
"aiaa-notification-service/internal/model"
)
// Evaluate checks all conditions against data. Returns true if ALL conditions pass.
// An empty conditions slice always returns true.
func Evaluate(conditions []model.Condition, data map[string]interface{}) bool {
if len(conditions) == 0 {
return true
}
for _, c := range conditions {
if !evaluateOne(c, data) {
return false
}
}
return true
}
func evaluateOne(c model.Condition, data map[string]interface{}) bool {
fieldVal, fieldExists := data[c.Field]
switch c.Op {
case "exists":
return fieldExists
case "not_exists":
return !fieldExists
case "eq":
if !fieldExists {
return false
}
return fmt.Sprintf("%v", fieldVal) == c.Value
case "ne":
if !fieldExists {
return false
}
return fmt.Sprintf("%v", fieldVal) != c.Value
case "contains":
if !fieldExists {
return false
}
return strings.Contains(fmt.Sprintf("%v", fieldVal), c.Value)
case "gt", "gte", "lt", "lte":
if !fieldExists {
return false
}
return compareNumeric(fieldVal, c.Value, c.Op)
default:
return false
}
}
func compareNumeric(fieldVal interface{}, value string, op string) bool {
fv, err := toFloat64(fieldVal)
if err != nil {
return false
}
cv, err := strconv.ParseFloat(value, 64)
if err != nil {
return false
}
switch op {
case "gt":
return fv > cv
case "gte":
return fv >= cv
case "lt":
return fv < cv
case "lte":
return fv <= cv
}
return false
}
func toFloat64(v interface{}) (float64, error) {
switch val := v.(type) {
case float64:
return val, nil
case float32:
return float64(val), nil
case int:
return float64(val), nil
case int64:
return float64(val), nil
case string:
return strconv.ParseFloat(val, 64)
case json.Number:
return val.Float64()
default:
return 0, fmt.Errorf("cannot convert %T to float64", v)
}
}
```
Note: needs `encoding/json` import for `json.Number`.
- [ ] **Step 2: Write tests**
File: `internal/condition/evaluator_test.go`
```go
package condition
import (
"testing"
"aiaa-notification-service/internal/model"
)
func TestEvaluate_Empty(t *testing.T) {
if !Evaluate(nil, nil) {
t.Error("empty conditions should pass")
}
}
func TestEvaluate_Exists(t *testing.T) {
conds := []model.Condition{{Field: "symbol", Op: "exists"}}
data := map[string]interface{}{"symbol": "BTC"}
if !Evaluate(conds, data) {
t.Error("symbol exists, should pass")
}
}
func TestEvaluate_NotExists(t *testing.T) {
conds := []model.Condition{{Field: "symbol", Op: "not_exists"}}
data := map[string]interface{}{"price": 100}
if !Evaluate(conds, data) {
t.Error("symbol not exists, should pass")
}
}
func TestEvaluate_Gt(t *testing.T) {
conds := []model.Condition{{Field: "price", Op: "gt", Value: "100"}}
data := map[string]interface{}{"price": float64(200)}
if !Evaluate(conds, data) {
t.Error("200 > 100, should pass")
}
}
func TestEvaluate_Fail(t *testing.T) {
conds := []model.Condition{
{Field: "symbol", Op: "exists"},
{Field: "price", Op: "lt", Value: "100"},
}
data := map[string]interface{}{"symbol": "BTC", "price": float64(200)}
if Evaluate(conds, data) {
t.Error("200 < 100 is false, should fail")
}
}
func TestEvaluate_Contains(t *testing.T) {
conds := []model.Condition{{Field: "msg", Op: "contains", Value: "error"}}
data := map[string]interface{}{"msg": "connection error occurred"}
if !Evaluate(conds, data) {
t.Error("msg contains 'error', should pass")
}
}
```
- [ ] **Step 3: Run tests**
```bash
go test ./internal/condition/... -v
```
Expected: all tests PASS.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: condition evaluator for rule filtering"
```
---
### Task 9: Channel Adapters — Interface + HTTP-based (DingTalk, WeCom, Bark)
**Files:**
- Create: `internal/adapter/adapter.go`
- Create: `internal/adapter/dingtalk.go`
- Create: `internal/adapter/wecom.go`
- Create: `internal/adapter/bark.go`
**Interfaces:**
- Produces: `adapter.ChannelSender` interface, `adapter.NewSender(channelType string, smtpCfg *config.SMTPConfig) (ChannelSender, error)`
- [ ] **Step 1: Write ChannelSender interface + factory**
File: `internal/adapter/adapter.go`
```go
package adapter
import (
"encoding/json"
"fmt"
"aiaa-notification-service/internal/config"
)
// ChannelSender sends a rendered message to a specific channel.
type ChannelSender interface {
Type() string
Send(title, content string, config json.RawMessage) error
}
func NewSender(channelType string, smtpCfg *config.SMTPConfig) (ChannelSender, error) {
switch channelType {
case "dingtalk":
return &DingTalkSender{}, nil
case "wecom":
return &WeComSender{}, nil
case "email":
if smtpCfg == nil {
return nil, fmt.Errorf("SMTP config required for email sender")
}
return NewEmailSender(*smtpCfg), nil
case "bark":
return &BarkSender{}, nil
default:
return nil, fmt.Errorf("unknown channel type: %s", channelType)
}
}
```
- [ ] **Step 2: Write DingTalk sender**
File: `internal/adapter/dingtalk.go`
```go
package adapter
import (
"bytes"
"crypto/hmac"
"crypto/sha256"
"encoding/base64"
"encoding/json"
"fmt"
"net/http"
"net/url"
"time"
)
type dingtalkConfig struct {
WebhookURL string `json:"webhook_url"`
Secret string `json:"secret,omitempty"`
}
type dingtalkMessage struct {
MsgType string `json:"msgtype"`
Markdown *dingtalkMD `json:"markdown"`
}
type dingtalkMD struct {
Title string `json:"title"`
Text string `json:"text"`
}
type DingTalkSender struct{}
func (s *DingTalkSender) Type() string { return "dingtalk" }
func (s *DingTalkSender) Send(title, content string, config json.RawMessage) error {
var cfg dingtalkConfig
if err := json.Unmarshal(config, &cfg); err != nil {
return fmt.Errorf("parse dingtalk config: %w", err)
}
reqURL := cfg.WebhookURL
if cfg.Secret != "" {
timestamp := time.Now().UnixMilli()
sign := dingtalkSign(timestamp, cfg.Secret)
reqURL = fmt.Sprintf("%s&timestamp=%d&sign=%s", cfg.WebhookURL, timestamp, sign)
}
msg := dingtalkMessage{
MsgType: "markdown",
Markdown: &dingtalkMD{
Title: title,
Text: content,
},
}
body, _ := json.Marshal(msg)
resp, err := http.Post(reqURL, "application/json", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("dingtalk send: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return fmt.Errorf("dingtalk returned status %d", resp.StatusCode)
}
return nil
}
func dingtalkSign(timestamp int64, secret string) string {
mac := hmac.New(sha256.New, []byte(secret))
fmt.Fprintf(mac, "%d\n%s", timestamp, secret)
signData := mac.Sum(nil)
return url.QueryEscape(base64.StdEncoding.EncodeToString(signData))
}
```
- [ ] **Step 3: Write WeCom (企业微信) sender**
File: `internal/adapter/wecom.go`
```go
package adapter
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
)
type wecomConfig struct {
WebhookURL string `json:"webhook_url"`
}
type wecomMessage struct {
MsgType string `json:"msgtype"`
Markdown *wecomMD `json:"markdown"`
}
type wecomMD struct {
Content string `json:"content"`
}
type WeComSender struct{}
func (s *WeComSender) Type() string { return "wecom" }
func (s *WeComSender) Send(title, content string, config json.RawMessage) error {
var cfg wecomConfig
if err := json.Unmarshal(config, &cfg); err != nil {
return fmt.Errorf("parse wecom config: %w", err)
}
msg := wecomMessage{
MsgType: "markdown",
Markdown: &wecomMD{
Content: fmt.Sprintf("## %s\n%s", title, content),
},
}
body, _ := json.Marshal(msg)
resp, err := http.Post(cfg.WebhookURL, "application/json", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("wecom send: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return fmt.Errorf("wecom returned status %d", resp.StatusCode)
}
return nil
}
```
- [ ] **Step 4: Write Bark sender**
File: `internal/adapter/bark.go`
```go
package adapter
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
)
type barkConfig struct {
URL string `json:"url"`
}
type barkPayload struct {
Title string `json:"title"`
Body string `json:"body"`
}
type BarkSender struct{}
func (s *BarkSender) Type() string { return "bark" }
func (s *BarkSender) Send(title, content string, config json.RawMessage) error {
var cfg barkConfig
if err := json.Unmarshal(config, &cfg); err != nil {
return fmt.Errorf("parse bark config: %w", err)
}
payload := barkPayload{Title: title, Body: content}
body, _ := json.Marshal(payload)
resp, err := http.Post(cfg.URL, "application/json", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("bark send: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode >= 400 {
return fmt.Errorf("bark returned status %d", resp.StatusCode)
}
return nil
}
```
- [ ] **Step 5: Verify compilation**
```bash
go build ./internal/adapter/...
```
Expected: no errors.
- [ ] **Step 6: Commit**
```bash
git add -A && git commit -m "feat: channel adapters — dingtalk, wecom, bark"
```
---
### Task 10: Email Adapter (SMTP)
**Files:**
- Create: `internal/adapter/email.go`
**Interfaces:**
- Produces: `EmailSender` implementing `ChannelSender`
- [ ] **Step 1: Write Email sender**
File: `internal/adapter/email.go`
```go
package adapter
import (
"crypto/tls"
"encoding/json"
"fmt"
"net"
"net/smtp"
"strings"
"aiaa-notification-service/internal/config"
)
type emailConfig struct {
To []string `json:"to"`
}
type EmailSender struct {
cfg config.SMTPConfig
}
func NewEmailSender(cfg config.SMTPConfig) *EmailSender {
return &EmailSender{cfg: cfg}
}
func (s *EmailSender) Type() string { return "email" }
func (s *EmailSender) Send(title, content string, config json.RawMessage) error {
var ecfg emailConfig
if err := json.Unmarshal(config, &ecfg); err != nil {
return fmt.Errorf("parse email config: %w", err)
}
if len(ecfg.To) == 0 {
return fmt.Errorf("email: 'to' list is empty")
}
msg := buildEmail(s.cfg.From, ecfg.To, title, content)
addr := fmt.Sprintf("%s:%d", s.cfg.Host, s.cfg.Port)
if s.cfg.Port == 465 {
return s.sendTLS(addr, ecfg.To, msg)
}
return s.sendSTARTTLS(addr, ecfg.To, msg)
}
func (s *EmailSender) sendTLS(addr string, to []string, msg []byte) error {
tlsCfg := &tls.Config{ServerName: s.cfg.Host}
conn, err := tls.Dial("tcp", addr, tlsCfg)
if err != nil {
return fmt.Errorf("tls dial: %w", err)
}
defer conn.Close()
client, err := smtp.NewClient(conn, s.cfg.Host)
if err != nil {
return fmt.Errorf("smtp client: %w", err)
}
defer client.Quit()
return s.authAndSend(client, to, msg)
}
func (s *EmailSender) sendSTARTTLS(addr string, to []string, msg []byte) error {
conn, err := net.Dial("tcp", addr)
if err != nil {
return fmt.Errorf("dial: %w", err)
}
defer conn.Close()
client, err := smtp.NewClient(conn, s.cfg.Host)
if err != nil {
return fmt.Errorf("smtp client: %w", err)
}
defer client.Quit()
if ok, _ := client.Extension("STARTTLS"); ok {
tlsCfg := &tls.Config{ServerName: s.cfg.Host}
if err := client.StartTLS(tlsCfg); err != nil {
return fmt.Errorf("starttls: %w", err)
}
}
return s.authAndSend(client, to, msg)
}
func (s *EmailSender) authAndSend(client *smtp.Client, to []string, msg []byte) error {
if s.cfg.User != "" {
auth := smtp.PlainAuth("", s.cfg.User, s.cfg.Password, s.cfg.Host)
if err := client.Auth(auth); err != nil {
return fmt.Errorf("auth: %w", err)
}
}
if err := client.Mail(s.cfg.From); err != nil {
return fmt.Errorf("mail: %w", err)
}
for _, recipient := range to {
if err := client.Rcpt(recipient); err != nil {
return fmt.Errorf("rcpt %s: %w", recipient, err)
}
}
w, err := client.Data()
if err != nil {
return fmt.Errorf("data: %w", err)
}
_, err = w.Write(msg)
if err != nil {
return fmt.Errorf("write: %w", err)
}
return w.Close()
}
func buildEmail(from string, to []string, subject, body string) []byte {
var sb strings.Builder
sb.WriteString(fmt.Sprintf("From: %s\r\n", from))
sb.WriteString(fmt.Sprintf("To: %s\r\n", strings.Join(to, ", ")))
sb.WriteString(fmt.Sprintf("Subject: %s\r\n", subject))
sb.WriteString("MIME-Version: 1.0\r\n")
sb.WriteString("Content-Type: text/plain; charset=utf-8\r\n")
sb.WriteString("\r\n")
sb.WriteString(body)
return []byte(sb.String())
}
```
Note: fix `s.cfg` references — they should be `s.cfg`.
- [ ] **Step 2: Verify compilation**
```bash
go build ./internal/adapter/...
```
Expected: no errors.
- [ ] **Step 3: Commit**
```bash
git add -A && git commit -m "feat: email adapter with SMTP TLS support"
```
---
### Task 11: Engine — Renderer, Matcher, Router
**Files:**
- Create: `internal/engine/renderer.go`
- Create: `internal/engine/matcher.go`
- Create: `internal/engine/router.go`
**Interfaces:**
- Produces: `engine.NewRenderer() *Renderer`, `Renderer.Render(templateContent string, data map[string]interface{}) (string, error)`
- Produces: `engine.NewMatcher(store *store.Store, cache *cache.Cache) *Matcher`, `Matcher.Match(ctx, sourceID, event) (*model.Rule, error)`
- Produces: `engine.NewRouter(store *store.Store, cache *cache.Cache, senderFactory func(string) (adapter.ChannelSender, error)) *Router`, `Router.Route(ctx, rule *model.Rule, content string) []SendResult`
- [ ] **Step 1: Write Template Renderer**
File: `internal/engine/renderer.go`
```go
package engine
import (
"bytes"
"fmt"
"text/template"
)
type Renderer struct{}
func NewRenderer() *Renderer {
return &Renderer{}
}
func (r *Renderer) Render(tmplContent string, data map[string]interface{}) (string, error) {
tmpl, err := template.New("notify").Parse(tmplContent)
if err != nil {
return "", fmt.Errorf("parse template: %w", err)
}
var buf bytes.Buffer
if err := tmpl.Execute(&buf, data); err != nil {
return "", fmt.Errorf("execute template: %w", err)
}
return buf.String(), nil
}
```
- [ ] **Step 2: Write Rule Matcher**
File: `internal/engine/matcher.go`
```go
package engine
import (
"context"
"encoding/json"
"fmt"
"aiaa-notification-service/internal/cache"
"aiaa-notification-service/internal/model"
"aiaa-notification-service/internal/store"
)
type Matcher struct {
store *store.Store
cache *cache.Cache
}
func NewMatcher(s *store.Store, c *cache.Cache) *Matcher {
return &Matcher{store: s, cache: c}
}
func (m *Matcher) Match(ctx context.Context, sourceID int, event string) (*model.Rule, error) {
// Try cache first
if m.cache != nil {
cr, err := m.cache.GetRule(ctx, sourceID, event)
if err == nil {
rule := &model.Rule{ID: cr.RuleID, TemplateID: cr.TemplateID, SourceID: sourceID, Event: event}
if cr.Conditions != "" && cr.Conditions != "null" {
raw := json.RawMessage(cr.Conditions)
rule.Conditions = &raw
}
return rule, nil
}
}
// Fall back to DB
rule, err := m.store.GetRuleBySourceEvent(ctx, sourceID, event)
if err != nil {
return nil, fmt.Errorf("match rule: %w", err)
}
// Warm cache
if m.cache != nil {
tmpl, err := m.store.GetTemplate(ctx, rule.TemplateID)
if err != nil {
return rule, nil // rule found but template fetch failed — still return rule
}
cr := &cache.CachedRule{
RuleID: rule.ID,
TemplateID: rule.TemplateID,
Content: tmpl.Content,
}
if rule.Conditions != nil {
cr.Conditions = string(*rule.Conditions)
}
_ = m.cache.SetRule(ctx, sourceID, event, cr)
}
return rule, nil
}
```
- [ ] **Step 3: Write Router**
File: `internal/engine/router.go`
```go
package engine
import (
"context"
"fmt"
"log/slog"
"aiaa-notification-service/internal/adapter"
"aiaa-notification-service/internal/cache"
"aiaa-notification-service/internal/model"
"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
senderFactory func(channelType string) (adapter.ChannelSender, error)
}
func NewRouter(s *store.Store, c *cache.Cache, sf func(channelType string) (adapter.ChannelSender, error)) *Router {
return &Router{store: s, cache: c, senderFactory: sf}
}
// Route sends content to all enabled channels for the rule. Returns immediately, sends async.
func (r *Router) Route(ctx context.Context, rule *model.Rule, title, content string) []string {
channels, err := r.getChannels(ctx, rule.ID)
if err != nil {
slog.Error("get channels for rule", "rule_id", rule.ID, "error", err)
return nil
}
channelNames := make([]string, 0, len(channels))
for _, ch := range channels {
channelNames = append(channelNames, fmt.Sprintf("%s:%d", ch.Type, ch.ID))
sender, err := r.senderFactory(ch.Type)
if err != nil {
slog.Error("create sender", "type", ch.Type, "error", err)
continue
}
go func(ch cache.CachedChannel, s adapter.ChannelSender) {
var cfg json.RawMessage
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)
} else {
slog.Info("sent", "channel_type", ch.Type, "channel_id", ch.ID)
}
}(ch, sender)
}
return channelNames
}
func (r *Router) getChannels(ctx context.Context, ruleID int) ([]cache.CachedChannel, error) {
if r.cache != nil {
chs, err := r.cache.GetChannels(ctx, ruleID)
if err == nil {
return chs, nil
}
}
rcs, err := r.store.GetRuleChannels(ctx, ruleID)
if err != nil {
return nil, err
}
var cached []cache.CachedChannel
for _, rc := range rcs {
ch, err := r.store.GetChannel(ctx, rc.ChannelID)
if err != nil {
continue
}
cc := cache.CachedChannel{ID: ch.ID, Type: ch.Type, Config: ch.Config}
cached = append(cached, cc)
}
if r.cache != nil {
_ = r.cache.SetChannels(ctx, ruleID, cached)
}
return cached, nil
}
```
Note: need `encoding/json` import in router.go.
- [ ] **Step 4: Write tests for renderer**
File: `internal/engine/renderer_test.go`
```go
package engine
import (
"strings"
"testing"
)
func TestRenderer(t *testing.T) {
r := NewRenderer()
tmpl := "🚀 {{.symbol}} 开仓通知\n价格: {{.price}}"
data := map[string]interface{}{"symbol": "BTC", "price": 65000}
result, err := r.Render(tmpl, data)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if !strings.Contains(result, "BTC") || !strings.Contains(result, "65000") {
t.Errorf("unexpected output: %s", result)
}
}
func TestRenderer_Error(t *testing.T) {
r := NewRenderer()
_, err := r.Render("{{.nonexistent}}", map[string]interface{}{})
if err == nil {
t.Error("expected error for missing field, got nil")
}
}
```
- [ ] **Step 5: Run tests**
```bash
go test ./internal/engine/... -v
```
Expected: renderer tests PASS.
- [ ] **Step 6: Commit**
```bash
git add -A && git commit -m "feat: engine — template renderer, rule matcher, channel router"
```
---
### Task 12: HTTP Middleware (Auth + Rate Limit)
**Files:**
- Create: `internal/handler/middleware.go`
**Interfaces:**
- Produces: `handler.AdminAuth(adminKey string) gin.HandlerFunc`, `handler.SourceAuth(store *store.Store) gin.HandlerFunc`, `handler.RateLimit(cache *cache.Cache, defaultLimit int) gin.HandlerFunc`
- [ ] **Step 1: Write middleware**
File: `internal/handler/middleware.go`
```go
package handler
import (
"net/http"
"strings"
"aiaa-notification-service/internal/cache"
"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]
}
```
Note: `rateLimit` needs the `model` import for `source.(*model.Source)`. Add:
```go
import (
"aiaa-notification-service/internal/model"
)
```
- [ ] **Step 2: Verify compilation**
```bash
go get github.com/gin-gonic/gin
go build ./internal/handler/...
```
Expected: no errors.
- [ ] **Step 3: Commit**
```bash
git add -A && git commit -m "feat: auth and rate limit middleware"
```
---
### Task 13: Notify Handler (POST /api/v1/notify)
**Files:**
- Create: `internal/handler/notify.go`
**Interfaces:**
- Consumes: `*store.Store`, `*cache.Cache`, `*engine.Matcher`, `*engine.Renderer`, `*engine.Router`, `*condition.Evaluator`
- Produces: `POST /api/v1/notify` endpoint
- [ ] **Step 1: Write notify handler**
File: `internal/handler/notify.go`
```go
package handler
import (
"encoding/json"
"io"
"log/slog"
"net/http"
"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 {
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)
go func() {
payloadJSON, _ := json.Marshal(msg.Data)
for _, chName := range channels {
ml := &model.MessageLog{
RuleID: rule.ID,
Source: src.Name,
Event: msg.Event,
Payload: payloadJSON,
Content: content,
Status: "pending",
}
_ = h.store.CreateMessageLog(c.Request.Context(), ml)
}
}()
slog.Info("notification accepted",
"source", src.Name,
"event", msg.Event,
"channels", channels,
)
c.JSON(http.StatusOK, gin.H{
"matched": true,
"channels": channels,
"accepted": true,
})
}
```
Note: need imports for `cache` and `engine` — add:
```go
import (
"aiaa-notification-service/internal/cache"
)
```
- [ ] **Step 2: Verify compilation**
```bash
go build ./internal/handler/...
```
Expected: no errors.
- [ ] **Step 3: Commit**
```bash
git add -A && git commit -m "feat: notify handler — parse, match, evaluate, render, route"
```
---
### Task 14: Management Handlers — Source, Template, Channel CRUD
**Files:**
- Create: `internal/handler/source.go`
- Create: `internal/handler/template.go`
- Create: `internal/handler/channel.go`
**Interfaces:**
- Consumes: `*store.Store`, `*cache.Cache`
- Produces: Source/Template/Channel CRUD endpoints
- [ ] **Step 1: Write Source CRUD handler**
File: `internal/handler/source.go`
```go
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})
}
```
- [ ] **Step 2: Write Template CRUD handler**
File: `internal/handler/template.go`
```go
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})
}
```
- [ ] **Step 3: Write Channel CRUD handler**
File: `internal/handler/channel.go`
```go
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})
}
```
- [ ] **Step 4: Verify compilation**
```bash
go build ./internal/handler/...
```
Expected: no errors.
- [ ] **Step 5: Commit**
```bash
git add -A && git commit -m "feat: management handlers — source, template, channel CRUD"
```
---
### Task 15: Management Handlers — Rule CRUD + MessageLog
**Files:**
- Create: `internal/handler/rule.go`
- Create: `internal/handler/message_log.go`
**Interfaces:**
- Produces: Rule CRUD + enable/disable/channel-toggle endpoints
- Produces: GET message-logs with pagination
- [ ] **Step 1: Write Rule CRUD handler**
File: `internal/handler/rule.go`
```go
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
}
```
- [ ] **Step 2: Write MessageLog handler**
File: `internal/handler/message_log.go`
```go
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,
})
}
```
- [ ] **Step 3: Verify compilation**
```bash
go build ./internal/handler/...
```
Expected: no errors.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: rule CRUD with enable/disable and message log query"
```
---
### Task 16: Retry Mechanism
**Files:**
- Create: `internal/retry/retry.go`
**Interfaces:**
- Produces: `retry.NewRetrier(maxRetries int, backoff []time.Duration) *Retrier`, `Retrier.Do(ctx context.Context, fn func() error) error`
- [ ] **Step 1: Write retry module**
File: `internal/retry/retry.go`
```go
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)
}
```
- [ ] **Step 2: Verify compilation**
```bash
go build ./internal/retry/...
```
Expected: no errors.
- [ ] **Step 3: Commit**
```bash
git add -A && git commit -m "feat: retry mechanism with exponential backoff"
```
---
### Task 17: Main Assembly — Wire Everything Together
**Files:**
- Modify: `cmd/server/main.go`
- Modify: `internal/handler/notify.go` (fix cache import)
**Interfaces:**
- Produces: fully functional HTTP server
- [ ] **Step 1: Write complete main.go**
File: `cmd/server/main.go`
```go
package main
import (
"context"
"fmt"
"log/slog"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"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"
)
func main() {
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
slog.SetDefault(logger)
// Load config
cfg, err := config.Load("config/config.yaml")
if err != nil {
slog.Error("failed to load config", "error", err)
os.Exit(1)
}
// Connect MySQL
store, err := store.NewStore(cfg.Database)
if err != nil {
slog.Error("failed to connect to MySQL", "error", err)
os.Exit(1)
}
defer store.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(store, redisCache)
renderer := engine.NewRenderer()
// Build sender factory
senderFactory := func(channelType string) (adapter.ChannelSender, error) {
return adapter.NewSender(channelType, &cfg.SMTP)
}
router := engine.NewRouter(store, redisCache, senderFactory)
// Build handlers
notifyH := handler.NewNotifyHandler(store, redisCache, matcher, renderer, router)
sourceH := handler.NewSourceHandler(store, redisCache)
templateH := handler.NewTemplateHandler(store, redisCache)
channelH := handler.NewChannelHandler(store, redisCache)
ruleH := handler.NewRuleHandler(store, redisCache)
msgLogH := handler.NewMessageLogHandler(store)
// 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(store))
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")
}
```
- [ ] **Step 2: Fix imports — add adapter import to main.go**
Add to imports:
```go
import (
"aiaa-notification-service/internal/adapter"
)
```
- [ ] **Step 3: Verify full build**
```bash
go mod tidy
go build ./cmd/server/...
```
Expected: builds successfully. Fix any import or type issues.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: main assembly — wire all components, gin router, graceful shutdown"
```
---
### Task 18: Docker & Docker Compose
**Files:**
- Create: `Dockerfile`
- Create: `docker-compose.yml`
- [ ] **Step 1: Write Dockerfile**
File: `Dockerfile`
```dockerfile
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 /etc/notification/config.yaml
EXPOSE 8080
ENTRYPOINT ["server"]
```
- [ ] **Step 2: Write docker-compose.yml**
File: `docker-compose.yml`
```yaml
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=
depends_on:
mysql:
condition: service_healthy
redis:
condition: service_healthy
restart: unless-stopped
volumes:
mysql_data:
```
- [ ] **Step 3: Update config for Docker**
The `config/config.yaml` should use Docker service names for hosts:
```yaml
database:
host: "mysql"
port: 3306
# ...
redis:
host: "redis"
port: 6379
# ...
```
Leave the existing config.yaml as-is (for local dev) and the docker-compose env vars will override as needed.
- [ ] **Step 4: Commit**
```bash
git add -A && git commit -m "feat: docker multi-stage build and docker-compose"
```
---
## Final Verification
After all tasks complete, run the full test suite and build:
```bash
# Tests
go test ./internal/... -v
# Build
go build -o bin/server ./cmd/server
# Start with docker-compose
docker-compose up -d
# Run migrations (inside container or locally)
# Then test with curl:
# Create a source
curl -X POST http://localhost:8080/api/v1/sources \
-H "Authorization: Bearer admin-sk-change-me" \
-H "Content-Type: application/json" \
-d '{"name":"test-system","parse_mode":"json"}'
# Create a template
curl -X POST http://localhost:8080/api/v1/templates \
-H "Authorization: Bearer admin-sk-change-me" \
-H "Content-Type: application/json" \
-d '{"name":"test_tmpl","content":"🚀 {{.symbol}} {{.price}}"}'
# Create a channel
curl -X POST http://localhost:8080/api/v1/channels \
-H "Authorization: Bearer admin-sk-change-me" \
-H "Content-Type: application/json" \
-d '{"name":"test-bark","type":"bark","config":{"url":"https://api.day.app/test"}}'
# Create a rule
curl -X POST http://localhost:8080/api/v1/rules \
-H "Authorization: Bearer admin-sk-change-me" \
-H "Content-Type: application/json" \
-d '{"source_name":"test-system","event":"trade.open","template_name":"test_tmpl","channels":["test-bark"]}'
# Send a notification (use the api_key from source creation response)
curl -X POST http://localhost:8080/api/v1/notify \
-H "Authorization: Bearer <SOURCE_API_KEY>" \
-H "Content-Type: application/json" \
-d '{"event":"trade.open","data":{"symbol":"BTC","price":65000}}'
```
---
Reference in New Issue View Git Blame Copy Permalink