221 lines
6.8 KiB
Go
221 lines
6.8 KiB
Go
package notification
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"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
|
|
}
|
|
|
|
// EmitVisibleStageReady writes the unique stage-ready request on the caller's fenced tx.
|
|
// Duplicate (stage, revision) or replayed source_event_id is success.
|
|
func (s *Service) EmitVisibleStageReady(ctx context.Context, tx pgx.Tx, sourceEventID string, creatorUserID int64, payload StageReadyPayload) error {
|
|
if tx == nil || sourceEventID == "" || creatorUserID <= 0 || payload.StageInstanceID <= 0 {
|
|
return fmt.Errorf("notification: visible stage ready requires source event, creator, and stage")
|
|
}
|
|
var advanceID int64
|
|
err := tx.QueryRow(ctx, `
|
|
INSERT INTO stage_advance_requests (
|
|
stage_instance_id, workflow_instance_id, stage_key, readiness_revision, status, source_event_id
|
|
) VALUES ($1, $2, $3, $4, 'pending', $5)
|
|
ON CONFLICT (stage_instance_id, readiness_revision) DO NOTHING
|
|
RETURNING id
|
|
`, payload.StageInstanceID, payload.WorkflowInstanceID, payload.StageKey, payload.ConfigRevision, sourceEventID).Scan(&advanceID)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return ignoreUnique(err)
|
|
}
|
|
|
|
raw, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var eventID int64
|
|
aggregateID := fmt.Sprintf("stage_instance:%d", payload.StageInstanceID)
|
|
if 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); err != nil {
|
|
return 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, source_event_id)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,'unread',$8)
|
|
ON CONFLICT (user_id, dedupe_key) DO NOTHING
|
|
`, creatorUserID, payload.ProjectID, eventID, TypeStageReadyToAdvance, title, body, dedupeKey, sourceEventID)
|
|
return ignoreUnique(err)
|
|
}
|
|
|
|
func ignoreUnique(err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
var pgErr *pgconn.PgError
|
|
if errors.As(err, &pgErr) && pgErr.Code == "23505" {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
// 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()
|
|
}
|