package notification import ( "context" "encoding/json" "fmt" "github.com/jackc/pgx/v5" ) const ( EventDocAgentCandidateReady = "document.agent_candidate_ready" EventDocAgentVersionCommitted = "document.agent_version_committed" EventDocHumanRevisionConflict = "document.human_revision_conflict" EventDocDownstreamImpact = "document.downstream_impact_detected" EventDocChangeReportDegraded = "document.change_report_degraded" EventDocProposalRequired = "document.proposal_required" EventDocUnclaimedEditApplied = "document.unclaimed_edit_applied" EventDocReviewRequested = "document.review_requested" EventDocApprovalRequested = "document.approval_requested" EventDocRejected = "document.rejected" EventDocReopened = "document.reopened" EventDocPendingSupplements = "document.pending_supplements" EventDocApproved = "document.approved" TypeDocAgentCandidateReady = "document_agent_candidate_ready" TypeDocAgentVersionCommitted = "document_agent_version_committed" TypeDocDownstreamImpact = "document_downstream_impact" TypeDocChangeReportDegraded = "document_change_report_degraded" TypeDocApproved = "document_approved" TypeDocProposalRequired = "document_proposal_required" TypeDocUnclaimedEditApplied = "document_unclaimed_edit_applied" TypeDocReviewRequested = "document_review_requested" TypeDocApprovalRequested = "document_approval_requested" TypeDocRejected = "document_rejected" TypeDocReopened = "document_reopened" TypeDocPendingSupplements = "document_pending_supplements" ) // DocumentEventPayload is a minimal, permission-safe notification body (no markdown excerpts). type DocumentEventPayload struct { ProjectID int64 `json:"project_id"` DocID string `json:"doc_id"` DocTitle string `json:"doc_title,omitempty"` VersionNo int `json:"version_no,omitempty"` CandidateID int64 `json:"candidate_id,omitempty"` ReportID int64 `json:"report_id,omitempty"` RiskLevel string `json:"risk_level,omitempty"` Summary string `json:"summary,omitempty"` ImpactedDocID string `json:"impacted_doc_id,omitempty"` ImpactedCount int `json:"impacted_count,omitempty"` RestrictedN int `json:"restricted_count,omitempty"` } // EmitDocumentEvent writes outbox + per-recipient notifications with dedupe. func (s *Service) EmitDocumentEvent(ctx context.Context, eventType, notifType, title, body, dedupeKey string, recipientUserID int64, payload DocumentEventPayload) (bool, error) { if !s.Available() { return false, ErrNotAvailable } if recipientUserID <= 0 || payload.DocID == "" { return false, nil } tx, err := s.pool.Begin(ctx) if err != nil { return false, err } defer tx.Rollback(ctx) raw, err := json.Marshal(payload) if err != nil { return false, err } var eventID int64 aggregateID := fmt.Sprintf("document:%s", payload.DocID) err = tx.QueryRow(ctx, ` INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload, project_id, published_at) VALUES ('document', $1, $2, $3::jsonb, $4, now()) RETURNING id `, aggregateID, eventType, string(raw), payload.ProjectID).Scan(&eventID) if err != nil { return false, err } tag, 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 `, recipientUserID, payload.ProjectID, eventID, notifType, title, body, dedupeKey) if err != nil { return false, err } if err := tx.Commit(ctx); err != nil { return false, err } return tag.RowsAffected() > 0, nil } // VisibleDedupeKey is the canonical inbox identity for a durable envelope. func VisibleDedupeKey(sourceEventID, notifType string, recipientUserID int64) string { return fmt.Sprintf("evt:%s:%s:%d", sourceEventID, notifType, recipientUserID) } // EmitVisibleDocumentEvent writes the unique visible notification on the caller's // fenced transaction. A replay that hits (user_id, dedupe_key) is success. func (s *Service) EmitVisibleDocumentEvent(ctx context.Context, tx pgx.Tx, sourceEventID, notifType, title, body string, recipientUserID int64, payload DocumentEventPayload) error { if tx == nil || recipientUserID <= 0 || sourceEventID == "" || notifType == "" { return fmt.Errorf("notification: visible outcome requires source event, type, and recipient") } dedupeKey := VisibleDedupeKey(sourceEventID, notifType, recipientUserID) _, err := tx.Exec(ctx, ` INSERT INTO notifications (user_id, project_id, type, title, body, dedupe_key, status, source_event_id) VALUES ($1,$2,$3,$4,$5,$6,'unread',$7) ON CONFLICT (user_id, dedupe_key) DO NOTHING `, recipientUserID, payload.ProjectID, notifType, title, body, dedupeKey, sourceEventID) return err } // UpdateNotificationStatus marks a notification read/acted/dismissed for the owner. func (s *Service) UpdateNotificationStatus(ctx context.Context, userID, notifID int64, status string) error { if !s.Available() { return ErrNotAvailable } switch status { case "read", "acted", "dismissed", "unread": default: return fmt.Errorf("notification: invalid status %q", status) } tag, err := s.pool.Exec(ctx, ` UPDATE notifications SET status = $1 WHERE id = $2 AND user_id = $3 `, status, notifID, userID) if err != nil { return err } if tag.RowsAffected() == 0 { return fmt.Errorf("notification: not found") } return nil }