Agent-System/internal/notification/reminder.go

package notification

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// stageReminderAfter 首次提醒后多久发送 24h 升级提醒（测试可覆盖）。
var stageReminderAfter = 24 * time.Hour

// SetStageReminderAfterForTest 仅供测试缩短等待窗口。
func SetStageReminderAfterForTest(d time.Duration) {
	if d > 0 {
		stageReminderAfter = d
	}
}

type pendingAdvanceRow struct {
	ID                 int64
	StageInstanceID    int64
	WorkflowInstanceID int64
	StageKey           string
	ReadinessRevision  int
	ProjectID          int64
	CreatorUserID      int64
	StageName          string
	RequiredDocCount   int
	ApprovedDocCount   int
}

// ProcessStageAdvanceReminders 扫描 pending 推进待办，对超过 24h 未处理的创建升级提醒（去重）。
func (s *Service) ProcessStageAdvanceReminders(ctx context.Context) (int, error) {
	if !s.Available() {
		return 0, ErrNotAvailable
	}

	cutoff := time.Now().Add(-stageReminderAfter)
	rows, err := s.pool.Query(ctx, `
		SELECT sar.id, sar.stage_instance_id, sar.workflow_instance_id, sar.stage_key, sar.readiness_revision,
		       pwi.project_id, COALESCE(p.creator_user_id, 0),
		       COALESCE(e.payload, '{}'::jsonb)
		FROM stage_advance_requests sar
		JOIN project_workflow_instances pwi ON pwi.id = sar.workflow_instance_id
		JOIN projects p ON p.id = pwi.project_id
		LEFT JOIN LATERAL (
			SELECT payload FROM outbox_events
			WHERE aggregate_type = 'stage_instance'
			  AND aggregate_id = 'stage_instance:' || sar.stage_instance_id::text
			  AND event_type = $1
			ORDER BY id DESC LIMIT 1
		) e ON true
		WHERE sar.status = 'pending' AND sar.requested_at <= $2
	`, EventStageReadyToAdvance, cutoff)
	if err != nil {
		return 0, err
	}
	defer rows.Close()

	sent := 0
	for rows.Next() {
		var row pendingAdvanceRow
		var payloadRaw []byte
		if err := rows.Scan(
			&row.ID, &row.StageInstanceID, &row.WorkflowInstanceID, &row.StageKey, &row.ReadinessRevision,
			&row.ProjectID, &row.CreatorUserID, &payloadRaw,
		); err != nil {
			return sent, err
		}
		if row.CreatorUserID <= 0 {
			continue
		}
		if len(payloadRaw) > 0 {
			var payload StageReadyPayload
			_ = json.Unmarshal(payloadRaw, &payload)
			row.StageName = payload.StageName
			row.RequiredDocCount = payload.RequiredDocCount
			row.ApprovedDocCount = payload.ApprovedDocCount
		}
		created, err := s.emitStageReminder24(ctx, row)
		if err != nil {
			return sent, err
		}
		if created {
			sent++
		}
	}
	return sent, rows.Err()
}

func (s *Service) emitStageReminder24(ctx context.Context, row pendingAdvanceRow) (bool, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return false, err
	}
	defer tx.Rollback(ctx)

	payload := StageReadyPayload{
		ProjectID:          row.ProjectID,
		WorkflowInstanceID: row.WorkflowInstanceID,
		StageInstanceID:    row.StageInstanceID,
		StageKey:           row.StageKey,
		StageName:          row.StageName,
		ConfigRevision:     row.ReadinessRevision,
		RequiredDocCount:   row.RequiredDocCount,
		ApprovedDocCount:   row.ApprovedDocCount,
	}
	raw, err := json.Marshal(payload)
	if err != nil {
		return false, err
	}

	dedupeKey := fmt.Sprintf("stage_ready_remind24:%d:%d", row.StageInstanceID, row.ReadinessRevision)
	var existingID int64
	err = tx.QueryRow(ctx, `
		SELECT id FROM notifications WHERE user_id = $1 AND dedupe_key = $2
	`, row.CreatorUserID, dedupeKey).Scan(&existingID)
	if err == nil && existingID > 0 {
		return false, nil
	}
	if err != nil && err != pgx.ErrNoRows {
		return false, err
	}

	var eventID int64
	aggregateID := fmt.Sprintf("stage_instance:%d", row.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, EventStageReadyReminder24, string(raw), row.ProjectID).Scan(&eventID)
	if err != nil {
		return false, err
	}

	stageLabel := row.StageName
	if stageLabel == "" {
		stageLabel = row.StageKey
	}
	title := fmt.Sprintf("【提醒】阶段「%s」仍可推进", stageLabel)
	body := "必需文档已全部批准，已超过 24 小时尚未确认推进。请尽快查看完成清单并确认是否进入下一阶段。"

	_, 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
	`, row.CreatorUserID, row.ProjectID, eventID, TypeStageReadyReminder24, title, body, dedupeKey)
	if err != nil {
		return false, err
	}

	if err := tx.Commit(ctx); err != nil {
		return false, err
	}
	return true, nil
}

// ConfirmStageAdvanceRequest 阶段已确认推进后关闭 pending 待办，停止后续提醒。
func (s *Service) ConfirmStageAdvanceRequest(ctx context.Context, stageInstanceID int64, readinessRevision int, decidedBy int64) error {
	if !s.Available() || stageInstanceID <= 0 {
		return nil
	}
	_, err := s.pool.Exec(ctx, `
		UPDATE stage_advance_requests
		SET status = 'confirmed', decided_by = $3, decided_at = now()
		WHERE stage_instance_id = $1 AND readiness_revision = $2 AND status = 'pending'
	`, stageInstanceID, readinessRevision, decidedBy)
	return err
}