feat: 配置列表分页与钉钉机器人分钟级排队限流
统一 sources/templates/channels/rules 列表为分页响应，避免配置增多时全量返回；按钉钉 access_token 限制每分钟发送并在超限时等待下一分钟，降低触发官方封禁风险。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
.PHONY: build run test docker-build docker-up docker-down migrate-up migrate-down
|
||||
.PHONY: build run test test-e2e docker-build docker-up docker-down migrate-up migrate-down
|
||||
|
||||
build:
|
||||
go build -o bin/server ./cmd/server
|
||||
@@ -9,6 +9,9 @@ run:
|
||||
test:
|
||||
go test ./internal/... -v -count=1
|
||||
|
||||
test-e2e:
|
||||
go test -tags e2e ./test/e2e/ -v -count=1 -timeout 2m
|
||||
|
||||
docker-build:
|
||||
docker build -t notification-service .
|
||||
|
||||
|
||||
@@ -94,6 +94,7 @@ make build && ./bin/server
|
||||
| `redis.*` | Redis（缓存 + 限流） | `127.0.0.1:6379` |
|
||||
| `smtp.*` | 邮件发送（email 渠道） | — |
|
||||
| `rate_limit.default` | 每 source 每秒请求上限 | `100` |
|
||||
| `rate_limit.dingtalk_per_min` | 同一钉钉机器人（access_token）每分钟发送上限；超限排队到下一分钟 | `18`（官方 20，留余量） |
|
||||
|
||||
健康检查：`GET /health` → `{"status":"ok"}`
|
||||
|
||||
@@ -265,7 +266,13 @@ Regex 示例：
|
||||
|
||||
#### `GET /api/v1/sources` — 列表
|
||||
|
||||
**200** → `Source[]`
|
||||
Query：`page`（默认 1）、`page_size`（默认 20）
|
||||
|
||||
**200：**
|
||||
|
||||
```json
|
||||
{ "data": [ /* Source[] */ ], "total": 10, "page": 1 }
|
||||
```
|
||||
|
||||
#### `GET /api/v1/sources/:id` — 详情
|
||||
|
||||
@@ -313,7 +320,11 @@ Text 模式示例：`{{.Body}}`
|
||||
|
||||
**201** → Template；冲突 → **409**
|
||||
|
||||
#### `GET /api/v1/templates` / `GET /api/v1/templates/:id`
|
||||
#### `GET /api/v1/templates` — 列表
|
||||
|
||||
Query：`page`、`page_size`（默认同 sources）。**200：** `{ "data": Template[], "total", "page" }`
|
||||
|
||||
#### `GET /api/v1/templates/:id`
|
||||
|
||||
#### `PUT /api/v1/templates/:id` / `DELETE /api/v1/templates/:id`
|
||||
|
||||
@@ -387,7 +398,11 @@ SMTP 使用全局 `config.yaml` 的 `smtp` 段；支持 587 STARTTLS / 465 TLS
|
||||
}
|
||||
```
|
||||
|
||||
#### `GET /api/v1/channels` / `GET /api/v1/channels/:id`
|
||||
#### `GET /api/v1/channels` — 列表
|
||||
|
||||
Query：`page`、`page_size`。**200：** `{ "data": Channel[], "total", "page" }`
|
||||
|
||||
#### `GET /api/v1/channels/:id`
|
||||
|
||||
#### `PUT /api/v1/channels/:id` / `DELETE /api/v1/channels/:id`
|
||||
|
||||
@@ -438,7 +453,11 @@ SMTP 使用全局 `config.yaml` 的 `smtp` 段；支持 587 STARTTLS / 465 TLS
|
||||
|
||||
**201** → Rule；冲突 → **409**
|
||||
|
||||
#### `GET /api/v1/rules` / `GET /api/v1/rules/:id`
|
||||
#### `GET /api/v1/rules` — 列表
|
||||
|
||||
Query：`page`、`page_size`。**200：** `{ "data": Rule[], "total", "page" }`
|
||||
|
||||
#### `GET /api/v1/rules/:id`
|
||||
|
||||
#### `PUT /api/v1/rules/:id` — 更新
|
||||
|
||||
|
||||
+11
-1
@@ -105,9 +105,19 @@ func main() {
|
||||
matcher := engine.NewMatcher(st, redisCache)
|
||||
renderer := engine.NewRenderer()
|
||||
|
||||
// DingTalk per-robot rate limit (queue/wait when over 18/min by default)
|
||||
var dingtalkLimiter *adapter.DingTalkLimiter
|
||||
limitPerMin := cfg.RateLimit.DingTalkPerMin
|
||||
if redisCache != nil {
|
||||
dingtalkLimiter = adapter.NewDingTalkLimiter(redisCache, limitPerMin)
|
||||
} else {
|
||||
dingtalkLimiter = adapter.NewMemoryDingTalkLimiter(limitPerMin)
|
||||
slog.Warn("dingtalk rate limit using in-memory store (single instance only)")
|
||||
}
|
||||
|
||||
// Build sender factory
|
||||
senderFactory := func(channelType string) (adapter.ChannelSender, error) {
|
||||
return adapter.NewSender(channelType, &cfg.SMTP)
|
||||
return adapter.NewSender(channelType, &cfg.SMTP, dingtalkLimiter)
|
||||
}
|
||||
router := engine.NewRouter(st, redisCache, senderFactory)
|
||||
|
||||
|
||||
@@ -25,6 +25,8 @@ smtp:
|
||||
|
||||
rate_limit:
|
||||
default: 100
|
||||
# 钉钉自定义机器人官方上限 20/分钟，超限封 10 分钟；本地默认 18 留余量
|
||||
dingtalk_per_min: 18
|
||||
|
||||
logbull:
|
||||
host: "https://log.516886.xyz"
|
||||
|
||||
@@ -111,7 +111,7 @@ curl -X POST 'http://localhost:8080/api/v1/sources' \
|
||||
### 列出 Source
|
||||
|
||||
```bash
|
||||
curl -X GET 'http://localhost:8080/api/v1/sources' \
|
||||
curl -X GET 'http://localhost:8080/api/v1/sources?page=1&page_size=20' \
|
||||
-H 'Authorization: Bearer admin-sk-change-me'
|
||||
```
|
||||
|
||||
@@ -161,7 +161,7 @@ curl -X POST 'http://localhost:8080/api/v1/templates' \
|
||||
### 列出 Template
|
||||
|
||||
```bash
|
||||
curl -X GET 'http://localhost:8080/api/v1/templates' \
|
||||
curl -X GET 'http://localhost:8080/api/v1/templates?page=1&page_size=20' \
|
||||
-H 'Authorization: Bearer admin-sk-change-me'
|
||||
```
|
||||
|
||||
@@ -263,7 +263,7 @@ curl -X POST 'http://localhost:8080/api/v1/channels' \
|
||||
### 列出 Channel
|
||||
|
||||
```bash
|
||||
curl -X GET 'http://localhost:8080/api/v1/channels' \
|
||||
curl -X GET 'http://localhost:8080/api/v1/channels?page=1&page_size=20' \
|
||||
-H 'Authorization: Bearer admin-sk-change-me'
|
||||
```
|
||||
|
||||
@@ -323,7 +323,7 @@ curl -X POST 'http://localhost:8080/api/v1/rules' \
|
||||
### 列出 Rule
|
||||
|
||||
```bash
|
||||
curl -X GET 'http://localhost:8080/api/v1/rules' \
|
||||
curl -X GET 'http://localhost:8080/api/v1/rules?page=1&page_size=20' \
|
||||
-H 'Authorization: Bearer admin-sk-change-me'
|
||||
```
|
||||
|
||||
|
||||
@@ -0,0 +1,294 @@
|
||||
# Remote E2E Notify Flow 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:** Add a Go e2e test (build tag `e2e`) that hits the deployed service and verifies source → channel×3 → template → rule → notify.
|
||||
|
||||
**Architecture:** Stdlib `net/http` client against `E2E_BASE_URL`. Admin Bearer for CRUD; source `api_key` for notify. Channel secrets from env. Unique resource names + `t.Cleanup` deletes.
|
||||
|
||||
**Tech Stack:** Go 1.22+, `testing`, `net/http`, `encoding/json`. No new dependencies.
|
||||
|
||||
## Global Constraints
|
||||
|
||||
- Build tag: `e2e` — must not run under default `make test`
|
||||
- Secrets only via env vars — never hardcode webhooks/secrets in source
|
||||
- Default base URL: `http://82.157.251.93:8080`
|
||||
- Default admin key: `admin-sk-change-me`
|
||||
- Channels: dingtalk + bark + email
|
||||
- Assert notify `matched`/`accepted` and 3 channels; poll message-logs up to 15s
|
||||
- Do not commit unless the user explicitly asks
|
||||
|
||||
---
|
||||
|
||||
## File Structure
|
||||
|
||||
| File | Responsibility |
|
||||
|------|----------------|
|
||||
| `test/e2e/notify_flow_test.go` | Full remote flow test + tiny HTTP helpers |
|
||||
| `Makefile` | Add `test-e2e` target |
|
||||
|
||||
---
|
||||
|
||||
### Task 1: E2E notify flow test + Makefile
|
||||
|
||||
**Files:**
|
||||
- Create: `test/e2e/notify_flow_test.go`
|
||||
- Modify: `Makefile`
|
||||
|
||||
**Interfaces:**
|
||||
- Produces: `TestNotifyFlow_SourceChannelTemplateRuleSend` runnable via `go test -tags e2e ./test/e2e/`
|
||||
|
||||
- [x] **Step 1: Create `test/e2e/notify_flow_test.go`**
|
||||
|
||||
```go
|
||||
//go:build e2e
|
||||
|
||||
package e2e
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestNotifyFlow_SourceChannelTemplateRuleSend(t *testing.T) {
|
||||
base := envOr("E2E_BASE_URL", "http://82.157.251.93:8080")
|
||||
adminKey := envOr("E2E_ADMIN_KEY", "admin-sk-change-me")
|
||||
dingWebhook := os.Getenv("E2E_DINGTALK_WEBHOOK")
|
||||
dingSecret := os.Getenv("E2E_DINGTALK_SECRET")
|
||||
barkURL := os.Getenv("E2E_BARK_URL")
|
||||
emailTo := os.Getenv("E2E_EMAIL_TO")
|
||||
if dingWebhook == "" || dingSecret == "" || barkURL == "" || emailTo == "" {
|
||||
t.Skip("missing E2E_DINGTALK_WEBHOOK / E2E_DINGTALK_SECRET / E2E_BARK_URL / E2E_EMAIL_TO")
|
||||
}
|
||||
|
||||
client := &http.Client{Timeout: 30 * time.Second}
|
||||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
srcName := "e2e-src-" + suffix
|
||||
tmplName := "e2e-tmpl-" + suffix
|
||||
chDing := "e2e-ding-" + suffix
|
||||
chBark := "e2e-bark-" + suffix
|
||||
chEmail := "e2e-email-" + suffix
|
||||
|
||||
// 1. Source
|
||||
src := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/sources", map[string]any{
|
||||
"name": srcName, "parse_mode": "json", "status": 1,
|
||||
}, http.StatusCreated)
|
||||
srcID := intFrom(src["id"])
|
||||
apiKey, _ := src["api_key"].(string)
|
||||
if apiKey == "" {
|
||||
t.Fatal("source api_key empty")
|
||||
}
|
||||
t.Cleanup(func() { _ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/sources/%d", srcID), nil) })
|
||||
|
||||
// 2. Channels
|
||||
ding := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/channels", map[string]any{
|
||||
"name": chDing, "type": "dingtalk", "status": 1,
|
||||
"config": map[string]string{"webhook_url": dingWebhook, "secret": dingSecret},
|
||||
}, http.StatusCreated)
|
||||
dingID := intFrom(ding["id"])
|
||||
t.Cleanup(func() { _ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/channels/%d", dingID), nil) })
|
||||
|
||||
bark := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/channels", map[string]any{
|
||||
"name": chBark, "type": "bark", "status": 1,
|
||||
"config": map[string]string{"url": barkURL},
|
||||
}, http.StatusCreated)
|
||||
barkID := intFrom(bark["id"])
|
||||
t.Cleanup(func() { _ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/channels/%d", barkID), nil) })
|
||||
|
||||
email := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/channels", map[string]any{
|
||||
"name": chEmail, "type": "email", "status": 1,
|
||||
"config": map[string]any{"to": []string{emailTo}},
|
||||
}, http.StatusCreated)
|
||||
emailID := intFrom(email["id"])
|
||||
t.Cleanup(func() { _ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/channels/%d", emailID), nil) })
|
||||
|
||||
// 3. Template
|
||||
tmpl := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/templates", map[string]any{
|
||||
"name": tmplName,
|
||||
"content": "### {{.symbol}} 开仓\n价格: {{.price}}",
|
||||
}, http.StatusCreated)
|
||||
tmplID := intFrom(tmpl["id"])
|
||||
t.Cleanup(func() { _ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/templates/%d", tmplID), nil) })
|
||||
|
||||
// 4. Rule
|
||||
rule := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/rules", map[string]any{
|
||||
"source_name": srcName, "event": "trade.open", "template_name": tmplName,
|
||||
"channels": []string{chDing, chBark, chEmail},
|
||||
"conditions": []map[string]string{
|
||||
{"field": "symbol", "op": "exists"},
|
||||
{"field": "price", "op": "gt", "value": "0"},
|
||||
},
|
||||
"enabled": 1,
|
||||
}, http.StatusCreated)
|
||||
ruleID := intFrom(rule["id"])
|
||||
t.Cleanup(func() { _ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/rules/%d", ruleID), nil) })
|
||||
|
||||
// 5. Notify
|
||||
notifyBody := map[string]any{
|
||||
"event": "trade.open",
|
||||
"data": map[string]any{"symbol": "BTC", "price": 65000},
|
||||
}
|
||||
notifyResp := mustJSON(t, client, http.MethodPost, base+"/api/v1/notify", apiKey, notifyBody, http.StatusOK)
|
||||
if notifyResp["matched"] != true {
|
||||
t.Fatalf("matched: %#v", notifyResp["matched"])
|
||||
}
|
||||
if notifyResp["accepted"] != true {
|
||||
t.Fatalf("accepted: %#v", notifyResp["accepted"])
|
||||
}
|
||||
chs, _ := notifyResp["channels"].([]any)
|
||||
if len(chs) != 3 {
|
||||
t.Fatalf("want 3 channels, got %#v", notifyResp["channels"])
|
||||
}
|
||||
|
||||
// 6. Message logs (async create)
|
||||
deadline := time.Now().Add(15 * time.Second)
|
||||
var total float64
|
||||
for time.Now().Before(deadline) {
|
||||
logs := mustAdminJSON(t, client, base, adminKey, http.MethodGet,
|
||||
fmt.Sprintf("/api/v1/message-logs?source=%s&event=trade.open&page=1&page_size=20", srcName),
|
||||
nil, http.StatusOK)
|
||||
if n, ok := logs["total"].(float64); ok {
|
||||
total = n
|
||||
if total >= 1 {
|
||||
break
|
||||
}
|
||||
}
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
if total < 1 {
|
||||
t.Fatalf("expected message logs, total=%v", total)
|
||||
}
|
||||
}
|
||||
|
||||
func envOr(k, def string) string {
|
||||
if v := os.Getenv(k); v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func intFrom(v any) int {
|
||||
switch n := v.(type) {
|
||||
case float64:
|
||||
return int(n)
|
||||
case int:
|
||||
return n
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func mustAdminJSON(t *testing.T, c *http.Client, base, key, method, path string, body any, want int) map[string]any {
|
||||
t.Helper()
|
||||
return mustJSON(t, c, method, base+path, key, body, want)
|
||||
}
|
||||
|
||||
func mustJSON(t *testing.T, c *http.Client, method, url, bearer string, body any, want int) map[string]any {
|
||||
t.Helper()
|
||||
code, raw, err := doJSON(c, method, url, bearer, body)
|
||||
if err != nil {
|
||||
t.Fatalf("%s %s: %v", method, url, err)
|
||||
}
|
||||
if code != want {
|
||||
t.Fatalf("%s %s: status %d want %d body=%s", method, url, code, want, raw)
|
||||
}
|
||||
var out map[string]any
|
||||
if err := json.Unmarshal(raw, &out); err != nil {
|
||||
t.Fatalf("decode: %v body=%s", err, raw)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func adminDo(c *http.Client, base, key, method, path string, body any) error {
|
||||
_, _, err := doJSON(c, method, base+path, key, body)
|
||||
return err
|
||||
}
|
||||
|
||||
func doJSON(c *http.Client, method, url, bearer string, body any) (int, []byte, error) {
|
||||
var rdr io.Reader
|
||||
if body != nil {
|
||||
b, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
rdr = bytes.NewReader(b)
|
||||
}
|
||||
req, err := http.NewRequest(method, url, rdr)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
if bearer != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+bearer)
|
||||
}
|
||||
if body != nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
resp, err := c.Do(req)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
raw, err := io.ReadAll(resp.Body)
|
||||
return resp.StatusCode, raw, err
|
||||
}
|
||||
```
|
||||
|
||||
Note: register cleanups so deletes run reverse of registration order — register source first, then channels, template, rule last so rule deletes before dependents. (Go runs cleanups LIFO.)
|
||||
|
||||
- [x] **Step 2: Update Makefile**
|
||||
|
||||
Add:
|
||||
|
||||
```make
|
||||
test-e2e:
|
||||
go test -tags e2e ./test/e2e/ -v -count=1 -timeout 2m
|
||||
```
|
||||
|
||||
- [x] **Step 3: Run without env — expect Skip**
|
||||
|
||||
```bash
|
||||
go test -tags e2e ./test/e2e/ -v -count=1
|
||||
```
|
||||
|
||||
Expected: `SKIP: missing E2E_...`
|
||||
|
||||
- [x] **Step 4: Run against remote with env — expect PASS**
|
||||
|
||||
```bash
|
||||
E2E_DINGTALK_WEBHOOK='https://oapi.dingtalk.com/robot/send?access_token=...' \
|
||||
E2E_DINGTALK_SECRET='SEC...' \
|
||||
E2E_BARK_URL='https://api.day.app/...' \
|
||||
E2E_EMAIL_TO='149516886@qq.com' \
|
||||
go test -tags e2e ./test/e2e/ -v -count=1 -timeout 2m
|
||||
```
|
||||
|
||||
Expected: `PASS`. Verify devices/inbox received messages manually if needed.
|
||||
|
||||
- [x] **Step 5: Confirm default unit tests still ignore e2e**
|
||||
|
||||
```bash
|
||||
go test ./internal/... -count=1
|
||||
```
|
||||
|
||||
Expected: no e2e package compiled.
|
||||
|
||||
---
|
||||
|
||||
## Spec coverage
|
||||
|
||||
| Spec item | Task |
|
||||
|-----------|------|
|
||||
| Build tag e2e | Task 1 |
|
||||
| Env vars + skip | Task 1 |
|
||||
| source→channel×3→template→rule→notify | Task 1 |
|
||||
| Assert matched/accepted/3 channels | Task 1 |
|
||||
| Poll message-logs 15s | Task 1 |
|
||||
| Cleanup | Task 1 |
|
||||
| Makefile test-e2e | Task 1 |
|
||||
| No secrets in repo | Task 1 |
|
||||
@@ -219,28 +219,28 @@ Response 429:
|
||||
```
|
||||
来源:
|
||||
POST /api/v1/sources
|
||||
GET /api/v1/sources
|
||||
GET /api/v1/sources?page=1&page_size=20
|
||||
GET /api/v1/sources/:id
|
||||
PUT /api/v1/sources/:id
|
||||
DELETE /api/v1/sources/:id
|
||||
|
||||
模板:
|
||||
POST /api/v1/templates
|
||||
GET /api/v1/templates
|
||||
GET /api/v1/templates?page=1&page_size=20
|
||||
GET /api/v1/templates/:id
|
||||
PUT /api/v1/templates/:id
|
||||
DELETE /api/v1/templates/:id
|
||||
|
||||
渠道:
|
||||
POST /api/v1/channels
|
||||
GET /api/v1/channels
|
||||
GET /api/v1/channels?page=1&page_size=20
|
||||
GET /api/v1/channels/:id
|
||||
PUT /api/v1/channels/:id
|
||||
DELETE /api/v1/channels/:id
|
||||
|
||||
规则:
|
||||
POST /api/v1/rules
|
||||
GET /api/v1/rules
|
||||
GET /api/v1/rules?page=1&page_size=20
|
||||
GET /api/v1/rules/:id
|
||||
PUT /api/v1/rules/:id
|
||||
DELETE /api/v1/rules/:id
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
# Remote E2E Notify Flow Test Design
|
||||
|
||||
**Date:** 2026-08-01
|
||||
**Status:** Approved (pending final user review of this doc)
|
||||
|
||||
## Goal
|
||||
|
||||
Add a Go end-to-end test that hits a **deployed** notification service over HTTP and exercises the full configuration → send path:
|
||||
|
||||
`source → channel(s) → template → rule → POST /notify`
|
||||
|
||||
Channels under test: **DingTalk**, **Bark**, **Email**. Real notifications are sent.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- Local `httptest` / in-process server tests
|
||||
- Mocking `ChannelSender`
|
||||
- CI by default (opt-in via build tag + env vars)
|
||||
- Committing secrets to the repo
|
||||
|
||||
## Placement
|
||||
|
||||
| Item | Path |
|
||||
|------|------|
|
||||
| Test | `test/e2e/notify_flow_test.go` |
|
||||
| Build tag | `//go:build e2e` |
|
||||
| Make target | `test-e2e` (optional helper) |
|
||||
|
||||
Default `make test` / `go test ./internal/...` must **not** run this suite.
|
||||
|
||||
## Target environment
|
||||
|
||||
| Setting | Default / source |
|
||||
|---------|------------------|
|
||||
| Base URL | `E2E_BASE_URL` → default `http://82.157.251.93:8080` |
|
||||
| Admin auth | `E2E_ADMIN_KEY` → default `admin-sk-change-me` |
|
||||
|
||||
## Required env vars (channels)
|
||||
|
||||
If any of these are missing, the test **Skips** (does not fail):
|
||||
|
||||
| Variable | Purpose |
|
||||
|----------|---------|
|
||||
| `E2E_DINGTALK_WEBHOOK` | DingTalk robot webhook URL |
|
||||
| `E2E_DINGTALK_SECRET` | DingTalk sign secret |
|
||||
| `E2E_BARK_URL` | Bark base URL, e.g. `https://api.day.app/<device_key>` |
|
||||
| `E2E_EMAIL_TO` | Email recipient |
|
||||
|
||||
Secrets live only in the runner environment / local shell, never in source or this spec.
|
||||
|
||||
## Flow
|
||||
|
||||
Resource names use a unique suffix (unix timestamp or random) to avoid collisions, e.g. `e2e-src-<suffix>`.
|
||||
|
||||
1. **Create Source** — `POST /api/v1/sources`
|
||||
- `parse_mode: json`, `status: 1`
|
||||
- Capture `id`, `api_key`, `name`
|
||||
|
||||
2. **Create Channels** — `POST /api/v1/channels` ×3
|
||||
- DingTalk: `type=dingtalk`, config `{webhook_url, secret}`
|
||||
- Bark: `type=bark`, config `{url}`
|
||||
- Email: `type=email`, config `{to: [E2E_EMAIL_TO]}`
|
||||
- All `status: 1`
|
||||
|
||||
3. **Create Template** — `POST /api/v1/templates`
|
||||
- Content includes `{{.symbol}}` and `{{.price}}`
|
||||
|
||||
4. **Create Rule** — `POST /api/v1/rules`
|
||||
- `source_name` / `template_name` / `channels` by name
|
||||
- `event: trade.open`
|
||||
- `enabled: 1`
|
||||
- Conditions: `symbol` exists AND `price` gt `0`
|
||||
|
||||
5. **Notify** — `POST /api/v1/notify`
|
||||
- `Authorization: Bearer <source.api_key>`
|
||||
- Body: `{"event":"trade.open","data":{"symbol":"BTC","price":65000}}`
|
||||
|
||||
6. **Assert notify response**
|
||||
- `matched == true`
|
||||
- `accepted == true`
|
||||
- `channels` length == 3
|
||||
|
||||
7. **Message-log check**
|
||||
- Poll `GET /api/v1/message-logs?source=...&event=trade.open` for up to ~15s
|
||||
- Assert at least one log entry appears (fail if none within timeout)
|
||||
|
||||
8. **Cleanup** via `t.Cleanup` (reverse order)
|
||||
- Delete rule → template → channels → source
|
||||
- Cleanup errors: log only, do not fail the test after a successful assertion path
|
||||
|
||||
## Run command
|
||||
|
||||
```bash
|
||||
E2E_DINGTALK_WEBHOOK='...' \
|
||||
E2E_DINGTALK_SECRET='...' \
|
||||
E2E_BARK_URL='https://api.day.app/<device_key>' \
|
||||
E2E_EMAIL_TO='149516886@qq.com' \
|
||||
go test -tags e2e ./test/e2e/ -v -count=1 -timeout 2m
|
||||
```
|
||||
|
||||
Optional Makefile:
|
||||
|
||||
```make
|
||||
test-e2e:
|
||||
go test -tags e2e ./test/e2e/ -v -count=1 -timeout 2m
|
||||
```
|
||||
|
||||
## Implementation notes
|
||||
|
||||
- Use `net/http` + `encoding/json` (stdlib); no new test deps required.
|
||||
- Helper for admin JSON requests and source-authed notify.
|
||||
- Failures must include HTTP status + response body for debugging.
|
||||
- Do not hardcode channel secrets in the test file.
|
||||
|
||||
## Success criteria
|
||||
|
||||
- With env vars set and remote healthy: test passes; DingTalk, Bark, and Email receive a message.
|
||||
- Without channel env vars: test skips cleanly.
|
||||
- Without `-tags e2e`: suite is not compiled/run.
|
||||
@@ -13,10 +13,10 @@ type ChannelSender interface {
|
||||
Send(title, content string, config json.RawMessage) error
|
||||
}
|
||||
|
||||
func NewSender(channelType string, smtpCfg *config.SMTPConfig) (ChannelSender, error) {
|
||||
func NewSender(channelType string, smtpCfg *config.SMTPConfig, dingtalkLimiter *DingTalkLimiter) (ChannelSender, error) {
|
||||
switch channelType {
|
||||
case "dingtalk":
|
||||
return &DingTalkSender{}, nil
|
||||
return &DingTalkSender{limiter: dingtalkLimiter}, nil
|
||||
case "wecom":
|
||||
return &WeComSender{}, nil
|
||||
case "bark":
|
||||
|
||||
@@ -2,6 +2,7 @@ package adapter
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
@@ -27,7 +28,9 @@ type dingtalkMD struct {
|
||||
Text string `json:"text"`
|
||||
}
|
||||
|
||||
type DingTalkSender struct{}
|
||||
type DingTalkSender struct {
|
||||
limiter *DingTalkLimiter
|
||||
}
|
||||
|
||||
func (s *DingTalkSender) Type() string { return "dingtalk" }
|
||||
|
||||
@@ -37,6 +40,12 @@ func (s *DingTalkSender) Send(title, content string, config json.RawMessage) err
|
||||
return fmt.Errorf("parse dingtalk config: %w", err)
|
||||
}
|
||||
|
||||
if s.limiter != nil {
|
||||
if err := s.limiter.Acquire(context.Background(), DingTalkLimitKey(cfg.WebhookURL)); err != nil {
|
||||
return fmt.Errorf("dingtalk rate limit: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
reqURL := cfg.WebhookURL
|
||||
if cfg.Secret != "" {
|
||||
timestamp := time.Now().UnixMilli()
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/url"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const DefaultDingTalkPerMin = 18
|
||||
|
||||
// MinuteWindowStore tracks per-key counters in a fixed UTC-minute window.
|
||||
type MinuteWindowStore interface {
|
||||
// TryIncr increments if under limit; returns (count, true) when acquired,
|
||||
// or (currentOrLimit, false) when the window is full.
|
||||
TryIncr(ctx context.Context, key string, minute int64, limit int) (count int64, ok bool, err error)
|
||||
}
|
||||
|
||||
// DingTalkLimiter enforces per-robot (access_token) send quota and waits for the next minute when full.
|
||||
type DingTalkLimiter struct {
|
||||
store MinuteWindowStore
|
||||
limit int
|
||||
now func() time.Time
|
||||
sleep func(time.Duration)
|
||||
}
|
||||
|
||||
func NewDingTalkLimiter(store MinuteWindowStore, limitPerMin int) *DingTalkLimiter {
|
||||
if limitPerMin <= 0 {
|
||||
limitPerMin = DefaultDingTalkPerMin
|
||||
}
|
||||
return &DingTalkLimiter{
|
||||
store: store,
|
||||
limit: limitPerMin,
|
||||
now: time.Now,
|
||||
sleep: time.Sleep,
|
||||
}
|
||||
}
|
||||
|
||||
// DingTalkLimitKey returns the rate-limit key for a webhook URL (access_token).
|
||||
func DingTalkLimitKey(webhookURL string) string {
|
||||
if webhookURL == "" {
|
||||
return "unknown"
|
||||
}
|
||||
u, err := url.Parse(webhookURL)
|
||||
if err != nil {
|
||||
return webhookURL
|
||||
}
|
||||
if tok := u.Query().Get("access_token"); tok != "" {
|
||||
return tok
|
||||
}
|
||||
return webhookURL
|
||||
}
|
||||
|
||||
// Acquire blocks until a send slot is available for key in the current minute window.
|
||||
func (l *DingTalkLimiter) Acquire(ctx context.Context, key string) error {
|
||||
if l == nil || l.store == nil {
|
||||
return nil
|
||||
}
|
||||
for {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
now := l.now()
|
||||
minute := now.Unix() / 60
|
||||
count, ok, err := l.store.TryIncr(ctx, key, minute, l.limit)
|
||||
if err != nil {
|
||||
return fmt.Errorf("dingtalk rate limit: %w", err)
|
||||
}
|
||||
if ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
next := time.Unix((minute+1)*60, 0)
|
||||
wait := next.Sub(now)
|
||||
if wait < time.Millisecond {
|
||||
wait = time.Millisecond
|
||||
}
|
||||
slog.Info("dingtalk rate limited, waiting for next minute",
|
||||
"key_suffix", maskKey(key),
|
||||
"count", count,
|
||||
"limit", l.limit,
|
||||
"wait", wait,
|
||||
)
|
||||
if err := l.sleepCtx(ctx, wait); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (l *DingTalkLimiter) sleepCtx(ctx context.Context, d time.Duration) error {
|
||||
// Prefer injectable sleep for tests; also honor context cancel via polling when possible.
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
l.sleep(d)
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-done:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func maskKey(key string) string {
|
||||
if len(key) <= 8 {
|
||||
return "****"
|
||||
}
|
||||
return key[:4] + "…" + key[len(key)-4:]
|
||||
}
|
||||
|
||||
// --- in-memory store (fallback / tests) ---
|
||||
|
||||
type memoryMinuteStore struct {
|
||||
mu sync.Mutex
|
||||
data map[string]int64 // key:minute -> count
|
||||
}
|
||||
|
||||
func newMemoryMinuteStore() *memoryMinuteStore {
|
||||
return &memoryMinuteStore{data: make(map[string]int64)}
|
||||
}
|
||||
|
||||
func (m *memoryMinuteStore) TryIncr(_ context.Context, key string, minute int64, limit int) (int64, bool, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
k := fmt.Sprintf("%s:%d", key, minute)
|
||||
cur := m.data[k]
|
||||
if cur >= int64(limit) {
|
||||
return cur, false, nil
|
||||
}
|
||||
cur++
|
||||
m.data[k] = cur
|
||||
return cur, true, nil
|
||||
}
|
||||
|
||||
// NewMemoryDingTalkLimiter builds a process-local limiter (single-instance only).
|
||||
func NewMemoryDingTalkLimiter(limitPerMin int) *DingTalkLimiter {
|
||||
return NewDingTalkLimiter(newMemoryMinuteStore(), limitPerMin)
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
package adapter
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestDingTalkLimitKey(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
url string
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "token only",
|
||||
url: "https://oapi.dingtalk.com/robot/send?access_token=abc123",
|
||||
want: "abc123",
|
||||
},
|
||||
{
|
||||
name: "token with other params",
|
||||
url: "https://oapi.dingtalk.com/robot/send?access_token=tok&foo=1",
|
||||
want: "tok",
|
||||
},
|
||||
{
|
||||
name: "missing token falls back to url",
|
||||
url: "https://oapi.dingtalk.com/robot/send?foo=1",
|
||||
want: "https://oapi.dingtalk.com/robot/send?foo=1",
|
||||
},
|
||||
{
|
||||
name: "empty",
|
||||
url: "",
|
||||
want: "unknown",
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := DingTalkLimitKey(tt.url); got != tt.want {
|
||||
t.Fatalf("DingTalkLimitKey(%q) = %q, want %q", tt.url, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDingTalkLimiterAcquireWaitsForNextWindow(t *testing.T) {
|
||||
now := time.Unix(1000, 0) // fixed
|
||||
var slept time.Duration
|
||||
store := newMemoryMinuteStore()
|
||||
lim := &DingTalkLimiter{
|
||||
store: store,
|
||||
limit: 2,
|
||||
now: func() time.Time { return now },
|
||||
sleep: func(d time.Duration) {
|
||||
slept += d
|
||||
now = now.Add(d)
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
key := "tok-1"
|
||||
if err := lim.Acquire(ctx, key); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := lim.Acquire(ctx, key); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if slept != 0 {
|
||||
t.Fatalf("unexpected sleep before limit: %v", slept)
|
||||
}
|
||||
|
||||
// third should wait until next minute boundary (60 - 1000%60 = 20s? 1000/60=16 rem 40, next at 17*60=1020, wait 20s)
|
||||
if err := lim.Acquire(ctx, key); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if slept != 20*time.Second {
|
||||
t.Fatalf("slept = %v, want 20s", slept)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDingTalkLimiterSameKeyShared(t *testing.T) {
|
||||
now := time.Unix(0, 0)
|
||||
store := newMemoryMinuteStore()
|
||||
lim := &DingTalkLimiter{
|
||||
store: store,
|
||||
limit: 1,
|
||||
now: func() time.Time { return now },
|
||||
sleep: func(d time.Duration) { now = now.Add(d) },
|
||||
}
|
||||
ctx := context.Background()
|
||||
if err := lim.Acquire(ctx, "shared"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := lim.Acquire(ctx, "shared"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if now.Unix() != 60 {
|
||||
t.Fatalf("expected wait to next minute boundary, now=%v", now)
|
||||
}
|
||||
}
|
||||
Vendored
+60
@@ -0,0 +1,60 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
// TryIncr implements adapter.MinuteWindowStore for DingTalk per-minute quotas.
|
||||
func (c *Cache) TryIncr(ctx context.Context, key string, minute int64, limit int) (int64, bool, error) {
|
||||
if c == nil || c.rdb == nil {
|
||||
return 0, false, fmt.Errorf("redis unavailable")
|
||||
}
|
||||
redisKey := fmt.Sprintf("dingtalk:rl:%s:%d", key, minute)
|
||||
script := redis.NewScript(`
|
||||
local key = KEYS[1]
|
||||
local limit = tonumber(ARGV[1])
|
||||
local cur = tonumber(redis.call('GET', key) or '0')
|
||||
if cur >= limit then
|
||||
return {cur, 0}
|
||||
end
|
||||
local n = redis.call('INCR', key)
|
||||
if n == 1 then
|
||||
redis.call('EXPIRE', key, 120)
|
||||
end
|
||||
return {n, 1}
|
||||
`)
|
||||
res, err := script.Run(ctx, c.rdb, []string{redisKey}, limit).Slice()
|
||||
if err != nil {
|
||||
return 0, false, fmt.Errorf("dingtalk TryIncr: %w", err)
|
||||
}
|
||||
if len(res) != 2 {
|
||||
return 0, false, fmt.Errorf("dingtalk TryIncr: unexpected result %#v", res)
|
||||
}
|
||||
count, err := toInt64(res[0])
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
}
|
||||
okFlag, err := toInt64(res[1])
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
}
|
||||
return count, okFlag == 1, nil
|
||||
}
|
||||
|
||||
func toInt64(v interface{}) (int64, error) {
|
||||
switch n := v.(type) {
|
||||
case int64:
|
||||
return n, nil
|
||||
case int:
|
||||
return int64(n), nil
|
||||
case string:
|
||||
var x int64
|
||||
_, err := fmt.Sscan(n, &x)
|
||||
return x, err
|
||||
default:
|
||||
return 0, fmt.Errorf("cannot convert %T to int64", v)
|
||||
}
|
||||
}
|
||||
@@ -56,6 +56,7 @@ type SMTPConfig struct {
|
||||
|
||||
type RateLimitConfig struct {
|
||||
Default int `mapstructure:"default"`
|
||||
DingTalkPerMin int `mapstructure:"dingtalk_per_min"` // per robot webhook; 0 => 18
|
||||
}
|
||||
|
||||
type LogbullConfig struct {
|
||||
|
||||
@@ -47,12 +47,19 @@ func (h *ChannelHandler) Create(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (h *ChannelHandler) List(c *gin.Context) {
|
||||
channels, err := h.store.ListChannels(c.Request.Context())
|
||||
var page store.PageFilter
|
||||
if err := c.ShouldBindQuery(&page); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
page.Normalize()
|
||||
|
||||
channels, total, err := h.store.ListChannels(c.Request.Context(), page)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, channels)
|
||||
c.JSON(http.StatusOK, gin.H{"data": channels, "total": total, "page": page.Page})
|
||||
}
|
||||
|
||||
func (h *ChannelHandler) Get(c *gin.Context) {
|
||||
|
||||
@@ -22,6 +22,7 @@ func (h *MessageLogHandler) List(c *gin.Context) {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
filter.Normalize()
|
||||
|
||||
logs, total, err := h.store.ListMessageLogs(c.Request.Context(), filter)
|
||||
if err != nil {
|
||||
|
||||
@@ -81,12 +81,19 @@ func (h *RuleHandler) Create(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (h *RuleHandler) List(c *gin.Context) {
|
||||
rules, err := h.store.ListRules(c.Request.Context())
|
||||
var page store.PageFilter
|
||||
if err := c.ShouldBindQuery(&page); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
page.Normalize()
|
||||
|
||||
rules, total, err := h.store.ListRules(c.Request.Context(), page)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, rules)
|
||||
c.JSON(http.StatusOK, gin.H{"data": rules, "total": total, "page": page.Page})
|
||||
}
|
||||
|
||||
func (h *RuleHandler) Get(c *gin.Context) {
|
||||
@@ -163,19 +170,35 @@ func (h *RuleHandler) Delete(c *gin.Context) {
|
||||
|
||||
func (h *RuleHandler) Enable(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
|
||||
}
|
||||
if err := h.store.SetRuleEnabled(c.Request.Context(), id, true); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if h.cache != nil {
|
||||
_ = h.cache.InvalidateRule(c.Request.Context(), rule.SourceID, rule.Event)
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
func (h *RuleHandler) Disable(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
|
||||
}
|
||||
if err := h.store.SetRuleEnabled(c.Request.Context(), id, false); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if h.cache != nil {
|
||||
_ = h.cache.InvalidateRule(c.Request.Context(), rule.SourceID, rule.Event)
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
@@ -186,6 +209,9 @@ func (h *RuleHandler) EnableChannel(c *gin.Context) {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if h.cache != nil {
|
||||
_ = h.cache.InvalidateChannels(c.Request.Context(), ruleID)
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
@@ -196,6 +222,9 @@ func (h *RuleHandler) DisableChannel(c *gin.Context) {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if h.cache != nil {
|
||||
_ = h.cache.InvalidateChannels(c.Request.Context(), ruleID)
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||||
}
|
||||
|
||||
|
||||
@@ -54,12 +54,19 @@ func (h *SourceHandler) Create(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (h *SourceHandler) List(c *gin.Context) {
|
||||
sources, err := h.store.ListSources(c.Request.Context())
|
||||
var page store.PageFilter
|
||||
if err := c.ShouldBindQuery(&page); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
page.Normalize()
|
||||
|
||||
sources, total, err := h.store.ListSources(c.Request.Context(), page)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, sources)
|
||||
c.JSON(http.StatusOK, gin.H{"data": sources, "total": total, "page": page.Page})
|
||||
}
|
||||
|
||||
func (h *SourceHandler) Get(c *gin.Context) {
|
||||
|
||||
@@ -40,12 +40,19 @@ func (h *TemplateHandler) Create(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (h *TemplateHandler) List(c *gin.Context) {
|
||||
templates, err := h.store.ListTemplates(c.Request.Context())
|
||||
var page store.PageFilter
|
||||
if err := c.ShouldBindQuery(&page); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
page.Normalize()
|
||||
|
||||
templates, total, err := h.store.ListTemplates(c.Request.Context(), page)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, templates)
|
||||
c.JSON(http.StatusOK, gin.H{"data": templates, "total": total, "page": page.Page})
|
||||
}
|
||||
|
||||
func (h *TemplateHandler) Get(c *gin.Context) {
|
||||
|
||||
@@ -47,10 +47,16 @@ func (s *Store) GetChannelByName(ctx context.Context, name string) (*model.Chann
|
||||
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 notification_channel ORDER BY id`)
|
||||
func (s *Store) ListChannels(ctx context.Context, page PageFilter) ([]model.Channel, int, error) {
|
||||
var count int
|
||||
if err := s.DB.GetContext(ctx, &count, `SELECT COUNT(*) FROM notification_channel`); err != nil {
|
||||
return nil, 0, fmt.Errorf("count channels: %w", err)
|
||||
}
|
||||
|
||||
page.Normalize()
|
||||
rows, err := s.DB.QueryContext(ctx, `SELECT id, name, type, config, status, created_at, updated_at FROM notification_channel ORDER BY id LIMIT ? OFFSET ?`, page.PageSize, page.Offset())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list channels: %w", err)
|
||||
return nil, 0, fmt.Errorf("list channels: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
@@ -59,13 +65,13 @@ func (s *Store) ListChannels(ctx context.Context) ([]model.Channel, error) {
|
||||
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)
|
||||
return nil, 0, fmt.Errorf("scan channel: %w", err)
|
||||
}
|
||||
raw := json.RawMessage(configBytes)
|
||||
ch.Config = &raw
|
||||
channels = append(channels, ch)
|
||||
}
|
||||
return channels, rows.Err()
|
||||
return channels, count, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Store) UpdateChannel(ctx context.Context, id int, ch *model.Channel) error {
|
||||
|
||||
@@ -11,8 +11,7 @@ type MessageLogFilter struct {
|
||||
Source string `form:"source"`
|
||||
Event string `form:"event"`
|
||||
Status string `form:"status"`
|
||||
Page int `form:"page"`
|
||||
PageSize int `form:"page_size"`
|
||||
PageFilter
|
||||
}
|
||||
|
||||
func (s *Store) CreateMessageLog(ctx context.Context, ml *model.MessageLog) error {
|
||||
@@ -54,17 +53,11 @@ func (s *Store) ListMessageLogs(ctx context.Context, filter MessageLogFilter) ([
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
if filter.Page <= 0 {
|
||||
filter.Page = 1
|
||||
}
|
||||
if filter.PageSize <= 0 {
|
||||
filter.PageSize = 20
|
||||
}
|
||||
offset := (filter.Page - 1) * filter.PageSize
|
||||
filter.Normalize()
|
||||
|
||||
logs := make([]model.MessageLog, 0)
|
||||
query := "SELECT * FROM notification_message_log " + where + " ORDER BY id DESC LIMIT ? OFFSET ?"
|
||||
args = append(args, filter.PageSize, offset)
|
||||
args = append(args, filter.PageSize, filter.Offset())
|
||||
if err := s.DB.SelectContext(ctx, &logs, query, args...); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
package store
|
||||
|
||||
// PageFilter is the shared pagination query for list endpoints.
|
||||
type PageFilter struct {
|
||||
Page int `form:"page"`
|
||||
PageSize int `form:"page_size"`
|
||||
}
|
||||
|
||||
// Normalize applies defaults: page=1, page_size=20.
|
||||
func (p *PageFilter) Normalize() {
|
||||
if p.Page <= 0 {
|
||||
p.Page = 1
|
||||
}
|
||||
if p.PageSize <= 0 {
|
||||
p.PageSize = 20
|
||||
}
|
||||
}
|
||||
|
||||
// Offset returns the SQL OFFSET for the current page.
|
||||
func (p PageFilter) Offset() int {
|
||||
return (p.Page - 1) * p.PageSize
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package store
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestPageFilterNormalize(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
in PageFilter
|
||||
wantPage int
|
||||
wantPageSize int
|
||||
wantOffset int
|
||||
}{
|
||||
{name: "defaults", in: PageFilter{}, wantPage: 1, wantPageSize: 20, wantOffset: 0},
|
||||
{name: "negative", in: PageFilter{Page: -1, PageSize: -5}, wantPage: 1, wantPageSize: 20, wantOffset: 0},
|
||||
{name: "page2", in: PageFilter{Page: 2, PageSize: 10}, wantPage: 2, wantPageSize: 10, wantOffset: 10},
|
||||
{name: "zero page size", in: PageFilter{Page: 3, PageSize: 0}, wantPage: 3, wantPageSize: 20, wantOffset: 40},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
p := tt.in
|
||||
p.Normalize()
|
||||
if p.Page != tt.wantPage || p.PageSize != tt.wantPageSize {
|
||||
t.Fatalf("Normalize() = %+v, want page=%d page_size=%d", p, tt.wantPage, tt.wantPageSize)
|
||||
}
|
||||
if got := p.Offset(); got != tt.wantOffset {
|
||||
t.Fatalf("Offset() = %d, want %d", got, tt.wantOffset)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+14
-4
@@ -66,13 +66,23 @@ func (s *Store) GetRuleBySourceEvent(ctx context.Context, sourceID int, event st
|
||||
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 notification_rule ORDER BY id`)
|
||||
func (s *Store) ListRules(ctx context.Context, page PageFilter) ([]model.Rule, int, error) {
|
||||
var count int
|
||||
if err := s.DB.GetContext(ctx, &count, `SELECT COUNT(*) FROM notification_rule`); err != nil {
|
||||
return nil, 0, fmt.Errorf("count rules: %w", err)
|
||||
}
|
||||
|
||||
page.Normalize()
|
||||
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 LIMIT ? OFFSET ?`, page.PageSize, page.Offset())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list rules: %w", err)
|
||||
return nil, 0, fmt.Errorf("list rules: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
return scanRules(rows)
|
||||
rules, err := scanRules(rows)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
return rules, count, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpdateRule(ctx context.Context, id int, r *model.Rule, channelIDs []int) error {
|
||||
|
||||
@@ -54,13 +54,19 @@ func (s *Store) GetSourceByName(ctx context.Context, name string) (*model.Source
|
||||
return &src, nil
|
||||
}
|
||||
|
||||
func (s *Store) ListSources(ctx context.Context) ([]model.Source, error) {
|
||||
sources := make([]model.Source, 0)
|
||||
err := s.DB.SelectContext(ctx, &sources, `SELECT * FROM notification_source ORDER BY id`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list sources: %w", err)
|
||||
func (s *Store) ListSources(ctx context.Context, page PageFilter) ([]model.Source, int, error) {
|
||||
var count int
|
||||
if err := s.DB.GetContext(ctx, &count, `SELECT COUNT(*) FROM notification_source`); err != nil {
|
||||
return nil, 0, fmt.Errorf("count sources: %w", err)
|
||||
}
|
||||
return sources, nil
|
||||
|
||||
page.Normalize()
|
||||
sources := make([]model.Source, 0)
|
||||
err := s.DB.SelectContext(ctx, &sources, `SELECT * FROM notification_source ORDER BY id LIMIT ? OFFSET ?`, page.PageSize, page.Offset())
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("list sources: %w", err)
|
||||
}
|
||||
return sources, count, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpdateSource(ctx context.Context, id int, src *model.Source) error {
|
||||
|
||||
@@ -36,13 +36,19 @@ func (s *Store) GetTemplateByName(ctx context.Context, name string) (*model.Temp
|
||||
return &t, nil
|
||||
}
|
||||
|
||||
func (s *Store) ListTemplates(ctx context.Context) ([]model.Template, error) {
|
||||
templates := make([]model.Template, 0)
|
||||
err := s.DB.SelectContext(ctx, &templates, `SELECT * FROM notification_template ORDER BY id`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list templates: %w", err)
|
||||
func (s *Store) ListTemplates(ctx context.Context, page PageFilter) ([]model.Template, int, error) {
|
||||
var count int
|
||||
if err := s.DB.GetContext(ctx, &count, `SELECT COUNT(*) FROM notification_template`); err != nil {
|
||||
return nil, 0, fmt.Errorf("count templates: %w", err)
|
||||
}
|
||||
return templates, nil
|
||||
|
||||
page.Normalize()
|
||||
templates := make([]model.Template, 0)
|
||||
err := s.DB.SelectContext(ctx, &templates, `SELECT * FROM notification_template ORDER BY id LIMIT ? OFFSET ?`, page.PageSize, page.Offset())
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("list templates: %w", err)
|
||||
}
|
||||
return templates, count, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpdateTemplate(ctx context.Context, id int, t *model.Template) error {
|
||||
|
||||
@@ -0,0 +1,506 @@
|
||||
//go:build e2e
|
||||
|
||||
package e2e
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type fixture struct {
|
||||
t *testing.T
|
||||
client *http.Client
|
||||
base string
|
||||
adminKey string
|
||||
apiKey string
|
||||
srcName string
|
||||
srcID int
|
||||
tmplID int
|
||||
ruleID int
|
||||
dingID int
|
||||
barkID int
|
||||
emailID int
|
||||
chDing string
|
||||
chBark string
|
||||
chEmail string
|
||||
}
|
||||
|
||||
func requireE2EEnv(t *testing.T) (base, adminKey, dingWebhook, dingSecret, barkURL, emailTo string) {
|
||||
t.Helper()
|
||||
base = envOr("E2E_BASE_URL", "http://82.157.251.93:8080")
|
||||
adminKey = envOr("E2E_ADMIN_KEY", "admin-sk-change-me")
|
||||
dingWebhook = os.Getenv("E2E_DINGTALK_WEBHOOK")
|
||||
dingSecret = os.Getenv("E2E_DINGTALK_SECRET")
|
||||
barkURL = os.Getenv("E2E_BARK_URL")
|
||||
emailTo = os.Getenv("E2E_EMAIL_TO")
|
||||
if dingWebhook == "" || dingSecret == "" || barkURL == "" || emailTo == "" {
|
||||
t.Skip("missing E2E_DINGTALK_WEBHOOK / E2E_DINGTALK_SECRET / E2E_BARK_URL / E2E_EMAIL_TO")
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func setupFixture(t *testing.T) *fixture {
|
||||
t.Helper()
|
||||
base, adminKey, dingWebhook, dingSecret, barkURL, emailTo := requireE2EEnv(t)
|
||||
client := &http.Client{Timeout: 30 * time.Second}
|
||||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
|
||||
f := &fixture{
|
||||
t: t,
|
||||
client: client,
|
||||
base: base,
|
||||
adminKey: adminKey,
|
||||
srcName: "e2e-src-" + suffix,
|
||||
chDing: "e2e-ding-" + suffix,
|
||||
chBark: "e2e-bark-" + suffix,
|
||||
chEmail: "e2e-email-" + suffix,
|
||||
}
|
||||
tmplName := "e2e-tmpl-" + suffix
|
||||
|
||||
// Cleanups run LIFO: register source first so rule (registered last) is deleted first.
|
||||
src := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/sources", map[string]any{
|
||||
"name": f.srcName, "parse_mode": "json", "status": 1,
|
||||
}, http.StatusCreated)
|
||||
f.srcID = intFrom(src["id"])
|
||||
apiKey, _ := src["api_key"].(string)
|
||||
if apiKey == "" {
|
||||
t.Fatal("source api_key empty")
|
||||
}
|
||||
f.apiKey = apiKey
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/sources/%d", f.srcID), nil)
|
||||
})
|
||||
|
||||
ding := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/channels", map[string]any{
|
||||
"name": f.chDing, "type": "dingtalk", "status": 1,
|
||||
"config": map[string]string{"webhook_url": dingWebhook, "secret": dingSecret},
|
||||
}, http.StatusCreated)
|
||||
f.dingID = intFrom(ding["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/channels/%d", f.dingID), nil)
|
||||
})
|
||||
|
||||
bark := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/channels", map[string]any{
|
||||
"name": f.chBark, "type": "bark", "status": 1,
|
||||
"config": map[string]string{"url": barkURL},
|
||||
}, http.StatusCreated)
|
||||
f.barkID = intFrom(bark["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/channels/%d", f.barkID), nil)
|
||||
})
|
||||
|
||||
email := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/channels", map[string]any{
|
||||
"name": f.chEmail, "type": "email", "status": 1,
|
||||
"config": map[string]any{"to": []string{emailTo}},
|
||||
}, http.StatusCreated)
|
||||
f.emailID = intFrom(email["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/channels/%d", f.emailID), nil)
|
||||
})
|
||||
|
||||
tmpl := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/templates", map[string]any{
|
||||
"name": tmplName,
|
||||
"content": "### {{.symbol}} 开仓\n价格: {{.price}}",
|
||||
}, http.StatusCreated)
|
||||
f.tmplID = intFrom(tmpl["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/templates/%d", f.tmplID), nil)
|
||||
})
|
||||
|
||||
rule := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/rules", map[string]any{
|
||||
"source_name": f.srcName,
|
||||
"event": "trade.open",
|
||||
"template_name": tmplName,
|
||||
"channels": []string{f.chDing, f.chBark, f.chEmail},
|
||||
"conditions": []map[string]string{
|
||||
{"field": "symbol", "op": "exists"},
|
||||
{"field": "price", "op": "gt", "value": "0"},
|
||||
},
|
||||
"enabled": 1,
|
||||
}, http.StatusCreated)
|
||||
f.ruleID = intFrom(rule["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/rules/%d", f.ruleID), nil)
|
||||
})
|
||||
|
||||
return f
|
||||
}
|
||||
|
||||
func (f *fixture) notify(body any) map[string]any {
|
||||
f.t.Helper()
|
||||
return mustJSON(f.t, f.client, http.MethodPost, f.base+"/api/v1/notify", f.apiKey, body, http.StatusOK)
|
||||
}
|
||||
|
||||
func (f *fixture) messageLogTotal() float64 {
|
||||
f.t.Helper()
|
||||
q := url.Values{}
|
||||
q.Set("source", f.srcName)
|
||||
q.Set("event", "trade.open")
|
||||
q.Set("page", "1")
|
||||
q.Set("page_size", "20")
|
||||
logs := mustAdminJSON(f.t, f.client, f.base, f.adminKey, http.MethodGet,
|
||||
"/api/v1/message-logs?"+q.Encode(), nil, http.StatusOK)
|
||||
n, _ := logs["total"].(float64)
|
||||
return n
|
||||
}
|
||||
|
||||
func (f *fixture) waitMessageLogs(min float64, timeout time.Duration) float64 {
|
||||
f.t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
var total float64
|
||||
for time.Now().Before(deadline) {
|
||||
total = f.messageLogTotal()
|
||||
if total >= min {
|
||||
return total
|
||||
}
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
f.t.Fatalf("expected message logs >= %v within %s, total=%v", min, timeout, total)
|
||||
return total
|
||||
}
|
||||
|
||||
func TestNotifyFlow_SourceChannelTemplateRuleSend(t *testing.T) {
|
||||
f := setupFixture(t)
|
||||
|
||||
resp := f.notify(map[string]any{
|
||||
"event": "trade.open",
|
||||
"data": map[string]any{"symbol": "BTC", "price": 65000},
|
||||
})
|
||||
if resp["matched"] != true {
|
||||
t.Fatalf("matched: %#v body=%v", resp["matched"], resp)
|
||||
}
|
||||
if resp["accepted"] != true {
|
||||
t.Fatalf("accepted: %#v body=%v", resp["accepted"], resp)
|
||||
}
|
||||
chs, _ := resp["channels"].([]any)
|
||||
if len(chs) != 3 {
|
||||
t.Fatalf("want 3 channels, got %#v", resp["channels"])
|
||||
}
|
||||
f.waitMessageLogs(1, 15*time.Second)
|
||||
}
|
||||
|
||||
func TestNotifyFlow_ConditionFiltered(t *testing.T) {
|
||||
f := setupFixture(t)
|
||||
before := f.messageLogTotal()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
data map[string]any
|
||||
}{
|
||||
{"price_not_gt_zero", map[string]any{"symbol": "BTC", "price": 0}},
|
||||
{"symbol_missing", map[string]any{"price": 65000}},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
resp := f.notify(map[string]any{"event": "trade.open", "data": tc.data})
|
||||
if resp["matched"] != true {
|
||||
t.Fatalf("matched: %#v body=%v", resp["matched"], resp)
|
||||
}
|
||||
if resp["filtered"] != true {
|
||||
t.Fatalf("filtered: %#v body=%v", resp["filtered"], resp)
|
||||
}
|
||||
if resp["reason"] != "condition not met" {
|
||||
t.Fatalf("reason: %#v body=%v", resp["reason"], resp)
|
||||
}
|
||||
if _, ok := resp["accepted"]; ok {
|
||||
t.Fatalf("filtered notify should not be accepted: %v", resp)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
time.Sleep(2 * time.Second)
|
||||
after := f.messageLogTotal()
|
||||
if after != before {
|
||||
t.Fatalf("filtered notify should not create logs: before=%v after=%v", before, after)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotifyFlow_NoMatchingRule(t *testing.T) {
|
||||
f := setupFixture(t)
|
||||
|
||||
resp := f.notify(map[string]any{
|
||||
"event": "unknown.event",
|
||||
"data": map[string]any{"symbol": "BTC", "price": 65000},
|
||||
})
|
||||
if resp["matched"] != false {
|
||||
t.Fatalf("matched: %#v body=%v", resp["matched"], resp)
|
||||
}
|
||||
if _, ok := resp["channels"]; ok {
|
||||
t.Fatalf("unmatched should not include channels: %v", resp)
|
||||
}
|
||||
if _, ok := resp["accepted"]; ok {
|
||||
t.Fatalf("unmatched should not be accepted: %v", resp)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotifyFlow_DisableChannel(t *testing.T) {
|
||||
f := setupFixture(t)
|
||||
|
||||
// Disable before any notify so channel cache is cold (safe even on older deploys).
|
||||
mustAdminJSON(t, f.client, f.base, f.adminKey, http.MethodPatch,
|
||||
fmt.Sprintf("/api/v1/rules/%d/channels/%d/disable", f.ruleID, f.dingID), nil, http.StatusOK)
|
||||
|
||||
resp := f.notify(map[string]any{
|
||||
"event": "trade.open",
|
||||
"data": map[string]any{"symbol": "BTC", "price": 65000},
|
||||
})
|
||||
if resp["matched"] != true || resp["accepted"] != true {
|
||||
t.Fatalf("unexpected notify resp: %v", resp)
|
||||
}
|
||||
chs, _ := resp["channels"].([]any)
|
||||
if len(chs) != 2 {
|
||||
t.Fatalf("want 2 channels after disable, got %#v", resp["channels"])
|
||||
}
|
||||
for _, ch := range chs {
|
||||
s, _ := ch.(string)
|
||||
if s == fmt.Sprintf("dingtalk:%d", f.dingID) {
|
||||
t.Fatalf("disabled dingtalk channel still routed: %#v", chs)
|
||||
}
|
||||
}
|
||||
f.waitMessageLogs(1, 15*time.Second)
|
||||
}
|
||||
|
||||
func TestNotifyFlow_DisableRule(t *testing.T) {
|
||||
f := setupFixture(t)
|
||||
|
||||
// Disable before any notify so rule cache is cold (safe even on older deploys).
|
||||
mustAdminJSON(t, f.client, f.base, f.adminKey, http.MethodPatch,
|
||||
fmt.Sprintf("/api/v1/rules/%d/disable", f.ruleID), nil, http.StatusOK)
|
||||
|
||||
resp := f.notify(map[string]any{
|
||||
"event": "trade.open",
|
||||
"data": map[string]any{"symbol": "BTC", "price": 65000},
|
||||
})
|
||||
if resp["matched"] != false {
|
||||
t.Fatalf("disabled rule should not match: %#v", resp)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNotifyFlow_DifferentTemplatesPerRule verifies rule→template binding:
|
||||
// same source, three events, each with a distinct template and a single channel.
|
||||
func TestNotifyFlow_DifferentTemplatesPerRule(t *testing.T) {
|
||||
base, adminKey, dingWebhook, dingSecret, barkURL, emailTo := requireE2EEnv(t)
|
||||
client := &http.Client{Timeout: 30 * time.Second}
|
||||
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
||||
srcName := "e2e-mt-" + suffix
|
||||
|
||||
src := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/sources", map[string]any{
|
||||
"name": srcName, "parse_mode": "json", "status": 1,
|
||||
}, http.StatusCreated)
|
||||
srcID := intFrom(src["id"])
|
||||
apiKey, _ := src["api_key"].(string)
|
||||
if apiKey == "" {
|
||||
t.Fatal("source api_key empty")
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/sources/%d", srcID), nil)
|
||||
})
|
||||
|
||||
type route struct {
|
||||
event string
|
||||
marker string
|
||||
chName string
|
||||
chType string
|
||||
chCfg map[string]any
|
||||
tmpl string
|
||||
channel string // set after create
|
||||
}
|
||||
routes := []route{
|
||||
{
|
||||
event: "e2e.ding",
|
||||
marker: "DING-TMPL",
|
||||
chName: "e2e-mtd-" + suffix,
|
||||
chType: "dingtalk",
|
||||
chCfg: map[string]any{"webhook_url": dingWebhook, "secret": dingSecret},
|
||||
tmpl: "DING-TMPL {{.symbol}} ding price={{.price}}",
|
||||
},
|
||||
{
|
||||
event: "e2e.bark",
|
||||
marker: "BARK-TMPL",
|
||||
chName: "e2e-mtb-" + suffix,
|
||||
chType: "bark",
|
||||
chCfg: map[string]any{"url": barkURL},
|
||||
tmpl: "BARK-TMPL {{.symbol}} bark price={{.price}}",
|
||||
},
|
||||
{
|
||||
event: "e2e.email",
|
||||
marker: "EMAIL-TMPL",
|
||||
chName: "e2e-mte-" + suffix,
|
||||
chType: "email",
|
||||
chCfg: map[string]any{"to": []string{emailTo}},
|
||||
tmpl: "EMAIL-TMPL {{.symbol}} email price={{.price}}",
|
||||
},
|
||||
}
|
||||
|
||||
for i := range routes {
|
||||
r := &routes[i]
|
||||
tmplName := "e2e-tmpl-" + r.marker + "-" + suffix
|
||||
|
||||
ch := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/channels", map[string]any{
|
||||
"name": r.chName, "type": r.chType, "status": 1, "config": r.chCfg,
|
||||
}, http.StatusCreated)
|
||||
chID := intFrom(ch["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/channels/%d", chID), nil)
|
||||
})
|
||||
|
||||
tmpl := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/templates", map[string]any{
|
||||
"name": tmplName, "content": r.tmpl,
|
||||
}, http.StatusCreated)
|
||||
tmplID := intFrom(tmpl["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/templates/%d", tmplID), nil)
|
||||
})
|
||||
|
||||
rule := mustAdminJSON(t, client, base, adminKey, http.MethodPost, "/api/v1/rules", map[string]any{
|
||||
"source_name": srcName,
|
||||
"event": r.event,
|
||||
"template_name": tmplName,
|
||||
"channels": []string{r.chName},
|
||||
"enabled": 1,
|
||||
}, http.StatusCreated)
|
||||
ruleID := intFrom(rule["id"])
|
||||
t.Cleanup(func() {
|
||||
_ = adminDo(client, base, adminKey, http.MethodDelete, fmt.Sprintf("/api/v1/rules/%d", ruleID), nil)
|
||||
})
|
||||
r.channel = fmt.Sprintf("%s:%d", r.chType, chID)
|
||||
}
|
||||
|
||||
payload := map[string]any{"symbol": "ETH", "price": 3200}
|
||||
for _, r := range routes {
|
||||
resp := mustJSON(t, client, http.MethodPost, base+"/api/v1/notify", apiKey, map[string]any{
|
||||
"event": r.event,
|
||||
"data": payload,
|
||||
}, http.StatusOK)
|
||||
if resp["matched"] != true || resp["accepted"] != true {
|
||||
t.Fatalf("event %s: unexpected resp %v", r.event, resp)
|
||||
}
|
||||
chs, _ := resp["channels"].([]any)
|
||||
if len(chs) != 1 || chs[0] != r.channel {
|
||||
t.Fatalf("event %s: want channels [%s], got %#v", r.event, r.channel, resp["channels"])
|
||||
}
|
||||
}
|
||||
|
||||
for _, r := range routes {
|
||||
want := strings.ReplaceAll(r.tmpl, "{{.symbol}}", "ETH")
|
||||
want = strings.ReplaceAll(want, "{{.price}}", "3200")
|
||||
content := waitLogContent(t, client, base, adminKey, srcName, r.event, r.marker, 15*time.Second)
|
||||
if content != want {
|
||||
t.Fatalf("event %s: content=%q want=%q", r.event, content, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func waitLogContent(t *testing.T, c *http.Client, base, adminKey, source, event, marker string, timeout time.Duration) string {
|
||||
t.Helper()
|
||||
q := url.Values{}
|
||||
q.Set("source", source)
|
||||
q.Set("event", event)
|
||||
q.Set("page", "1")
|
||||
q.Set("page_size", "20")
|
||||
path := "/api/v1/message-logs?" + q.Encode()
|
||||
|
||||
deadline := time.Now().Add(timeout)
|
||||
var last string
|
||||
for time.Now().Before(deadline) {
|
||||
logs := mustAdminJSON(t, c, base, adminKey, http.MethodGet, path, nil, http.StatusOK)
|
||||
data, _ := logs["data"].([]any)
|
||||
for _, item := range data {
|
||||
m, _ := item.(map[string]any)
|
||||
content, _ := m["content"].(string)
|
||||
if strings.Contains(content, marker) {
|
||||
return content
|
||||
}
|
||||
last = content
|
||||
}
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("no message log with marker %q for event %s (last=%q)", marker, event, last)
|
||||
return ""
|
||||
}
|
||||
|
||||
func envOr(k, def string) string {
|
||||
if v := os.Getenv(k); v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func intFrom(v any) int {
|
||||
switch n := v.(type) {
|
||||
case float64:
|
||||
return int(n)
|
||||
case json.Number:
|
||||
i, _ := n.Int64()
|
||||
return int(i)
|
||||
case int:
|
||||
return n
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
func mustAdminJSON(t *testing.T, c *http.Client, base, key, method, path string, body any, want int) map[string]any {
|
||||
t.Helper()
|
||||
return mustJSON(t, c, method, base+path, key, body, want)
|
||||
}
|
||||
|
||||
func mustJSON(t *testing.T, c *http.Client, method, urlStr, bearer string, body any, want int) map[string]any {
|
||||
t.Helper()
|
||||
code, raw, err := doJSON(c, method, urlStr, bearer, body)
|
||||
if err != nil {
|
||||
t.Fatalf("%s %s: %v", method, urlStr, err)
|
||||
}
|
||||
if code != want {
|
||||
t.Fatalf("%s %s: status %d want %d body=%s", method, urlStr, code, want, raw)
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
return map[string]any{}
|
||||
}
|
||||
var out map[string]any
|
||||
if err := json.Unmarshal(raw, &out); err != nil {
|
||||
t.Fatalf("decode: %v body=%s", err, raw)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func adminDo(c *http.Client, base, key, method, path string, body any) error {
|
||||
_, _, err := doJSON(c, method, base+path, key, body)
|
||||
return err
|
||||
}
|
||||
|
||||
func doJSON(c *http.Client, method, urlStr, bearer string, body any) (int, []byte, error) {
|
||||
var rdr io.Reader
|
||||
if body != nil {
|
||||
b, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
rdr = bytes.NewReader(b)
|
||||
}
|
||||
req, err := http.NewRequest(method, urlStr, rdr)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
if bearer != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+bearer)
|
||||
}
|
||||
if body != nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
resp, err := c.Do(req)
|
||||
if err != nil {
|
||||
return 0, nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
raw, err := io.ReadAll(resp.Body)
|
||||
return resp.StatusCode, raw, err
|
||||
}
|
||||
Reference in New Issue
Block a user