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, ¬ifID) 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(¬ifID) 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.