Agent-System/internal/notification/document_events.go

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"

	TypeDocAgentCandidateReady   = "document_agent_candidate_ready"
	TypeDocAgentVersionCommitted = "document_agent_version_committed"
	TypeDocDownstreamImpact      = "document_downstream_impact"
	TypeDocChangeReportDegraded  = "document_change_report_degraded"
	TypeDocApproved              = "document_approved"
)

// 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
}