Agent-System/internal/notification/digest.go

package notification

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

const (
	DigestKindProposalRequired   = "proposal_required"
	DigestKindUnclaimedEdit      = "unclaimed_edit"
	DigestKindPendingSupplements = "pending_supplements"
	DigestKindReviewRequested    = "review_requested"
	DigestKindApprovalRequested  = "approval_requested"
	SeverityActionRequired       = "action_required"
	SeverityInfo                 = "info"
	maxDigestItems               = 20
)

// DigestItem is one occurrence folded into a daily digest.
type DigestItem struct {
	At          time.Time `json:"at"`
	ActorID     int64     `json:"actor_id"`
	ActorName   string    `json:"actor_name"`
	Summary     string    `json:"summary"`
	ProposalID  string    `json:"proposal_id,omitempty"`
	ChangeCount int       `json:"change_count,omitempty"`
}

// UpsertDigestParams drives ON CONFLICT DO UPDATE aggregation.
type UpsertDigestParams struct {
	UserID      int64
	ProjectID   int64
	DigestKind  string
	DocID       string
	DocTitle    string
	Severity    string
	ActorID     int64
	ActorName   string
	Summary     string
	ProposalID  string
	ChangeCount int
	WindowDate  time.Time // truncated to date; zero → today UTC
}

