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 <cursoragent@cursor.com>
This commit is contained in:
weichen
committed by
weichen
weichen
co-authored by
Cursor
Cursor
parent
ff81170f47
commit
10fa91de61
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
Reference in New Issue
Block a user