From 10fa91de6117de6801ada768a4f3512ea56a54aa Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 12 Sep 2026 03:14:30 +0000 Subject: [PATCH] feat(notification): emit deduplicated stage ready to advance events. Write outbox row and creator notification when a stage first becomes advance-ready. Co-authored-by: Cursor --- internal/notification/service.go | 156 ++++++++++++++++++++++++++ internal/notification/service_test.go | 122 ++++++++++++++++++++ internal/notification/types.go | 34 ++++++ 3 files changed, 312 insertions(+) create mode 100644 internal/notification/service.go create mode 100644 internal/notification/service_test.go create mode 100644 internal/notification/types.go diff --git a/internal/notification/service.go b/internal/notification/service.go new file mode 100644 index 00000000..7277090c --- /dev/null +++ b/internal/notification/service.go @@ -0,0 +1,156 @@ +package notification + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +var ErrNotAvailable = errors.New("notification: service not available") + +// Service Outbox 与站内通知。 +type Service struct { + pool *pgxpool.Pool +} + +func NewService(pool *pgxpool.Pool) *Service { + return &Service{pool: pool} +} + +func (s *Service) Available() bool { + if s == nil || s.pool == nil { + return false + } + var ok bool + _ = s.pool.QueryRow(context.Background(), ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'outbox_events' + ) + `).Scan(&ok) + return ok +} + +// EmitStageReadyToAdvance 写入 Outbox 与创建者通知；去重键 (stage_instance_id, readiness_revision)。 +func (s *Service) EmitStageReadyToAdvance(ctx context.Context, creatorUserID int64, payload StageReadyPayload) (bool, error) { + if !s.Available() { + return false, ErrNotAvailable + } + if creatorUserID <= 0 || payload.StageInstanceID <= 0 { + return false, nil + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return false, err + } + defer tx.Rollback(ctx) + + var advanceID int64 + err = tx.QueryRow(ctx, ` + INSERT INTO stage_advance_requests ( + stage_instance_id, workflow_instance_id, stage_key, readiness_revision, status + ) VALUES ($1, $2, $3, $4, 'pending') + ON CONFLICT (stage_instance_id, readiness_revision) DO NOTHING + RETURNING id + `, payload.StageInstanceID, payload.WorkflowInstanceID, payload.StageKey, payload.ConfigRevision).Scan(&advanceID) + if errors.Is(err, pgx.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + + raw, err := json.Marshal(payload) + if err != nil { + return false, err + } + + var eventID int64 + aggregateID := fmt.Sprintf("stage_instance:%d", payload.StageInstanceID) + err = tx.QueryRow(ctx, ` + INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload, project_id, published_at) + VALUES ('stage_instance', $1, $2, $3::jsonb, $4, now()) + RETURNING id + `, aggregateID, EventStageReadyToAdvance, string(raw), payload.ProjectID).Scan(&eventID) + if err != nil { + return false, err + } + + stageLabel := payload.StageName + if stageLabel == "" { + stageLabel = payload.StageKey + } + title := fmt.Sprintf("阶段「%s」可推进", stageLabel) + body := fmt.Sprintf( + "%d/%d 个必需文档已批准，没有阻断项。请确认是否推进到下一阶段。", + payload.ApprovedDocCount, payload.RequiredDocCount, + ) + dedupeKey := fmt.Sprintf("stage_ready:%d:%d", payload.StageInstanceID, payload.ConfigRevision) + + _, err = tx.Exec(ctx, ` + INSERT INTO notifications (user_id, project_id, event_id, type, title, body, dedupe_key, status) + VALUES ($1, $2, $3, $4, $5, $6, $7, 'unread') + ON CONFLICT (user_id, dedupe_key) DO NOTHING + `, creatorUserID, payload.ProjectID, eventID, TypeStageReadyToAdvance, title, body, dedupeKey) + if err != nil { + return false, err + } + + if err := tx.Commit(ctx); err != nil { + return false, err + } + return true, nil +} + +// ListProjectNotifications 列出项目下当前用户的通知（可选 since_id 增量）。 +func (s *Service) ListProjectNotifications(ctx context.Context, userID, projectID int64, sinceID int64, limit int) ([]Notification, error) { + if !s.Available() { + return nil, ErrNotAvailable + } + if limit <= 0 || limit > 100 { + limit = 50 + } + + query := ` + SELECT n.id, n.user_id, n.project_id, COALESCE(n.event_id, 0), n.type, n.title, n.body, n.status, n.created_at, + COALESCE(e.payload, '{}'::jsonb) + FROM notifications n + LEFT JOIN outbox_events e ON e.id = n.event_id + WHERE n.user_id = $1 AND n.project_id = $2 + ` + args := []any{userID, projectID} + if sinceID > 0 { + query += ` AND n.id > $3` + args = append(args, sinceID) + } + query += ` ORDER BY n.id ASC LIMIT $` + fmt.Sprint(len(args)+1) + args = append(args, limit) + + rows, err := s.pool.Query(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make([]Notification, 0) + for rows.Next() { + var n Notification + var payloadRaw []byte + if err := rows.Scan( + &n.ID, &n.UserID, &n.ProjectID, &n.EventID, &n.Type, &n.Title, &n.Body, &n.Status, &n.CreatedAt, + &payloadRaw, + ); err != nil { + return nil, err + } + if len(payloadRaw) > 0 { + _ = json.Unmarshal(payloadRaw, &n.Payload) + } + out = append(out, n) + } + return out, rows.Err() +} diff --git a/internal/notification/service_test.go b/internal/notification/service_test.go new file mode 100644 index 00000000..44cbfe4f --- /dev/null +++ b/internal/notification/service_test.go @@ -0,0 +1,122 @@ +package notification_test + +import ( + "context" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/db" + "chat-agent-system/internal/notification" +) + +func openPool(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("TEST_DATABASE_URL") + if dsn == "" { + dsn = "postgres://app:dev_secret_db_pass@127.0.0.1:5433/agent_system_dev?sslmode=disable" + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Skip(err) + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + t.Skip(err) + } + migCtx, migCancel := context.WithTimeout(context.Background(), 30*time.Second) + defer migCancel() + if err := db.RunMigrations(migCtx, pool, db.ResolveMigrationsDir()); err != nil { + pool.Close() + t.Skipf("migration failed: %v", err) + } + return pool +} + +func TestPG_EmitStageReadyDedupe(t *testing.T) { + pool := openPool(t) + defer pool.Close() + + svc := notification.NewService(pool) + if !svc.Available() { + t.Fatal("notification service should be available after migration") + } + + ctx := context.Background() + var projectID, instanceID, stageInstID int64 + code := "notify-" + time.Now().Format("150405.000") + err := pool.QueryRow(ctx, ` + INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status) + VALUES ($1, 'Notify Test', 'active', 2, 'active') RETURNING id + `, code).Scan(&projectID) + if err != nil { + t.Fatalf("insert project: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + }) + + var wfVersionID int64 + err = pool.QueryRow(ctx, `SELECT id FROM workflow_versions ORDER BY id DESC LIMIT 1`).Scan(&wfVersionID) + if err != nil { + t.Skip("no workflow version for notification test") + } + + err = pool.QueryRow(ctx, ` + INSERT INTO project_workflow_instances (project_id, workflow_version_id, instance_no, status, config_revision, plan_hash, config_snapshot, created_by) + VALUES ($1, $2, 1, 'active', 3, 'h', '{}'::jsonb, 2) RETURNING id + `, projectID, wfVersionID).Scan(&instanceID) + if err != nil { + t.Fatalf("insert instance: %v", err) + } + + err = pool.QueryRow(ctx, ` + INSERT INTO project_stage_instances (workflow_instance_id, stage_key, position, enabled, status) + VALUES ($1, 'biz_intake', 0, true, 'active') RETURNING id + `, instanceID).Scan(&stageInstID) + if err != nil { + t.Fatalf("insert stage: %v", err) + } + + payload := notification.StageReadyPayload{ + ProjectID: projectID, + WorkflowInstanceID: instanceID, + StageInstanceID: stageInstID, + StageKey: "biz_intake", + StageName: "商务摸底", + ConfigRevision: 3, + RequiredDocCount: 1, + ApprovedDocCount: 1, + } + + created, err := svc.EmitStageReadyToAdvance(ctx, 2, payload) + if err != nil { + t.Fatalf("emit: %v", err) + } + if !created { + t.Fatal("expected first emit to create notification") + } + + again, err := svc.EmitStageReadyToAdvance(ctx, 2, payload) + if err != nil { + t.Fatalf("second emit: %v", err) + } + if again { + t.Fatal("expected dedupe to suppress second notification") + } + + var count int + err = pool.QueryRow(ctx, ` + SELECT COUNT(*) FROM notifications WHERE user_id = 2 AND project_id = $1 AND type = $2 + `, projectID, notification.TypeStageReadyToAdvance).Scan(&count) + if err != nil { + t.Fatalf("count notifications: %v", err) + } + if count != 1 { + t.Fatalf("expected exactly one notification, got %d", count) + } +} diff --git a/internal/notification/types.go b/internal/notification/types.go new file mode 100644 index 00000000..4e6b7773 --- /dev/null +++ b/internal/notification/types.go @@ -0,0 +1,34 @@ +package notification + +import "time" + +const ( + EventStageReadyToAdvance = "stage.ready_to_advance" + TypeStageReadyToAdvance = "stage_ready_to_advance" +) + +// StageReadyPayload Outbox / 通知载荷。 +type StageReadyPayload struct { + ProjectID int64 `json:"project_id"` + WorkflowInstanceID int64 `json:"workflow_instance_id"` + StageInstanceID int64 `json:"stage_instance_id"` + StageKey string `json:"stage_key"` + StageName string `json:"stage_name"` + ConfigRevision int `json:"config_revision"` + RequiredDocCount int `json:"required_doc_count"` + ApprovedDocCount int `json:"approved_doc_count"` +} + +// Notification 用户站内通知。 +type Notification struct { + ID int64 `json:"id"` + UserID int64 `json:"user_id"` + ProjectID int64 `json:"project_id"` + EventID int64 `json:"event_id,omitempty"` + Type string `json:"type"` + Title string `json:"title"` + Body string `json:"body"` + Status string `json:"status"` + CreatedAt time.Time `json:"created_at"` + Payload StageReadyPayload `json:"payload,omitempty"` +}