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>
2026-09-12 03:14:30 +00:00
parent 00b0bb6882
commit 6fb28aabb7
3 changed files with 312 additions and 0 deletions
@@ -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"`
+}