You've already forked agentic-coding-workflow
243 lines
7.5 KiB
Go
243 lines
7.5 KiB
Go
// Package notify 内的 handler.go 是通知 inbox 的 HTTP 接入层。它把已经
|
|
// 鉴权完成的请求路由到 Repository 的对应方法，并按统一的 response.go
|
|
// 风格（Google API 风格错误体）返回。
|
|
//
|
|
// 端点（统一前缀 /api/v1/notifications，全部需要登录）：
|
|
//
|
|
// GET / → 列表，参数 ?unread=true&limit=N
|
|
// GET /unread-count → 未读计数
|
|
// PATCH /read-all → 全部标已读
|
|
// PATCH /{id}/read → 单条标已读
|
|
package notify
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
|
|
httpx "github.com/yan1h/agent-coding-workflow/internal/transport/http"
|
|
"github.com/yan1h/agent-coding-workflow/internal/transport/http/middleware"
|
|
)
|
|
|
|
// Encryptor 是 prefs handler 加密 im_webhook_url 所需的窄接口，由 crypto.Encryptor
|
|
// 满足。以接口注入避免 notify 依赖 crypto 包。
|
|
type Encryptor interface {
|
|
Encrypt(plaintext []byte) ([]byte, error)
|
|
}
|
|
|
|
// Handler 暴露 notify 模块的 HTTP 接口；依赖 Repository 与 SessionResolver。
|
|
// 不直接持有 Dispatcher——发送通知由其它模块通过 Dispatcher 完成，handler
|
|
// 仅服务 inbox 读取与状态变更。
|
|
type Handler struct {
|
|
repo Repository
|
|
resolver middleware.SessionResolver
|
|
// hub 为可选的实时推送中心；非 nil 时暴露 GET /stream SSE 端点。
|
|
hub StreamHub
|
|
// prefs/enc 为可选的偏好仓库与加密器；二者均非 nil 时暴露 GET/PUT /prefs。
|
|
prefs PrefsRepository
|
|
enc Encryptor
|
|
}
|
|
|
|
// NewHandler 构造 notify 模块的 HTTP handler。
|
|
func NewHandler(repo Repository, resolver middleware.SessionResolver) *Handler {
|
|
return &Handler{repo: repo, resolver: resolver}
|
|
}
|
|
|
|
// WithStreamHub 注入实时推送中心，启用 GET /api/v1/notifications/stream SSE。
|
|
func (h *Handler) WithStreamHub(hub StreamHub) *Handler {
|
|
h.hub = hub
|
|
return h
|
|
}
|
|
|
|
// WithPrefs 注入偏好仓库与加密器，启用 GET/PUT /api/v1/notifications/prefs。
|
|
func (h *Handler) WithPrefs(prefs PrefsRepository, enc Encryptor) *Handler {
|
|
h.prefs = prefs
|
|
h.enc = enc
|
|
return h
|
|
}
|
|
|
|
// Mount 把 inbox 路由挂到给定 chi.Router 上。Auth 中间件在子路由内统一注入。
|
|
func (h *Handler) Mount(r chi.Router) {
|
|
r.Route("/api/v1/notifications", func(r chi.Router) {
|
|
r.Use(middleware.Auth(h.resolver, middleware.AuthOptions{}))
|
|
r.Get("/", h.list)
|
|
r.Get("/unread-count", h.unreadCount)
|
|
r.Patch("/read-all", h.readAll)
|
|
r.Patch("/{id}/read", h.markRead)
|
|
if h.hub != nil {
|
|
r.Get("/stream", h.stream)
|
|
}
|
|
if h.prefs != nil && h.enc != nil {
|
|
r.Get("/prefs", h.getPrefs)
|
|
r.Put("/prefs", h.putPrefs)
|
|
}
|
|
})
|
|
}
|
|
|
|
// stream 实现 GET /api/v1/notifications/stream：把该用户的实时通知以 SSE 推送。
|
|
// 客户端用 EventSource 订阅；连接关闭时（r.Context().Done）自动退订。
|
|
func (h *Handler) stream(w http.ResponseWriter, r *http.Request) {
|
|
uid, ok := middleware.UserIDFromContext(r.Context())
|
|
if !ok {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()),
|
|
errs.New(errs.CodeUnauthorized, "未认证"))
|
|
return
|
|
}
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
ch, cancel := h.hub.Subscribe(uid)
|
|
defer cancel()
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
w.WriteHeader(http.StatusOK)
|
|
flusher.Flush()
|
|
|
|
for {
|
|
select {
|
|
case <-r.Context().Done():
|
|
return
|
|
case m, ok := <-ch:
|
|
if !ok {
|
|
return
|
|
}
|
|
writeNotificationSSE(w, flusher, m)
|
|
}
|
|
}
|
|
}
|
|
|
|
// writeNotificationSSE 把一条通知序列化为 SSE 帧并 flush。
|
|
func writeNotificationSSE(w http.ResponseWriter, f http.Flusher, m Message) {
|
|
dto := notificationDTO{
|
|
ID: m.ID.String(),
|
|
Topic: m.Topic,
|
|
Severity: string(m.Severity),
|
|
Title: m.Title,
|
|
Body: m.Body,
|
|
Link: m.Link,
|
|
Metadata: m.Metadata,
|
|
CreatedAt: m.CreatedAt.UTC().Format(time.RFC3339),
|
|
}
|
|
if m.ReadAt != nil {
|
|
s := m.ReadAt.UTC().Format(time.RFC3339)
|
|
dto.ReadAt = &s
|
|
}
|
|
data, _ := json.Marshal(dto)
|
|
fmt.Fprintf(w, "event: notification\ndata: %s\n\n", data)
|
|
f.Flush()
|
|
}
|
|
|
|
// notificationDTO 是 inbox 单条通知的对外 JSON 形态。created_at/read_at
|
|
// 统一序列化为 RFC3339 字符串；read_at 为 nil 时省略（omitempty）。
|
|
type notificationDTO struct {
|
|
ID string `json:"id"`
|
|
Topic string `json:"topic"`
|
|
Severity string `json:"severity"`
|
|
Title string `json:"title"`
|
|
Body string `json:"body"`
|
|
Link string `json:"link,omitempty"`
|
|
Metadata map[string]any `json:"metadata,omitempty"`
|
|
ReadAt *string `json:"read_at,omitempty"`
|
|
CreatedAt string `json:"created_at"`
|
|
}
|
|
|
|
func (h *Handler) list(w http.ResponseWriter, r *http.Request) {
|
|
uid, ok := middleware.UserIDFromContext(r.Context())
|
|
if !ok {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()),
|
|
errs.New(errs.CodeUnauthorized, "未认证"))
|
|
return
|
|
}
|
|
unread := r.URL.Query().Get("unread") == "true"
|
|
limit := 50
|
|
if v := r.URL.Query().Get("limit"); v != "" {
|
|
if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 200 {
|
|
limit = n
|
|
}
|
|
}
|
|
msgs, err := h.repo.List(r.Context(), uid, unread, limit)
|
|
if err != nil {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err)
|
|
return
|
|
}
|
|
out := make([]notificationDTO, 0, len(msgs))
|
|
for _, m := range msgs {
|
|
dto := notificationDTO{
|
|
ID: m.ID.String(),
|
|
Topic: m.Topic,
|
|
Severity: string(m.Severity),
|
|
Title: m.Title,
|
|
Body: m.Body,
|
|
Link: m.Link,
|
|
Metadata: m.Metadata,
|
|
CreatedAt: m.CreatedAt.UTC().Format(time.RFC3339),
|
|
}
|
|
if m.ReadAt != nil {
|
|
s := m.ReadAt.UTC().Format(time.RFC3339)
|
|
dto.ReadAt = &s
|
|
}
|
|
out = append(out, dto)
|
|
}
|
|
httpx.WriteJSON(w, http.StatusOK, map[string]any{"items": out})
|
|
}
|
|
|
|
func (h *Handler) unreadCount(w http.ResponseWriter, r *http.Request) {
|
|
uid, ok := middleware.UserIDFromContext(r.Context())
|
|
if !ok {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()),
|
|
errs.New(errs.CodeUnauthorized, "未认证"))
|
|
return
|
|
}
|
|
n, err := h.repo.CountUnread(r.Context(), uid)
|
|
if err != nil {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err)
|
|
return
|
|
}
|
|
httpx.WriteJSON(w, http.StatusOK, map[string]any{"count": n})
|
|
}
|
|
|
|
func (h *Handler) markRead(w http.ResponseWriter, r *http.Request) {
|
|
uid, ok := middleware.UserIDFromContext(r.Context())
|
|
if !ok {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()),
|
|
errs.New(errs.CodeUnauthorized, "未认证"))
|
|
return
|
|
}
|
|
id, err := uuid.Parse(chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()),
|
|
errs.New(errs.CodeInvalidInput, "id 格式错误"))
|
|
return
|
|
}
|
|
if err := h.repo.MarkRead(r.Context(), uid, id); err != nil {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err)
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
func (h *Handler) readAll(w http.ResponseWriter, r *http.Request) {
|
|
uid, ok := middleware.UserIDFromContext(r.Context())
|
|
if !ok {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()),
|
|
errs.New(errs.CodeUnauthorized, "未认证"))
|
|
return
|
|
}
|
|
if err := h.repo.MarkAllRead(r.Context(), uid); err != nil {
|
|
httpx.WriteError(w, middleware.RequestIDFromContext(r.Context()), err)
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|