// UpsertDocumentDigest aggregates same-day notifications per (user, kind, doc).
func (s *Service) UpsertDocumentDigest(ctx context.Context, p UpsertDigestParams) error {
	if !s.Available() {
		return ErrNotAvailable
	}
	if p.UserID <= 0 || p.DocID == "" || p.DigestKind == "" {
		return fmt.Errorf("notification: digest requires user, doc, kind")
	}
	if p.Severity == "" {
		p.Severity = SeverityInfo
	}
	window := p.WindowDate
	if window.IsZero() {
		now := time.Now().UTC()
		window = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
	}
	item := DigestItem{
		At: time.Now().UTC(), ActorID: p.ActorID, ActorName: p.ActorName,
		Summary: truncateRunes(p.Summary, 60), ProposalID: p.ProposalID, ChangeCount: p.ChangeCount,
	}
	itemJSON, _ := json.Marshal(item)

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)

	var digestID, notifID int64
	var count int
	err = tx.QueryRow(ctx, `
		INSERT INTO notification_digests (
		  user_id, digest_kind, doc_id, window_date, occurrence_count,
		  last_actor_id, last_actor_name, last_at, item_summaries, truncated_count
		) VALUES ($1,$2,$3,$4,1,$5,$6,now(), jsonb_build_array($7::jsonb), 0)
		ON CONFLICT (user_id, digest_kind, doc_id, window_date) DO UPDATE SET
		  occurrence_count = notification_digests.occurrence_count + 1,
		  last_actor_id = EXCLUDED.last_actor_id,
		  last_actor_name = EXCLUDED.last_actor_name,
		  last_at = now(),
		  item_summaries = CASE
		    WHEN jsonb_array_length(notification_digests.item_summaries) >= $8 THEN notification_digests.item_summaries
		    ELSE notification_digests.item_summaries || EXCLUDED.item_summaries
		  END,
		  truncated_count = CASE
		    WHEN jsonb_array_length(notification_digests.item_summaries) >= $8
		      THEN notification_digests.truncated_count + 1
		    ELSE notification_digests.truncated_count
		  END,
		  updated_at = now()
		RETURNING id, occurrence_count, COALESCE(notification_id, 0)
	`, p.UserID, p.DigestKind, p.DocID, window, p.ActorID, p.ActorName, string(itemJSON), maxDigestItems).
		Scan(&digestID, &count, &notifID)
	if err != nil {
		return err
	}

	title, body := digestCopy(p, count)
	dedupeKey := fmt.Sprintf("digest:%s:%s:%s", p.DigestKind, p.DocID, window.Format("2006-01-02"))
	notifType := typeForDigestKind(p.DigestKind)

	if notifID == 0 {
		err = tx.QueryRow(ctx, `
			INSERT INTO notifications (user_id, project_id, type, title, body, dedupe_key, status, digest_id, severity)
			VALUES ($1,$2,$3,$4,$5,$6,'unread',$7,$8)
			ON CONFLICT (user_id, dedupe_key) DO UPDATE SET
			  title = EXCLUDED.title,
			  body = EXCLUDED.body,
			  status = 'unread',
			  severity = EXCLUDED.severity,
			  digest_id = EXCLUDED.digest_id
			RETURNING id
		`, p.UserID, p.ProjectID, notifType, title, body, dedupeKey, digestID, p.Severity).Scan(&notifID)
		if err != nil {
			return err
		}
		_, err = tx.Exec(ctx, `UPDATE notification_digests SET notification_id=$1 WHERE id=$2`, notifID, digestID)
		if err != nil {
			return err
		}
	} else {
		_, err = tx.Exec(ctx, `
			UPDATE notifications SET title=$1, body=$2, status='unread', severity=$3
			WHERE id=$4 AND user_id=$5
		`, title, body, p.Severity, notifID, p.UserID)
		if err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func typeForDigestKind(kind string) string {
	switch kind {
	case DigestKindProposalRequired:
		return TypeDocProposalRequired
	case DigestKindUnclaimedEdit:
		return TypeDocUnclaimedEditApplied
	case DigestKindPendingSupplements:
		return TypeDocPendingSupplements
	case DigestKindReviewRequested:
		return TypeDocReviewRequested
	case DigestKindApprovalRequested:
		return TypeDocApprovalRequested
	default:
		return kind
	}
}

func digestCopy(p UpsertDigestParams, count int) (title, body string) {
	titleDoc := strings.TrimSpace(p.DocTitle)
	if titleDoc == "" {
		titleDoc = p.DocID
	}
	actor := strings.TrimSpace(p.ActorName)
	if actor == "" {
		actor = "成员"
	}
	summary := truncateRunes(p.Summary, 60)
	switch p.DigestKind {
	case DigestKindProposalRequired:
		if count <= 1 {
			return fmt.Sprintf("《%s》有变更待您确认", titleDoc),
				fmt.Sprintf("%s 补充了「%s」，拟修改本文档。文档已被您认领，需您确认后由您发起修改。", actor, summary)
		}
		return fmt.Sprintf("《%s》累计 %d 次变更待您确认", titleDoc, count),
			fmt.Sprintf("最近一次：%s。%d 次变更均来自他人补充材料，请逐份拍板。", actor, count)
	case DigestKindUnclaimedEdit:
		if count <= 1 {
			return fmt.Sprintf("《%s》已被更新（无需您确认）", titleDoc),
				fmt.Sprintf("%s 通过助手更新了本文档，来源：%s。本文档尚未被您认领，系统已直接应用。", actor, summary)
		}
		return fmt.Sprintf("《%s》今日已有 %d 次更新", titleDoc, count),
			fmt.Sprintf("最近一次：%s；累计 %d 次更新。本文档尚未被您认领。", actor, count)
	case DigestKindPendingSupplements:
		if count <= 1 {
			return fmt.Sprintf("《%s》有补充信息待消化", titleDoc),
				fmt.Sprintf("%s 在文档锁定期间补充了「%s」。文档已回编制期，请审阅并消化积压补充。", actor, summary)
		}
		return fmt.Sprintf("《%s》累计 %d 条补充待消化", titleDoc, count),
			fmt.Sprintf("最近一次：%s。文档已回编制期，请逐条审阅积压补充。", actor)
	case DigestKindReviewRequested:
		return fmt.Sprintf("《%s》待您审核", titleDoc),
			fmt.Sprintf("%s 已送审，请审核本文档。", actor)
	case DigestKindApprovalRequested:
		return fmt.Sprintf("《%s》待您批准", titleDoc),
			fmt.Sprintf("%s 审核已通过，请批准本文档。", actor)
	default:
		return titleDoc, summary
	}
}

func truncateRunes(s string, n int) string {
	r := []rune(strings.TrimSpace(s))
	if len(r) <= n {
		return string(r)
	}
	return string(r[:n]) + "…"
}

// Ensure unused import guard for pgx when scanning with interfaces.