From 39f377494031408d9321d2a09e03eb6f3c8a1688 Mon Sep 17 00:00:00 2001 From: ryan Date: Mon, 10 Aug 2026 00:34:26 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=85=8D=E7=BD=AE=E5=88=97=E8=A1=A8?= =?UTF-8?q?=E5=88=86=E9=A1=B5=E4=B8=8E=E9=92=89=E9=92=89=E6=9C=BA=E5=99=A8?= =?UTF-8?q?=E4=BA=BA=E5=88=86=E9=92=9F=E7=BA=A7=E6=8E=92=E9=98=9F=E9=99=90?= =?UTF-8?q?=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 统一 sources/templates/channels/rules 列表为分页响应，避免配置增多时全量返回；按钉钉 access_token 限制每分钟发送并在超限时等待下一分钟，降低触发官方封禁风险。 Co-authored-by: Cursor --- Makefile | 5 +- README.md | 27 +- cmd/server/main.go | 12 +- config/config.yaml | 2 + docs/httpie/curls.md | 8 +- .../2026-08-01-remote-e2e-notify-flow.md | 294 ++++++++++ .../2025-06-27-notification-service-design.md | 8 +- ...026-08-01-remote-e2e-notify-flow-design.md | 119 ++++ internal/adapter/adapter.go | 4 +- internal/adapter/dingtalk.go | 11 +- internal/adapter/dingtalk_limit.go | 141 +++++ internal/adapter/dingtalk_limit_test.go | 99 ++++ internal/cache/dingtalk_limit.go | 60 +++ internal/config/config.go | 3 +- internal/handler/channel.go | 11 +- internal/handler/message_log.go | 1 + internal/handler/rule.go | 33 +- internal/handler/source.go | 11 +- internal/handler/template.go | 11 +- internal/store/channel.go | 16 +- internal/store/message_log.go | 19 +- internal/store/page.go | 22 + internal/store/page_test.go | 30 ++ internal/store/rule.go | 18 +- internal/store/source.go | 18 +- internal/store/template.go | 18 +- test/e2e/notify_flow_test.go | 506 ++++++++++++++++++ 27 files changed, 1447 insertions(+), 60 deletions(-) create mode 100644 docs/superpowers/plans/2026-08-01-remote-e2e-notify-flow.md create mode 100644 docs/superpowers/specs/2026-08-01-remote-e2e-notify-flow-design.md create mode 100644 internal/adapter/dingtalk_limit.go create mode 100644 internal/adapter/dingtalk_limit_test.go create mode 100644 internal/cache/dingtalk_limit.go create mode 100644 internal/store/page.go create mode 100644 internal/store/page_test.go create mode 100644 test/e2e/notify_flow_test.go diff --git a/Makefile b/Makefile index ca20c1a..07a2944 100644 --- a/Makefile +++ b/Makefile @@ -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 . diff --git a/README.md b/README.md index 16ffeed..e5ffddb 100644 --- a/README.md +++ b/README.md @@ -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` — 更新 diff --git a/cmd/server/main.go b/cmd/server/main.go index 04e9ba9..2fa45c4 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -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) diff --git a/config/config.yaml b/config/config.yaml index 1bcc820..35f4946 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -25,6 +25,8 @@ smtp: rate_limit: default: 100 + # 钉钉自定义机器人官方上限 20/分钟，超限封 10 分钟；本地默认 18 留余量 + dingtalk_per_min: 18 logbull: host: "https://log.516886.xyz" diff --git a/docs/httpie/curls.md b/docs/httpie/curls.md index 154d40c..bc07285 100644 --- a/docs/httpie/curls.md +++ b/docs/httpie/curls.md @@ -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' ``` diff --git a/docs/superpowers/plans/2026-08-01-remote-e2e-notify-flow.md b/docs/superpowers/plans/2026-08-01-remote-e2e-notify-flow.md new file mode 100644 index 0000000..5282787 --- /dev/null +++ b/docs/superpowers/plans/2026-08-01-remote-e2e-notify-flow.md @@ -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 | diff --git a/docs/superpowers/specs/2025-06-27-notification-service-design.md b/docs/superpowers/specs/2025-06-27-notification-service-design.md index 5dfb38f..a48499f 100644 --- a/docs/superpowers/specs/2025-06-27-notification-service-design.md +++ b/docs/superpowers/specs/2025-06-27-notification-service-design.md @@ -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 diff --git a/docs/superpowers/specs/2026-08-01-remote-e2e-notify-flow-design.md b/docs/superpowers/specs/2026-08-01-remote-e2e-notify-flow-design.md new file mode 100644 index 0000000..e576018 --- /dev/null +++ b/docs/superpowers/specs/2026-08-01-remote-e2e-notify-flow-design.md @@ -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/` | +| `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-`. + +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 ` + - 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/' \ +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. diff --git a/internal/adapter/adapter.go b/internal/adapter/adapter.go index 75e0ce2..997d89d 100644 --- a/internal/adapter/adapter.go +++ b/internal/adapter/adapter.go @@ -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": diff --git a/internal/adapter/dingtalk.go b/internal/adapter/dingtalk.go index dd94c38..2149107 100644 --- a/internal/adapter/dingtalk.go +++ b/internal/adapter/dingtalk.go @@ -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() diff --git a/internal/adapter/dingtalk_limit.go b/internal/adapter/dingtalk_limit.go new file mode 100644 index 0000000..fabcf20 --- /dev/null +++ b/internal/adapter/dingtalk_limit.go @@ -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) +} diff --git a/internal/adapter/dingtalk_limit_test.go b/internal/adapter/dingtalk_limit_test.go new file mode 100644 index 0000000..9f906de --- /dev/null +++ b/internal/adapter/dingtalk_limit_test.go @@ -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) + } +} diff --git a/internal/cache/dingtalk_limit.go b/internal/cache/dingtalk_limit.go new file mode 100644 index 0000000..2a0f313 --- /dev/null +++ b/internal/cache/dingtalk_limit.go @@ -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) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 6b602d5..6ec9344 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -55,7 +55,8 @@ type SMTPConfig struct { } type RateLimitConfig struct { - Default int `mapstructure:"default"` + Default int `mapstructure:"default"` + DingTalkPerMin int `mapstructure:"dingtalk_per_min"` // per robot webhook; 0 => 18 } type LogbullConfig struct { diff --git a/internal/handler/channel.go b/internal/handler/channel.go index aabf6f2..3b26c56 100644 --- a/internal/handler/channel.go +++ b/internal/handler/channel.go @@ -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) { diff --git a/internal/handler/message_log.go b/internal/handler/message_log.go index ccb1cdd..1a2521e 100644 --- a/internal/handler/message_log.go +++ b/internal/handler/message_log.go @@ -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 { diff --git a/internal/handler/rule.go b/internal/handler/rule.go index 0e1958d..b4942af 100644 --- a/internal/handler/rule.go +++ b/internal/handler/rule.go @@ -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}) } diff --git a/internal/handler/source.go b/internal/handler/source.go index 8f489d0..f1a2f38 100644 --- a/internal/handler/source.go +++ b/internal/handler/source.go @@ -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) { diff --git a/internal/handler/template.go b/internal/handler/template.go index b33062f..14935d1 100644 --- a/internal/handler/template.go +++ b/internal/handler/template.go @@ -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) { diff --git a/internal/store/channel.go b/internal/store/channel.go index a028554..fe9eaab 100644 --- a/internal/store/channel.go +++ b/internal/store/channel.go @@ -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 { diff --git a/internal/store/message_log.go b/internal/store/message_log.go index 1e71d6e..aa68fd3 100644 --- a/internal/store/message_log.go +++ b/internal/store/message_log.go @@ -8,11 +8,10 @@ import ( ) type MessageLogFilter struct { - Source string `form:"source"` - Event string `form:"event"` - Status string `form:"status"` - Page int `form:"page"` - PageSize int `form:"page_size"` + Source string `form:"source"` + Event string `form:"event"` + Status string `form:"status"` + 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 } diff --git a/internal/store/page.go b/internal/store/page.go new file mode 100644 index 0000000..6fe9a0a --- /dev/null +++ b/internal/store/page.go @@ -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 +} diff --git a/internal/store/page_test.go b/internal/store/page_test.go new file mode 100644 index 0000000..245a6e9 --- /dev/null +++ b/internal/store/page_test.go @@ -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) + } + }) + } +} diff --git a/internal/store/rule.go b/internal/store/rule.go index acfb6dd..d86347b 100644 --- a/internal/store/rule.go +++ b/internal/store/rule.go @@ -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 { diff --git a/internal/store/source.go b/internal/store/source.go index e2fe63a..83c9ded 100644 --- a/internal/store/source.go +++ b/internal/store/source.go @@ -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 { diff --git a/internal/store/template.go b/internal/store/template.go index b7a3d31..9bee7f8 100644 --- a/internal/store/template.go +++ b/internal/store/template.go @@ -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 { diff --git a/test/e2e/notify_flow_test.go b/test/e2e/notify_flow_test.go new file mode 100644 index 0000000..8a134aa --- /dev/null +++ b/test/e2e/notify_flow_test.go @@ -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 +}