Agent-System/internal/app/docnotify/subscriber.go

package docnotify

import (
	"context"
	"fmt"

	"chat-agent-system/internal/docgov"
	"chat-agent-system/internal/docrevision"
	"chat-agent-system/internal/notification"
)

// DocumentLookup resolves a document for recipient and project fill-in.
type DocumentLookup interface {
	GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error)
}

// Subscriber is the unique consumer for document OnOutbox notification events.
// Bind once from the app composition root; do not assign OnOutbox in server.
type Subscriber struct {
	Notify notificationEmitter
	Docs   DocumentLookup
	// SkipWrite is true when the durable notify consumer is live.
	// Bind stays registered while mode=off; the closure must not dual-write after switch.
	SkipWrite func(ctx context.Context) bool
}

type notificationEmitter interface {
	Available() bool
	EmitDocumentEvent(ctx context.Context, eventType, notifType, title, body, dedupeKey string, recipientUserID int64, payload notification.DocumentEventPayload) (bool, error)
}

// Bind assigns the subscriber exactly once. Repeat calls keep the first handler.
func Bind(rev *docrevision.Service, sub *Subscriber) bool {
	if rev == nil || sub == nil || rev.OnOutbox != nil {
		return false
	}
	rev.OnOutbox = sub.Handle
	return true
}

// Handle maps one document outbox event onto a permission-safe inbox notification.
func (s *Subscriber) Handle(ctx context.Context, eventType string, payload map[string]any) error {
	if s == nil {
		return nil
	}
	if s.SkipWrite != nil && s.SkipWrite(ctx) {
		return nil
	}
	if s.Notify == nil || !s.Notify.Available() {
		return nil
	}
	docID, _ := payload["doc_id"].(string)
	projectID, _ := payload["project_id"].(int64)
	risk, _ := payload["risk"].(string)
	var recipient int64
	if s.Docs != nil && docID != "" {
		if doc, err := s.Docs.GetDocument(ctx, docID); err == nil && doc != nil && doc.AuthorID > 0 {
			recipient = doc.AuthorID
			projectID = doc.ProjectID
		}
	}
	if recipient == 0 {
		return nil
	}
	p := notification.DocumentEventPayload{
		ProjectID: projectID, DocID: docID, RiskLevel: risk,
	}
	switch eventType {
	case notification.EventDocAgentCandidateReady:
		if v, ok := payload["candidate_id"].(int64); ok {
			p.CandidateID = v
		}
		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentCandidateReady,
			"智能体变更待确认", "文档有高风险智能体候选变更，请确认后生效。",
			fmt.Sprintf("doc_candidate:%s:%d:%d", docID, p.CandidateID, recipient), recipient, p)
		return err
	case notification.EventDocAgentVersionCommitted:
		if v, ok := payload["version_no"].(int); ok {
			p.VersionNo = v
		}
		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocAgentVersionCommitted,
			"智能体版本已生效", "新的智能体文档版本已提交。",
			fmt.Sprintf("doc_agent_version:%s:%d:%d", docID, p.VersionNo, recipient), recipient, p)
		return err
	case notification.EventDocDownstreamImpact:
		impacted, _ := payload["impacted_doc_id"].(string)
		p.ImpactedDocID = impacted
		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocDownstreamImpact,
			"下游文档受影响", "上游文档变更，请复核关联文档。",
			fmt.Sprintf("doc_impact:%s:%s:%d", docID, impacted, recipient), recipient, p)
		return err
	case notification.EventDocChangeReportDegraded:
		if v, ok := payload["report_id"].(int64); ok {
			p.ReportID = v
		}
		_, err := s.Notify.EmitDocumentEvent(ctx, eventType, notification.TypeDocChangeReportDegraded,
			"变更报告未完整评估", "变更报告降级，受控文档暂不可送审。",
			fmt.Sprintf("doc_report_degraded:%s:%d:%d", docID, p.ReportID, recipient), recipient, p)
		return err
	default:
		return nil
	}
}